1package uidriver
2
3import (
4 "bufio"
5 "context"
6 "encoding/json"
7 "errors"
8 "io"
9 "sync"
10
11 clockadapter "github.com/bnema/vev/internal/adapters/clock"
12 "github.com/bnema/vev/internal/ports"
13)
14
15const (
16 responseSlots = 3
17 errorResponseReserve = 1024
18)
19
20// Server shares attachment-wide connection, action, and serialized-response
21// limits. Closing a stream cancels its request waiters, never the attachment's
22// Runner.
23type Server struct {
24 service ports.UIService
25 clock ports.Clock
26
27 mu sync.Mutex
28 connections int
29 serialized int
30 actionBusy bool
31}
32
33func New(service ports.UIService, clock ports.Clock) *Server {
34 if clock == nil {
35 clock = clockadapter.New()
36 }
37 return &Server{service: service, clock: clock}
38}
39
40func (s *Server) reserve(bytes int) bool {
41 s.mu.Lock()
42 defer s.mu.Unlock()
43 // Keep room for one small error in each connection's response slots.
44 limit := maxSerializedBytes
45 if bytes > errorResponseReserve {
46 limit -= maxConnections * responseSlots * errorResponseReserve
47 }
48 if s.serialized+bytes > limit {
49 return false
50 }
51 s.serialized += bytes
52 return true
53}
54
55func (s *Server) release(bytes int) {
56 s.mu.Lock()
57 s.serialized -= bytes
58 s.mu.Unlock()
59}
60
61func (s *Server) admitAction() bool {
62 s.mu.Lock()
63 defer s.mu.Unlock()
64 if s.actionBusy {
65 return false
66 }
67 s.actionBusy = true
68 return true
69}
70
71func (s *Server) finishAction() {
72 s.mu.Lock()
73 s.actionBusy = false
74 s.mu.Unlock()
75}
76
77type queuedResponse struct {
78 data []byte
79 reserved int
80 id uint64
81 ownsID bool
82}
83
84type actionCompletionTracker interface {
85 ActionComplete(actionID uint64) <-chan struct{}
86}
87
88func (s *Server) releaseActionWhenDone(ctx context.Context, done <-chan struct{}) {
89 select {
90 case <-done:
91 s.finishAction()
92 case <-ctx.Done():
93 s.finishAction()
94 }
95}
96
97// Serve owns stream until return. Close must interrupt its blocked reads and
98// writes (Unix sockets and the app's owned stdio pipes satisfy this contract).
99func (s *Server) Serve(ctx context.Context, stream io.ReadWriteCloser, ready Ready) error {
100 s.mu.Lock()
101 if s.connections >= maxConnections {
102 s.mu.Unlock()
103 _ = stream.Close()
104 return &ports.UIError{Code: ports.UIErrBusy}
105 }
106 s.connections++
107 s.mu.Unlock()
108 defer func() {
109 s.mu.Lock()
110 s.connections--
111 s.mu.Unlock()
112 }()
113 defer func() { _ = stream.Close() }()
114
115 serving, cancelServing := context.WithCancel(ctx)
116 defer cancelServing()
117 requests, cancelRequests := context.WithCancel(serving)
118 defer cancelRequests()
119 var closeOnce sync.Once
120 closeStream := func() { closeOnce.Do(func() { _ = stream.Close() }) }
121 watcherDone := make(chan struct{})
122 stopWatcher := make(chan struct{})
123 go func() {
124 defer close(watcherDone)
125 select {
126 case <-serving.Done():
127 closeStream()
128 case <-stopWatcher:
129 }
130 }()
131 defer func() {
132 close(stopWatcher)
133 <-watcherDone
134 }()
135
136 queue := make(chan queuedResponse, responseSlots)
137 slots := make(chan struct{}, responseSlots)
138 var stateMu sync.Mutex
139 ids := make(map[uint64]bool)
140 var waitActive bool
141
142 writerDone := make(chan error, 1)
143 go func() {
144 var writeErr error
145 for response := range queue {
146 if writeErr == nil {
147 writeErr = s.writeResponse(serving, closeStream, stream, response.data)
148 }
149 s.release(response.reserved)
150 <-slots
151 if response.ownsID {
152 stateMu.Lock()
153 delete(ids, response.id)
154 stateMu.Unlock()
155 }
156 }
157 writerDone <- writeErr
158 }()
159
160 respond := func(id uint64, ownsID bool, build func() envelope) bool {
161 select {
162 case slots <- struct{}{}:
163 case <-serving.Done():
164 return false
165 default:
166 cancelServing()
167 closeStream()
168 return false
169 }
170
171 reserved := maxResponseBytes
172 var response envelope
173 if !s.reserve(reserved) {
174 reserved = errorResponseReserve
175 if !s.reserve(reserved) {
176 <-slots
177 cancelServing()
178 closeStream()
179 return false
180 }
181 response = errorEnvelope(id, &ports.UIError{Code: ports.UIErrBusy})
182 } else {
183 // Build outside all adapter and terminal locks. Capture owns its own
184 // immutable snapshot and may be evaluated while a wait is blocked.
185 response = build()
186 }
187 data, err := marshalResponse(response)
188 if err != nil {
189 response = errorEnvelope(id, &ports.UIError{Code: ports.UIErrCaptureTooLarge})
190 data, err = marshalResponse(response)
191 if err != nil {
192 // The fixed error envelope is deliberately tiny; reaching this
193 // branch means the process cannot produce a valid response.
194 s.release(reserved)
195 <-slots
196 cancelServing()
197 closeStream()
198 return false
199 }
200 }
201 queued := queuedResponse{data: data, reserved: reserved, id: id, ownsID: ownsID}
202 select {
203 case queue <- queued:
204 return true
205 case <-serving.Done():
206 s.release(reserved)
207 <-slots
208 return false
209 }
210 }
211
212 if !respond(0, false, func() envelope { return envelope{Version: apiVersion, Result: ready} }) {
213 cancelServing()
214 closeStream()
215 }
216
217 var workers sync.WaitGroup
218 scanner := bufio.NewScanner(stream)
219 scanner.Buffer(make([]byte, 4096), maxRequestBytes+2)
220 for scanner.Scan() {
221 parsed, err := decodeRequest(scanner.Bytes())
222 if err != nil {
223 if !respond(parsed.ID, false, func() envelope { return errorEnvelope(parsed.ID, err) }) {
224 break
225 }
226 continue
227 }
228
229 stateMu.Lock()
230 duplicate := ids[parsed.ID]
231 if !duplicate {
232 ids[parsed.ID] = true
233 }
234 isAction := parsed.Op == opKeys || parsed.Op == opText
235 busy := false
236 if !duplicate && isAction && ready.Control {
237 busy = !s.admitAction()
238 }
239 if !duplicate && parsed.Op == opWait {
240 busy = waitActive
241 if !busy {
242 waitActive = true
243 }
244 }
245 stateMu.Unlock()
246
247 if duplicate || busy {
248 code := ports.UIErrBusy
249 if duplicate {
250 code = ports.UIErrInvalidRequest
251 }
252 if !respond(parsed.ID, !duplicate, func() envelope {
253 return errorEnvelope(parsed.ID, &ports.UIError{Code: code})
254 }) {
255 break
256 }
257 continue
258 }
259
260 execute := func() {
261 holdAction := false
262 defer func() {
263 stateMu.Lock()
264 if parsed.Op == opWait {
265 waitActive = false
266 }
267 stateMu.Unlock()
268 if isAction && ready.Control && !holdAction {
269 s.finishAction()
270 }
271 }()
272 if isAction && !ready.Control {
273 respond(parsed.ID, true, func() envelope {
274 return errorEnvelope(parsed.ID, &ports.UIError{Code: ports.UIErrPermissionDenied})
275 })
276 return
277 }
278 switch parsed.Op {
279 case opCapture:
280 respond(parsed.ID, true, func() envelope {
281 snapshot, err := s.service.Capture(parsed.Attachment)
282 if err != nil {
283 return errorEnvelope(parsed.ID, err)
284 }
285 capture, err := publicCapture(snapshot, parsed.Format)
286 if err != nil {
287 return errorEnvelope(parsed.ID, err)
288 }
289 return envelope{Version: apiVersion, ID: parsed.ID, Result: capture}
290 })
291 case opKeys, opText:
292 result, err := s.service.Action(serving, parsed.Action)
293 if err != nil {
294 var uiErr *ports.UIError
295 if errors.As(err, &uiErr) && uiErr.Accepted && uiErr.ActionID != 0 {
296 if tracker, ok := s.service.(actionCompletionTracker); ok {
297 if done := tracker.ActionComplete(uiErr.ActionID); done != nil {
298 holdAction = true
299 go s.releaseActionWhenDone(serving, done)
300 }
301 }
302 }
303 }
304 respond(parsed.ID, true, func() envelope {
305 if err != nil {
306 return errorEnvelope(parsed.ID, err)
307 }
308 return envelope{Version: apiVersion, ID: parsed.ID, Result: actionResponse{ActionID: result.ActionID, Accepted: result.Accepted, Status: result.Status, Revision: result.Revision, Context: publicContext(result.Context)}}
309 })
310 case opWait:
311 result, err := s.service.Wait(requests, parsed.Wait)
312 respond(parsed.ID, true, func() envelope {
313 if err != nil {
314 return errorEnvelope(parsed.ID, err)
315 }
316 return envelope{Version: apiVersion, ID: parsed.ID, Result: waitResponse{ActionID: result.ActionID, ActionStatus: result.ActionStatus, Revision: result.Revision, Context: publicContext(result.Context)}}
317 })
318 }
319 }
320
321 // Capture is kept on the read loop so a fast query is never starved by
322 // an unbounded operation worker. Wait and action must be independent of
323 // subsequent reads.
324 if parsed.Op == opCapture {
325 execute()
326 } else {
327 workers.Add(1)
328 go func() {
329 defer workers.Done()
330 execute()
331 }()
332 }
333 if serving.Err() != nil {
334 break
335 }
336 }
337
338 readErr := scanner.Err()
339 if readErr != nil && serving.Err() == nil {
340 respond(0, false, func() envelope { return errorEnvelope(0, invalidRequest()) })
341 }
342 cancelRequests()
343 workers.Wait()
344 close(queue)
345 writeErr := <-writerDone
346 if ctx.Err() != nil {
347 return ctx.Err()
348 }
349 if writeErr != nil {
350 return &ports.UIError{Code: ports.UIErrUnavailable}
351 }
352 if readErr != nil && !errors.Is(readErr, io.EOF) {
353 return invalidRequest()
354 }
355 return nil
356}
357
358func (s *Server) writeResponse(ctx context.Context, closeStream func(), stream io.Writer, data []byte) error {
359 timer := s.clock.NewTimer(writeTimeout)
360 writeDone := make(chan struct{})
361 deadlineDone := make(chan struct{})
362 go func() {
363 defer close(deadlineDone)
364 select {
365 case <-timer.C():
366 closeStream()
367 case <-writeDone:
368 case <-ctx.Done():
369 closeStream()
370 }
371 }()
372 n, err := stream.Write(data)
373 close(writeDone)
374 timer.Stop()
375 <-deadlineDone
376 if err != nil {
377 return err
378 }
379 if n != len(data) {
380 return io.ErrShortWrite
381 }
382 return nil
383}
384
385func marshalResponse(response envelope) ([]byte, error) {
386 data, err := json.Marshal(response)
387 if err != nil {
388 return nil, err
389 }
390 if len(data)+1 > maxResponseBytes {
391 return nil, errors.New("response exceeds configured size")
392 }
393 return append(data, '\n'), nil
394}