Federated via graft
32 kB
878 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 return nil
220}
221
222// GitCursor returns the last-synced commit SHAs on each side for a repo pair.
223func (s *Store) GitCursor(repoPair string) (forgejoSHA, radicleSHA string, err error) {
224 row := s.db.QueryRow(`SELECT forgejo_sha, radicle_sha FROM git_cursor WHERE repo_pair = ?`, repoPair)
225 err = row.Scan(&forgejoSHA, &radicleSHA)
226 if err == sql.ErrNoRows {
227 return "", "", nil
228 }
229 return forgejoSHA, radicleSHA, err
230}
231
232// SetGitCursor records the commit SHAs last observed on each side.
233func (s *Store) SetGitCursor(repoPair, forgejoSHA, radicleSHA string) error {
234 _, err := s.db.Exec(`
235INSERT INTO git_cursor (repo_pair, forgejo_sha, radicle_sha) VALUES (?, ?, ?)
236ON CONFLICT (repo_pair) DO UPDATE SET forgejo_sha = excluded.forgejo_sha, radicle_sha = excluded.radicle_sha
237`, repoPair, forgejoSHA, radicleSHA)
238 return err
239}
240
241// ItemMapping records the correspondence between a Forgejo issue/PR and its
242// mirrored Radicle issue/patch, plus a content hash used to detect edits
243// that need re-syncing.
244type ItemMapping struct {
245 Kind string // "issue" or "patch"
246 ForgejoID int64
247 RadicleID string
248 ContentHash string
249}
250
251// FindByForgejoID looks up an existing mapping created for a Forgejo item.
252func (s *Store) FindByForgejoID(repoPair, kind string, forgejoID int64) (*ItemMapping, error) {
253 row := s.db.QueryRow(`
254SELECT radicle_id, content_hash FROM item_mapping
255WHERE repo_pair = ? AND kind = ? AND forgejo_id = ?
256`, repoPair, kind, forgejoID)
257 m := &ItemMapping{Kind: kind, ForgejoID: forgejoID}
258 err := row.Scan(&m.RadicleID, &m.ContentHash)
259 if err == sql.ErrNoRows {
260 return nil, nil
261 }
262 return m, err
263}
264
265// FindByRadicleID looks up an existing mapping created for a Radicle item.
266func (s *Store) FindByRadicleID(repoPair, kind, radicleID string) (*ItemMapping, error) {
267 row := s.db.QueryRow(`
268SELECT forgejo_id, content_hash FROM item_mapping
269WHERE repo_pair = ? AND kind = ? AND radicle_id = ?
270`, repoPair, kind, radicleID)
271 m := &ItemMapping{Kind: kind, RadicleID: radicleID}
272 err := row.Scan(&m.ForgejoID, &m.ContentHash)
273 if err == sql.ErrNoRows {
274 return nil, nil
275 }
276 return m, err
277}
278
279// Upsert records or updates a mapping between a Forgejo item and its
280// Radicle counterpart.
281func (s *Store) Upsert(repoPair string, m ItemMapping) error {
282 _, err := s.db.Exec(`
283INSERT INTO item_mapping (repo_pair, kind, forgejo_id, radicle_id, content_hash, synced_at)
284VALUES (?, ?, ?, ?, ?, datetime('now'))
285ON CONFLICT (repo_pair, kind, forgejo_id, radicle_id)
286DO UPDATE SET content_hash = excluded.content_hash, synced_at = excluded.synced_at
287`, repoPair, m.Kind, m.ForgejoID, m.RadicleID, m.ContentHash)
288 return err
289}
290
291// Activity is one thing that was actually mirrored, for the status page.
292type Activity struct {
293 RepoPair string // the sync pair that did it, e.g. "graft-test-v2-f1"
294 Series string // the human-facing repo group this pair belongs to,
295 // e.g. "graft-test-v2" — several pairs (one per forge/Radicle side)
296 // can share a series so the dashboard shows one row per repo, not
297 // one per pair.
298 SeriesURL string // where clicking the series' own name should go
299 Kind string // "git", "issue", "patch"
300 Direction string // ForgejoToRadicle or RadicleToForgejo
301 Summary string
302 URL string // where to send someone who clicks this entry — the
303 // forge or Radicle page for the thing that now exists because of
304 // this sync; "" if none is known.
305 ForgejoID int64 // issue/PR number on Forgejo, for kind "issue"/"patch"; 0 for "git"
306 RadicleID string // issue/patch id on Radicle, for kind "issue"/"patch"; "" for "git"
307 // Origin names where a "comment" kind entry actually happened —
308 // "forgejo", "radicle", "activitypub", or "atproto". Empty for every
309 // other kind. Display-only: shown in the dashboard tooltip, never
310 // drives a cell's color (see status.event.Origin).
311 Origin string
312}
313
314// Direction values for Activity.Direction, kept as constants so callers
315// can't typo a value the status page's rendering silently fails to
316// recognize.
317const (
318 ForgejoToRadicle = "forgejo_to_radicle"
319 RadicleToForgejo = "radicle_to_forgejo"
320)
321
322// LogActivity records one Activity and returns its new row id — used by
323// the Bluesky bridge to link a posted note back to what it was about, so
324// a later reply can be traced to the right item. Call it once per
325// commit/issue/patch/comment, not once per sync pass.
326func (s *Store) LogActivity(a Activity) (int64, error) {
327 res, err := s.db.Exec(`
328INSERT INTO activity_log (repo_pair, kind, direction, summary, url, series, series_url, forgejo_id, radicle_id, origin, occurred_at)
329VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
330`, 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))
331 if err != nil {
332 return 0, err
333 }
334 return res.LastInsertId()
335}
336
337// ActivityEntry is one row of activity_log.
338type ActivityEntry struct {
339 ID int64 // activity_log.id, used as the stable suffix of ActivityPub Note URIs
340 Activity
341 OccurredAt time.Time
342}
343
344// ActivitySince returns every activity entry at or after since, newest first.
345func (s *Store) ActivitySince(since time.Time) ([]ActivityEntry, error) {
346 rows, err := s.db.Query(`
347SELECT id, repo_pair, kind, direction, summary, url, series, series_url, forgejo_id, radicle_id, origin, occurred_at FROM activity_log
348WHERE occurred_at >= ?
349ORDER BY occurred_at DESC
350`, since.UTC().Format(time.RFC3339))
351 if err != nil {
352 return nil, err
353 }
354 return scanActivityRows(rows)
355}
356
357// ActivityForSeries returns the most recent activity entries for one
358// series, newest first, capped at limit — used to build that series'
359// ActivityPub outbox.
360func (s *Store) ActivityForSeries(series string, limit int) ([]ActivityEntry, error) {
361 rows, err := s.db.Query(`
362SELECT id, repo_pair, kind, direction, summary, url, series, series_url, forgejo_id, radicle_id, origin, occurred_at FROM activity_log
363WHERE series = ?
364ORDER BY occurred_at DESC
365LIMIT ?
366`, series, limit)
367 if err != nil {
368 return nil, err
369 }
370 return scanActivityRows(rows)
371}
372
373// ActivityByID looks up one activity_log row by id, e.g. to resolve an
374// ActivityPub Note URI (/actors/{series}/notes/{id}) back to the mirrored
375// item it describes.
376func (s *Store) ActivityByID(id int64) (*ActivityEntry, error) {
377 row := s.db.QueryRow(`
378SELECT id, repo_pair, kind, direction, summary, url, series, series_url, forgejo_id, radicle_id, origin, occurred_at FROM activity_log
379WHERE id = ?
380`, id)
381 var e ActivityEntry
382 var occurredAt string
383 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)
384 if err == sql.ErrNoRows {
385 return nil, nil
386 }
387 if err != nil {
388 return nil, err
389 }
390 e.OccurredAt, err = time.Parse(time.RFC3339, occurredAt)
391 if err != nil {
392 return nil, fmt.Errorf("parse occurred_at %q: %w", occurredAt, err)
393 }
394 return &e, nil
395}
396
397func scanActivityRows(rows *sql.Rows) ([]ActivityEntry, error) {
398 defer rows.Close()
399 var out []ActivityEntry
400 for rows.Next() {
401 var e ActivityEntry
402 var occurredAt string
403 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 {
404 return nil, err
405 }
406 var err error
407 e.OccurredAt, err = time.Parse(time.RFC3339, occurredAt)
408 if err != nil {
409 return nil, fmt.Errorf("parse occurred_at %q: %w", occurredAt, err)
410 }
411 out = append(out, e)
412 }
413 return out, rows.Err()
414}
415
416// ActorKeys is one series' ActivityPub actor keypair, PEM-encoded.
417type ActorKeys struct {
418 PrivateKeyPEM string
419 PublicKeyPEM string
420}
421
422// ActorKeys returns the stored keypair for a series, or nil if none has
423// been generated yet.
424func (s *Store) ActorKeys(series string) (*ActorKeys, error) {
425 row := s.db.QueryRow(`SELECT private_key, public_key FROM ap_actor WHERE series = ?`, series)
426 var k ActorKeys
427 err := row.Scan(&k.PrivateKeyPEM, &k.PublicKeyPEM)
428 if err == sql.ErrNoRows {
429 return nil, nil
430 }
431 if err != nil {
432 return nil, err
433 }
434 return &k, nil
435}
436
437// SaveActorKeys stores a newly generated keypair for a series. Safe to call
438// concurrently: if another goroutine already inserted one, this becomes a
439// no-op via INSERT OR IGNORE, and the caller should re-read with
440// ActorKeys to get whichever copy won.
441func (s *Store) SaveActorKeys(series string, k ActorKeys) error {
442 _, err := s.db.Exec(`
443INSERT OR IGNORE INTO ap_actor (series, private_key, public_key, created_at)
444VALUES (?, ?, ?, ?)
445`, series, k.PrivateKeyPEM, k.PublicKeyPEM, time.Now().UTC().Format(time.RFC3339))
446 return err
447}
448
449// Follower is one remote actor following a series' ActivityPub actor.
450type Follower struct {
451 ActorURI string
452 InboxURI string
453}
454
455// SaveFollower records (or updates the inbox URL of) a follower.
456func (s *Store) SaveFollower(series, actorURI, inboxURI string) error {
457 _, err := s.db.Exec(`
458INSERT INTO ap_follower (series, actor_uri, inbox_uri) VALUES (?, ?, ?)
459ON CONFLICT (series, actor_uri) DO UPDATE SET inbox_uri = excluded.inbox_uri
460`, series, actorURI, inboxURI)
461 return err
462}
463
464// RemoveFollower drops a follower, e.g. on Undo{Follow}.
465func (s *Store) RemoveFollower(series, actorURI string) error {
466 _, err := s.db.Exec(`DELETE FROM ap_follower WHERE series = ? AND actor_uri = ?`, series, actorURI)
467 return err
468}
469
470// Followers lists everyone following a series.
471func (s *Store) Followers(series string) ([]Follower, error) {
472 rows, err := s.db.Query(`SELECT actor_uri, inbox_uri FROM ap_follower WHERE series = ?`, series)
473 if err != nil {
474 return nil, err
475 }
476 defer rows.Close()
477 var out []Follower
478 for rows.Next() {
479 var f Follower
480 if err := rows.Scan(&f.ActorURI, &f.InboxURI); err != nil {
481 return nil, err
482 }
483 out = append(out, f)
484 }
485 return out, rows.Err()
486}
487
488// FollowerCount is len(Followers(series)) without building the slice.
489func (s *Store) FollowerCount(series string) (int, error) {
490 row := s.db.QueryRow(`SELECT COUNT(*) FROM ap_follower WHERE series = ?`, series)
491 var n int
492 err := row.Scan(&n)
493 return n, err
494}
495
496// SeriesWithFollowers lists every series that has at least one follower —
497// the set worth checking for new activity to deliver.
498func (s *Store) SeriesWithFollowers() ([]string, error) {
499 rows, err := s.db.Query(`SELECT DISTINCT series FROM ap_follower`)
500 if err != nil {
501 return nil, err
502 }
503 defer rows.Close()
504 var out []string
505 for rows.Next() {
506 var series string
507 if err := rows.Scan(&series); err != nil {
508 return nil, err
509 }
510 out = append(out, series)
511 }
512 return out, rows.Err()
513}
514
515// DeliveryCursor returns how far ActivityPub delivery has gotten through
516// activity_log for a series (0 if nothing has been delivered yet).
517func (s *Store) DeliveryCursor(series string) (int64, error) {
518 row := s.db.QueryRow(`SELECT last_delivered_id FROM ap_delivery_cursor WHERE series = ?`, series)
519 var id int64
520 err := row.Scan(&id)
521 if err == sql.ErrNoRows {
522 return 0, nil
523 }
524 return id, err
525}
526
527// SetDeliveryCursor records how far delivery has gotten for a series.
528func (s *Store) SetDeliveryCursor(series string, id int64) error {
529 _, err := s.db.Exec(`
530INSERT INTO ap_delivery_cursor (series, last_delivered_id) VALUES (?, ?)
531ON CONFLICT (series) DO UPDATE SET last_delivered_id = excluded.last_delivered_id
532`, series, id)
533 return err
534}
535
536// NewActivityForSeries returns activity_log entries for a series with
537// id > afterID, oldest first — the delivery order for ActivityPub.
538func (s *Store) NewActivityForSeries(series string, afterID int64) ([]ActivityEntry, error) {
539 rows, err := s.db.Query(`
540SELECT id, repo_pair, kind, direction, summary, url, series, series_url, forgejo_id, radicle_id, origin, occurred_at FROM activity_log
541WHERE series = ? AND id > ?
542ORDER BY id ASC
543`, series, afterID)
544 if err != nil {
545 return nil, err
546 }
547 return scanActivityRows(rows)
548}
549
550// TokenShare is one encrypted one-time secret. Nonce and Ciphertext are
551// base64-encoded AES-256-GCM output; the decryption key is derived from
552// the passcode the claimant supplies plus the share's own id (which is
553// never stored — only IDHash is), so a database dump alone never yields
554// enough to decrypt.
555type TokenShare struct {
556 IDHash string
557 Nonce string
558 Ciphertext string
559 Attempts int
560 ExpiresAt time.Time
561}
562
563// CreateTokenShare stores a new encrypted share, indexed by the hash of
564// its id (the plaintext id is only ever known to whoever holds the URL).
565func (s *Store) CreateTokenShare(idHash, nonce, ciphertext string, expiresAt time.Time) error {
566 _, err := s.db.Exec(`
567INSERT INTO token_share (id_hash, nonce, ciphertext, attempts, created_at, expires_at)
568VALUES (?, ?, ?, 0, ?, ?)
569`, idHash, nonce, ciphertext, time.Now().UTC().Format(time.RFC3339), expiresAt.UTC().Format(time.RFC3339))
570 return err
571}
572
573// GetTokenShare looks up a share by id hash. Returns nil, nil if it
574// doesn't exist or has already expired (expiry is checked here rather
575// than left to a background sweep, so a stale row can never be claimed
576// even if cleanup hasn't run yet).
577func (s *Store) GetTokenShare(idHash string) (*TokenShare, error) {
578 row := s.db.QueryRow(`SELECT id_hash, nonce, ciphertext, attempts, expires_at FROM token_share WHERE id_hash = ?`, idHash)
579 var t TokenShare
580 var expiresAt string
581 if err := row.Scan(&t.IDHash, &t.Nonce, &t.Ciphertext, &t.Attempts, &expiresAt); err != nil {
582 if err == sql.ErrNoRows {
583 return nil, nil
584 }
585 return nil, err
586 }
587 exp, err := time.Parse(time.RFC3339, expiresAt)
588 if err != nil {
589 return nil, err
590 }
591 t.ExpiresAt = exp
592 if time.Now().UTC().After(exp) {
593 _ = s.DeleteTokenShare(idHash)
594 return nil, nil
595 }
596 return &t, nil
597}
598
599// IncrementTokenShareAttempts records one failed passcode attempt and
600// returns the new total, so the caller can destroy the share once too
601// many wrong guesses have been made — the actual brute-force defense,
602// since a 6-digit passcode alone is far too small a space to rely on
603// key-derivation cost against a determined online attacker.
604func (s *Store) IncrementTokenShareAttempts(idHash string) (int, error) {
605 _, err := s.db.Exec(`UPDATE token_share SET attempts = attempts + 1 WHERE id_hash = ?`, idHash)
606 if err != nil {
607 return 0, err
608 }
609 row := s.db.QueryRow(`SELECT attempts FROM token_share WHERE id_hash = ?`, idHash)
610 var n int
611 if err := row.Scan(&n); err != nil {
612 return 0, err
613 }
614 return n, nil
615}
616
617// DeleteTokenShare destroys a share — called on a successful claim (it's
618// one-time by design), on exceeding the attempt limit (fail closed), or
619// on expiry.
620func (s *Store) DeleteTokenShare(idHash string) error {
621 _, err := s.db.Exec(`DELETE FROM token_share WHERE id_hash = ?`, idHash)
622 return err
623}
624
625// DynamicRepo is one repo pair added through the dashboard's self-service
626// onboarding forms rather than a config.yaml edit.
627type DynamicRepo struct {
628 ID int64
629 Name string
630 Series string
631 ForgejoBaseURL string
632 ForgejoOwner string
633 ForgejoRepo string
634 ForgejoTokenFile string
635 RadicleRID string
636 RadicleHTTPBaseURL string
637 RadicleExplorerURL string
638 RadicleRadHome string
639 SyncGit bool
640 SyncIssues bool
641 SyncPatches bool
642 Materialized bool
643 Approved bool
644}
645
646// CreateDynamicRepo inserts a newly-submitted repo pair as an unapproved
647// request — it never joins the sync loop until ApproveDynamicRepo is
648// called. Fails if the name is already taken — the same uniqueness
649// graft would need across config.yaml pairs anyway.
650func (s *Store) CreateDynamicRepo(r DynamicRepo) (int64, error) {
651 res, err := s.db.Exec(`
652INSERT INTO dynamic_repo (
653 name, series, forgejo_base_url, forgejo_owner, forgejo_repo, forgejo_token_file,
654 radicle_rid, radicle_http_base_url, radicle_explorer_url, radicle_rad_home,
655 sync_git, sync_issues, sync_patches, created_at, materialized, approved
656) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0, 0)
657`, r.Name, r.Series, r.ForgejoBaseURL, r.ForgejoOwner, r.ForgejoRepo, r.ForgejoTokenFile,
658 r.RadicleRID, r.RadicleHTTPBaseURL, r.RadicleExplorerURL, r.RadicleRadHome,
659 boolToInt(r.SyncGit), boolToInt(r.SyncIssues), boolToInt(r.SyncPatches), time.Now().UTC().Format(time.RFC3339))
660 if err != nil {
661 return 0, err
662 }
663 return res.LastInsertId()
664}
665
666const dynamicRepoColumns = `id, name, series, forgejo_base_url, forgejo_owner, forgejo_repo, forgejo_token_file,
667 radicle_rid, radicle_http_base_url, radicle_explorer_url, radicle_rad_home,
668 sync_git, sync_issues, sync_patches, materialized, approved`
669
670func scanDynamicRepoRows(rows *sql.Rows) ([]DynamicRepo, error) {
671 defer rows.Close()
672 var out []DynamicRepo
673 for rows.Next() {
674 var r DynamicRepo
675 var g, i, p, m, a int
676 if err := rows.Scan(&r.ID, &r.Name, &r.Series, &r.ForgejoBaseURL, &r.ForgejoOwner, &r.ForgejoRepo, &r.ForgejoTokenFile,
677 &r.RadicleRID, &r.RadicleHTTPBaseURL, &r.RadicleExplorerURL, &r.RadicleRadHome, &g, &i, &p, &m, &a); err != nil {
678 return nil, err
679 }
680 r.SyncGit, r.SyncIssues, r.SyncPatches, r.Materialized, r.Approved = g != 0, i != 0, p != 0, m != 0, a != 0
681 out = append(out, r)
682 }
683 return out, rows.Err()
684}
685
686// PendingDynamicRepos returns every submitted peer/repo request still
687// awaiting admin approval.
688func (s *Store) PendingDynamicRepos() ([]DynamicRepo, error) {
689 rows, err := s.db.Query(`SELECT ` + dynamicRepoColumns + ` FROM dynamic_repo WHERE approved = 0`)
690 if err != nil {
691 return nil, err
692 }
693 return scanDynamicRepoRows(rows)
694}
695
696// GetDynamicRepo looks up one row by id, e.g. to recover its token file
697// path before deleting a rejected request.
698func (s *Store) GetDynamicRepo(id int64) (*DynamicRepo, error) {
699 rows, err := s.db.Query(`SELECT `+dynamicRepoColumns+` FROM dynamic_repo WHERE id = ?`, id)
700 if err != nil {
701 return nil, err
702 }
703 out, err := scanDynamicRepoRows(rows)
704 if err != nil || len(out) == 0 {
705 return nil, err
706 }
707 return &out[0], nil
708}
709
710// ApproveDynamicRepo marks a pending request approved — the next sync
711// pass turns it into a live RepoSyncer (see UnmaterializedDynamicRepos).
712func (s *Store) ApproveDynamicRepo(id int64) error {
713 _, err := s.db.Exec(`UPDATE dynamic_repo SET approved = 1 WHERE id = ?`, id)
714 return err
715}
716
717// RejectDynamicRepo permanently removes a pending request.
718func (s *Store) RejectDynamicRepo(id int64) error {
719 _, err := s.db.Exec(`DELETE FROM dynamic_repo WHERE id = ?`, id)
720 return err
721}
722
723// UnmaterializedDynamicRepos returns every approved dynamic_repo row the
724// running daemon hasn't yet turned into a live RepoSyncer.
725func (s *Store) UnmaterializedDynamicRepos() ([]DynamicRepo, error) {
726 rows, err := s.db.Query(`SELECT ` + dynamicRepoColumns + ` FROM dynamic_repo WHERE approved = 1 AND materialized = 0`)
727 if err != nil {
728 return nil, err
729 }
730 return scanDynamicRepoRows(rows)
731}
732
733// MarkDynamicRepoMaterialized flips a row so it's never re-registered on
734// a later pass.
735func (s *Store) MarkDynamicRepoMaterialized(id int64) error {
736 _, err := s.db.Exec(`UPDATE dynamic_repo SET materialized = 1 WHERE id = ?`, id)
737 return err
738}
739
740// AllDynamicRepos returns every approved dynamic_repo row regardless of
741// materialization state — used at startup to rebuild the full set of
742// live pairs after a restart, since these rows persist across it but
743// config.yaml doesn't know about them. Never includes a row still
744// awaiting approval.
745func (s *Store) AllDynamicRepos() ([]DynamicRepo, error) {
746 rows, err := s.db.Query(`SELECT ` + dynamicRepoColumns + ` FROM dynamic_repo WHERE approved = 1`)
747 if err != nil {
748 return nil, err
749 }
750 return scanDynamicRepoRows(rows)
751}
752
753func boolToInt(b bool) int {
754 if b {
755 return 1
756 }
757 return 0
758}
759
760// CommentSeen reports whether a comment body's hash has already been
761// dealt with (mirrored, or recognized as already present on both sides)
762// for one item, so a comment sync pass never re-posts the same content
763// back and forth between Forgejo and Radicle.
764func (s *Store) CommentSeen(repoPair, kind, itemID, hash string) (bool, error) {
765 row := s.db.QueryRow(`SELECT 1 FROM comment_seen WHERE repo_pair = ? AND kind = ? AND item_id = ? AND hash = ?`, repoPair, kind, itemID, hash)
766 var one int
767 err := row.Scan(&one)
768 if err == sql.ErrNoRows {
769 return false, nil
770 }
771 return err == nil, err
772}
773
774// MarkCommentSeen records a comment body's hash as dealt with.
775func (s *Store) MarkCommentSeen(repoPair, kind, itemID, hash string) error {
776 _, err := s.db.Exec(`INSERT OR IGNORE INTO comment_seen (repo_pair, kind, item_id, hash) VALUES (?, ?, ?, ?)`, repoPair, kind, itemID, hash)
777 return err
778}
779
780// SaveATProtoPost records a Bluesky post graft made, so a later reply's
781// parent URI can be traced back to the activity_log entry it was about.
782func (s *Store) SaveATProtoPost(uri string, activityID int64, series string) error {
783 _, err := s.db.Exec(`INSERT OR IGNORE INTO atproto_post (uri, activity_id, series, created_at) VALUES (?, ?, ?, ?)`,
784 uri, activityID, series, time.Now().UTC().Format(time.RFC3339))
785 return err
786}
787
788// ATProtoPostActivity looks up which activity_log entry a Bluesky post
789// URI was about, for resolving an inbound reply back to the underlying
790// Forgejo/Radicle item.
791func (s *Store) ATProtoPostActivity(uri string) (int64, bool, error) {
792 row := s.db.QueryRow(`SELECT activity_id FROM atproto_post WHERE uri = ?`, uri)
793 var id int64
794 err := row.Scan(&id)
795 if err == sql.ErrNoRows {
796 return 0, false, nil
797 }
798 return id, err == nil, err
799}
800
801// ATProtoCursor returns how far notification polling has gotten for a
802// series' Bluesky account ("" if never polled).
803func (s *Store) ATProtoCursor(series string) (string, error) {
804 row := s.db.QueryRow(`SELECT cursor FROM atproto_cursor WHERE series = ?`, series)
805 var c string
806 err := row.Scan(&c)
807 if err == sql.ErrNoRows {
808 return "", nil
809 }
810 return c, err
811}
812
813// SetATProtoCursor records how far notification polling has gotten.
814func (s *Store) SetATProtoCursor(series, cursor string) error {
815 _, err := s.db.Exec(`
816INSERT INTO atproto_cursor (series, cursor) VALUES (?, ?)
817ON CONFLICT (series) DO UPDATE SET cursor = excluded.cursor
818`, series, cursor)
819 return err
820}
821
822// PendingRadiclePeer is one submitted "add a Radicle peer" request
823// awaiting admin approval.
824type PendingRadiclePeer struct {
825 ID int64
826 Series string
827 NodeID string
828 Address string
829}
830
831// CreatePendingRadiclePeer records a submitted request.
832func (s *Store) CreatePendingRadiclePeer(series, nodeID, address string) (int64, error) {
833 res, err := s.db.Exec(`INSERT INTO pending_radicle_peer (series, node_id, address, created_at) VALUES (?, ?, ?, ?)`,
834 series, nodeID, address, time.Now().UTC().Format(time.RFC3339))
835 if err != nil {
836 return 0, err
837 }
838 return res.LastInsertId()
839}
840
841// ListPendingRadiclePeers returns every request awaiting approval.
842func (s *Store) ListPendingRadiclePeers() ([]PendingRadiclePeer, error) {
843 rows, err := s.db.Query(`SELECT id, series, node_id, address FROM pending_radicle_peer`)
844 if err != nil {
845 return nil, err
846 }
847 defer rows.Close()
848 var out []PendingRadiclePeer
849 for rows.Next() {
850 var p PendingRadiclePeer
851 if err := rows.Scan(&p.ID, &p.Series, &p.NodeID, &p.Address); err != nil {
852 return nil, err
853 }
854 out = append(out, p)
855 }
856 return out, rows.Err()
857}
858
859// GetPendingRadiclePeer looks up one request by id.
860func (s *Store) GetPendingRadiclePeer(id int64) (*PendingRadiclePeer, error) {
861 row := s.db.QueryRow(`SELECT id, series, node_id, address FROM pending_radicle_peer WHERE id = ?`, id)
862 var p PendingRadiclePeer
863 err := row.Scan(&p.ID, &p.Series, &p.NodeID, &p.Address)
864 if err == sql.ErrNoRows {
865 return nil, nil
866 }
867 if err != nil {
868 return nil, err
869 }
870 return &p, nil
871}
872
873// DeletePendingRadiclePeer removes a request — called after it's been
874// approved and acted on, or rejected.
875func (s *Store) DeletePendingRadiclePeer(id int64) error {
876 _, err := s.db.Exec(`DELETE FROM pending_radicle_peer WHERE id = ?`, id)
877 return err
878}