Federated via graft
8.0 kB
232 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 if err != nil {
80 return err
81 }
82 // These columns were added after the table already shipped once; each
83 // ALTER is a no-op (ignored) on a fresh db where the CREATE above
84 // already included it.
85 for _, stmt := range []string{
86 `ALTER TABLE activity_log ADD COLUMN url TEXT NOT NULL DEFAULT ''`,
87 `ALTER TABLE activity_log ADD COLUMN series TEXT NOT NULL DEFAULT ''`,
88 `ALTER TABLE activity_log ADD COLUMN series_url TEXT NOT NULL DEFAULT ''`,
89 } {
90 if _, err := db.Exec(stmt); err != nil && !strings.Contains(err.Error(), "duplicate column") {
91 return err
92 }
93 }
94 return nil
95}
96
97// GitCursor returns the last-synced commit SHAs on each side for a repo pair.
98func (s *Store) GitCursor(repoPair string) (forgejoSHA, radicleSHA string, err error) {
99 row := s.db.QueryRow(`SELECT forgejo_sha, radicle_sha FROM git_cursor WHERE repo_pair = ?`, repoPair)
100 err = row.Scan(&forgejoSHA, &radicleSHA)
101 if err == sql.ErrNoRows {
102 return "", "", nil
103 }
104 return forgejoSHA, radicleSHA, err
105}
106
107// SetGitCursor records the commit SHAs last observed on each side.
108func (s *Store) SetGitCursor(repoPair, forgejoSHA, radicleSHA string) error {
109 _, err := s.db.Exec(`
110INSERT INTO git_cursor (repo_pair, forgejo_sha, radicle_sha) VALUES (?, ?, ?)
111ON CONFLICT (repo_pair) DO UPDATE SET forgejo_sha = excluded.forgejo_sha, radicle_sha = excluded.radicle_sha
112`, repoPair, forgejoSHA, radicleSHA)
113 return err
114}
115
116// ItemMapping records the correspondence between a Forgejo issue/PR and its
117// mirrored Radicle issue/patch, plus a content hash used to detect edits
118// that need re-syncing.
119type ItemMapping struct {
120 Kind string // "issue" or "patch"
121 ForgejoID int64
122 RadicleID string
123 ContentHash string
124}
125
126// FindByForgejoID looks up an existing mapping created for a Forgejo item.
127func (s *Store) FindByForgejoID(repoPair, kind string, forgejoID int64) (*ItemMapping, error) {
128 row := s.db.QueryRow(`
129SELECT radicle_id, content_hash FROM item_mapping
130WHERE repo_pair = ? AND kind = ? AND forgejo_id = ?
131`, repoPair, kind, forgejoID)
132 m := &ItemMapping{Kind: kind, ForgejoID: forgejoID}
133 err := row.Scan(&m.RadicleID, &m.ContentHash)
134 if err == sql.ErrNoRows {
135 return nil, nil
136 }
137 return m, err
138}
139
140// FindByRadicleID looks up an existing mapping created for a Radicle item.
141func (s *Store) FindByRadicleID(repoPair, kind, radicleID string) (*ItemMapping, error) {
142 row := s.db.QueryRow(`
143SELECT forgejo_id, content_hash FROM item_mapping
144WHERE repo_pair = ? AND kind = ? AND radicle_id = ?
145`, repoPair, kind, radicleID)
146 m := &ItemMapping{Kind: kind, RadicleID: radicleID}
147 err := row.Scan(&m.ForgejoID, &m.ContentHash)
148 if err == sql.ErrNoRows {
149 return nil, nil
150 }
151 return m, err
152}
153
154// Upsert records or updates a mapping between a Forgejo item and its
155// Radicle counterpart.
156func (s *Store) Upsert(repoPair string, m ItemMapping) error {
157 _, err := s.db.Exec(`
158INSERT INTO item_mapping (repo_pair, kind, forgejo_id, radicle_id, content_hash, synced_at)
159VALUES (?, ?, ?, ?, ?, datetime('now'))
160ON CONFLICT (repo_pair, kind, forgejo_id, radicle_id)
161DO UPDATE SET content_hash = excluded.content_hash, synced_at = excluded.synced_at
162`, repoPair, m.Kind, m.ForgejoID, m.RadicleID, m.ContentHash)
163 return err
164}
165
166// Activity is one thing that was actually mirrored, for the status page.
167type Activity struct {
168 RepoPair string // the sync pair that did it, e.g. "graft-test-v2-f1"
169 Series string // the human-facing repo group this pair belongs to,
170 // e.g. "graft-test-v2" — several pairs (one per forge/Radicle side)
171 // can share a series so the dashboard shows one row per repo, not
172 // one per pair.
173 SeriesURL string // where clicking the series' own name should go
174 Kind string // "git", "issue", "patch"
175 Direction string // ForgejoToRadicle or RadicleToForgejo
176 Summary string
177 URL string // where to send someone who clicks this entry — the
178 // forge or Radicle page for the thing that now exists because of
179 // this sync; "" if none is known.
180}
181
182// Direction values for Activity.Direction, kept as constants so callers
183// can't typo a value the status page's rendering silently fails to
184// recognize.
185const (
186 ForgejoToRadicle = "forgejo_to_radicle"
187 RadicleToForgejo = "radicle_to_forgejo"
188)
189
190// LogActivity records one Activity. Call it once per commit/issue/patch,
191// not once per sync pass.
192func (s *Store) LogActivity(a Activity) error {
193 _, err := s.db.Exec(`
194INSERT INTO activity_log (repo_pair, kind, direction, summary, url, series, series_url, occurred_at)
195VALUES (?, ?, ?, ?, ?, ?, ?, ?)
196`, a.RepoPair, a.Kind, a.Direction, a.Summary, a.URL, a.Series, a.SeriesURL, time.Now().UTC().Format(time.RFC3339))
197 return err
198}
199
200// ActivityEntry is one row of activity_log.
201type ActivityEntry struct {
202 Activity
203 OccurredAt time.Time
204}
205
206// ActivitySince returns every activity entry at or after since, newest first.
207func (s *Store) ActivitySince(since time.Time) ([]ActivityEntry, error) {
208 rows, err := s.db.Query(`
209SELECT repo_pair, kind, direction, summary, url, series, series_url, occurred_at FROM activity_log
210WHERE occurred_at >= ?
211ORDER BY occurred_at DESC
212`, since.UTC().Format(time.RFC3339))
213 if err != nil {
214 return nil, err
215 }
216 defer rows.Close()
217
218 var out []ActivityEntry
219 for rows.Next() {
220 var e ActivityEntry
221 var occurredAt string
222 if err := rows.Scan(&e.RepoPair, &e.Kind, &e.Direction, &e.Summary, &e.URL, &e.Series, &e.SeriesURL, &occurredAt); err != nil {
223 return nil, err
224 }
225 e.OccurredAt, err = time.Parse(time.RFC3339, occurredAt)
226 if err != nil {
227 return nil, fmt.Errorf("parse occurred_at %q: %w", occurredAt, err)
228 }
229 out = append(out, e)
230 }
231 return out, rows.Err()
232}