1package persist
2
3import (
4 "errors"
5 "fmt"
6 "os"
7 "path/filepath"
8 "sort"
9 "sync"
10
11 "github.com/bnema/vev/internal/domain"
12 "github.com/bnema/vev/internal/ports"
13 "github.com/bnema/vev/pkg/kv"
14)
15
16const filename = "sessions.kv"
17
18var errPersistenceUnavailable = errors.New("persist: catalogue unavailable")
19
20// ErrCatalogueDurability reports a failed identity write whose durability is
21// ambiguous. A Persister remains fenced after this error so rejected state
22// cannot be made durable by a later flush.
23var ErrCatalogueDurability = errors.New("persist: catalogue durability rejected")
24
25// ErrCatalogueUnreadable reports durable state that exists but cannot be
26// decoded. vev never repairs or erases it; the operator resets explicitly.
27var ErrCatalogueUnreadable = errors.New("persist: catalogue unreadable")
28
29// OpenResult is the outcome of opening durable session state at startup.
30type OpenResult struct {
31 Catalogue *Persister
32 Records []domain.CatalogueRecord
33 Migration Migration
34 NewInstall bool
35}
36
37func StorePath(dir string) string { return filepath.Join(dir, filename) }
38
39type Persister struct {
40 store ports.Store
41 mu sync.Mutex
42
43 incarnationOwners map[domain.IncarnationID]string
44 nameIncarnations map[string]domain.IncarnationID
45 incarnationIndex bool
46 terminalErr error
47}
48
49// KVStore adapts the reusable whole-file store to the persistence port.
50type KVStore struct{ store *kv.Store }
51
52func OpenStore(path string) (*KVStore, error) {
53 store, err := kv.Open(path)
54 if err != nil {
55 return nil, err
56 }
57 return &KVStore{store: store}, nil
58}
59
60func (s *KVStore) Get(key []byte) ([]byte, bool) { return s.store.Get(key) }
61func (s *KVStore) Set(key, value []byte) error { return s.store.Set(key, value) }
62func (s *KVStore) Delete(key []byte) error { return s.store.Delete(key) }
63func (s *KVStore) Range(fn func(key, value []byte) bool) {
64 s.store.Range(fn)
65}
66func (s *KVStore) Sync() error { return s.store.Sync() }
67func (s *KVStore) Close() error { return s.store.Close() }
68
69func (s *KVStore) CloseWithoutSync() error { return s.store.CloseWithoutSync() }
70
71// Open opens an existing strict catalogue and never creates an unproven empty one.
72func Open(dir string) (*Persister, error) {
73 p, _, err := openCurrentCatalogue(dir, false)
74 return p, err
75}
76
77// OpenOrCreate opens the catalogue, or creates a proven-empty one when no
78// durable state exists. Any state that exists but does not decode is returned
79// as ErrCatalogueUnreadable and left untouched on disk.
80func OpenOrCreate(dir string) (OpenResult, error) {
81 path := StorePath(dir)
82 _, statErr := os.Stat(path)
83 existed := statErr == nil
84 if statErr != nil && !errors.Is(statErr, os.ErrNotExist) {
85 return OpenResult{}, fmt.Errorf("%w: %s: %w", ErrCatalogueUnreadable, path, statErr)
86 }
87 catalogue, records, err := openCurrentCatalogue(dir, true)
88 if err != nil {
89 return OpenResult{}, fmt.Errorf("%w: %s: %w", ErrCatalogueUnreadable, path, err)
90 }
91 migration, err := catalogue.migrateLegacyRecords(path, records)
92 if err != nil {
93 return OpenResult{}, errors.Join(fmt.Errorf("%w: %s: %w", ErrCatalogueUnreadable, path, err), catalogue.Close())
94 }
95 return OpenResult{Catalogue: catalogue, Records: records, Migration: migration, NewInstall: !existed}, nil
96}
97
98func New(store ports.Store) *Persister { return &Persister{store: store} }
99
100func cataloguePresent(path string) (bool, error) {
101 if _, err := os.Stat(path); err == nil {
102 return true, nil
103 } else if !errors.Is(err, os.ErrNotExist) {
104 return false, err
105 }
106 return false, nil
107}
108
109func openCurrentCatalogue(dir string, createProvenEmpty bool) (*Persister, []domain.CatalogueRecord, error) {
110 path := StorePath(dir)
111 present, err := cataloguePresent(path)
112 if err != nil {
113 return nil, nil, err
114 }
115 if !present && !createProvenEmpty {
116 return nil, nil, fmt.Errorf("%w: no catalogue candidates", errPersistenceUnavailable)
117 }
118 store, err := OpenStore(path)
119 if err != nil {
120 return nil, nil, err
121 }
122 p := New(store)
123 records, err := p.LoadCatalogue()
124 if err != nil {
125 return nil, nil, errors.Join(err, p.Close())
126 }
127 return p, records, nil
128}
129
130func (p *Persister) Save(record domain.CatalogueRecord) error { return p.Replace(record.Name, record) }
131func (p *Persister) Create(record domain.CatalogueRecord) error {
132 if p == nil {
133 return nil
134 }
135 if p.store == nil {
136 return errPersistenceUnavailable
137 }
138 p.mu.Lock()
139 defer p.mu.Unlock()
140 if err := p.terminalLocked(); err != nil {
141 return err
142 }
143 if _, exists := p.store.Get([]byte(record.Name)); exists {
144 return errors.New("persist: session already exists")
145 }
146 return p.applyLocked(map[string]*domain.CatalogueRecord{record.Name: &record}, true)
147}
148
149func (p *Persister) Records() ([]domain.CatalogueRecord, error) { return p.LoadCatalogue() }
150func (p *Persister) Record(name string) (domain.CatalogueRecord, bool, error) {
151 if p == nil {
152 return domain.CatalogueRecord{}, false, nil
153 }
154 if p.store == nil {
155 return domain.CatalogueRecord{}, false, errPersistenceUnavailable
156 }
157 p.mu.Lock()
158 defer p.mu.Unlock()
159 if err := p.terminalLocked(); err != nil {
160 return domain.CatalogueRecord{}, false, err
161 }
162 value, ok := p.store.Get([]byte(name))
163 if !ok {
164 return domain.CatalogueRecord{}, false, nil
165 }
166 record, err := decodeRecordValue(name, value)
167 if err != nil {
168 return domain.CatalogueRecord{}, false, err
169 }
170 return record, true, nil
171}
172func (p *Persister) Apply(records map[string]*domain.CatalogueRecord) error {
173 if p == nil {
174 return nil
175 }
176 if p.store == nil {
177 return errPersistenceUnavailable
178 }
179 p.mu.Lock()
180 defer p.mu.Unlock()
181 if err := p.terminalLocked(); err != nil {
182 return err
183 }
184 return p.applyLocked(records, true)
185}
186func (p *Persister) applyLocked(records map[string]*domain.CatalogueRecord, durable bool) error {
187 if len(records) == 0 {
188 return errors.New("persist: empty catalogue batch")
189 }
190 if err := p.ensureIncarnationIndexLocked(); err != nil {
191 return err
192 }
193 type mutation struct {
194 key []byte
195 value []byte
196 delete bool
197 }
198 changes := make([]mutation, 0, len(records))
199 keys := make([]string, 0, len(records))
200 for name := range records {
201 keys = append(keys, name)
202 }
203 sort.Strings(keys)
204 for _, name := range keys {
205 record := records[name]
206 if err := domain.ValidateSessionName(name); err != nil {
207 return err
208 }
209 if record == nil {
210 changes = append(changes, mutation{key: []byte(name), delete: true})
211 continue
212 }
213 if record.Name != name {
214 return errors.New("persist: catalogue key/name mismatch")
215 }
216 value, err := encodeRecordValue(*record)
217 if err != nil {
218 return err
219 }
220 changes = append(changes, mutation{key: []byte(name), value: value})
221 }
222 incoming := make(map[domain.IncarnationID]string, len(records))
223 for name, record := range records {
224 if record == nil {
225 continue
226 }
227 if previous, ok := incoming[record.IncarnationID]; ok && previous != name {
228 return fmt.Errorf("persist: duplicate incarnation for %q and %q", previous, name)
229 }
230 incoming[record.IncarnationID] = name
231 if owner, ok := p.incarnationOwners[record.IncarnationID]; ok && owner != name {
232 ownerNext, ownerTouched := records[owner]
233 if !ownerTouched || (ownerNext != nil && ownerNext.IncarnationID == record.IncarnationID) {
234 return fmt.Errorf("persist: duplicate incarnation for %q and %q", owner, name)
235 }
236 }
237 }
238 previous := make(map[string]storedValue, len(changes))
239 for _, change := range changes {
240 value, ok := p.store.Get(change.key)
241 previous[string(change.key)] = storedValue{value: value, exists: ok}
242 }
243 for _, change := range changes {
244 var err error
245 if change.delete {
246 err = p.store.Delete(change.key)
247 } else {
248 err = p.store.Set(change.key, change.value)
249 }
250 if err != nil {
251 return p.fenceLocked(errors.Join(err, p.restoreLocked(previous)))
252 }
253 }
254 if durable {
255 if err := p.store.Sync(); err != nil {
256 return p.fenceLocked(errors.Join(err, p.restoreLocked(previous)))
257 }
258 }
259 for name := range records {
260 if id, ok := p.nameIncarnations[name]; ok {
261 delete(p.incarnationOwners, id)
262 delete(p.nameIncarnations, name)
263 }
264 }
265 for name, record := range records {
266 if record != nil {
267 p.incarnationOwners[record.IncarnationID] = name
268 p.nameIncarnations[name] = record.IncarnationID
269 }
270 }
271 return nil
272}
273
274type storedValue struct {
275 value []byte
276 exists bool
277}
278
279// restoreLocked restores only keys affected by the rejected identity mutation.
280// It intentionally does not Sync: the Persister is fenced immediately after.
281func (p *Persister) restoreLocked(previous map[string]storedValue) error {
282 var err error
283 for key, old := range previous {
284 if old.exists {
285 err = errors.Join(err, p.store.Set([]byte(key), old.value))
286 } else {
287 err = errors.Join(err, p.store.Delete([]byte(key)))
288 }
289 }
290 return err
291}
292
293func (p *Persister) fenceLocked(cause error) error {
294 if p.terminalErr == nil {
295 p.terminalErr = fmt.Errorf("%w: %w", ErrCatalogueDurability, cause)
296 }
297 p.incarnationIndex = false
298 return p.terminalErr
299}
300
301func (p *Persister) terminalLocked() error { return p.terminalErr }
302
303func (p *Persister) ensureIncarnationIndexLocked() error {
304 if err := p.terminalLocked(); err != nil {
305 return err
306 }
307 if p.incarnationIndex {
308 return nil
309 }
310 owners := make(map[domain.IncarnationID]string)
311 names := make(map[string]domain.IncarnationID)
312 var indexErr error
313 p.store.Range(func(key, value []byte) bool {
314 name := string(key)
315 record, err := decodeRecordValue(name, value)
316 if err != nil {
317 indexErr = err
318 return false
319 }
320 if previous, ok := owners[record.IncarnationID]; ok {
321 indexErr = fmt.Errorf("persist: duplicate incarnation for %q and %q", previous, name)
322 return false
323 }
324 owners[record.IncarnationID] = name
325 names[name] = record.IncarnationID
326 return true
327 })
328 if indexErr != nil {
329 return indexErr
330 }
331 p.incarnationOwners = owners
332 p.nameIncarnations = names
333 p.incarnationIndex = true
334 return nil
335}
336func (p *Persister) Rename(oldName string, next domain.CatalogueRecord) error {
337 if p == nil {
338 return nil
339 }
340 if p.store == nil {
341 return errPersistenceUnavailable
342 }
343 if oldName == next.Name {
344 return errors.New("persist: rename requires distinct names")
345 }
346 return p.Apply(map[string]*domain.CatalogueRecord{oldName: nil, next.Name: &next})
347}
348func (p *Persister) Replace(name string, next domain.CatalogueRecord) error {
349 return p.Apply(map[string]*domain.CatalogueRecord{name: &next})
350}
351
352// UpdateMetadata buffers ordinary session metadata on the existing authoritative
353// incarnation. Recovery and transaction fields are retained from the stored
354// record rather than accepted from a potentially stale runtime copy. Sync or the
355// next identity operation makes the update durable.
356func (p *Persister) UpdateMetadata(update domain.CatalogueMetadataUpdate) error {
357 if p == nil {
358 return nil
359 }
360 if p.store == nil {
361 return errPersistenceUnavailable
362 }
363 p.mu.Lock()
364 defer p.mu.Unlock()
365 if err := p.terminalLocked(); err != nil {
366 return err
367 }
368 value, ok := p.store.Get([]byte(update.Name))
369 if !ok {
370 return errors.New("persist: session not found")
371 }
372 current, err := decodeRecordValue(update.Name, value)
373 if err != nil {
374 return err
375 }
376 if update.IncarnationID == (domain.IncarnationID{}) || current.IncarnationID != update.IncarnationID {
377 return errors.New("persist: session incarnation changed")
378 }
379 if update.Cwd != nil {
380 current.Cwd = *update.Cwd
381 }
382 if update.UpdatedAt != nil {
383 current.UpdatedAt = *update.UpdatedAt
384 }
385 if update.LastUsedSeq != nil {
386 current.LastUsedSeq = *update.LastUsedSeq
387 }
388 if update.TabNames != nil {
389 current.TabNames = append([]string(nil), (*update.TabNames)...)
390 }
391 if update.TabRecords != nil {
392 current.TabRecords = append([]domain.CatalogueTabRecord(nil), (*update.TabRecords)...)
393 }
394 return p.applyLocked(map[string]*domain.CatalogueRecord{current.Name: ¤t}, false)
395}
396
397// Sync makes every buffered catalogue mutation durable.
398func (p *Persister) Sync() error {
399 if p == nil {
400 return nil
401 }
402 if p.store == nil {
403 return errPersistenceUnavailable
404 }
405 p.mu.Lock()
406 defer p.mu.Unlock()
407 if err := p.terminalLocked(); err != nil {
408 return err
409 }
410 if err := p.store.Sync(); err != nil {
411 return p.fenceLocked(err)
412 }
413 return nil
414}
415
416func (p *Persister) Delete(name string) error {
417 return p.Apply(map[string]*domain.CatalogueRecord{name: nil})
418}
419func (p *Persister) LoadAll() ([]domain.CatalogueRecord, error) { return p.LoadCatalogue() }
420
421func (p *Persister) LoadCatalogue() ([]domain.CatalogueRecord, error) {
422 if p == nil {
423 return []domain.CatalogueRecord{}, nil
424 }
425 if p.store == nil {
426 return nil, errPersistenceUnavailable
427 }
428 p.mu.Lock()
429 defer p.mu.Unlock()
430 if err := p.terminalLocked(); err != nil {
431 return nil, err
432 }
433 data := make(map[string][]byte)
434 p.store.Range(func(key, value []byte) bool { data[string(key)] = append([]byte(nil), value...); return true })
435 return decodeAll(data)
436}
437func (p *Persister) Close() error {
438 if p == nil || p.store == nil {
439 return nil
440 }
441 p.mu.Lock()
442 defer p.mu.Unlock()
443 if err := p.terminalLocked(); err != nil {
444 if closeErr := p.store.CloseWithoutSync(); closeErr != nil {
445 return errors.Join(err, closeErr)
446 }
447 return err
448 }
449 return p.store.Close()
450}
451
452func decodeAll(data map[string][]byte) ([]domain.CatalogueRecord, error) {
453 records := make([]domain.CatalogueRecord, 0, len(data))
454 for name, value := range data {
455 record, err := decodeRecordValue(name, value)
456 if err != nil {
457 return nil, fmt.Errorf("persist: decode catalogue key %q: %w", name, err)
458 }
459 records = append(records, record)
460 }
461 sort.Slice(records, func(i, j int) bool { return records[i].Name < records[j].Name })
462 if err := validateUniqueIncarnations(records); err != nil {
463 return nil, err
464 }
465 return records, nil
466}
467func validateUniqueIncarnations(records []domain.CatalogueRecord) error {
468 seen := make(map[domain.IncarnationID]string, len(records))
469 for _, record := range records {
470 if err := record.Validate(); err != nil {
471 return err
472 }
473 if previous, ok := seen[record.IncarnationID]; ok {
474 return fmt.Errorf("persist: duplicate incarnation for %q and %q", previous, record.Name)
475 }
476 seen[record.IncarnationID] = record.Name
477 }
478 return nil
479}