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 IncompatibleRecords []domain.CatalogueRecord
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 incompatible, err := catalogue.loadIncompatibleRecords()
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, IncompatibleRecords: incompatible, 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 LoadReadOnly(dir string) ([]domain.CatalogueRecord, error) { return LoadCatalogueReadOnly(dir) }
131
132// LoadCatalogueReadOnly loads the catalogue without retaining a store handle.
133// A fresh install with no main catalogue is not an error. Stray temporary
134// files are ignored, while malformed main files fail closed.
135func LoadCatalogueReadOnly(dir string) ([]domain.CatalogueRecord, error) {
136 p, records, err := openCurrentCatalogue(dir, false)
137 if err != nil {
138 if errors.Is(err, errPersistenceUnavailable) {
139 return []domain.CatalogueRecord{}, nil
140 }
141 return nil, fmt.Errorf("%w: %s: %w", ErrCatalogueUnreadable, StorePath(dir), err)
142 }
143 if err := p.Close(); err != nil {
144 return nil, err
145 }
146 return records, nil
147}
148
149func (p *Persister) Save(record domain.CatalogueRecord) error { return p.Replace(record.Name, record) }
150func (p *Persister) Create(record domain.CatalogueRecord) error {
151 if p == nil {
152 return nil
153 }
154 if p.store == nil {
155 return errPersistenceUnavailable
156 }
157 p.mu.Lock()
158 defer p.mu.Unlock()
159 if err := p.terminalLocked(); err != nil {
160 return err
161 }
162 if _, exists := p.store.Get([]byte(record.Name)); exists {
163 return errors.New("persist: session already exists")
164 }
165 return p.applyLocked(map[string]*domain.CatalogueRecord{record.Name: &record}, true)
166}
167
168func (p *Persister) Records() ([]domain.CatalogueRecord, error) { return p.LoadCatalogue() }
169func (p *Persister) Record(name string) (domain.CatalogueRecord, bool, error) {
170 if p == nil {
171 return domain.CatalogueRecord{}, false, nil
172 }
173 if p.store == nil {
174 return domain.CatalogueRecord{}, false, errPersistenceUnavailable
175 }
176 p.mu.Lock()
177 defer p.mu.Unlock()
178 if err := p.terminalLocked(); err != nil {
179 return domain.CatalogueRecord{}, false, err
180 }
181 value, ok := p.store.Get([]byte(name))
182 if !ok {
183 return domain.CatalogueRecord{}, false, nil
184 }
185 record, err := decodeRecordValue(name, value)
186 if err != nil {
187 return domain.CatalogueRecord{}, false, err
188 }
189 return record, true, nil
190}
191func (p *Persister) Apply(records map[string]*domain.CatalogueRecord) error {
192 if p == nil {
193 return nil
194 }
195 if p.store == nil {
196 return errPersistenceUnavailable
197 }
198 p.mu.Lock()
199 defer p.mu.Unlock()
200 if err := p.terminalLocked(); err != nil {
201 return err
202 }
203 return p.applyLocked(records, true)
204}
205func (p *Persister) applyLocked(records map[string]*domain.CatalogueRecord, durable bool) error {
206 if len(records) == 0 {
207 return errors.New("persist: empty catalogue batch")
208 }
209 if err := p.ensureIncarnationIndexLocked(); err != nil {
210 return err
211 }
212 type mutation struct {
213 key []byte
214 value []byte
215 delete bool
216 }
217 changes := make([]mutation, 0, len(records))
218 keys := make([]string, 0, len(records))
219 for name := range records {
220 keys = append(keys, name)
221 }
222 sort.Strings(keys)
223 for _, name := range keys {
224 record := records[name]
225 if err := domain.ValidateSessionName(name); err != nil {
226 return err
227 }
228 if record == nil {
229 changes = append(changes, mutation{key: []byte(name), delete: true})
230 continue
231 }
232 if record.Name != name {
233 return errors.New("persist: catalogue key/name mismatch")
234 }
235 value, err := encodeRecordValue(*record)
236 if err != nil {
237 return err
238 }
239 changes = append(changes, mutation{key: []byte(name), value: value})
240 }
241 incoming := make(map[domain.IncarnationID]string, len(records))
242 for name, record := range records {
243 if record == nil {
244 continue
245 }
246 if previous, ok := incoming[record.IncarnationID]; ok && previous != name {
247 return fmt.Errorf("persist: duplicate incarnation for %q and %q", previous, name)
248 }
249 incoming[record.IncarnationID] = name
250 if owner, ok := p.incarnationOwners[record.IncarnationID]; ok && owner != name {
251 ownerNext, ownerTouched := records[owner]
252 if !ownerTouched || (ownerNext != nil && ownerNext.IncarnationID == record.IncarnationID) {
253 return fmt.Errorf("persist: duplicate incarnation for %q and %q", owner, name)
254 }
255 }
256 }
257 previous := make(map[string]storedValue, len(changes))
258 for _, change := range changes {
259 value, ok := p.store.Get(change.key)
260 previous[string(change.key)] = storedValue{value: value, exists: ok}
261 }
262 for _, change := range changes {
263 var err error
264 if change.delete {
265 err = p.store.Delete(change.key)
266 } else {
267 err = p.store.Set(change.key, change.value)
268 }
269 if err != nil {
270 return p.fenceLocked(errors.Join(err, p.restoreLocked(previous)))
271 }
272 }
273 if durable {
274 if err := p.store.Sync(); err != nil {
275 return p.fenceLocked(errors.Join(err, p.restoreLocked(previous)))
276 }
277 }
278 for name := range records {
279 if id, ok := p.nameIncarnations[name]; ok {
280 delete(p.incarnationOwners, id)
281 delete(p.nameIncarnations, name)
282 }
283 }
284 for name, record := range records {
285 if record != nil {
286 p.incarnationOwners[record.IncarnationID] = name
287 p.nameIncarnations[name] = record.IncarnationID
288 }
289 }
290 return nil
291}
292
293type storedValue struct {
294 value []byte
295 exists bool
296}
297
298// restoreLocked restores only keys affected by the rejected identity mutation.
299// It intentionally does not Sync: the Persister is fenced immediately after.
300func (p *Persister) restoreLocked(previous map[string]storedValue) error {
301 var err error
302 for key, old := range previous {
303 if old.exists {
304 err = errors.Join(err, p.store.Set([]byte(key), old.value))
305 } else {
306 err = errors.Join(err, p.store.Delete([]byte(key)))
307 }
308 }
309 return err
310}
311
312func (p *Persister) fenceLocked(cause error) error {
313 if p.terminalErr == nil {
314 p.terminalErr = fmt.Errorf("%w: %w", ErrCatalogueDurability, cause)
315 }
316 p.incarnationIndex = false
317 return p.terminalErr
318}
319
320func (p *Persister) terminalLocked() error { return p.terminalErr }
321
322func (p *Persister) ensureIncarnationIndexLocked() error {
323 if err := p.terminalLocked(); err != nil {
324 return err
325 }
326 if p.incarnationIndex {
327 return nil
328 }
329 owners := make(map[domain.IncarnationID]string)
330 names := make(map[string]domain.IncarnationID)
331 var indexErr error
332 p.store.Range(func(key, value []byte) bool {
333 name := string(key)
334 record, err := decodeRecordValue(name, value)
335 if err != nil {
336 indexErr = err
337 return false
338 }
339 if previous, ok := owners[record.IncarnationID]; ok {
340 indexErr = fmt.Errorf("persist: duplicate incarnation for %q and %q", previous, name)
341 return false
342 }
343 owners[record.IncarnationID] = name
344 names[name] = record.IncarnationID
345 return true
346 })
347 if indexErr != nil {
348 return indexErr
349 }
350 p.incarnationOwners = owners
351 p.nameIncarnations = names
352 p.incarnationIndex = true
353 return nil
354}
355func (p *Persister) Rename(oldName string, next domain.CatalogueRecord) error {
356 if p == nil {
357 return nil
358 }
359 if p.store == nil {
360 return errPersistenceUnavailable
361 }
362 if oldName == next.Name {
363 return errors.New("persist: rename requires distinct names")
364 }
365 return p.Apply(map[string]*domain.CatalogueRecord{oldName: nil, next.Name: &next})
366}
367func (p *Persister) Replace(name string, next domain.CatalogueRecord) error {
368 return p.Apply(map[string]*domain.CatalogueRecord{name: &next})
369}
370
371// UpdateMetadata buffers ordinary session metadata on the existing authoritative
372// incarnation. Recovery and transaction fields are retained from the stored
373// record rather than accepted from a potentially stale runtime copy. Sync or the
374// next identity operation makes the update durable.
375func (p *Persister) UpdateMetadata(update domain.CatalogueMetadataUpdate) error {
376 if p == nil {
377 return nil
378 }
379 if p.store == nil {
380 return errPersistenceUnavailable
381 }
382 p.mu.Lock()
383 defer p.mu.Unlock()
384 if err := p.terminalLocked(); err != nil {
385 return err
386 }
387 value, ok := p.store.Get([]byte(update.Name))
388 if !ok {
389 return errors.New("persist: session not found")
390 }
391 current, err := decodeRecordValue(update.Name, value)
392 if err != nil {
393 return err
394 }
395 if update.IncarnationID == (domain.IncarnationID{}) || current.IncarnationID != update.IncarnationID {
396 return errors.New("persist: session incarnation changed")
397 }
398 if update.Cwd != nil {
399 current.Cwd = *update.Cwd
400 }
401 if update.UpdatedAt != nil {
402 current.UpdatedAt = *update.UpdatedAt
403 }
404 if update.LastUsedSeq != nil {
405 current.LastUsedSeq = *update.LastUsedSeq
406 }
407 if update.TabNames != nil {
408 current.TabNames = append([]string(nil), (*update.TabNames)...)
409 }
410 if update.TabRecords != nil {
411 current.TabRecords = append([]domain.CatalogueTabRecord(nil), (*update.TabRecords)...)
412 }
413 return p.applyLocked(map[string]*domain.CatalogueRecord{current.Name: ¤t}, false)
414}
415
416// Sync makes every buffered catalogue mutation durable.
417func (p *Persister) Sync() error {
418 if p == nil {
419 return nil
420 }
421 if p.store == nil {
422 return errPersistenceUnavailable
423 }
424 p.mu.Lock()
425 defer p.mu.Unlock()
426 if err := p.terminalLocked(); err != nil {
427 return err
428 }
429 if err := p.store.Sync(); err != nil {
430 return p.fenceLocked(err)
431 }
432 return nil
433}
434
435func (p *Persister) Delete(name string) error {
436 return p.Apply(map[string]*domain.CatalogueRecord{name: nil})
437}
438func (p *Persister) LoadAll() ([]domain.CatalogueRecord, error) { return p.LoadCatalogue() }
439
440func (p *Persister) loadIncompatibleRecords() ([]domain.CatalogueRecord, error) {
441 if p == nil || p.store == nil {
442 return nil, errPersistenceUnavailable
443 }
444 p.mu.Lock()
445 defer p.mu.Unlock()
446 if err := p.terminalLocked(); err != nil {
447 return nil, err
448 }
449 var records []domain.CatalogueRecord
450 var decodeErr error
451 p.store.Range(func(key, value []byte) bool {
452 stored, err := decodeStoredRecordValue(string(key), value)
453 if err != nil {
454 decodeErr = err
455 return false
456 }
457 if stored.protocolVersion != ports.ProtocolVersion {
458 records = append(records, stored.record)
459 }
460 return true
461 })
462 sort.Slice(records, func(i, j int) bool { return records[i].Name < records[j].Name })
463 return records, decodeErr
464}
465
466func (p *Persister) LoadCatalogue() ([]domain.CatalogueRecord, error) {
467 if p == nil {
468 return []domain.CatalogueRecord{}, nil
469 }
470 if p.store == nil {
471 return nil, errPersistenceUnavailable
472 }
473 p.mu.Lock()
474 defer p.mu.Unlock()
475 if err := p.terminalLocked(); err != nil {
476 return nil, err
477 }
478 data := make(map[string][]byte)
479 p.store.Range(func(key, value []byte) bool { data[string(key)] = append([]byte(nil), value...); return true })
480 return decodeAll(data)
481}
482func (p *Persister) Close() error {
483 if p == nil || p.store == nil {
484 return nil
485 }
486 p.mu.Lock()
487 defer p.mu.Unlock()
488 if err := p.terminalLocked(); err != nil {
489 if closeErr := p.store.CloseWithoutSync(); closeErr != nil {
490 return errors.Join(err, closeErr)
491 }
492 return err
493 }
494 return p.store.Close()
495}
496
497func decodeAll(data map[string][]byte) ([]domain.CatalogueRecord, error) {
498 records := make([]domain.CatalogueRecord, 0, len(data))
499 for name, value := range data {
500 record, err := decodeRecordValue(name, value)
501 if err != nil {
502 return nil, fmt.Errorf("persist: decode catalogue key %q: %w", name, err)
503 }
504 records = append(records, record)
505 }
506 sort.Slice(records, func(i, j int) bool { return records[i].Name < records[j].Name })
507 if err := validateUniqueIncarnations(records); err != nil {
508 return nil, err
509 }
510 return records, nil
511}
512func validateUniqueIncarnations(records []domain.CatalogueRecord) error {
513 seen := make(map[domain.IncarnationID]string, len(records))
514 for _, record := range records {
515 if err := record.Validate(); err != nil {
516 return err
517 }
518 if previous, ok := seen[record.IncarnationID]; ok {
519 return fmt.Errorf("persist: duplicate incarnation for %q and %q", previous, record.Name)
520 }
521 seen[record.IncarnationID] = record.Name
522 }
523 return nil
524}