Skip to main content

max / makenotwork

15.0 KB · 417 lines History Blame Raw
1 //! Schema migrations and pool construction. Migrations are an ordered list of
2 //! (version, description, SQL) applied in one transaction each; pre-migration
3 //! databases are detected and stamped as version 1.
4
5 use super::{FromStr, Path, Result, SqliteConnectOptions, SqlitePool, SqlitePoolOptions};
6 use tracing::{info, instrument};
7
8 /// Each migration is a (version, description, SQL) tuple. Versions start at 1.
9 /// The SQL may contain multiple statements separated by semicolons.
10 const MIGRATIONS: &[(i64, &str, &str)] = &[
11 (
12 1,
13 "initial schema",
14 r"
15 CREATE TABLE IF NOT EXISTS health_checks (
16 id INTEGER PRIMARY KEY AUTOINCREMENT,
17 target TEXT NOT NULL,
18 status TEXT NOT NULL,
19 checked_at TEXT NOT NULL,
20 response_time_ms INTEGER NOT NULL,
21 details_json TEXT,
22 error TEXT
23 );
24 CREATE TABLE IF NOT EXISTS test_runs (
25 id INTEGER PRIMARY KEY AUTOINCREMENT,
26 target TEXT NOT NULL,
27 started_at TEXT NOT NULL,
28 finished_at TEXT,
29 duration_secs INTEGER,
30 exit_code INTEGER,
31 passed INTEGER NOT NULL,
32 summary_json TEXT NOT NULL,
33 raw_output TEXT NOT NULL,
34 filter TEXT
35 );
36 CREATE TABLE IF NOT EXISTS peer_identities (
37 peer_name TEXT PRIMARY KEY,
38 instance_id TEXT NOT NULL,
39 first_seen TEXT NOT NULL
40 );
41 CREATE TABLE IF NOT EXISTS peer_heartbeats (
42 id INTEGER PRIMARY KEY AUTOINCREMENT,
43 peer_name TEXT NOT NULL,
44 status TEXT NOT NULL,
45 latency_ms INTEGER NOT NULL,
46 checked_at TEXT NOT NULL
47 );
48 CREATE INDEX IF NOT EXISTS idx_health_checks_target_id ON health_checks(target, id DESC);
49 CREATE INDEX IF NOT EXISTS idx_health_checks_target_checked ON health_checks(target, checked_at);
50 CREATE INDEX IF NOT EXISTS idx_test_runs_target_id ON test_runs(target, id DESC);
51 CREATE INDEX IF NOT EXISTS idx_peer_heartbeats_peer_id ON peer_heartbeats(peer_name, id DESC);
52 ",
53 ),
54 (
55 2,
56 "add alerts table",
57 r"
58 CREATE TABLE IF NOT EXISTS alerts (
59 id INTEGER PRIMARY KEY AUTOINCREMENT,
60 target TEXT NOT NULL,
61 alert_type TEXT NOT NULL,
62 from_status TEXT,
63 to_status TEXT,
64 sent_at TEXT NOT NULL,
65 error TEXT
66 );
67 CREATE INDEX IF NOT EXISTS idx_alerts_target_sent ON alerts(target, sent_at);
68 ",
69 ),
70 (
71 3,
72 "add tls_checks table",
73 r"
74 CREATE TABLE IF NOT EXISTS tls_checks (
75 id INTEGER PRIMARY KEY AUTOINCREMENT,
76 target TEXT NOT NULL,
77 host TEXT NOT NULL,
78 valid INTEGER NOT NULL,
79 days_remaining INTEGER NOT NULL,
80 not_before TEXT NOT NULL,
81 not_after TEXT NOT NULL,
82 subject TEXT NOT NULL,
83 issuer TEXT NOT NULL,
84 checked_at TEXT NOT NULL,
85 error TEXT
86 );
87 CREATE INDEX IF NOT EXISTS idx_tls_checks_target_id ON tls_checks(target, id DESC);
88 ",
89 ),
90 (
91 4,
92 "add incidents table",
93 r"
94 CREATE TABLE IF NOT EXISTS incidents (
95 id INTEGER PRIMARY KEY AUTOINCREMENT,
96 target TEXT NOT NULL,
97 started_at TEXT NOT NULL,
98 ended_at TEXT,
99 duration_secs INTEGER,
100 from_status TEXT NOT NULL,
101 to_status TEXT NOT NULL
102 );
103 CREATE INDEX IF NOT EXISTS idx_incidents_target_id ON incidents(target, id DESC);
104 ",
105 ),
106 (
107 5,
108 "add route_checks table",
109 r"
110 CREATE TABLE IF NOT EXISTS route_checks (
111 id INTEGER PRIMARY KEY AUTOINCREMENT,
112 target TEXT NOT NULL,
113 path TEXT NOT NULL,
114 status_code INTEGER NOT NULL,
115 ok INTEGER NOT NULL,
116 response_time_ms INTEGER NOT NULL,
117 checked_at TEXT NOT NULL,
118 error TEXT
119 );
120 CREATE INDEX IF NOT EXISTS idx_route_checks_target_path ON route_checks(target, path, id DESC);
121 CREATE INDEX IF NOT EXISTS idx_route_checks_target ON route_checks(target, checked_at DESC);
122 ",
123 ),
124 (
125 6,
126 "add dns_checks and whois_checks tables",
127 r"
128 CREATE TABLE IF NOT EXISTS dns_checks (
129 id INTEGER PRIMARY KEY AUTOINCREMENT,
130 target TEXT NOT NULL,
131 name TEXT NOT NULL,
132 record_type TEXT NOT NULL,
133 expected TEXT NOT NULL,
134 actual TEXT NOT NULL,
135 matches INTEGER NOT NULL,
136 checked_at TEXT NOT NULL,
137 error TEXT
138 );
139 CREATE INDEX IF NOT EXISTS idx_dns_checks_target ON dns_checks(target, name, id DESC);
140
141 CREATE TABLE IF NOT EXISTS whois_checks (
142 id INTEGER PRIMARY KEY AUTOINCREMENT,
143 target TEXT NOT NULL,
144 domain TEXT NOT NULL,
145 registrar TEXT,
146 expiry_date TEXT,
147 days_remaining INTEGER,
148 nameservers TEXT,
149 checked_at TEXT NOT NULL,
150 error TEXT
151 );
152 CREATE INDEX IF NOT EXISTS idx_whois_checks_target ON whois_checks(target, id DESC);
153 ",
154 ),
155 (
156 7,
157 "add test_details table",
158 r"
159 CREATE TABLE IF NOT EXISTS test_details (
160 id INTEGER PRIMARY KEY AUTOINCREMENT,
161 run_id INTEGER NOT NULL REFERENCES test_runs(id) ON DELETE CASCADE,
162 test_name TEXT NOT NULL,
163 passed INTEGER NOT NULL,
164 duration_ms INTEGER
165 );
166 CREATE INDEX IF NOT EXISTS idx_test_details_run_id ON test_details(run_id);
167 CREATE INDEX IF NOT EXISTS idx_test_details_name ON test_details(test_name, run_id DESC);
168 ",
169 ),
170 (
171 8,
172 "add cors_checks table",
173 r"
174 CREATE TABLE IF NOT EXISTS cors_checks (
175 id INTEGER PRIMARY KEY AUTOINCREMENT,
176 target TEXT NOT NULL,
177 url TEXT NOT NULL,
178 origin TEXT NOT NULL,
179 method TEXT NOT NULL,
180 passes INTEGER NOT NULL,
181 checked_at TEXT NOT NULL,
182 error TEXT
183 );
184 CREATE INDEX IF NOT EXISTS idx_cors_target_id ON cors_checks(target, id DESC);
185 ",
186 ),
187 (
188 9,
189 "add backup_checks table",
190 r"
191 CREATE TABLE IF NOT EXISTS backup_checks (
192 id INTEGER PRIMARY KEY AUTOINCREMENT,
193 target TEXT NOT NULL,
194 database_name TEXT NOT NULL,
195 status TEXT NOT NULL,
196 last_backup_at TEXT,
197 size_bytes INTEGER,
198 age_hours INTEGER,
199 checked_at TEXT NOT NULL,
200 error TEXT
201 );
202 CREATE INDEX IF NOT EXISTS idx_backup_checks_target ON backup_checks(target, id DESC);
203 ",
204 ),
205 (
206 10,
207 "add pending_alerts retry queue",
208 r"
209 CREATE TABLE IF NOT EXISTS pending_alerts (
210 id INTEGER PRIMARY KEY AUTOINCREMENT,
211 alert_key TEXT NOT NULL,
212 category TEXT NOT NULL,
213 channel TEXT NOT NULL, -- 'wam' or 'email'
214 subject TEXT NOT NULL,
215 body TEXT NOT NULL,
216 priority TEXT, -- WAM only
217 source TEXT, -- WAM only
218 source_ref TEXT, -- WAM only
219 from_status TEXT,
220 to_status TEXT,
221 error TEXT,
222 attempts INTEGER NOT NULL DEFAULT 0,
223 created_at TEXT NOT NULL,
224 next_retry_at TEXT NOT NULL
225 );
226 CREATE INDEX IF NOT EXISTS idx_pending_alerts_due ON pending_alerts(next_retry_at, id);
227 ",
228 ),
229 (
230 11,
231 "add scan_pipeline_checks table",
232 r"
233 CREATE TABLE IF NOT EXISTS scan_pipeline_checks (
234 id INTEGER PRIMARY KEY AUTOINCREMENT,
235 target TEXT NOT NULL,
236 status TEXT NOT NULL,
237 issues TEXT NOT NULL, -- JSON array of fired-threshold lines
238 queue_pending INTEGER NOT NULL,
239 queue_running INTEGER NOT NULL,
240 queue_stuck INTEGER NOT NULL,
241 held_total INTEGER NOT NULL,
242 checked_at TEXT NOT NULL,
243 error TEXT
244 );
245 CREATE INDEX IF NOT EXISTS idx_scan_pipeline_checks_target ON scan_pipeline_checks(target, id DESC);
246 ",
247 ),
248 (
249 12,
250 "add systemd_checks table",
251 r"
252 CREATE TABLE IF NOT EXISTS systemd_checks (
253 id INTEGER PRIMARY KEY AUTOINCREMENT,
254 target TEXT NOT NULL,
255 status TEXT NOT NULL,
256 units TEXT NOT NULL, -- JSON array of unit snapshots
257 failed_units TEXT NOT NULL, -- JSON array of host-wide failed unit names
258 issues TEXT NOT NULL, -- JSON array of fired-threshold lines
259 checked_at TEXT NOT NULL,
260 error TEXT
261 );
262 CREATE INDEX IF NOT EXISTS idx_systemd_checks_target ON systemd_checks(target, id DESC);
263 ",
264 ),
265 (
266 13,
267 "add synckit_fleet_checks table",
268 r"
269 CREATE TABLE IF NOT EXISTS synckit_fleet_checks (
270 id INTEGER PRIMARY KEY AUTOINCREMENT,
271 target TEXT NOT NULL,
272 window_days INTEGER NOT NULL,
273 devices INTEGER NOT NULL,
274 versions TEXT NOT NULL, -- JSON array of version snapshots
275 checked_at TEXT NOT NULL,
276 error TEXT
277 );
278 CREATE INDEX IF NOT EXISTS idx_synckit_fleet_checks_target ON synckit_fleet_checks(target, id DESC);
279 ",
280 ),
281 ];
282
283 #[instrument(skip_all)]
284 pub async fn connect(path: &Path) -> Result<SqlitePool> {
285 let opts = SqliteConnectOptions::from_str(&format!("sqlite:{}", path.display()))?
286 .create_if_missing(true)
287 .foreign_keys(true)
288 .journal_mode(sqlx::sqlite::SqliteJournalMode::Wal);
289
290 let pool = SqlitePoolOptions::new()
291 .max_connections(5)
292 .connect_with(opts)
293 .await?;
294
295 run_migrations(&pool).await?;
296 Ok(pool)
297 }
298
299 #[instrument(skip_all)]
300 pub async fn connect_in_memory() -> Result<SqlitePool> {
301 let opts = SqliteConnectOptions::from_str("sqlite::memory:")?.foreign_keys(true);
302 let pool = SqlitePoolOptions::new()
303 .max_connections(1)
304 .connect_with(opts)
305 .await?;
306
307 run_migrations(&pool).await?;
308 Ok(pool)
309 }
310
311 /// Run pending schema migrations. Detects pre-migration databases by checking
312 /// for existing tables and stamps them as version 1 without re-running.
313 #[instrument(skip_all)]
314 pub async fn run_migrations(pool: &SqlitePool) -> Result<()> {
315 // Ensure the schema_version table exists
316 sqlx::query(
317 "CREATE TABLE IF NOT EXISTS schema_version (
318 version INTEGER NOT NULL,
319 description TEXT NOT NULL,
320 applied_at TEXT NOT NULL
321 )",
322 )
323 .execute(pool)
324 .await?;
325
326 let current_version = get_schema_version(pool).await?;
327
328 // Detect pre-migration databases: if schema_version is empty but tables exist,
329 // this is an existing database that predates the migration system.
330 if current_version == 0 && has_existing_tables(pool).await? {
331 info!("detected pre-migration database, stamping as version 1");
332 stamp_version(pool, 1, "initial schema (pre-existing)").await?;
333 // Run remaining migrations (2+) if any
334 for &(version, description, sql) in MIGRATIONS {
335 if version > 1 {
336 run_one_migration(pool, version, description, sql).await?;
337 }
338 }
339 return Ok(());
340 }
341
342 // Run all migrations newer than current version
343 for &(version, description, sql) in MIGRATIONS {
344 if version > current_version {
345 run_one_migration(pool, version, description, sql).await?;
346 }
347 }
348
349 Ok(())
350 }
351
352 /// Get the current schema version (0 if no migrations have been applied).
353 #[instrument(skip_all)]
354 pub async fn get_schema_version(pool: &SqlitePool) -> Result<i64> {
355 let row = sqlx::query_as::<_, (i64,)>("SELECT COALESCE(MAX(version), 0) FROM schema_version")
356 .fetch_one(pool)
357 .await?;
358 Ok(row.0)
359 }
360
361 /// Check whether the database has existing tables from before the migration system.
362 async fn has_existing_tables(pool: &SqlitePool) -> Result<bool> {
363 let row = sqlx::query_as::<_, (i64,)>(
364 "SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = 'health_checks'",
365 )
366 .fetch_one(pool)
367 .await?;
368 Ok(row.0 > 0)
369 }
370
371 /// Execute a single migration's SQL and record it in schema_version.
372 /// Wrapped in an explicit transaction so partial failures roll back cleanly.
373 async fn run_one_migration(
374 pool: &SqlitePool,
375 version: i64,
376 description: &str,
377 sql: &str,
378 ) -> Result<()> {
379 info!(version, description, "running migration");
380
381 let mut tx = pool.begin().await?;
382
383 // Execute the whole migration as a batch via `raw_sql`, which hands the
384 // string to SQLite's own parser to split on statement boundaries. A naive
385 // `split(';')` breaks on the first `;` inside a trigger body or string
386 // literal (fuzz-2026-07-06 migration split trap); today's migrations have
387 // none, but this makes the first such migration safe rather than a startup
388 // failure.
389 sqlx::raw_sql(sqlx::AssertSqlSafe(sql))
390 .execute(&mut *tx)
391 .await?;
392
393 // Record the version inside the same transaction
394 let now = chrono::Utc::now().to_rfc3339();
395 sqlx::query("INSERT INTO schema_version (version, description, applied_at) VALUES (?, ?, ?)")
396 .bind(version)
397 .bind(description)
398 .bind(&now)
399 .execute(&mut *tx)
400 .await?;
401
402 tx.commit().await?;
403 Ok(())
404 }
405
406 /// Record a version in the schema_version table.
407 async fn stamp_version(pool: &SqlitePool, version: i64, description: &str) -> Result<()> {
408 let now = chrono::Utc::now().to_rfc3339();
409 sqlx::query("INSERT INTO schema_version (version, description, applied_at) VALUES (?, ?, ?)")
410 .bind(version)
411 .bind(description)
412 .bind(&now)
413 .execute(pool)
414 .await?;
415 Ok(())
416 }
417