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