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