Skip to main content

max / makenotwork

6.3 KB · 213 lines History Blame Raw
1 //! Health check storage and the queries built on it: history, latest snapshot,
2 //! uptime percentage, and the response-time series used for drift detection.
3
4 use super::{HealthDetails, HealthSnapshot, HealthStatus, Result, SqlitePool};
5 use tracing::instrument;
6
7 #[instrument(skip_all)]
8 pub async fn insert_health_check(pool: &SqlitePool, snapshot: &HealthSnapshot) -> 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
60 .into_iter()
61 .map(HealthCheckRow::into_snapshot)
62 .collect())
63 }
64
65 #[instrument(skip_all)]
66 pub async fn get_latest_health(pool: &SqlitePool, target: &str) -> 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(HealthCheckRow::into_snapshot))
76 }
77
78 /// When the target's *current* version was first seen, i.e. the earliest check
79 /// in the unbroken run of checks reporting it.
80 ///
81 /// The closest PoM can honestly get to "when was this deployed": it observes
82 /// versions, never deployments. Two consequences worth knowing at the call
83 /// site. A failed check reports no version at all, and is skipped rather than
84 /// treated as a version change, so an outage does not read as a redeploy. And
85 /// the answer is bounded by retention: a version older than the prune window
86 /// looks like it was first seen at the oldest row still held.
87 #[instrument(skip_all)]
88 pub async fn get_version_first_seen(
89 pool: &SqlitePool,
90 target: &str,
91 version: &str,
92 ) -> Result<Option<String>> {
93 let row = sqlx::query_as::<_, (Option<String>,)>(
94 "SELECT MIN(checked_at) FROM health_checks
95 WHERE target = ?1
96 AND json_extract(details_json, '$.version') = ?2
97 AND id > COALESCE(
98 (SELECT MAX(id) FROM health_checks
99 WHERE target = ?1
100 AND json_extract(details_json, '$.version') IS NOT NULL
101 AND json_extract(details_json, '$.version') != ?2),
102 0)",
103 )
104 .bind(target)
105 .bind(version)
106 .fetch_optional(pool)
107 .await?;
108
109 Ok(row.and_then(|r| r.0))
110 }
111
112 /// Calculate uptime percentage for a target over the given number of hours.
113 /// Returns the percentage of health checks with "operational" status.
114 #[instrument(skip_all)]
115 pub async fn get_uptime_percent(
116 pool: &SqlitePool,
117 target: &str,
118 hours: i64,
119 ) -> Result<Option<f64>> {
120 let cutoff = chrono::Utc::now() - chrono::Duration::hours(hours);
121 let cutoff_str = cutoff.to_rfc3339();
122
123 let row = sqlx::query_as::<_, (i64, i64)>(
124 "SELECT
125 COUNT(*) as total,
126 SUM(CASE WHEN status = 'operational' THEN 1 ELSE 0 END) as operational
127 FROM health_checks
128 WHERE target = ? AND checked_at >= ?",
129 )
130 .bind(target)
131 .bind(&cutoff_str)
132 .fetch_one(pool)
133 .await?;
134
135 if row.0 == 0 {
136 Ok(None)
137 } else {
138 Ok(Some(row.1 as f64 / row.0 as f64 * 100.0))
139 }
140 }
141
142 /// Fetch all response times for a target since a given timestamp, ordered ASC.
143 #[instrument(skip_all)]
144 pub async fn get_response_times(
145 pool: &SqlitePool,
146 target: &str,
147 since_rfc3339: &str,
148 ) -> Result<Vec<(String, i64)>> {
149 let rows = sqlx::query_as::<_, (String, i64)>(
150 "SELECT checked_at, response_time_ms FROM health_checks
151 WHERE target = ? AND checked_at >= ?
152 ORDER BY checked_at ASC",
153 )
154 .bind(target)
155 .bind(since_rfc3339)
156 .fetch_all(pool)
157 .await?;
158 Ok(rows)
159 }
160
161 /// Fetch the last N response times for **operational** checks only (most recent first).
162 #[instrument(skip_all)]
163 pub async fn get_recent_response_times(
164 pool: &SqlitePool,
165 target: &str,
166 count: i64,
167 ) -> Result<Vec<i64>> {
168 let rows = sqlx::query_as::<_, (i64,)>(
169 "SELECT response_time_ms FROM health_checks
170 WHERE target = ? AND status = 'operational'
171 ORDER BY id DESC LIMIT ?",
172 )
173 .bind(target)
174 .bind(count)
175 .fetch_all(pool)
176 .await?;
177 Ok(rows.into_iter().map(|r| r.0).collect())
178 }
179
180 #[derive(sqlx::FromRow)]
181 struct HealthCheckRow {
182 id: i64,
183 target: String,
184 status: String,
185 checked_at: String,
186 response_time_ms: i64,
187 details_json: Option<String>,
188 error: Option<String>,
189 }
190
191 impl HealthCheckRow {
192 fn into_snapshot(self) -> HealthSnapshot {
193 let status = self
194 .status
195 .parse::<HealthStatus>()
196 .unwrap_or(HealthStatus::Error);
197 let details = self
198 .details_json
199 .as_deref()
200 .and_then(|s| serde_json::from_str::<HealthDetails>(s).ok());
201
202 HealthSnapshot {
203 id: Some(self.id),
204 target: self.target,
205 status,
206 checked_at: self.checked_at,
207 response_time_ms: self.response_time_ms,
208 details,
209 error: self.error,
210 }
211 }
212 }
213