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