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