Federated via graft
16 kB
450 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 if err != nil {
115 return err
116 }
117 // These columns were added after the table already shipped once; each
118 // ALTER is a no-op (ignored) on a fresh db where the CREATE above
119 // already included it.
120 for _, stmt := range []string{
121 `ALTER TABLE activity_log ADD COLUMN url TEXT NOT NULL DEFAULT ''`,
122 `ALTER TABLE activity_log ADD COLUMN series TEXT NOT NULL DEFAULT ''`,
123 `ALTER TABLE activity_log ADD COLUMN series_url TEXT NOT NULL DEFAULT ''`,
124 `ALTER TABLE activity_log ADD COLUMN forgejo_id INTEGER NOT NULL DEFAULT 0`,
125 `ALTER TABLE activity_log ADD COLUMN radicle_id TEXT NOT NULL DEFAULT ''`,
126 } {
127 if _, err := db.Exec(stmt); err != nil && !strings.Contains(err.Error(), "duplicate column") {
128 return err
129 }
130 }
131 return nil
132}
133
134// GitCursor returns the last-synced commit SHAs on each side for a repo pair.
135func (s *Store) GitCursor(repoPair string) (forgejoSHA, radicleSHA string, err error) {
136 row := s.db.QueryRow(`SELECT forgejo_sha, radicle_sha FROM git_cursor WHERE repo_pair = ?`, repoPair)
137 err = row.Scan(&forgejoSHA, &radicleSHA)
138 if err == sql.ErrNoRows {
139 return "", "", nil
140 }
141 return forgejoSHA, radicleSHA, err
142}
143
144// SetGitCursor records the commit SHAs last observed on each side.
145func (s *Store) SetGitCursor(repoPair, forgejoSHA, radicleSHA string) error {
146 _, err := s.db.Exec(`
147INSERT INTO git_cursor (repo_pair, forgejo_sha, radicle_sha) VALUES (?, ?, ?)
148ON CONFLICT (repo_pair) DO UPDATE SET forgejo_sha = excluded.forgejo_sha, radicle_sha = excluded.radicle_sha
149`, repoPair, forgejoSHA, radicleSHA)
150 return err
151}
152
153// ItemMapping records the correspondence between a Forgejo issue/PR and its
154// mirrored Radicle issue/patch, plus a content hash used to detect edits
155// that need re-syncing.
156type ItemMapping struct {
157 Kind string // "issue" or "patch"
158 ForgejoID int64
159 RadicleID string
160 ContentHash string
161}
162
163// FindByForgejoID looks up an existing mapping created for a Forgejo item.
164func (s *Store) FindByForgejoID(repoPair, kind string, forgejoID int64) (*ItemMapping, error) {
165 row := s.db.QueryRow(`
166SELECT radicle_id, content_hash FROM item_mapping
167WHERE repo_pair = ? AND kind = ? AND forgejo_id = ?
168`, repoPair, kind, forgejoID)
169 m := &ItemMapping{Kind: kind, ForgejoID: forgejoID}
170 err := row.Scan(&m.RadicleID, &m.ContentHash)
171 if err == sql.ErrNoRows {
172 return nil, nil
173 }
174 return m, err
175}
176
177// FindByRadicleID looks up an existing mapping created for a Radicle item.
178func (s *Store) FindByRadicleID(repoPair, kind, radicleID string) (*ItemMapping, error) {
179 row := s.db.QueryRow(`
180SELECT forgejo_id, content_hash FROM item_mapping
181WHERE repo_pair = ? AND kind = ? AND radicle_id = ?
182`, repoPair, kind, radicleID)
183 m := &ItemMapping{Kind: kind, RadicleID: radicleID}
184 err := row.Scan(&m.ForgejoID, &m.ContentHash)
185 if err == sql.ErrNoRows {
186 return nil, nil
187 }
188 return m, err
189}
190
191// Upsert records or updates a mapping between a Forgejo item and its
192// Radicle counterpart.
193func (s *Store) Upsert(repoPair string, m ItemMapping) error {
194 _, err := s.db.Exec(`
195INSERT INTO item_mapping (repo_pair, kind, forgejo_id, radicle_id, content_hash, synced_at)
196VALUES (?, ?, ?, ?, ?, datetime('now'))
197ON CONFLICT (repo_pair, kind, forgejo_id, radicle_id)
198DO UPDATE SET content_hash = excluded.content_hash, synced_at = excluded.synced_at
199`, repoPair, m.Kind, m.ForgejoID, m.RadicleID, m.ContentHash)
200 return err
201}
202
203// Activity is one thing that was actually mirrored, for the status page.
204type Activity struct {
205 RepoPair string // the sync pair that did it, e.g. "graft-test-v2-f1"
206 Series string // the human-facing repo group this pair belongs to,
207 // e.g. "graft-test-v2" — several pairs (one per forge/Radicle side)
208 // can share a series so the dashboard shows one row per repo, not
209 // one per pair.
210 SeriesURL string // where clicking the series' own name should go
211 Kind string // "git", "issue", "patch"
212 Direction string // ForgejoToRadicle or RadicleToForgejo
213 Summary string
214 URL string // where to send someone who clicks this entry — the
215 // forge or Radicle page for the thing that now exists because of
216 // this sync; "" if none is known.
217 ForgejoID int64 // issue/PR number on Forgejo, for kind "issue"/"patch"; 0 for "git"
218 RadicleID string // issue/patch id on Radicle, for kind "issue"/"patch"; "" for "git"
219}
220
221// Direction values for Activity.Direction, kept as constants so callers
222// can't typo a value the status page's rendering silently fails to
223// recognize.
224const (
225 ForgejoToRadicle = "forgejo_to_radicle"
226 RadicleToForgejo = "radicle_to_forgejo"
227)
228
229// LogActivity records one Activity. Call it once per commit/issue/patch,
230// not once per sync pass.
231func (s *Store) LogActivity(a Activity) error {
232 _, err := s.db.Exec(`
233INSERT INTO activity_log (repo_pair, kind, direction, summary, url, series, series_url, forgejo_id, radicle_id, occurred_at)
234VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
235`, a.RepoPair, a.Kind, a.Direction, a.Summary, a.URL, a.Series, a.SeriesURL, a.ForgejoID, a.RadicleID, time.Now().UTC().Format(time.RFC3339))
236 return err
237}
238
239// ActivityEntry is one row of activity_log.
240type ActivityEntry struct {
241 ID int64 // activity_log.id, used as the stable suffix of ActivityPub Note URIs
242 Activity
243 OccurredAt time.Time
244}
245
246// ActivitySince returns every activity entry at or after since, newest first.
247func (s *Store) ActivitySince(since time.Time) ([]ActivityEntry, error) {
248 rows, err := s.db.Query(`
249SELECT id, repo_pair, kind, direction, summary, url, series, series_url, forgejo_id, radicle_id, occurred_at FROM activity_log
250WHERE occurred_at >= ?
251ORDER BY occurred_at DESC
252`, since.UTC().Format(time.RFC3339))
253 if err != nil {
254 return nil, err
255 }
256 return scanActivityRows(rows)
257}
258
259// ActivityForSeries returns the most recent activity entries for one
260// series, newest first, capped at limit — used to build that series'
261// ActivityPub outbox.
262func (s *Store) ActivityForSeries(series string, limit int) ([]ActivityEntry, error) {
263 rows, err := s.db.Query(`
264SELECT id, repo_pair, kind, direction, summary, url, series, series_url, forgejo_id, radicle_id, occurred_at FROM activity_log
265WHERE series = ?
266ORDER BY occurred_at DESC
267LIMIT ?
268`, series, limit)
269 if err != nil {
270 return nil, err
271 }
272 return scanActivityRows(rows)
273}
274
275// ActivityByID looks up one activity_log row by id, e.g. to resolve an
276// ActivityPub Note URI (/actors/{series}/notes/{id}) back to the mirrored
277// item it describes.
278func (s *Store) ActivityByID(id int64) (*ActivityEntry, error) {
279 row := s.db.QueryRow(`
280SELECT id, repo_pair, kind, direction, summary, url, series, series_url, forgejo_id, radicle_id, occurred_at FROM activity_log
281WHERE id = ?
282`, id)
283 var e ActivityEntry
284 var occurredAt string
285 err := row.Scan(&e.ID, &e.RepoPair, &e.Kind, &e.Direction, &e.Summary, &e.URL, &e.Series, &e.SeriesURL, &e.ForgejoID, &e.RadicleID, &occurredAt)
286 if err == sql.ErrNoRows {
287 return nil, nil
288 }
289 if err != nil {
290 return nil, err
291 }
292 e.OccurredAt, err = time.Parse(time.RFC3339, occurredAt)
293 if err != nil {
294 return nil, fmt.Errorf("parse occurred_at %q: %w", occurredAt, err)
295 }
296 return &e, nil
297}
298
299func scanActivityRows(rows *sql.Rows) ([]ActivityEntry, error) {
300 defer rows.Close()
301 var out []ActivityEntry
302 for rows.Next() {
303 var e ActivityEntry
304 var occurredAt string
305 if err := rows.Scan(&e.ID, &e.RepoPair, &e.Kind, &e.Direction, &e.Summary, &e.URL, &e.Series, &e.SeriesURL, &e.ForgejoID, &e.RadicleID, &occurredAt); err != nil {
306 return nil, err
307 }
308 var err error
309 e.OccurredAt, err = time.Parse(time.RFC3339, occurredAt)
310 if err != nil {
311 return nil, fmt.Errorf("parse occurred_at %q: %w", occurredAt, err)
312 }
313 out = append(out, e)
314 }
315 return out, rows.Err()
316}
317
318// ActorKeys is one series' ActivityPub actor keypair, PEM-encoded.
319type ActorKeys struct {
320 PrivateKeyPEM string
321 PublicKeyPEM string
322}
323
324// ActorKeys returns the stored keypair for a series, or nil if none has
325// been generated yet.
326func (s *Store) ActorKeys(series string) (*ActorKeys, error) {
327 row := s.db.QueryRow(`SELECT private_key, public_key FROM ap_actor WHERE series = ?`, series)
328 var k ActorKeys
329 err := row.Scan(&k.PrivateKeyPEM, &k.PublicKeyPEM)
330 if err == sql.ErrNoRows {
331 return nil, nil
332 }
333 if err != nil {
334 return nil, err
335 }
336 return &k, nil
337}
338
339// SaveActorKeys stores a newly generated keypair for a series. Safe to call
340// concurrently: if another goroutine already inserted one, this becomes a
341// no-op via INSERT OR IGNORE, and the caller should re-read with
342// ActorKeys to get whichever copy won.
343func (s *Store) SaveActorKeys(series string, k ActorKeys) error {
344 _, err := s.db.Exec(`
345INSERT OR IGNORE INTO ap_actor (series, private_key, public_key, created_at)
346VALUES (?, ?, ?, ?)
347`, series, k.PrivateKeyPEM, k.PublicKeyPEM, time.Now().UTC().Format(time.RFC3339))
348 return err
349}
350
351// Follower is one remote actor following a series' ActivityPub actor.
352type Follower struct {
353 ActorURI string
354 InboxURI string
355}
356
357// SaveFollower records (or updates the inbox URL of) a follower.
358func (s *Store) SaveFollower(series, actorURI, inboxURI string) error {
359 _, err := s.db.Exec(`
360INSERT INTO ap_follower (series, actor_uri, inbox_uri) VALUES (?, ?, ?)
361ON CONFLICT (series, actor_uri) DO UPDATE SET inbox_uri = excluded.inbox_uri
362`, series, actorURI, inboxURI)
363 return err
364}
365
366// RemoveFollower drops a follower, e.g. on Undo{Follow}.
367func (s *Store) RemoveFollower(series, actorURI string) error {
368 _, err := s.db.Exec(`DELETE FROM ap_follower WHERE series = ? AND actor_uri = ?`, series, actorURI)
369 return err
370}
371
372// Followers lists everyone following a series.
373func (s *Store) Followers(series string) ([]Follower, error) {
374 rows, err := s.db.Query(`SELECT actor_uri, inbox_uri FROM ap_follower WHERE series = ?`, series)
375 if err != nil {
376 return nil, err
377 }
378 defer rows.Close()
379 var out []Follower
380 for rows.Next() {
381 var f Follower
382 if err := rows.Scan(&f.ActorURI, &f.InboxURI); err != nil {
383 return nil, err
384 }
385 out = append(out, f)
386 }
387 return out, rows.Err()
388}
389
390// FollowerCount is len(Followers(series)) without building the slice.
391func (s *Store) FollowerCount(series string) (int, error) {
392 row := s.db.QueryRow(`SELECT COUNT(*) FROM ap_follower WHERE series = ?`, series)
393 var n int
394 err := row.Scan(&n)
395 return n, err
396}
397
398// SeriesWithFollowers lists every series that has at least one follower —
399// the set worth checking for new activity to deliver.
400func (s *Store) SeriesWithFollowers() ([]string, error) {
401 rows, err := s.db.Query(`SELECT DISTINCT series FROM ap_follower`)
402 if err != nil {
403 return nil, err
404 }
405 defer rows.Close()
406 var out []string
407 for rows.Next() {
408 var series string
409 if err := rows.Scan(&series); err != nil {
410 return nil, err
411 }
412 out = append(out, series)
413 }
414 return out, rows.Err()
415}
416
417// DeliveryCursor returns how far ActivityPub delivery has gotten through
418// activity_log for a series (0 if nothing has been delivered yet).
419func (s *Store) DeliveryCursor(series string) (int64, error) {
420 row := s.db.QueryRow(`SELECT last_delivered_id FROM ap_delivery_cursor WHERE series = ?`, series)
421 var id int64
422 err := row.Scan(&id)
423 if err == sql.ErrNoRows {
424 return 0, nil
425 }
426 return id, err
427}
428
429// SetDeliveryCursor records how far delivery has gotten for a series.
430func (s *Store) SetDeliveryCursor(series string, id int64) error {
431 _, err := s.db.Exec(`
432INSERT INTO ap_delivery_cursor (series, last_delivered_id) VALUES (?, ?)
433ON CONFLICT (series) DO UPDATE SET last_delivered_id = excluded.last_delivered_id
434`, series, id)
435 return err
436}
437
438// NewActivityForSeries returns activity_log entries for a series with
439// id > afterID, oldest first — the delivery order for ActivityPub.
440func (s *Store) NewActivityForSeries(series string, afterID int64) ([]ActivityEntry, error) {
441 rows, err := s.db.Query(`
442SELECT id, repo_pair, kind, direction, summary, url, series, series_url, forgejo_id, radicle_id, occurred_at FROM activity_log
443WHERE series = ? AND id > ?
444ORDER BY id ASC
445`, series, afterID)
446 if err != nil {
447 return nil, err
448 }
449 return scanActivityRows(rows)
450}