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