1package app
2
3import (
4 "context"
5 "fmt"
6 "sync"
7 "time"
8
9 "github.com/bnema/neferwl/internal/adapters/capture"
10
11 "github.com/bnema/neferwl/internal/ports"
12)
13
14// outputRun drives one output until ctx ends: it renders the scenes and
15// surface contents it receives, with the cursor the client asks for.
16type outputRun func(ctx context.Context, scenes <-chan ports.Scene, contents <-chan ports.SurfaceContent, cursor <-chan ports.CursorChange, captures <-chan ports.CaptureRequest, security <-chan ports.SecurityState, instance ports.OutputInstance) error
17
18// outputSet owns the running outputs and feeds them. Core sends one scene
19// per output; each goes to its output, latest first. Surface contents go to
20// every output, since windows move between outputs; the latest content of
21// each window is kept so a new output starts with every window drawn.
22// Only the goroutine running loop touches the set.
23type outputSet struct {
24 outs map[string]*runningOutput
25 latest map[ports.WindowID]ports.SurfaceContent
26 cursor ports.CursorChange
27 stopped chan outputStopped
28 captured chan<- ports.CaptureDone
29 ctx context.Context
30 quit chan struct{} // closed by wait: nobody reads stopped any more
31 security ports.SessionSecurity
32 securityEvents chan<- ports.SecurityBackendEvent
33 state ports.SecurityState
34 nextInstance ports.OutputInstance
35 startErr error
36 barriers chan ports.SecurityBackendEvent
37 forwarders sync.WaitGroup
38}
39
40// A stop is tied to a lifetime, never just a reusable connector name.
41type outputStopped struct {
42 name string
43 instance ports.OutputInstance
44}
45
46type runningOutput struct {
47 instance ports.OutputInstance
48 security chan ports.SecurityState
49 securityEvents chan ports.SecurityBackendEvent
50 pending []ports.SurfaceContent
51 scenes chan ports.Scene
52 contents chan ports.SurfaceContent
53 cursor chan ports.CursorChange
54 captures chan ports.CaptureRequest
55 ctx context.Context
56 stop context.CancelFunc
57 err error
58 done chan struct{}
59}
60
61func newOutputSet(ctx context.Context, captured chan<- ports.CaptureDone) *outputSet {
62 return &outputSet{captured: captured, ctx: ctx, outs: map[string]*runningOutput{}, latest: map[ports.WindowID]ports.SurfaceContent{}, stopped: make(chan outputStopped), quit: make(chan struct{})}
63}
64
65// start runs an output in its own goroutine and replays the window contents.
66func (s *outputSet) start(ctx context.Context, name string, run outputRun, wire ...*chan<- ports.SecurityBackendEvent) bool {
67 // Exhaustion is permanent: never wrap to zero or reuse a lifetime.
68 if s.nextInstance == ^ports.OutputInstance(0) {
69 s.startErr = fmt.Errorf("output instance exhausted")
70 return false
71 }
72 // Register authoritatively before an owner can render or replay contents.
73 s.nextInstance++
74 instance := s.nextInstance
75 if !s.securityEvent(ports.SecurityOutputAdded{Instance: instance, Output: name}) {
76 return false
77 }
78 if s.security != nil {
79 s.setSecurity(s.security.Snapshot())
80 }
81 octx, stop := context.WithCancel(ctx)
82 r := &runningOutput{instance: instance, security: make(chan ports.SecurityState, 1), scenes: make(chan ports.Scene, 1), contents: make(chan ports.SurfaceContent, 64), cursor: make(chan ports.CursorChange, 1), captures: make(chan ports.CaptureRequest, 16), ctx: octx, stop: stop, done: make(chan struct{})}
83 r.security <- s.state // available before any old scene/content replay
84 r.securityEvents = s.forwardSecurity()
85 for _, destination := range wire {
86 *destination = r.securityEvents
87 }
88 for _, c := range s.latest {
89 select {
90 case r.contents <- c:
91 default:
92 r.pending = append(r.pending, c)
93 }
94 }
95 s.outs[name] = r
96 go func() {
97 r.err = run(octx, r.scenes, r.contents, r.cursor, r.captures, r.security, r.instance)
98 if r.securityEvents != nil {
99 close(r.securityEvents)
100 }
101 stop()
102 close(r.done)
103 select {
104 case s.stopped <- outputStopped{name: name, instance: instance}:
105 case <-s.quit:
106 }
107 }()
108 r.cursor <- s.cursor
109 return true
110}
111
112func (s *outputSet) wireSecurity(ch outputChannels) {
113 s.security, s.securityEvents = ch.security, ch.securityEvents
114 if ch.securityEvents != nil {
115 s.barriers = s.forwardSecurity()
116 }
117 if s.security != nil {
118 s.setSecurity(s.security.Snapshot())
119 }
120}
121
122// securityEvent registers/removes directly, while barriers use a bounded FIFO.
123// No authoritative event is merged or discarded during an active backend.
124func (s *outputSet) securityEvent(ev ports.SecurityBackendEvent) bool {
125 if s.securityEvents == nil {
126 return true // security disabled
127 }
128 destination := s.securityEvents
129 if _, ok := ev.(ports.SecurityBackendBarrier); ok && s.barriers != nil {
130 destination = s.barriers
131 }
132 select {
133 case destination <- ev:
134 return true
135 case <-s.ctx.Done():
136 return false
137 }
138}
139
140// Each producer has a finite, FIFO outbox. Saturation applies backpressure;
141// events are never merged. Registration/removal bypass it for lifecycle ordering.
142func (s *outputSet) forwardSecurity() chan ports.SecurityBackendEvent {
143 if s.securityEvents == nil {
144 return nil
145 }
146 incoming := make(chan ports.SecurityBackendEvent, 16)
147 s.forwarders.Add(1)
148 go func() {
149 defer s.forwarders.Done()
150 for {
151 select {
152 case ev, ok := <-incoming:
153 if !ok {
154 return
155 }
156 select {
157 case s.securityEvents <- ev:
158 case <-s.ctx.Done():
159 return
160 case <-s.quit:
161 return
162 }
163 case <-s.ctx.Done():
164 return
165 case <-s.quit:
166 return
167 }
168 }
169 }()
170 return incoming
171}
172
173func (s *outputSet) setSecurity(state ports.SecurityState) bool {
174 if s.security != nil {
175 if newest := s.security.Snapshot(); newest.Generation > state.Generation {
176 state = newest
177 }
178 }
179 if state.Generation < s.state.Generation || state == s.state {
180 return false
181 }
182 s.state = state
183 for _, r := range s.outs {
184 select {
185 case <-r.security:
186 default:
187 }
188 r.security <- state
189 }
190 return true
191}
192
193// stoppedCurrent rejects delayed stop messages from a previous incarnation.
194func (s *outputSet) stoppedCurrent(stopped outputStopped) bool {
195 r := s.outs[stopped.name]
196 return r != nil && r.instance == stopped.instance
197}
198
199// setCursor gives every output the latest cursor, replacing an unread one.
200func (s *outputSet) setCursor(c ports.CursorChange) {
201 s.cursor = c
202 for _, r := range s.outs {
203 select {
204 case <-r.cursor:
205 default:
206 }
207 r.cursor <- c
208 }
209}
210
211// scenes routes each scene to its output, replacing an unread one.
212func (s *outputSet) scenes(set []ports.Scene) {
213 for _, sc := range set {
214 if r := s.outs[sc.Output]; r != nil {
215 select {
216 case <-r.scenes:
217 default:
218 }
219 r.scenes <- sc
220 }
221 }
222}
223
224func (s *outputSet) content(c ports.SurfaceContent) {
225 if c.Empty() {
226 delete(s.latest, c.ID)
227 } else {
228 s.latest[c.ID] = c
229 }
230 for _, r := range s.outs {
231 if r.ctx.Err() != nil {
232 continue
233 }
234 if len(r.pending) == 0 {
235 select {
236 case r.contents <- c:
237 continue
238 default:
239 }
240 }
241 // The backend disabled intake until the preceding broadcast drained.
242 if cap(r.pending) == 0 {
243 r.pending = make([]ports.SurfaceContent, 0, 1)
244 }
245 r.pending = append(r.pending, c)
246 }
247}
248
249// At most one admitted broadcast waits behind each owner's finite replay.
250// Disable new content intake until all copies have been delivered.
251func (s *outputSet) contentOut(incoming <-chan ports.SurfaceContent) (<-chan ports.SurfaceContent, chan<- ports.SurfaceContent, ports.SurfaceContent, *runningOutput) {
252 for _, r := range s.outs {
253 if len(r.pending) != 0 && r.ctx.Err() == nil {
254 return nil, r.contents, r.pending[0], r
255 }
256 }
257 return incoming, nil, ports.SurfaceContent{}, nil
258}
259
260func (r *runningOutput) contentSent() {
261 r.pending[0] = ports.SurfaceContent{}
262 if len(r.pending) == 1 {
263 r.pending = r.pending[:0] // retain the one-slot broadcast buffer
264 } else {
265 r.pending = r.pending[1:]
266 }
267}
268
269// finish forgets a stopped output and returns its error.
270func (s *outputSet) finish(name string) error {
271 r := s.outs[name]
272 if r == nil {
273 return nil
274 }
275 <-r.done
276 r.stop()
277 s.failQueued(r)
278 delete(s.outs, name)
279 return r.err
280}
281
282// wait stops every output and returns the first error.
283func (s *outputSet) wait() error {
284 var first error = s.startErr
285 // Cancel producers before closing the forwarding quit path. Owners may
286 // be backpressured on a full FIFO, and need their context to exit.
287 for _, r := range s.outs {
288 r.stop()
289 }
290 close(s.quit)
291 s.forwarders.Wait()
292 for name, r := range s.outs {
293 r.stop()
294 <-r.done
295 s.failQueued(r)
296 if first == nil && r.err != nil {
297 first = r.err
298 }
299 delete(s.outs, name)
300 }
301 return first
302}
303
304// cursors routes pointer moves from the input goroutine to the cursor of
305// the output under the pointer and hides the others. After hideAfter
306// without motion it hides that cursor too; the next motion shows it.
307type cursors struct {
308 mu sync.Mutex
309 all map[string]cursor
310 last string
311 clock ports.Clock
312 // hideAfter is 0 when the cursor never hides. lastMove is when the
313 // latest motion came; armed tells the timer runs; hidden tells the
314 // cursor is idle-hidden. The timer is not reset on every motion: when
315 // it fires early it re-arms for the time left.
316 hideAfter time.Duration
317 lastMove time.Time
318 timer ports.Timer
319 armed bool
320 hidden bool
321}
322
323// cursor is a hardware or software cursor of one output.
324type cursor interface {
325 Move(x, y float64)
326 Hide()
327}
328
329func newCursors(clock ports.Clock, hideAfter time.Duration) *cursors {
330 return &cursors{all: map[string]cursor{}, clock: clock, hideAfter: hideAfter}
331}
332
333// setHideAfter changes the idle delay; 0 never hides the cursor. A cursor
334// already hidden stays hidden until the next motion.
335func (c *cursors) setHideAfter(d time.Duration) {
336 c.mu.Lock()
337 defer c.mu.Unlock()
338 if d == c.hideAfter {
339 return
340 }
341 c.hideAfter = d
342 c.disarm()
343 if d > 0 && c.last != "" && !c.hidden {
344 c.lastMove = c.clock.Now()
345 c.arm(d)
346 }
347}
348
349// stop cancels a pending hide; the router is not used after.
350func (c *cursors) stop() {
351 c.mu.Lock()
352 defer c.mu.Unlock()
353 c.disarm()
354}
355
356// disarm stops the idle timer. Callers hold mu.
357func (c *cursors) disarm() {
358 if c.timer != nil {
359 c.timer.Stop()
360 }
361 c.armed = false
362}
363
364// arm starts the idle timer for d. Callers hold mu.
365func (c *cursors) arm(d time.Duration) {
366 c.armed = true
367 if c.timer == nil {
368 c.timer = c.clock.AfterFunc(d, c.idle)
369 return
370 }
371 c.timer.Reset(d)
372}
373
374// idle hides the cursor once hideAfter passed since the last move, or
375// re-arms the timer for the time left.
376func (c *cursors) idle() {
377 c.mu.Lock()
378 defer c.mu.Unlock()
379 if !c.armed || c.hideAfter <= 0 {
380 return
381 }
382 if left := c.hideAfter - c.clock.Now().Sub(c.lastMove); left > 0 {
383 c.timer.Reset(left)
384 return
385 }
386 c.armed, c.hidden = false, true
387 if cur := c.all[c.last]; cur != nil {
388 cur.Hide()
389 }
390}
391
392// set registers the cursor of an output; nil removes it.
393func (c *cursors) set(output string, cur cursor) {
394 c.mu.Lock()
395 defer c.mu.Unlock()
396 if cur == nil {
397 delete(c.all, output)
398 return
399 }
400 c.all[output] = cur
401 // Only the output under the pointer shows it; before the first move,
402 // none does (a new cursor would show at its top-left corner). An idle
403 // hidden cursor stays hidden.
404 if output != c.last || c.hidden {
405 cur.Hide()
406 }
407}
408
409// move places the cursor on output at physical (x, y). motion is false
410// when the pointer is only placed again (layout or constraint change): it
411// neither shows an idle-hidden cursor nor delays the hide.
412func (c *cursors) move(output string, x, y float64, motion bool) {
413 c.mu.Lock()
414 defer c.mu.Unlock()
415 if motion {
416 c.hidden = false
417 }
418 if output != c.last {
419 for name, cur := range c.all {
420 if name != output {
421 cur.Hide()
422 }
423 }
424 c.last = output
425 }
426 if c.hidden {
427 return // the next motion moves it with fresh coordinates
428 }
429 if cur := c.all[output]; cur != nil {
430 cur.Move(x, y)
431 }
432 if c.hideAfter > 0 && (motion || !c.armed) {
433 c.lastMove = c.clock.Now()
434 if !c.armed {
435 c.arm(c.hideAfter)
436 }
437 }
438}
439
440// routeCapture transfers the request to its output or closes it on failure.
441func (s *outputSet) routeCapture(req ports.CaptureRequest) {
442 r := s.outs[req.Output]
443 if r == nil {
444 capture.Fail(s.ctx, req, fmt.Errorf("output %q unavailable", req.Output), s.captured)
445 return
446 }
447 select {
448 case r.captures <- req:
449 case <-r.done:
450 r.stop()
451 capture.Fail(s.ctx, req, fmt.Errorf("output %q stopped", req.Output), s.captured)
452 case <-r.ctx.Done():
453 capture.Fail(s.ctx, req, fmt.Errorf("output %q stopped", req.Output), s.captured)
454 default:
455 capture.Fail(s.ctx, req, fmt.Errorf("output %q capture queue full", req.Output), s.captured)
456 }
457}
458
459func (s *outputSet) failQueued(r *runningOutput) {
460 for {
461 select {
462 case req := <-r.captures:
463 capture.Fail(s.ctx, req, fmt.Errorf("output stopped"), s.captured)
464 default:
465 return
466 }
467 }
468}
469
470// drainCaptures releases requests left in the app queue. Call it once the
471// producer (the wayland server) has stopped, so none can arrive after it.
472func drainCaptures(ctx context.Context, incoming <-chan ports.CaptureRequest, replies chan<- ports.CaptureDone) {
473 for {
474 select {
475 case q, ok := <-incoming:
476 if !ok {
477 return
478 }
479 capture.Fail(ctx, q, fmt.Errorf("output stopped"), replies)
480 default:
481 return
482 }
483 }
484}