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