1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package kvstore
4
5import (
6 "container/heap"
7 "context"
8 "errors"
9 "fmt"
10 "slices"
11 "strconv"
12 "strings"
13
14 kv "bnema.dev/ermite-kv"
15 storagepb "bnema.dev/ermite/gen/ermite/storage"
16 "bnema.dev/ermite/internal/domain"
17 "bnema.dev/ermite/internal/ports"
18 "google.golang.org/protobuf/proto"
19)
20
21const issueRefPrefix = "isref/"
22const mrRefPrefix = "mrref/"
23
24// IssueLinks stores the derived issue-to-merge-request index.
25type IssueLinks struct{ db *kv.DB }
26
27var _ ports.IssueLinkStore = (*IssueLinks)(nil)
28
29// IssueLinks returns a link store sharing this Store's database and lifetime.
30func (s *Store) IssueLinks() *IssueLinks { return &IssueLinks{db: s.db} }
31
32func mrRefKey(repo domain.RepoName, mr int64) []byte {
33 return []byte(mrRefPrefix + repo.String() + "/" + fmt.Sprintf(numberFormat, mr))
34}
35func issueRefKeys(repo domain.RepoName, issue int64) string {
36 return issueRefPrefix + repo.String() + "/" + fmt.Sprintf(numberFormat, issue) + "/"
37}
38func issueRefKey(repo domain.RepoName, issue, mr int64) []byte {
39 return []byte(issueRefKeys(repo, issue) + fmt.Sprintf(numberFormat, mr))
40}
41
42// ReplaceLinks atomically replaces both directions of the index.
43func (l *IssueLinks) ReplaceLinks(ctx context.Context, repo domain.RepoName, mr int64, issues, closes []int64, ref ports.ReferenceEvent) (added, removed []domain.IssueLink, err error) {
44 key := mrRefKey(repo, mr)
45 item, exists := l.db.Get(ctx, key)
46 var old storagepb.MergeRequestReferences
47 if exists {
48 if err := proto.Unmarshal(item.Value, &old); err != nil {
49 return nil, nil, fmt.Errorf("decode MR references: %w", err)
50 }
51 }
52 issues = slices.Sorted(slices.Values(issues))
53 issues = slices.Compact(issues)
54 closes = slices.Sorted(slices.Values(closes))
55 closes = slices.Compact(closes)
56 for _, n := range issues {
57 if n < 1 {
58 return nil, nil, domain.ErrInvalid
59 }
60 }
61 issueSet, closeSet := toSet(issues), toSet(closes)
62 oldIssues, oldCloses := toSet(old.GetIssues()), toSet(old.GetCloses())
63 for _, n := range closes {
64 if _, ok := issueSet[n]; !ok {
65 return nil, nil, domain.ErrInvalid
66 }
67 }
68 var b kv.Batch
69 if err := guardRepoKey(ctx, l.db, &b, repo); err != nil {
70 return nil, nil, err
71 }
72 if exists {
73 b.RequireVersion(key, item.Version)
74 } else {
75 b.RequireAbsent(key)
76 }
77 for _, n := range old.GetIssues() {
78 _, wasClosing := oldCloses[n]
79 if _, ok := issueSet[n]; !ok {
80 b.Delete(issueRefKey(repo, n, mr))
81 removed = append(removed, domain.IssueLink{Issue: n, MergeRequest: mr, Closes: wasClosing})
82 } else if _, still := closeSet[n]; wasClosing && !still {
83 removed = append(removed, domain.IssueLink{Issue: n, MergeRequest: mr, Closes: true})
84 }
85 }
86 for _, n := range issues {
87 _, closing := closeSet[n]
88 b.Put(issueRefKey(repo, n, mr), mustMarshal(&storagepb.IssueLink{Closes: closing}))
89 _, was := oldIssues[n]
90 _, wasClosing := oldCloses[n]
91 if !was || closing && !wasClosing {
92 added = append(added, domain.IssueLink{Issue: n, MergeRequest: mr, Closes: closing})
93 }
94 }
95 for _, link := range added {
96 if err := prepareIssueEvents(ctx, l.db, &b, repo, link.Issue, []domain.IssueEvent{{Kind: domain.IssueEventReferenced, MergeRequest: mr, Closes: link.Closes, Actor: ref.Actor, CreatedAt: ref.At}}); err != nil {
97 return nil, nil, err
98 }
99 }
100 b.Put(key, mustMarshal(&storagepb.MergeRequestReferences{Issues: issues, Closes: closes}))
101 if _, err := l.db.Apply(ctx, &b); err != nil {
102 if errors.Is(err, kv.ErrConflict) {
103 return nil, nil, domain.ErrConflict
104 }
105 return nil, nil, fmt.Errorf("replace issue links: %w", err)
106 }
107 return added, removed, nil
108}
109
110// LinksOfIssue returns links ordered by merge request number.
111func (l *IssueLinks) LinksOfIssue(ctx context.Context, repo domain.RepoName, issue int64, limit int) ([]domain.IssueLink, error) {
112 var result []domain.IssueLink
113 prefix := issueRefKeys(repo, issue)
114 for key, item := range l.db.Snapshot(ctx).Prefix(ctx, []byte(prefix)) {
115 if err := ctx.Err(); err != nil {
116 return nil, err
117 }
118 mr, err := strconv.ParseInt(strings.TrimPrefix(string(key), prefix), 10, 64)
119 if err != nil {
120 return nil, fmt.Errorf("decode issue link key: %w", err)
121 }
122 var pb storagepb.IssueLink
123 if err := proto.Unmarshal(item.Value, &pb); err != nil {
124 return nil, fmt.Errorf("decode issue link: %w", err)
125 }
126 result = append(result, domain.IssueLink{Issue: issue, MergeRequest: mr, Closes: pb.GetCloses()})
127 if limit > 0 && len(result) >= limit {
128 break
129 }
130 }
131 return result, nil
132}
133
134// LinksOfMergeRequest returns the links recorded by its last rebuild.
135func (l *IssueLinks) LinksOfMergeRequest(ctx context.Context, repo domain.RepoName, mr int64) ([]domain.IssueLink, error) {
136 item, ok := l.db.Get(ctx, mrRefKey(repo, mr))
137 if !ok {
138 return nil, nil
139 }
140 var pb storagepb.MergeRequestReferences
141 if err := proto.Unmarshal(item.Value, &pb); err != nil {
142 return nil, fmt.Errorf("decode MR references: %w", err)
143 }
144 var result []domain.IssueLink
145 for _, n := range pb.GetIssues() {
146 result = append(result, domain.IssueLink{Issue: n, MergeRequest: mr, Closes: slices.Contains(pb.GetCloses(), n)})
147 }
148 return result, nil
149}
150
151func toSet(numbers []int64) map[int64]struct{} {
152 set := make(map[int64]struct{}, len(numbers))
153 for _, n := range numbers {
154 set[n] = struct{}{}
155 }
156 return set
157}
158
159// EventsOfMergeRequest returns history, including references no longer linked,
160// ordered by time, then issue and sequence, capped at MaxIssueEvents.
161func (l *IssueLinks) EventsOfMergeRequest(ctx context.Context, repo domain.RepoName, mr int64) ([]domain.IssueEvent, error) {
162 out := make(issueEventMaxHeap, 0, MaxIssueEvents)
163 prefix := mrEventsPrefix(repo, mr)
164 for key, item := range l.db.Snapshot(ctx).Prefix(ctx, []byte(prefix)) {
165 if err := ctx.Err(); err != nil {
166 return nil, err
167 }
168 num, _, ok := strings.Cut(strings.TrimPrefix(string(key), prefix), "/")
169 if !ok {
170 return nil, fmt.Errorf("invalid MR event key %q", key)
171 }
172 issue, err := strconv.ParseInt(num, 10, 64)
173 if err != nil {
174 return nil, fmt.Errorf("decode MR event key: %w", err)
175 }
176 ev, err := decodeIssueEvent(item.Value, issue)
177 if err != nil {
178 return nil, err
179 }
180 if len(out) < MaxIssueEvents {
181 heap.Push(&out, ev)
182 } else if domain.CompareIssueEvents(ev, out[0]) < 0 {
183 out[0] = ev
184 heap.Fix(&out, 0)
185 }
186 }
187 slices.SortFunc(out, domain.CompareIssueEvents)
188 return out, ctx.Err()
189}
190
191// issueEventMaxHeap keeps the latest retained event at the root.
192type issueEventMaxHeap []domain.IssueEvent
193
194func (h issueEventMaxHeap) Len() int { return len(h) }
195func (h issueEventMaxHeap) Less(i, j int) bool { return domain.CompareIssueEvents(h[i], h[j]) > 0 }
196func (h issueEventMaxHeap) Swap(i, j int) { h[i], h[j] = h[j], h[i] }
197func (h *issueEventMaxHeap) Push(x any) { *h = append(*h, x.(domain.IssueEvent)) }
198func (h *issueEventMaxHeap) Pop() any {
199 old := *h
200 n := len(old) - 1
201 x := old[n]
202 old[n] = domain.IssueEvent{}
203 *h = old[:n]
204 return x
205}