1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package kvstore
4
5import (
6 "context"
7 "errors"
8 "fmt"
9 "net/url"
10 "slices"
11 "strconv"
12 "strings"
13 "unicode/utf8"
14
15 kv "bnema.dev/ermite-kv"
16 "github.com/bnema/zerowrap"
17 "google.golang.org/protobuf/proto"
18
19 storagepb "bnema.dev/ermite/gen/ermite/storage"
20 "bnema.dev/ermite/internal/domain"
21 "bnema.dev/ermite/internal/ports"
22)
23
24// Key layout of the merge requests. Repository names hold no '/', so a prefix
25// ending in "/" never matches another repository; numbers and comment ids are
26// zero-padded so that keys sort numerically.
27const (
28 mrPrefix = "mr/" // mr/<repo>/<number:010d>
29 commentPrefix = "mrc/" // mrc/<repo>/<number:010d>/<id:010d>
30 readPrefix = "mrread/" // mrread/<repo>/<number:010d>
31 seqPrefix = "mrseq/" // mrseq/<repo>
32 // The indexes of the open merge requests by source branch and by target
33 // branch, which a push looks up (the values are not read). They are written in
34 // the same batch as the merge request, so they are never out of step with it,
35 // and built once, together, for the data that predates them (BuildOpenIndex,
36 // mrOpenIndexMarker).
37 mrOpenPrefix = "mropen/" // mropen/<repo>/<escaped source>/<number:010d>
38 mrTargetPrefix = "mrtarget/" // mrtarget/<repo>/<escaped target>/<number:010d>
39 mrOpenIndexMarker = "meta/mropen" // present once the indexes cover every merge request
40 numberFormat = "%010d"
41 createTries = 10
42)
43
44// MergeRequests is the ermite-kv backed ports.MergeRequestStore. It shares the
45// database of the Store it comes from.
46type MergeRequests struct {
47 db *kv.DB
48}
49
50var _ ports.MergeRequestStore = (*MergeRequests)(nil)
51
52// MergeRequests returns the merge request store over the same database. It is
53// valid until the Store is closed.
54func (s *Store) MergeRequests() *MergeRequests { return &MergeRequests{db: s.db} }
55
56func mrRepoPrefix(repo domain.RepoName) string { return mrPrefix + repo.String() + "/" }
57func mrKey(repo domain.RepoName, n int64) []byte {
58 return []byte(mrRepoPrefix(repo) + fmt.Sprintf(numberFormat, n))
59}
60func commentsPrefix(repo domain.RepoName, n int64) string {
61 return commentPrefix + repo.String() + "/" + fmt.Sprintf(numberFormat, n) + "/"
62}
63func commentKey(repo domain.RepoName, n, id int64) []byte {
64 return []byte(commentsPrefix(repo, n) + fmt.Sprintf(numberFormat, id))
65}
66func readKey(repo domain.RepoName, n int64) []byte {
67 return []byte(readPrefix + repo.String() + "/" + fmt.Sprintf(numberFormat, n))
68}
69func seqKey(repo domain.RepoName) []byte { return []byte(seqPrefix + repo.String()) }
70
71// mrOpenRepoPrefix is where the open merge requests of a repository are indexed
72// by source branch.
73func mrOpenRepoPrefix(repo domain.RepoName) string { return mrOpenPrefix + repo.String() + "/" }
74
75// indexBranchPrefix is where the open merge requests of one branch are indexed
76// under prefix (mrOpenPrefix or mrTargetPrefix): the branch is escaped, so it
77// holds no '/' and the prefix matches that branch only.
78func indexBranchPrefix(prefix string, repo domain.RepoName, branch string) string {
79 return prefix + repo.String() + "/" + url.PathEscape(branch) + "/"
80}
81
82func indexKey(prefix string, repo domain.RepoName, branch string, n int64) []byte {
83 return []byte(indexBranchPrefix(prefix, repo, branch) + fmt.Sprintf(numberFormat, n))
84}
85
86// indexOpen adds to b the writes that keep the indexes of the open merge
87// requests (by source and by target) in step with a write of mr over old (nil
88// when old is not known, as for a new request or a comment: the keys of mr are
89// then put or deleted, whatever they were). Deleting a key that is not there is
90// not an error.
91func indexOpen(b *kv.Batch, old *domain.MergeRequest, mr domain.MergeRequest) {
92 if old != nil && old.IsOpen() {
93 if !mr.IsOpen() || old.Source != mr.Source {
94 b.Delete(indexKey(mrOpenPrefix, old.Repo, old.Source, old.Number))
95 }
96 if !mr.IsOpen() || old.Target != mr.Target {
97 b.Delete(indexKey(mrTargetPrefix, old.Repo, old.Target, old.Number))
98 }
99 }
100 switch {
101 case mr.IsOpen():
102 b.Put(indexKey(mrOpenPrefix, mr.Repo, mr.Source, mr.Number), []byte{1})
103 b.Put(indexKey(mrTargetPrefix, mr.Repo, mr.Target, mr.Number), []byte{1})
104 case old == nil:
105 b.Delete(indexKey(mrOpenPrefix, mr.Repo, mr.Source, mr.Number))
106 b.Delete(indexKey(mrTargetPrefix, mr.Repo, mr.Target, mr.Number))
107 }
108}
109
110// BuildOpenIndex builds the indexes of the open merge requests (by source and
111// by target) when the data does not have them yet (it predates them): one pass
112// over the merge requests, in batches, then the marker. It runs when the store is opened and is idempotent: a pass
113// cut short by a crash is simply done again, writing the same keys.
114//
115//hexcheck:ignore suspicious-business-logic-in-adapter one-time storage backfill of an index: a storage concern, no policy
116func (s *Store) BuildOpenIndex(ctx context.Context) error {
117 if _, ok := s.db.Get(ctx, []byte(mrOpenIndexMarker)); ok {
118 return nil
119 }
120 var b kv.Batch
121 n := 0
122 flush := func() error {
123 if b.Len() == 0 {
124 return nil
125 }
126 if _, err := s.db.Apply(ctx, &b); err != nil {
127 return fmt.Errorf("build the index of open merge requests: %w", err)
128 }
129 b.Reset()
130 return nil
131 }
132 for key, item := range s.db.Snapshot(ctx).Prefix(ctx, []byte(mrPrefix)) {
133 repoName, _, ok := strings.Cut(strings.TrimPrefix(string(key), mrPrefix), "/")
134 if !ok {
135 continue
136 }
137 repo, err := domain.ParseRepoName(repoName)
138 if err != nil {
139 continue
140 }
141 mr, err := decodeMergeRequest(repo, item)
142 if err != nil {
143 return fmt.Errorf("build the index of open merge requests: decode %s: %w", key, err)
144 }
145 if !mr.IsOpen() {
146 continue
147 }
148 b.Put(indexKey(mrOpenPrefix, mr.Repo, mr.Source, mr.Number), []byte{1})
149 b.Put(indexKey(mrTargetPrefix, mr.Repo, mr.Target, mr.Number), []byte{1})
150 if n++; n%(deleteChunk/2) == 0 {
151 if err := flush(); err != nil {
152 return err
153 }
154 }
155 }
156 if err := flush(); err != nil {
157 return err
158 }
159 b.Put([]byte(mrOpenIndexMarker), []byte{1})
160 if _, err := s.db.Apply(ctx, &b); err != nil {
161 return fmt.Errorf("build the index of open merge requests: %w", err)
162 }
163 log := zerowrap.FromCtx(ctx)
164 log.Info().Int("open", n).Msg("index of open merge requests built")
165 return nil
166}
167
168// mergeRequestKeys lists every key of the snapshot that belongs to the
169// repository's merge requests.
170func mergeRequestKeys(ctx context.Context, snap *kv.Snapshot, repo domain.RepoName) [][]byte {
171 var keys [][]byte
172 for _, p := range []string{
173 mrRepoPrefix(repo), commentPrefix + repo.String() + "/", readPrefix + repo.String() + "/",
174 mrOpenRepoPrefix(repo), mrTargetPrefix + repo.String() + "/",
175 } {
176 for k := range snap.Prefix(ctx, []byte(p)) {
177 keys = append(keys, slices.Clone(k))
178 }
179 }
180 if _, ok := snap.Get(ctx, seqKey(repo)); ok {
181 keys = append(keys, seqKey(repo))
182 }
183 return keys
184}
185
186// guardRepo is guardRepoKey on the database of the merge requests; see there.
187func (m *MergeRequests) guardRepo(ctx context.Context, b *kv.Batch, repo domain.RepoName) error {
188 return guardRepoKey(ctx, m.db, b, repo)
189}
190
191// Create takes the next number of the repository from mrseq/<repo> and stores
192// the merge request under it, in one batch that requires the sequence to be
193// unchanged since it was read: two concurrent creations never share a number.
194// The batch also requires the repository to be unchanged since it was read and
195// rewrites it (see guardRepo), so that a repository deleted meanwhile is not
196// left with a stray merge request (a push recording pushed_at makes it retry). A missing repository gives
197// domain.ErrNotFound.
198//
199//hexcheck:ignore suspicious-business-logic-in-adapter takes the next number under a version precondition: a storage concern, no policy
200func (m *MergeRequests) Create(ctx context.Context, mr domain.MergeRequest) (domain.MergeRequest, error) {
201 for range createTries {
202 repoItem, repoOK := m.db.Get(ctx, repoKey(mr.Repo))
203 if !repoOK {
204 return domain.MergeRequest{}, domain.ErrNotFound
205 }
206 var last int64
207 item, ok := m.db.Get(ctx, seqKey(mr.Repo))
208 if ok {
209 var seq storagepb.Sequence
210 if err := proto.Unmarshal(item.Value, &seq); err != nil {
211 return domain.MergeRequest{}, fmt.Errorf("decode merge request sequence of %q: %w", mr.Repo, err)
212 }
213 last = seq.GetLast()
214 }
215 mr.Number = last + 1
216 val, err := encodeMergeRequest(mr)
217 if err != nil {
218 return domain.MergeRequest{}, err
219 }
220 var b kv.Batch
221 b.RequireVersion(repoKey(mr.Repo), repoItem.Version)
222 b.Put(repoKey(mr.Repo), repoItem.Value) // see guardRepo
223 if ok {
224 b.RequireVersion(seqKey(mr.Repo), item.Version)
225 } else {
226 b.RequireAbsent(seqKey(mr.Repo))
227 }
228 b.Put(seqKey(mr.Repo), mustMarshal(&storagepb.Sequence{Last: mr.Number}))
229 b.RequireAbsent(mrKey(mr.Repo, mr.Number))
230 b.Put(mrKey(mr.Repo, mr.Number), val)
231 indexOpen(&b, nil, mr)
232 _, err = m.db.Apply(ctx, &b)
233 if errors.Is(err, kv.ErrConflict) {
234 continue
235 }
236 if err != nil {
237 return domain.MergeRequest{}, fmt.Errorf("create merge request of %q: %w", mr.Repo, err)
238 }
239 return m.Get(ctx, mr.Repo, mr.Number)
240 }
241 return domain.MergeRequest{}, fmt.Errorf("create merge request of %q: %w", mr.Repo, domain.ErrConflict)
242}
243
244// Get reads mr/<repo>/<number>.
245func (m *MergeRequests) Get(ctx context.Context, repo domain.RepoName, number int64) (domain.MergeRequest, error) {
246 if number < 1 {
247 return domain.MergeRequest{}, domain.ErrNotFound
248 }
249 item, ok := m.db.Get(ctx, mrKey(repo, number))
250 if !ok {
251 return domain.MergeRequest{}, domain.ErrNotFound
252 }
253 return decodeMergeRequest(repo, item)
254}
255
256// List returns the merge requests of the repository, highest number first.
257func (m *MergeRequests) List(ctx context.Context, repo domain.RepoName) ([]domain.MergeRequest, error) {
258 var out []domain.MergeRequest
259 for key, item := range m.db.Snapshot(ctx).Prefix(ctx, []byte(mrRepoPrefix(repo))) {
260 mr, err := decodeMergeRequest(repo, item)
261 if err != nil {
262 return nil, fmt.Errorf("decode %s: %w", key, err)
263 }
264 out = append(out, mr)
265 }
266 slices.Reverse(out)
267 return out, nil
268}
269
270// scanIndex reads the numbers under the prefixes of an index, in key order. It
271// reads keys only: no merge request is decoded. With a limit above 0 it stops as
272// soon as it has seen limit+1 keys and returns the first limit with truncated
273// set, so the work does not grow with the number of open requests on a branch.
274// The context is checked before every prefix and every key (the scan of ermite-kv
275// itself ignores it): on cancellation it stops and returns what it has with the
276// error of the context.
277func (m *MergeRequests) scanIndex(ctx context.Context, prefixes []string, limit int) (numbers []int64, truncated bool, err error) {
278 snap := m.db.Snapshot(ctx)
279 for _, prefix := range prefixes {
280 if err := ctx.Err(); err != nil {
281 return numbers, true, err
282 }
283 for key := range snap.Prefix(ctx, []byte(prefix)) {
284 if err := ctx.Err(); err != nil {
285 return numbers, true, err
286 }
287 if limit > 0 && len(numbers) == limit {
288 return numbers, true, nil // the limit+1th key: more exist
289 }
290 n, perr := strconv.ParseInt(string(key[strings.LastIndexByte(string(key), '/')+1:]), 10, 64)
291 if perr != nil {
292 return numbers, false, fmt.Errorf("index key %q: %w", key, perr)
293 }
294 numbers = append(numbers, n)
295 }
296 }
297 return numbers, false, nil
298}
299
300func (m *MergeRequests) byBranch(ctx context.Context, index string, repo domain.RepoName, branches []string, limit int) ([]int64, bool, error) {
301 prefixes := make([]string, 0, len(branches))
302 for _, b := range branches {
303 prefixes = append(prefixes, indexBranchPrefix(index, repo, b))
304 }
305 return m.scanIndex(ctx, prefixes, limit)
306}
307
308// OpenBySource returns the numbers of the open merge requests of the repository
309// whose source branch is one of branches, from the index: a scan of the keys of
310// those branches only, whatever the number of merge requests the repository
311// ever had, stopping after limit of them (truncated).
312func (m *MergeRequests) OpenBySource(ctx context.Context, repo domain.RepoName, branches []string, limit int) ([]int64, bool, error) {
313 return m.byBranch(ctx, mrOpenPrefix, repo, branches, limit)
314}
315
316// OpenByTarget is OpenBySource for the target branch.
317func (m *MergeRequests) OpenByTarget(ctx context.Context, repo domain.RepoName, branches []string, limit int) ([]int64, bool, error) {
318 return m.byBranch(ctx, mrTargetPrefix, repo, branches, limit)
319}
320
321// ListOpen returns the open merge requests of the repository, highest number
322// first, through the index: only the open ones are read.
323//
324//hexcheck:ignore suspicious-business-logic-in-adapter reads the open merge requests through their index: a storage concern, no policy
325func (m *MergeRequests) ListOpen(ctx context.Context, repo domain.RepoName) ([]domain.MergeRequest, error) {
326 numbers, _, err := m.scanIndex(ctx, []string{mrOpenRepoPrefix(repo)}, 0)
327 if err != nil {
328 return nil, err
329 }
330 slices.Sort(numbers)
331 var out []domain.MergeRequest
332 for _, n := range slices.Backward(numbers) {
333 mr, err := m.Get(ctx, repo, n)
334 if errors.Is(err, domain.ErrNotFound) {
335 continue // closed or deleted since the snapshot of the index
336 }
337 if err != nil {
338 return nil, err
339 }
340 if mr.IsOpen() {
341 out = append(out, mr)
342 }
343 }
344 return out, nil
345}
346
347// CountOpen counts the open merge requests of the repository from the keys of
348// the index: no request is decoded.
349func (m *MergeRequests) CountOpen(ctx context.Context, repo domain.RepoName) (int, error) {
350 numbers, _, err := m.scanIndex(ctx, []string{mrOpenRepoPrefix(repo)}, 0)
351 return len(numbers), err
352}
353
354// Update replaces the merge request with a version precondition.
355//
356//hexcheck:ignore suspicious-business-logic-in-adapter port method name only
357func (m *MergeRequests) Update(ctx context.Context, mr domain.MergeRequest) error {
358 val, err := encodeMergeRequest(mr)
359 if err != nil {
360 return err
361 }
362 key := mrKey(mr.Repo, mr.Number)
363 cur, ok := m.db.Get(ctx, key)
364 if !ok {
365 return domain.ErrNotFound
366 }
367 if mr.Version == 0 {
368 return fmt.Errorf("update merge request %d of %q: read it first: %w", mr.Number, mr.Repo, domain.ErrConflict)
369 }
370 var b kv.Batch
371 b.RequireVersion(key, mr.Version)
372 b.Put(key, val)
373 // The stored request is the one mr was read at (the version is required
374 // above): its index key moves with the write.
375 if old, err := decodeMergeRequest(mr.Repo, cur); err == nil {
376 indexOpen(&b, &old, mr)
377 } else {
378 indexOpen(&b, nil, mr)
379 }
380 if _, err := m.db.Apply(ctx, &b); err != nil {
381 if errors.Is(err, kv.ErrConflict) {
382 return domain.ErrConflict
383 }
384 return fmt.Errorf("update merge request %d of %q: %w", mr.Number, mr.Repo, err)
385 }
386 return nil
387}
388
389// ListComments returns the comments of the merge request, oldest first.
390func (m *MergeRequests) ListComments(ctx context.Context, repo domain.RepoName, number int64) ([]domain.Comment, error) {
391 var out []domain.Comment
392 for key, item := range m.db.Snapshot(ctx).Prefix(ctx, []byte(commentsPrefix(repo, number))) {
393 c, err := decodeComment(item)
394 if err != nil {
395 return nil, fmt.Errorf("decode %s: %w", key, err)
396 }
397 out = append(out, c)
398 }
399 return out, nil
400}
401
402// GetComment reads one comment.
403func (m *MergeRequests) GetComment(ctx context.Context, repo domain.RepoName, number, id int64) (domain.Comment, error) {
404 if number < 1 || id < 1 {
405 return domain.Comment{}, domain.ErrNotFound
406 }
407 item, ok := m.db.Get(ctx, commentKey(repo, number, id))
408 if !ok {
409 return domain.Comment{}, domain.ErrNotFound
410 }
411 return decodeComment(item)
412}
413
414// SaveComment writes the comment and the merge request in one batch, each with
415// its version precondition (absent for a new comment), and requires the
416// repository to exist (see guardRepo): domain.ErrNotFound otherwise.
417func (m *MergeRequests) SaveComment(ctx context.Context, mr domain.MergeRequest, c domain.Comment) error {
418 mrVal, err := encodeMergeRequest(mr)
419 if err != nil {
420 return err
421 }
422 cVal, err := encodeComment(c)
423 if err != nil {
424 return err
425 }
426 if mr.Version == 0 {
427 return fmt.Errorf("save comment on %d of %q: read the merge request first: %w", mr.Number, mr.Repo, domain.ErrConflict)
428 }
429 var b kv.Batch
430 if err := m.guardRepo(ctx, &b, mr.Repo); err != nil {
431 return err
432 }
433 b.RequireVersion(mrKey(mr.Repo, mr.Number), mr.Version)
434 b.Put(mrKey(mr.Repo, mr.Number), mrVal)
435 indexOpen(&b, nil, mr)
436 if c.Version == 0 {
437 b.RequireAbsent(commentKey(mr.Repo, mr.Number, c.ID))
438 } else {
439 b.RequireVersion(commentKey(mr.Repo, mr.Number, c.ID), c.Version)
440 }
441 b.Put(commentKey(mr.Repo, mr.Number, c.ID), cVal)
442 if _, err := m.db.Apply(ctx, &b); err != nil {
443 if errors.Is(err, kv.ErrConflict) {
444 return domain.ErrConflict
445 }
446 return fmt.Errorf("save comment on %d of %q: %w", mr.Number, mr.Repo, err)
447 }
448 return nil
449}
450
451// GetRead reads mrread/<repo>/<number>: the id of a commit, or nothing.
452func (m *MergeRequests) GetRead(ctx context.Context, repo domain.RepoName, number int64) (domain.ObjectID, error) {
453 item, ok := m.db.Get(ctx, readKey(repo, number))
454 if !ok {
455 return domain.ObjectID{}, nil
456 }
457 id, err := domain.ParseObjectID(string(item.Value))
458 if err != nil {
459 return domain.ObjectID{}, fmt.Errorf("decode read position of %d in %q: %w", number, repo, err)
460 }
461 return id, nil
462}
463
464// SetRead writes mrread/<repo>/<number>. The repository must exist (see
465// guardRepo): domain.ErrNotFound otherwise.
466//
467//hexcheck:ignore suspicious-business-logic-in-adapter retries a write that lost a race with the repository key: a storage concern, no policy
468func (m *MergeRequests) SetRead(ctx context.Context, repo domain.RepoName, number int64, head domain.ObjectID) error {
469 if head.IsZero() {
470 return fmt.Errorf("%w: empty read position", domain.ErrInvalidObjectID)
471 }
472 for range createTries {
473 var b kv.Batch
474 if err := m.guardRepo(ctx, &b, repo); err != nil {
475 return err
476 }
477 b.Put(readKey(repo, number), []byte(head.String()))
478 _, err := m.db.Apply(ctx, &b)
479 if errors.Is(err, kv.ErrConflict) {
480 continue
481 }
482 if err != nil {
483 return fmt.Errorf("record read position of %d in %q: %w", number, repo, err)
484 }
485 return nil
486 }
487 return fmt.Errorf("record read position of %d in %q: %w", number, repo, domain.ErrConflict)
488}
489
490func encodeMergeRequest(mr domain.MergeRequest) ([]byte, error) {
491 if mr.Repo.IsZero() {
492 return nil, fmt.Errorf("%w: empty name", domain.ErrInvalidName)
493 }
494 if !utf8.ValidString(mr.Author) {
495 return nil, errors.New("merge request author is not valid UTF-8")
496 }
497 state, err := encodeState(mr.State)
498 if err != nil {
499 return nil, err
500 }
501 mode, err := encodeMode(mr.MergeMode)
502 if err != nil {
503 return nil, err
504 }
505 return mustMarshal(&storagepb.MergeRequest{
506 Number: mr.Number,
507 Title: mr.Title.String(),
508 Description: mr.Description.String(),
509 SourceBranch: mr.Source,
510 TargetBranch: mr.Target,
511 State: state,
512 Author: mr.Author,
513 CreatedAt: encodeTime(mr.CreatedAt),
514 UpdatedAt: encodeTime(mr.UpdatedAt),
515 HeadId: mr.HeadID.String(),
516 BaseId: mr.BaseID.String(),
517 MergedCommitId: mr.MergedCommitID.String(),
518 MergeMode: mode,
519 CommentSeq: mr.CommentSeq,
520 }), nil
521}
522
523//hexcheck:ignore suspicious-business-logic-in-adapter schema decoding, not a policy
524func decodeMergeRequest(repo domain.RepoName, item kv.Item) (domain.MergeRequest, error) {
525 var pb storagepb.MergeRequest
526 if err := proto.Unmarshal(item.Value, &pb); err != nil {
527 return domain.MergeRequest{}, fmt.Errorf("decode merge request: %w", err)
528 }
529 fail := func(err error) (domain.MergeRequest, error) {
530 return domain.MergeRequest{}, fmt.Errorf("decode merge request %d of %q: %w", pb.GetNumber(), repo, err)
531 }
532 title, err := domain.ParseMRTitle(pb.GetTitle())
533 if err != nil {
534 return fail(err)
535 }
536 desc, err := domain.ParsePlainText(pb.GetDescription())
537 if err != nil {
538 return fail(err)
539 }
540 state, err := decodeState(pb.GetState())
541 if err != nil {
542 return fail(err)
543 }
544 mr := domain.MergeRequest{
545 Repo: repo, Number: pb.GetNumber(), Title: title, Description: desc,
546 Source: pb.GetSourceBranch(), Target: pb.GetTargetBranch(), State: state,
547 Author: pb.GetAuthor(),
548 CreatedAt: decodeTime(pb.GetCreatedAt()), UpdatedAt: decodeTime(pb.GetUpdatedAt()),
549 MergeMode: decodeMode(pb.GetMergeMode()),
550 CommentSeq: pb.GetCommentSeq(),
551 Version: item.Version,
552 }
553 for _, f := range []struct {
554 dst *domain.ObjectID
555 src string
556 }{{&mr.HeadID, pb.GetHeadId()}, {&mr.BaseID, pb.GetBaseId()}, {&mr.MergedCommitID, pb.GetMergedCommitId()}} {
557 if f.src == "" {
558 continue
559 }
560 if *f.dst, err = domain.ParseObjectID(f.src); err != nil {
561 return fail(err)
562 }
563 }
564 return mr, nil
565}
566
567func encodeComment(c domain.Comment) ([]byte, error) {
568 side, err := encodeSide(c.Side)
569 if err != nil {
570 return nil, err
571 }
572 return mustMarshal(&storagepb.Comment{
573 Id: c.ID,
574 Author: c.Author,
575 Path: c.Path,
576 Line: int32(c.Line), //nolint:gosec // validated against the diff, far below 2^31
577 Side: side,
578 CommitId: c.CommitID.String(),
579 Text: c.Text.String(),
580 Resolved: c.Resolved,
581 CreatedAt: encodeTime(c.CreatedAt),
582 }), nil
583}
584
585//hexcheck:ignore suspicious-business-logic-in-adapter schema decoding, not a policy
586func decodeComment(item kv.Item) (domain.Comment, error) {
587 var pb storagepb.Comment
588 if err := proto.Unmarshal(item.Value, &pb); err != nil {
589 return domain.Comment{}, fmt.Errorf("decode comment: %w", err)
590 }
591 text, err := domain.ParsePlainText(pb.GetText())
592 if err != nil {
593 return domain.Comment{}, fmt.Errorf("decode comment %d: %w", pb.GetId(), err)
594 }
595 c := domain.Comment{
596 ID: pb.GetId(), Author: pb.GetAuthor(), Path: pb.GetPath(), Line: int(pb.GetLine()),
597 Side: decodeSide(pb.GetSide()), Text: text, Resolved: pb.GetResolved(),
598 CreatedAt: decodeTime(pb.GetCreatedAt()), Version: item.Version,
599 }
600 if pb.GetCommitId() != "" {
601 if c.CommitID, err = domain.ParseObjectID(pb.GetCommitId()); err != nil {
602 return domain.Comment{}, fmt.Errorf("decode comment %d: %w", c.ID, err)
603 }
604 }
605 return c, nil
606}
607
608func encodeState(s domain.MRState) (storagepb.MergeRequestState, error) {
609 switch s {
610 case domain.MRStateOpen:
611 return storagepb.MergeRequestState_MERGE_REQUEST_STATE_OPEN, nil
612 case domain.MRStateMerged:
613 return storagepb.MergeRequestState_MERGE_REQUEST_STATE_MERGED, nil
614 case domain.MRStateClosed:
615 return storagepb.MergeRequestState_MERGE_REQUEST_STATE_CLOSED, nil
616 }
617 return 0, fmt.Errorf("%w: merge request state %q", domain.ErrInvalid, s)
618}
619
620func decodeState(s storagepb.MergeRequestState) (domain.MRState, error) {
621 switch s {
622 case storagepb.MergeRequestState_MERGE_REQUEST_STATE_OPEN:
623 return domain.MRStateOpen, nil
624 case storagepb.MergeRequestState_MERGE_REQUEST_STATE_MERGED:
625 return domain.MRStateMerged, nil
626 case storagepb.MergeRequestState_MERGE_REQUEST_STATE_CLOSED:
627 return domain.MRStateClosed, nil
628 }
629 return "", fmt.Errorf("unknown merge request state %d", s)
630}
631
632func encodeMode(m domain.MergeMode) (storagepb.MergeMode, error) {
633 switch m {
634 case "":
635 return storagepb.MergeMode_MERGE_MODE_UNSPECIFIED, nil
636 case domain.MergeModeFastForward:
637 return storagepb.MergeMode_MERGE_MODE_FAST_FORWARD, nil
638 case domain.MergeModeSquash:
639 return storagepb.MergeMode_MERGE_MODE_SQUASH, nil
640 }
641 return 0, fmt.Errorf("%w: merge mode %q", domain.ErrInvalid, m)
642}
643
644// decodeMode maps an unknown mode to none: it only labels a merged request.
645func decodeMode(m storagepb.MergeMode) domain.MergeMode {
646 switch m {
647 case storagepb.MergeMode_MERGE_MODE_FAST_FORWARD:
648 return domain.MergeModeFastForward
649 case storagepb.MergeMode_MERGE_MODE_SQUASH:
650 return domain.MergeModeSquash
651 }
652 return ""
653}
654
655func encodeSide(s domain.DiffSide) (storagepb.DiffSide, error) {
656 switch s {
657 case domain.SideNone:
658 return storagepb.DiffSide_DIFF_SIDE_UNSPECIFIED, nil
659 case domain.SideOld:
660 return storagepb.DiffSide_DIFF_SIDE_OLD, nil
661 case domain.SideNew:
662 return storagepb.DiffSide_DIFF_SIDE_NEW, nil
663 }
664 return 0, fmt.Errorf("%w: diff side %q", domain.ErrInvalid, s)
665}
666
667func decodeSide(s storagepb.DiffSide) domain.DiffSide {
668 switch s {
669 case storagepb.DiffSide_DIFF_SIDE_OLD:
670 return domain.SideOld
671 case storagepb.DiffSide_DIFF_SIDE_NEW:
672 return domain.SideNew
673 }
674 return domain.SideNone
675}
676
677func encodeSettings(s domain.RepoSettings) (*storagepb.RepoSettings, error) {
678 pb := &storagepb.RepoSettings{FailingChecksBlock: s.FailingChecksBlock}
679 for _, m := range s.AllowedMergeModes {
680 mode, err := encodeMode(m)
681 if err != nil || mode == storagepb.MergeMode_MERGE_MODE_UNSPECIFIED {
682 return nil, fmt.Errorf("%w: merge mode %q in the settings", domain.ErrInvalid, m)
683 }
684 pb.AllowedMergeModes = append(pb.AllowedMergeModes, mode)
685 }
686 return pb, nil
687}
688
689// decodeSettings returns the defaults for a repository stored before merge
690// requests existed, and for settings left with no mode that this build knows.
691//
692//hexcheck:ignore suspicious-business-logic-in-adapter schema decoding with the documented defaults
693func decodeSettings(pb *storagepb.RepoSettings) domain.RepoSettings {
694 if pb == nil {
695 return domain.DefaultRepoSettings()
696 }
697 var modes []domain.MergeMode
698 for _, m := range pb.GetAllowedMergeModes() {
699 if mode := decodeMode(m); mode != "" {
700 modes = append(modes, mode)
701 }
702 }
703 if len(modes) == 0 {
704 return domain.DefaultRepoSettings()
705 }
706 normalized, err := domain.NormalizeMergeModes(modes)
707 if err != nil {
708 return domain.DefaultRepoSettings()
709 }
710 return domain.RepoSettings{AllowedMergeModes: normalized, FailingChecksBlock: pb.GetFailingChecksBlock()}
711}