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.
97func (s *PendingIssues) List(ctx context.Context, repo domain.RepoName, page ports.ListPendingParams) (ports.PendingIssuePage, error) {
98 prefix := pendingPrefix
99 if !repo.IsZero() {
100 prefix = pendingRepoPrefix(repo)
101 }
102 var after pendingCursor
103 if page.PageToken != "" {
104 var err error
105 if after, err = parsePendingCursor(page.PageToken); err != nil {
106 return ports.PendingIssuePage{}, err
107 }
108 }
109 var all []domain.PendingIssue
110 for key, item := range s.db.Snapshot(ctx).Prefix(ctx, []byte(prefix)) {
111 if err := ctx.Err(); err != nil {
112 return ports.PendingIssuePage{}, err
113 }
114 r, id, err := splitPendingKey(string(key))
115 if err != nil {
116 return ports.PendingIssuePage{}, err
117 }
118 p, err := decodePending(r, id, item)
119 if err != nil {
120 return ports.PendingIssuePage{}, fmt.Errorf("decode %s: %w", key, err)
121 }
122 all = append(all, p)
123 }
124 slices.SortFunc(all, comparePending)
125 if page.PageToken != "" {
126 all = slices.DeleteFunc(all, func(p domain.PendingIssue) bool { return comparePending(p, after.item()) <= 0 })
127 }
128 var out ports.PendingIssuePage
129 if page.PageSize > 0 && len(all) > page.PageSize {
130 all = all[:page.PageSize]
131 out.NextPageToken = encodePendingCursor(all[len(all)-1])
132 }
133 out.Issues = all
134 return out, nil
135}
136
137func comparePending(a, b domain.PendingIssue) int {
138 if c := a.CreatedAt.Compare(b.CreatedAt); c != 0 {
139 return c
140 }
141 if c := strings.Compare(a.Repo.String(), b.Repo.String()); c != 0 {
142 return c
143 }
144 return strings.Compare(a.ID.String(), b.ID.String())
145}
146
147type pendingCursor struct {
148 nanos int64
149 repo domain.RepoName
150 id domain.PendingIssueID
151}
152
153func (c pendingCursor) item() domain.PendingIssue {
154 return domain.PendingIssue{Repo: c.repo, ID: c.id, CreatedAt: time.Unix(0, c.nanos)}
155}
156
157func encodePendingCursor(p domain.PendingIssue) string {
158 return strconv.FormatInt(p.CreatedAt.UnixNano(), 10) + "/" + p.Repo.String() + "/" + p.ID.String()
159}
160
161func parsePendingCursor(s string) (pendingCursor, error) {
162 bad := fmt.Errorf("%w: page token", domain.ErrInvalid)
163 parts := strings.Split(s, "/")
164 if len(parts) != 3 {
165 return pendingCursor{}, bad
166 }
167 n, err := strconv.ParseInt(parts[0], 10, 64)
168 if err != nil {
169 return pendingCursor{}, bad
170 }
171 repo, err := domain.ParseRepoName(parts[1])
172 if err != nil {
173 return pendingCursor{}, bad
174 }
175 id, err := domain.ParsePendingIssueID(parts[2])
176 if err != nil {
177 return pendingCursor{}, bad
178 }
179 return pendingCursor{nanos: n, repo: repo, id: id}, nil
180}
181
182func splitPendingKey(key string) (domain.RepoName, domain.PendingIssueID, error) {
183 rest := strings.TrimPrefix(key, pendingPrefix)
184 repoName, idStr, ok := strings.Cut(rest, "/")
185 if !ok {
186 return domain.RepoName{}, "", fmt.Errorf("pending issue key %q", key)
187 }
188 repo, err := domain.ParseRepoName(repoName)
189 if err != nil {
190 return domain.RepoName{}, "", fmt.Errorf("pending issue key %q: %w", key, err)
191 }
192 id, err := domain.ParsePendingIssueID(idStr)
193 if err != nil {
194 return domain.RepoName{}, "", fmt.Errorf("pending issue key %q: %w", key, err)
195 }
196 return repo, id, nil
197}
198
199// Count counts the keys under ispend/, stopping after MaxPendingIssues+1.
200func (s *PendingIssues) Count(ctx context.Context) (int, error) {
201 n := 0
202 for range s.db.Snapshot(ctx).Prefix(ctx, []byte(pendingPrefix)) {
203 if err := ctx.Err(); err != nil {
204 return 0, err
205 }
206 n++
207 if n > domain.MaxPendingIssues {
208 break
209 }
210 }
211 return n, nil
212}
213
214// Approve numbers the pending issue from mrseq/<repo> (see createNumbered),
215// writes the open issue and its index key and deletes the pending record, in
216// one batch that requires the pending record to be unchanged.
217func (s *PendingIssues) Approve(ctx context.Context, repo domain.RepoName, id domain.PendingIssueID, now time.Time) (domain.Issue, error) {
218 var number int64
219 err := createNumbered(ctx, s.db, s.locks, repo, func(b *kv.Batch, n int64) error {
220 key := pendingKey(repo, id)
221 item, ok := s.db.Get(ctx, key)
222 if !ok {
223 return domain.ErrNotFound
224 }
225 p, err := decodePending(repo, id, item)
226 if err != nil {
227 return err
228 }
229 issue := domain.ApprovedIssue(p, now)
230 issue.Repo, issue.Number = repo, n
231 val, err := encodeIssue(issue)
232 if err != nil {
233 return err
234 }
235 b.RequireVersion(key, item.Version)
236 b.Delete(key)
237 b.RequireAbsent(issueKey(repo, n))
238 b.Put(issueKey(repo, n), val)
239 indexOpenIssue(b, issue)
240 number = n
241 return nil
242 })
243 if err != nil {
244 if errors.Is(err, domain.ErrNotFound) {
245 return domain.Issue{}, domain.ErrNotFound
246 }
247 return domain.Issue{}, fmt.Errorf("approve pending issue of %q: %w", repo, err)
248 }
249 return (&Issues{db: s.db, locks: s.locks}).Get(ctx, repo, number)
250}
251
252// Delete removes the pending record, or returns domain.ErrNotFound.
253func (s *PendingIssues) Delete(ctx context.Context, repo domain.RepoName, id domain.PendingIssueID) error {
254 key := pendingKey(repo, id)
255 item, ok := s.db.Get(ctx, key)
256 if !ok {
257 return domain.ErrNotFound
258 }
259 var b kv.Batch
260 b.RequireVersion(key, item.Version)
261 b.Delete(key)
262 if _, err := s.db.Apply(ctx, &b); err != nil {
263 if errors.Is(err, kv.ErrConflict) {
264 return domain.ErrNotFound // gone or approved meanwhile
265 }
266 return fmt.Errorf("delete pending issue of %q: %w", repo, err)
267 }
268 return nil
269}
270
271func encodePending(p domain.PendingIssue) ([]byte, error) {
272 if p.Repo.IsZero() {
273 return nil, fmt.Errorf("%w: empty name", domain.ErrInvalidName)
274 }
275 if _, err := domain.ParsePendingIssueID(p.ID.String()); err != nil {
276 return nil, err
277 }
278 if !utf8.ValidString(p.Author) {
279 return nil, errors.New("pending issue author is not valid UTF-8")
280 }
281 if len(p.IPHash) != domain.IPHashLen {
282 return nil, fmt.Errorf("%w: ip hash is %d bytes, want %d", domain.ErrInvalid, len(p.IPHash), domain.IPHashLen)
283 }
284 return mustMarshal(&storagepb.PendingIssue{
285 Title: p.Title.String(), Body: p.Body.String(), Author: p.Author,
286 CreatedAt: encodeTime(p.CreatedAt), IpHash: p.IPHash,
287 }), nil
288}
289
290func decodePending(repo domain.RepoName, id domain.PendingIssueID, item kv.Item) (domain.PendingIssue, error) {
291 var pb storagepb.PendingIssue
292 if err := proto.Unmarshal(item.Value, &pb); err != nil {
293 return domain.PendingIssue{}, fmt.Errorf("decode pending issue: %w", err)
294 }
295 title, err := domain.ParseIssueTitle(pb.GetTitle())
296 if err != nil {
297 return domain.PendingIssue{}, fmt.Errorf("decode pending issue %s of %q: %w", id, repo, err)
298 }
299 body, err := domain.ParseIssueBody(pb.GetBody())
300 if err != nil {
301 return domain.PendingIssue{}, fmt.Errorf("decode pending issue %s of %q: %w", id, repo, err)
302 }
303 return domain.PendingIssue{
304 ID: id, Repo: repo, Title: title, Body: body, Author: pb.GetAuthor(),
305 CreatedAt: decodeTime(pb.GetCreatedAt()), IPHash: slices.Clone(pb.GetIpHash()), Version: item.Version,
306 }, nil
307}
308
309// IssueKeys is the ermite-kv backed ports.IssueKeyStore.
310type IssueKeys struct {
311 db *kv.DB
312}
313
314var _ ports.IssueKeyStore = (*IssueKeys)(nil)
315
316// IssueKeys returns the issue key store over the same database. It is valid
317// until the Store is closed.
318func (s *Store) IssueKeys() *IssueKeys { return &IssueKeys{db: s.db} }
319
320// Key reads meta/issuekey and generates it when absent. The key is never
321// logged.
322func (k *IssueKeys) Key(ctx context.Context) ([]byte, error) {
323 for range updateAttempts {
324 if item, ok := k.db.Get(ctx, []byte(issueKeyKey)); ok {
325 var m storagepb.IssueKey
326 if err := proto.Unmarshal(item.Value, &m); err != nil {
327 return nil, fmt.Errorf("decode issue key: %w", err)
328 }
329 if len(m.GetKey()) != issueKeyLen {
330 return nil, fmt.Errorf("decode issue key: %d bytes, want %d", len(m.GetKey()), issueKeyLen)
331 }
332 return append([]byte(nil), m.GetKey()...), nil
333 }
334 key := make([]byte, issueKeyLen)
335 if _, err := rand.Read(key); err != nil {
336 return nil, fmt.Errorf("generate issue key: %w", err)
337 }
338 var b kv.Batch
339 b.RequireAbsent([]byte(issueKeyKey))
340 b.Put([]byte(issueKeyKey), mustMarshal(&storagepb.IssueKey{Key: key}))
341 _, err := k.db.Apply(ctx, &b)
342 if errors.Is(err, kv.ErrConflict) {
343 continue // written meanwhile: read it
344 }
345 if err != nil {
346 return nil, fmt.Errorf("store issue key: %w", err)
347 }
348 log := zerowrap.FromCtx(ctx)
349 log.Info().Msg("issue key generated")
350 return key, nil
351 }
352 return nil, fmt.Errorf("store issue key: %w", domain.ErrConflict)
353}
354
355// IssueBans is the ermite-kv backed ports.IssueBanStore.
356type IssueBans struct {
357 db *kv.DB
358}
359
360var _ ports.IssueBanStore = (*IssueBans)(nil)
361
362// IssueBans returns the ban store over the same database. It is valid until
363// the Store is closed.
364func (s *Store) IssueBans() *IssueBans { return &IssueBans{db: s.db} }
365
366func banKey(hash []byte) []byte { return []byte(banPrefix + hex.EncodeToString(hash)) }
367
368// Ban writes isban/<hex hash> with its expiry.
369func (s *IssueBans) Ban(ctx context.Context, hash []byte, until time.Time) error {
370 if len(hash) != domain.IPHashLen {
371 return fmt.Errorf("%w: ban hash is %d bytes, want %d", domain.ErrInvalid, len(hash), domain.IPHashLen)
372 }
373 var b kv.Batch
374 b.Put(banKey(hash), mustMarshal(&storagepb.IssueBan{Until: encodeTime(until)}))
375 if _, err := s.db.Apply(ctx, &b); err != nil {
376 return fmt.Errorf("store ban: %w", err)
377 }
378 return nil
379}
380
381// Banned reads the ban; one that expired is not a ban and is deleted (if it
382// has not been replaced meanwhile).
383func (s *IssueBans) Banned(ctx context.Context, hash []byte, now time.Time) (bool, error) {
384 key := banKey(hash)
385 item, ok := s.db.Get(ctx, key)
386 if !ok {
387 return false, nil
388 }
389 var pb storagepb.IssueBan
390 if err := proto.Unmarshal(item.Value, &pb); err != nil {
391 return false, fmt.Errorf("decode ban: %w", err)
392 }
393 if now.Before(decodeTime(pb.GetUntil())) {
394 return true, nil
395 }
396 var b kv.Batch
397 b.RequireVersion(key, item.Version)
398 b.Delete(key)
399 if _, err := s.db.Apply(ctx, &b); err != nil && !errors.Is(err, kv.ErrConflict) {
400 return false, fmt.Errorf("delete expired ban: %w", err)
401 }
402 return false, nil
403}