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 issuePrefix = "is/" // is/<repo>/<number:010d>
27 issueCommentPrefix = "isc/" // isc/<repo>/<number:010d>/<id:010d>
28 issueOpenPrefix = "isopen/" // isopen/<repo>/<number:010d>, written with the issue
29)
30
31// Issues is the ermite-kv backed ports.IssueStore. It shares the database of
32// the Store it comes from.
33type Issues struct {
34 db *kv.DB
35 locks *numberLocks
36}
37
38var _ ports.IssueStore = (*Issues)(nil)
39
40// Issues returns the issue store over the same database. It is valid until the
41// Store is closed.
42func (s *Store) Issues() *Issues { return &Issues{db: s.db, locks: s.locks} }
43
44func issueRepoPrefix(repo domain.RepoName) string { return issuePrefix + repo.String() + "/" }
45func issueKey(repo domain.RepoName, n int64) []byte {
46 return []byte(issueRepoPrefix(repo) + fmt.Sprintf(numberFormat, n))
47}
48func issueCommentsPrefix(repo domain.RepoName, n int64) string {
49 return issueCommentPrefix + repo.String() + "/" + fmt.Sprintf(numberFormat, n) + "/"
50}
51func issueCommentKey(repo domain.RepoName, n, id int64) []byte {
52 return []byte(issueCommentsPrefix(repo, n) + fmt.Sprintf(numberFormat, id))
53}
54func issueOpenRepoPrefix(repo domain.RepoName) string { return issueOpenPrefix + repo.String() + "/" }
55func issueOpenKey(repo domain.RepoName, n int64) []byte {
56 return []byte(issueOpenRepoPrefix(repo) + fmt.Sprintf(numberFormat, n))
57}
58
59// indexOpenIssue adds to b the write that keeps the index of the open issues in
60// step with the issue: the key is put when it is open, deleted otherwise
61// (deleting a key that is not there is not an error).
62func indexOpenIssue(b *kv.Batch, issue domain.Issue) {
63 if issue.IsOpen() {
64 b.Put(issueOpenKey(issue.Repo, issue.Number), []byte{1})
65 return
66 }
67 b.Delete(issueOpenKey(issue.Repo, issue.Number))
68}
69
70// issueKeys lists every key of the snapshot that belongs to the repository's
71// issues.
72func issueKeys(ctx context.Context, snap *kv.Snapshot, repo domain.RepoName) [][]byte {
73 var keys [][]byte
74 for _, p := range []string{
75 issueRepoPrefix(repo), issueCommentPrefix + repo.String() + "/", issueOpenRepoPrefix(repo),
76 } {
77 for k := range snap.Prefix(ctx, []byte(p)) {
78 keys = append(keys, slices.Clone(k))
79 }
80 }
81 return keys
82}
83
84// Create takes the next number of the repository from mrseq/<repo>, the
85// sequence of the merge requests, and stores the issue under it; see
86// createNumbered. A missing repository gives domain.ErrNotFound.
87func (i *Issues) Create(ctx context.Context, issue domain.Issue) (domain.Issue, error) {
88 err := createNumbered(ctx, i.db, i.locks, issue.Repo, func(b *kv.Batch, n int64) error {
89 issue.Number = n
90 val, err := encodeIssue(issue)
91 if err != nil {
92 return err
93 }
94 b.RequireAbsent(issueKey(issue.Repo, n))
95 b.Put(issueKey(issue.Repo, n), val)
96 indexOpenIssue(b, issue)
97 return nil
98 })
99 if err != nil {
100 return domain.Issue{}, fmt.Errorf("create issue of %q: %w", issue.Repo, err)
101 }
102 return i.Get(ctx, issue.Repo, issue.Number)
103}
104
105// Get reads is/<repo>/<number>.
106func (i *Issues) Get(ctx context.Context, repo domain.RepoName, number int64) (domain.Issue, error) {
107 if number < 1 {
108 return domain.Issue{}, domain.ErrNotFound
109 }
110 item, ok := i.db.Get(ctx, issueKey(repo, number))
111 if !ok {
112 return domain.Issue{}, domain.ErrNotFound
113 }
114 return decodeIssue(repo, item)
115}
116
117// List returns a page of the issues of the repository, highest number first. The
118// page token is the last number of the previous page.
119func (i *Issues) List(ctx context.Context, repo domain.RepoName, params ports.ListIssuesParams) (ports.IssuePage, error) {
120 before := int64(0) // 0: from the highest
121 if params.PageToken != "" {
122 n, err := strconv.ParseInt(params.PageToken, 10, 64)
123 if err != nil || n < 1 {
124 return ports.IssuePage{}, fmt.Errorf("%w: page token", domain.ErrInvalid)
125 }
126 before = n
127 }
128 // The keys sort by number: walk [first, cursor) ascending, keep the last
129 // PageSize+1 matching records undecoded (the extra one tells whether
130 // another page follows), then decode only those.
131 start := []byte(issueRepoPrefix(repo))
132 end := append(slices.Clone(start[:len(start)-1]), '/'+1) // the prefix ends in '/'
133 if before > 0 {
134 end = issueKey(repo, before)
135 }
136 keep := params.PageSize + 1
137 var window []kv.Item
138 for _, item := range i.db.Snapshot(ctx).Range(ctx, start, end) {
139 if params.State != "" {
140 var pb storagepb.Issue
141 if err := proto.Unmarshal(item.Value, &pb); err != nil {
142 return ports.IssuePage{}, fmt.Errorf("decode issue: %w", err)
143 }
144 st, err := decodeIssueState(pb.GetState())
145 if err != nil {
146 return ports.IssuePage{}, fmt.Errorf("decode issue of %q: %w", repo, err)
147 }
148 if st != params.State {
149 continue
150 }
151 }
152 window = append(window, item)
153 if params.PageSize > 0 && len(window) >= 2*keep {
154 window = append(window[:0], window[len(window)-keep:]...)
155 }
156 }
157 if params.PageSize > 0 && len(window) > keep {
158 window = window[len(window)-keep:]
159 }
160 all := make([]domain.Issue, 0, len(window))
161 for _, item := range slices.Backward(window) {
162 issue, err := decodeIssue(repo, item)
163 if err != nil {
164 return ports.IssuePage{}, err
165 }
166 all = append(all, issue)
167 }
168 var page ports.IssuePage
169 if params.PageSize > 0 && len(all) > params.PageSize {
170 all = all[:params.PageSize]
171 page.NextPageToken = strconv.FormatInt(all[len(all)-1].Number, 10)
172 }
173 page.Issues = all
174 return page, nil
175}
176
177// ListOpenAll returns the open issues of every repository through the isopen/
178// index: the issues it names are read, no other.
179//
180//hexcheck:ignore suspicious-business-logic-in-adapter reads the open issues of all repositories through their index: a storage concern, no policy
181func (i *Issues) ListOpenAll(ctx context.Context) ([]domain.Issue, error) {
182 type ref struct {
183 repo domain.RepoName
184 n int64
185 }
186 var refs []ref
187 for key := range i.db.Snapshot(ctx).Prefix(ctx, []byte(issueOpenPrefix)) {
188 if err := ctx.Err(); err != nil {
189 return nil, err
190 }
191 repoName, num, ok := strings.Cut(strings.TrimPrefix(string(key), issueOpenPrefix), "/")
192 if !ok {
193 return nil, fmt.Errorf("index key %q", key)
194 }
195 repo, err := domain.ParseRepoName(repoName)
196 if err != nil {
197 return nil, fmt.Errorf("index key %q: %w", key, err)
198 }
199 n, err := strconv.ParseInt(num, 10, 64)
200 if err != nil {
201 return nil, fmt.Errorf("index key %q: %w", key, err)
202 }
203 refs = append(refs, ref{repo, n})
204 }
205 var out []domain.Issue
206 for _, r := range refs {
207 issue, err := i.Get(ctx, r.repo, r.n)
208 if errors.Is(err, domain.ErrNotFound) {
209 continue // closed or deleted since the snapshot of the index
210 }
211 if err != nil {
212 return nil, err
213 }
214 if issue.IsOpen() {
215 out = append(out, issue)
216 }
217 }
218 return out, nil
219}
220
221// Update replaces the issue with a version precondition and moves its index key
222// in the same batch. The repository must exist (see guardRepoKey), so that a
223// concurrent delete of the repository leaves no index key behind.
224func (i *Issues) Update(ctx context.Context, issue domain.Issue) error {
225 val, err := encodeIssue(issue)
226 if err != nil {
227 return err
228 }
229 key := issueKey(issue.Repo, issue.Number)
230 if _, ok := i.db.Get(ctx, key); !ok {
231 return domain.ErrNotFound
232 }
233 if issue.Version == 0 {
234 return fmt.Errorf("update issue %d of %q: read it first: %w", issue.Number, issue.Repo, domain.ErrConflict)
235 }
236 var b kv.Batch
237 if err := guardRepoKey(ctx, i.db, &b, issue.Repo); err != nil {
238 return err
239 }
240 b.RequireVersion(key, issue.Version)
241 b.Put(key, val)
242 indexOpenIssue(&b, issue)
243 if _, err := i.db.Apply(ctx, &b); err != nil {
244 if errors.Is(err, kv.ErrConflict) {
245 return domain.ErrConflict
246 }
247 return fmt.Errorf("update issue %d of %q: %w", issue.Number, issue.Repo, err)
248 }
249 return nil
250}
251
252// ListComments returns the comments of the issue, oldest first.
253func (i *Issues) ListComments(ctx context.Context, repo domain.RepoName, number int64) ([]domain.IssueComment, error) {
254 var out []domain.IssueComment
255 for key, item := range i.db.Snapshot(ctx).Prefix(ctx, []byte(issueCommentsPrefix(repo, number))) {
256 c, err := decodeIssueComment(item)
257 if err != nil {
258 return nil, fmt.Errorf("decode %s: %w", key, err)
259 }
260 out = append(out, c)
261 }
262 return out, nil
263}
264
265// SaveComment writes the comment and the issue in one batch, each with its
266// version precondition (absent for a new comment), and requires the repository
267// to exist (see guardRepoKey): domain.ErrNotFound otherwise.
268func (i *Issues) SaveComment(ctx context.Context, issue domain.Issue, c domain.IssueComment) error {
269 issueVal, err := encodeIssue(issue)
270 if err != nil {
271 return err
272 }
273 cVal, err := encodeIssueComment(c)
274 if err != nil {
275 return err
276 }
277 if issue.Version == 0 {
278 return fmt.Errorf("save comment on %d of %q: read the issue first: %w", issue.Number, issue.Repo, domain.ErrConflict)
279 }
280 var b kv.Batch
281 if err := guardRepoKey(ctx, i.db, &b, issue.Repo); err != nil {
282 return err
283 }
284 b.RequireVersion(issueKey(issue.Repo, issue.Number), issue.Version)
285 b.Put(issueKey(issue.Repo, issue.Number), issueVal)
286 indexOpenIssue(&b, issue)
287 ckey := issueCommentKey(issue.Repo, issue.Number, c.ID)
288 if c.Version == 0 {
289 b.RequireAbsent(ckey)
290 } else {
291 b.RequireVersion(ckey, c.Version)
292 }
293 b.Put(ckey, cVal)
294 if _, err := i.db.Apply(ctx, &b); err != nil {
295 if errors.Is(err, kv.ErrConflict) {
296 return domain.ErrConflict
297 }
298 return fmt.Errorf("save comment on %d of %q: %w", issue.Number, issue.Repo, err)
299 }
300 return nil
301}
302
303func encodeIssue(issue domain.Issue) ([]byte, error) {
304 if issue.Repo.IsZero() {
305 return nil, fmt.Errorf("%w: empty name", domain.ErrInvalidName)
306 }
307 if !utf8.ValidString(issue.Author) {
308 return nil, errors.New("issue author is not valid UTF-8")
309 }
310 state, err := encodeIssueState(issue.State)
311 if err != nil {
312 return nil, err
313 }
314 return mustMarshal(&storagepb.Issue{
315 Number: issue.Number,
316 Title: issue.Title.String(),
317 Body: issue.Body.String(),
318 State: state,
319 Author: issue.Author,
320 CreatedAt: encodeTime(issue.CreatedAt),
321 UpdatedAt: encodeTime(issue.UpdatedAt),
322 CommentSeq: issue.CommentSeq,
323 External: issue.External,
324 }), nil
325}
326
327func decodeIssue(repo domain.RepoName, item kv.Item) (domain.Issue, error) {
328 var pb storagepb.Issue
329 if err := proto.Unmarshal(item.Value, &pb); err != nil {
330 return domain.Issue{}, fmt.Errorf("decode issue: %w", err)
331 }
332 fail := func(err error) (domain.Issue, error) {
333 return domain.Issue{}, fmt.Errorf("decode issue %d of %q: %w", pb.GetNumber(), repo, err)
334 }
335 title, err := domain.ParseIssueTitle(pb.GetTitle())
336 if err != nil {
337 return fail(err)
338 }
339 body, err := domain.ParseIssueBody(pb.GetBody())
340 if err != nil {
341 return fail(err)
342 }
343 state, err := decodeIssueState(pb.GetState())
344 if err != nil {
345 return fail(err)
346 }
347 return domain.Issue{
348 Repo: repo, Number: pb.GetNumber(), Title: title, Body: body, State: state,
349 Author: pb.GetAuthor(),
350 CreatedAt: decodeTime(pb.GetCreatedAt()), UpdatedAt: decodeTime(pb.GetUpdatedAt()),
351 CommentSeq: pb.GetCommentSeq(),
352 External: pb.GetExternal(),
353 Version: item.Version,
354 }, nil
355}
356
357func encodeIssueComment(c domain.IssueComment) ([]byte, error) {
358 if !utf8.ValidString(c.Author) {
359 return nil, errors.New("issue comment author is not valid UTF-8")
360 }
361 return mustMarshal(&storagepb.IssueComment{
362 Id: c.ID,
363 Author: c.Author,
364 Text: c.Text.String(),
365 CreatedAt: encodeTime(c.CreatedAt),
366 }), nil
367}
368
369func decodeIssueComment(item kv.Item) (domain.IssueComment, error) {
370 var pb storagepb.IssueComment
371 if err := proto.Unmarshal(item.Value, &pb); err != nil {
372 return domain.IssueComment{}, fmt.Errorf("decode issue comment: %w", err)
373 }
374 text, err := domain.ParseIssueBody(pb.GetText())
375 if err != nil {
376 return domain.IssueComment{}, fmt.Errorf("decode issue comment %d: %w", pb.GetId(), err)
377 }
378 return domain.IssueComment{
379 ID: pb.GetId(), Author: pb.GetAuthor(), Text: text,
380 CreatedAt: decodeTime(pb.GetCreatedAt()), Version: item.Version,
381 }, nil
382}
383
384func encodeIssueState(s domain.IssueState) (storagepb.IssueState, error) {
385 switch s {
386 case domain.IssueStateOpen:
387 return storagepb.IssueState_ISSUE_STATE_OPEN, nil
388 case domain.IssueStateClosed:
389 return storagepb.IssueState_ISSUE_STATE_CLOSED, nil
390 }
391 return 0, fmt.Errorf("%w: issue state %q", domain.ErrInvalid, s)
392}
393
394func decodeIssueState(s storagepb.IssueState) (domain.IssueState, error) {
395 switch s {
396 case storagepb.IssueState_ISSUE_STATE_OPEN:
397 return domain.IssueStateOpen, nil
398 case storagepb.IssueState_ISSUE_STATE_CLOSED:
399 return domain.IssueStateClosed, nil
400 case storagepb.IssueState_ISSUE_STATE_PENDING:
401 return "", errors.New("stored issue is pending: a numbered issue is open or closed")
402 }
403 return "", fmt.Errorf("unknown issue state %d", s)
404}
405
406func encodeIssuesMode(m domain.IssuesMode) (storagepb.IssuesMode, error) {
407 switch m {
408 case "":
409 return storagepb.IssuesMode_ISSUES_MODE_UNSPECIFIED, nil
410 case domain.IssuesOff:
411 return storagepb.IssuesMode_ISSUES_MODE_OFF, nil
412 case domain.IssuesOwner:
413 return storagepb.IssuesMode_ISSUES_MODE_OWNER, nil
414 case domain.IssuesPublic:
415 return storagepb.IssuesMode_ISSUES_MODE_PUBLIC, nil
416 }
417 return 0, fmt.Errorf("%w: issues mode %q in the settings", domain.ErrInvalid, m)
418}
419
420// decodeIssuesMode reads an unspecified or unknown mode as off: the safe side.
421func decodeIssuesMode(m storagepb.IssuesMode) domain.IssuesMode {
422 switch m {
423 case storagepb.IssuesMode_ISSUES_MODE_OWNER:
424 return domain.IssuesOwner
425 case storagepb.IssuesMode_ISSUES_MODE_PUBLIC:
426 return domain.IssuesPublic
427 }
428 return domain.IssuesOff
429}