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