Federated via graft
33 kB
888 lines
1// Package state persists sync bookkeeping so the daemon can tell, across
2// restarts, what has already been mirrored and avoid duplicating issues,
3// patches or commits on every poll.
4package state
5
6import (
7 "database/sql"
8 "fmt"
9 "os"
10 "strings"
11 "time"
12
13 _ "modernc.org/sqlite"
14)
15
16// Store wraps the local SQLite database used to track sync state.
17type Store struct {
18 db *sql.DB
19}
20
21// Open creates (if needed) and opens the state database at path.
22func Open(path string) (*Store, error) {
23 db, err := sql.Open("sqlite", path)
24 if err != nil {
25 return nil, fmt.Errorf("open state db: %w", err)
26 }
27 if err := migrate(db); err != nil {
28 db.Close()
29 return nil, fmt.Errorf("migrate state db: %w", err)
30 }
31 // This database holds every series' ActivityPub private key
32 // (ap_actor.private_key) in plaintext — it must never be
33 // group/world-readable, regardless of the umask the daemon happened to
34 // start with. Best-effort: sqlite may not have created the file yet on
35 // the very first Open before migrate runs, but by this point it has.
36 if err := os.Chmod(path, 0o600); err != nil {
37 return nil, fmt.Errorf("restrict state db permissions: %w", err)
38 }
39 return &Store{db: db}, nil
40}
41
42func (s *Store) Close() error {
43 return s.db.Close()
44}
45
46func migrate(db *sql.DB) error {
47 _, err := db.Exec(`
48CREATE TABLE IF NOT EXISTS git_cursor (
49 repo_pair TEXT PRIMARY KEY,
50 forgejo_sha TEXT NOT NULL DEFAULT '',
51 radicle_sha TEXT NOT NULL DEFAULT ''
52);
53
54CREATE TABLE IF NOT EXISTS item_mapping (
55 repo_pair TEXT NOT NULL,
56 kind TEXT NOT NULL, -- 'issue' or 'patch'
57 forgejo_id INTEGER NOT NULL DEFAULT 0,
58 radicle_id TEXT NOT NULL DEFAULT '',
59 content_hash TEXT NOT NULL DEFAULT '',
60 synced_at TEXT NOT NULL,
61 PRIMARY KEY (repo_pair, kind, forgejo_id, radicle_id)
62);
63
64CREATE INDEX IF NOT EXISTS idx_item_mapping_forgejo
65 ON item_mapping (repo_pair, kind, forgejo_id);
66CREATE INDEX IF NOT EXISTS idx_item_mapping_radicle
67 ON item_mapping (repo_pair, kind, radicle_id);
68
69-- One row per thing actually mirrored: a commit, an issue, a patch. This is
70-- what the status page's activity calendar and commit list read from — it's
71-- an append-only log, never updated, so it can't itself be a source of the
72-- kind of state-loss bug that hit item_mapping.
73CREATE TABLE IF NOT EXISTS activity_log (
74 id INTEGER PRIMARY KEY AUTOINCREMENT,
75 repo_pair TEXT NOT NULL,
76 kind TEXT NOT NULL, -- 'git', 'issue', 'patch'
77 direction TEXT NOT NULL, -- 'forgejo_to_radicle' or 'radicle_to_forgejo'
78 summary TEXT NOT NULL,
79 url TEXT NOT NULL DEFAULT '',
80 series TEXT NOT NULL DEFAULT '',
81 series_url TEXT NOT NULL DEFAULT '',
82 forgejo_id INTEGER NOT NULL DEFAULT 0,
83 radicle_id TEXT NOT NULL DEFAULT '',
84 occurred_at TEXT NOT NULL
85);
86
87CREATE INDEX IF NOT EXISTS idx_activity_log_occurred_at
88 ON activity_log (occurred_at);
89
90-- One row per series' ActivityPub actor: its RSA keypair, generated once
91-- lazily the first time that series is served over ActivityPub.
92CREATE TABLE IF NOT EXISTS ap_actor (
93 series TEXT PRIMARY KEY,
94 private_key TEXT NOT NULL,
95 public_key TEXT NOT NULL,
96 created_at TEXT NOT NULL
97);
98
99-- One row per remote actor following a series' ActivityPub actor.
100CREATE TABLE IF NOT EXISTS ap_follower (
101 series TEXT NOT NULL,
102 actor_uri TEXT NOT NULL,
103 inbox_uri TEXT NOT NULL,
104 PRIMARY KEY (series, actor_uri)
105);
106
107-- How far delivery has gotten through activity_log for each series, so a
108-- restart doesn't redeliver everything to every follower.
109CREATE TABLE IF NOT EXISTS ap_delivery_cursor (
110 series TEXT PRIMARY KEY,
111 last_delivered_id INTEGER NOT NULL DEFAULT 0
112);
113
114-- One-time secret shares: an admin (graft's own, or a peer's) encrypts a
115-- secret (typically a Forgejo token) under a key derived from a 6-digit
116-- passcode plus the high-entropy id itself. Only id_hash (never the id in
117-- reversible form) is stored, so a database dump alone can never be used
118-- to derive the decryption key. Deleted the moment it's successfully
119-- claimed, or after too many wrong passcode attempts, or past expires_at —
120-- whichever comes first.
121CREATE TABLE IF NOT EXISTS token_share (
122 id_hash TEXT PRIMARY KEY,
123 nonce TEXT NOT NULL,
124 ciphertext TEXT NOT NULL,
125 attempts INTEGER NOT NULL DEFAULT 0,
126 created_at TEXT NOT NULL,
127 expires_at TEXT NOT NULL
128);
129
130-- A repo pair added through the dashboard's self-service "add a peer" /
131-- "start a new repo" forms, gated by the shared admin password rather
132-- than a config.yaml edit + restart. forgejo_token_file (not the token
133-- itself) follows the same convention as every statically-configured
134-- pair: graft writes the submitted token out to its own chmod-600 file
135-- under the same directory config.yaml's token_file paths live in, and
136-- only the path is kept here. materialized flips to 1 once the running
137-- daemon has turned this row into a live RepoSyncer — it happens on the
138-- next sync pass, never mid-request, so a submission is never trusted
139-- before graft itself has validated it end to end.
140CREATE TABLE IF NOT EXISTS dynamic_repo (
141 id INTEGER PRIMARY KEY AUTOINCREMENT,
142 name TEXT NOT NULL UNIQUE,
143 series TEXT NOT NULL,
144 forgejo_base_url TEXT NOT NULL,
145 forgejo_owner TEXT NOT NULL,
146 forgejo_repo TEXT NOT NULL,
147 forgejo_token_file TEXT NOT NULL,
148 radicle_rid TEXT NOT NULL,
149 radicle_http_base_url TEXT NOT NULL,
150 radicle_explorer_url TEXT NOT NULL DEFAULT '',
151 radicle_rad_home TEXT NOT NULL,
152 sync_git INTEGER NOT NULL DEFAULT 1,
153 sync_issues INTEGER NOT NULL DEFAULT 1,
154 sync_patches INTEGER NOT NULL DEFAULT 1,
155 created_at TEXT NOT NULL,
156 materialized INTEGER NOT NULL DEFAULT 0
157);
158
159-- Dedupes comment mirroring across Forgejo and Radicle: a comment body's
160-- hash, once mirrored (or recognized as already present on both sides),
161-- is recorded here so a later pass never re-posts it back the other way.
162-- item_id is the Radicle-side issue/patch id, stable per repo_pair+kind
163-- regardless of which Forgejo peer in a federation is being scanned.
164CREATE TABLE IF NOT EXISTS comment_seen (
165 repo_pair TEXT NOT NULL,
166 kind TEXT NOT NULL, -- 'issue' or 'patch'
167 item_id TEXT NOT NULL, -- radicle issue/patch id
168 hash TEXT NOT NULL,
169 PRIMARY KEY (repo_pair, kind, item_id, hash)
170);
171
172-- One row per Bluesky post graft has made, so an inbound reply's parent
173-- URI can be traced back to the activity_log entry (and from there, the
174-- underlying Forgejo/Radicle item) it was posted about.
175CREATE TABLE IF NOT EXISTS atproto_post (
176 uri TEXT PRIMARY KEY,
177 activity_id INTEGER NOT NULL,
178 series TEXT NOT NULL,
179 created_at TEXT NOT NULL
180);
181
182-- How far AT Proto notification polling has gotten for each series'
183-- Bluesky account, so a restart doesn't re-process old replies.
184CREATE TABLE IF NOT EXISTS atproto_cursor (
185 series TEXT PRIMARY KEY,
186 cursor TEXT NOT NULL DEFAULT ''
187);
188
189-- A submitted "add a Radicle peer" request awaiting admin approval.
190-- Unlike dynamic_repo this never becomes an ongoing sync pair — approval
191-- just runs rad node connect + rad seed once, then the row is gone.
192CREATE TABLE IF NOT EXISTS pending_radicle_peer (
193 id INTEGER PRIMARY KEY AUTOINCREMENT,
194 series TEXT NOT NULL,
195 node_id TEXT NOT NULL,
196 address TEXT NOT NULL,
197 created_at TEXT NOT NULL
198);
199`)
200 if err != nil {
201 return err
202 }
203 // These columns were added after the table already shipped once; each
204 // ALTER is a no-op (ignored) on a fresh db where the CREATE above
205 // already included it.
206 for _, stmt := range []string{
207 `ALTER TABLE activity_log ADD COLUMN url TEXT NOT NULL DEFAULT ''`,
208 `ALTER TABLE activity_log ADD COLUMN series TEXT NOT NULL DEFAULT ''`,
209 `ALTER TABLE activity_log ADD COLUMN series_url TEXT NOT NULL DEFAULT ''`,
210 `ALTER TABLE activity_log ADD COLUMN forgejo_id INTEGER NOT NULL DEFAULT 0`,
211 `ALTER TABLE activity_log ADD COLUMN radicle_id TEXT NOT NULL DEFAULT ''`,
212 `ALTER TABLE activity_log ADD COLUMN origin TEXT NOT NULL DEFAULT ''`,
213 `ALTER TABLE dynamic_repo ADD COLUMN approved INTEGER NOT NULL DEFAULT 0`,
214 } {
215 if _, err := db.Exec(stmt); err != nil && !strings.Contains(err.Error(), "duplicate column") {
216 return err
217 }
218 }
219 // Backfill: a dynamic_repo row that was already materialized before
220 // the approved column existed was live and trusted under the old
221 // model — the ALTER above defaults it to approved=0 along with every
222 // other row, which would silently pull an already-active peer back
223 // out of the sync loop on the next restart. Idempotent (materialized
224 // never reverts to 0), so safe to run unconditionally every startup,
225 // not just the first one after the column was added.
226 if _, err := db.Exec(`UPDATE dynamic_repo SET approved = 1 WHERE materialized = 1 AND approved = 0`); err != nil {
227 return err
228 }
229 return nil
230}
231
232// GitCursor returns the last-synced commit SHAs on each side for a repo pair.
233func (s *Store) GitCursor(repoPair string) (forgejoSHA, radicleSHA string, err error) {
234 row := s.db.QueryRow(`SELECT forgejo_sha, radicle_sha FROM git_cursor WHERE repo_pair = ?`, repoPair)
235 err = row.Scan(&forgejoSHA, &radicleSHA)
236 if err == sql.ErrNoRows {
237 return "", "", nil
238 }
239 return forgejoSHA, radicleSHA, err
240}
241
242// SetGitCursor records the commit SHAs last observed on each side.
243func (s *Store) SetGitCursor(repoPair, forgejoSHA, radicleSHA string) error {
244 _, err := s.db.Exec(`
245INSERT INTO git_cursor (repo_pair, forgejo_sha, radicle_sha) VALUES (?, ?, ?)
246ON CONFLICT (repo_pair) DO UPDATE SET forgejo_sha = excluded.forgejo_sha, radicle_sha = excluded.radicle_sha
247`, repoPair, forgejoSHA, radicleSHA)
248 return err
249}
250
251// ItemMapping records the correspondence between a Forgejo issue/PR and its
252// mirrored Radicle issue/patch, plus a content hash used to detect edits
253// that need re-syncing.
254type ItemMapping struct {
255 Kind string // "issue" or "patch"
256 ForgejoID int64
257 RadicleID string
258 ContentHash string
259}
260
261// FindByForgejoID looks up an existing mapping created for a Forgejo item.
262func (s *Store) FindByForgejoID(repoPair, kind string, forgejoID int64) (*ItemMapping, error) {
263 row := s.db.QueryRow(`
264SELECT radicle_id, content_hash FROM item_mapping
265WHERE repo_pair = ? AND kind = ? AND forgejo_id = ?
266`, repoPair, kind, forgejoID)
267 m := &ItemMapping{Kind: kind, ForgejoID: forgejoID}
268 err := row.Scan(&m.RadicleID, &m.ContentHash)
269 if err == sql.ErrNoRows {
270 return nil, nil
271 }
272 return m, err
273}
274
275// FindByRadicleID looks up an existing mapping created for a Radicle item.
276func (s *Store) FindByRadicleID(repoPair, kind, radicleID string) (*ItemMapping, error) {
277 row := s.db.QueryRow(`
278SELECT forgejo_id, content_hash FROM item_mapping
279WHERE repo_pair = ? AND kind = ? AND radicle_id = ?
280`, repoPair, kind, radicleID)
281 m := &ItemMapping{Kind: kind, RadicleID: radicleID}
282 err := row.Scan(&m.ForgejoID, &m.ContentHash)
283 if err == sql.ErrNoRows {
284 return nil, nil
285 }
286 return m, err
287}
288
289// Upsert records or updates a mapping between a Forgejo item and its
290// Radicle counterpart.
291func (s *Store) Upsert(repoPair string, m ItemMapping) error {
292 _, err := s.db.Exec(`
293INSERT INTO item_mapping (repo_pair, kind, forgejo_id, radicle_id, content_hash, synced_at)
294VALUES (?, ?, ?, ?, ?, datetime('now'))
295ON CONFLICT (repo_pair, kind, forgejo_id, radicle_id)
296DO UPDATE SET content_hash = excluded.content_hash, synced_at = excluded.synced_at
297`, repoPair, m.Kind, m.ForgejoID, m.RadicleID, m.ContentHash)
298 return err
299}
300
301// Activity is one thing that was actually mirrored, for the status page.
302type Activity struct {
303 RepoPair string // the sync pair that did it, e.g. "graft-test-v2-f1"
304 Series string // the human-facing repo group this pair belongs to,
305 // e.g. "graft-test-v2" — several pairs (one per forge/Radicle side)
306 // can share a series so the dashboard shows one row per repo, not
307 // one per pair.
308 SeriesURL string // where clicking the series' own name should go
309 Kind string // "git", "issue", "patch"
310 Direction string // ForgejoToRadicle or RadicleToForgejo
311 Summary string
312 URL string // where to send someone who clicks this entry — the
313 // forge or Radicle page for the thing that now exists because of
314 // this sync; "" if none is known.
315 ForgejoID int64 // issue/PR number on Forgejo, for kind "issue"/"patch"; 0 for "git"
316 RadicleID string // issue/patch id on Radicle, for kind "issue"/"patch"; "" for "git"
317 // Origin names where a "comment" kind entry actually happened —
318 // "forgejo", "radicle", "activitypub", or "atproto". Empty for every
319 // other kind. Display-only: shown in the dashboard tooltip, never
320 // drives a cell's color (see status.event.Origin).
321 Origin string
322}
323
324// Direction values for Activity.Direction, kept as constants so callers
325// can't typo a value the status page's rendering silently fails to
326// recognize.
327const (
328 ForgejoToRadicle = "forgejo_to_radicle"
329 RadicleToForgejo = "radicle_to_forgejo"
330)
331
332// LogActivity records one Activity and returns its new row id — used by
333// the Bluesky bridge to link a posted note back to what it was about, so
334// a later reply can be traced to the right item. Call it once per
335// commit/issue/patch/comment, not once per sync pass.
336func (s *Store) LogActivity(a Activity) (int64, error) {
337 res, err := s.db.Exec(`
338INSERT INTO activity_log (repo_pair, kind, direction, summary, url, series, series_url, forgejo_id, radicle_id, origin, occurred_at)
339VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
340`, a.RepoPair, a.Kind, a.Direction, a.Summary, a.URL, a.Series, a.SeriesURL, a.ForgejoID, a.RadicleID, a.Origin, time.Now().UTC().Format(time.RFC3339))
341 if err != nil {
342 return 0, err
343 }
344 return res.LastInsertId()
345}
346
347// ActivityEntry is one row of activity_log.
348type ActivityEntry struct {
349 ID int64 // activity_log.id, used as the stable suffix of ActivityPub Note URIs
350 Activity
351 OccurredAt time.Time
352}
353
354// ActivitySince returns every activity entry at or after since, newest first.
355func (s *Store) ActivitySince(since time.Time) ([]ActivityEntry, error) {
356 rows, err := s.db.Query(`
357SELECT id, repo_pair, kind, direction, summary, url, series, series_url, forgejo_id, radicle_id, origin, occurred_at FROM activity_log
358WHERE occurred_at >= ?
359ORDER BY occurred_at DESC
360`, since.UTC().Format(time.RFC3339))
361 if err != nil {
362 return nil, err
363 }
364 return scanActivityRows(rows)
365}
366
367// ActivityForSeries returns the most recent activity entries for one
368// series, newest first, capped at limit — used to build that series'
369// ActivityPub outbox.
370func (s *Store) ActivityForSeries(series string, limit int) ([]ActivityEntry, error) {
371 rows, err := s.db.Query(`
372SELECT id, repo_pair, kind, direction, summary, url, series, series_url, forgejo_id, radicle_id, origin, occurred_at FROM activity_log
373WHERE series = ?
374ORDER BY occurred_at DESC
375LIMIT ?
376`, series, limit)
377 if err != nil {
378 return nil, err
379 }
380 return scanActivityRows(rows)
381}
382
383// ActivityByID looks up one activity_log row by id, e.g. to resolve an
384// ActivityPub Note URI (/actors/{series}/notes/{id}) back to the mirrored
385// item it describes.
386func (s *Store) ActivityByID(id int64) (*ActivityEntry, error) {
387 row := s.db.QueryRow(`
388SELECT id, repo_pair, kind, direction, summary, url, series, series_url, forgejo_id, radicle_id, origin, occurred_at FROM activity_log
389WHERE id = ?
390`, id)
391 var e ActivityEntry
392 var occurredAt string
393 err := row.Scan(&e.ID, &e.RepoPair, &e.Kind, &e.Direction, &e.Summary, &e.URL, &e.Series, &e.SeriesURL, &e.ForgejoID, &e.RadicleID, &e.Origin, &occurredAt)
394 if err == sql.ErrNoRows {
395 return nil, nil
396 }
397 if err != nil {
398 return nil, err
399 }
400 e.OccurredAt, err = time.Parse(time.RFC3339, occurredAt)
401 if err != nil {
402 return nil, fmt.Errorf("parse occurred_at %q: %w", occurredAt, err)
403 }
404 return &e, nil
405}
406
407func scanActivityRows(rows *sql.Rows) ([]ActivityEntry, error) {
408 defer rows.Close()
409 var out []ActivityEntry
410 for rows.Next() {
411 var e ActivityEntry
412 var occurredAt string
413 if err := rows.Scan(&e.ID, &e.RepoPair, &e.Kind, &e.Direction, &e.Summary, &e.URL, &e.Series, &e.SeriesURL, &e.ForgejoID, &e.RadicleID, &e.Origin, &occurredAt); err != nil {
414 return nil, err
415 }
416 var err error
417 e.OccurredAt, err = time.Parse(time.RFC3339, occurredAt)
418 if err != nil {
419 return nil, fmt.Errorf("parse occurred_at %q: %w", occurredAt, err)
420 }
421 out = append(out, e)
422 }
423 return out, rows.Err()
424}
425
426// ActorKeys is one series' ActivityPub actor keypair, PEM-encoded.
427type ActorKeys struct {
428 PrivateKeyPEM string
429 PublicKeyPEM string
430}
431
432// ActorKeys returns the stored keypair for a series, or nil if none has
433// been generated yet.
434func (s *Store) ActorKeys(series string) (*ActorKeys, error) {
435 row := s.db.QueryRow(`SELECT private_key, public_key FROM ap_actor WHERE series = ?`, series)
436 var k ActorKeys
437 err := row.Scan(&k.PrivateKeyPEM, &k.PublicKeyPEM)
438 if err == sql.ErrNoRows {
439 return nil, nil
440 }
441 if err != nil {
442 return nil, err
443 }
444 return &k, nil
445}
446
447// SaveActorKeys stores a newly generated keypair for a series. Safe to call
448// concurrently: if another goroutine already inserted one, this becomes a
449// no-op via INSERT OR IGNORE, and the caller should re-read with
450// ActorKeys to get whichever copy won.
451func (s *Store) SaveActorKeys(series string, k ActorKeys) error {
452 _, err := s.db.Exec(`
453INSERT OR IGNORE INTO ap_actor (series, private_key, public_key, created_at)
454VALUES (?, ?, ?, ?)
455`, series, k.PrivateKeyPEM, k.PublicKeyPEM, time.Now().UTC().Format(time.RFC3339))
456 return err
457}
458
459// Follower is one remote actor following a series' ActivityPub actor.
460type Follower struct {
461 ActorURI string
462 InboxURI string
463}
464
465// SaveFollower records (or updates the inbox URL of) a follower.
466func (s *Store) SaveFollower(series, actorURI, inboxURI string) error {
467 _, err := s.db.Exec(`
468INSERT INTO ap_follower (series, actor_uri, inbox_uri) VALUES (?, ?, ?)
469ON CONFLICT (series, actor_uri) DO UPDATE SET inbox_uri = excluded.inbox_uri
470`, series, actorURI, inboxURI)
471 return err
472}
473
474// RemoveFollower drops a follower, e.g. on Undo{Follow}.
475func (s *Store) RemoveFollower(series, actorURI string) error {
476 _, err := s.db.Exec(`DELETE FROM ap_follower WHERE series = ? AND actor_uri = ?`, series, actorURI)
477 return err
478}
479
480// Followers lists everyone following a series.
481func (s *Store) Followers(series string) ([]Follower, error) {
482 rows, err := s.db.Query(`SELECT actor_uri, inbox_uri FROM ap_follower WHERE series = ?`, series)
483 if err != nil {
484 return nil, err
485 }
486 defer rows.Close()
487 var out []Follower
488 for rows.Next() {
489 var f Follower
490 if err := rows.Scan(&f.ActorURI, &f.InboxURI); err != nil {
491 return nil, err
492 }
493 out = append(out, f)
494 }
495 return out, rows.Err()
496}
497
498// FollowerCount is len(Followers(series)) without building the slice.
499func (s *Store) FollowerCount(series string) (int, error) {
500 row := s.db.QueryRow(`SELECT COUNT(*) FROM ap_follower WHERE series = ?`, series)
501 var n int
502 err := row.Scan(&n)
503 return n, err
504}
505
506// SeriesWithFollowers lists every series that has at least one follower —
507// the set worth checking for new activity to deliver.
508func (s *Store) SeriesWithFollowers() ([]string, error) {
509 rows, err := s.db.Query(`SELECT DISTINCT series FROM ap_follower`)
510 if err != nil {
511 return nil, err
512 }
513 defer rows.Close()
514 var out []string
515 for rows.Next() {
516 var series string
517 if err := rows.Scan(&series); err != nil {
518 return nil, err
519 }
520 out = append(out, series)
521 }
522 return out, rows.Err()
523}
524
525// DeliveryCursor returns how far ActivityPub delivery has gotten through
526// activity_log for a series (0 if nothing has been delivered yet).
527func (s *Store) DeliveryCursor(series string) (int64, error) {
528 row := s.db.QueryRow(`SELECT last_delivered_id FROM ap_delivery_cursor WHERE series = ?`, series)
529 var id int64
530 err := row.Scan(&id)
531 if err == sql.ErrNoRows {
532 return 0, nil
533 }
534 return id, err
535}
536
537// SetDeliveryCursor records how far delivery has gotten for a series.
538func (s *Store) SetDeliveryCursor(series string, id int64) error {
539 _, err := s.db.Exec(`
540INSERT INTO ap_delivery_cursor (series, last_delivered_id) VALUES (?, ?)
541ON CONFLICT (series) DO UPDATE SET last_delivered_id = excluded.last_delivered_id
542`, series, id)
543 return err
544}
545
546// NewActivityForSeries returns activity_log entries for a series with
547// id > afterID, oldest first — the delivery order for ActivityPub.
548func (s *Store) NewActivityForSeries(series string, afterID int64) ([]ActivityEntry, error) {
549 rows, err := s.db.Query(`
550SELECT id, repo_pair, kind, direction, summary, url, series, series_url, forgejo_id, radicle_id, origin, occurred_at FROM activity_log
551WHERE series = ? AND id > ?
552ORDER BY id ASC
553`, series, afterID)
554 if err != nil {
555 return nil, err
556 }
557 return scanActivityRows(rows)
558}
559
560// TokenShare is one encrypted one-time secret. Nonce and Ciphertext are
561// base64-encoded AES-256-GCM output; the decryption key is derived from
562// the passcode the claimant supplies plus the share's own id (which is
563// never stored — only IDHash is), so a database dump alone never yields
564// enough to decrypt.
565type TokenShare struct {
566 IDHash string
567 Nonce string
568 Ciphertext string
569 Attempts int
570 ExpiresAt time.Time
571}
572
573// CreateTokenShare stores a new encrypted share, indexed by the hash of
574// its id (the plaintext id is only ever known to whoever holds the URL).
575func (s *Store) CreateTokenShare(idHash, nonce, ciphertext string, expiresAt time.Time) error {
576 _, err := s.db.Exec(`
577INSERT INTO token_share (id_hash, nonce, ciphertext, attempts, created_at, expires_at)
578VALUES (?, ?, ?, 0, ?, ?)
579`, idHash, nonce, ciphertext, time.Now().UTC().Format(time.RFC3339), expiresAt.UTC().Format(time.RFC3339))
580 return err
581}
582
583// GetTokenShare looks up a share by id hash. Returns nil, nil if it
584// doesn't exist or has already expired (expiry is checked here rather
585// than left to a background sweep, so a stale row can never be claimed
586// even if cleanup hasn't run yet).
587func (s *Store) GetTokenShare(idHash string) (*TokenShare, error) {
588 row := s.db.QueryRow(`SELECT id_hash, nonce, ciphertext, attempts, expires_at FROM token_share WHERE id_hash = ?`, idHash)
589 var t TokenShare
590 var expiresAt string
591 if err := row.Scan(&t.IDHash, &t.Nonce, &t.Ciphertext, &t.Attempts, &expiresAt); err != nil {
592 if err == sql.ErrNoRows {
593 return nil, nil
594 }
595 return nil, err
596 }
597 exp, err := time.Parse(time.RFC3339, expiresAt)
598 if err != nil {
599 return nil, err
600 }
601 t.ExpiresAt = exp
602 if time.Now().UTC().After(exp) {
603 _ = s.DeleteTokenShare(idHash)
604 return nil, nil
605 }
606 return &t, nil
607}
608
609// IncrementTokenShareAttempts records one failed passcode attempt and
610// returns the new total, so the caller can destroy the share once too
611// many wrong guesses have been made — the actual brute-force defense,
612// since a 6-digit passcode alone is far too small a space to rely on
613// key-derivation cost against a determined online attacker.
614func (s *Store) IncrementTokenShareAttempts(idHash string) (int, error) {
615 _, err := s.db.Exec(`UPDATE token_share SET attempts = attempts + 1 WHERE id_hash = ?`, idHash)
616 if err != nil {
617 return 0, err
618 }
619 row := s.db.QueryRow(`SELECT attempts FROM token_share WHERE id_hash = ?`, idHash)
620 var n int
621 if err := row.Scan(&n); err != nil {
622 return 0, err
623 }
624 return n, nil
625}
626
627// DeleteTokenShare destroys a share — called on a successful claim (it's
628// one-time by design), on exceeding the attempt limit (fail closed), or
629// on expiry.
630func (s *Store) DeleteTokenShare(idHash string) error {
631 _, err := s.db.Exec(`DELETE FROM token_share WHERE id_hash = ?`, idHash)
632 return err
633}
634
635// DynamicRepo is one repo pair added through the dashboard's self-service
636// onboarding forms rather than a config.yaml edit.
637type DynamicRepo struct {
638 ID int64
639 Name string
640 Series string
641 ForgejoBaseURL string
642 ForgejoOwner string
643 ForgejoRepo string
644 ForgejoTokenFile string
645 RadicleRID string
646 RadicleHTTPBaseURL string
647 RadicleExplorerURL string
648 RadicleRadHome string
649 SyncGit bool
650 SyncIssues bool
651 SyncPatches bool
652 Materialized bool
653 Approved bool
654}
655
656// CreateDynamicRepo inserts a newly-submitted repo pair as an unapproved
657// request — it never joins the sync loop until ApproveDynamicRepo is
658// called. Fails if the name is already taken — the same uniqueness
659// graft would need across config.yaml pairs anyway.
660func (s *Store) CreateDynamicRepo(r DynamicRepo) (int64, error) {
661 res, err := s.db.Exec(`
662INSERT INTO dynamic_repo (
663 name, series, forgejo_base_url, forgejo_owner, forgejo_repo, forgejo_token_file,
664 radicle_rid, radicle_http_base_url, radicle_explorer_url, radicle_rad_home,
665 sync_git, sync_issues, sync_patches, created_at, materialized, approved
666) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0, 0)
667`, r.Name, r.Series, r.ForgejoBaseURL, r.ForgejoOwner, r.ForgejoRepo, r.ForgejoTokenFile,
668 r.RadicleRID, r.RadicleHTTPBaseURL, r.RadicleExplorerURL, r.RadicleRadHome,
669 boolToInt(r.SyncGit), boolToInt(r.SyncIssues), boolToInt(r.SyncPatches), time.Now().UTC().Format(time.RFC3339))
670 if err != nil {
671 return 0, err
672 }
673 return res.LastInsertId()
674}
675
676const dynamicRepoColumns = `id, name, series, forgejo_base_url, forgejo_owner, forgejo_repo, forgejo_token_file,
677 radicle_rid, radicle_http_base_url, radicle_explorer_url, radicle_rad_home,
678 sync_git, sync_issues, sync_patches, materialized, approved`
679
680func scanDynamicRepoRows(rows *sql.Rows) ([]DynamicRepo, error) {
681 defer rows.Close()
682 var out []DynamicRepo
683 for rows.Next() {
684 var r DynamicRepo
685 var g, i, p, m, a int
686 if err := rows.Scan(&r.ID, &r.Name, &r.Series, &r.ForgejoBaseURL, &r.ForgejoOwner, &r.ForgejoRepo, &r.ForgejoTokenFile,
687 &r.RadicleRID, &r.RadicleHTTPBaseURL, &r.RadicleExplorerURL, &r.RadicleRadHome, &g, &i, &p, &m, &a); err != nil {
688 return nil, err
689 }
690 r.SyncGit, r.SyncIssues, r.SyncPatches, r.Materialized, r.Approved = g != 0, i != 0, p != 0, m != 0, a != 0
691 out = append(out, r)
692 }
693 return out, rows.Err()
694}
695
696// PendingDynamicRepos returns every submitted peer/repo request still
697// awaiting admin approval.
698func (s *Store) PendingDynamicRepos() ([]DynamicRepo, error) {
699 rows, err := s.db.Query(`SELECT ` + dynamicRepoColumns + ` FROM dynamic_repo WHERE approved = 0`)
700 if err != nil {
701 return nil, err
702 }
703 return scanDynamicRepoRows(rows)
704}
705
706// GetDynamicRepo looks up one row by id, e.g. to recover its token file
707// path before deleting a rejected request.
708func (s *Store) GetDynamicRepo(id int64) (*DynamicRepo, error) {
709 rows, err := s.db.Query(`SELECT `+dynamicRepoColumns+` FROM dynamic_repo WHERE id = ?`, id)
710 if err != nil {
711 return nil, err
712 }
713 out, err := scanDynamicRepoRows(rows)
714 if err != nil || len(out) == 0 {
715 return nil, err
716 }
717 return &out[0], nil
718}
719
720// ApproveDynamicRepo marks a pending request approved — the next sync
721// pass turns it into a live RepoSyncer (see UnmaterializedDynamicRepos).
722func (s *Store) ApproveDynamicRepo(id int64) error {
723 _, err := s.db.Exec(`UPDATE dynamic_repo SET approved = 1 WHERE id = ?`, id)
724 return err
725}
726
727// RejectDynamicRepo permanently removes a pending request.
728func (s *Store) RejectDynamicRepo(id int64) error {
729 _, err := s.db.Exec(`DELETE FROM dynamic_repo WHERE id = ?`, id)
730 return err
731}
732
733// UnmaterializedDynamicRepos returns every approved dynamic_repo row the
734// running daemon hasn't yet turned into a live RepoSyncer.
735func (s *Store) UnmaterializedDynamicRepos() ([]DynamicRepo, error) {
736 rows, err := s.db.Query(`SELECT ` + dynamicRepoColumns + ` FROM dynamic_repo WHERE approved = 1 AND materialized = 0`)
737 if err != nil {
738 return nil, err
739 }
740 return scanDynamicRepoRows(rows)
741}
742
743// MarkDynamicRepoMaterialized flips a row so it's never re-registered on
744// a later pass.
745func (s *Store) MarkDynamicRepoMaterialized(id int64) error {
746 _, err := s.db.Exec(`UPDATE dynamic_repo SET materialized = 1 WHERE id = ?`, id)
747 return err
748}
749
750// AllDynamicRepos returns every approved dynamic_repo row regardless of
751// materialization state — used at startup to rebuild the full set of
752// live pairs after a restart, since these rows persist across it but
753// config.yaml doesn't know about them. Never includes a row still
754// awaiting approval.
755func (s *Store) AllDynamicRepos() ([]DynamicRepo, error) {
756 rows, err := s.db.Query(`SELECT ` + dynamicRepoColumns + ` FROM dynamic_repo WHERE approved = 1`)
757 if err != nil {
758 return nil, err
759 }
760 return scanDynamicRepoRows(rows)
761}
762
763func boolToInt(b bool) int {
764 if b {
765 return 1
766 }
767 return 0
768}
769
770// CommentSeen reports whether a comment body's hash has already been
771// dealt with (mirrored, or recognized as already present on both sides)
772// for one item, so a comment sync pass never re-posts the same content
773// back and forth between Forgejo and Radicle.
774func (s *Store) CommentSeen(repoPair, kind, itemID, hash string) (bool, error) {
775 row := s.db.QueryRow(`SELECT 1 FROM comment_seen WHERE repo_pair = ? AND kind = ? AND item_id = ? AND hash = ?`, repoPair, kind, itemID, hash)
776 var one int
777 err := row.Scan(&one)
778 if err == sql.ErrNoRows {
779 return false, nil
780 }
781 return err == nil, err
782}
783
784// MarkCommentSeen records a comment body's hash as dealt with.
785func (s *Store) MarkCommentSeen(repoPair, kind, itemID, hash string) error {
786 _, err := s.db.Exec(`INSERT OR IGNORE INTO comment_seen (repo_pair, kind, item_id, hash) VALUES (?, ?, ?, ?)`, repoPair, kind, itemID, hash)
787 return err
788}
789
790// SaveATProtoPost records a Bluesky post graft made, so a later reply's
791// parent URI can be traced back to the activity_log entry it was about.
792func (s *Store) SaveATProtoPost(uri string, activityID int64, series string) error {
793 _, err := s.db.Exec(`INSERT OR IGNORE INTO atproto_post (uri, activity_id, series, created_at) VALUES (?, ?, ?, ?)`,
794 uri, activityID, series, time.Now().UTC().Format(time.RFC3339))
795 return err
796}
797
798// ATProtoPostActivity looks up which activity_log entry a Bluesky post
799// URI was about, for resolving an inbound reply back to the underlying
800// Forgejo/Radicle item.
801func (s *Store) ATProtoPostActivity(uri string) (int64, bool, error) {
802 row := s.db.QueryRow(`SELECT activity_id FROM atproto_post WHERE uri = ?`, uri)
803 var id int64
804 err := row.Scan(&id)
805 if err == sql.ErrNoRows {
806 return 0, false, nil
807 }
808 return id, err == nil, err
809}
810
811// ATProtoCursor returns how far notification polling has gotten for a
812// series' Bluesky account ("" if never polled).
813func (s *Store) ATProtoCursor(series string) (string, error) {
814 row := s.db.QueryRow(`SELECT cursor FROM atproto_cursor WHERE series = ?`, series)
815 var c string
816 err := row.Scan(&c)
817 if err == sql.ErrNoRows {
818 return "", nil
819 }
820 return c, err
821}
822
823// SetATProtoCursor records how far notification polling has gotten.
824func (s *Store) SetATProtoCursor(series, cursor string) error {
825 _, err := s.db.Exec(`
826INSERT INTO atproto_cursor (series, cursor) VALUES (?, ?)
827ON CONFLICT (series) DO UPDATE SET cursor = excluded.cursor
828`, series, cursor)
829 return err
830}
831
832// PendingRadiclePeer is one submitted "add a Radicle peer" request
833// awaiting admin approval.
834type PendingRadiclePeer struct {
835 ID int64
836 Series string
837 NodeID string
838 Address string
839}
840
841// CreatePendingRadiclePeer records a submitted request.
842func (s *Store) CreatePendingRadiclePeer(series, nodeID, address string) (int64, error) {
843 res, err := s.db.Exec(`INSERT INTO pending_radicle_peer (series, node_id, address, created_at) VALUES (?, ?, ?, ?)`,
844 series, nodeID, address, time.Now().UTC().Format(time.RFC3339))
845 if err != nil {
846 return 0, err
847 }
848 return res.LastInsertId()
849}
850
851// ListPendingRadiclePeers returns every request awaiting approval.
852func (s *Store) ListPendingRadiclePeers() ([]PendingRadiclePeer, error) {
853 rows, err := s.db.Query(`SELECT id, series, node_id, address FROM pending_radicle_peer`)
854 if err != nil {
855 return nil, err
856 }
857 defer rows.Close()
858 var out []PendingRadiclePeer
859 for rows.Next() {
860 var p PendingRadiclePeer
861 if err := rows.Scan(&p.ID, &p.Series, &p.NodeID, &p.Address); err != nil {
862 return nil, err
863 }
864 out = append(out, p)
865 }
866 return out, rows.Err()
867}
868
869// GetPendingRadiclePeer looks up one request by id.
870func (s *Store) GetPendingRadiclePeer(id int64) (*PendingRadiclePeer, error) {
871 row := s.db.QueryRow(`SELECT id, series, node_id, address FROM pending_radicle_peer WHERE id = ?`, id)
872 var p PendingRadiclePeer
873 err := row.Scan(&p.ID, &p.Series, &p.NodeID, &p.Address)
874 if err == sql.ErrNoRows {
875 return nil, nil
876 }
877 if err != nil {
878 return nil, err
879 }
880 return &p, nil
881}
882
883// DeletePendingRadiclePeer removes a request — called after it's been
884// approved and acted on, or rejected.
885func (s *Store) DeletePendingRadiclePeer(id int64) error {
886 _, err := s.db.Exec(`DELETE FROM pending_radicle_peer WHERE id = ?`, id)
887 return err
888}