Skip to main content

max / makenotwork

Split pom db.rs into per-domain modules Break the 1634-line db.rs into a db/ directory of 13 per-domain modules (migrations, health, test_runs, alerts, tls, incidents, routes, dns, whois, cors, backup, maintenance, peers) behind a db/mod.rs facade. mod.rs keeps the module doc, carries the shared named imports, and re-exports every item via pub use <mod>::*, so all pom::db::* paths are unchanged. Migration helpers stay private; the private HealthCheckRow/TestRunRow row structs and their into_* impls travel with their callers. Each module re-imports the tracing macros it uses (use super::* does not carry macros). No behavior change; fn multiset identical, build and clippy clean, all 129 db tests pass. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-07-13 16:14 UTC
Commit: af35b4c0dfdf4203e245195fabf10a9200159ec3
Parent: 400bb5a
15 files changed, +1651 insertions, -500 deletions
D pom/src/db.rs -500
@@ -1,1634 +0,0 @@
1 - //! SQLite persistence — schema, health checks, test runs, and peer data.
2 - //!
3 - //! Uses a migration versioning system: each schema change is a numbered migration
4 - //! stored in [`MIGRATIONS`]. On startup, [`run_migrations`] checks the current
5 - //! version and runs any pending migrations. Existing databases (pre-migration)
6 - //! are detected by the presence of the `health_checks` table and marked as v1.
7 -
8 - use sqlx::sqlite::{SqliteConnectOptions, SqlitePool, SqlitePoolOptions};
9 - use std::path::Path;
10 - use std::str::FromStr;
11 - use tracing::{info, instrument};
12 -
13 - use crate::error::Result;
14 - use crate::types::{BackupCheckResult, CorsCheckResult, DnsCheckResult, HealthDetails, HealthSnapshot, HealthStatus, TestDetail, TestRun, TestRunId, TestSummary, TlsStatus, WhoisResult};
15 -
16 - /// Each migration is a (version, description, SQL) tuple. Versions start at 1.
17 - /// The SQL may contain multiple statements separated by semicolons.
18 - const MIGRATIONS: &[(i64, &str, &str)] = &[
19 - (1, "initial schema", r#"
20 - CREATE TABLE IF NOT EXISTS health_checks (
21 - id INTEGER PRIMARY KEY AUTOINCREMENT,
22 - target TEXT NOT NULL,
23 - status TEXT NOT NULL,
24 - checked_at TEXT NOT NULL,
25 - response_time_ms INTEGER NOT NULL,
26 - details_json TEXT,
27 - error TEXT
28 - );
29 - CREATE TABLE IF NOT EXISTS test_runs (
30 - id INTEGER PRIMARY KEY AUTOINCREMENT,
31 - target TEXT NOT NULL,
32 - started_at TEXT NOT NULL,
33 - finished_at TEXT,
34 - duration_secs INTEGER,
35 - exit_code INTEGER,
36 - passed INTEGER NOT NULL,
37 - summary_json TEXT NOT NULL,
38 - raw_output TEXT NOT NULL,
39 - filter TEXT
40 - );
41 - CREATE TABLE IF NOT EXISTS peer_identities (
42 - peer_name TEXT PRIMARY KEY,
43 - instance_id TEXT NOT NULL,
44 - first_seen TEXT NOT NULL
45 - );
46 - CREATE TABLE IF NOT EXISTS peer_heartbeats (
47 - id INTEGER PRIMARY KEY AUTOINCREMENT,
48 - peer_name TEXT NOT NULL,
49 - status TEXT NOT NULL,
50 - latency_ms INTEGER NOT NULL,
51 - checked_at TEXT NOT NULL
52 - );
53 - CREATE INDEX IF NOT EXISTS idx_health_checks_target_id ON health_checks(target, id DESC);
54 - CREATE INDEX IF NOT EXISTS idx_health_checks_target_checked ON health_checks(target, checked_at);
55 - CREATE INDEX IF NOT EXISTS idx_test_runs_target_id ON test_runs(target, id DESC);
56 - CREATE INDEX IF NOT EXISTS idx_peer_heartbeats_peer_id ON peer_heartbeats(peer_name, id DESC);
57 - "#),
58 - (2, "add alerts table", 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 - (3, "add tls_checks table", r#"
71 - CREATE TABLE IF NOT EXISTS tls_checks (
72 - id INTEGER PRIMARY KEY AUTOINCREMENT,
73 - target TEXT NOT NULL,
74 - host TEXT NOT NULL,
75 - valid INTEGER NOT NULL,
76 - days_remaining INTEGER NOT NULL,
77 - not_before TEXT NOT NULL,
78 - not_after TEXT NOT NULL,
79 - subject TEXT NOT NULL,
80 - issuer TEXT NOT NULL,
81 - checked_at TEXT NOT NULL,
82 - error TEXT
83 - );
84 - CREATE INDEX IF NOT EXISTS idx_tls_checks_target_id ON tls_checks(target, id DESC);
85 - "#),
86 - (4, "add incidents table", r#"
87 - CREATE TABLE IF NOT EXISTS incidents (
88 - id INTEGER PRIMARY KEY AUTOINCREMENT,
89 - target TEXT NOT NULL,
90 - started_at TEXT NOT NULL,
91 - ended_at TEXT,
92 - duration_secs INTEGER,
93 - from_status TEXT NOT NULL,
94 - to_status TEXT NOT NULL
95 - );
96 - CREATE INDEX IF NOT EXISTS idx_incidents_target_id ON incidents(target, id DESC);
97 - "#),
98 - (5, "add route_checks table", r#"
99 - CREATE TABLE IF NOT EXISTS route_checks (
100 - id INTEGER PRIMARY KEY AUTOINCREMENT,
101 - target TEXT NOT NULL,
102 - path TEXT NOT NULL,
103 - status_code INTEGER NOT NULL,
104 - ok INTEGER NOT NULL,
105 - response_time_ms INTEGER NOT NULL,
106 - checked_at TEXT NOT NULL,
107 - error TEXT
108 - );
109 - CREATE INDEX IF NOT EXISTS idx_route_checks_target_path ON route_checks(target, path, id DESC);
110 - CREATE INDEX IF NOT EXISTS idx_route_checks_target ON route_checks(target, checked_at DESC);
111 - "#),
112 - (6, "add dns_checks and whois_checks tables", r#"
113 - CREATE TABLE IF NOT EXISTS dns_checks (
114 - id INTEGER PRIMARY KEY AUTOINCREMENT,
115 - target TEXT NOT NULL,
116 - name TEXT NOT NULL,
117 - record_type TEXT NOT NULL,
118 - expected TEXT NOT NULL,
119 - actual TEXT NOT NULL,
120 - matches INTEGER NOT NULL,
121 - checked_at TEXT NOT NULL,
122 - error TEXT
123 - );
124 - CREATE INDEX IF NOT EXISTS idx_dns_checks_target ON dns_checks(target, name, id DESC);
125 -
126 - CREATE TABLE IF NOT EXISTS whois_checks (
127 - id INTEGER PRIMARY KEY AUTOINCREMENT,
128 - target TEXT NOT NULL,
129 - domain TEXT NOT NULL,
130 - registrar TEXT,
131 - expiry_date TEXT,
132 - days_remaining INTEGER,
133 - nameservers TEXT,
134 - checked_at TEXT NOT NULL,
135 - error TEXT
136 - );
137 - CREATE INDEX IF NOT EXISTS idx_whois_checks_target ON whois_checks(target, id DESC);
138 - "#),
139 - (7, "add test_details table", r#"
140 - CREATE TABLE IF NOT EXISTS test_details (
141 - id INTEGER PRIMARY KEY AUTOINCREMENT,
142 - run_id INTEGER NOT NULL REFERENCES test_runs(id) ON DELETE CASCADE,
143 - test_name TEXT NOT NULL,
144 - passed INTEGER NOT NULL,
145 - duration_ms INTEGER
146 - );
147 - CREATE INDEX IF NOT EXISTS idx_test_details_run_id ON test_details(run_id);
148 - CREATE INDEX IF NOT EXISTS idx_test_details_name ON test_details(test_name, run_id DESC);
149 - "#),
150 - (8, "add cors_checks table", r#"
151 - CREATE TABLE IF NOT EXISTS cors_checks (
152 - id INTEGER PRIMARY KEY AUTOINCREMENT,
153 - target TEXT NOT NULL,
154 - url TEXT NOT NULL,
155 - origin TEXT NOT NULL,
156 - method TEXT NOT NULL,
157 - passes INTEGER NOT NULL,
158 - checked_at TEXT NOT NULL,
159 - error TEXT
160 - );
161 - CREATE INDEX IF NOT EXISTS idx_cors_target_id ON cors_checks(target, id DESC);
162 - "#),
163 - (9, "add backup_checks table", r#"
164 - CREATE TABLE IF NOT EXISTS backup_checks (
165 - id INTEGER PRIMARY KEY AUTOINCREMENT,
166 - target TEXT NOT NULL,
167 - database_name TEXT NOT NULL,
168 - status TEXT NOT NULL,
169 - last_backup_at TEXT,
170 - size_bytes INTEGER,
171 - age_hours INTEGER,
172 - checked_at TEXT NOT NULL,
173 - error TEXT
174 - );
175 - CREATE INDEX IF NOT EXISTS idx_backup_checks_target ON backup_checks(target, id DESC);
176 - "#),
177 - (10, "add pending_alerts retry queue", r#"
178 - CREATE TABLE IF NOT EXISTS pending_alerts (
179 - id INTEGER PRIMARY KEY AUTOINCREMENT,
180 - alert_key TEXT NOT NULL,
181 - category TEXT NOT NULL,
182 - channel TEXT NOT NULL, -- 'wam' or 'email'
183 - subject TEXT NOT NULL,
184 - body TEXT NOT NULL,
185 - priority TEXT, -- WAM only
186 - source TEXT, -- WAM only
187 - source_ref TEXT, -- WAM only
188 - from_status TEXT,
189 - to_status TEXT,
190 - error TEXT,
191 - attempts INTEGER NOT NULL DEFAULT 0,
192 - created_at TEXT NOT NULL,
193 - next_retry_at TEXT NOT NULL
194 - );
195 - CREATE INDEX IF NOT EXISTS idx_pending_alerts_due ON pending_alerts(next_retry_at, id);
196 - "#),
197 - ];
198 -
199 - #[instrument(skip_all)]
200 - pub async fn connect(path: &Path) -> Result<SqlitePool> {
201 - let opts = SqliteConnectOptions::from_str(&format!("sqlite:{}", path.display()))?
202 - .create_if_missing(true)
203 - .foreign_keys(true)
204 - .journal_mode(sqlx::sqlite::SqliteJournalMode::Wal);
205 -
206 - let pool = SqlitePoolOptions::new()
207 - .max_connections(5)
208 - .connect_with(opts)
209 - .await?;
210 -
211 - run_migrations(&pool).await?;
212 - Ok(pool)
213 - }
214 -
215 - #[instrument(skip_all)]
216 - pub async fn connect_in_memory() -> Result<SqlitePool> {
217 - let opts = SqliteConnectOptions::from_str("sqlite::memory:")?.foreign_keys(true);
218 - let pool = SqlitePoolOptions::new()
219 - .max_connections(1)
220 - .connect_with(opts)
221 - .await?;
222 -
223 - run_migrations(&pool).await?;
224 - Ok(pool)
225 - }
226 -
227 - /// Run pending schema migrations. Detects pre-migration databases by checking
228 - /// for existing tables and stamps them as version 1 without re-running.
229 - #[instrument(skip_all)]
230 - pub async fn run_migrations(pool: &SqlitePool) -> Result<()> {
231 - // Ensure the schema_version table exists
232 - sqlx::query(
233 - "CREATE TABLE IF NOT EXISTS schema_version (
234 - version INTEGER NOT NULL,
235 - description TEXT NOT NULL,
236 - applied_at TEXT NOT NULL
237 - )",
238 - )
239 - .execute(pool)
240 - .await?;
241 -
242 - let current_version = get_schema_version(pool).await?;
243 -
244 - // Detect pre-migration databases: if schema_version is empty but tables exist,
245 - // this is an existing database that predates the migration system.
246 - if current_version == 0 && has_existing_tables(pool).await? {
247 - info!("detected pre-migration database, stamping as version 1");
248 - stamp_version(pool, 1, "initial schema (pre-existing)").await?;
249 - // Run remaining migrations (2+) if any
250 - for &(version, description, sql) in MIGRATIONS {
251 - if version > 1 {
252 - run_one_migration(pool, version, description, sql).await?;
253 - }
254 - }
255 - return Ok(());
256 - }
257 -
258 - // Run all migrations newer than current version
259 - for &(version, description, sql) in MIGRATIONS {
260 - if version > current_version {
261 - run_one_migration(pool, version, description, sql).await?;
262 - }
263 - }
264 -
265 - Ok(())
266 - }
267 -
268 - /// Get the current schema version (0 if no migrations have been applied).
269 - #[instrument(skip_all)]
270 - pub async fn get_schema_version(pool: &SqlitePool) -> Result<i64> {
271 - let row = sqlx::query_as::<_, (i64,)>(
272 - "SELECT COALESCE(MAX(version), 0) FROM schema_version",
273 - )
274 - .fetch_one(pool)
275 - .await?;
276 - Ok(row.0)
277 - }
278 -
279 - /// Check whether the database has existing tables from before the migration system.
280 - async fn has_existing_tables(pool: &SqlitePool) -> Result<bool> {
281 - let row = sqlx::query_as::<_, (i64,)>(
282 - "SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = 'health_checks'",
283 - )
284 - .fetch_one(pool)
285 - .await?;
286 - Ok(row.0 > 0)
287 - }
288 -
289 - /// Execute a single migration's SQL and record it in schema_version.
290 - /// Wrapped in an explicit transaction so partial failures roll back cleanly.
291 - async fn run_one_migration(
292 - pool: &SqlitePool,
293 - version: i64,
294 - description: &str,
295 - sql: &str,
296 - ) -> Result<()> {
297 - info!(version, description, "running migration");
298 -
299 - let mut tx = pool.begin().await?;
300 -
301 - // Execute the whole migration as a batch via `raw_sql`, which hands the
302 - // string to SQLite's own parser to split on statement boundaries. A naive
303 - // `split(';')` breaks on the first `;` inside a trigger body or string
304 - // literal (fuzz-2026-07-06 migration split trap); today's migrations have
305 - // none, but this makes the first such migration safe rather than a startup
306 - // failure.
307 - sqlx::raw_sql(sql).execute(&mut *tx).await?;
308 -
309 - // Record the version inside the same transaction
310 - let now = chrono::Utc::now().to_rfc3339();
311 - sqlx::query(
312 - "INSERT INTO schema_version (version, description, applied_at) VALUES (?, ?, ?)",
313 - )
314 - .bind(version)
315 - .bind(description)
316 - .bind(&now)
317 - .execute(&mut *tx)
318 - .await?;
319 -
320 - tx.commit().await?;
321 - Ok(())
322 - }
323 -
324 - /// Record a version in the schema_version table.
325 - async fn stamp_version(pool: &SqlitePool, version: i64, description: &str) -> Result<()> {
326 - let now = chrono::Utc::now().to_rfc3339();
327 - sqlx::query(
328 - "INSERT INTO schema_version (version, description, applied_at) VALUES (?, ?, ?)",
329 - )
330 - .bind(version)
331 - .bind(description)
332 - .bind(&now)
333 - .execute(pool)
334 - .await?;
335 - Ok(())
336 - }
337 -
338 - // --- Health check queries ---
339 -
340 - #[instrument(skip_all)]
341 - pub async fn insert_health_check(
342 - pool: &SqlitePool,
343 - snapshot: &HealthSnapshot,
344 - ) -> Result<i64> {
345 - let status = snapshot.status.to_string();
346 - let details_json = snapshot
347 - .details
348 - .as_ref()
349 - .map(|d| serde_json::to_string(d).unwrap_or_default());
350 -
351 - let result = sqlx::query(
352 - "INSERT INTO health_checks (target, status, checked_at, response_time_ms, details_json, error)
353 - VALUES (?, ?, ?, ?, ?, ?)",
354 - )
355 - .bind(&snapshot.target)
356 - .bind(&status)
357 - .bind(&snapshot.checked_at)
358 - .bind(snapshot.response_time_ms)
359 - .bind(&details_json)
360 - .bind(&snapshot.error)
361 - .execute(pool)
362 - .await?;
363 -
364 - Ok(result.last_insert_rowid())
365 - }
366 -
367 - #[instrument(skip_all)]
368 - pub async fn get_health_history(
369 - pool: &SqlitePool,
370 - target: Option<&str>,
371 - limit: i64,
372 - ) -> Result<Vec<HealthSnapshot>> {
373 - let rows = match target {
374 - Some(t) => {
375 - sqlx::query_as::<_, HealthCheckRow>(
376 - "SELECT id, target, status, checked_at, response_time_ms, details_json, error
377 - FROM health_checks WHERE target = ? ORDER BY id DESC LIMIT ?",
378 - )
379 - .bind(t)
380 - .bind(limit)
381 - .fetch_all(pool)
382 - .await?
383 - }
384 - None => {
385 - sqlx::query_as::<_, HealthCheckRow>(
386 - "SELECT id, target, status, checked_at, response_time_ms, details_json, error
387 - FROM health_checks ORDER BY id DESC LIMIT ?",
388 - )
389 - .bind(limit)
390 - .fetch_all(pool)
391 - .await?
392 - }
393 - };
394 -
395 - Ok(rows.into_iter().map(|r| r.into_snapshot()).collect())
396 - }
397 -
398 - #[instrument(skip_all)]
399 - pub async fn get_latest_health(
400 - pool: &SqlitePool,
401 - target: &str,
402 - ) -> Result<Option<HealthSnapshot>> {
403 - let row = sqlx::query_as::<_, HealthCheckRow>(
404 - "SELECT id, target, status, checked_at, response_time_ms, details_json, error
405 - FROM health_checks WHERE target = ? ORDER BY id DESC LIMIT 1",
406 - )
407 - .bind(target)
408 - .fetch_optional(pool)
409 - .await?;
410 -
411 - Ok(row.map(|r| r.into_snapshot()))
412 - }
413 -
414 - // --- Test run queries ---
415 -
416 - #[instrument(skip_all)]
417 - pub async fn insert_test_run(
418 - pool: &SqlitePool,
419 - run: &TestRun,
420 - ) -> Result<TestRunId> {
421 - let summary_json = serde_json::to_string(&run.summary).unwrap_or_default();
422 -
423 - let result = sqlx::query(
424 - "INSERT INTO test_runs (target, started_at, finished_at, duration_secs, exit_code, passed, summary_json, raw_output, filter)
425 - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
426 - )
427 - .bind(&run.target)
428 - .bind(&run.started_at)
429 - .bind(&run.finished_at)
430 - .bind(run.duration_secs)
431 - .bind(run.exit_code)
432 - .bind(run.passed)
433 - .bind(&summary_json)
434 - .bind(&run.raw_output)
435 - .bind(&run.filter)
436 - .execute(pool)
437 - .await?;
438 -
439 - Ok(TestRunId(result.last_insert_rowid()))
440 - }
441 -
442 - #[instrument(skip_all)]
443 - pub async fn get_test_history(
444 - pool: &SqlitePool,
445 - target: Option<&str>,
446 - limit: i64,
447 - ) -> Result<Vec<TestRun>> {
448 - let rows = match target {
449 - Some(t) => {
450 - sqlx::query_as::<_, TestRunRow>(
451 - "SELECT id, target, started_at, finished_at, duration_secs, exit_code, passed, summary_json, raw_output, filter
452 - FROM test_runs WHERE target = ? ORDER BY id DESC LIMIT ?",
453 - )
454 - .bind(t)
455 - .bind(limit)
456 - .fetch_all(pool)
457 - .await?
458 - }
459 - None => {
460 - sqlx::query_as::<_, TestRunRow>(
461 - "SELECT id, target, started_at, finished_at, duration_secs, exit_code, passed, summary_json, raw_output, filter
462 - FROM test_runs ORDER BY id DESC LIMIT ?",
463 - )
464 - .bind(limit)
465 - .fetch_all(pool)
466 - .await?
467 - }
468 - };
469 -
470 - Ok(rows.into_iter().map(|r| r.into_test_run()).collect())
471 - }
472 -
473 - #[instrument(skip_all)]
474 - pub async fn get_latest_test_run(
475 - pool: &SqlitePool,
476 - target: &str,
477 - ) -> Result<Option<TestRun>> {
478 - let row = sqlx::query_as::<_, TestRunRow>(
479 - "SELECT id, target, started_at, finished_at, duration_secs, exit_code, passed, summary_json, raw_output, filter
480 - FROM test_runs WHERE target = ? ORDER BY id DESC LIMIT 1",
481 - )
482 - .bind(target)
483 - .fetch_optional(pool)
484 - .await?;
485 -
486 - Ok(row.map(|r| r.into_test_run()))
487 - }
488 -
489 - // --- Test detail queries ---
490 -
491 - /// Insert per-test results for a given test run.
492 - #[instrument(skip_all)]
493 - pub async fn insert_test_details(
494 - pool: &SqlitePool,
495 - run_id: TestRunId,
496 - details: &[TestDetail],
497 - ) -> Result<()> {
498 - // One transaction for the whole detail set: a crash mid-loop otherwise left a
499 - // partial set, which skews get_test_regressions (fuzz-2026-07-06 crash
500 - // atomicity). All-or-nothing.
Lines truncated
A pom/src/db/alerts.rs +173
@@ -0,0 +1,173 @@
1 + use super::*;
2 + use tracing::instrument;
3 +
4 + #[derive(Debug, sqlx::FromRow)]
5 + pub struct AlertRow {
6 + pub id: i64,
7 + pub target: String,
8 + pub alert_type: String,
9 + pub from_status: Option<String>,
10 + pub to_status: Option<String>,
11 + pub sent_at: String,
12 + pub error: Option<String>,
13 + }
14 +
15 + #[instrument(skip_all)]
16 + pub async fn insert_alert(
17 + pool: &SqlitePool,
18 + target: &str,
19 + alert_type: &str,
20 + from_status: Option<&str>,
21 + to_status: Option<&str>,
22 + error: Option<&str>,
23 + ) -> Result<i64> {
24 + let now = chrono::Utc::now().to_rfc3339();
25 + let result = sqlx::query(
26 + "INSERT INTO alerts (target, alert_type, from_status, to_status, sent_at, error)
27 + VALUES (?, ?, ?, ?, ?, ?)",
28 + )
29 + .bind(target)
30 + .bind(alert_type)
31 + .bind(from_status)
32 + .bind(to_status)
33 + .bind(&now)
34 + .bind(error)
35 + .execute(pool)
36 + .await?;
37 + Ok(result.last_insert_rowid())
38 + }
39 +
40 + #[instrument(skip_all)]
41 + pub async fn get_latest_alert_for_target(
42 + pool: &SqlitePool,
43 + target: &str,
44 + ) -> Result<Option<AlertRow>> {
45 + Ok(sqlx::query_as::<_, AlertRow>(
46 + "SELECT id, target, alert_type, from_status, to_status, sent_at, error
47 + FROM alerts WHERE target = ? AND alert_type NOT LIKE '%recovery%'
48 + ORDER BY id DESC LIMIT 1",
49 + )
50 + .bind(target)
51 + .fetch_optional(pool)
52 + .await?)
53 + }
54 +
55 + /// The most recent alert row for a target of a specific type-set. Unlike
56 + /// [`get_latest_alert_for_target`] (which excludes recoveries), this can target
57 + /// recovery rows so recoveries get their own cooldown, and so a recorded recovery
58 + /// can clear a prior failure's cooldown (fail→recover→fail). `like` is a SQL
59 + /// LIKE pattern matched against `alert_type` (e.g. `'%recovery%'`).
60 + #[instrument(skip_all)]
61 + pub async fn get_latest_alert_matching(
62 + pool: &SqlitePool,
63 + target: &str,
64 + like: &str,
65 + ) -> Result<Option<AlertRow>> {
66 + Ok(sqlx::query_as::<_, AlertRow>(
67 + "SELECT id, target, alert_type, from_status, to_status, sent_at, error
68 + FROM alerts WHERE target = ? AND alert_type LIKE ?
69 + ORDER BY id DESC LIMIT 1",
70 + )
71 + .bind(target)
72 + .bind(like)
73 + .fetch_optional(pool)
74 + .await?)
75 + }
76 +
77 + /// A delivery that failed and must be retried, so a transient WAM/Postmark error
78 + /// at a status transition can't silence the whole outage (the transition fires
79 + /// once; this queue is the durable state that outlives it).
80 + #[derive(Debug, Clone, sqlx::FromRow)]
81 + pub struct PendingAlertRow {
82 + pub id: i64,
83 + pub alert_key: String,
84 + pub category: String,
85 + pub channel: String,
86 + pub subject: String,
87 + pub body: String,
88 + pub priority: Option<String>,
89 + pub source: Option<String>,
90 + pub source_ref: Option<String>,
91 + pub from_status: Option<String>,
92 + pub to_status: Option<String>,
93 + pub error: Option<String>,
94 + pub attempts: i64,
95 + }
96 +
97 + /// Fields needed to enqueue an undelivered alert for retry.
98 + #[derive(Debug, Clone)]
99 + pub struct NewPendingAlert<'a> {
100 + pub alert_key: &'a str,
101 + pub category: &'a str,
102 + pub channel: &'a str,
103 + pub subject: &'a str,
104 + pub body: &'a str,
105 + pub priority: Option<&'a str>,
106 + pub source: Option<&'a str>,
107 + pub source_ref: Option<&'a str>,
108 + pub from_status: Option<&'a str>,
109 + pub to_status: Option<&'a str>,
110 + pub error: Option<&'a str>,
111 + }
112 +
113 + #[instrument(skip_all)]
114 + pub async fn enqueue_pending_alert(pool: &SqlitePool, a: &NewPendingAlert<'_>) -> Result<()> {
115 + let now = chrono::Utc::now().to_rfc3339();
116 + sqlx::query(
117 + "INSERT INTO pending_alerts
118 + (alert_key, category, channel, subject, body, priority, source, source_ref,
119 + from_status, to_status, error, attempts, created_at, next_retry_at)
120 + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0, ?, ?)",
121 + )
122 + .bind(a.alert_key)
123 + .bind(a.category)
124 + .bind(a.channel)
125 + .bind(a.subject)
126 + .bind(a.body)
127 + .bind(a.priority)
128 + .bind(a.source)
129 + .bind(a.source_ref)
130 + .bind(a.from_status)
131 + .bind(a.to_status)
132 + .bind(a.error)
133 + .bind(&now)
134 + .bind(&now) // due immediately for the first retry
135 + .execute(pool)
136 + .await?;
137 + Ok(())
138 + }
139 +
140 + /// Pending alerts whose `next_retry_at` is due (<= now), oldest first.
141 + #[instrument(skip_all)]
142 + pub async fn due_pending_alerts(pool: &SqlitePool, limit: i64) -> Result<Vec<PendingAlertRow>> {
143 + let now = chrono::Utc::now().to_rfc3339();
144 + Ok(sqlx::query_as::<_, PendingAlertRow>(
145 + "SELECT id, alert_key, category, channel, subject, body, priority, source, source_ref,
146 + from_status, to_status, error, attempts
147 + FROM pending_alerts WHERE next_retry_at <= ? ORDER BY id ASC LIMIT ?",
148 + )
149 + .bind(&now)
150 + .bind(limit)
151 + .fetch_all(pool)
152 + .await?)
153 + }
154 +
155 + #[instrument(skip_all)]
156 + pub async fn delete_pending_alert(pool: &SqlitePool, id: i64) -> Result<()> {
157 + sqlx::query("DELETE FROM pending_alerts WHERE id = ?")
158 + .bind(id)
159 + .execute(pool)
160 + .await?;
161 + Ok(())
162 + }
163 +
164 + /// Record a failed retry: bump the attempt count and push `next_retry_at` out.
165 + #[instrument(skip_all)]
166 + pub async fn bump_pending_alert(pool: &SqlitePool, id: i64, next_retry_at: &str) -> Result<()> {
167 + sqlx::query("UPDATE pending_alerts SET attempts = attempts + 1, next_retry_at = ? WHERE id = ?")
168 + .bind(next_retry_at)
169 + .bind(id)
170 + .execute(pool)
171 + .await?;
172 + Ok(())
173 + }
A pom/src/db/backup.rs +54
@@ -0,0 +1,54 @@
1 + use super::*;
2 + use tracing::instrument;
3 +
4 + #[instrument(skip_all)]
5 + pub async fn insert_backup_check(
6 + pool: &SqlitePool,
7 + result: &BackupCheckResult,
8 + ) -> Result<i64> {
9 + let row = sqlx::query(
10 + "INSERT INTO backup_checks (target, database_name, status, last_backup_at, size_bytes, age_hours, checked_at, error)
11 + VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
12 + )
13 + .bind(&result.target)
14 + .bind(&result.database_name)
15 + .bind(&result.status)
16 + .bind(&result.last_backup_at)
17 + .bind(result.size_bytes)
18 + .bind(result.age_hours)
19 + .bind(&result.checked_at)
20 + .bind(&result.error)
21 + .execute(pool)
22 + .await?;
23 + Ok(row.last_insert_rowid())
24 + }
25 +
26 + /// Get the latest backup check for a specific target and database.
27 + #[instrument(skip_all)]
28 + pub async fn get_latest_backup_check(
29 + pool: &SqlitePool,
30 + target: &str,
31 + database_name: &str,
32 + ) -> Result<Option<BackupCheckRow>> {
33 + Ok(sqlx::query_as::<_, BackupCheckRow>(
34 + "SELECT id, target, database_name, status, last_backup_at, size_bytes, age_hours, checked_at, error
35 + FROM backup_checks WHERE target = ? AND database_name = ? ORDER BY id DESC LIMIT 1",
36 + )
37 + .bind(target)
38 + .bind(database_name)
39 + .fetch_optional(pool)
40 + .await?)
41 + }
42 +
43 + #[derive(Debug, Clone, sqlx::FromRow, serde::Serialize)]
44 + pub struct BackupCheckRow {
45 + pub id: i64,
46 + pub target: String,
47 + pub database_name: String,
48 + pub status: String,
49 + pub last_backup_at: Option<String>,
50 + pub size_bytes: Option<i64>,
51 + pub age_hours: Option<i64>,
52 + pub checked_at: String,
53 + pub error: Option<String>,
54 + }
A pom/src/db/cors.rs +54
@@ -0,0 +1,54 @@
1 + use super::*;
2 + use tracing::instrument;
3 +
4 + #[instrument(skip_all)]
5 + pub async fn insert_cors_check(
6 + pool: &SqlitePool,
7 + result: &CorsCheckResult,
8 + ) -> Result<i64> {
9 + let row = sqlx::query(
10 + "INSERT INTO cors_checks (target, url, origin, method, passes, checked_at, error)
11 + VALUES (?, ?, ?, ?, ?, ?, ?)",
12 + )
13 + .bind(&result.target)
14 + .bind(&result.url)
15 + .bind(&result.origin)
16 + .bind(&result.method)
17 + .bind(result.passes)
18 + .bind(&result.checked_at)
19 + .bind(&result.error)
20 + .execute(pool)
21 + .await?;
22 + Ok(row.last_insert_rowid())
23 + }
24 +
25 + /// Get the latest CORS check per URL for a target.
26 + #[instrument(skip_all)]
27 + pub async fn get_latest_cors_checks(
28 + pool: &SqlitePool,
29 + target: &str,
30 + ) -> Result<Vec<CorsCheckResult>> {
31 + let rows = sqlx::query_as::<_, (String, String, String, String, bool, String, Option<String>)>(
32 + "SELECT c.target, c.url, c.origin, c.method, c.passes, c.checked_at, c.error
33 + FROM cors_checks c
34 + INNER JOIN (SELECT url, MAX(id) as max_id FROM cors_checks WHERE target = ? GROUP BY url) latest
35 + ON c.id = latest.max_id
36 + ORDER BY c.url",
37 + )
38 + .bind(target)
39 + .fetch_all(pool)
40 + .await?;
41 +
42 + Ok(rows
43 + .into_iter()
44 + .map(|(target, url, origin, method, passes, checked_at, error)| CorsCheckResult {
45 + target,
46 + url,
47 + origin,
48 + method,
49 + passes,
50 + checked_at,
51 + error,
52 + })
53 + .collect())
54 + }
A pom/src/db/dns.rs +92
@@ -0,0 +1,92 @@
1 + use super::*;
2 + use tracing::instrument;
3 +
4 + #[derive(Debug, Clone, sqlx::FromRow, serde::Serialize)]
5 + pub struct DnsCheckRow {
6 + pub id: i64,
7 + pub target: String,
8 + pub name: String,
9 + pub record_type: String,
10 + pub expected: String,
11 + pub actual: String,
12 + pub matches: bool,
13 + pub checked_at: String,
14 + pub error: Option<String>,
15 + }
16 +
17 + #[instrument(skip_all)]
18 + pub async fn insert_dns_check(
19 + pool: &SqlitePool,
20 + result: &DnsCheckResult,
21 + ) -> Result<i64> {
22 + let expected = serde_json::to_string(&result.expected).unwrap_or_default();
23 + let actual = serde_json::to_string(&result.actual).unwrap_or_default();
24 + let record_type_str = result.record_type.to_string();
25 +
26 + let row = sqlx::query(
27 + "INSERT INTO dns_checks (target, name, record_type, expected, actual, matches, checked_at, error)
28 + VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
29 + )
30 + .bind(&result.target)
31 + .bind(&result.name)
32 + .bind(&record_type_str)
33 + .bind(&expected)
34 + .bind(&actual)
35 + .bind(result.matches)
36 + .bind(&result.checked_at)
37 + .bind(&result.error)
38 + .execute(pool)
39 + .await?;
40 + Ok(row.last_insert_rowid())
41 + }
42 +
43 + /// Get the latest DNS check per (name, record_type) for a target.
44 + #[instrument(skip_all)]
45 + pub async fn get_latest_dns_checks(
46 + pool: &SqlitePool,
47 + target: &str,
48 + ) -> Result<Vec<DnsCheckRow>> {
49 + Ok(sqlx::query_as::<_, DnsCheckRow>(
50 + "SELECT d.id, d.target, d.name, d.record_type, d.expected, d.actual, d.matches, d.checked_at, d.error
51 + FROM dns_checks d
52 + INNER JOIN (SELECT name, record_type, MAX(id) as max_id FROM dns_checks WHERE target = ? GROUP BY name, record_type) latest
53 + ON d.id = latest.max_id
54 + ORDER BY d.name, d.record_type",
55 + )
56 + .bind(target)
57 + .fetch_all(pool)
58 + .await?)
59 + }
60 +
61 + /// Delete dns_checks for (name, record_type) pairs no longer in the config.
62 + #[instrument(skip_all)]
63 + pub async fn prune_stale_dns(
64 + pool: &SqlitePool,
65 + target: &str,
66 + expected_dns: &[(String, String)],
67 + ) -> Result<u64> {
68 + if expected_dns.is_empty() {
69 + // No configured DNS records — delete all DNS checks for this target
70 + let result = sqlx::query("DELETE FROM dns_checks WHERE target = ?")
71 + .bind(target)
72 + .execute(pool)
73 + .await?;
74 + return Ok(result.rows_affected());
75 + }
76 +
77 + // Build a compound condition: keep rows matching any configured (name, record_type) pair
78 + let conditions: Vec<String> = expected_dns
79 + .iter()
80 + .map(|_| "(name = ? AND record_type = ?)".to_string())
81 + .collect();
82 + let sql = format!(
83 + "DELETE FROM dns_checks WHERE target = ? AND NOT ({})",
84 + conditions.join(" OR ")
85 + );
86 + let mut query = sqlx::query(&sql).bind(target);
87 + for (name, record_type) in expected_dns {
88 + query = query.bind(name).bind(record_type);
89 + }
90 + let result = query.execute(pool).await?;
91 + Ok(result.rows_affected())
92 + }
A pom/src/db/health.rs +178
@@ -0,0 +1,178 @@
1 + use super::*;
2 + use tracing::instrument;
3 +
4 + #[instrument(skip_all)]
5 + pub async fn insert_health_check(
6 + pool: &SqlitePool,
7 + snapshot: &HealthSnapshot,
8 + ) -> Result<i64> {
9 + let status = snapshot.status.to_string();
10 + let details_json = snapshot
11 + .details
12 + .as_ref()
13 + .map(|d| serde_json::to_string(d).unwrap_or_default());
14 +
15 + let result = sqlx::query(
16 + "INSERT INTO health_checks (target, status, checked_at, response_time_ms, details_json, error)
17 + VALUES (?, ?, ?, ?, ?, ?)",
18 + )
19 + .bind(&snapshot.target)
20 + .bind(&status)
21 + .bind(&snapshot.checked_at)
22 + .bind(snapshot.response_time_ms)
23 + .bind(&details_json)
24 + .bind(&snapshot.error)
25 + .execute(pool)
26 + .await?;
27 +
28 + Ok(result.last_insert_rowid())
29 + }
30 +
31 + #[instrument(skip_all)]
32 + pub async fn get_health_history(
33 + pool: &SqlitePool,
34 + target: Option<&str>,
35 + limit: i64,
36 + ) -> Result<Vec<HealthSnapshot>> {
37 + let rows = match target {
38 + Some(t) => {
39 + sqlx::query_as::<_, HealthCheckRow>(
40 + "SELECT id, target, status, checked_at, response_time_ms, details_json, error
41 + FROM health_checks WHERE target = ? ORDER BY id DESC LIMIT ?",
42 + )
43 + .bind(t)
44 + .bind(limit)
45 + .fetch_all(pool)
46 + .await?
47 + }
48 + None => {
49 + sqlx::query_as::<_, HealthCheckRow>(
50 + "SELECT id, target, status, checked_at, response_time_ms, details_json, error
51 + FROM health_checks ORDER BY id DESC LIMIT ?",
52 + )
53 + .bind(limit)
54 + .fetch_all(pool)
55 + .await?
56 + }
57 + };
58 +
59 + Ok(rows.into_iter().map(|r| r.into_snapshot()).collect())
60 + }
61 +
62 + #[instrument(skip_all)]
63 + pub async fn get_latest_health(
64 + pool: &SqlitePool,
65 + target: &str,
66 + ) -> Result<Option<HealthSnapshot>> {
67 + let row = sqlx::query_as::<_, HealthCheckRow>(
68 + "SELECT id, target, status, checked_at, response_time_ms, details_json, error
69 + FROM health_checks WHERE target = ? ORDER BY id DESC LIMIT 1",
70 + )
71 + .bind(target)
72 + .fetch_optional(pool)
73 + .await?;
74 +
75 + Ok(row.map(|r| r.into_snapshot()))
76 + }
77 +
78 + /// Calculate uptime percentage for a target over the given number of hours.
79 + /// Returns the percentage of health checks with "operational" status.
80 + #[instrument(skip_all)]
81 + pub async fn get_uptime_percent(
82 + pool: &SqlitePool,
83 + target: &str,
84 + hours: i64,
85 + ) -> Result<Option<f64>> {
86 + let cutoff = chrono::Utc::now() - chrono::Duration::hours(hours);
87 + let cutoff_str = cutoff.to_rfc3339();
88 +
89 + let row = sqlx::query_as::<_, (i64, i64)>(
90 + "SELECT
91 + COUNT(*) as total,
92 + SUM(CASE WHEN status = 'operational' THEN 1 ELSE 0 END) as operational
93 + FROM health_checks
94 + WHERE target = ? AND checked_at >= ?",
95 + )
96 + .bind(target)
97 + .bind(&cutoff_str)
98 + .fetch_one(pool)
99 + .await?;
100 +
101 + if row.0 == 0 {
102 + Ok(None)
103 + } else {
104 + Ok(Some(row.1 as f64 / row.0 as f64 * 100.0))
105 + }
106 + }
107 +
108 + /// Fetch all response times for a target since a given timestamp, ordered ASC.
109 + #[instrument(skip_all)]
110 + pub async fn get_response_times(
111 + pool: &SqlitePool,
112 + target: &str,
113 + since_rfc3339: &str,
114 + ) -> Result<Vec<(String, i64)>> {
115 + let rows = sqlx::query_as::<_, (String, i64)>(
116 + "SELECT checked_at, response_time_ms FROM health_checks
117 + WHERE target = ? AND checked_at >= ?
118 + ORDER BY checked_at ASC",
119 + )
120 + .bind(target)
121 + .bind(since_rfc3339)
122 + .fetch_all(pool)
123 + .await?;
124 + Ok(rows)
125 + }
126 +
127 + /// Fetch the last N response times for **operational** checks only (most recent first).
128 + #[instrument(skip_all)]
129 + pub async fn get_recent_response_times(
130 + pool: &SqlitePool,
131 + target: &str,
132 + count: i64,
133 + ) -> Result<Vec<i64>> {
134 + let rows = sqlx::query_as::<_, (i64,)>(
135 + "SELECT response_time_ms FROM health_checks
136 + WHERE target = ? AND status = 'operational'
137 + ORDER BY id DESC LIMIT ?",
138 + )
139 + .bind(target)
140 + .bind(count)
141 + .fetch_all(pool)
142 + .await?;
143 + Ok(rows.into_iter().map(|r| r.0).collect())
144 + }
145 +
146 + #[derive(sqlx::FromRow)]
147 + struct HealthCheckRow {
148 + id: i64,
149 + target: String,
150 + status: String,
151 + checked_at: String,
152 + response_time_ms: i64,
153 + details_json: Option<String>,
154 + error: Option<String>,
155 + }
156 +
157 + impl HealthCheckRow {
158 + fn into_snapshot(self) -> HealthSnapshot {
159 + let status = self
160 + .status
161 + .parse::<HealthStatus>()
162 + .unwrap_or(HealthStatus::Error);
163 + let details = self
164 + .details_json
165 + .as_deref()
166 + .and_then(|s| serde_json::from_str::<HealthDetails>(s).ok());
167 +
168 + HealthSnapshot {
169 + id: Some(self.id),
170 + target: self.target,
171 + status,
172 + checked_at: self.checked_at,
173 + response_time_ms: self.response_time_ms,
174 + details,
175 + error: self.error,
176 + }
177 + }
178 + }
A pom/src/db/incidents.rs +117
@@ -0,0 +1,117 @@
1 + use super::*;
2 + use tracing::instrument;
3 +
4 + #[derive(Debug, Clone, sqlx::FromRow, serde::Serialize)]
5 + pub struct IncidentRow {
6 + pub id: i64,
7 + pub target: String,
8 + pub started_at: String,
9 + pub ended_at: Option<String>,
10 + pub duration_secs: Option<i64>,
11 + pub from_status: String,
12 + pub to_status: String,
13 + }
14 +
15 + #[instrument(skip_all)]
16 + pub async fn insert_incident(
17 + pool: &SqlitePool,
18 + target: &str,
19 + from_status: &str,
20 + to_status: &str,
21 + ) -> Result<i64> {
22 + let now = chrono::Utc::now().to_rfc3339();
23 + let result = sqlx::query(
24 + "INSERT INTO incidents (target, started_at, from_status, to_status)
25 + VALUES (?, ?, ?, ?)",
26 + )
27 + .bind(target)
28 + .bind(&now)
29 + .bind(from_status)
30 + .bind(to_status)
31 + .execute(pool)
32 + .await?;
33 + Ok(result.last_insert_rowid())
34 + }
35 +
36 + #[instrument(skip_all)]
37 + pub async fn close_open_incidents(
38 + pool: &SqlitePool,
39 + target: &str,
40 + ) -> Result<u64> {
41 + let now = chrono::Utc::now().to_rfc3339();
42 + let result = sqlx::query(
43 + "UPDATE incidents SET ended_at = ?, duration_secs = CAST((julianday(?) - julianday(started_at)) * 86400 AS INTEGER)
44 + WHERE target = ? AND ended_at IS NULL",
45 + )
46 + .bind(&now)
47 + .bind(&now)
48 + .bind(target)
49 + .execute(pool)
50 + .await?;
51 + Ok(result.rows_affected())
52 + }
53 +
54 + /// Close any open incidents for `target` and open a new one, atomically. A status
55 + /// change like `error -> unreachable` closes the old incident and opens the new
56 + /// one; done as two pool calls, a crash between them left the old incident closed
57 + /// and the new one never recorded (fuzz-2026-07-06 crash atomicity). One tx.
58 + #[instrument(skip_all)]
59 + pub async fn close_and_open_incident(
60 + pool: &SqlitePool,
61 + target: &str,
62 + from_status: &str,
63 + to_status: &str,
64 + ) -> Result<()> {
65 + let now = chrono::Utc::now().to_rfc3339();
66 + let mut tx = pool.begin().await?;
67 + sqlx::query(
68 + "UPDATE incidents SET ended_at = ?, duration_secs = CAST((julianday(?) - julianday(started_at)) * 86400 AS INTEGER)
69 + WHERE target = ? AND ended_at IS NULL",
70 + )
71 + .bind(&now)
72 + .bind(&now)
73 + .bind(target)
74 + .execute(&mut *tx)
75 + .await?;
76 + sqlx::query(
77 + "INSERT INTO incidents (target, started_at, from_status, to_status) VALUES (?, ?, ?, ?)",
78 + )
79 + .bind(target)
80 + .bind(&now)
81 + .bind(from_status)
82 + .bind(to_status)
83 + .execute(&mut *tx)
84 + .await?;
85 + tx.commit().await?;
86 + Ok(())
87 + }
88 +
89 + #[instrument(skip_all)]
90 + pub async fn get_open_incident(
91 + pool: &SqlitePool,
92 + target: &str,
93 + ) -> Result<Option<IncidentRow>> {
94 + Ok(sqlx::query_as::<_, IncidentRow>(
95 + "SELECT id, target, started_at, ended_at, duration_secs, from_status, to_status
96 + FROM incidents WHERE target = ? AND ended_at IS NULL ORDER BY id DESC LIMIT 1",
97 + )
98 + .bind(target)
99 + .fetch_optional(pool)
100 + .await?)
101 + }
102 +
103 + #[instrument(skip_all)]
104 + pub async fn get_recent_incidents(
105 + pool: &SqlitePool,
106 + target: &str,
107 + limit: i64,
108 + ) -> Result<Vec<IncidentRow>> {
109 + Ok(sqlx::query_as::<_, IncidentRow>(
110 + "SELECT id, target, started_at, ended_at, duration_secs, from_status, to_status
111 + FROM incidents WHERE target = ? ORDER BY id DESC LIMIT ?",
112 + )
113 + .bind(target)
114 + .bind(limit)
115 + .fetch_all(pool)
116 + .await?)
117 + }
A pom/src/db/maintenance.rs +105
@@ -0,0 +1,105 @@
1 + use super::*;
2 + use tracing::instrument;
3 +
4 + /// Prune result with counts for each table.
5 + pub struct PruneResult {
6 + pub health: u64,
7 + pub tests: u64,
8 + pub test_details: u64,
9 + pub heartbeats: u64,
10 + pub alerts: u64,
11 + pub tls: u64,
12 + pub incidents: u64,
13 + pub routes: u64,
14 + pub dns: u64,
15 + pub whois: u64,
16 + pub backups: u64,
17 + }
18 +
19 + /// Delete records older than `days` from all tables.
20 + /// Only closed incidents (with a non-NULL `ended_at`) are pruned.
21 + #[instrument(skip_all)]
22 + pub async fn prune_old_records(
23 + pool: &SqlitePool,
24 + days: i64,
25 + ) -> Result<PruneResult> {
26 + // Guard: days <= 0 would set cutoff to now (or the future), deleting
27 + // everything. Treat this as a no-op instead.
28 + if days <= 0 {
29 + return Ok(PruneResult { health: 0, tests: 0, test_details: 0, heartbeats: 0, alerts: 0, tls: 0, incidents: 0, routes: 0, dns: 0, whois: 0, backups: 0 });
30 + }
31 +
32 + let cutoff = chrono::Utc::now() - chrono::Duration::days(days);
33 + let cutoff_str = cutoff.to_rfc3339();
34 +
35 + let health_result = sqlx::query("DELETE FROM health_checks WHERE checked_at < ?")
36 + .bind(&cutoff_str)
37 + .execute(pool)
38 + .await?;
39 +
40 + let test_result = sqlx::query("DELETE FROM test_runs WHERE started_at < ?")
41 + .bind(&cutoff_str)
42 + .execute(pool)
43 + .await?;
44 +
45 + // Prune orphaned test_details (run was deleted above).
46 + let test_details_result = sqlx::query(
47 + "DELETE FROM test_details WHERE run_id NOT IN (SELECT id FROM test_runs)",
48 + )
49 + .execute(pool)
50 + .await?;
51 +
52 + let peer_hb_result = sqlx::query("DELETE FROM peer_heartbeats WHERE checked_at < ?")
53 + .bind(&cutoff_str)
54 + .execute(pool)
55 + .await?;
56 +
57 + let alerts_result = sqlx::query("DELETE FROM alerts WHERE sent_at < ?")
58 + .bind(&cutoff_str)
59 + .execute(pool)
60 + .await?;
61 +
62 + let tls_result = sqlx::query("DELETE FROM tls_checks WHERE checked_at < ?")
63 + .bind(&cutoff_str)
64 + .execute(pool)
65 + .await?;
66 +
67 + let incidents_result = sqlx::query("DELETE FROM incidents WHERE ended_at IS NOT NULL AND ended_at < ?")
68 + .bind(&cutoff_str)
69 + .execute(pool)
70 + .await?;
71 +
72 + let routes_result = sqlx::query("DELETE FROM route_checks WHERE checked_at < ?")
73 + .bind(&cutoff_str)
74 + .execute(pool)
75 + .await?;
76 +
77 + let dns_result = sqlx::query("DELETE FROM dns_checks WHERE checked_at < ?")
78 + .bind(&cutoff_str)
79 + .execute(pool)
80 + .await?;
81 +
82 + let whois_result = sqlx::query("DELETE FROM whois_checks WHERE checked_at < ?")
83 + .bind(&cutoff_str)
84 + .execute(pool)
85 + .await?;
86 +
87 + let backups_result = sqlx::query("DELETE FROM backup_checks WHERE checked_at < ?")
88 + .bind(&cutoff_str)
89 + .execute(pool)
90 + .await?;
91 +
92 + Ok(PruneResult {
93 + health: health_result.rows_affected(),
94 + tests: test_result.rows_affected(),
95 + test_details: test_details_result.rows_affected(),
96 + heartbeats: peer_hb_result.rows_affected(),
97 + alerts: alerts_result.rows_affected(),
98 + tls: tls_result.rows_affected(),
99 + incidents: incidents_result.rows_affected(),
100 + routes: routes_result.rows_affected(),
101 + dns: dns_result.rows_affected(),
102 + whois: whois_result.rows_affected(),
103 + backups: backups_result.rows_affected(),
104 + })
105 + }
A pom/src/db/migrations.rs +324
@@ -0,0 +1,324 @@
1 + use super::*;
2 + use tracing::{info, instrument};
3 +
4 + /// Each migration is a (version, description, SQL) tuple. Versions start at 1.
5 + /// The SQL may contain multiple statements separated by semicolons.
6 + const MIGRATIONS: &[(i64, &str, &str)] = &[
7 + (1, "initial schema", r#"
8 + CREATE TABLE IF NOT EXISTS health_checks (
9 + id INTEGER PRIMARY KEY AUTOINCREMENT,
10 + target TEXT NOT NULL,
11 + status TEXT NOT NULL,
12 + checked_at TEXT NOT NULL,
13 + response_time_ms INTEGER NOT NULL,
14 + details_json TEXT,
15 + error TEXT
16 + );
17 + CREATE TABLE IF NOT EXISTS test_runs (
18 + id INTEGER PRIMARY KEY AUTOINCREMENT,
19 + target TEXT NOT NULL,
20 + started_at TEXT NOT NULL,
21 + finished_at TEXT,
22 + duration_secs INTEGER,
23 + exit_code INTEGER,
24 + passed INTEGER NOT NULL,
25 + summary_json TEXT NOT NULL,
26 + raw_output TEXT NOT NULL,
27 + filter TEXT
28 + );
29 + CREATE TABLE IF NOT EXISTS peer_identities (
30 + peer_name TEXT PRIMARY KEY,
31 + instance_id TEXT NOT NULL,
32 + first_seen TEXT NOT NULL
33 + );
34 + CREATE TABLE IF NOT EXISTS peer_heartbeats (
35 + id INTEGER PRIMARY KEY AUTOINCREMENT,
36 + peer_name TEXT NOT NULL,
37 + status TEXT NOT NULL,
38 + latency_ms INTEGER NOT NULL,
39 + checked_at TEXT NOT NULL
40 + );
41 + CREATE INDEX IF NOT EXISTS idx_health_checks_target_id ON health_checks(target, id DESC);
42 + CREATE INDEX IF NOT EXISTS idx_health_checks_target_checked ON health_checks(target, checked_at);
43 + CREATE INDEX IF NOT EXISTS idx_test_runs_target_id ON test_runs(target, id DESC);
44 + CREATE INDEX IF NOT EXISTS idx_peer_heartbeats_peer_id ON peer_heartbeats(peer_name, id DESC);
45 + "#),
46 + (2, "add alerts table", r#"
47 + CREATE TABLE IF NOT EXISTS alerts (
48 + id INTEGER PRIMARY KEY AUTOINCREMENT,
49 + target TEXT NOT NULL,
50 + alert_type TEXT NOT NULL,
51 + from_status TEXT,
52 + to_status TEXT,
53 + sent_at TEXT NOT NULL,
54 + error TEXT
55 + );
56 + CREATE INDEX IF NOT EXISTS idx_alerts_target_sent ON alerts(target, sent_at);
57 + "#),
58 + (3, "add tls_checks table", r#"
59 + CREATE TABLE IF NOT EXISTS tls_checks (
60 + id INTEGER PRIMARY KEY AUTOINCREMENT,
61 + target TEXT NOT NULL,
62 + host TEXT NOT NULL,
63 + valid INTEGER NOT NULL,
64 + days_remaining INTEGER NOT NULL,
65 + not_before TEXT NOT NULL,
66 + not_after TEXT NOT NULL,
67 + subject TEXT NOT NULL,
68 + issuer TEXT NOT NULL,
69 + checked_at TEXT NOT NULL,
70 + error TEXT
71 + );
72 + CREATE INDEX IF NOT EXISTS idx_tls_checks_target_id ON tls_checks(target, id DESC);
73 + "#),
74 + (4, "add incidents table", r#"
75 + CREATE TABLE IF NOT EXISTS incidents (
76 + id INTEGER PRIMARY KEY AUTOINCREMENT,
77 + target TEXT NOT NULL,
78 + started_at TEXT NOT NULL,
79 + ended_at TEXT,
80 + duration_secs INTEGER,
81 + from_status TEXT NOT NULL,
82 + to_status TEXT NOT NULL
83 + );
84 + CREATE INDEX IF NOT EXISTS idx_incidents_target_id ON incidents(target, id DESC);
85 + "#),
86 + (5, "add route_checks table", r#"
87 + CREATE TABLE IF NOT EXISTS route_checks (
88 + id INTEGER PRIMARY KEY AUTOINCREMENT,
89 + target TEXT NOT NULL,
90 + path TEXT NOT NULL,
91 + status_code INTEGER NOT NULL,
92 + ok INTEGER NOT NULL,
93 + response_time_ms INTEGER NOT NULL,
94 + checked_at TEXT NOT NULL,
95 + error TEXT
96 + );
97 + CREATE INDEX IF NOT EXISTS idx_route_checks_target_path ON route_checks(target, path, id DESC);
98 + CREATE INDEX IF NOT EXISTS idx_route_checks_target ON route_checks(target, checked_at DESC);
99 + "#),
100 + (6, "add dns_checks and whois_checks tables", r#"
101 + CREATE TABLE IF NOT EXISTS dns_checks (
102 + id INTEGER PRIMARY KEY AUTOINCREMENT,
103 + target TEXT NOT NULL,
104 + name TEXT NOT NULL,
105 + record_type TEXT NOT NULL,
106 + expected TEXT NOT NULL,
107 + actual TEXT NOT NULL,
108 + matches INTEGER NOT NULL,
109 + checked_at TEXT NOT NULL,
110 + error TEXT
111 + );
112 + CREATE INDEX IF NOT EXISTS idx_dns_checks_target ON dns_checks(target, name, id DESC);
113 +
114 + CREATE TABLE IF NOT EXISTS whois_checks (
115 + id INTEGER PRIMARY KEY AUTOINCREMENT,
116 + target TEXT NOT NULL,
117 + domain TEXT NOT NULL,
118 + registrar TEXT,
119 + expiry_date TEXT,
120 + days_remaining INTEGER,
121 + nameservers TEXT,
122 + checked_at TEXT NOT NULL,
123 + error TEXT
124 + );
125 + CREATE INDEX IF NOT EXISTS idx_whois_checks_target ON whois_checks(target, id DESC);
126 + "#),
127 + (7, "add test_details table", r#"
128 + CREATE TABLE IF NOT EXISTS test_details (
129 + id INTEGER PRIMARY KEY AUTOINCREMENT,
130 + run_id INTEGER NOT NULL REFERENCES test_runs(id) ON DELETE CASCADE,
131 + test_name TEXT NOT NULL,
132 + passed INTEGER NOT NULL,
133 + duration_ms INTEGER
134 + );
135 + CREATE INDEX IF NOT EXISTS idx_test_details_run_id ON test_details(run_id);
136 + CREATE INDEX IF NOT EXISTS idx_test_details_name ON test_details(test_name, run_id DESC);
137 + "#),
138 + (8, "add cors_checks table", r#"
139 + CREATE TABLE IF NOT EXISTS cors_checks (
140 + id INTEGER PRIMARY KEY AUTOINCREMENT,
141 + target TEXT NOT NULL,
142 + url TEXT NOT NULL,
143 + origin TEXT NOT NULL,
144 + method TEXT NOT NULL,
145 + passes INTEGER NOT NULL,
146 + checked_at TEXT NOT NULL,
147 + error TEXT
148 + );
149 + CREATE INDEX IF NOT EXISTS idx_cors_target_id ON cors_checks(target, id DESC);
150 + "#),
151 + (9, "add backup_checks table", r#"
152 + CREATE TABLE IF NOT EXISTS backup_checks (
153 + id INTEGER PRIMARY KEY AUTOINCREMENT,
154 + target TEXT NOT NULL,
155 + database_name TEXT NOT NULL,
156 + status TEXT NOT NULL,
157 + last_backup_at TEXT,
158 + size_bytes INTEGER,
159 + age_hours INTEGER,
160 + checked_at TEXT NOT NULL,
161 + error TEXT
162 + );
163 + CREATE INDEX IF NOT EXISTS idx_backup_checks_target ON backup_checks(target, id DESC);
164 + "#),
165 + (10, "add pending_alerts retry queue", r#"
166 + CREATE TABLE IF NOT EXISTS pending_alerts (
167 + id INTEGER PRIMARY KEY AUTOINCREMENT,
168 + alert_key TEXT NOT NULL,
169 + category TEXT NOT NULL,
170 + channel TEXT NOT NULL, -- 'wam' or 'email'
171 + subject TEXT NOT NULL,
172 + body TEXT NOT NULL,
173 + priority TEXT, -- WAM only
174 + source TEXT, -- WAM only
175 + source_ref TEXT, -- WAM only
176 + from_status TEXT,
177 + to_status TEXT,
178 + error TEXT,
179 + attempts INTEGER NOT NULL DEFAULT 0,
180 + created_at TEXT NOT NULL,
181 + next_retry_at TEXT NOT NULL
182 + );
183 + CREATE INDEX IF NOT EXISTS idx_pending_alerts_due ON pending_alerts(next_retry_at, id);
184 + "#),
185 + ];
186 +
187 + #[instrument(skip_all)]
188 + pub async fn connect(path: &Path) -> Result<SqlitePool> {
189 + let opts = SqliteConnectOptions::from_str(&format!("sqlite:{}", path.display()))?
190 + .create_if_missing(true)
191 + .foreign_keys(true)
192 + .journal_mode(sqlx::sqlite::SqliteJournalMode::Wal);
193 +
194 + let pool = SqlitePoolOptions::new()
195 + .max_connections(5)
196 + .connect_with(opts)
197 + .await?;
198 +
199 + run_migrations(&pool).await?;
200 + Ok(pool)
201 + }
202 +
203 + #[instrument(skip_all)]
204 + pub async fn connect_in_memory() -> Result<SqlitePool> {
205 + let opts = SqliteConnectOptions::from_str("sqlite::memory:")?.foreign_keys(true);
206 + let pool = SqlitePoolOptions::new()
207 + .max_connections(1)
208 + .connect_with(opts)
209 + .await?;
210 +
211 + run_migrations(&pool).await?;
212 + Ok(pool)
213 + }
214 +
215 + /// Run pending schema migrations. Detects pre-migration databases by checking
216 + /// for existing tables and stamps them as version 1 without re-running.
217 + #[instrument(skip_all)]
218 + pub async fn run_migrations(pool: &SqlitePool) -> Result<()> {
219 + // Ensure the schema_version table exists
220 + sqlx::query(
221 + "CREATE TABLE IF NOT EXISTS schema_version (
222 + version INTEGER NOT NULL,
223 + description TEXT NOT NULL,
224 + applied_at TEXT NOT NULL
225 + )",
226 + )
227 + .execute(pool)
228 + .await?;
229 +
230 + let current_version = get_schema_version(pool).await?;
231 +
232 + // Detect pre-migration databases: if schema_version is empty but tables exist,
233 + // this is an existing database that predates the migration system.
234 + if current_version == 0 && has_existing_tables(pool).await? {
235 + info!("detected pre-migration database, stamping as version 1");
236 + stamp_version(pool, 1, "initial schema (pre-existing)").await?;
237 + // Run remaining migrations (2+) if any
238 + for &(version, description, sql) in MIGRATIONS {
239 + if version > 1 {
240 + run_one_migration(pool, version, description, sql).await?;
241 + }
242 + }
243 + return Ok(());
244 + }
245 +
246 + // Run all migrations newer than current version
247 + for &(version, description, sql) in MIGRATIONS {
248 + if version > current_version {
249 + run_one_migration(pool, version, description, sql).await?;
250 + }
251 + }
252 +
253 + Ok(())
254 + }
255 +
256 + /// Get the current schema version (0 if no migrations have been applied).
257 + #[instrument(skip_all)]
258 + pub async fn get_schema_version(pool: &SqlitePool) -> Result<i64> {
259 + let row = sqlx::query_as::<_, (i64,)>(
260 + "SELECT COALESCE(MAX(version), 0) FROM schema_version",
261 + )
262 + .fetch_one(pool)
263 + .await?;
264 + Ok(row.0)
265 + }
266 +
267 + /// Check whether the database has existing tables from before the migration system.
268 + async fn has_existing_tables(pool: &SqlitePool) -> Result<bool> {
269 + let row = sqlx::query_as::<_, (i64,)>(
270 + "SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = 'health_checks'",
271 + )
272 + .fetch_one(pool)
273 + .await?;
274 + Ok(row.0 > 0)
275 + }
276 +
277 + /// Execute a single migration's SQL and record it in schema_version.
278 + /// Wrapped in an explicit transaction so partial failures roll back cleanly.
279 + async fn run_one_migration(
280 + pool: &SqlitePool,
281 + version: i64,
282 + description: &str,
283 + sql: &str,
284 + ) -> Result<()> {
285 + info!(version, description, "running migration");
286 +
287 + let mut tx = pool.begin().await?;
288 +
289 + // Execute the whole migration as a batch via `raw_sql`, which hands the
290 + // string to SQLite's own parser to split on statement boundaries. A naive
291 + // `split(';')` breaks on the first `;` inside a trigger body or string
292 + // literal (fuzz-2026-07-06 migration split trap); today's migrations have
293 + // none, but this makes the first such migration safe rather than a startup
294 + // failure.
295 + sqlx::raw_sql(sql).execute(&mut *tx).await?;
296 +
297 + // Record the version inside the same transaction
298 + let now = chrono::Utc::now().to_rfc3339();
299 + sqlx::query(
300 + "INSERT INTO schema_version (version, description, applied_at) VALUES (?, ?, ?)",
301 + )
302 + .bind(version)
303 + .bind(description)
304 + .bind(&now)
305 + .execute(&mut *tx)
306 + .await?;
307 +
308 + tx.commit().await?;
309 + Ok(())
310 + }
311 +
312 + /// Record a version in the schema_version table.
313 + async fn stamp_version(pool: &SqlitePool, version: i64, description: &str) -> Result<()> {
314 + let now = chrono::Utc::now().to_rfc3339();
315 + sqlx::query(
316 + "INSERT INTO schema_version (version, description, applied_at) VALUES (?, ?, ?)",
317 + )
318 + .bind(version)
319 + .bind(description)
320 + .bind(&now)
321 + .execute(pool)
322 + .await?;
323 + Ok(())
324 + }
A pom/src/db/mod.rs +41
@@ -0,0 +1,41 @@
1 + //! SQLite persistence — schema, health checks, test runs, and peer data.
2 + //!
3 + //! Uses a migration versioning system: each schema change is a numbered migration
4 + //! stored in [`MIGRATIONS`]. On startup, [`run_migrations`] checks the current
5 + //! version and runs any pending migrations. Existing databases (pre-migration)
6 + //! are detected by the presence of the `health_checks` table and marked as v1.
7 +
8 + use sqlx::sqlite::{SqliteConnectOptions, SqlitePool, SqlitePoolOptions};
9 + use std::path::Path;
10 + use std::str::FromStr;
11 +
12 + use crate::error::Result;
13 + use crate::types::{BackupCheckResult, CorsCheckResult, DnsCheckResult, HealthDetails, HealthSnapshot, HealthStatus, TestDetail, TestRun, TestRunId, TestSummary, TlsStatus, WhoisResult};
14 +
15 + mod migrations;
16 + mod health;
17 + mod test_runs;
18 + mod alerts;
19 + mod tls;
20 + mod incidents;
21 + mod routes;
22 + mod dns;
23 + mod whois;
24 + mod cors;
25 + mod backup;
26 + mod maintenance;
27 + mod peers;
28 +
29 + pub use migrations::*;
30 + pub use health::*;
31 + pub use test_runs::*;
32 + pub use alerts::*;
33 + pub use tls::*;
34 + pub use incidents::*;
35 + pub use routes::*;
36 + pub use dns::*;
37 + pub use whois::*;
38 + pub use cors::*;
39 + pub use backup::*;
40 + pub use maintenance::*;
41 + pub use peers::*;
A pom/src/db/peers.rs +103
@@ -0,0 +1,103 @@
1 + use super::*;
2 + use tracing::instrument;
3 +
4 + /// Store a peer's identity (first-seen UUID). INSERT OR IGNORE — first wins.
5 + #[instrument(skip_all)]
6 + pub async fn store_peer_identity(
7 + pool: &SqlitePool,
8 + peer_name: &str,
9 + instance_id: &str,
10 + ) -> Result<()> {
11 + let now = chrono::Utc::now().to_rfc3339();
12 + sqlx::query(
13 + "INSERT OR IGNORE INTO peer_identities (peer_name, instance_id, first_seen)
14 + VALUES (?, ?, ?)",
15 + )
16 + .bind(peer_name)
17 + .bind(instance_id)
18 + .bind(&now)
19 + .execute(pool)
20 + .await?;
21 + Ok(())
22 + }
23 +
24 + /// Update a peer's identity (UUID changed, e.g. after reinstall).
25 + #[instrument(skip_all)]
26 + pub async fn update_peer_identity(
27 + pool: &SqlitePool,
28 + peer_name: &str,
29 + instance_id: &str,
30 + ) -> Result<()> {
31 + let now = chrono::Utc::now().to_rfc3339();
32 + sqlx::query(
33 + "UPDATE peer_identities SET instance_id = ?, first_seen = ?
34 + WHERE peer_name = ?",
35 + )
36 + .bind(instance_id)
37 + .bind(&now)
38 + .bind(peer_name)
39 + .execute(pool)
40 + .await?;
41 + Ok(())
42 + }
43 +
44 + /// Get a peer's first-seen instance ID.
45 + #[instrument(skip_all)]
46 + pub async fn get_peer_identity(
47 + pool: &SqlitePool,
48 + peer_name: &str,
49 + ) -> Result<Option<String>> {
50 + let row = sqlx::query_as::<_, (String,)>(
51 + "SELECT instance_id FROM peer_identities WHERE peer_name = ?",
52 + )
53 + .bind(peer_name)
54 + .fetch_optional(pool)
55 + .await?;
56 + Ok(row.map(|r| r.0))
57 + }
58 +
59 + #[instrument(skip_all)]
60 + pub async fn insert_peer_heartbeat(
61 + pool: &SqlitePool,
62 + peer_name: &str,
63 + status: &str,
64 + latency_ms: i64,
65 + ) -> Result<i64> {
66 + let now = chrono::Utc::now().to_rfc3339();
67 + let result = sqlx::query(
68 + "INSERT INTO peer_heartbeats (peer_name, status, latency_ms, checked_at)
69 + VALUES (?, ?, ?, ?)",
70 + )
71 + .bind(peer_name)
72 + .bind(status)
73 + .bind(latency_ms)
74 + .bind(&now)
75 + .execute(pool)
76 + .await?;
77 + Ok(result.last_insert_rowid())
78 + }
79 +
80 + #[instrument(skip_all)]
81 + pub async fn get_peer_heartbeat_history(
82 + pool: &SqlitePool,
83 + peer_name: &str,
84 + limit: i64,
85 + ) -> Result<Vec<PeerHeartbeatRow>> {
86 + Ok(sqlx::query_as::<_, PeerHeartbeatRow>(
87 + "SELECT id, peer_name, status, latency_ms, checked_at
88 + FROM peer_heartbeats WHERE peer_name = ? ORDER BY id DESC LIMIT ?",
89 + )
90 + .bind(peer_name)
91 + .bind(limit)
92 + .fetch_all(pool)
93 + .await?)
94 + }
95 +
96 + #[derive(Debug, sqlx::FromRow, serde::Serialize)]
97 + pub struct PeerHeartbeatRow {
98 + pub id: i64,
99 + pub peer_name: String,
100 + pub status: String,
101 + pub latency_ms: i64,
102 + pub checked_at: String,
103 + }
A pom/src/db/routes.rs +83
@@ -0,0 +1,83 @@
1 + use super::*;
2 + use tracing::instrument;
3 +
4 + #[derive(Debug, Clone, sqlx::FromRow, serde::Serialize)]
5 + pub struct RouteCheckRow {
6 + pub id: i64,
7 + pub target: String,
8 + pub path: String,
9 + pub status_code: i64,
10 + pub ok: bool,
11 + pub response_time_ms: i64,
12 + pub checked_at: String,
13 + pub error: Option<String>,
14 + }
15 +
16 + #[instrument(skip_all)]
17 + pub async fn insert_route_check(
18 + pool: &SqlitePool,
19 + result: &crate::checks::routes::RouteCheckResult,
20 + ) -> Result<i64> {
21 + let row = sqlx::query(
22 + "INSERT INTO route_checks (target, path, status_code, ok, response_time_ms, checked_at, error)
23 + VALUES (?, ?, ?, ?, ?, ?, ?)",
24 + )
25 + .bind(&result.target)
26 + .bind(&result.path)
27 + .bind(result.status_code as i64)
28 + .bind(result.ok)
29 + .bind(result.response_time_ms)
30 + .bind(&result.checked_at)
31 + .bind(&result.error)
32 + .execute(pool)
33 + .await?;
34 + Ok(row.last_insert_rowid())
35 + }
36 +
37 + /// Get the latest route check per path for a target.
38 + #[instrument(skip_all)]
39 + pub async fn get_latest_route_checks(
40 + pool: &SqlitePool,
41 + target: &str,
42 + ) -> Result<Vec<RouteCheckRow>> {
43 + Ok(sqlx::query_as::<_, RouteCheckRow>(
44 + "SELECT r.id, r.target, r.path, r.status_code, r.ok, r.response_time_ms, r.checked_at, r.error
45 + FROM route_checks r
46 + INNER JOIN (SELECT path, MAX(id) as max_id FROM route_checks WHERE target = ? GROUP BY path) latest
47 + ON r.id = latest.max_id
48 + ORDER BY r.path",
49 + )
50 + .bind(target)
51 + .fetch_all(pool)
52 + .await?)
53 + }
54 +
55 + /// Delete route_checks for paths no longer in the config.
56 + #[instrument(skip_all)]
57 + pub async fn prune_stale_routes(
58 + pool: &SqlitePool,
59 + target: &str,
60 + expected_routes: &[String],
61 + ) -> Result<u64> {
62 + if expected_routes.is_empty() {
63 + // No configured routes — delete all route checks for this target
64 + let result = sqlx::query("DELETE FROM route_checks WHERE target = ?")
65 + .bind(target)
66 + .execute(pool)
67 + .await?;
68 + return Ok(result.rows_affected());
69 + }
70 +
71 + // Build placeholders for the IN clause
72 + let placeholders: Vec<&str> = expected_routes.iter().map(|_| "?").collect();
73 + let sql = format!(
74 + "DELETE FROM route_checks WHERE target = ? AND path NOT IN ({})",
75 + placeholders.join(", ")
76 + );
77 + let mut query = sqlx::query(&sql).bind(target);
78 + for route in expected_routes {
79 + query = query.bind(route);
80 + }
81 + let result = query.execute(pool).await?;
82 + Ok(result.rows_affected())
83 + }
A pom/src/db/test_runs.rs +218
@@ -0,0 +1,218 @@
1 + use super::*;
2 + use tracing::instrument;
3 +
4 + #[instrument(skip_all)]
5 + pub async fn insert_test_run(
6 + pool: &SqlitePool,
7 + run: &TestRun,
8 + ) -> Result<TestRunId> {
9 + let summary_json = serde_json::to_string(&run.summary).unwrap_or_default();
10 +
11 + let result = sqlx::query(
12 + "INSERT INTO test_runs (target, started_at, finished_at, duration_secs, exit_code, passed, summary_json, raw_output, filter)
13 + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
14 + )
15 + .bind(&run.target)
16 + .bind(&run.started_at)
17 + .bind(&run.finished_at)
18 + .bind(run.duration_secs)
19 + .bind(run.exit_code)
20 + .bind(run.passed)
21 + .bind(&summary_json)
22 + .bind(&run.raw_output)
23 + .bind(&run.filter)
24 + .execute(pool)
25 + .await?;
26 +
27 + Ok(TestRunId(result.last_insert_rowid()))
28 + }
29 +
30 + #[instrument(skip_all)]
31 + pub async fn get_test_history(
32 + pool: &SqlitePool,
33 + target: Option<&str>,
34 + limit: i64,
35 + ) -> Result<Vec<TestRun>> {
36 + let rows = match target {
37 + Some(t) => {
38 + sqlx::query_as::<_, TestRunRow>(
39 + "SELECT id, target, started_at, finished_at, duration_secs, exit_code, passed, summary_json, raw_output, filter
40 + FROM test_runs WHERE target = ? ORDER BY id DESC LIMIT ?",
41 + )
42 + .bind(t)
43 + .bind(limit)
44 + .fetch_all(pool)
45 + .await?
46 + }
47 + None => {
48 + sqlx::query_as::<_, TestRunRow>(
49 + "SELECT id, target, started_at, finished_at, duration_secs, exit_code, passed, summary_json, raw_output, filter
50 + FROM test_runs ORDER BY id DESC LIMIT ?",
51 + )
52 + .bind(limit)
53 + .fetch_all(pool)
54 + .await?
55 + }
56 + };
57 +
58 + Ok(rows.into_iter().map(|r| r.into_test_run()).collect())
59 + }
60 +
61 + #[instrument(skip_all)]
62 + pub async fn get_latest_test_run(
63 + pool: &SqlitePool,
64 + target: &str,
65 + ) -> Result<Option<TestRun>> {
66 + let row = sqlx::query_as::<_, TestRunRow>(
67 + "SELECT id, target, started_at, finished_at, duration_secs, exit_code, passed, summary_json, raw_output, filter
68 + FROM test_runs WHERE target = ? ORDER BY id DESC LIMIT 1",
69 + )
70 + .bind(target)
71 + .fetch_optional(pool)
72 + .await?;
73 +
74 + Ok(row.map(|r| r.into_test_run()))
75 + }
76 +
77 + /// Insert per-test results for a given test run.
78 + #[instrument(skip_all)]
79 + pub async fn insert_test_details(
80 + pool: &SqlitePool,
81 + run_id: TestRunId,
82 + details: &[TestDetail],
83 + ) -> Result<()> {
84 + // One transaction for the whole detail set: a crash mid-loop otherwise left a
85 + // partial set, which skews get_test_regressions (fuzz-2026-07-06 crash
86 + // atomicity). All-or-nothing.
87 + let mut tx = pool.begin().await?;
88 + for detail in details {
89 + sqlx::query(
90 + "INSERT INTO test_details (run_id, test_name, passed) VALUES (?, ?, ?)",
91 + )
92 + .bind(run_id.0)
93 + .bind(&detail.test_name)
94 + .bind(detail.passed)
95 + .execute(&mut *tx)
96 + .await?;
97 + }
98 + tx.commit().await?;
99 + Ok(())
100 + }
101 +
102 + /// Find tests that passed in the previous run but failed in this one (regressions).
103 + #[instrument(skip_all)]
104 + pub async fn get_test_regressions(
105 + pool: &SqlitePool,
106 + target: &str,
107 + current_run_id: TestRunId,
108 + ) -> Result<Vec<String>> {
109 + // Find the run immediately before this one for the same target
110 + let prev_run = sqlx::query_as::<_, (i64,)>(
111 + "SELECT id FROM test_runs
112 + WHERE target = ? AND id < ?
113 + ORDER BY id DESC LIMIT 1",
114 + )
115 + .bind(target)
116 + .bind(current_run_id.0)
117 + .fetch_optional(pool)
118 + .await?;
119 +
120 + let Some((prev_id,)) = prev_run else {
121 + return Ok(vec![]);
122 + };
123 +
124 + // Tests that passed in prev run but failed in current run
125 + let rows = sqlx::query_as::<_, (String,)>(
126 + "SELECT curr.test_name FROM test_details curr
127 + INNER JOIN test_details prev ON prev.test_name = curr.test_name AND prev.run_id = ?
128 + WHERE curr.run_id = ? AND curr.passed = 0 AND prev.passed = 1",
129 + )
130 + .bind(prev_id)
131 + .bind(current_run_id.0)
132 + .fetch_all(pool)
133 + .await?;
134 +
135 + Ok(rows.into_iter().map(|r| r.0).collect())
136 + }
137 +
138 + /// Get test duration history for a target (most recent first).
139 + #[instrument(skip_all)]
140 + pub async fn get_test_durations(
141 + pool: &SqlitePool,
142 + target: &str,
143 + limit: i64,
144 + ) -> Result<Vec<(String, i64)>> {
145 + let rows = sqlx::query_as::<_, (String, i64)>(
146 + "SELECT started_at, duration_secs FROM test_runs
147 + WHERE target = ? AND duration_secs IS NOT NULL
148 + ORDER BY id DESC LIMIT ?",
149 + )
150 + .bind(target)
151 + .bind(limit)
152 + .fetch_all(pool)
153 + .await?;
154 + Ok(rows)
155 + }
156 +
157 + /// Get the version from the health check closest to (but before) a given timestamp.
158 + #[instrument(skip_all)]
159 + pub async fn get_version_at_time(
160 + pool: &SqlitePool,
161 + target: &str,
162 + before_rfc3339: &str,
163 + ) -> Result<Option<String>> {
164 + let row = sqlx::query_as::<_, (Option<String>,)>(
165 + "SELECT details_json FROM health_checks
166 + WHERE target = ? AND checked_at <= ?
167 + ORDER BY checked_at DESC LIMIT 1",
168 + )
169 + .bind(target)
170 + .bind(before_rfc3339)
171 + .fetch_optional(pool)
172 + .await?;
173 +
174 + let version = row
175 + .and_then(|r| r.0)
176 + .and_then(|json_str| serde_json::from_str::<serde_json::Value>(&json_str).ok())
177 + .and_then(|json| json.get("version").and_then(|v| v.as_str()).map(String::from));
178 +
179 + Ok(version)
180 + }
181 +
182 + #[derive(sqlx::FromRow)]
183 + struct TestRunRow {
184 + id: i64,
185 + target: String,
186 + started_at: String,
187 + finished_at: Option<String>,
188 + duration_secs: Option<i64>,
189 + exit_code: Option<i32>,
190 + passed: bool,
191 + summary_json: String,
192 + raw_output: String,
193 + filter: Option<String>,
194 + }
195 +
196 + impl TestRunRow {
197 + fn into_test_run(self) -> TestRun {
198 + let summary = serde_json::from_str::<TestSummary>(&self.summary_json).unwrap_or(TestSummary {
199 + steps: vec![],
200 + total_passed: None,
201 + total_failed: None,
202 + details: vec![],
203 + });
204 +
205 + TestRun {
206 + id: Some(self.id),
207 + target: self.target,
208 + started_at: self.started_at,
209 + finished_at: self.finished_at,
210 + duration_secs: self.duration_secs,
211 + exit_code: self.exit_code,
212 + passed: self.passed,
213 + summary,
214 + raw_output: self.raw_output,
215 + filter: self.filter,
216 + }
217 + }
218 + }
A pom/src/db/tls.rs +56
@@ -0,0 +1,56 @@
1 + use super::*;
2 + use tracing::instrument;
3 +
4 + #[derive(Debug, sqlx::FromRow, serde::Serialize)]
5 + pub struct TlsCheckRow {
6 + pub id: i64,
7 + pub target: String,
8 + pub host: String,
9 + pub valid: bool,
10 + pub days_remaining: i64,
11 + pub not_before: String,
12 + pub not_after: String,
13 + pub subject: String,
14 + pub issuer: String,
15 + pub checked_at: String,
16 + pub error: Option<String>,
17 + }
18 +
19 + #[instrument(skip_all)]
20 + pub async fn insert_tls_check(
21 + pool: &SqlitePool,
22 + status: &TlsStatus,
23 + ) -> Result<i64> {
24 + let result = sqlx::query(
25 + "INSERT INTO tls_checks (target, host, valid, days_remaining, not_before, not_after, subject, issuer, checked_at, error)
26 + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
27 + )
28 + .bind(&status.target)
29 + .bind(&status.host)
30 + .bind(status.valid)
31 + .bind(status.days_remaining)
32 + .bind(&status.not_before)
33 + .bind(&status.not_after)
34 + .bind(&status.subject)
35 + .bind(&status.issuer)
36 + .bind(&status.checked_at)
37 + .bind(&status.error)
38 + .execute(pool)
39 + .await?;
40 +
41 + Ok(result.last_insert_rowid())
42 + }
43 +
44 + #[instrument(skip_all)]
45 + pub async fn get_latest_tls_check(
46 + pool: &SqlitePool,
47 + target: &str,
48 + ) -> Result<Option<TlsCheckRow>> {
49 + Ok(sqlx::query_as::<_, TlsCheckRow>(
50 + "SELECT id, target, host, valid, days_remaining, not_before, not_after, subject, issuer, checked_at, error
51 + FROM tls_checks WHERE target = ? ORDER BY id DESC LIMIT 1",
52 + )
53 + .bind(target)
54 + .fetch_optional(pool)
55 + .await?)
56 + }
A pom/src/db/whois.rs +53
@@ -0,0 +1,53 @@
1 + use super::*;
2 + use tracing::instrument;
3 +
4 + #[derive(Debug, Clone, sqlx::FromRow, serde::Serialize)]
5 + pub struct WhoisCheckRow {
6 + pub id: i64,
7 + pub target: String,
8 + pub domain: String,
9 + pub registrar: Option<String>,
10 + pub expiry_date: Option<String>,
11 + pub days_remaining: Option<i64>,
12 + pub nameservers: Option<String>,
13 + pub checked_at: String,
14 + pub error: Option<String>,
15 + }
16 +
17 + #[instrument(skip_all)]
18 + pub async fn insert_whois_check(
19 + pool: &SqlitePool,
20 + result: &WhoisResult,
21 + ) -> Result<i64> {
22 + let nameservers = serde_json::to_string(&result.nameservers).unwrap_or_default();
23 +
24 + let row = sqlx::query(
25 + "INSERT INTO whois_checks (target, domain, registrar, expiry_date, days_remaining, nameservers, checked_at, error)
26 + VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
27 + )
28 + .bind(&result.target)
29 + .bind(&result.domain)
30 + .bind(&result.registrar)
31 + .bind(&result.expiry_date)
32 + .bind(result.days_remaining)
33 + .bind(&nameservers)
34 + .bind(&result.checked_at)
35 + .bind(&result.error)
36 + .execute(pool)
37 + .await?;
38 + Ok(row.last_insert_rowid())
39 + }
40 +
41 + #[instrument(skip_all)]
42 + pub async fn get_latest_whois_check(
43 + pool: &SqlitePool,
44 + target: &str,
45 + ) -> Result<Option<WhoisCheckRow>> {
46 + Ok(sqlx::query_as::<_, WhoisCheckRow>(
47 + "SELECT id, target, domain, registrar, expiry_date, days_remaining, nameservers, checked_at, error
48 + FROM whois_checks WHERE target = ? ORDER BY id DESC LIMIT 1",
49 + )
50 + .bind(target)
51 + .fetch_optional(pool)
52 + .await?)
53 + }