feat(kvstore): store pending issues, bans and the issue form key

commit: 6a3e65dac5833f8b52bd6273fbdbf9585cb371bb
author: brice · 3d ago · 2026-10-19 06:54 UTC
parent: dca1fe7
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

6 files · +912 -13

M gen/ermite/storage/issue.pb.go +199 -12 generated file: collapsed

A internal/adapters/kvstore/issues_pending.go +416 -0

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

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

A internal/adapters/kvstore/issues_pending_test.go +269 -0

@@ -0,0 +1,269 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	package kvstore_test
	4	+
	5	+	import (
	6	+	"bytes"
	7	+	"context"
	8	+	"sync"
	9	+	"testing"
	10	+	"time"
	11	+
	12	+	"github.com/stretchr/testify/assert"
	13	+	"github.com/stretchr/testify/require"
	14	+
	15	+	"bnema.dev/ermite/internal/domain"
	16	+	"bnema.dev/ermite/internal/ports"
	17	+	)
	18	+
	19	+	func newPending(t *testing.T, repo domain.RepoName, title, author string, at time.Time) domain.PendingIssue {
	20	+	t.Helper()
	21	+	id, err := domain.NewPendingIssueID()
	22	+	require.NoError(t, err)
	23	+	ti, err := domain.ParseIssueTitle(title)
	24	+	require.NoError(t, err)
	25	+	body, err := domain.ParseIssueBody("details")
	26	+	require.NoError(t, err)
	27	+	return domain.PendingIssue{
	28	+	ID: id, Repo: repo, Title: ti, Body: body, Author: author, CreatedAt: at,
	29	+	IPHash: bytes.Repeat([]byte{7}, domain.IPHashLen),
	30	+	}
	31	+	}
	32	+
	33	+	func TestPendingIssuePutGetListCount(t *testing.T) {
	34	+	ctx := context.Background()
	35	+	s, _, n := repoWithMRs(t)
	36	+	other := name(t, "other")
	37	+	require.NoError(t, s.Create(ctx, domain.Repo{Name: other, CreatedAt: created, Visibility: domain.VisibilityPublic}))
	38	+	ps := s.PendingIssues()
	39	+
	40	+	a := newPending(t, n, "a", "Alex", created.Add(time.Minute))
	41	+	b := newPending(t, other, "b", "", created.Add(2*time.Minute))
	42	+	c := newPending(t, n, "c", "", created.Add(3*time.Minute))
	43	+	for _, p := range []domain.PendingIssue{c, a, b} {
	44	+	require.NoError(t, ps.Put(ctx, p))
	45	+	}
	46	+	require.ErrorIs(t, ps.Put(ctx, a), domain.ErrConflict, "an id is used once")
	47	+	require.ErrorIs(t, ps.Put(ctx, newPending(t, name(t, "ghost"), "g", "", created)), domain.ErrNotFound)
	48	+
	49	+	got, err := ps.Get(ctx, n, a.ID)
	50	+	require.NoError(t, err)
	51	+	assert.Equal(t, a.Title, got.Title)
	52	+	assert.Equal(t, "Alex", got.Author)
	53	+	assert.Equal(t, a.IPHash, got.IPHash)
	54	+	assert.True(t, got.CreatedAt.Equal(a.CreatedAt))
	55	+	assert.NotZero(t, got.Version)
	56	+	_, err = ps.Get(ctx, other, a.ID)
	57	+	assert.ErrorIs(t, err, domain.ErrNotFound, "ids are per repository")
	58	+
	59	+	all, err := ps.List(ctx, domain.RepoName{}, ports.ListPendingParams{})
	60	+	require.NoError(t, err)
	61	+	require.Len(t, all.Issues, 3)
	62	+	assert.Equal(t, []domain.PendingIssueID{a.ID, b.ID, c.ID}, []domain.PendingIssueID{all.Issues[0].ID, all.Issues[1].ID, all.Issues[2].ID}, "oldest first")
	63	+
	64	+	mine, err := ps.List(ctx, n, ports.ListPendingParams{})
	65	+	require.NoError(t, err)
	66	+	assert.Len(t, mine.Issues, 2)
	67	+
	68	+	p1, err := ps.List(ctx, domain.RepoName{}, ports.ListPendingParams{PageSize: 2})
	69	+	require.NoError(t, err)
	70	+	require.Len(t, p1.Issues, 2)
	71	+	require.NotEmpty(t, p1.NextPageToken)
	72	+	p2, err := ps.List(ctx, domain.RepoName{}, ports.ListPendingParams{PageSize: 2, PageToken: p1.NextPageToken})
	73	+	require.NoError(t, err)
	74	+	require.Len(t, p2.Issues, 1)
	75	+	assert.Equal(t, c.ID, p2.Issues[0].ID)
	76	+	assert.Empty(t, p2.NextPageToken)
	77	+	_, err = ps.List(ctx, domain.RepoName{}, ports.ListPendingParams{PageToken: "x"})
	78	+	assert.ErrorIs(t, err, domain.ErrInvalid)
	79	+
	80	+	cnt, err := ps.Count(ctx)
	81	+	require.NoError(t, err)
	82	+	assert.Equal(t, 3, cnt)
	83	+
	84	+	require.NoError(t, ps.Delete(ctx, n, a.ID))
	85	+	assert.ErrorIs(t, ps.Delete(ctx, n, a.ID), domain.ErrNotFound)
	86	+	cnt, err = ps.Count(ctx)
	87	+	require.NoError(t, err)
	88	+	assert.Equal(t, 2, cnt)
	89	+	}
	90	+
	91	+	func TestPendingIssueCountIsCapped(t *testing.T) {
	92	+	ctx := context.Background()
	93	+	s, _, n := repoWithMRs(t)
	94	+	ps := s.PendingIssues()
	95	+	for i := range domain.MaxPendingIssues + 5 {
	96	+	require.NoError(t, ps.Put(ctx, newPending(t, n, "t", "", created.Add(time.Duration(i)*time.Second))))
	97	+	}
	98	+	cnt, err := ps.Count(ctx)
	99	+	require.NoError(t, err)
	100	+	assert.Equal(t, domain.MaxPendingIssues+1, cnt)
	101	+	}
	102	+
	103	+	func TestPendingIssueApproveSharesNumbers(t *testing.T) {
	104	+	ctx := context.Background()
	105	+	s, mrs, n := repoWithMRs(t)
	106	+	ps := s.PendingIssues()
	107	+
	108	+	mr, err := mrs.Create(ctx, newMR(t, n, "mr"))
	109	+	require.NoError(t, err)
	110	+	assert.Equal(t, int64(1), mr.Number)
	111	+
	112	+	named := newPending(t, n, "named", "Alex", created)
	113	+	anon := newPending(t, n, "anon", "", created.Add(time.Second))
	114	+	require.NoError(t, ps.Put(ctx, named))
	115	+	require.NoError(t, ps.Put(ctx, anon))
	116	+
	117	+	issue, err := ps.Approve(ctx, n, named.ID)
	118	+	require.NoError(t, err)
	119	+	assert.Equal(t, int64(2), issue.Number)
	120	+	assert.Equal(t, domain.IssueStateOpen, issue.State)
	121	+	assert.Equal(t, "Alex", issue.Author)
	122	+	assert.Equal(t, "named", issue.Title.String())
	123	+
	124	+	_, err = ps.Get(ctx, n, named.ID)
	125	+	assert.ErrorIs(t, err, domain.ErrNotFound, "the pending record is gone")
	126	+	_, err = ps.Approve(ctx, n, named.ID)
	127	+	assert.ErrorIs(t, err, domain.ErrNotFound, "approved once")
	128	+
	129	+	got, err := s.Issues().Get(ctx, n, 2)
	130	+	require.NoError(t, err)
	131	+	assert.Equal(t, issue.Title, got.Title)
	132	+	open, err := s.Issues().ListOpen(ctx, n)
	133	+	require.NoError(t, err)
	134	+	require.Len(t, open, 1)
	135	+	assert.Equal(t, int64(2), open[0].Number)
	136	+
	137	+	second, err := ps.Approve(ctx, n, anon.ID)
	138	+	require.NoError(t, err)
	139	+	assert.Equal(t, int64(3), second.Number)
	140	+	assert.Equal(t, "anonymous", second.Author)
	141	+
	142	+	next, err := s.Issues().Create(ctx, newIssue(t, n, "owner"))
	143	+	require.NoError(t, err)
	144	+	assert.Equal(t, int64(4), next.Number, "issues and merge requests share the counter")
	145	+	}
	146	+
	147	+	func TestPendingIssueApproveConcurrent(t *testing.T) {
	148	+	ctx := context.Background()
	149	+	s, mrs, n := repoWithMRs(t)
	150	+	ps := s.PendingIssues()
	151	+	const k = 8
	152	+	ids := make([]domain.PendingIssueID, k)
	153	+	for i := range k {
	154	+	p := newPending(t, n, "t", "", created.Add(time.Duration(i)*time.Second))
	155	+	ids[i] = p.ID
	156	+	require.NoError(t, ps.Put(ctx, p))
	157	+	}
	158	+	var mu sync.Mutex
	159	+	seen := map[int64]bool{}
	160	+	var wg sync.WaitGroup
	161	+	for _, id := range ids {
	162	+	wg.Add(1)
	163	+	go func() {
	164	+	defer wg.Done()
	165	+	is, err := ps.Approve(ctx, n, id)
	166	+	if assert.NoError(t, err) {
	167	+	mu.Lock()
	168	+	seen[is.Number] = true
	169	+	mu.Unlock()
	170	+	}
	171	+	}()
	172	+	}
	173	+	wg.Add(1)
	174	+	go func() {
	175	+	defer wg.Done()
	176	+	mr, err := mrs.Create(ctx, newMR(t, n, "mr"))
	177	+	if assert.NoError(t, err) {
	178	+	mu.Lock()
	179	+	seen[mr.Number] = true
	180	+	mu.Unlock()
	181	+	}
	182	+	}()
	183	+	wg.Wait()
	184	+	assert.Len(t, seen, k+1, "no number is given twice")
	185	+	}
	186	+
	187	+	func TestPendingIssueApproveOnDeletedRepoFails(t *testing.T) {
	188	+	ctx := context.Background()
	189	+	s, _, n := repoWithMRs(t)
	190	+	ps := s.PendingIssues()
	191	+	p := newPending(t, n, "t", "", created)
	192	+	require.NoError(t, ps.Put(ctx, p))
	193	+	require.NoError(t, s.Delete(ctx, n))
	194	+
	195	+	_, err := ps.Approve(ctx, n, p.ID)
	196	+	assert.ErrorIs(t, err, domain.ErrNotFound)
	197	+	assert.ErrorIs(t, ps.Put(ctx, newPending(t, n, "t", "", created)), domain.ErrNotFound)
	198	+	assert.Zero(t, countKeys(t, s, "is/"))
	199	+	assert.Zero(t, countKeys(t, s, "isopen/"))
	200	+	}
	201	+
	202	+	func TestRepoDeleteRemovesPendingIssues(t *testing.T) {
	203	+	ctx := context.Background()
	204	+	s, _, n := repoWithMRs(t)
	205	+	other := name(t, "other")
	206	+	require.NoError(t, s.Create(ctx, domain.Repo{Name: other, CreatedAt: created, Visibility: domain.VisibilityPublic}))
	207	+	ps := s.PendingIssues()
	208	+	require.NoError(t, ps.Put(ctx, newPending(t, n, "t", "", created)))
	209	+	require.NoError(t, ps.Put(ctx, newPending(t, n, "u", "", created)))
	210	+	keep := newPending(t, other, "k", "", created)
	211	+	require.NoError(t, ps.Put(ctx, keep))
	212	+
	213	+	require.NoError(t, s.Delete(ctx, n))
	214	+	assert.Zero(t, countKeys(t, s, "ispend/"+n.String()+"/"))
	215	+	_, err := ps.Get(ctx, other, keep.ID)
	216	+	assert.NoError(t, err, "another repository keeps its pending issues")
	217	+	cnt, err := ps.Count(ctx)
	218	+	require.NoError(t, err)
	219	+	assert.Equal(t, 1, cnt)
	220	+	}
	221	+
	222	+	func TestIssueBansExpire(t *testing.T) {
	223	+	ctx := context.Background()
	224	+	s := openStore(t, t.TempDir())
	225	+	bans := s.IssueBans()
	226	+	h := bytes.Repeat([]byte{1}, domain.IPHashLen)
	227	+	other := bytes.Repeat([]byte{2}, domain.IPHashLen)
	228	+	now := created
	229	+
	230	+	banned, err := bans.Banned(ctx, h, now)
	231	+	require.NoError(t, err)
	232	+	assert.False(t, banned)
	233	+
	234	+	require.NoError(t, bans.Ban(ctx, h, now.Add(domain.BanDuration)))
	235	+	require.ErrorIs(t, bans.Ban(ctx, []byte("short"), now), domain.ErrInvalid)
	236	+	banned, err = bans.Banned(ctx, h, now.Add(time.Hour))
	237	+	require.NoError(t, err)
	238	+	assert.True(t, banned)
	239	+	banned, err = bans.Banned(ctx, other, now.Add(time.Hour))
	240	+	require.NoError(t, err)
	241	+	assert.False(t, banned, "another hash is not banned")
	242	+	assert.Equal(t, 1, countKeys(t, s, "isban/"))
	243	+
	244	+	banned, err = bans.Banned(ctx, h, now.Add(domain.BanDuration))
	245	+	require.NoError(t, err)
	246	+	assert.False(t, banned, "expired")
	247	+	assert.Zero(t, countKeys(t, s, "isban/"), "an expired ban is deleted")
	248	+	}
	249	+
	250	+	func TestIssueKeyPersists(t *testing.T) {
	251	+	ctx := context.Background()
	252	+	dir := t.TempDir()
	253	+	s := openStore(t, dir)
	254	+	key, err := s.IssueKeys().Key(ctx)
	255	+	require.NoError(t, err)
	256	+	assert.Len(t, key, 32)
	257	+	again, err := s.IssueKeys().Key(ctx)
	258	+	require.NoError(t, err)
	259	+	assert.Equal(t, key, again)
	260	+	sess, err := s.Sessions().Key(ctx)
	261	+	require.NoError(t, err)
	262	+	assert.NotEqual(t, sess, key, "not the session key")
	263	+	require.NoError(t, s.Close(ctx))
	264	+
	265	+	s = openStore(t, dir)
	266	+	reopened, err := s.IssueKeys().Key(ctx)
	267	+	require.NoError(t, err)
	268	+	assert.Equal(t, key, reopened, "a restart keeps the key")
	269	+	}

@@ -0,0 +1,269 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	package kvstore_test
	4	+
	5	+	import (
	6	+	"bytes"
	7	+	"context"
	8	+	"sync"
	9	+	"testing"
	10	+	"time"
	11	+
	12	+	"github.com/stretchr/testify/assert"
	13	+	"github.com/stretchr/testify/require"
	14	+
	15	+	"bnema.dev/ermite/internal/domain"
	16	+	"bnema.dev/ermite/internal/ports"
	17	+	)
	18	+
	19	+	func newPending(t *testing.T, repo domain.RepoName, title, author string, at time.Time) domain.PendingIssue {
	20	+	t.Helper()
	21	+	id, err := domain.NewPendingIssueID()
	22	+	require.NoError(t, err)
	23	+	ti, err := domain.ParseIssueTitle(title)
	24	+	require.NoError(t, err)
	25	+	body, err := domain.ParseIssueBody("details")
	26	+	require.NoError(t, err)
	27	+	return domain.PendingIssue{
	28	+	ID: id, Repo: repo, Title: ti, Body: body, Author: author, CreatedAt: at,
	29	+	IPHash: bytes.Repeat([]byte{7}, domain.IPHashLen),
	30	+	}
	31	+	}
	32	+
	33	+	func TestPendingIssuePutGetListCount(t *testing.T) {
	34	+	ctx := context.Background()
	35	+	s, _, n := repoWithMRs(t)
	36	+	other := name(t, "other")
	37	+	require.NoError(t, s.Create(ctx, domain.Repo{Name: other, CreatedAt: created, Visibility: domain.VisibilityPublic}))
	38	+	ps := s.PendingIssues()
	39	+
	40	+	a := newPending(t, n, "a", "Alex", created.Add(time.Minute))
	41	+	b := newPending(t, other, "b", "", created.Add(2*time.Minute))
	42	+	c := newPending(t, n, "c", "", created.Add(3*time.Minute))
	43	+	for _, p := range []domain.PendingIssue{c, a, b} {
	44	+	require.NoError(t, ps.Put(ctx, p))
	45	+	}
	46	+	require.ErrorIs(t, ps.Put(ctx, a), domain.ErrConflict, "an id is used once")
	47	+	require.ErrorIs(t, ps.Put(ctx, newPending(t, name(t, "ghost"), "g", "", created)), domain.ErrNotFound)
	48	+
	49	+	got, err := ps.Get(ctx, n, a.ID)
	50	+	require.NoError(t, err)
	51	+	assert.Equal(t, a.Title, got.Title)
	52	+	assert.Equal(t, "Alex", got.Author)
	53	+	assert.Equal(t, a.IPHash, got.IPHash)
	54	+	assert.True(t, got.CreatedAt.Equal(a.CreatedAt))
	55	+	assert.NotZero(t, got.Version)
	56	+	_, err = ps.Get(ctx, other, a.ID)
	57	+	assert.ErrorIs(t, err, domain.ErrNotFound, "ids are per repository")
	58	+
	59	+	all, err := ps.List(ctx, domain.RepoName{}, ports.ListPendingParams{})
	60	+	require.NoError(t, err)
	61	+	require.Len(t, all.Issues, 3)
	62	+	assert.Equal(t, []domain.PendingIssueID{a.ID, b.ID, c.ID}, []domain.PendingIssueID{all.Issues[0].ID, all.Issues[1].ID, all.Issues[2].ID}, "oldest first")
	63	+
	64	+	mine, err := ps.List(ctx, n, ports.ListPendingParams{})
	65	+	require.NoError(t, err)
	66	+	assert.Len(t, mine.Issues, 2)
	67	+
	68	+	p1, err := ps.List(ctx, domain.RepoName{}, ports.ListPendingParams{PageSize: 2})
	69	+	require.NoError(t, err)
	70	+	require.Len(t, p1.Issues, 2)
	71	+	require.NotEmpty(t, p1.NextPageToken)
	72	+	p2, err := ps.List(ctx, domain.RepoName{}, ports.ListPendingParams{PageSize: 2, PageToken: p1.NextPageToken})
	73	+	require.NoError(t, err)
	74	+	require.Len(t, p2.Issues, 1)
	75	+	assert.Equal(t, c.ID, p2.Issues[0].ID)
	76	+	assert.Empty(t, p2.NextPageToken)
	77	+	_, err = ps.List(ctx, domain.RepoName{}, ports.ListPendingParams{PageToken: "x"})
	78	+	assert.ErrorIs(t, err, domain.ErrInvalid)
	79	+
	80	+	cnt, err := ps.Count(ctx)
	81	+	require.NoError(t, err)
	82	+	assert.Equal(t, 3, cnt)
	83	+
	84	+	require.NoError(t, ps.Delete(ctx, n, a.ID))
	85	+	assert.ErrorIs(t, ps.Delete(ctx, n, a.ID), domain.ErrNotFound)
	86	+	cnt, err = ps.Count(ctx)
	87	+	require.NoError(t, err)
	88	+	assert.Equal(t, 2, cnt)
	89	+	}
	90	+
	91	+	func TestPendingIssueCountIsCapped(t *testing.T) {
	92	+	ctx := context.Background()
	93	+	s, _, n := repoWithMRs(t)
	94	+	ps := s.PendingIssues()
	95	+	for i := range domain.MaxPendingIssues + 5 {
	96	+	require.NoError(t, ps.Put(ctx, newPending(t, n, "t", "", created.Add(time.Duration(i)*time.Second))))
	97	+	}
	98	+	cnt, err := ps.Count(ctx)
	99	+	require.NoError(t, err)
	100	+	assert.Equal(t, domain.MaxPendingIssues+1, cnt)
	101	+	}
	102	+
	103	+	func TestPendingIssueApproveSharesNumbers(t *testing.T) {
	104	+	ctx := context.Background()
	105	+	s, mrs, n := repoWithMRs(t)
	106	+	ps := s.PendingIssues()
	107	+
	108	+	mr, err := mrs.Create(ctx, newMR(t, n, "mr"))
	109	+	require.NoError(t, err)
	110	+	assert.Equal(t, int64(1), mr.Number)
	111	+
	112	+	named := newPending(t, n, "named", "Alex", created)
	113	+	anon := newPending(t, n, "anon", "", created.Add(time.Second))
	114	+	require.NoError(t, ps.Put(ctx, named))
	115	+	require.NoError(t, ps.Put(ctx, anon))
	116	+
	117	+	issue, err := ps.Approve(ctx, n, named.ID)
	118	+	require.NoError(t, err)
	119	+	assert.Equal(t, int64(2), issue.Number)
	120	+	assert.Equal(t, domain.IssueStateOpen, issue.State)
	121	+	assert.Equal(t, "Alex", issue.Author)
	122	+	assert.Equal(t, "named", issue.Title.String())
	123	+
	124	+	_, err = ps.Get(ctx, n, named.ID)
	125	+	assert.ErrorIs(t, err, domain.ErrNotFound, "the pending record is gone")
	126	+	_, err = ps.Approve(ctx, n, named.ID)
	127	+	assert.ErrorIs(t, err, domain.ErrNotFound, "approved once")
	128	+
	129	+	got, err := s.Issues().Get(ctx, n, 2)
	130	+	require.NoError(t, err)
	131	+	assert.Equal(t, issue.Title, got.Title)
	132	+	open, err := s.Issues().ListOpen(ctx, n)
	133	+	require.NoError(t, err)
	134	+	require.Len(t, open, 1)
	135	+	assert.Equal(t, int64(2), open[0].Number)
	136	+
	137	+	second, err := ps.Approve(ctx, n, anon.ID)
	138	+	require.NoError(t, err)
	139	+	assert.Equal(t, int64(3), second.Number)
	140	+	assert.Equal(t, "anonymous", second.Author)
	141	+
	142	+	next, err := s.Issues().Create(ctx, newIssue(t, n, "owner"))
	143	+	require.NoError(t, err)
	144	+	assert.Equal(t, int64(4), next.Number, "issues and merge requests share the counter")
	145	+	}
	146	+
	147	+	func TestPendingIssueApproveConcurrent(t *testing.T) {
	148	+	ctx := context.Background()
	149	+	s, mrs, n := repoWithMRs(t)
	150	+	ps := s.PendingIssues()
	151	+	const k = 8
	152	+	ids := make([]domain.PendingIssueID, k)
	153	+	for i := range k {
	154	+	p := newPending(t, n, "t", "", created.Add(time.Duration(i)*time.Second))
	155	+	ids[i] = p.ID
	156	+	require.NoError(t, ps.Put(ctx, p))
	157	+	}
	158	+	var mu sync.Mutex
	159	+	seen := map[int64]bool{}
	160	+	var wg sync.WaitGroup
	161	+	for _, id := range ids {
	162	+	wg.Add(1)
	163	+	go func() {
	164	+	defer wg.Done()
	165	+	is, err := ps.Approve(ctx, n, id)
	166	+	if assert.NoError(t, err) {
	167	+	mu.Lock()
	168	+	seen[is.Number] = true
	169	+	mu.Unlock()
	170	+	}
	171	+	}()
	172	+	}
	173	+	wg.Add(1)
	174	+	go func() {
	175	+	defer wg.Done()
	176	+	mr, err := mrs.Create(ctx, newMR(t, n, "mr"))
	177	+	if assert.NoError(t, err) {
	178	+	mu.Lock()
	179	+	seen[mr.Number] = true
	180	+	mu.Unlock()
	181	+	}
	182	+	}()
	183	+	wg.Wait()
	184	+	assert.Len(t, seen, k+1, "no number is given twice")
	185	+	}
	186	+
	187	+	func TestPendingIssueApproveOnDeletedRepoFails(t *testing.T) {
	188	+	ctx := context.Background()
	189	+	s, _, n := repoWithMRs(t)
	190	+	ps := s.PendingIssues()
	191	+	p := newPending(t, n, "t", "", created)
	192	+	require.NoError(t, ps.Put(ctx, p))
	193	+	require.NoError(t, s.Delete(ctx, n))
	194	+
	195	+	_, err := ps.Approve(ctx, n, p.ID)
	196	+	assert.ErrorIs(t, err, domain.ErrNotFound)
	197	+	assert.ErrorIs(t, ps.Put(ctx, newPending(t, n, "t", "", created)), domain.ErrNotFound)
	198	+	assert.Zero(t, countKeys(t, s, "is/"))
	199	+	assert.Zero(t, countKeys(t, s, "isopen/"))
	200	+	}
	201	+
	202	+	func TestRepoDeleteRemovesPendingIssues(t *testing.T) {
	203	+	ctx := context.Background()
	204	+	s, _, n := repoWithMRs(t)
	205	+	other := name(t, "other")
	206	+	require.NoError(t, s.Create(ctx, domain.Repo{Name: other, CreatedAt: created, Visibility: domain.VisibilityPublic}))
	207	+	ps := s.PendingIssues()
	208	+	require.NoError(t, ps.Put(ctx, newPending(t, n, "t", "", created)))
	209	+	require.NoError(t, ps.Put(ctx, newPending(t, n, "u", "", created)))
	210	+	keep := newPending(t, other, "k", "", created)
	211	+	require.NoError(t, ps.Put(ctx, keep))
	212	+
	213	+	require.NoError(t, s.Delete(ctx, n))
	214	+	assert.Zero(t, countKeys(t, s, "ispend/"+n.String()+"/"))
	215	+	_, err := ps.Get(ctx, other, keep.ID)
	216	+	assert.NoError(t, err, "another repository keeps its pending issues")
	217	+	cnt, err := ps.Count(ctx)
	218	+	require.NoError(t, err)
	219	+	assert.Equal(t, 1, cnt)
	220	+	}
	221	+
	222	+	func TestIssueBansExpire(t *testing.T) {
	223	+	ctx := context.Background()
	224	+	s := openStore(t, t.TempDir())
	225	+	bans := s.IssueBans()
	226	+	h := bytes.Repeat([]byte{1}, domain.IPHashLen)
	227	+	other := bytes.Repeat([]byte{2}, domain.IPHashLen)
	228	+	now := created
	229	+
	230	+	banned, err := bans.Banned(ctx, h, now)
	231	+	require.NoError(t, err)
	232	+	assert.False(t, banned)
	233	+
	234	+	require.NoError(t, bans.Ban(ctx, h, now.Add(domain.BanDuration)))
	235	+	require.ErrorIs(t, bans.Ban(ctx, []byte("short"), now), domain.ErrInvalid)
	236	+	banned, err = bans.Banned(ctx, h, now.Add(time.Hour))
	237	+	require.NoError(t, err)
	238	+	assert.True(t, banned)
	239	+	banned, err = bans.Banned(ctx, other, now.Add(time.Hour))
	240	+	require.NoError(t, err)
	241	+	assert.False(t, banned, "another hash is not banned")
	242	+	assert.Equal(t, 1, countKeys(t, s, "isban/"))
	243	+
	244	+	banned, err = bans.Banned(ctx, h, now.Add(domain.BanDuration))
	245	+	require.NoError(t, err)
	246	+	assert.False(t, banned, "expired")
	247	+	assert.Zero(t, countKeys(t, s, "isban/"), "an expired ban is deleted")
	248	+	}
	249	+
	250	+	func TestIssueKeyPersists(t *testing.T) {
	251	+	ctx := context.Background()
	252	+	dir := t.TempDir()
	253	+	s := openStore(t, dir)
	254	+	key, err := s.IssueKeys().Key(ctx)
	255	+	require.NoError(t, err)
	256	+	assert.Len(t, key, 32)
	257	+	again, err := s.IssueKeys().Key(ctx)
	258	+	require.NoError(t, err)
	259	+	assert.Equal(t, key, again)
	260	+	sess, err := s.Sessions().Key(ctx)
	261	+	require.NoError(t, err)
	262	+	assert.NotEqual(t, sess, key, "not the session key")
	263	+	require.NoError(t, s.Close(ctx))
	264	+
	265	+	s = openStore(t, dir)
	266	+	reopened, err := s.IssueKeys().Key(ctx)
	267	+	require.NoError(t, err)
	268	+	assert.Equal(t, key, reopened, "a restart keeps the key")
	269	+	}

M internal/adapters/kvstore/repos.go +1 -0


@@ -145,6 +145,7 @@ func (s *Store) Delete(ctx context.Context, name domain.RepoName) error {
145	145		snap := s.db.Snapshot(ctx)
146	146		repoItem, repoOK := snap.Get(ctx, repoKey(name))
147	147		keys := append(mergeRequestKeys(ctx, snap, name), issueKeys(ctx, snap, name)...)
	148	+	keys = append(keys, pendingKeys(ctx, snap, name)...)
148	149		keys = append(keys, ciKeys(ctx, snap, name)...)
149	150		keys = append(keys, settingsKeys(ctx, snap, name)...)
150	151		for len(keys) > deleteChunk {


@@ -145,6 +145,7 @@ func (s *Store) Delete(ctx context.Context, name domain.RepoName) error {
145	snap := s.db.Snapshot(ctx)	145		snap := s.db.Snapshot(ctx)
146	repoItem, repoOK := snap.Get(ctx, repoKey(name))	146		repoItem, repoOK := snap.Get(ctx, repoKey(name))
147	keys := append(mergeRequestKeys(ctx, snap, name), issueKeys(ctx, snap, name)...)	147		keys := append(mergeRequestKeys(ctx, snap, name), issueKeys(ctx, snap, name)...)
		148	+	keys = append(keys, pendingKeys(ctx, snap, name)...)
148	keys = append(keys, ciKeys(ctx, snap, name)...)	149		keys = append(keys, ciKeys(ctx, snap, name)...)
149	keys = append(keys, settingsKeys(ctx, snap, name)...)	150		keys = append(keys, settingsKeys(ctx, snap, name)...)
150	for len(keys) > deleteChunk {	151		for len(keys) > deleteChunk {

M internal/adapters/kvstore/store.go +4 -1


@@ -33,7 +33,10 @@ import (
33	33		// isopen/, numbered from the same mrseq/ sequence as the merge requests, and
34	34		// RepoSettings.issues): an older build would not delete them with their
35	35		// repository, nor keep the numbers apart, and it would misread the issue:write
36		-	// scope, so it must refuse the data. The upgrade also adds issue:write to the
	36	+	// scope, so it must refuse the data. It also adds the pending issues of the
	37	+	// public form (ispend/<repo>/<id>, removed with their repository), the bans of
	38	+	// their submitters (isban/<hex hash>) and the form's server key
	39	+	// (meta/issuekey). The upgrade also adds issue:write to the
37	40		// stored tokens that held all nine earlier scopes with no repository list and
38	41		// no branch prefix, so that they keep the owner rights (see upgradeTokens).
39	42		const SchemaVersion = 4

 // isopen/, numbered from the same mrseq/ sequence as the merge requests, and
 // RepoSettings.issues): an older build would not delete them with their
 // repository, nor keep the numbers apart, and it would misread the issue:write
-// scope, so it must refuse the data. The upgrade also adds issue:write to the
+// scope, so it must refuse the data. It also adds the pending issues of the
+// public form (ispend/<repo>/<id>, removed with their repository), the bans of
+// their submitters (isban/<hex hash>) and the form's server key
+// (meta/issuekey). The upgrade also adds issue:write to the
 // stored tokens that held all nine earlier scopes with no repository list and
 // no branch prefix, so that they keep the owner rights (see upgradeTokens).
 const SchemaVersion = 4

M proto/ermite/storage/issue.proto +23 -0


@@ -43,3 +43,26 @@ message IssueComment {
43	43		string text = 3;
44	44		google.protobuf.Timestamp created_at = 4;
45	45		}
	46	+
	47	+	// PendingIssue is an issue submitted through the public form, stored under
	48	+	// the key "ispend/<repo>/<id>" where id is 16 random bytes in lowercase base32.
	49	+	// It has no number until the owner approves it.
	50	+	message PendingIssue {
	51	+	string title = 1;
	52	+	string body = 2;
	53	+	// Display name given by the submitter. May be empty.
	54	+	string author = 3;
	55	+	google.protobuf.Timestamp created_at = 4;
	56	+	// HMAC-SHA256 of the client key (32 bytes).
	57	+	bytes ip_hash = 5;
	58	+	}
	59	+
	60	+	// IssueBan is stored under the key "isban/<hex of the hash>".
	61	+	message IssueBan {
	62	+	google.protobuf.Timestamp until = 1;
	63	+	}
	64	+
	65	+	// IssueKey is stored under the key "meta/issuekey".
	66	+	message IssueKey {
	67	+	bytes key = 1;
	68	+	}


@@ -43,3 +43,26 @@ message IssueComment {
43	string text = 3;	43		string text = 3;
44	google.protobuf.Timestamp created_at = 4;	44		google.protobuf.Timestamp created_at = 4;
45	}	45		}
		46	+
		47	+	// PendingIssue is an issue submitted through the public form, stored under
		48	+	// the key "ispend/<repo>/<id>" where id is 16 random bytes in lowercase base32.
		49	+	// It has no number until the owner approves it.
		50	+	message PendingIssue {
		51	+	string title = 1;
		52	+	string body = 2;
		53	+	// Display name given by the submitter. May be empty.
		54	+	string author = 3;
		55	+	google.protobuf.Timestamp created_at = 4;
		56	+	// HMAC-SHA256 of the client key (32 bytes).
		57	+	bytes ip_hash = 5;
		58	+	}
		59	+
		60	+	// IssueBan is stored under the key "isban/<hex of the hash>".
		61	+	message IssueBan {
		62	+	google.protobuf.Timestamp until = 1;
		63	+	}
		64	+
		65	+	// IssueKey is stored under the key "meta/issuekey".
		66	+	message IssueKey {
		67	+	bytes key = 1;
		68	+	}