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