1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package usecase
4
5import (
6 "context"
7 "errors"
8 "slices"
9 "strings"
10 "sync"
11 "time"
12
13 "bnema.dev/ermite/internal/domain"
14 "bnema.dev/ermite/internal/ports"
15 "github.com/bnema/zerowrap"
16)
17
18const MaxIssueLinkCommits = ports.MaxMergeRequestCommits
19const MaxIssueLinksPerIssue = 100
20const maxLinkedMRsPerPush = 50
21
22type IssueLinksConfig struct {
23 Meta ports.RepoMetaStore
24 Issues ports.IssueStore
25 MRs ports.MergeRequestStore
26 Links ports.IssueLinkStore
27 Git ports.MergeGit
28 Bus ports.EventBus
29 Clock ports.Clock
30}
31
32// IssueLinks maintains the derived index and applies closing references on merge.
33type IssueLinks struct {
34 cfg IssueLinksConfig
35 mu sync.Mutex
36 locks map[linkKey]*linkLock
37}
38
39type linkKey struct {
40 repo domain.RepoName
41 number int64
42}
43
44type linkLock struct {
45 ch chan struct{}
46 refs int
47}
48
49// lockMR serializes reindexes of one merge request, waiting until ctx ends.
50// Entries are removed when no one holds or waits for them.
51func (l *IssueLinks) lockMR(ctx context.Context, repo domain.RepoName, number int64) (func(), error) {
52 key := linkKey{repo, number}
53 l.mu.Lock()
54 e, ok := l.locks[key]
55 if !ok {
56 e = &linkLock{ch: make(chan struct{}, 1)}
57 l.locks[key] = e
58 }
59 e.refs++
60 l.mu.Unlock()
61 release := func() {
62 l.mu.Lock()
63 if e.refs--; e.refs == 0 {
64 delete(l.locks, key)
65 }
66 l.mu.Unlock()
67 }
68 select {
69 case e.ch <- struct{}{}:
70 return func() { <-e.ch; release() }, nil
71 case <-ctx.Done():
72 release()
73 return nil, ctx.Err()
74 }
75}
76
77var _ ports.IssueLinkTrigger = (*IssueLinks)(nil)
78
79func NewIssueLinks(cfg IssueLinksConfig) *IssueLinks {
80 return &IssueLinks{cfg: cfg, locks: map[linkKey]*linkLock{}}
81}
82
83// Reindex replaces links from the current texts and bounded commit range.
84func (l *IssueLinks) Reindex(ctx context.Context, repo domain.Repo, number int64) error {
85 _, err := l.reindex(ctx, repo, number)
86 return err
87}
88func (l *IssueLinks) reindex(ctx context.Context, repo domain.Repo, number int64) ([]int64, error) {
89 unlock, err := l.lockMR(ctx, repo.Name, number)
90 if err != nil {
91 return nil, err
92 }
93 defer unlock()
94 mr, err := l.cfg.MRs.Get(ctx, repo.Name, number)
95 if err != nil {
96 return nil, err
97 }
98 notes, err := l.cfg.MRs.ListComments(ctx, repo.Name, number)
99 if err != nil {
100 return nil, err
101 }
102 var texts, messages []string
103 for _, note := range notes {
104 if !note.Deleted {
105 texts = append(texts, note.Text.String())
106 }
107 }
108 if mr.IsOpen() {
109 mr.HeadID, err = l.cfg.Git.BranchHead(ctx, repo.Name, mr.Source)
110 if errors.Is(err, domain.ErrNotFound) {
111 mr.HeadID = domain.ObjectID{}
112 } else if err != nil {
113 return nil, err
114 }
115 }
116 exclude, _, err := mrRange(ctx, l.cfg.Git, mr)
117 if err != nil {
118 return nil, err
119 }
120 has := !mr.HeadID.IsZero()
121 if has && !mr.IsOpen() {
122 has, err = l.cfg.Git.HasCommit(ctx, repo.Name, mr.HeadID)
123 if err != nil {
124 return nil, err
125 }
126 }
127 if has {
128 commits, more, err := l.cfg.Git.ListRange(ctx, repo.Name, exclude, mr.HeadID, MaxIssueLinkCommits)
129 if err != nil {
130 return nil, err
131 }
132 if more {
133 log := zerowrap.FromCtx(ctx)
134 log.Debug().Msg("issue link commit range truncated")
135 }
136 for _, commit := range commits {
137 messages = append(messages, commit.Message)
138 }
139 }
140 refs := domain.ScanMergeRequest(mr.Title.String(), mr.Description.String(), texts, messages)
141 var mentions, closes []int64
142 for _, n := range refs.Mentions {
143 _, err := l.cfg.Issues.Get(ctx, repo.Name, n)
144 if errors.Is(err, domain.ErrNotFound) {
145 continue
146 }
147 if err != nil {
148 return nil, err
149 }
150 mentions = append(mentions, n)
151 if slices.Contains(refs.Closes, n) {
152 closes = append(closes, n)
153 }
154 }
155 for range maxWriteRetries {
156 added, removed, err := l.cfg.Links.ReplaceLinks(ctx, repo.Name, number, mentions, closes)
157 if errors.Is(err, domain.ErrConflict) {
158 continue
159 }
160 if err != nil {
161 return nil, err
162 }
163 type pair struct{ issue, mr int64 }
164 sent := map[pair]struct{}{}
165 for _, link := range slices.Concat(added, removed) {
166 p := pair{link.Issue, link.MergeRequest}
167 if _, ok := sent[p]; ok {
168 continue
169 }
170 sent[p] = struct{}{}
171 l.notify(ctx, repo, link)
172 }
173 return closes, nil
174 }
175 return nil, domain.ErrConflict
176}
177func (l *IssueLinks) notify(ctx context.Context, repo domain.Repo, link domain.IssueLink) {
178 l.cfg.Bus.Publish(ctx, domain.IssueReferenced{Repo: repo.Name, Issue: link.Issue, MergeRequest: link.MergeRequest, At: l.cfg.Clock.Now().UTC(), Visibility: repo.Visibility, CreatedAt: repo.CreatedAt})
179}
180func linkContext(ctx context.Context) (context.Context, context.CancelFunc) {
181 return context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
182}
183func linkWarning(ctx context.Context, err error) {
184 if err != nil {
185 log := zerowrap.FromCtx(ctx)
186 log.Warn().Err(err).Msg("issue links not updated")
187 }
188}
189func (l *IssueLinks) OnMergeRequestChanged(ctx context.Context, repo domain.Repo, number int64) {
190 ctx, cancel := linkContext(ctx)
191 defer cancel()
192 linkWarning(ctx, l.Reindex(ctx, repo, number))
193}
194func (l *IssueLinks) settled(ctx context.Context, repo domain.Repo, mr domain.MergeRequest) {
195 links, err := l.cfg.Links.LinksOfMergeRequest(ctx, repo.Name, mr.Number)
196 linkWarning(ctx, err)
197 if err == nil {
198 for _, link := range links {
199 l.notify(ctx, repo, link)
200 }
201 }
202}
203func (l *IssueLinks) OnMergeRequestSettled(ctx context.Context, repo domain.Repo, mr domain.MergeRequest) {
204 ctx, cancel := linkContext(ctx)
205 defer cancel()
206 l.settled(ctx, repo, mr)
207}
208func (l *IssueLinks) OnBranchesMoved(ctx context.Context, name domain.RepoName, updates []domain.RefUpdate) {
209 ctx, cancel := linkContext(ctx)
210 defer cancel()
211 var branches []string
212 for _, up := range updates {
213 if b, ok := strings.CutPrefix(up.Ref, "refs/heads/"); ok && !up.IsDelete() {
214 branches = append(branches, b)
215 }
216 }
217 if len(branches) == 0 {
218 return
219 }
220 repo, err := l.cfg.Meta.Get(ctx, name)
221 if err != nil {
222 linkWarning(ctx, err)
223 return
224 }
225 numbers, truncated, err := l.cfg.MRs.OpenBySource(ctx, name, branches, maxLinkedMRsPerPush)
226 if err != nil {
227 linkWarning(ctx, err)
228 return
229 }
230 if truncated {
231 log := zerowrap.FromCtx(ctx)
232 log.Warn().Str("repo", name.String()).Msg("issue link merge requests truncated")
233 }
234 for _, n := range numbers {
235 linkWarning(ctx, l.Reindex(ctx, repo, n))
236 }
237}
238func (l *IssueLinks) OnMerged(ctx context.Context, p domain.Principal, repo domain.Repo, mr domain.MergeRequest) {
239 ctx, cancel := linkContext(ctx)
240 defer cancel()
241 defer l.settled(ctx, repo, mr)
242 closes, err := l.reindex(ctx, repo, mr.Number)
243 if err != nil {
244 linkWarning(ctx, err)
245 return
246 }
247 if !repo.Settings.IssuesEnabled() {
248 log := zerowrap.FromCtx(ctx)
249 log.Debug().Msg("merge issue closing disabled")
250 return
251 }
252 for _, n := range closes {
253 err := l.close(ctx, p, repo, mr.Number, n)
254 if err != nil {
255 log := zerowrap.FromCtx(ctx)
256 log.Warn().Err(err).Str("repo", repo.Name.String()).Int64("number", n).Int64("merge_request", mr.Number).Msg("issue not closed by the merge")
257 }
258 }
259}
260func (l *IssueLinks) close(ctx context.Context, p domain.Principal, repo domain.Repo, mr, n int64) error {
261 for range maxWriteRetries {
262 issue, err := l.cfg.Issues.Get(ctx, repo.Name, n)
263 if err != nil {
264 return err
265 }
266 if !issue.IsOpen() {
267 return nil
268 }
269 issue.State = domain.IssueStateClosed
270 issue.ClosedByMergeRequest = mr
271 issue.UpdatedAt = l.cfg.Clock.Now().UTC()
272 err = l.cfg.Issues.Update(ctx, issue)
273 if errors.Is(err, domain.ErrConflict) {
274 continue
275 }
276 if err != nil {
277 return err
278 }
279 recordActivity(ctx, l.cfg.Meta, repo.Name, issue.UpdatedAt)
280 log := zerowrap.FromCtx(ctx)
281 log.Info().Str("repo", repo.Name.String()).Int64("number", n).Int64("merge_request", mr).Str("principal", p.Label()).Str("token_id", p.TokenID()).Msg("issue closed by merge")
282 l.cfg.Bus.Publish(ctx, domain.IssueClosed{Issue: issue, At: issue.UpdatedAt, Visibility: repo.Visibility, CreatedAt: repo.CreatedAt})
283 return nil
284 }
285 return domain.ErrConflict
286}