Skip to main content

max / makenotwork

13.7 KB · 384 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
250 #[instrument(skip_all)]
251 pub async fn connect(path: &Path) -> Result<SqlitePool> {
252 let opts = SqliteConnectOptions::from_str(&format!("sqlite:{}", path.display()))?
253 .create_if_missing(true)
254 .foreign_keys(true)
255 .journal_mode(sqlx::sqlite::SqliteJournalMode::Wal);
256
257 let pool = SqlitePoolOptions::new()
258 .max_connections(5)
259 .connect_with(opts)
260 .await?;
261
262 run_migrations(&pool).await?;
263 Ok(pool)
264 }
265
266 #[instrument(skip_all)]
267 pub async fn connect_in_memory() -> Result<SqlitePool> {
268 let opts = SqliteConnectOptions::from_str("sqlite::memory:")?.foreign_keys(true);
269 let pool = SqlitePoolOptions::new()
270 .max_connections(1)
271 .connect_with(opts)
272 .await?;
273
274 run_migrations(&pool).await?;
275 Ok(pool)
276 }
277
278 /// Run pending schema migrations. Detects pre-migration databases by checking
279 /// for existing tables and stamps them as version 1 without re-running.
280 #[instrument(skip_all)]
281 pub async fn run_migrations(pool: &SqlitePool) -> Result<()> {
282 // Ensure the schema_version table exists
283 sqlx::query(
284 "CREATE TABLE IF NOT EXISTS schema_version (
285 version INTEGER NOT NULL,
286 description TEXT NOT NULL,
287 applied_at TEXT NOT NULL
288 )",
289 )
290 .execute(pool)
291 .await?;
292
293 let current_version = get_schema_version(pool).await?;
294
295 // Detect pre-migration databases: if schema_version is empty but tables exist,
296 // this is an existing database that predates the migration system.
297 if current_version == 0 && has_existing_tables(pool).await? {
298 info!("detected pre-migration database, stamping as version 1");
299 stamp_version(pool, 1, "initial schema (pre-existing)").await?;
300 // Run remaining migrations (2+) if any
301 for &(version, description, sql) in MIGRATIONS {
302 if version > 1 {
303 run_one_migration(pool, version, description, sql).await?;
304 }
305 }
306 return Ok(());
307 }
308
309 // Run all migrations newer than current version
310 for &(version, description, sql) in MIGRATIONS {
311 if version > current_version {
312 run_one_migration(pool, version, description, sql).await?;
313 }
314 }
315
316 Ok(())
317 }
318
319 /// Get the current schema version (0 if no migrations have been applied).
320 #[instrument(skip_all)]
321 pub async fn get_schema_version(pool: &SqlitePool) -> Result<i64> {
322 let row = sqlx::query_as::<_, (i64,)>("SELECT COALESCE(MAX(version), 0) FROM schema_version")
323 .fetch_one(pool)
324 .await?;
325 Ok(row.0)
326 }
327
328 /// Check whether the database has existing tables from before the migration system.
329 async fn has_existing_tables(pool: &SqlitePool) -> Result<bool> {
330 let row = sqlx::query_as::<_, (i64,)>(
331 "SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = 'health_checks'",
332 )
333 .fetch_one(pool)
334 .await?;
335 Ok(row.0 > 0)
336 }
337
338 /// Execute a single migration's SQL and record it in schema_version.
339 /// Wrapped in an explicit transaction so partial failures roll back cleanly.
340 async fn run_one_migration(
341 pool: &SqlitePool,
342 version: i64,
343 description: &str,
344 sql: &str,
345 ) -> Result<()> {
346 info!(version, description, "running migration");
347
348 let mut tx = pool.begin().await?;
349
350 // Execute the whole migration as a batch via `raw_sql`, which hands the
351 // string to SQLite's own parser to split on statement boundaries. A naive
352 // `split(';')` breaks on the first `;` inside a trigger body or string
353 // literal (fuzz-2026-07-06 migration split trap); today's migrations have
354 // none, but this makes the first such migration safe rather than a startup
355 // failure.
356 sqlx::raw_sql(sqlx::AssertSqlSafe(sql))
357 .execute(&mut *tx)
358 .await?;
359
360 // Record the version inside the same transaction
361 let now = chrono::Utc::now().to_rfc3339();
362 sqlx::query("INSERT INTO schema_version (version, description, applied_at) VALUES (?, ?, ?)")
363 .bind(version)
364 .bind(description)
365 .bind(&now)
366 .execute(&mut *tx)
367 .await?;
368
369 tx.commit().await?;
370 Ok(())
371 }
372
373 /// Record a version in the schema_version table.
374 async fn stamp_version(pool: &SqlitePool, version: i64, description: &str) -> Result<()> {
375 let now = chrono::Utc::now().to_rfc3339();
376 sqlx::query("INSERT INTO schema_version (version, description, applied_at) VALUES (?, ?, ?)")
377 .bind(version)
378 .bind(description)
379 .bind(&now)
380 .execute(pool)
381 .await?;
382 Ok(())
383 }
384