remote_catalog_state.go RawHistoryFolder

1package daemon
2
3import (
4	"context"
5	"slices"
6	"sort"
7	"sync"
8	"time"
9
10	"github.com/bnema/vev/internal/ports"
11)
12
13type remoteHostStatus uint8
14
15const (
16	remoteHostCached remoteHostStatus = iota
17	remoteHostRefreshing
18	remoteHostFresh
19	remoteHostStale
20	remoteHostUnreachable
21	remoteHostVersionMismatch
22	remoteHostMalformed
23)
24
25// remoteCatalogState owns the cache-derived discovery state. Its mutex never
26// covers cache or transport I/O; writeMu only serializes complete cache writes.
27type remoteCatalogState struct {
28	mu      sync.Mutex
29	cache   map[string]ports.RemoteCatalogCacheEntry
30	status  map[string]remoteHostStatus
31	failure map[string]time.Time
32	refresh uint64
33	cancel  context.CancelFunc
34	pickers map[*attachedClient]struct{}
35	writeMu sync.Mutex
36}
37
38func newRemoteCatalogState() remoteCatalogState {
39	return remoteCatalogState{
40		cache:   make(map[string]ports.RemoteCatalogCacheEntry),
41		status:  make(map[string]remoteHostStatus),
42		failure: make(map[string]time.Time),
43		pickers: make(map[*attachedClient]struct{}),
44	}
45}
46
47func cloneRemoteCatalogEntry(entry ports.RemoteCatalogCacheEntry) ports.RemoteCatalogCacheEntry {
48	sessions := make([]ports.RemoteCatalogSession, len(entry.Sessions))
49	for i, session := range entry.Sessions {
50		sessions[i] = session
51		if tabs := ports.CatalogTabs(session); tabs != nil {
52			sessions[i].Tabs = slices.Clone(tabs)
53		}
54	}
55	return ports.RemoteCatalogCacheEntry{
56		Host:      entry.Host,
57		FetchedAt: entry.FetchedAt,
58		Sessions:  sessions,
59	}
60}
61
62func (s *remoteCatalogState) replaceCache(entries []ports.RemoteCatalogCacheEntry) {
63	if s == nil {
64		return
65	}
66	s.mu.Lock()
67	defer s.mu.Unlock()
68	s.cache = make(map[string]ports.RemoteCatalogCacheEntry, len(entries))
69	s.status = make(map[string]remoteHostStatus, len(entries))
70	s.failure = make(map[string]time.Time)
71	for _, entry := range entries {
72		s.cache[entry.Host] = cloneRemoteCatalogEntry(entry)
73		s.status[entry.Host] = remoteHostCached
74	}
75}
76
77func (s *remoteCatalogState) cacheSnapshot() []ports.RemoteCatalogCacheEntry {
78	if s == nil {
79		return nil
80	}
81	s.mu.Lock()
82	defer s.mu.Unlock()
83	entries := make([]ports.RemoteCatalogCacheEntry, 0, len(s.cache))
84	for _, entry := range s.cache {
85		entries = append(entries, cloneRemoteCatalogEntry(entry))
86	}
87	sort.Slice(entries, func(i, j int) bool { return entries[i].Host < entries[j].Host })
88	return entries
89}
90
91func (d *Daemon) loadRemoteCatalogCache() {
92	if d == nil || d.remoteCatalogCache == nil {
93		return
94	}
95	entries, err := d.remoteCatalogCache.Load()
96	if err != nil {
97		d.log.Warn("loading remote catalog cache failed; starting empty", "err", err)
98		return
99	}
100	d.remoteCatalog.replaceCache(entries)
101}
102
103// persistRemoteCatalogCache writes the newest complete cache snapshot. Cache
104// I/O happens after releasing the state mutex and remains serialized with
105// other writes so an older snapshot cannot win a later atomic replacement.
106func (d *Daemon) persistRemoteCatalogCache() {
107	if d == nil || d.remoteCatalogCache == nil {
108		return
109	}
110	d.remoteCatalog.writeMu.Lock()
111	defer d.remoteCatalog.writeMu.Unlock()
112
113	entries := d.remoteCatalog.cacheSnapshot()
114	if err := d.remoteCatalogCache.Store(entries); err != nil {
115		d.log.Warn("storing remote catalog cache failed", "err", err)
116	}
117}