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