1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package kvstore
4
5import (
6 "context"
7 "errors"
8 "fmt"
9 "slices"
10 "strconv"
11 "strings"
12
13 kv "bnema.dev/ermite-kv"
14 storagepb "bnema.dev/ermite/gen/ermite/storage"
15 "bnema.dev/ermite/internal/domain"
16 "bnema.dev/ermite/internal/ports"
17 "google.golang.org/protobuf/proto"
18)
19
20const issueRefPrefix = "isref/"
21const mrRefPrefix = "mrref/"
22
23// IssueLinks stores the derived issue-to-merge-request index.
24type IssueLinks struct{ db *kv.DB }
25
26var _ ports.IssueLinkStore = (*IssueLinks)(nil)
27
28// IssueLinks returns a link store sharing this Store's database and lifetime.
29func (s *Store) IssueLinks() *IssueLinks { return &IssueLinks{db: s.db} }
30
31func mrRefKey(repo domain.RepoName, mr int64) []byte {
32 return []byte(mrRefPrefix + repo.String() + "/" + fmt.Sprintf(numberFormat, mr))
33}
34func issueRefKeys(repo domain.RepoName, issue int64) string {
35 return issueRefPrefix + repo.String() + "/" + fmt.Sprintf(numberFormat, issue) + "/"
36}
37func issueRefKey(repo domain.RepoName, issue, mr int64) []byte {
38 return []byte(issueRefKeys(repo, issue) + fmt.Sprintf(numberFormat, mr))
39}
40
41// ReplaceLinks atomically replaces both directions of the index.
42func (l *IssueLinks) ReplaceLinks(ctx context.Context, repo domain.RepoName, mr int64, issues, closes []int64) (added, removed []domain.IssueLink, err error) {
43 key := mrRefKey(repo, mr)
44 item, exists := l.db.Get(ctx, key)
45 var old storagepb.MergeRequestReferences
46 if exists {
47 if err := proto.Unmarshal(item.Value, &old); err != nil {
48 return nil, nil, fmt.Errorf("decode MR references: %w", err)
49 }
50 }
51 issues = slices.Sorted(slices.Values(issues))
52 issues = slices.Compact(issues)
53 closes = slices.Sorted(slices.Values(closes))
54 closes = slices.Compact(closes)
55 for _, n := range issues {
56 if n < 1 {
57 return nil, nil, domain.ErrInvalid
58 }
59 }
60 issueSet, closeSet := toSet(issues), toSet(closes)
61 oldIssues, oldCloses := toSet(old.GetIssues()), toSet(old.GetCloses())
62 for _, n := range closes {
63 if _, ok := issueSet[n]; !ok {
64 return nil, nil, domain.ErrInvalid
65 }
66 }
67 var b kv.Batch
68 if err := guardRepoKey(ctx, l.db, &b, repo); err != nil {
69 return nil, nil, err
70 }
71 if exists {
72 b.RequireVersion(key, item.Version)
73 } else {
74 b.RequireAbsent(key)
75 }
76 for _, n := range old.GetIssues() {
77 _, wasClosing := oldCloses[n]
78 if _, ok := issueSet[n]; !ok {
79 b.Delete(issueRefKey(repo, n, mr))
80 removed = append(removed, domain.IssueLink{Issue: n, MergeRequest: mr, Closes: wasClosing})
81 } else if _, still := closeSet[n]; wasClosing && !still {
82 removed = append(removed, domain.IssueLink{Issue: n, MergeRequest: mr, Closes: true})
83 }
84 }
85 for _, n := range issues {
86 _, closing := closeSet[n]
87 b.Put(issueRefKey(repo, n, mr), mustMarshal(&storagepb.IssueLink{Closes: closing}))
88 _, was := oldIssues[n]
89 _, wasClosing := oldCloses[n]
90 if !was || closing && !wasClosing {
91 added = append(added, domain.IssueLink{Issue: n, MergeRequest: mr, Closes: closing})
92 }
93 }
94 b.Put(key, mustMarshal(&storagepb.MergeRequestReferences{Issues: issues, Closes: closes}))
95 if _, err := l.db.Apply(ctx, &b); err != nil {
96 if errors.Is(err, kv.ErrConflict) {
97 return nil, nil, domain.ErrConflict
98 }
99 return nil, nil, fmt.Errorf("replace issue links: %w", err)
100 }
101 return added, removed, nil
102}
103
104// LinksOfIssue returns links ordered by merge request number.
105func (l *IssueLinks) LinksOfIssue(ctx context.Context, repo domain.RepoName, issue int64, limit int) ([]domain.IssueLink, error) {
106 var result []domain.IssueLink
107 prefix := issueRefKeys(repo, issue)
108 for key, item := range l.db.Snapshot(ctx).Prefix(ctx, []byte(prefix)) {
109 if err := ctx.Err(); err != nil {
110 return nil, err
111 }
112 mr, err := strconv.ParseInt(strings.TrimPrefix(string(key), prefix), 10, 64)
113 if err != nil {
114 return nil, fmt.Errorf("decode issue link key: %w", err)
115 }
116 var pb storagepb.IssueLink
117 if err := proto.Unmarshal(item.Value, &pb); err != nil {
118 return nil, fmt.Errorf("decode issue link: %w", err)
119 }
120 result = append(result, domain.IssueLink{Issue: issue, MergeRequest: mr, Closes: pb.GetCloses()})
121 if limit > 0 && len(result) >= limit {
122 break
123 }
124 }
125 return result, nil
126}
127
128// LinksOfMergeRequest returns the links recorded by its last rebuild.
129func (l *IssueLinks) LinksOfMergeRequest(ctx context.Context, repo domain.RepoName, mr int64) ([]domain.IssueLink, error) {
130 item, ok := l.db.Get(ctx, mrRefKey(repo, mr))
131 if !ok {
132 return nil, nil
133 }
134 var pb storagepb.MergeRequestReferences
135 if err := proto.Unmarshal(item.Value, &pb); err != nil {
136 return nil, fmt.Errorf("decode MR references: %w", err)
137 }
138 var result []domain.IssueLink
139 for _, n := range pb.GetIssues() {
140 result = append(result, domain.IssueLink{Issue: n, MergeRequest: mr, Closes: slices.Contains(pb.GetCloses(), n)})
141 }
142 return result, nil
143}
144
145func toSet(numbers []int64) map[int64]struct{} {
146 set := make(map[int64]struct{}, len(numbers))
147 for _, n := range numbers {
148 set[n] = struct{}{}
149 }
150 return set
151}