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). Version 5 grants mr:force-merge to unrestricted tokens
43// holding all eleven previous scopes, preserving their owner rights. Version 6
44// adds releases and release:write; older builds cannot clean up their metadata
45// or preserve the owner rights of upgraded tokens (see upgradeReleaseTokens).
46// Version 7 adds the latest-release index; older writers would leave it stale.
47// Version 8 adds resolved issues and history (isev/, mrev/), and rewrites
48// merge-closed issues as resolved; older builds cannot decode state 4.
49const SchemaVersion = 8
50
51const schemaKey = "meta/schema"
52
53// ErrSchemaTooNew means the data was written by a newer ermite.
54var ErrSchemaTooNew = errors.New("storage schema is newer than this ermite supports")
55
56// Store is the ermite-kv backed metadata store.
57type Store struct {
58 db *kv.DB
59 locks *numberLocks // shared by the merge request and issue stores
60}
61
62var _ ports.RepoMetaStore = (*Store)(nil)
63
64// Open opens (or creates) the database in dir and checks the storage schema:
65// a missing version is written as SchemaVersion, a lower one is upgraded to it,
66// a higher one stops with ErrSchemaTooNew. dir must exist.
67func Open(ctx context.Context, dir string) (*Store, error) {
68 db, err := kv.Open(ctx, dir, kv.Options{})
69 if err != nil {
70 return nil, fmt.Errorf("open ermite-kv in %s: %w", dir, err)
71 }
72 s := &Store{db: db, locks: new(numberLocks)}
73 if err := s.checkSchema(ctx); err != nil {
74 _ = db.Close(context.WithoutCancel(ctx))
75 return nil, err
76 }
77 // Part of the schema 3 upgrade: the index of the open merge requests.
78 if err := s.BuildOpenIndex(ctx); err != nil {
79 _ = db.Close(context.WithoutCancel(ctx))
80 return nil, err
81 }
82 if err := s.buildReleaseLatestIndex(ctx); err != nil {
83 _ = db.Close(context.WithoutCancel(ctx))
84 return nil, err
85 }
86 return s, nil
87}
88
89// Close stops the database after the writes in progress.
90func (s *Store) Close(ctx context.Context) error {
91 if err := s.db.Close(ctx); err != nil {
92 return fmt.Errorf("close ermite-kv: %w", err)
93 }
94 return nil
95}
96
97// Check reports whether the database accepts writes. It returns the failure
98// once ermite-kv is closed or failed, and writes nothing.
99//
100// ermite-kv has no state accessor, and an empty batch sent through Apply would
101// queue behind the writes in progress. Apply looks at the state of the database
102// before it looks at its context, though, and an ended context stops it before
103// the writer channel: Apply on a cancelled context therefore answers ErrClosed
104// or ErrFailed when the database is stopped, and context.Canceled when it is
105// healthy, without waiting for the writer.
106func (s *Store) Check(ctx context.Context) error {
107 if err := ctx.Err(); err != nil {
108 return err
109 }
110 probe, cancel := context.WithCancel(context.WithoutCancel(ctx))
111 cancel()
112 var b kv.Batch
113 if _, err := s.db.Apply(probe, &b); err != nil && !errors.Is(err, context.Canceled) {
114 return fmt.Errorf("ermite-kv: %w", err)
115 }
116 return nil
117}
118
119func (s *Store) checkSchema(ctx context.Context) error {
120 log := zerowrap.FromCtx(ctx)
121 key := []byte(schemaKey)
122 for range 5 {
123 item, ok := s.db.Get(ctx, key)
124 if !ok {
125 var b kv.Batch
126 b.RequireAbsent(key)
127 b.Put(key, mustMarshal(&storagepb.SchemaVersion{Version: SchemaVersion}))
128 _, err := s.db.Apply(ctx, &b)
129 if errors.Is(err, kv.ErrConflict) {
130 continue // written meanwhile: read it
131 }
132 if err != nil {
133 return fmt.Errorf("write storage schema: %w", err)
134 }
135 log.Info().Uint32("schema", SchemaVersion).Msg("storage schema initialized")
136 return nil
137 }
138 var sv storagepb.SchemaVersion
139 if err := proto.Unmarshal(item.Value, &sv); err != nil {
140 return fmt.Errorf("decode storage schema: %w", err)
141 }
142 switch {
143 case sv.GetVersion() > SchemaVersion:
144 return fmt.Errorf("%w: found %d, supported %d", ErrSchemaTooNew, sv.GetVersion(), SchemaVersion)
145 case sv.GetVersion() == 0:
146 return fmt.Errorf("decode storage schema: invalid version 0")
147 }
148 if sv.GetVersion() < SchemaVersion {
149 // Older data stays readable as it is (the changes are additive);
150 // the new version is recorded so that older builds refuse it.
151 // The tokens are upgraded in the same batch, so that a crash
152 // leaves the old version and the upgrade runs again.
153 target := uint32(SchemaVersion)
154 switch {
155 case sv.GetVersion() < 4:
156 target = 4
157 case sv.GetVersion() == 4:
158 target = 5
159 }
160 var b kv.Batch
161 b.RequireVersion(key, item.Version)
162 if sv.GetVersion() < 4 {
163 if err := s.upgradeTokens(ctx, &b); err != nil {
164 return err
165 }
166 }
167 if sv.GetVersion() == 4 {
168 if err := s.upgradeForceMergeTokens(ctx, &b); err != nil {
169 return err
170 }
171 }
172 if sv.GetVersion() == 5 {
173 if err := s.upgradeReleaseTokens(ctx, &b); err != nil {
174 return err
175 }
176 }
177 if target == SchemaVersion {
178 if err := s.upgradeResolvedIssues(ctx, &b); err != nil {
179 return err
180 }
181 }
182 b.Put(key, mustMarshal(&storagepb.SchemaVersion{Version: target}))
183 _, err := s.db.Apply(ctx, &b)
184 if errors.Is(err, kv.ErrConflict) {
185 continue // changed meanwhile: read it again
186 }
187 if err != nil {
188 return fmt.Errorf("upgrade storage schema: %w", err)
189 }
190 log.Info().Uint32("from", sv.GetVersion()).Uint32("schema", target).Msg("storage schema upgraded")
191 if target < SchemaVersion {
192 continue
193 }
194 return nil
195 }
196 log.Info().Uint32("schema", sv.GetVersion()).Msg("storage schema checked")
197 return nil
198 }
199 return errors.New("write storage schema: concurrent initialization did not settle")
200}
201
202func mustMarshal(m proto.Message) []byte {
203 b, err := proto.MarshalOptions{Deterministic: true}.Marshal(m)
204 if err != nil {
205 panic(fmt.Sprintf("kvstore: marshal %T: %v", m, err)) // only for invalid UTF-8 strings, checked by callers
206 }
207 return b
208}
209
210// preIssueScopes are the nine original scopes, before mr:review and issue:write.
211var preIssueScopes = []domain.Scope{
212 domain.ScopeRepoRead, domain.ScopeRepoWrite, domain.ScopeRepoAdmin,
213 domain.ScopeMRWrite, domain.ScopeMRMerge,
214 domain.ScopeCIWrite, domain.ScopeSecretsWrite, domain.ScopeRunnerRegister,
215 domain.ScopeTokenAdmin,
216}
217
218// upgradeTokens adds to b the schema 4 upgrade of the tokens: a token that
219// holds all of preIssueScopes, limited to no repository and no branch prefix,
220// had the owner rights, which schema 4 needed mr:review and issue:write
221// for, so it gets each of them if missing (this is no escalation). Every other
222// token is left as it is. The batch requires each token to be at the version it
223// was read at.
224func (s *Store) upgradeTokens(ctx context.Context, b *kv.Batch) error {
225 return s.upgradeTokenScopes(ctx, b, preIssueScopes, []domain.Scope{domain.ScopeMRReview, domain.ScopeIssueWrite})
226}
227
228// preForceMergeScopes are the eleven scopes of schema 4, before mr:force-merge.
229var preForceMergeScopes = append(slices.Clone(preIssueScopes), domain.ScopeMRReview, domain.ScopeIssueWrite)
230
231// upgradeForceMergeTokens adds to b the schema 5 upgrade of the tokens: a
232// token that holds all of preForceMergeScopes, limited to no repository and no
233// branch prefix, had the owner rights, which AllScopes also needs
234// mr:force-merge for, so it gets it. Every other token is left as it is.
235func (s *Store) upgradeForceMergeTokens(ctx context.Context, b *kv.Batch) error {
236 return s.upgradeTokenScopes(ctx, b, preForceMergeScopes, []domain.Scope{domain.ScopeMRForceMerge})
237}
238
239// preReleaseScopes are the twelve scopes of schema 5, before release:write.
240var preReleaseScopes = append(slices.Clone(preForceMergeScopes), domain.ScopeMRForceMerge)
241
242// upgradeReleaseTokens adds to b the schema 6 upgrade of the tokens: a token
243// that holds all of preReleaseScopes, limited to no repository and no branch
244// prefix, had the owner rights, which AllScopes also needs release:write for,
245// so it gets it. Restricted or partial tokens are left as they are.
246func (s *Store) upgradeReleaseTokens(ctx context.Context, b *kv.Batch) error {
247 return s.upgradeTokenScopes(ctx, b, preReleaseScopes, []domain.Scope{domain.ScopeReleaseWrite})
248}
249
250func (s *Store) upgradeTokenScopes(ctx context.Context, b *kv.Batch, required, addedScopes []domain.Scope) error {
251 for key, item := range s.db.Snapshot(ctx).Prefix(ctx, []byte(tokenPrefix)) {
252 pb := &storagepb.Token{}
253 if err := proto.Unmarshal(item.Value, pb); err != nil {
254 return fmt.Errorf("upgrade token %s: %w", key, err)
255 }
256 if len(pb.GetRepos()) > 0 || len(pb.GetBranchPrefixes()) > 0 {
257 continue
258 }
259 full := true
260 for _, sc := range required {
261 if !slices.Contains(pb.GetScopes(), sc.String()) {
262 full = false
263 break
264 }
265 }
266 if !full {
267 continue
268 }
269 added := false
270 for _, sc := range addedScopes {
271 if !slices.Contains(pb.GetScopes(), sc.String()) {
272 pb.Scopes = append(pb.Scopes, sc.String())
273 added = true
274 }
275 }
276 if !added {
277 continue
278 }
279 b.RequireVersion(slices.Clone(key), item.Version)
280 b.Put(slices.Clone(key), mustMarshal(pb))
281 }
282 return nil
283}
284
285// upgradeResolvedIssues runs in the schema batch, without retroactive events or
286// timestamp/index changes. Typical records are <1 KiB; the 16 MiB batch limit
287// accommodates over 10k issues, consistent with the unchunked token upgrades.
288func (s *Store) upgradeResolvedIssues(ctx context.Context, b *kv.Batch) error {
289 count := 0
290 for key, item := range s.db.Snapshot(ctx).Prefix(ctx, []byte(issuePrefix)) {
291 if err := ctx.Err(); err != nil {
292 return err
293 }
294 var pb storagepb.Issue
295 if err := proto.Unmarshal(item.Value, &pb); err != nil {
296 return fmt.Errorf("upgrade issue %s: %w", key, err)
297 }
298 if pb.GetState() != storagepb.IssueState_ISSUE_STATE_CLOSED || pb.GetClosedByMergeRequest() == 0 {
299 continue
300 }
301 pb.State = storagepb.IssueState_ISSUE_STATE_RESOLVED
302 b.RequireVersion(key, item.Version)
303 b.Put(key, mustMarshal(&pb))
304 count++
305 }
306 log := zerowrap.FromCtx(ctx)
307 log.Info().Int("count", count).Msg("merge-closed issues upgraded to resolved")
308 return ctx.Err()
309}