1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package kvstore
4
5import (
6 "context"
7 "errors"
8 "fmt"
9 "slices"
10 "unicode/utf8"
11
12 kv "bnema.dev/ermite-kv"
13 "google.golang.org/protobuf/proto"
14
15 storagepb "bnema.dev/ermite/gen/ermite/storage"
16 "bnema.dev/ermite/internal/domain"
17 "bnema.dev/ermite/internal/ports"
18)
19
20// Key layout of the merge requests. Repository names hold no '/', so a prefix
21// ending in "/" never matches another repository; numbers and comment ids are
22// zero-padded so that keys sort numerically.
23const (
24 mrPrefix = "mr/" // mr/<repo>/<number:010d>
25 commentPrefix = "mrc/" // mrc/<repo>/<number:010d>/<id:010d>
26 readPrefix = "mrread/" // mrread/<repo>/<number:010d>
27 seqPrefix = "mrseq/" // mrseq/<repo>
28 numberFormat = "%010d"
29 createTries = 10
30)
31
32// MergeRequests is the ermite-kv backed ports.MergeRequestStore. It shares the
33// database of the Store it comes from.
34type MergeRequests struct {
35 db *kv.DB
36}
37
38var _ ports.MergeRequestStore = (*MergeRequests)(nil)
39
40// MergeRequests returns the merge request store over the same database. It is
41// valid until the Store is closed.
42func (s *Store) MergeRequests() *MergeRequests { return &MergeRequests{db: s.db} }
43
44func mrRepoPrefix(repo domain.RepoName) string { return mrPrefix + repo.String() + "/" }
45func mrKey(repo domain.RepoName, n int64) []byte {
46 return []byte(mrRepoPrefix(repo) + fmt.Sprintf(numberFormat, n))
47}
48func commentsPrefix(repo domain.RepoName, n int64) string {
49 return commentPrefix + repo.String() + "/" + fmt.Sprintf(numberFormat, n) + "/"
50}
51func commentKey(repo domain.RepoName, n, id int64) []byte {
52 return []byte(commentsPrefix(repo, n) + fmt.Sprintf(numberFormat, id))
53}
54func readKey(repo domain.RepoName, n int64) []byte {
55 return []byte(readPrefix + repo.String() + "/" + fmt.Sprintf(numberFormat, n))
56}
57func seqKey(repo domain.RepoName) []byte { return []byte(seqPrefix + repo.String()) }
58
59// mergeRequestKeys lists every key of the snapshot that belongs to the
60// repository's merge requests.
61func mergeRequestKeys(ctx context.Context, snap *kv.Snapshot, repo domain.RepoName) [][]byte {
62 var keys [][]byte
63 for _, p := range []string{mrRepoPrefix(repo), commentPrefix + repo.String() + "/", readPrefix + repo.String() + "/"} {
64 for k := range snap.Prefix(ctx, []byte(p)) {
65 keys = append(keys, slices.Clone(k))
66 }
67 }
68 if _, ok := snap.Get(ctx, seqKey(repo)); ok {
69 keys = append(keys, seqKey(repo))
70 }
71 return keys
72}
73
74// guardRepo makes the batch depend on the repository: it requires the
75// repository key to be at the version just read and rewrites it, which gives it
76// a new version. Every write that adds a key under the repository's merge
77// requests (a request, a comment, a read position) goes through it, so that
78//
79// - such a write fails with domain.ErrNotFound once the repository is deleted,
80// and leaves no key that a repository created again under the name would
81// inherit;
82// - Store.Delete, which requires the version of the repository key it listed
83// the keys at, notices a write that got in after the listing and lists again.
84//
85// It returns domain.ErrNotFound when the repository does not exist.
86func (m *MergeRequests) guardRepo(ctx context.Context, b *kv.Batch, repo domain.RepoName) error {
87 item, ok := m.db.Get(ctx, repoKey(repo))
88 if !ok {
89 return domain.ErrNotFound
90 }
91 b.RequireVersion(repoKey(repo), item.Version)
92 b.Put(repoKey(repo), item.Value)
93 return nil
94}
95
96// Create takes the next number of the repository from mrseq/<repo> and stores
97// the merge request under it, in one batch that requires the sequence to be
98// unchanged since it was read: two concurrent creations never share a number.
99// The batch also requires the repository to be unchanged since it was read and
100// rewrites it (see guardRepo), so that a repository deleted meanwhile is not
101// left with a stray merge request (a push recording pushed_at makes it retry). A missing repository gives
102// domain.ErrNotFound.
103//
104//hexcheck:ignore suspicious-business-logic-in-adapter takes the next number under a version precondition: a storage concern, no policy
105func (m *MergeRequests) Create(ctx context.Context, mr domain.MergeRequest) (domain.MergeRequest, error) {
106 for range createTries {
107 repoItem, repoOK := m.db.Get(ctx, repoKey(mr.Repo))
108 if !repoOK {
109 return domain.MergeRequest{}, domain.ErrNotFound
110 }
111 var last int64
112 item, ok := m.db.Get(ctx, seqKey(mr.Repo))
113 if ok {
114 var seq storagepb.Sequence
115 if err := proto.Unmarshal(item.Value, &seq); err != nil {
116 return domain.MergeRequest{}, fmt.Errorf("decode merge request sequence of %q: %w", mr.Repo, err)
117 }
118 last = seq.GetLast()
119 }
120 mr.Number = last + 1
121 val, err := encodeMergeRequest(mr)
122 if err != nil {
123 return domain.MergeRequest{}, err
124 }
125 var b kv.Batch
126 b.RequireVersion(repoKey(mr.Repo), repoItem.Version)
127 b.Put(repoKey(mr.Repo), repoItem.Value) // see guardRepo
128 if ok {
129 b.RequireVersion(seqKey(mr.Repo), item.Version)
130 } else {
131 b.RequireAbsent(seqKey(mr.Repo))
132 }
133 b.Put(seqKey(mr.Repo), mustMarshal(&storagepb.Sequence{Last: mr.Number}))
134 b.RequireAbsent(mrKey(mr.Repo, mr.Number))
135 b.Put(mrKey(mr.Repo, mr.Number), val)
136 _, err = m.db.Apply(ctx, &b)
137 if errors.Is(err, kv.ErrConflict) {
138 continue
139 }
140 if err != nil {
141 return domain.MergeRequest{}, fmt.Errorf("create merge request of %q: %w", mr.Repo, err)
142 }
143 return m.Get(ctx, mr.Repo, mr.Number)
144 }
145 return domain.MergeRequest{}, fmt.Errorf("create merge request of %q: %w", mr.Repo, domain.ErrConflict)
146}
147
148// Get reads mr/<repo>/<number>.
149func (m *MergeRequests) Get(ctx context.Context, repo domain.RepoName, number int64) (domain.MergeRequest, error) {
150 if number < 1 {
151 return domain.MergeRequest{}, domain.ErrNotFound
152 }
153 item, ok := m.db.Get(ctx, mrKey(repo, number))
154 if !ok {
155 return domain.MergeRequest{}, domain.ErrNotFound
156 }
157 return decodeMergeRequest(repo, item)
158}
159
160// List returns the merge requests of the repository, highest number first.
161func (m *MergeRequests) List(ctx context.Context, repo domain.RepoName) ([]domain.MergeRequest, error) {
162 var out []domain.MergeRequest
163 for key, item := range m.db.Snapshot(ctx).Prefix(ctx, []byte(mrRepoPrefix(repo))) {
164 mr, err := decodeMergeRequest(repo, item)
165 if err != nil {
166 return nil, fmt.Errorf("decode %s: %w", key, err)
167 }
168 out = append(out, mr)
169 }
170 slices.Reverse(out)
171 return out, nil
172}
173
174// Update replaces the merge request with a version precondition.
175//
176//hexcheck:ignore suspicious-business-logic-in-adapter port method name only
177func (m *MergeRequests) Update(ctx context.Context, mr domain.MergeRequest) error {
178 val, err := encodeMergeRequest(mr)
179 if err != nil {
180 return err
181 }
182 key := mrKey(mr.Repo, mr.Number)
183 if _, ok := m.db.Get(ctx, key); !ok {
184 return domain.ErrNotFound
185 }
186 if mr.Version == 0 {
187 return fmt.Errorf("update merge request %d of %q: read it first: %w", mr.Number, mr.Repo, domain.ErrConflict)
188 }
189 var b kv.Batch
190 b.RequireVersion(key, mr.Version)
191 b.Put(key, val)
192 if _, err := m.db.Apply(ctx, &b); err != nil {
193 if errors.Is(err, kv.ErrConflict) {
194 return domain.ErrConflict
195 }
196 return fmt.Errorf("update merge request %d of %q: %w", mr.Number, mr.Repo, err)
197 }
198 return nil
199}
200
201// ListComments returns the comments of the merge request, oldest first.
202func (m *MergeRequests) ListComments(ctx context.Context, repo domain.RepoName, number int64) ([]domain.Comment, error) {
203 var out []domain.Comment
204 for key, item := range m.db.Snapshot(ctx).Prefix(ctx, []byte(commentsPrefix(repo, number))) {
205 c, err := decodeComment(item)
206 if err != nil {
207 return nil, fmt.Errorf("decode %s: %w", key, err)
208 }
209 out = append(out, c)
210 }
211 return out, nil
212}
213
214// GetComment reads one comment.
215func (m *MergeRequests) GetComment(ctx context.Context, repo domain.RepoName, number, id int64) (domain.Comment, error) {
216 if number < 1 || id < 1 {
217 return domain.Comment{}, domain.ErrNotFound
218 }
219 item, ok := m.db.Get(ctx, commentKey(repo, number, id))
220 if !ok {
221 return domain.Comment{}, domain.ErrNotFound
222 }
223 return decodeComment(item)
224}
225
226// SaveComment writes the comment and the merge request in one batch, each with
227// its version precondition (absent for a new comment), and requires the
228// repository to exist (see guardRepo): domain.ErrNotFound otherwise.
229func (m *MergeRequests) SaveComment(ctx context.Context, mr domain.MergeRequest, c domain.Comment) error {
230 mrVal, err := encodeMergeRequest(mr)
231 if err != nil {
232 return err
233 }
234 cVal, err := encodeComment(c)
235 if err != nil {
236 return err
237 }
238 if mr.Version == 0 {
239 return fmt.Errorf("save comment on %d of %q: read the merge request first: %w", mr.Number, mr.Repo, domain.ErrConflict)
240 }
241 var b kv.Batch
242 if err := m.guardRepo(ctx, &b, mr.Repo); err != nil {
243 return err
244 }
245 b.RequireVersion(mrKey(mr.Repo, mr.Number), mr.Version)
246 b.Put(mrKey(mr.Repo, mr.Number), mrVal)
247 if c.Version == 0 {
248 b.RequireAbsent(commentKey(mr.Repo, mr.Number, c.ID))
249 } else {
250 b.RequireVersion(commentKey(mr.Repo, mr.Number, c.ID), c.Version)
251 }
252 b.Put(commentKey(mr.Repo, mr.Number, c.ID), cVal)
253 if _, err := m.db.Apply(ctx, &b); err != nil {
254 if errors.Is(err, kv.ErrConflict) {
255 return domain.ErrConflict
256 }
257 return fmt.Errorf("save comment on %d of %q: %w", mr.Number, mr.Repo, err)
258 }
259 return nil
260}
261
262// GetRead reads mrread/<repo>/<number>: the id of a commit, or nothing.
263func (m *MergeRequests) GetRead(ctx context.Context, repo domain.RepoName, number int64) (domain.ObjectID, error) {
264 item, ok := m.db.Get(ctx, readKey(repo, number))
265 if !ok {
266 return domain.ObjectID{}, nil
267 }
268 id, err := domain.ParseObjectID(string(item.Value))
269 if err != nil {
270 return domain.ObjectID{}, fmt.Errorf("decode read position of %d in %q: %w", number, repo, err)
271 }
272 return id, nil
273}
274
275// SetRead writes mrread/<repo>/<number>. The repository must exist (see
276// guardRepo): domain.ErrNotFound otherwise.
277//
278//hexcheck:ignore suspicious-business-logic-in-adapter retries a write that lost a race with the repository key: a storage concern, no policy
279func (m *MergeRequests) SetRead(ctx context.Context, repo domain.RepoName, number int64, head domain.ObjectID) error {
280 if head.IsZero() {
281 return fmt.Errorf("%w: empty read position", domain.ErrInvalidObjectID)
282 }
283 for range createTries {
284 var b kv.Batch
285 if err := m.guardRepo(ctx, &b, repo); err != nil {
286 return err
287 }
288 b.Put(readKey(repo, number), []byte(head.String()))
289 _, err := m.db.Apply(ctx, &b)
290 if errors.Is(err, kv.ErrConflict) {
291 continue
292 }
293 if err != nil {
294 return fmt.Errorf("record read position of %d in %q: %w", number, repo, err)
295 }
296 return nil
297 }
298 return fmt.Errorf("record read position of %d in %q: %w", number, repo, domain.ErrConflict)
299}
300
301func encodeMergeRequest(mr domain.MergeRequest) ([]byte, error) {
302 if mr.Repo.IsZero() {
303 return nil, fmt.Errorf("%w: empty name", domain.ErrInvalidName)
304 }
305 if !utf8.ValidString(mr.Author) {
306 return nil, errors.New("merge request author is not valid UTF-8")
307 }
308 state, err := encodeState(mr.State)
309 if err != nil {
310 return nil, err
311 }
312 mode, err := encodeMode(mr.MergeMode)
313 if err != nil {
314 return nil, err
315 }
316 return mustMarshal(&storagepb.MergeRequest{
317 Number: mr.Number,
318 Title: mr.Title.String(),
319 Description: mr.Description.String(),
320 SourceBranch: mr.Source,
321 TargetBranch: mr.Target,
322 State: state,
323 Author: mr.Author,
324 CreatedAt: encodeTime(mr.CreatedAt),
325 UpdatedAt: encodeTime(mr.UpdatedAt),
326 HeadId: mr.HeadID.String(),
327 BaseId: mr.BaseID.String(),
328 MergedCommitId: mr.MergedCommitID.String(),
329 MergeMode: mode,
330 CommentSeq: mr.CommentSeq,
331 }), nil
332}
333
334//hexcheck:ignore suspicious-business-logic-in-adapter schema decoding, not a policy
335func decodeMergeRequest(repo domain.RepoName, item kv.Item) (domain.MergeRequest, error) {
336 var pb storagepb.MergeRequest
337 if err := proto.Unmarshal(item.Value, &pb); err != nil {
338 return domain.MergeRequest{}, fmt.Errorf("decode merge request: %w", err)
339 }
340 fail := func(err error) (domain.MergeRequest, error) {
341 return domain.MergeRequest{}, fmt.Errorf("decode merge request %d of %q: %w", pb.GetNumber(), repo, err)
342 }
343 title, err := domain.ParseMRTitle(pb.GetTitle())
344 if err != nil {
345 return fail(err)
346 }
347 desc, err := domain.ParsePlainText(pb.GetDescription())
348 if err != nil {
349 return fail(err)
350 }
351 state, err := decodeState(pb.GetState())
352 if err != nil {
353 return fail(err)
354 }
355 mr := domain.MergeRequest{
356 Repo: repo, Number: pb.GetNumber(), Title: title, Description: desc,
357 Source: pb.GetSourceBranch(), Target: pb.GetTargetBranch(), State: state,
358 Author: pb.GetAuthor(),
359 CreatedAt: decodeTime(pb.GetCreatedAt()), UpdatedAt: decodeTime(pb.GetUpdatedAt()),
360 MergeMode: decodeMode(pb.GetMergeMode()),
361 CommentSeq: pb.GetCommentSeq(),
362 Version: item.Version,
363 }
364 for _, f := range []struct {
365 dst *domain.ObjectID
366 src string
367 }{{&mr.HeadID, pb.GetHeadId()}, {&mr.BaseID, pb.GetBaseId()}, {&mr.MergedCommitID, pb.GetMergedCommitId()}} {
368 if f.src == "" {
369 continue
370 }
371 if *f.dst, err = domain.ParseObjectID(f.src); err != nil {
372 return fail(err)
373 }
374 }
375 return mr, nil
376}
377
378func encodeComment(c domain.Comment) ([]byte, error) {
379 side, err := encodeSide(c.Side)
380 if err != nil {
381 return nil, err
382 }
383 return mustMarshal(&storagepb.Comment{
384 Id: c.ID,
385 Author: c.Author,
386 Path: c.Path,
387 Line: int32(c.Line), //nolint:gosec // validated against the diff, far below 2^31
388 Side: side,
389 CommitId: c.CommitID.String(),
390 Text: c.Text.String(),
391 Resolved: c.Resolved,
392 CreatedAt: encodeTime(c.CreatedAt),
393 }), nil
394}
395
396//hexcheck:ignore suspicious-business-logic-in-adapter schema decoding, not a policy
397func decodeComment(item kv.Item) (domain.Comment, error) {
398 var pb storagepb.Comment
399 if err := proto.Unmarshal(item.Value, &pb); err != nil {
400 return domain.Comment{}, fmt.Errorf("decode comment: %w", err)
401 }
402 text, err := domain.ParsePlainText(pb.GetText())
403 if err != nil {
404 return domain.Comment{}, fmt.Errorf("decode comment %d: %w", pb.GetId(), err)
405 }
406 c := domain.Comment{
407 ID: pb.GetId(), Author: pb.GetAuthor(), Path: pb.GetPath(), Line: int(pb.GetLine()),
408 Side: decodeSide(pb.GetSide()), Text: text, Resolved: pb.GetResolved(),
409 CreatedAt: decodeTime(pb.GetCreatedAt()), Version: item.Version,
410 }
411 if pb.GetCommitId() != "" {
412 if c.CommitID, err = domain.ParseObjectID(pb.GetCommitId()); err != nil {
413 return domain.Comment{}, fmt.Errorf("decode comment %d: %w", c.ID, err)
414 }
415 }
416 return c, nil
417}
418
419func encodeState(s domain.MRState) (storagepb.MergeRequestState, error) {
420 switch s {
421 case domain.MRStateOpen:
422 return storagepb.MergeRequestState_MERGE_REQUEST_STATE_OPEN, nil
423 case domain.MRStateMerged:
424 return storagepb.MergeRequestState_MERGE_REQUEST_STATE_MERGED, nil
425 case domain.MRStateClosed:
426 return storagepb.MergeRequestState_MERGE_REQUEST_STATE_CLOSED, nil
427 }
428 return 0, fmt.Errorf("%w: merge request state %q", domain.ErrInvalid, s)
429}
430
431func decodeState(s storagepb.MergeRequestState) (domain.MRState, error) {
432 switch s {
433 case storagepb.MergeRequestState_MERGE_REQUEST_STATE_OPEN:
434 return domain.MRStateOpen, nil
435 case storagepb.MergeRequestState_MERGE_REQUEST_STATE_MERGED:
436 return domain.MRStateMerged, nil
437 case storagepb.MergeRequestState_MERGE_REQUEST_STATE_CLOSED:
438 return domain.MRStateClosed, nil
439 }
440 return "", fmt.Errorf("unknown merge request state %d", s)
441}
442
443func encodeMode(m domain.MergeMode) (storagepb.MergeMode, error) {
444 switch m {
445 case "":
446 return storagepb.MergeMode_MERGE_MODE_UNSPECIFIED, nil
447 case domain.MergeModeFastForward:
448 return storagepb.MergeMode_MERGE_MODE_FAST_FORWARD, nil
449 case domain.MergeModeSquash:
450 return storagepb.MergeMode_MERGE_MODE_SQUASH, nil
451 }
452 return 0, fmt.Errorf("%w: merge mode %q", domain.ErrInvalid, m)
453}
454
455// decodeMode maps an unknown mode to none: it only labels a merged request.
456func decodeMode(m storagepb.MergeMode) domain.MergeMode {
457 switch m {
458 case storagepb.MergeMode_MERGE_MODE_FAST_FORWARD:
459 return domain.MergeModeFastForward
460 case storagepb.MergeMode_MERGE_MODE_SQUASH:
461 return domain.MergeModeSquash
462 }
463 return ""
464}
465
466func encodeSide(s domain.DiffSide) (storagepb.DiffSide, error) {
467 switch s {
468 case domain.SideNone:
469 return storagepb.DiffSide_DIFF_SIDE_UNSPECIFIED, nil
470 case domain.SideOld:
471 return storagepb.DiffSide_DIFF_SIDE_OLD, nil
472 case domain.SideNew:
473 return storagepb.DiffSide_DIFF_SIDE_NEW, nil
474 }
475 return 0, fmt.Errorf("%w: diff side %q", domain.ErrInvalid, s)
476}
477
478func decodeSide(s storagepb.DiffSide) domain.DiffSide {
479 switch s {
480 case storagepb.DiffSide_DIFF_SIDE_OLD:
481 return domain.SideOld
482 case storagepb.DiffSide_DIFF_SIDE_NEW:
483 return domain.SideNew
484 }
485 return domain.SideNone
486}
487
488func encodeSettings(s domain.RepoSettings) (*storagepb.RepoSettings, error) {
489 pb := &storagepb.RepoSettings{FailingChecksBlock: s.FailingChecksBlock}
490 for _, m := range s.AllowedMergeModes {
491 mode, err := encodeMode(m)
492 if err != nil || mode == storagepb.MergeMode_MERGE_MODE_UNSPECIFIED {
493 return nil, fmt.Errorf("%w: merge mode %q in the settings", domain.ErrInvalid, m)
494 }
495 pb.AllowedMergeModes = append(pb.AllowedMergeModes, mode)
496 }
497 return pb, nil
498}
499
500// decodeSettings returns the defaults for a repository stored before merge
501// requests existed, and for settings left with no mode that this build knows.
502//
503//hexcheck:ignore suspicious-business-logic-in-adapter schema decoding with the documented defaults
504func decodeSettings(pb *storagepb.RepoSettings) domain.RepoSettings {
505 if pb == nil {
506 return domain.DefaultRepoSettings()
507 }
508 var modes []domain.MergeMode
509 for _, m := range pb.GetAllowedMergeModes() {
510 if mode := decodeMode(m); mode != "" {
511 modes = append(modes, mode)
512 }
513 }
514 if len(modes) == 0 {
515 return domain.DefaultRepoSettings()
516 }
517 normalized, err := domain.NormalizeMergeModes(modes)
518 if err != nil {
519 return domain.DefaultRepoSettings()
520 }
521 return domain.RepoSettings{AllowedMergeModes: normalized, FailingChecksBlock: pb.GetFailingChecksBlock()}
522}