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 "unicode/utf8"
13
14 kv "bnema.dev/ermite-kv"
15 "google.golang.org/protobuf/proto"
16
17 storagepb "bnema.dev/ermite/gen/ermite/storage"
18 "bnema.dev/ermite/internal/domain"
19 "bnema.dev/ermite/internal/ports"
20)
21
22// Key layout of the issues, like that of the merge requests. The numbers come
23// from the sequence of the merge requests (mrseq/<repo>): an issue and a merge
24// request never share one.
25const (
26 issueEventPrefix = "isev/"
27 mrEventPrefix = "mrev/"
28 issuePrefix = "is/" // is/<repo>/<number:010d>
29 issueCommentPrefix = "isc/" // isc/<repo>/<number:010d>/<id:010d>
30 issueOpenPrefix = "isopen/" // isopen/<repo>/<number:010d>, written with the issue
31)
32
33// Issues is the ermite-kv backed ports.IssueStore. It shares the database of
34// the Store it comes from.
35type Issues struct {
36 db *kv.DB
37 locks *numberLocks
38}
39
40var _ ports.IssueStore = (*Issues)(nil)
41
42// Issues returns the issue store over the same database. It is valid until the
43// Store is closed.
44func (s *Store) Issues() *Issues { return &Issues{db: s.db, locks: s.locks} }
45
46func issueRepoPrefix(repo domain.RepoName) string { return issuePrefix + repo.String() + "/" }
47func issueKey(repo domain.RepoName, n int64) []byte {
48 return []byte(issueRepoPrefix(repo) + fmt.Sprintf(numberFormat, n))
49}
50func issueCommentsPrefix(repo domain.RepoName, n int64) string {
51 return issueCommentPrefix + repo.String() + "/" + fmt.Sprintf(numberFormat, n) + "/"
52}
53func issueCommentKey(repo domain.RepoName, n, id int64) []byte {
54 return []byte(issueCommentsPrefix(repo, n) + fmt.Sprintf(numberFormat, id))
55}
56func issueOpenRepoPrefix(repo domain.RepoName) string { return issueOpenPrefix + repo.String() + "/" }
57func issueOpenKey(repo domain.RepoName, n int64) []byte {
58 return []byte(issueOpenRepoPrefix(repo) + fmt.Sprintf(numberFormat, n))
59}
60
61// indexOpenIssue adds to b the write that keeps the index of the open issues in
62// step with the issue: the key is put when it is open, deleted otherwise
63// (deleting a key that is not there is not an error).
64func indexOpenIssue(b *kv.Batch, issue domain.Issue) {
65 if issue.IsOpen() {
66 b.Put(issueOpenKey(issue.Repo, issue.Number), []byte{1})
67 return
68 }
69 b.Delete(issueOpenKey(issue.Repo, issue.Number))
70}
71
72// issueKeys lists every key of the snapshot that belongs to the repository's
73// issues.
74func issueKeys(ctx context.Context, snap *kv.Snapshot, repo domain.RepoName) [][]byte {
75 var keys [][]byte
76 for _, p := range []string{
77 issueEventPrefix + repo.String() + "/", mrEventPrefix + repo.String() + "/",
78 issueRepoPrefix(repo), issueCommentPrefix + repo.String() + "/", issueOpenRepoPrefix(repo), issueRefPrefix + repo.String() + "/",
79 } {
80 for k := range snap.Prefix(ctx, []byte(p)) {
81 keys = append(keys, slices.Clone(k))
82 }
83 }
84 return keys
85}
86
87// Create takes the next number of the repository from mrseq/<repo>, the
88// sequence of the merge requests, and stores the issue under it; see
89// createNumbered. A missing repository gives domain.ErrNotFound.
90func (i *Issues) Create(ctx context.Context, issue domain.Issue) (domain.Issue, error) {
91 err := createNumbered(ctx, i.db, i.locks, issue.Repo, func(b *kv.Batch, n int64) error {
92 issue.Number = n
93 val, err := encodeIssue(issue)
94 if err != nil {
95 return err
96 }
97 b.RequireAbsent(issueKey(issue.Repo, n))
98 b.Put(issueKey(issue.Repo, n), val)
99 indexOpenIssue(b, issue)
100 return nil
101 })
102 if err != nil {
103 return domain.Issue{}, fmt.Errorf("create issue of %q: %w", issue.Repo, err)
104 }
105 return i.Get(ctx, issue.Repo, issue.Number)
106}
107
108// Get reads is/<repo>/<number>.
109func (i *Issues) Get(ctx context.Context, repo domain.RepoName, number int64) (domain.Issue, error) {
110 if number < 1 {
111 return domain.Issue{}, domain.ErrNotFound
112 }
113 item, ok := i.db.Get(ctx, issueKey(repo, number))
114 if !ok {
115 return domain.Issue{}, domain.ErrNotFound
116 }
117 return decodeIssue(repo, item)
118}
119
120// List returns a page of the issues of the repository, highest number first. The
121// page token is the last number of the previous page.
122func (i *Issues) List(ctx context.Context, repo domain.RepoName, params ports.ListIssuesParams) (ports.IssuePage, error) {
123 before := int64(0) // 0: from the highest
124 if params.PageToken != "" {
125 n, err := strconv.ParseInt(params.PageToken, 10, 64)
126 if err != nil || n < 1 {
127 return ports.IssuePage{}, fmt.Errorf("%w: page token", domain.ErrInvalid)
128 }
129 before = n
130 }
131 // The keys sort by number: walk [first, cursor) ascending, keep the last
132 // PageSize+1 matching records undecoded (the extra one tells whether
133 // another page follows), then decode only those.
134 start := []byte(issueRepoPrefix(repo))
135 end := append(slices.Clone(start[:len(start)-1]), '/'+1) // the prefix ends in '/'
136 if before > 0 {
137 end = issueKey(repo, before)
138 }
139 keep := params.PageSize + 1
140 var window []kv.Item
141 for _, item := range i.db.Snapshot(ctx).Range(ctx, start, end) {
142 if params.State != "" {
143 var pb storagepb.Issue
144 if err := proto.Unmarshal(item.Value, &pb); err != nil {
145 return ports.IssuePage{}, fmt.Errorf("decode issue: %w", err)
146 }
147 st, err := decodeIssueState(pb.GetState())
148 if err != nil {
149 return ports.IssuePage{}, fmt.Errorf("decode issue of %q: %w", repo, err)
150 }
151 if st != params.State {
152 continue
153 }
154 }
155 window = append(window, item)
156 if params.PageSize > 0 && len(window) >= 2*keep {
157 window = append(window[:0], window[len(window)-keep:]...)
158 }
159 }
160 if params.PageSize > 0 && len(window) > keep {
161 window = window[len(window)-keep:]
162 }
163 all := make([]domain.Issue, 0, len(window))
164 for _, item := range slices.Backward(window) {
165 issue, err := decodeIssue(repo, item)
166 if err != nil {
167 return ports.IssuePage{}, err
168 }
169 all = append(all, issue)
170 }
171 var page ports.IssuePage
172 if params.PageSize > 0 && len(all) > params.PageSize {
173 all = all[:params.PageSize]
174 page.NextPageToken = strconv.FormatInt(all[len(all)-1].Number, 10)
175 }
176 page.Issues = all
177 return page, nil
178}
179
180// ListOpenAll returns the open issues of every repository through the isopen/
181// index: the issues it names are read, no other.
182//
183//hexcheck:ignore suspicious-business-logic-in-adapter reads the open issues of all repositories through their index: a storage concern, no policy
184func (i *Issues) ListOpenAll(ctx context.Context) ([]domain.Issue, error) {
185 type ref struct {
186 repo domain.RepoName
187 n int64
188 }
189 var refs []ref
190 for key := range i.db.Snapshot(ctx).Prefix(ctx, []byte(issueOpenPrefix)) {
191 if err := ctx.Err(); err != nil {
192 return nil, err
193 }
194 repoName, num, ok := strings.Cut(strings.TrimPrefix(string(key), issueOpenPrefix), "/")
195 if !ok {
196 return nil, fmt.Errorf("index key %q", key)
197 }
198 repo, err := domain.ParseRepoName(repoName)
199 if err != nil {
200 return nil, fmt.Errorf("index key %q: %w", key, err)
201 }
202 n, err := strconv.ParseInt(num, 10, 64)
203 if err != nil {
204 return nil, fmt.Errorf("index key %q: %w", key, err)
205 }
206 refs = append(refs, ref{repo, n})
207 }
208 var out []domain.Issue
209 for _, r := range refs {
210 issue, err := i.Get(ctx, r.repo, r.n)
211 if errors.Is(err, domain.ErrNotFound) {
212 continue // closed or deleted since the snapshot of the index
213 }
214 if err != nil {
215 return nil, err
216 }
217 if issue.IsOpen() {
218 out = append(out, issue)
219 }
220 }
221 return out, nil
222}
223
224// Update replaces the issue with a version precondition and moves its index key
225// in the same batch. The repository must exist (see guardRepoKey), so that a
226// concurrent delete of the repository leaves no index key behind.
227func (i *Issues) Update(ctx context.Context, issue domain.Issue, events ...domain.IssueEvent) error {
228 val, err := encodeIssue(issue)
229 if err != nil {
230 return err
231 }
232 key := issueKey(issue.Repo, issue.Number)
233 if _, ok := i.db.Get(ctx, key); !ok {
234 return domain.ErrNotFound
235 }
236 if issue.Version == 0 {
237 return fmt.Errorf("update issue %d of %q: read it first: %w", issue.Number, issue.Repo, domain.ErrConflict)
238 }
239 var b kv.Batch
240 if err := guardRepoKey(ctx, i.db, &b, issue.Repo); err != nil {
241 return err
242 }
243 b.RequireVersion(key, issue.Version)
244 b.Put(key, val)
245 indexOpenIssue(&b, issue)
246 if err := prepareIssueEvents(ctx, i.db, &b, issue.Repo, issue.Number, events); err != nil {
247 return err
248 }
249 if _, err := i.db.Apply(ctx, &b); err != nil {
250 if errors.Is(err, kv.ErrConflict) {
251 return domain.ErrConflict
252 }
253 return fmt.Errorf("update issue %d of %q: %w", issue.Number, issue.Repo, err)
254 }
255 return nil
256}
257
258// ListComments returns the comments of the issue, oldest first.
259func (i *Issues) ListComments(ctx context.Context, repo domain.RepoName, number int64) ([]domain.IssueComment, error) {
260 var out []domain.IssueComment
261 for key, item := range i.db.Snapshot(ctx).Prefix(ctx, []byte(issueCommentsPrefix(repo, number))) {
262 c, err := decodeIssueComment(item)
263 if err != nil {
264 return nil, fmt.Errorf("decode %s: %w", key, err)
265 }
266 out = append(out, c)
267 }
268 return out, nil
269}
270
271// SaveComment writes the comment and the issue in one batch, each with its
272// version precondition (absent for a new comment), and requires the repository
273// to exist (see guardRepoKey): domain.ErrNotFound otherwise.
274func (i *Issues) SaveComment(ctx context.Context, issue domain.Issue, c domain.IssueComment) error {
275 issueVal, err := encodeIssue(issue)
276 if err != nil {
277 return err
278 }
279 cVal, err := encodeIssueComment(c)
280 if err != nil {
281 return err
282 }
283 if issue.Version == 0 {
284 return fmt.Errorf("save comment on %d of %q: read the issue first: %w", issue.Number, issue.Repo, domain.ErrConflict)
285 }
286 var b kv.Batch
287 if err := guardRepoKey(ctx, i.db, &b, issue.Repo); err != nil {
288 return err
289 }
290 b.RequireVersion(issueKey(issue.Repo, issue.Number), issue.Version)
291 b.Put(issueKey(issue.Repo, issue.Number), issueVal)
292 indexOpenIssue(&b, issue)
293 ckey := issueCommentKey(issue.Repo, issue.Number, c.ID)
294 if c.Version == 0 {
295 b.RequireAbsent(ckey)
296 } else {
297 b.RequireVersion(ckey, c.Version)
298 }
299 b.Put(ckey, cVal)
300 if _, err := i.db.Apply(ctx, &b); err != nil {
301 if errors.Is(err, kv.ErrConflict) {
302 return domain.ErrConflict
303 }
304 return fmt.Errorf("save comment on %d of %q: %w", issue.Number, issue.Repo, err)
305 }
306 return nil
307}
308
309func encodeIssue(issue domain.Issue) ([]byte, error) {
310 if issue.Repo.IsZero() {
311 return nil, fmt.Errorf("%w: empty name", domain.ErrInvalidName)
312 }
313 if !utf8.ValidString(issue.Author) {
314 return nil, errors.New("issue author is not valid UTF-8")
315 }
316 state, err := encodeIssueState(issue.State)
317 if err != nil {
318 return nil, err
319 }
320 return mustMarshal(&storagepb.Issue{
321 Number: issue.Number,
322 Title: issue.Title.String(),
323 Body: issue.Body.String(),
324 State: state,
325 Author: issue.Author,
326 CreatedAt: encodeTime(issue.CreatedAt),
327 UpdatedAt: encodeTime(issue.UpdatedAt),
328 CommentSeq: issue.CommentSeq,
329 External: issue.External,
330 ClosedByMergeRequest: issue.ClosedByMergeRequest,
331 }), nil
332}
333
334func decodeIssue(repo domain.RepoName, item kv.Item) (domain.Issue, error) {
335 var pb storagepb.Issue
336 if err := proto.Unmarshal(item.Value, &pb); err != nil {
337 return domain.Issue{}, fmt.Errorf("decode issue: %w", err)
338 }
339 fail := func(err error) (domain.Issue, error) {
340 return domain.Issue{}, fmt.Errorf("decode issue %d of %q: %w", pb.GetNumber(), repo, err)
341 }
342 title, err := domain.ParseIssueTitle(pb.GetTitle())
343 if err != nil {
344 return fail(err)
345 }
346 body, err := domain.ParseIssueBody(pb.GetBody())
347 if err != nil {
348 return fail(err)
349 }
350 state, err := decodeIssueState(pb.GetState())
351 if err != nil {
352 return fail(err)
353 }
354 return domain.Issue{
355 Repo: repo, Number: pb.GetNumber(), Title: title, Body: body, State: state,
356 Author: pb.GetAuthor(),
357 CreatedAt: decodeTime(pb.GetCreatedAt()), UpdatedAt: decodeTime(pb.GetUpdatedAt()),
358 CommentSeq: pb.GetCommentSeq(),
359 External: pb.GetExternal(),
360 ClosedByMergeRequest: pb.GetClosedByMergeRequest(),
361 Version: item.Version,
362 }, nil
363}
364
365func encodeIssueComment(c domain.IssueComment) ([]byte, error) {
366 if !utf8.ValidString(c.Author) {
367 return nil, errors.New("issue comment author is not valid UTF-8")
368 }
369 return mustMarshal(&storagepb.IssueComment{
370 Id: c.ID,
371 Author: c.Author,
372 Text: c.Text.String(),
373 CreatedAt: encodeTime(c.CreatedAt),
374 }), nil
375}
376
377func decodeIssueComment(item kv.Item) (domain.IssueComment, error) {
378 var pb storagepb.IssueComment
379 if err := proto.Unmarshal(item.Value, &pb); err != nil {
380 return domain.IssueComment{}, fmt.Errorf("decode issue comment: %w", err)
381 }
382 text, err := domain.ParseIssueBody(pb.GetText())
383 if err != nil {
384 return domain.IssueComment{}, fmt.Errorf("decode issue comment %d: %w", pb.GetId(), err)
385 }
386 return domain.IssueComment{
387 ID: pb.GetId(), Author: pb.GetAuthor(), Text: text,
388 CreatedAt: decodeTime(pb.GetCreatedAt()), Version: item.Version,
389 }, nil
390}
391
392func encodeIssueState(s domain.IssueState) (storagepb.IssueState, error) {
393 switch s {
394 case domain.IssueStateOpen:
395 return storagepb.IssueState_ISSUE_STATE_OPEN, nil
396 case domain.IssueStateResolved:
397 return storagepb.IssueState_ISSUE_STATE_RESOLVED, nil
398 case domain.IssueStateClosed:
399 return storagepb.IssueState_ISSUE_STATE_CLOSED, nil
400 }
401 return 0, fmt.Errorf("%w: issue state %q", domain.ErrInvalid, s)
402}
403
404func decodeIssueState(s storagepb.IssueState) (domain.IssueState, error) {
405 switch s {
406 case storagepb.IssueState_ISSUE_STATE_OPEN:
407 return domain.IssueStateOpen, nil
408 case storagepb.IssueState_ISSUE_STATE_RESOLVED:
409 return domain.IssueStateResolved, nil
410 case storagepb.IssueState_ISSUE_STATE_CLOSED:
411 return domain.IssueStateClosed, nil
412 case storagepb.IssueState_ISSUE_STATE_PENDING:
413 return "", errors.New("stored issue is pending: a numbered issue is open or closed")
414 }
415 return "", fmt.Errorf("unknown issue state %d", s)
416}
417
418func encodeIssuesMode(m domain.IssuesMode) (storagepb.IssuesMode, error) {
419 switch m {
420 case "":
421 return storagepb.IssuesMode_ISSUES_MODE_UNSPECIFIED, nil
422 case domain.IssuesOff:
423 return storagepb.IssuesMode_ISSUES_MODE_OFF, nil
424 case domain.IssuesOwner:
425 return storagepb.IssuesMode_ISSUES_MODE_OWNER, nil
426 case domain.IssuesPublic:
427 return storagepb.IssuesMode_ISSUES_MODE_PUBLIC, nil
428 }
429 return 0, fmt.Errorf("%w: issues mode %q in the settings", domain.ErrInvalid, m)
430}
431
432// decodeIssuesMode reads an unspecified or unknown mode as off: the safe side.
433func decodeIssuesMode(m storagepb.IssuesMode) domain.IssuesMode {
434 switch m {
435 case storagepb.IssuesMode_ISSUES_MODE_OWNER:
436 return domain.IssuesOwner
437 case storagepb.IssuesMode_ISSUES_MODE_PUBLIC:
438 return domain.IssuesPublic
439 }
440 return domain.IssuesOff
441}
442
443// MaxIssueEvents caps the append-only history of each issue. Further events
444// are silently dropped without preventing the associated write.
445const MaxIssueEvents = 500
446
447// ListEvents returns the append-only issue history in sequence order.
448func (i *Issues) ListEvents(ctx context.Context, repo domain.RepoName, number int64) ([]domain.IssueEvent, error) {
449 var out []domain.IssueEvent
450 for _, item := range i.db.Snapshot(ctx).Prefix(ctx, []byte(issueEventsPrefix(repo, number))) {
451 if err := ctx.Err(); err != nil {
452 return nil, err
453 }
454 ev, err := decodeIssueEvent(item.Value, number)
455 if err != nil {
456 return nil, err
457 }
458 out = append(out, ev)
459 if len(out) == MaxIssueEvents {
460 break
461 }
462 }
463 return out, ctx.Err()
464}
465
466func issueEventsPrefix(repo domain.RepoName, issue int64) string {
467 return issueEventPrefix + repo.String() + "/" + fmt.Sprintf(numberFormat, issue) + "/"
468}
469
470// prepareIssueEvents allocates dense per-issue sequences. RequireAbsent makes
471// concurrent appenders conflict; callers retry the entire associated write.
472// Scan once for a group so multiple events in one batch get distinct sequences.
473func prepareIssueEvents(ctx context.Context, db *kv.DB, b *kv.Batch, repo domain.RepoName, issue int64, events []domain.IssueEvent) error {
474 if len(events) == 0 {
475 return nil
476 }
477 prefix := issueEventsPrefix(repo, issue)
478 var seq int64
479 count := 0
480 for key := range db.Snapshot(ctx).Prefix(ctx, []byte(prefix)) {
481 if err := ctx.Err(); err != nil {
482 return err
483 }
484 n, err := strconv.ParseInt(strings.TrimPrefix(string(key), prefix), 10, 64)
485 if err != nil {
486 return fmt.Errorf("decode issue event key: %w", err)
487 }
488 seq = max(seq, n)
489 count++
490 }
491 for _, ev := range events {
492 if count >= MaxIssueEvents {
493 break
494 }
495 seq++
496 ev.ID = seq
497 val, err := encodeIssueEvent(ev)
498 if err != nil {
499 return err
500 }
501 key := []byte(prefix + fmt.Sprintf(numberFormat, seq))
502 b.RequireAbsent(key)
503 b.Put(key, val)
504 if ev.MergeRequest != 0 {
505 b.Put([]byte(mrEventsPrefix(repo, ev.MergeRequest)+fmt.Sprintf(numberFormat, issue)+"/"+fmt.Sprintf(numberFormat, seq)), val)
506 }
507 count++
508 }
509 return ctx.Err()
510}
511
512func mrEventsPrefix(repo domain.RepoName, mr int64) string {
513 return mrEventPrefix + repo.String() + "/" + fmt.Sprintf(numberFormat, mr) + "/"
514}
515
516func encodeIssueEvent(ev domain.IssueEvent) ([]byte, error) {
517 var kind storagepb.IssueEventKind
518 switch ev.Kind {
519 case domain.IssueEventReferenced:
520 kind = storagepb.IssueEventKind_ISSUE_EVENT_KIND_REFERENCED
521 case domain.IssueEventClosed:
522 kind = storagepb.IssueEventKind_ISSUE_EVENT_KIND_CLOSED
523 case domain.IssueEventReopened:
524 kind = storagepb.IssueEventKind_ISSUE_EVENT_KIND_REOPENED
525 case domain.IssueEventResolved:
526 kind = storagepb.IssueEventKind_ISSUE_EVENT_KIND_RESOLVED
527 default:
528 return nil, fmt.Errorf("%w: issue event kind %q", domain.ErrInvalid, ev.Kind)
529 }
530 if !utf8.ValidString(ev.Actor) {
531 return nil, errors.New("issue event actor is not valid UTF-8")
532 }
533 return mustMarshal(&storagepb.IssueEvent{Seq: ev.ID, Kind: kind, MergeRequest: ev.MergeRequest, Closes: ev.Closes, Actor: ev.Actor, CreatedAt: encodeTime(ev.CreatedAt)}), nil
534}
535
536func decodeIssueEvent(val []byte, issue int64) (domain.IssueEvent, error) {
537 var pb storagepb.IssueEvent
538 if err := proto.Unmarshal(val, &pb); err != nil {
539 return domain.IssueEvent{}, fmt.Errorf("decode issue event: %w", err)
540 }
541 var kind domain.IssueEventKind
542 switch pb.GetKind() {
543 case storagepb.IssueEventKind_ISSUE_EVENT_KIND_REFERENCED:
544 kind = domain.IssueEventReferenced
545 case storagepb.IssueEventKind_ISSUE_EVENT_KIND_CLOSED:
546 kind = domain.IssueEventClosed
547 case storagepb.IssueEventKind_ISSUE_EVENT_KIND_REOPENED:
548 kind = domain.IssueEventReopened
549 case storagepb.IssueEventKind_ISSUE_EVENT_KIND_RESOLVED:
550 kind = domain.IssueEventResolved
551 default:
552 return domain.IssueEvent{}, fmt.Errorf("unknown issue event kind %d", pb.GetKind())
553 }
554 return domain.IssueEvent{ID: pb.GetSeq(), Issue: issue, Kind: kind, MergeRequest: pb.GetMergeRequest(), Closes: pb.GetCloses(), Actor: pb.GetActor(), CreatedAt: decodeTime(pb.GetCreatedAt())}, nil
555}