1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package kvstore
4
5import (
6 "context"
7 "crypto/rand"
8 "encoding/hex"
9 "errors"
10 "fmt"
11 "slices"
12 "strconv"
13 "strings"
14 "time"
15 "unicode/utf8"
16
17 kv "bnema.dev/ermite-kv"
18 "github.com/bnema/zerowrap"
19 "google.golang.org/protobuf/proto"
20
21 storagepb "bnema.dev/ermite/gen/ermite/storage"
22 "bnema.dev/ermite/internal/domain"
23 "bnema.dev/ermite/internal/ports"
24)
25
26// Key layout of the public issue form.
27const (
28 pendingPrefix = "ispend/" // ispend/<repo>/<id>
29 banPrefix = "isban/" // isban/<hex of the hash>
30 issueKeyKey = "meta/issuekey"
31 issueKeyLen = 32
32)
33
34// PendingIssues is the ermite-kv backed ports.PendingIssueStore. It shares the
35// database of the Store it comes from.
36type PendingIssues struct {
37 db *kv.DB
38 locks *numberLocks
39}
40
41var _ ports.PendingIssueStore = (*PendingIssues)(nil)
42
43// PendingIssues returns the pending issue store over the same database. It is
44// valid until the Store is closed.
45func (s *Store) PendingIssues() *PendingIssues { return &PendingIssues{db: s.db, locks: s.locks} }
46
47func pendingRepoPrefix(repo domain.RepoName) string { return pendingPrefix + repo.String() + "/" }
48func pendingKey(repo domain.RepoName, id domain.PendingIssueID) []byte {
49 return []byte(pendingRepoPrefix(repo) + id.String())
50}
51
52// pendingKeys lists every key of the snapshot that holds a pending issue of
53// the repository.
54func pendingKeys(ctx context.Context, snap *kv.Snapshot, repo domain.RepoName) [][]byte {
55 var keys [][]byte
56 for k := range snap.Prefix(ctx, []byte(pendingRepoPrefix(repo))) {
57 keys = append(keys, slices.Clone(k))
58 }
59 return keys
60}
61
62// Put stores the pending issue under a new key. The repository must exist (see
63// guardRepoKey): domain.ErrNotFound otherwise.
64func (s *PendingIssues) Put(ctx context.Context, p domain.PendingIssue) error {
65 val, err := encodePending(p)
66 if err != nil {
67 return err
68 }
69 var b kv.Batch
70 if err := guardRepoKey(ctx, s.db, &b, p.Repo); err != nil {
71 return err
72 }
73 b.RequireAbsent(pendingKey(p.Repo, p.ID))
74 b.Put(pendingKey(p.Repo, p.ID), val)
75 if _, err := s.db.Apply(ctx, &b); err != nil {
76 if errors.Is(err, kv.ErrConflict) {
77 return domain.ErrConflict
78 }
79 return fmt.Errorf("store pending issue of %q: %w", p.Repo, err)
80 }
81 return nil
82}
83
84// Get reads ispend/<repo>/<id>.
85func (s *PendingIssues) Get(ctx context.Context, repo domain.RepoName, id domain.PendingIssueID) (domain.PendingIssue, error) {
86 item, ok := s.db.Get(ctx, pendingKey(repo, id))
87 if !ok {
88 return domain.PendingIssue{}, domain.ErrNotFound
89 }
90 return decodePending(repo, id, item)
91}
92
93// List reads the pending issues of the repository, or of all, sorted by
94// creation time then repository then id. There are at most
95// domain.MaxPendingIssues of them (a little more under a race), so they are
96// read and sorted. The page token is the position of the last item returned.
97//
98//hexcheck:ignore suspicious-business-logic-in-adapter pages through the keys of a prefix: a storage concern, no policy
99func (s *PendingIssues) List(ctx context.Context, repo domain.RepoName, page ports.ListPendingParams) (ports.PendingIssuePage, error) {
100 prefix := pendingPrefix
101 if !repo.IsZero() {
102 prefix = pendingRepoPrefix(repo)
103 }
104 var after pendingCursor
105 if page.PageToken != "" {
106 var err error
107 if after, err = parsePendingCursor(page.PageToken); err != nil {
108 return ports.PendingIssuePage{}, err
109 }
110 }
111 var all []domain.PendingIssue
112 for key, item := range s.db.Snapshot(ctx).Prefix(ctx, []byte(prefix)) {
113 if err := ctx.Err(); err != nil {
114 return ports.PendingIssuePage{}, err
115 }
116 r, id, err := splitPendingKey(string(key))
117 if err != nil {
118 return ports.PendingIssuePage{}, err
119 }
120 p, err := decodePending(r, id, item)
121 if err != nil {
122 return ports.PendingIssuePage{}, fmt.Errorf("decode %s: %w", key, err)
123 }
124 all = append(all, p)
125 }
126 slices.SortFunc(all, comparePending)
127 if page.PageToken != "" {
128 all = slices.DeleteFunc(all, func(p domain.PendingIssue) bool { return comparePending(p, after.item()) <= 0 })
129 }
130 var out ports.PendingIssuePage
131 if page.PageSize > 0 && len(all) > page.PageSize {
132 all = all[:page.PageSize]
133 out.NextPageToken = encodePendingCursor(all[len(all)-1])
134 }
135 out.Issues = all
136 return out, nil
137}
138
139func comparePending(a, b domain.PendingIssue) int {
140 if c := a.CreatedAt.Compare(b.CreatedAt); c != 0 {
141 return c
142 }
143 if c := strings.Compare(a.Repo.String(), b.Repo.String()); c != 0 {
144 return c
145 }
146 return strings.Compare(a.ID.String(), b.ID.String())
147}
148
149type pendingCursor struct {
150 nanos int64
151 repo domain.RepoName
152 id domain.PendingIssueID
153}
154
155func (c pendingCursor) item() domain.PendingIssue {
156 return domain.PendingIssue{Repo: c.repo, ID: c.id, CreatedAt: time.Unix(0, c.nanos)}
157}
158
159func encodePendingCursor(p domain.PendingIssue) string {
160 return strconv.FormatInt(p.CreatedAt.UnixNano(), 10) + "/" + p.Repo.String() + "/" + p.ID.String()
161}
162
163func parsePendingCursor(s string) (pendingCursor, error) {
164 bad := fmt.Errorf("%w: page token", domain.ErrInvalid)
165 parts := strings.Split(s, "/")
166 if len(parts) != 3 {
167 return pendingCursor{}, bad
168 }
169 n, err := strconv.ParseInt(parts[0], 10, 64)
170 if err != nil {
171 return pendingCursor{}, bad
172 }
173 repo, err := domain.ParseRepoName(parts[1])
174 if err != nil {
175 return pendingCursor{}, bad
176 }
177 id, err := domain.ParsePendingIssueID(parts[2])
178 if err != nil {
179 return pendingCursor{}, bad
180 }
181 return pendingCursor{nanos: n, repo: repo, id: id}, nil
182}
183
184func splitPendingKey(key string) (domain.RepoName, domain.PendingIssueID, error) {
185 rest := strings.TrimPrefix(key, pendingPrefix)
186 repoName, idStr, ok := strings.Cut(rest, "/")
187 if !ok {
188 return domain.RepoName{}, "", fmt.Errorf("pending issue key %q", key)
189 }
190 repo, err := domain.ParseRepoName(repoName)
191 if err != nil {
192 return domain.RepoName{}, "", fmt.Errorf("pending issue key %q: %w", key, err)
193 }
194 id, err := domain.ParsePendingIssueID(idStr)
195 if err != nil {
196 return domain.RepoName{}, "", fmt.Errorf("pending issue key %q: %w", key, err)
197 }
198 return repo, id, nil
199}
200
201// Count counts the keys under ispend/, stopping after MaxPendingIssues+1.
202func (s *PendingIssues) Count(ctx context.Context) (int, error) {
203 n := 0
204 for range s.db.Snapshot(ctx).Prefix(ctx, []byte(pendingPrefix)) {
205 if err := ctx.Err(); err != nil {
206 return 0, err
207 }
208 n++
209 if n > domain.MaxPendingIssues {
210 break
211 }
212 }
213 return n, nil
214}
215
216// Approve numbers the pending issue from mrseq/<repo> (see createNumbered),
217// writes the open issue and its index key and deletes the pending record, in
218// one batch that requires the pending record to be unchanged.
219//
220//hexcheck:ignore suspicious-business-logic-in-adapter only sets the number under a version precondition; the issue itself is built by domain.ApprovedIssue
221func (s *PendingIssues) Approve(ctx context.Context, repo domain.RepoName, id domain.PendingIssueID, now time.Time) (domain.Issue, error) {
222 var number int64
223 err := createNumbered(ctx, s.db, s.locks, repo, func(b *kv.Batch, n int64) error {
224 key := pendingKey(repo, id)
225 item, ok := s.db.Get(ctx, key)
226 if !ok {
227 return domain.ErrNotFound
228 }
229 p, err := decodePending(repo, id, item)
230 if err != nil {
231 return err
232 }
233 issue := domain.ApprovedIssue(p, now)
234 issue.Repo, issue.Number = repo, n
235 val, err := encodeIssue(issue)
236 if err != nil {
237 return err
238 }
239 b.RequireVersion(key, item.Version)
240 b.Delete(key)
241 b.RequireAbsent(issueKey(repo, n))
242 b.Put(issueKey(repo, n), val)
243 indexOpenIssue(b, issue)
244 number = n
245 return nil
246 })
247 if err != nil {
248 if errors.Is(err, domain.ErrNotFound) {
249 return domain.Issue{}, domain.ErrNotFound
250 }
251 return domain.Issue{}, fmt.Errorf("approve pending issue of %q: %w", repo, err)
252 }
253 return (&Issues{db: s.db, locks: s.locks}).Get(ctx, repo, number)
254}
255
256// Delete removes the pending record, or returns domain.ErrNotFound.
257func (s *PendingIssues) Delete(ctx context.Context, repo domain.RepoName, id domain.PendingIssueID) error {
258 key := pendingKey(repo, id)
259 item, ok := s.db.Get(ctx, key)
260 if !ok {
261 return domain.ErrNotFound
262 }
263 var b kv.Batch
264 b.RequireVersion(key, item.Version)
265 b.Delete(key)
266 if _, err := s.db.Apply(ctx, &b); err != nil {
267 if errors.Is(err, kv.ErrConflict) {
268 return domain.ErrNotFound // gone or approved meanwhile
269 }
270 return fmt.Errorf("delete pending issue of %q: %w", repo, err)
271 }
272 return nil
273}
274
275func encodePending(p domain.PendingIssue) ([]byte, error) {
276 if p.Repo.IsZero() {
277 return nil, fmt.Errorf("%w: empty name", domain.ErrInvalidName)
278 }
279 if _, err := domain.ParsePendingIssueID(p.ID.String()); err != nil {
280 return nil, err
281 }
282 if !utf8.ValidString(p.Author) {
283 return nil, errors.New("pending issue author is not valid UTF-8")
284 }
285 if len(p.IPHash) != domain.IPHashLen {
286 return nil, fmt.Errorf("%w: ip hash is %d bytes, want %d", domain.ErrInvalid, len(p.IPHash), domain.IPHashLen)
287 }
288 return mustMarshal(&storagepb.PendingIssue{
289 Title: p.Title.String(), Body: p.Body.String(), Author: p.Author,
290 CreatedAt: encodeTime(p.CreatedAt), IpHash: p.IPHash,
291 }), nil
292}
293
294//hexcheck:ignore suspicious-business-logic-in-adapter schema decoding, not a policy
295func decodePending(repo domain.RepoName, id domain.PendingIssueID, item kv.Item) (domain.PendingIssue, error) {
296 var pb storagepb.PendingIssue
297 if err := proto.Unmarshal(item.Value, &pb); err != nil {
298 return domain.PendingIssue{}, fmt.Errorf("decode pending issue: %w", err)
299 }
300 title, err := domain.ParseIssueTitle(pb.GetTitle())
301 if err != nil {
302 return domain.PendingIssue{}, fmt.Errorf("decode pending issue %s of %q: %w", id, repo, err)
303 }
304 body, err := domain.ParseIssueBody(pb.GetBody())
305 if err != nil {
306 return domain.PendingIssue{}, fmt.Errorf("decode pending issue %s of %q: %w", id, repo, err)
307 }
308 return domain.PendingIssue{
309 ID: id, Repo: repo, Title: title, Body: body, Author: pb.GetAuthor(),
310 CreatedAt: decodeTime(pb.GetCreatedAt()), IPHash: slices.Clone(pb.GetIpHash()), Version: item.Version,
311 }, nil
312}
313
314// IssueKeys is the ermite-kv backed ports.IssueKeyStore.
315type IssueKeys struct {
316 db *kv.DB
317}
318
319var _ ports.IssueKeyStore = (*IssueKeys)(nil)
320
321// IssueKeys returns the issue key store over the same database. It is valid
322// until the Store is closed.
323func (s *Store) IssueKeys() *IssueKeys { return &IssueKeys{db: s.db} }
324
325// Key reads meta/issuekey and generates it when absent. The key is never
326// logged.
327func (k *IssueKeys) Key(ctx context.Context) ([]byte, error) {
328 for range updateAttempts {
329 if item, ok := k.db.Get(ctx, []byte(issueKeyKey)); ok {
330 var m storagepb.IssueKey
331 if err := proto.Unmarshal(item.Value, &m); err != nil {
332 return nil, fmt.Errorf("decode issue key: %w", err)
333 }
334 if len(m.GetKey()) != issueKeyLen {
335 return nil, fmt.Errorf("decode issue key: %d bytes, want %d", len(m.GetKey()), issueKeyLen)
336 }
337 return append([]byte(nil), m.GetKey()...), nil
338 }
339 key := make([]byte, issueKeyLen)
340 if _, err := rand.Read(key); err != nil {
341 return nil, fmt.Errorf("generate issue key: %w", err)
342 }
343 var b kv.Batch
344 b.RequireAbsent([]byte(issueKeyKey))
345 b.Put([]byte(issueKeyKey), mustMarshal(&storagepb.IssueKey{Key: key}))
346 _, err := k.db.Apply(ctx, &b)
347 if errors.Is(err, kv.ErrConflict) {
348 continue // written meanwhile: read it
349 }
350 if err != nil {
351 return nil, fmt.Errorf("store issue key: %w", err)
352 }
353 log := zerowrap.FromCtx(ctx)
354 log.Info().Msg("issue key generated")
355 return key, nil
356 }
357 return nil, fmt.Errorf("store issue key: %w", domain.ErrConflict)
358}
359
360// IssueBans is the ermite-kv backed ports.IssueBanStore.
361type IssueBans struct {
362 db *kv.DB
363}
364
365var _ ports.IssueBanStore = (*IssueBans)(nil)
366
367// IssueBans returns the ban store over the same database. It is valid until
368// the Store is closed.
369func (s *Store) IssueBans() *IssueBans { return &IssueBans{db: s.db} }
370
371func banKey(hash []byte) []byte { return []byte(banPrefix + hex.EncodeToString(hash)) }
372
373// Ban writes isban/<hex hash> with its expiry.
374func (s *IssueBans) Ban(ctx context.Context, hash []byte, until time.Time) error {
375 if len(hash) != domain.IPHashLen {
376 return fmt.Errorf("%w: ban hash is %d bytes, want %d", domain.ErrInvalid, len(hash), domain.IPHashLen)
377 }
378 var b kv.Batch
379 b.Put(banKey(hash), mustMarshal(&storagepb.IssueBan{Until: encodeTime(until)}))
380 if _, err := s.db.Apply(ctx, &b); err != nil {
381 return fmt.Errorf("store ban: %w", err)
382 }
383 return nil
384}
385
386// Banned reads the ban; one that expired is not a ban and is deleted (if it
387// has not been replaced meanwhile).
388func (s *IssueBans) Banned(ctx context.Context, hash []byte, now time.Time) (bool, error) {
389 key := banKey(hash)
390 item, ok := s.db.Get(ctx, key)
391 if !ok {
392 return false, nil
393 }
394 var pb storagepb.IssueBan
395 if err := proto.Unmarshal(item.Value, &pb); err != nil {
396 return false, fmt.Errorf("decode ban: %w", err)
397 }
398 if now.Before(decodeTime(pb.GetUntil())) {
399 return true, nil
400 }
401 var b kv.Batch
402 b.RequireVersion(key, item.Version)
403 b.Delete(key)
404 if _, err := s.db.Apply(ctx, &b); err != nil && !errors.Is(err, kv.ErrConflict) {
405 return false, fmt.Errorf("delete expired ban: %w", err)
406 }
407 return false, nil
408}