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