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 "slices"
12
13 kv "bnema.dev/ermite-kv"
14 "github.com/bnema/zerowrap"
15 "google.golang.org/protobuf/proto"
16
17 storagepb "bnema.dev/ermite/gen/ermite/storage"
18 "bnema.dev/ermite/internal/domain"
19 "bnema.dev/ermite/internal/ports"
20)
21
22// SchemaVersion is the storage schema this build reads and writes. Version 2
23// adds branch prefixes to tokens: a build that does not know them would treat
24// a branch-limited token as unrestricted, so it must refuse the data
25// (ErrSchemaTooNew) instead. Version 3 adds the records of CI-3 that an older
26// build would misread: the call and scoped jobs of reusable workflows (it would
27// run the jobs of a called workflow as jobs of the caller's file, and plan the
28// call itself), the head commit of the runs of merge requests
29// (CIRun.head_sha: it would attach their checks to the merge commit, where
30// nothing looks for them), and the index of the open merge requests by source
31// branch (mropen/: an older build would not keep it up to date; the upgrade
32// builds it once, see BuildOpenIndex). Version 4 adds the issues (is/, isc/,
33// isopen/, numbered from the same mrseq/ sequence as the merge requests, and
34// RepoSettings.issues): an older build would not delete them with their
35// repository, nor keep the numbers apart, and it would misread the issue:write
36// scope, so it must refuse the data. It also adds the pending issues of the
37// public form (ispend/<repo>/<id>, removed with their repository), the bans of
38// their submitters (isban/<hex hash>) and the form's server key
39// (meta/issuekey). The upgrade also adds mr:review and issue:write (each only
40// if missing) to the stored tokens that held all nine original scopes with no
41// repository list and no branch prefix, so that they keep the owner rights (see
42// upgradeTokens).
43const SchemaVersion = 4
44
45const schemaKey = "meta/schema"
46
47// ErrSchemaTooNew means the data was written by a newer ermite.
48var ErrSchemaTooNew = errors.New("storage schema is newer than this ermite supports")
49
50// Store is the ermite-kv backed metadata store.
51type Store struct {
52 db *kv.DB
53 locks *numberLocks // shared by the merge request and issue stores
54}
55
56var _ ports.RepoMetaStore = (*Store)(nil)
57
58// Open opens (or creates) the database in dir and checks the storage schema:
59// a missing version is written as SchemaVersion, a lower one is upgraded to it,
60// a higher one stops with ErrSchemaTooNew. dir must exist.
61func Open(ctx context.Context, dir string) (*Store, error) {
62 db, err := kv.Open(ctx, dir, kv.Options{})
63 if err != nil {
64 return nil, fmt.Errorf("open ermite-kv in %s: %w", dir, err)
65 }
66 s := &Store{db: db, locks: new(numberLocks)}
67 if err := s.checkSchema(ctx); err != nil {
68 _ = db.Close(context.WithoutCancel(ctx))
69 return nil, err
70 }
71 // Part of the schema 3 upgrade: the index of the open merge requests.
72 if err := s.BuildOpenIndex(ctx); err != nil {
73 _ = db.Close(context.WithoutCancel(ctx))
74 return nil, err
75 }
76 return s, nil
77}
78
79// Close stops the database after the writes in progress.
80func (s *Store) Close(ctx context.Context) error {
81 if err := s.db.Close(ctx); err != nil {
82 return fmt.Errorf("close ermite-kv: %w", err)
83 }
84 return nil
85}
86
87// Check reports whether the database accepts writes. It returns the failure
88// once ermite-kv is closed or failed, and writes nothing.
89//
90// ermite-kv has no state accessor, and an empty batch sent through Apply would
91// queue behind the writes in progress. Apply looks at the state of the database
92// before it looks at its context, though, and an ended context stops it before
93// the writer channel: Apply on a cancelled context therefore answers ErrClosed
94// or ErrFailed when the database is stopped, and context.Canceled when it is
95// healthy, without waiting for the writer.
96func (s *Store) Check(ctx context.Context) error {
97 if err := ctx.Err(); err != nil {
98 return err
99 }
100 probe, cancel := context.WithCancel(context.WithoutCancel(ctx))
101 cancel()
102 var b kv.Batch
103 if _, err := s.db.Apply(probe, &b); err != nil && !errors.Is(err, context.Canceled) {
104 return fmt.Errorf("ermite-kv: %w", err)
105 }
106 return nil
107}
108
109func (s *Store) checkSchema(ctx context.Context) error {
110 log := zerowrap.FromCtx(ctx)
111 key := []byte(schemaKey)
112 for range 2 {
113 item, ok := s.db.Get(ctx, key)
114 if !ok {
115 var b kv.Batch
116 b.RequireAbsent(key)
117 b.Put(key, mustMarshal(&storagepb.SchemaVersion{Version: SchemaVersion}))
118 _, err := s.db.Apply(ctx, &b)
119 if errors.Is(err, kv.ErrConflict) {
120 continue // written meanwhile: read it
121 }
122 if err != nil {
123 return fmt.Errorf("write storage schema: %w", err)
124 }
125 log.Info().Uint32("schema", SchemaVersion).Msg("storage schema initialized")
126 return nil
127 }
128 var sv storagepb.SchemaVersion
129 if err := proto.Unmarshal(item.Value, &sv); err != nil {
130 return fmt.Errorf("decode storage schema: %w", err)
131 }
132 switch {
133 case sv.GetVersion() > SchemaVersion:
134 return fmt.Errorf("%w: found %d, supported %d", ErrSchemaTooNew, sv.GetVersion(), SchemaVersion)
135 case sv.GetVersion() == 0:
136 return fmt.Errorf("decode storage schema: invalid version 0")
137 }
138 if sv.GetVersion() < SchemaVersion {
139 // Older data stays readable as it is (the changes are additive);
140 // the new version is recorded so that older builds refuse it.
141 // The tokens are upgraded in the same batch, so that a crash
142 // leaves the old version and the upgrade runs again.
143 var b kv.Batch
144 b.RequireVersion(key, item.Version)
145 if sv.GetVersion() < 4 {
146 if err := s.upgradeTokens(ctx, &b); err != nil {
147 return err
148 }
149 }
150 b.Put(key, mustMarshal(&storagepb.SchemaVersion{Version: SchemaVersion}))
151 _, err := s.db.Apply(ctx, &b)
152 if errors.Is(err, kv.ErrConflict) {
153 continue // changed meanwhile: read it again
154 }
155 if err != nil {
156 return fmt.Errorf("upgrade storage schema: %w", err)
157 }
158 log.Info().Uint32("from", sv.GetVersion()).Uint32("schema", SchemaVersion).Msg("storage schema upgraded")
159 return nil
160 }
161 log.Info().Uint32("schema", sv.GetVersion()).Msg("storage schema checked")
162 return nil
163 }
164 return errors.New("write storage schema: concurrent initialization did not settle")
165}
166
167func mustMarshal(m proto.Message) []byte {
168 b, err := proto.MarshalOptions{Deterministic: true}.Marshal(m)
169 if err != nil {
170 panic(fmt.Sprintf("kvstore: marshal %T: %v", m, err)) // only for invalid UTF-8 strings, checked by callers
171 }
172 return b
173}
174
175// preIssueScopes are the nine original scopes, before mr:review and issue:write.
176var preIssueScopes = []domain.Scope{
177 domain.ScopeRepoRead, domain.ScopeRepoWrite, domain.ScopeRepoAdmin,
178 domain.ScopeMRWrite, domain.ScopeMRMerge,
179 domain.ScopeCIWrite, domain.ScopeSecretsWrite, domain.ScopeRunnerRegister,
180 domain.ScopeTokenAdmin,
181}
182
183// upgradeTokens adds to b the schema 4 upgrade of the tokens: a token that
184// holds all of preIssueScopes, limited to no repository and no branch prefix,
185// had the owner rights, which AllScopes now needs mr:review and issue:write
186// for, so it gets each of them if missing (this is no escalation). Every other
187// token is left as it is. The batch requires each token to be at the version it
188// was read at.
189func (s *Store) upgradeTokens(ctx context.Context, b *kv.Batch) error {
190 for key, item := range s.db.Snapshot(ctx).Prefix(ctx, []byte(tokenPrefix)) {
191 pb := &storagepb.Token{}
192 if err := proto.Unmarshal(item.Value, pb); err != nil {
193 return fmt.Errorf("upgrade token %s: %w", key, err)
194 }
195 if len(pb.GetRepos()) > 0 || len(pb.GetBranchPrefixes()) > 0 {
196 continue
197 }
198 full := true
199 for _, sc := range preIssueScopes {
200 if !slices.Contains(pb.GetScopes(), sc.String()) {
201 full = false
202 break
203 }
204 }
205 if !full {
206 continue
207 }
208 added := false
209 for _, sc := range []domain.Scope{domain.ScopeMRReview, domain.ScopeIssueWrite} {
210 if !slices.Contains(pb.GetScopes(), sc.String()) {
211 pb.Scopes = append(pb.Scopes, sc.String())
212 added = true
213 }
214 }
215 if !added {
216 continue
217 }
218 b.RequireVersion(slices.Clone(key), item.Version)
219 b.Put(slices.Clone(key), mustMarshal(pb))
220 }
221 return nil
222}