Federated via graft
6.5 kB
199 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 "time"
10
11 _ "modernc.org/sqlite"
12)
13
14// Store wraps the local SQLite database used to track sync state.
15type Store struct {
16 db *sql.DB
17}
18
19// Open creates (if needed) and opens the state database at path.
20func Open(path string) (*Store, error) {
21 db, err := sql.Open("sqlite", path)
22 if err != nil {
23 return nil, fmt.Errorf("open state db: %w", err)
24 }
25 if err := migrate(db); err != nil {
26 db.Close()
27 return nil, fmt.Errorf("migrate state db: %w", err)
28 }
29 return &Store{db: db}, nil
30}
31
32func (s *Store) Close() error {
33 return s.db.Close()
34}
35
36func migrate(db *sql.DB) error {
37 _, err := db.Exec(`
38CREATE TABLE IF NOT EXISTS git_cursor (
39 repo_pair TEXT PRIMARY KEY,
40 forgejo_sha TEXT NOT NULL DEFAULT '',
41 radicle_sha TEXT NOT NULL DEFAULT ''
42);
43
44CREATE TABLE IF NOT EXISTS item_mapping (
45 repo_pair TEXT NOT NULL,
46 kind TEXT NOT NULL, -- 'issue' or 'patch'
47 forgejo_id INTEGER NOT NULL DEFAULT 0,
48 radicle_id TEXT NOT NULL DEFAULT '',
49 content_hash TEXT NOT NULL DEFAULT '',
50 synced_at TEXT NOT NULL,
51 PRIMARY KEY (repo_pair, kind, forgejo_id, radicle_id)
52);
53
54CREATE INDEX IF NOT EXISTS idx_item_mapping_forgejo
55 ON item_mapping (repo_pair, kind, forgejo_id);
56CREATE INDEX IF NOT EXISTS idx_item_mapping_radicle
57 ON item_mapping (repo_pair, kind, radicle_id);
58
59-- One row per thing actually mirrored: a commit, an issue, a patch. This is
60-- what the status page's activity calendar and commit list read from — it's
61-- an append-only log, never updated, so it can't itself be a source of the
62-- kind of state-loss bug that hit item_mapping.
63CREATE TABLE IF NOT EXISTS activity_log (
64 id INTEGER PRIMARY KEY AUTOINCREMENT,
65 repo_pair TEXT NOT NULL,
66 kind TEXT NOT NULL, -- 'git', 'issue', 'patch'
67 direction TEXT NOT NULL, -- 'forgejo_to_radicle' or 'radicle_to_forgejo'
68 summary TEXT NOT NULL,
69 occurred_at TEXT NOT NULL
70);
71
72CREATE INDEX IF NOT EXISTS idx_activity_log_occurred_at
73 ON activity_log (occurred_at);
74`)
75 return err
76}
77
78// GitCursor returns the last-synced commit SHAs on each side for a repo pair.
79func (s *Store) GitCursor(repoPair string) (forgejoSHA, radicleSHA string, err error) {
80 row := s.db.QueryRow(`SELECT forgejo_sha, radicle_sha FROM git_cursor WHERE repo_pair = ?`, repoPair)
81 err = row.Scan(&forgejoSHA, &radicleSHA)
82 if err == sql.ErrNoRows {
83 return "", "", nil
84 }
85 return forgejoSHA, radicleSHA, err
86}
87
88// SetGitCursor records the commit SHAs last observed on each side.
89func (s *Store) SetGitCursor(repoPair, forgejoSHA, radicleSHA string) error {
90 _, err := s.db.Exec(`
91INSERT INTO git_cursor (repo_pair, forgejo_sha, radicle_sha) VALUES (?, ?, ?)
92ON CONFLICT (repo_pair) DO UPDATE SET forgejo_sha = excluded.forgejo_sha, radicle_sha = excluded.radicle_sha
93`, repoPair, forgejoSHA, radicleSHA)
94 return err
95}
96
97// ItemMapping records the correspondence between a Forgejo issue/PR and its
98// mirrored Radicle issue/patch, plus a content hash used to detect edits
99// that need re-syncing.
100type ItemMapping struct {
101 Kind string // "issue" or "patch"
102 ForgejoID int64
103 RadicleID string
104 ContentHash string
105}
106
107// FindByForgejoID looks up an existing mapping created for a Forgejo item.
108func (s *Store) FindByForgejoID(repoPair, kind string, forgejoID int64) (*ItemMapping, error) {
109 row := s.db.QueryRow(`
110SELECT radicle_id, content_hash FROM item_mapping
111WHERE repo_pair = ? AND kind = ? AND forgejo_id = ?
112`, repoPair, kind, forgejoID)
113 m := &ItemMapping{Kind: kind, ForgejoID: forgejoID}
114 err := row.Scan(&m.RadicleID, &m.ContentHash)
115 if err == sql.ErrNoRows {
116 return nil, nil
117 }
118 return m, err
119}
120
121// FindByRadicleID looks up an existing mapping created for a Radicle item.
122func (s *Store) FindByRadicleID(repoPair, kind, radicleID string) (*ItemMapping, error) {
123 row := s.db.QueryRow(`
124SELECT forgejo_id, content_hash FROM item_mapping
125WHERE repo_pair = ? AND kind = ? AND radicle_id = ?
126`, repoPair, kind, radicleID)
127 m := &ItemMapping{Kind: kind, RadicleID: radicleID}
128 err := row.Scan(&m.ForgejoID, &m.ContentHash)
129 if err == sql.ErrNoRows {
130 return nil, nil
131 }
132 return m, err
133}
134
135// Upsert records or updates a mapping between a Forgejo item and its
136// Radicle counterpart.
137func (s *Store) Upsert(repoPair string, m ItemMapping) error {
138 _, err := s.db.Exec(`
139INSERT INTO item_mapping (repo_pair, kind, forgejo_id, radicle_id, content_hash, synced_at)
140VALUES (?, ?, ?, ?, ?, datetime('now'))
141ON CONFLICT (repo_pair, kind, forgejo_id, radicle_id)
142DO UPDATE SET content_hash = excluded.content_hash, synced_at = excluded.synced_at
143`, repoPair, m.Kind, m.ForgejoID, m.RadicleID, m.ContentHash)
144 return err
145}
146
147// Direction names for LogActivity, kept as constants so callers can't typo
148// a value the status page's rendering silently fails to recognize.
149const (
150 ForgejoToRadicle = "forgejo_to_radicle"
151 RadicleToForgejo = "radicle_to_forgejo"
152)
153
154// LogActivity records one thing that was actually mirrored, for the status
155// page. Call it once per commit/issue/patch, not once per sync pass.
156func (s *Store) LogActivity(repoPair, kind, direction, summary string) error {
157 _, err := s.db.Exec(`
158INSERT INTO activity_log (repo_pair, kind, direction, summary, occurred_at)
159VALUES (?, ?, ?, ?, ?)
160`, repoPair, kind, direction, summary, time.Now().UTC().Format(time.RFC3339))
161 return err
162}
163
164// ActivityEntry is one row of activity_log.
165type ActivityEntry struct {
166 RepoPair string
167 Kind string
168 Direction string
169 Summary string
170 OccurredAt time.Time
171}
172
173// ActivitySince returns every activity entry at or after since, newest first.
174func (s *Store) ActivitySince(since time.Time) ([]ActivityEntry, error) {
175 rows, err := s.db.Query(`
176SELECT repo_pair, kind, direction, summary, occurred_at FROM activity_log
177WHERE occurred_at >= ?
178ORDER BY occurred_at DESC
179`, since.UTC().Format(time.RFC3339))
180 if err != nil {
181 return nil, err
182 }
183 defer rows.Close()
184
185 var out []ActivityEntry
186 for rows.Next() {
187 var e ActivityEntry
188 var occurredAt string
189 if err := rows.Scan(&e.RepoPair, &e.Kind, &e.Direction, &e.Summary, &occurredAt); err != nil {
190 return nil, err
191 }
192 e.OccurredAt, err = time.Parse(time.RFC3339, occurredAt)
193 if err != nil {
194 return nil, fmt.Errorf("parse occurred_at %q: %w", occurredAt, err)
195 }
196 out = append(out, e)
197 }
198 return out, rows.Err()
199}