1package client
2
3import (
4 "slices"
5 "sort"
6
7 "github.com/bnema/vev/internal/domain"
8 "github.com/bnema/vev/internal/ports"
9 "github.com/bnema/vev/internal/protocol"
10 "github.com/bnema/vev/internal/protocol/catalogue"
11)
12
13// Client-owned route ledger.
14//
15// The serving daemon renders the status-bar MRU, the other-session bells, the
16// jump-recent hints, and the palette's remote, recent-route, and
17// create-on-host rows from one bounded protocol.RecentRouteSnapshot the client
18// publishes. The ledger builds that snapshot from the broker catalogue, which
19// already observes every daemon (local and remote), so no daemon monitors
20// another one.
21//
22// It is pure: no clock, no I/O, no lock. The supervisor's run goroutine owns
23// it and feeds it each broker publication together with the attachment's
24// committed target.
25//
26// - Every session lifecycle, and every daemon as a creation host, gets a
27// RouteRef that stays stable while it remains in the catalogue. A new
28// lifecycle of the same name is a new reference.
29// - Entries are ordered by this client's attach recency, then live local
30// sessions by catalogue LastUsedSeq, then live remote sessions in
31// catalogue order, then stopped sessions.
32// - Visited marks entries this client attached to. Attach recency sorts
33// first, so visited entries are always a prefix of Entries and Previous
34// is the first of them; the daemon's status history and jump-recent
35// ranks show that prefix plus any ringing entry after it, so visited
36// ranks stay stable, while the palette shows every entry.
37// - The attached lifecycle is metadata only (ActiveEntry), never an entry.
38// - Attention is any(tab.Attention). AttentionSeq orders onsets as the
39// client observed them, across daemons, so the serving daemon can jump to
40// the oldest one.
41// - Generation advances only when the published content changes.
42
43// routeAuthority names one daemon in the broker catalogue.
44type routeAuthority struct {
45 local bool
46 endpoint string
47}
48
49func routeAuthorityOf(observation ports.BrokerDaemonObservation) routeAuthority {
50 if observation.Local {
51 return routeAuthority{local: true}
52 }
53 return routeAuthority{endpoint: observation.Endpoint}
54}
55
56// routeID names one session lifecycle on one daemon. A zero lifecycle names
57// the daemon itself as a creation host.
58type routeID struct {
59 authority routeAuthority
60 lifecycle domain.SessionLifecycleID
61}
62
63// routeLedgerTarget is what one published reference resolves to.
64type routeLedgerTarget struct {
65 authority routeAuthority
66 // target is zero for a creation host.
67 target protocol.ExactSessionTarget
68 host bool
69}
70
71// routeActive is the attachment the snapshot is published over.
72type routeActive struct {
73 known bool
74 authority routeAuthority
75 target protocol.ExactSessionTarget
76}
77
78type routeLedger struct {
79 nextKey uint64
80 keys map[routeID]uint64
81 attachSeq uint64
82 attached map[routeID]uint64
83 attentionSeq uint64
84 attention map[routeID]uint64
85 lastActive routeID
86 generation uint64
87 published protocol.RecentRouteSnapshot
88 targets map[protocol.RouteRef]routeLedgerTarget
89}
90
91func newRouteLedger() *routeLedger {
92 return &routeLedger{
93 keys: make(map[routeID]uint64),
94 attached: make(map[routeID]uint64),
95 attention: make(map[routeID]uint64),
96 targets: make(map[protocol.RouteRef]routeLedgerTarget),
97 }
98}
99
100// routeCandidate is one catalogue session before ordering.
101type routeCandidate struct {
102 id routeID
103 entry protocol.RecentRouteEntry
104 tier int
105 lastUsed uint64
106 daemon int
107 session int
108}
109
110const (
111 routeTierLiveLocal = iota
112 routeTierLiveRemote
113 routeTierStopped
114)
115
116// build folds one broker publication and the attachment's committed target
117// into the next snapshot. It reports false when nothing visible changed, in
118// which case the previously published snapshot is returned unchanged.
119func (l *routeLedger) build(snapshot ports.BrokerSnapshot, active routeActive) (protocol.RecentRouteSnapshot, bool) {
120 activeID := routeID{}
121 if active.known {
122 activeID = routeID{authority: active.authority, lifecycle: active.target.LifecycleID}
123 if activeID != l.lastActive {
124 l.attachSeq++
125 l.attached[activeID] = l.attachSeq
126 l.lastActive = activeID
127 }
128 }
129
130 present := make(map[routeID]struct{})
131 candidates := make([]routeCandidate, 0)
132 attentive := make(map[routeID]struct{})
133 var activeObservation *ports.BrokerDaemonObservation
134 for i := range snapshot.Daemons {
135 daemon := snapshot.Daemons[i]
136 authority := routeAuthorityOf(daemon)
137 if active.known && authority == active.authority {
138 activeObservation = &snapshot.Daemons[i]
139 }
140 for j, session := range daemon.Sessions {
141 if session.State == catalogue.RemoteCatalogSessionBroken {
142 continue
143 }
144 id := routeID{authority: authority, lifecycle: session.LifecycleID}
145 if _, dup := present[id]; dup {
146 continue
147 }
148 entry := routeEntryFor(daemon, session)
149 if validateRouteEntryShape(entry) != nil {
150 continue
151 }
152 present[id] = struct{}{}
153 if entry.Attention {
154 attentive[id] = struct{}{}
155 }
156 tier := routeTierStopped
157 if session.State == catalogue.RemoteCatalogSessionUp {
158 tier = routeTierLiveRemote
159 if daemon.Local {
160 tier = routeTierLiveLocal
161 }
162 }
163 candidates = append(candidates, routeCandidate{id: id, entry: entry, tier: tier, lastUsed: session.LastUsedSeq, daemon: i, session: j})
164 }
165 }
166
167 // Attention onsets keep their order until the attention clears.
168 for id := range l.attention {
169 if _, ok := attentive[id]; !ok {
170 delete(l.attention, id)
171 }
172 }
173 for _, candidate := range candidates {
174 if _, ok := attentive[candidate.id]; ok && l.attention[candidate.id] == 0 {
175 l.attentionSeq++
176 l.attention[candidate.id] = l.attentionSeq
177 }
178 }
179
180 sort.SliceStable(candidates, func(a, b int) bool {
181 left, right := candidates[a], candidates[b]
182 if la, ra := l.attached[left.id], l.attached[right.id]; la != ra {
183 return la > ra
184 }
185 if left.tier != right.tier {
186 return left.tier < right.tier
187 }
188 leftLocal, rightLocal := left.id.authority.local, right.id.authority.local
189 if leftLocal != rightLocal {
190 return leftLocal
191 }
192 if leftLocal && left.lastUsed != right.lastUsed {
193 return left.lastUsed > right.lastUsed
194 }
195 if left.daemon != right.daemon {
196 return left.daemon < right.daemon
197 }
198 return left.session < right.session
199 })
200
201 live := make(map[routeID]struct{}, len(present)+len(snapshot.Daemons)+1)
202 next := protocol.RecentRouteSnapshot{}
203 targets := make(map[protocol.RouteRef]routeLedgerTarget)
204 if active.known {
205 live[activeID] = struct{}{}
206 entry := routeActiveEntry(candidates, activeID, active, activeObservation)
207 entry.Key, entry.Generation = l.keyFor(activeID), 1
208 entry.AttentionSeq = l.attention[activeID]
209 entry.Visited = true
210 if validateRouteEntryShape(entry) == nil {
211 next.ActiveEntry = entry
212 next.Active = protocol.RouteRef{Key: entry.Key, Generation: entry.Generation}
213 targets[next.Active] = routeLedgerTarget{authority: active.authority, target: entry.Target}
214 if active.authority.local {
215 next.Home = next.Active
216 }
217 }
218 }
219 for _, candidate := range candidates {
220 if candidate.id == activeID && active.known {
221 continue
222 }
223 if len(next.Entries) == protocol.RouteSnapshotMaxEntries {
224 break
225 }
226 entry := candidate.entry
227 entry.Key, entry.Generation = l.keyFor(candidate.id), 1
228 entry.AttentionSeq = l.attention[candidate.id]
229 entry.Visited = l.attached[candidate.id] != 0
230 ref := protocol.RouteRef{Key: entry.Key, Generation: entry.Generation}
231 if next.Previous.IsZero() && l.attached[candidate.id] != 0 {
232 next.Previous = ref
233 }
234 next.Entries = append(next.Entries, entry)
235 targets[ref] = routeLedgerTarget{authority: candidate.id.authority, target: entry.Target}
236 live[candidate.id] = struct{}{}
237 }
238 for _, daemon := range snapshot.Daemons {
239 authority := routeAuthorityOf(daemon)
240 if active.known && authority == active.authority {
241 // The serving daemon creates in place; it is never a host row.
242 continue
243 }
244 if len(next.Hosts) == protocol.RouteSnapshotMaxHosts {
245 break
246 }
247 if daemon.Availability == domain.RemoteAvailabilityIncompatible || pickerObservationVersionMismatch(daemon) {
248 continue
249 }
250 kind := protocol.RouteKindRemote
251 if daemon.Local {
252 kind = protocol.RouteKindLocal
253 }
254 label := pickerOriginLabel(daemon)
255 if protocol.ValidateRouteLabel(label, false) != nil {
256 continue
257 }
258 id := routeID{authority: authority}
259 host := protocol.RouteHost{Key: l.keyFor(id), Generation: 1, Label: label, Kind: kind}
260 next.Hosts = append(next.Hosts, host)
261 targets[protocol.RouteRef{Key: host.Key, Generation: host.Generation}] = routeLedgerTarget{authority: authority, host: true}
262 live[id] = struct{}{}
263 }
264 if len(next.Entries) == 0 {
265 next.Entries = nil
266 }
267
268 // Identities that left the catalogue lose their reference and recency.
269 for id := range l.keys {
270 if _, ok := live[id]; !ok {
271 delete(l.keys, id)
272 }
273 }
274 for id := range l.attached {
275 if _, ok := live[id]; !ok {
276 delete(l.attached, id)
277 }
278 }
279
280 if l.generation != 0 && routeSnapshotContentEqual(next, l.published) {
281 return l.published, false
282 }
283 next.Generation = l.generation + 1
284 if next.Validate() != nil {
285 return l.published, false
286 }
287 l.generation = next.Generation
288 l.published = next
289 l.targets = targets
290 return next, true
291}
292
293// resolve returns what one reference of the latest published snapshot names.
294func (l *routeLedger) resolve(ref protocol.RouteRef) (routeLedgerTarget, bool) {
295 target, ok := l.targets[ref]
296 return target, ok
297}
298
299func (l *routeLedger) keyFor(id routeID) uint64 {
300 if key, ok := l.keys[id]; ok {
301 return key
302 }
303 l.nextKey++
304 l.keys[id] = l.nextKey
305 return l.nextKey
306}
307
308// routeEntryFor projects one catalogue session. Identity fields are filled by
309// the caller.
310func routeEntryFor(daemon ports.BrokerDaemonObservation, session catalogue.RemoteCatalogSession) protocol.RecentRouteEntry {
311 entry := protocol.RecentRouteEntry{
312 Target: protocol.ExactSessionTarget{LifecycleID: session.LifecycleID, SessionName: session.Name},
313 Name: session.Name,
314 Kind: protocol.RouteKindLocal,
315 Ephemeral: session.Ephemeral,
316 Reachability: routeReachability(daemon),
317 }
318 if !daemon.Local {
319 entry.Kind = protocol.RouteKindRemote
320 entry.HostLabel = pickerOriginLabel(daemon)
321 }
322 for _, tab := range session.Tabs {
323 if tab.Attention {
324 entry.Attention = true
325 break
326 }
327 }
328 return entry
329}
330
331// routeActiveEntry is the attached lifecycle's presentation: its catalogue
332// row when the broker already observed it, otherwise the committed target.
333func routeActiveEntry(candidates []routeCandidate, id routeID, active routeActive, observation *ports.BrokerDaemonObservation) protocol.RecentRouteEntry {
334 for _, candidate := range candidates {
335 if candidate.id == id {
336 return candidate.entry
337 }
338 }
339 entry := protocol.RecentRouteEntry{
340 Target: active.target,
341 Name: active.target.SessionName,
342 Kind: protocol.RouteKindLocal,
343 }
344 if !active.authority.local {
345 entry.Kind = protocol.RouteKindRemote
346 if observation != nil {
347 entry.HostLabel = pickerOriginLabel(*observation)
348 }
349 }
350 return entry
351}
352
353func routeReachability(daemon ports.BrokerDaemonObservation) protocol.RouteReachability {
354 switch {
355 case daemon.Availability == domain.RemoteAvailabilityReachable:
356 return protocol.RouteReachabilityReachable
357 case daemon.Availability.Failing():
358 return protocol.RouteReachabilityUnavailable
359 default:
360 return protocol.RouteReachabilityUnknown
361 }
362}
363
364// validateRouteEntryShape checks the fields a catalogue row contributes, with
365// a placeholder identity.
366func validateRouteEntryShape(entry protocol.RecentRouteEntry) error {
367 entry.Key, entry.Generation = 1, 1
368 return protocol.RecentRouteSnapshot{Generation: 1, Entries: []protocol.RecentRouteEntry{entry}}.Validate()
369}
370
371func routeSnapshotContentEqual(a, b protocol.RecentRouteSnapshot) bool {
372 return a.Active == b.Active && a.ActiveEntry == b.ActiveEntry && a.Previous == b.Previous && a.Home == b.Home &&
373 slices.Equal(a.Entries, b.Entries) && slices.Equal(a.Hosts, b.Hosts)
374}