Skip to main content

max / makenotwork

17.7 KB · 485 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 crate::error::PomError;
7 use tracing::{info, instrument};
8
9 /// Each migration is a (version, description, SQL) tuple. Versions start at 1.
10 /// The SQL may contain multiple statements separated by semicolons.
11 const MIGRATIONS: &[(i64, &str, &str)] = &[
12 (
13 1,
14 "initial schema",
15 r"
16 CREATE TABLE IF NOT EXISTS health_checks (
17 id INTEGER PRIMARY KEY AUTOINCREMENT,
18 target TEXT NOT NULL,
19 status TEXT NOT NULL,
20 checked_at TEXT NOT NULL,
21 response_time_ms INTEGER NOT NULL,
22 details_json TEXT,
23 error TEXT
24 );
25 CREATE TABLE IF NOT EXISTS test_runs (
26 id INTEGER PRIMARY KEY AUTOINCREMENT,
27 target TEXT NOT NULL,
28 started_at TEXT NOT NULL,
29 finished_at TEXT,
30 duration_secs INTEGER,
31 exit_code INTEGER,
32 passed INTEGER NOT NULL,
33 summary_json TEXT NOT NULL,
34 raw_output TEXT NOT NULL,
35 filter TEXT
36 );
37 CREATE TABLE IF NOT EXISTS peer_identities (
38 peer_name TEXT PRIMARY KEY,
39 instance_id TEXT NOT NULL,
40 first_seen TEXT NOT NULL
41 );
42 CREATE TABLE IF NOT EXISTS peer_heartbeats (
43 id INTEGER PRIMARY KEY AUTOINCREMENT,
44 peer_name TEXT NOT NULL,
45 status TEXT NOT NULL,
46 latency_ms INTEGER NOT NULL,
47 checked_at TEXT NOT NULL
48 );
49 CREATE INDEX IF NOT EXISTS idx_health_checks_target_id ON health_checks(target, id DESC);
50 CREATE INDEX IF NOT EXISTS idx_health_checks_target_checked ON health_checks(target, checked_at);
51 CREATE INDEX IF NOT EXISTS idx_test_runs_target_id ON test_runs(target, id DESC);
52 CREATE INDEX IF NOT EXISTS idx_peer_heartbeats_peer_id ON peer_heartbeats(peer_name, id DESC);
53 ",
54 ),
55 (
56 2,
57 "add alerts table",
58 r"
59 CREATE TABLE IF NOT EXISTS alerts (
60 id INTEGER PRIMARY KEY AUTOINCREMENT,
61 target TEXT NOT NULL,
62 alert_type TEXT NOT NULL,
63 from_status TEXT,
64 to_status TEXT,
65 sent_at TEXT NOT NULL,
66 error TEXT
67 );
68 CREATE INDEX IF NOT EXISTS idx_alerts_target_sent ON alerts(target, sent_at);
69 ",
70 ),
71 (
72 3,
73 "add tls_checks table",
74 r"
75 CREATE TABLE IF NOT EXISTS tls_checks (
76 id INTEGER PRIMARY KEY AUTOINCREMENT,
77 target TEXT NOT NULL,
78 host TEXT NOT NULL,
79 valid INTEGER NOT NULL,
80 days_remaining INTEGER NOT NULL,
81 not_before TEXT NOT NULL,
82 not_after TEXT NOT NULL,
83 subject TEXT NOT NULL,
84 issuer TEXT NOT NULL,
85 checked_at TEXT NOT NULL,
86 error TEXT
87 );
88 CREATE INDEX IF NOT EXISTS idx_tls_checks_target_id ON tls_checks(target, id DESC);
89 ",
90 ),
91 (
92 4,
93 "add incidents table",
94 r"
95 CREATE TABLE IF NOT EXISTS incidents (
96 id INTEGER PRIMARY KEY AUTOINCREMENT,
97 target TEXT NOT NULL,
98 started_at TEXT NOT NULL,
99 ended_at TEXT,
100 duration_secs INTEGER,
101 from_status TEXT NOT NULL,
102 to_status TEXT NOT NULL
103 );
104 CREATE INDEX IF NOT EXISTS idx_incidents_target_id ON incidents(target, id DESC);
105 ",
106 ),
107 (
108 5,
109 "add route_checks table",
110 r"
111 CREATE TABLE IF NOT EXISTS route_checks (
112 id INTEGER PRIMARY KEY AUTOINCREMENT,
113 target TEXT NOT NULL,
114 path TEXT NOT NULL,
115 status_code INTEGER NOT NULL,
116 ok INTEGER NOT NULL,
117 response_time_ms INTEGER NOT NULL,
118 checked_at TEXT NOT NULL,
119 error TEXT
120 );
121 CREATE INDEX IF NOT EXISTS idx_route_checks_target_path ON route_checks(target, path, id DESC);
122 CREATE INDEX IF NOT EXISTS idx_route_checks_target ON route_checks(target, checked_at DESC);
123 ",
124 ),
125 (
126 6,
127 "add dns_checks and whois_checks tables",
128 r"
129 CREATE TABLE IF NOT EXISTS dns_checks (
130 id INTEGER PRIMARY KEY AUTOINCREMENT,
131 target TEXT NOT NULL,
132 name TEXT NOT NULL,
133 record_type TEXT NOT NULL,
134 expected TEXT NOT NULL,
135 actual TEXT NOT NULL,
136 matches INTEGER NOT NULL,
137 checked_at TEXT NOT NULL,
138 error TEXT
139 );
140 CREATE INDEX IF NOT EXISTS idx_dns_checks_target ON dns_checks(target, name, id DESC);
141
142 CREATE TABLE IF NOT EXISTS whois_checks (
143 id INTEGER PRIMARY KEY AUTOINCREMENT,
144 target TEXT NOT NULL,
145 domain TEXT NOT NULL,
146 registrar TEXT,
147 expiry_date TEXT,
148 days_remaining INTEGER,
149 nameservers TEXT,
150 checked_at TEXT NOT NULL,
151 error TEXT
152 );
153 CREATE INDEX IF NOT EXISTS idx_whois_checks_target ON whois_checks(target, id DESC);
154 ",
155 ),
156 (
157 7,
158 "add test_details table",
159 r"
160 CREATE TABLE IF NOT EXISTS test_details (
161 id INTEGER PRIMARY KEY AUTOINCREMENT,
162 run_id INTEGER NOT NULL REFERENCES test_runs(id) ON DELETE CASCADE,
163 test_name TEXT NOT NULL,
164 passed INTEGER NOT NULL,
165 duration_ms INTEGER
166 );
167 CREATE INDEX IF NOT EXISTS idx_test_details_run_id ON test_details(run_id);
168 CREATE INDEX IF NOT EXISTS idx_test_details_name ON test_details(test_name, run_id DESC);
169 ",
170 ),
171 (
172 8,
173 "add cors_checks table",
174 r"
175 CREATE TABLE IF NOT EXISTS cors_checks (
176 id INTEGER PRIMARY KEY AUTOINCREMENT,
177 target TEXT NOT NULL,
178 url TEXT NOT NULL,
179 origin TEXT NOT NULL,
180 method TEXT NOT NULL,
181 passes INTEGER NOT NULL,
182 checked_at TEXT NOT NULL,
183 error TEXT
184 );
185 CREATE INDEX IF NOT EXISTS idx_cors_target_id ON cors_checks(target, id DESC);
186 ",
187 ),
188 (
189 9,
190 "add backup_checks table",
191 r"
192 CREATE TABLE IF NOT EXISTS backup_checks (
193 id INTEGER PRIMARY KEY AUTOINCREMENT,
194 target TEXT NOT NULL,
195 database_name TEXT NOT NULL,
196 status TEXT NOT NULL,
197 last_backup_at TEXT,
198 size_bytes INTEGER,
199 age_hours INTEGER,
200 checked_at TEXT NOT NULL,
201 error TEXT
202 );
203 CREATE INDEX IF NOT EXISTS idx_backup_checks_target ON backup_checks(target, id DESC);
204 ",
205 ),
206 (
207 10,
208 "add pending_alerts retry queue",
209 r"
210 CREATE TABLE IF NOT EXISTS pending_alerts (
211 id INTEGER PRIMARY KEY AUTOINCREMENT,
212 alert_key TEXT NOT NULL,
213 category TEXT NOT NULL,
214 channel TEXT NOT NULL, -- 'wam' or 'email'
215 subject TEXT NOT NULL,
216 body TEXT NOT NULL,
217 priority TEXT, -- WAM only
218 source TEXT, -- WAM only
219 source_ref TEXT, -- WAM only
220 from_status TEXT,
221 to_status TEXT,
222 error TEXT,
223 attempts INTEGER NOT NULL DEFAULT 0,
224 created_at TEXT NOT NULL,
225 next_retry_at TEXT NOT NULL
226 );
227 CREATE INDEX IF NOT EXISTS idx_pending_alerts_due ON pending_alerts(next_retry_at, id);
228 ",
229 ),
230 (
231 11,
232 "add scan_pipeline_checks table",
233 r"
234 CREATE TABLE IF NOT EXISTS scan_pipeline_checks (
235 id INTEGER PRIMARY KEY AUTOINCREMENT,
236 target TEXT NOT NULL,
237 status TEXT NOT NULL,
238 issues TEXT NOT NULL, -- JSON array of fired-threshold lines
239 queue_pending INTEGER NOT NULL,
240 queue_running INTEGER NOT NULL,
241 queue_stuck INTEGER NOT NULL,
242 held_total INTEGER NOT NULL,
243 checked_at TEXT NOT NULL,
244 error TEXT
245 );
246 CREATE INDEX IF NOT EXISTS idx_scan_pipeline_checks_target ON scan_pipeline_checks(target, id DESC);
247 ",
248 ),
249 (
250 12,
251 "add systemd_checks table",
252 r"
253 CREATE TABLE IF NOT EXISTS systemd_checks (
254 id INTEGER PRIMARY KEY AUTOINCREMENT,
255 target TEXT NOT NULL,
256 status TEXT NOT NULL,
257 units TEXT NOT NULL, -- JSON array of unit snapshots
258 failed_units TEXT NOT NULL, -- JSON array of host-wide failed unit names
259 issues TEXT NOT NULL, -- JSON array of fired-threshold lines
260 checked_at TEXT NOT NULL,
261 error TEXT
262 );
263 CREATE INDEX IF NOT EXISTS idx_systemd_checks_target ON systemd_checks(target, id DESC);
264 ",
265 ),
266 (
267 13,
268 "add synckit_fleet_checks table",
269 r"
270 CREATE TABLE IF NOT EXISTS synckit_fleet_checks (
271 id INTEGER PRIMARY KEY AUTOINCREMENT,
272 target TEXT NOT NULL,
273 window_days INTEGER NOT NULL,
274 devices INTEGER NOT NULL,
275 versions TEXT NOT NULL, -- JSON array of version snapshots
276 checked_at TEXT NOT NULL,
277 error TEXT
278 );
279 CREATE INDEX IF NOT EXISTS idx_synckit_fleet_checks_target ON synckit_fleet_checks(target, id DESC);
280 ",
281 ),
282 (
283 14,
284 "record per-trust-store results on tls_checks",
285 // `check_tls` has always probed both the bundled web-PKI roots and the
286 // host's own trust store, then folded only the first into `valid` and
287 // dropped the second on the floor. The platform result is the one that
288 // answers "can a client on this machine still validate a public chain",
289 // which is exactly the host-CA-bundle failure mode nothing was watching.
290 //
291 // Nullable with no default on purpose: rows written before this
292 // migration have no trust readings, and NULL says so. A DEFAULT 0 would
293 // claim every historical check found the chain untrusted.
294 r"
295 ALTER TABLE tls_checks ADD COLUMN webpki_trusted INTEGER;
296 ALTER TABLE tls_checks ADD COLUMN platform_trusted INTEGER;
297 ALTER TABLE tls_checks ADD COLUMN webpki_error TEXT;
298 ALTER TABLE tls_checks ADD COLUMN platform_error TEXT;
299 ",
300 ),
301 (
302 15,
303 "add ca_bundle_checks table",
304 r"
305 CREATE TABLE IF NOT EXISTS ca_bundle_checks (
306 id INTEGER PRIMARY KEY AUTOINCREMENT,
307 target TEXT NOT NULL,
308 status TEXT NOT NULL,
309 package TEXT NOT NULL,
310 installed TEXT,
311 candidate TEXT,
312 cert_count INTEGER,
313 lists_age_hours INTEGER,
314 issues TEXT NOT NULL, -- JSON array of fired issue lines
315 checked_at TEXT NOT NULL,
316 error TEXT
317 );
318 CREATE INDEX IF NOT EXISTS idx_ca_bundle_checks_target ON ca_bundle_checks(target, id DESC);
319 ",
320 ),
321 ];
322
323 /// What to do when the database file is not there yet.
324 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
325 pub enum OnMissingDb {
326 /// Create the file (and its parent directory) — `pom --init`.
327 Create,
328 /// Fail, naming the path. Creating on open is what turned one wrong path
329 /// into a silent second database: five suites ran, passed, and reported
330 /// into a file nothing served.
331 Fail,
332 }
333
334 #[instrument(skip(on_missing))]
335 pub async fn connect(path: &Path, on_missing: OnMissingDb) -> Result<SqlitePool> {
336 if !path.exists() {
337 match on_missing {
338 OnMissingDb::Fail => {
339 return Err(PomError::Config(format!(
340 "no database at {}. Run `pom --init` to create it, or point \
341 storage.db_path at the existing one",
342 path.display()
343 )));
344 }
345 OnMissingDb::Create => {
346 if let Some(dir) = path.parent() {
347 std::fs::create_dir_all(dir)?;
348 }
349 }
350 }
351 }
352
353 let opts = SqliteConnectOptions::from_str(&format!("sqlite:{}", path.display()))?
354 .create_if_missing(on_missing == OnMissingDb::Create)
355 .foreign_keys(true)
356 .journal_mode(sqlx::sqlite::SqliteJournalMode::Wal);
357
358 let pool = SqlitePoolOptions::new()
359 .max_connections(5)
360 .connect_with(opts)
361 .await?;
362
363 run_migrations(&pool).await?;
364 Ok(pool)
365 }
366
367 #[instrument(skip_all)]
368 pub async fn connect_in_memory() -> Result<SqlitePool> {
369 let opts = SqliteConnectOptions::from_str("sqlite::memory:")?.foreign_keys(true);
370 let pool = SqlitePoolOptions::new()
371 .max_connections(1)
372 .connect_with(opts)
373 .await?;
374
375 run_migrations(&pool).await?;
376 Ok(pool)
377 }
378
379 /// Run pending schema migrations. Detects pre-migration databases by checking
380 /// for existing tables and stamps them as version 1 without re-running.
381 #[instrument(skip_all)]
382 pub async fn run_migrations(pool: &SqlitePool) -> Result<()> {
383 // Ensure the schema_version table exists
384 sqlx::query(
385 "CREATE TABLE IF NOT EXISTS schema_version (
386 version INTEGER NOT NULL,
387 description TEXT NOT NULL,
388 applied_at TEXT NOT NULL
389 )",
390 )
391 .execute(pool)
392 .await?;
393
394 let current_version = get_schema_version(pool).await?;
395
396 // Detect pre-migration databases: if schema_version is empty but tables exist,
397 // this is an existing database that predates the migration system.
398 if current_version == 0 && has_existing_tables(pool).await? {
399 info!("detected pre-migration database, stamping as version 1");
400 stamp_version(pool, 1, "initial schema (pre-existing)").await?;
401 // Run remaining migrations (2+) if any
402 for &(version, description, sql) in MIGRATIONS {
403 if version > 1 {
404 run_one_migration(pool, version, description, sql).await?;
405 }
406 }
407 return Ok(());
408 }
409
410 // Run all migrations newer than current version
411 for &(version, description, sql) in MIGRATIONS {
412 if version > current_version {
413 run_one_migration(pool, version, description, sql).await?;
414 }
415 }
416
417 Ok(())
418 }
419
420 /// Get the current schema version (0 if no migrations have been applied).
421 #[instrument(skip_all)]
422 pub async fn get_schema_version(pool: &SqlitePool) -> Result<i64> {
423 let row = sqlx::query_as::<_, (i64,)>("SELECT COALESCE(MAX(version), 0) FROM schema_version")
424 .fetch_one(pool)
425 .await?;
426 Ok(row.0)
427 }
428
429 /// Check whether the database has existing tables from before the migration system.
430 async fn has_existing_tables(pool: &SqlitePool) -> Result<bool> {
431 let row = sqlx::query_as::<_, (i64,)>(
432 "SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = 'health_checks'",
433 )
434 .fetch_one(pool)
435 .await?;
436 Ok(row.0 > 0)
437 }
438
439 /// Execute a single migration's SQL and record it in schema_version.
440 /// Wrapped in an explicit transaction so partial failures roll back cleanly.
441 async fn run_one_migration(
442 pool: &SqlitePool,
443 version: i64,
444 description: &str,
445 sql: &str,
446 ) -> Result<()> {
447 info!(version, description, "running migration");
448
449 let mut tx = pool.begin().await?;
450
451 // Execute the whole migration as a batch via `raw_sql`, which hands the
452 // string to SQLite's own parser to split on statement boundaries. A naive
453 // `split(';')` breaks on the first `;` inside a trigger body or string
454 // literal (fuzz-2026-07-06 migration split trap); today's migrations have
455 // none, but this makes the first such migration safe rather than a startup
456 // failure.
457 sqlx::raw_sql(sqlx::AssertSqlSafe(sql))
458 .execute(&mut *tx)
459 .await?;
460
461 // Record the version inside the same transaction
462 let now = chrono::Utc::now().to_rfc3339();
463 sqlx::query("INSERT INTO schema_version (version, description, applied_at) VALUES (?, ?, ?)")
464 .bind(version)
465 .bind(description)
466 .bind(&now)
467 .execute(&mut *tx)
468 .await?;
469
470 tx.commit().await?;
471 Ok(())
472 }
473
474 /// Record a version in the schema_version table.
475 async fn stamp_version(pool: &SqlitePool, version: i64, description: &str) -> Result<()> {
476 let now = chrono::Utc::now().to_rfc3339();
477 sqlx::query("INSERT INTO schema_version (version, description, applied_at) VALUES (?, ?, ?)")
478 .bind(version)
479 .bind(description)
480 .bind(&now)
481 .execute(pool)
482 .await?;
483 Ok(())
484 }
485