1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3// Package kvstore implements the metadata ports over ermite-kv. Values are
4// the protobuf messages of proto/ermite/storage.
5package kvstore
6
7import (
8 "context"
9 "errors"
10 "fmt"
11
12 kv "bnema.dev/ermite-kv"
13 "github.com/bnema/zerowrap"
14 "google.golang.org/protobuf/proto"
15
16 storagepb "bnema.dev/ermite/gen/ermite/storage"
17 "bnema.dev/ermite/internal/ports"
18)
19
20// SchemaVersion is the storage schema this build reads and writes.
21const SchemaVersion = 1
22
23const schemaKey = "meta/schema"
24
25// ErrSchemaTooNew means the data was written by a newer ermite.
26var ErrSchemaTooNew = errors.New("storage schema is newer than this ermite supports")
27
28// Store is the ermite-kv backed metadata store.
29type Store struct {
30 db *kv.DB
31}
32
33var _ ports.RepoMetaStore = (*Store)(nil)
34
35// Open opens (or creates) the database in dir and checks the storage schema:
36// a missing version is written as SchemaVersion, a higher one stops with
37// ErrSchemaTooNew. dir must exist.
38func Open(ctx context.Context, dir string) (*Store, error) {
39 db, err := kv.Open(ctx, dir, kv.Options{})
40 if err != nil {
41 return nil, fmt.Errorf("open ermite-kv in %s: %w", dir, err)
42 }
43 s := &Store{db: db}
44 if err := s.checkSchema(ctx); err != nil {
45 _ = db.Close(context.WithoutCancel(ctx))
46 return nil, err
47 }
48 return s, nil
49}
50
51// Close stops the database after the writes in progress.
52func (s *Store) Close(ctx context.Context) error {
53 if err := s.db.Close(ctx); err != nil {
54 return fmt.Errorf("close ermite-kv: %w", err)
55 }
56 return nil
57}
58
59// Check reports whether the database accepts writes. It returns the failure
60// once ermite-kv is closed or failed, and writes nothing.
61func (s *Store) Check(ctx context.Context) error {
62 var b kv.Batch // no ops: only the state of the database is checked
63 if _, err := s.db.Apply(ctx, &b); err != nil {
64 return fmt.Errorf("ermite-kv: %w", err)
65 }
66 return nil
67}
68
69func (s *Store) checkSchema(ctx context.Context) error {
70 log := zerowrap.FromCtx(ctx)
71 key := []byte(schemaKey)
72 for range 2 {
73 item, ok := s.db.Get(ctx, key)
74 if !ok {
75 var b kv.Batch
76 b.RequireAbsent(key)
77 b.Put(key, mustMarshal(&storagepb.SchemaVersion{Version: SchemaVersion}))
78 _, err := s.db.Apply(ctx, &b)
79 if errors.Is(err, kv.ErrConflict) {
80 continue // written meanwhile: read it
81 }
82 if err != nil {
83 return fmt.Errorf("write storage schema: %w", err)
84 }
85 log.Info().Uint32("schema", SchemaVersion).Msg("storage schema initialized")
86 return nil
87 }
88 var sv storagepb.SchemaVersion
89 if err := proto.Unmarshal(item.Value, &sv); err != nil {
90 return fmt.Errorf("decode storage schema: %w", err)
91 }
92 switch {
93 case sv.GetVersion() > SchemaVersion:
94 return fmt.Errorf("%w: found %d, supported %d", ErrSchemaTooNew, sv.GetVersion(), SchemaVersion)
95 case sv.GetVersion() == 0:
96 return fmt.Errorf("decode storage schema: invalid version 0")
97 }
98 log.Info().Uint32("schema", sv.GetVersion()).Msg("storage schema checked")
99 return nil
100 }
101 return errors.New("write storage schema: concurrent initialization did not settle")
102}
103
104func mustMarshal(m proto.Message) []byte {
105 b, err := proto.MarshalOptions{Deterministic: true}.Marshal(m)
106 if err != nil {
107 panic(fmt.Sprintf("kvstore: marshal %T: %v", m, err)) // only for invalid UTF-8 strings, checked by callers
108 }
109 return b
110}