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