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