Federated via graft
11 kB
319 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 if err != nil {
89 return err
90 }
91 // These columns were added after the table already shipped once; each
92 // ALTER is a no-op (ignored) on a fresh db where the CREATE above
93 // already included it.
94 for _, stmt := range []string{
95 `ALTER TABLE activity_log ADD COLUMN url TEXT NOT NULL DEFAULT ''`,
96 `ALTER TABLE activity_log ADD COLUMN series TEXT NOT NULL DEFAULT ''`,
97 `ALTER TABLE activity_log ADD COLUMN series_url TEXT NOT NULL DEFAULT ''`,
98 } {
99 if _, err := db.Exec(stmt); err != nil && !strings.Contains(err.Error(), "duplicate column") {
100 return err
101 }
102 }
103 return nil
104}
105
106// GitCursor returns the last-synced commit SHAs on each side for a repo pair.
107func (s *Store) GitCursor(repoPair string) (forgejoSHA, radicleSHA string, err error) {
108 row := s.db.QueryRow(`SELECT forgejo_sha, radicle_sha FROM git_cursor WHERE repo_pair = ?`, repoPair)
109 err = row.Scan(&forgejoSHA, &radicleSHA)
110 if err == sql.ErrNoRows {
111 return "", "", nil
112 }
113 return forgejoSHA, radicleSHA, err
114}
115
116// SetGitCursor records the commit SHAs last observed on each side.
117func (s *Store) SetGitCursor(repoPair, forgejoSHA, radicleSHA string) error {
118 _, err := s.db.Exec(`
119INSERT INTO git_cursor (repo_pair, forgejo_sha, radicle_sha) VALUES (?, ?, ?)
120ON CONFLICT (repo_pair) DO UPDATE SET forgejo_sha = excluded.forgejo_sha, radicle_sha = excluded.radicle_sha
121`, repoPair, forgejoSHA, radicleSHA)
122 return err
123}
124
125// ItemMapping records the correspondence between a Forgejo issue/PR and its
126// mirrored Radicle issue/patch, plus a content hash used to detect edits
127// that need re-syncing.
128type ItemMapping struct {
129 Kind string // "issue" or "patch"
130 ForgejoID int64
131 RadicleID string
132 ContentHash string
133}
134
135// FindByForgejoID looks up an existing mapping created for a Forgejo item.
136func (s *Store) FindByForgejoID(repoPair, kind string, forgejoID int64) (*ItemMapping, error) {
137 row := s.db.QueryRow(`
138SELECT radicle_id, content_hash FROM item_mapping
139WHERE repo_pair = ? AND kind = ? AND forgejo_id = ?
140`, repoPair, kind, forgejoID)
141 m := &ItemMapping{Kind: kind, ForgejoID: forgejoID}
142 err := row.Scan(&m.RadicleID, &m.ContentHash)
143 if err == sql.ErrNoRows {
144 return nil, nil
145 }
146 return m, err
147}
148
149// FindByRadicleID looks up an existing mapping created for a Radicle item.
150func (s *Store) FindByRadicleID(repoPair, kind, radicleID string) (*ItemMapping, error) {
151 row := s.db.QueryRow(`
152SELECT forgejo_id, content_hash FROM item_mapping
153WHERE repo_pair = ? AND kind = ? AND radicle_id = ?
154`, repoPair, kind, radicleID)
155 m := &ItemMapping{Kind: kind, RadicleID: radicleID}
156 err := row.Scan(&m.ForgejoID, &m.ContentHash)
157 if err == sql.ErrNoRows {
158 return nil, nil
159 }
160 return m, err
161}
162
163// Upsert records or updates a mapping between a Forgejo item and its
164// Radicle counterpart.
165func (s *Store) Upsert(repoPair string, m ItemMapping) error {
166 _, err := s.db.Exec(`
167INSERT INTO item_mapping (repo_pair, kind, forgejo_id, radicle_id, content_hash, synced_at)
168VALUES (?, ?, ?, ?, ?, datetime('now'))
169ON CONFLICT (repo_pair, kind, forgejo_id, radicle_id)
170DO UPDATE SET content_hash = excluded.content_hash, synced_at = excluded.synced_at
171`, repoPair, m.Kind, m.ForgejoID, m.RadicleID, m.ContentHash)
172 return err
173}
174
175// Activity is one thing that was actually mirrored, for the status page.
176type Activity struct {
177 RepoPair string // the sync pair that did it, e.g. "graft-test-v2-f1"
178 Series string // the human-facing repo group this pair belongs to,
179 // e.g. "graft-test-v2" — several pairs (one per forge/Radicle side)
180 // can share a series so the dashboard shows one row per repo, not
181 // one per pair.
182 SeriesURL string // where clicking the series' own name should go
183 Kind string // "git", "issue", "patch"
184 Direction string // ForgejoToRadicle or RadicleToForgejo
185 Summary string
186 URL string // where to send someone who clicks this entry — the
187 // forge or Radicle page for the thing that now exists because of
188 // this sync; "" if none is known.
189}
190
191// Direction values for Activity.Direction, kept as constants so callers
192// can't typo a value the status page's rendering silently fails to
193// recognize.
194const (
195 ForgejoToRadicle = "forgejo_to_radicle"
196 RadicleToForgejo = "radicle_to_forgejo"
197)
198
199// LogActivity records one Activity. Call it once per commit/issue/patch,
200// not once per sync pass.
201func (s *Store) LogActivity(a Activity) error {
202 _, err := s.db.Exec(`
203INSERT INTO activity_log (repo_pair, kind, direction, summary, url, series, series_url, occurred_at)
204VALUES (?, ?, ?, ?, ?, ?, ?, ?)
205`, a.RepoPair, a.Kind, a.Direction, a.Summary, a.URL, a.Series, a.SeriesURL, time.Now().UTC().Format(time.RFC3339))
206 return err
207}
208
209// ActivityEntry is one row of activity_log.
210type ActivityEntry struct {
211 ID int64 // activity_log.id, used as the stable suffix of ActivityPub Note URIs
212 Activity
213 OccurredAt time.Time
214}
215
216// ActivitySince returns every activity entry at or after since, newest first.
217func (s *Store) ActivitySince(since time.Time) ([]ActivityEntry, error) {
218 rows, err := s.db.Query(`
219SELECT id, repo_pair, kind, direction, summary, url, series, series_url, occurred_at FROM activity_log
220WHERE occurred_at >= ?
221ORDER BY occurred_at DESC
222`, since.UTC().Format(time.RFC3339))
223 if err != nil {
224 return nil, err
225 }
226 return scanActivityRows(rows)
227}
228
229// ActivityForSeries returns the most recent activity entries for one
230// series, newest first, capped at limit — used to build that series'
231// ActivityPub outbox.
232func (s *Store) ActivityForSeries(series string, limit int) ([]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 series = ?
236ORDER BY occurred_at DESC
237LIMIT ?
238`, series, limit)
239 if err != nil {
240 return nil, err
241 }
242 return scanActivityRows(rows)
243}
244
245// ActivityByID looks up one activity_log row by id, e.g. to resolve an
246// ActivityPub Note URI (/actors/{series}/notes/{id}) back to the mirrored
247// item it describes.
248func (s *Store) ActivityByID(id int64) (*ActivityEntry, error) {
249 row := s.db.QueryRow(`
250SELECT id, repo_pair, kind, direction, summary, url, series, series_url, occurred_at FROM activity_log
251WHERE id = ?
252`, id)
253 var e ActivityEntry
254 var occurredAt string
255 err := row.Scan(&e.ID, &e.RepoPair, &e.Kind, &e.Direction, &e.Summary, &e.URL, &e.Series, &e.SeriesURL, &occurredAt)
256 if err == sql.ErrNoRows {
257 return nil, nil
258 }
259 if err != nil {
260 return nil, err
261 }
262 e.OccurredAt, err = time.Parse(time.RFC3339, occurredAt)
263 if err != nil {
264 return nil, fmt.Errorf("parse occurred_at %q: %w", occurredAt, err)
265 }
266 return &e, nil
267}
268
269func scanActivityRows(rows *sql.Rows) ([]ActivityEntry, error) {
270 defer rows.Close()
271 var out []ActivityEntry
272 for rows.Next() {
273 var e ActivityEntry
274 var occurredAt string
275 if err := rows.Scan(&e.ID, &e.RepoPair, &e.Kind, &e.Direction, &e.Summary, &e.URL, &e.Series, &e.SeriesURL, &occurredAt); err != nil {
276 return nil, err
277 }
278 var err error
279 e.OccurredAt, err = time.Parse(time.RFC3339, occurredAt)
280 if err != nil {
281 return nil, fmt.Errorf("parse occurred_at %q: %w", occurredAt, err)
282 }
283 out = append(out, e)
284 }
285 return out, rows.Err()
286}
287
288// ActorKeys is one series' ActivityPub actor keypair, PEM-encoded.
289type ActorKeys struct {
290 PrivateKeyPEM string
291 PublicKeyPEM string
292}
293
294// ActorKeys returns the stored keypair for a series, or nil if none has
295// been generated yet.
296func (s *Store) ActorKeys(series string) (*ActorKeys, error) {
297 row := s.db.QueryRow(`SELECT private_key, public_key FROM ap_actor WHERE series = ?`, series)
298 var k ActorKeys
299 err := row.Scan(&k.PrivateKeyPEM, &k.PublicKeyPEM)
300 if err == sql.ErrNoRows {
301 return nil, nil
302 }
303 if err != nil {
304 return nil, err
305 }
306 return &k, nil
307}
308
309// SaveActorKeys stores a newly generated keypair for a series. Safe to call
310// concurrently: if another goroutine already inserted one, this becomes a
311// no-op via INSERT OR IGNORE, and the caller should re-read with
312// ActorKeys to get whichever copy won.
313func (s *Store) SaveActorKeys(series string, k ActorKeys) error {
314 _, err := s.db.Exec(`
315INSERT OR IGNORE INTO ap_actor (series, private_key, public_key, created_at)
316VALUES (?, ?, ?, ?)
317`, series, k.PrivateKeyPEM, k.PublicKeyPEM, time.Now().UTC().Format(time.RFC3339))
318 return err
319}