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// Update replaces the merge request with a version precondition.
348//
349//hexcheck:ignore suspicious-business-logic-in-adapter port method name only
350func (m *MergeRequests) Update(ctx context.Context, mr domain.MergeRequest) error {
351 val, err := encodeMergeRequest(mr)
352 if err != nil {
353 return err
354 }
355 key := mrKey(mr.Repo, mr.Number)
356 cur, ok := m.db.Get(ctx, key)
357 if !ok {
358 return domain.ErrNotFound
359 }
360 if mr.Version == 0 {
361 return fmt.Errorf("update merge request %d of %q: read it first: %w", mr.Number, mr.Repo, domain.ErrConflict)
362 }
363 var b kv.Batch
364 b.RequireVersion(key, mr.Version)
365 b.Put(key, val)
366 // The stored request is the one mr was read at (the version is required
367 // above): its index key moves with the write.
368 if old, err := decodeMergeRequest(mr.Repo, cur); err == nil {
369 indexOpen(&b, &old, mr)
370 } else {
371 indexOpen(&b, nil, mr)
372 }
373 if _, err := m.db.Apply(ctx, &b); err != nil {
374 if errors.Is(err, kv.ErrConflict) {
375 return domain.ErrConflict
376 }
377 return fmt.Errorf("update merge request %d of %q: %w", mr.Number, mr.Repo, err)
378 }
379 return nil
380}
381
382// ListComments returns the comments of the merge request, oldest first.
383func (m *MergeRequests) ListComments(ctx context.Context, repo domain.RepoName, number int64) ([]domain.Comment, error) {
384 var out []domain.Comment
385 for key, item := range m.db.Snapshot(ctx).Prefix(ctx, []byte(commentsPrefix(repo, number))) {
386 c, err := decodeComment(item)
387 if err != nil {
388 return nil, fmt.Errorf("decode %s: %w", key, err)
389 }
390 out = append(out, c)
391 }
392 return out, nil
393}
394
395// GetComment reads one comment.
396func (m *MergeRequests) GetComment(ctx context.Context, repo domain.RepoName, number, id int64) (domain.Comment, error) {
397 if number < 1 || id < 1 {
398 return domain.Comment{}, domain.ErrNotFound
399 }
400 item, ok := m.db.Get(ctx, commentKey(repo, number, id))
401 if !ok {
402 return domain.Comment{}, domain.ErrNotFound
403 }
404 return decodeComment(item)
405}
406
407// SaveComment writes the comment and the merge request in one batch, each with
408// its version precondition (absent for a new comment), and requires the
409// repository to exist (see guardRepo): domain.ErrNotFound otherwise.
410func (m *MergeRequests) SaveComment(ctx context.Context, mr domain.MergeRequest, c domain.Comment) error {
411 mrVal, err := encodeMergeRequest(mr)
412 if err != nil {
413 return err
414 }
415 cVal, err := encodeComment(c)
416 if err != nil {
417 return err
418 }
419 if mr.Version == 0 {
420 return fmt.Errorf("save comment on %d of %q: read the merge request first: %w", mr.Number, mr.Repo, domain.ErrConflict)
421 }
422 var b kv.Batch
423 if err := m.guardRepo(ctx, &b, mr.Repo); err != nil {
424 return err
425 }
426 b.RequireVersion(mrKey(mr.Repo, mr.Number), mr.Version)
427 b.Put(mrKey(mr.Repo, mr.Number), mrVal)
428 indexOpen(&b, nil, mr)
429 if c.Version == 0 {
430 b.RequireAbsent(commentKey(mr.Repo, mr.Number, c.ID))
431 } else {
432 b.RequireVersion(commentKey(mr.Repo, mr.Number, c.ID), c.Version)
433 }
434 b.Put(commentKey(mr.Repo, mr.Number, c.ID), cVal)
435 if _, err := m.db.Apply(ctx, &b); err != nil {
436 if errors.Is(err, kv.ErrConflict) {
437 return domain.ErrConflict
438 }
439 return fmt.Errorf("save comment on %d of %q: %w", mr.Number, mr.Repo, err)
440 }
441 return nil
442}
443
444// GetRead reads mrread/<repo>/<number>: the id of a commit, or nothing.
445func (m *MergeRequests) GetRead(ctx context.Context, repo domain.RepoName, number int64) (domain.ObjectID, error) {
446 item, ok := m.db.Get(ctx, readKey(repo, number))
447 if !ok {
448 return domain.ObjectID{}, nil
449 }
450 id, err := domain.ParseObjectID(string(item.Value))
451 if err != nil {
452 return domain.ObjectID{}, fmt.Errorf("decode read position of %d in %q: %w", number, repo, err)
453 }
454 return id, nil
455}
456
457// SetRead writes mrread/<repo>/<number>. The repository must exist (see
458// guardRepo): domain.ErrNotFound otherwise.
459//
460//hexcheck:ignore suspicious-business-logic-in-adapter retries a write that lost a race with the repository key: a storage concern, no policy
461func (m *MergeRequests) SetRead(ctx context.Context, repo domain.RepoName, number int64, head domain.ObjectID) error {
462 if head.IsZero() {
463 return fmt.Errorf("%w: empty read position", domain.ErrInvalidObjectID)
464 }
465 for range createTries {
466 var b kv.Batch
467 if err := m.guardRepo(ctx, &b, repo); err != nil {
468 return err
469 }
470 b.Put(readKey(repo, number), []byte(head.String()))
471 _, err := m.db.Apply(ctx, &b)
472 if errors.Is(err, kv.ErrConflict) {
473 continue
474 }
475 if err != nil {
476 return fmt.Errorf("record read position of %d in %q: %w", number, repo, err)
477 }
478 return nil
479 }
480 return fmt.Errorf("record read position of %d in %q: %w", number, repo, domain.ErrConflict)
481}
482
483func encodeMergeRequest(mr domain.MergeRequest) ([]byte, error) {
484 if mr.Repo.IsZero() {
485 return nil, fmt.Errorf("%w: empty name", domain.ErrInvalidName)
486 }
487 if !utf8.ValidString(mr.Author) {
488 return nil, errors.New("merge request author is not valid UTF-8")
489 }
490 state, err := encodeState(mr.State)
491 if err != nil {
492 return nil, err
493 }
494 mode, err := encodeMode(mr.MergeMode)
495 if err != nil {
496 return nil, err
497 }
498 return mustMarshal(&storagepb.MergeRequest{
499 Number: mr.Number,
500 Title: mr.Title.String(),
501 Description: mr.Description.String(),
502 SourceBranch: mr.Source,
503 TargetBranch: mr.Target,
504 State: state,
505 Author: mr.Author,
506 CreatedAt: encodeTime(mr.CreatedAt),
507 UpdatedAt: encodeTime(mr.UpdatedAt),
508 HeadId: mr.HeadID.String(),
509 BaseId: mr.BaseID.String(),
510 MergedCommitId: mr.MergedCommitID.String(),
511 MergeMode: mode,
512 CommentSeq: mr.CommentSeq,
513 }), nil
514}
515
516//hexcheck:ignore suspicious-business-logic-in-adapter schema decoding, not a policy
517func decodeMergeRequest(repo domain.RepoName, item kv.Item) (domain.MergeRequest, error) {
518 var pb storagepb.MergeRequest
519 if err := proto.Unmarshal(item.Value, &pb); err != nil {
520 return domain.MergeRequest{}, fmt.Errorf("decode merge request: %w", err)
521 }
522 fail := func(err error) (domain.MergeRequest, error) {
523 return domain.MergeRequest{}, fmt.Errorf("decode merge request %d of %q: %w", pb.GetNumber(), repo, err)
524 }
525 title, err := domain.ParseMRTitle(pb.GetTitle())
526 if err != nil {
527 return fail(err)
528 }
529 desc, err := domain.ParsePlainText(pb.GetDescription())
530 if err != nil {
531 return fail(err)
532 }
533 state, err := decodeState(pb.GetState())
534 if err != nil {
535 return fail(err)
536 }
537 mr := domain.MergeRequest{
538 Repo: repo, Number: pb.GetNumber(), Title: title, Description: desc,
539 Source: pb.GetSourceBranch(), Target: pb.GetTargetBranch(), State: state,
540 Author: pb.GetAuthor(),
541 CreatedAt: decodeTime(pb.GetCreatedAt()), UpdatedAt: decodeTime(pb.GetUpdatedAt()),
542 MergeMode: decodeMode(pb.GetMergeMode()),
543 CommentSeq: pb.GetCommentSeq(),
544 Version: item.Version,
545 }
546 for _, f := range []struct {
547 dst *domain.ObjectID
548 src string
549 }{{&mr.HeadID, pb.GetHeadId()}, {&mr.BaseID, pb.GetBaseId()}, {&mr.MergedCommitID, pb.GetMergedCommitId()}} {
550 if f.src == "" {
551 continue
552 }
553 if *f.dst, err = domain.ParseObjectID(f.src); err != nil {
554 return fail(err)
555 }
556 }
557 return mr, nil
558}
559
560func encodeComment(c domain.Comment) ([]byte, error) {
561 side, err := encodeSide(c.Side)
562 if err != nil {
563 return nil, err
564 }
565 return mustMarshal(&storagepb.Comment{
566 Id: c.ID,
567 Author: c.Author,
568 Path: c.Path,
569 Line: int32(c.Line), //nolint:gosec // validated against the diff, far below 2^31
570 Side: side,
571 CommitId: c.CommitID.String(),
572 Text: c.Text.String(),
573 Resolved: c.Resolved,
574 CreatedAt: encodeTime(c.CreatedAt),
575 }), nil
576}
577
578//hexcheck:ignore suspicious-business-logic-in-adapter schema decoding, not a policy
579func decodeComment(item kv.Item) (domain.Comment, error) {
580 var pb storagepb.Comment
581 if err := proto.Unmarshal(item.Value, &pb); err != nil {
582 return domain.Comment{}, fmt.Errorf("decode comment: %w", err)
583 }
584 text, err := domain.ParsePlainText(pb.GetText())
585 if err != nil {
586 return domain.Comment{}, fmt.Errorf("decode comment %d: %w", pb.GetId(), err)
587 }
588 c := domain.Comment{
589 ID: pb.GetId(), Author: pb.GetAuthor(), Path: pb.GetPath(), Line: int(pb.GetLine()),
590 Side: decodeSide(pb.GetSide()), Text: text, Resolved: pb.GetResolved(),
591 CreatedAt: decodeTime(pb.GetCreatedAt()), Version: item.Version,
592 }
593 if pb.GetCommitId() != "" {
594 if c.CommitID, err = domain.ParseObjectID(pb.GetCommitId()); err != nil {
595 return domain.Comment{}, fmt.Errorf("decode comment %d: %w", c.ID, err)
596 }
597 }
598 return c, nil
599}
600
601func encodeState(s domain.MRState) (storagepb.MergeRequestState, error) {
602 switch s {
603 case domain.MRStateOpen:
604 return storagepb.MergeRequestState_MERGE_REQUEST_STATE_OPEN, nil
605 case domain.MRStateMerged:
606 return storagepb.MergeRequestState_MERGE_REQUEST_STATE_MERGED, nil
607 case domain.MRStateClosed:
608 return storagepb.MergeRequestState_MERGE_REQUEST_STATE_CLOSED, nil
609 }
610 return 0, fmt.Errorf("%w: merge request state %q", domain.ErrInvalid, s)
611}
612
613func decodeState(s storagepb.MergeRequestState) (domain.MRState, error) {
614 switch s {
615 case storagepb.MergeRequestState_MERGE_REQUEST_STATE_OPEN:
616 return domain.MRStateOpen, nil
617 case storagepb.MergeRequestState_MERGE_REQUEST_STATE_MERGED:
618 return domain.MRStateMerged, nil
619 case storagepb.MergeRequestState_MERGE_REQUEST_STATE_CLOSED:
620 return domain.MRStateClosed, nil
621 }
622 return "", fmt.Errorf("unknown merge request state %d", s)
623}
624
625func encodeMode(m domain.MergeMode) (storagepb.MergeMode, error) {
626 switch m {
627 case "":
628 return storagepb.MergeMode_MERGE_MODE_UNSPECIFIED, nil
629 case domain.MergeModeFastForward:
630 return storagepb.MergeMode_MERGE_MODE_FAST_FORWARD, nil
631 case domain.MergeModeSquash:
632 return storagepb.MergeMode_MERGE_MODE_SQUASH, nil
633 }
634 return 0, fmt.Errorf("%w: merge mode %q", domain.ErrInvalid, m)
635}
636
637// decodeMode maps an unknown mode to none: it only labels a merged request.
638func decodeMode(m storagepb.MergeMode) domain.MergeMode {
639 switch m {
640 case storagepb.MergeMode_MERGE_MODE_FAST_FORWARD:
641 return domain.MergeModeFastForward
642 case storagepb.MergeMode_MERGE_MODE_SQUASH:
643 return domain.MergeModeSquash
644 }
645 return ""
646}
647
648func encodeSide(s domain.DiffSide) (storagepb.DiffSide, error) {
649 switch s {
650 case domain.SideNone:
651 return storagepb.DiffSide_DIFF_SIDE_UNSPECIFIED, nil
652 case domain.SideOld:
653 return storagepb.DiffSide_DIFF_SIDE_OLD, nil
654 case domain.SideNew:
655 return storagepb.DiffSide_DIFF_SIDE_NEW, nil
656 }
657 return 0, fmt.Errorf("%w: diff side %q", domain.ErrInvalid, s)
658}
659
660func decodeSide(s storagepb.DiffSide) domain.DiffSide {
661 switch s {
662 case storagepb.DiffSide_DIFF_SIDE_OLD:
663 return domain.SideOld
664 case storagepb.DiffSide_DIFF_SIDE_NEW:
665 return domain.SideNew
666 }
667 return domain.SideNone
668}
669
670func encodeSettings(s domain.RepoSettings) (*storagepb.RepoSettings, error) {
671 pb := &storagepb.RepoSettings{FailingChecksBlock: s.FailingChecksBlock}
672 for _, m := range s.AllowedMergeModes {
673 mode, err := encodeMode(m)
674 if err != nil || mode == storagepb.MergeMode_MERGE_MODE_UNSPECIFIED {
675 return nil, fmt.Errorf("%w: merge mode %q in the settings", domain.ErrInvalid, m)
676 }
677 pb.AllowedMergeModes = append(pb.AllowedMergeModes, mode)
678 }
679 return pb, nil
680}
681
682// decodeSettings returns the defaults for a repository stored before merge
683// requests existed, and for settings left with no mode that this build knows.
684//
685//hexcheck:ignore suspicious-business-logic-in-adapter schema decoding with the documented defaults
686func decodeSettings(pb *storagepb.RepoSettings) domain.RepoSettings {
687 if pb == nil {
688 return domain.DefaultRepoSettings()
689 }
690 var modes []domain.MergeMode
691 for _, m := range pb.GetAllowedMergeModes() {
692 if mode := decodeMode(m); mode != "" {
693 modes = append(modes, mode)
694 }
695 }
696 if len(modes) == 0 {
697 return domain.DefaultRepoSettings()
698 }
699 normalized, err := domain.NormalizeMergeModes(modes)
700 if err != nil {
701 return domain.DefaultRepoSettings()
702 }
703 return domain.RepoSettings{AllowedMergeModes: normalized, FailingChecksBlock: pb.GetFailingChecksBlock()}
704}