Single-binary self-hosted uptime monitoring and status pages on Rust axum: HTTP probes, Lighthouse audits, SEO crawler, and PDF reports.
axumdockerrustself-hostedsqlitestatus-pageuptime-monitoringvite
1use crate::alerts;
2use crate::db::now_ms;
3use crate::models::PropertyRow;
4use anyhow::{anyhow, Context};
5use chrono::Timelike;
6use hickory_resolver::TokioAsyncResolver;
7use rustls::pki_types::ServerName;
8use serde_json::json;
9use sqlx::SqlitePool;
10use std::collections::BTreeMap;
11use std::net::SocketAddr;
12use std::sync::Arc;
13use std::time::{Duration, Instant};
14use tokio::net::TcpStream;
15use tokio_rustls::TlsConnector;
16use url::Url;
17use uuid::Uuid;
18
19const USER_AGENT: &str = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 \
20 (KHTML, like Gecko) Chrome/102.0.5005.115 Safari/537.36 Status/2.0.0";
21const HTTP_TIMEOUT_SECS: u64 = 10;
22const MAX_REDIRECTS: usize = 5;
23
24/// Phase-by-phase timings for a single HTTP probe. `None` means the phase
25/// didn't run (the probe errored before reaching it). `total_ms` is
26/// wall-clock first-hop end-to-end and is also written to
27/// `checks.response_ms` for backward compat with the alert email avg.
28/// Fields are `Option<i64>` so that pre-rewrite rows (migration 0002 left
29/// them NULL) keep deserializing.
30#[derive(Debug, Default, Clone, Copy)]
31pub struct PhaseTimings {
32 pub dns_ms: Option<i64>,
33 pub tcp_ms: Option<i64>,
34 pub tls_ms: Option<i64>,
35 pub ttfb_ms: Option<i64>,
36 pub total_ms: i64,
37}
38
39struct ProbeOutcome {
40 status_code: i64,
41 headers_json: String,
42 timings: PhaseTimings,
43}
44
45/// One-hop result with everything we need to decide whether to follow a
46/// redirect.
47struct HopResult {
48 status_code: i64,
49 headers: BTreeMap<String, String>,
50 raw_headers_json: String,
51 timings: PhaseTimings,
52}
53
54/// Build a fresh rustls config per probe. Keeps the "fresh client per
55/// probe = real handshake cost" invariant. ALPN-pinned to `h2` only:
56/// servers that don't speak HTTP/2 will fail the handshake (mapped to
57/// 526), which matches the project's "no HTTP/1.1" stance.
58fn tls_config_h2() -> Arc<rustls::ClientConfig> {
59 let mut roots = rustls::RootCertStore::empty();
60 roots.extend(webpki_roots::TLS_SERVER_ROOTS.iter().cloned());
61 let mut cfg = rustls::ClientConfig::builder()
62 .with_root_certificates(roots)
63 .with_no_client_auth();
64 cfg.alpn_protocols = vec![b"h2".to_vec()];
65 Arc::new(cfg)
66}
67
68/// Round a duration to whole milliseconds, but report a sub-ms phase as
69/// 1 rather than 0. The Linux kernel routes traffic destined for the
70/// host's own public IP via `lo`, so loopback TCP/handshake phases
71/// genuinely take 200-500 microseconds, which `as_millis()` truncates to
72/// zero. A `0` in the chart reads as "this phase didn't happen" rather
73/// than "this phase was instant", so floor it at 1 ms when the phase did
74/// run. Total is unaffected: any real probe takes well over 1 ms.
75fn elapsed_ms_atleast1(d: std::time::Duration) -> i64 {
76 let ms = d.as_millis() as i64;
77 if ms > 0 || d.is_zero() {
78 ms
79 } else {
80 1
81 }
82}
83
84fn looks_like_ssl_error(e: &anyhow::Error) -> bool {
85 let s = format!("{e:?}").to_lowercase();
86 s.contains("certificate")
87 || s.contains("invalidcertificate")
88 || s.contains("tls")
89 || s.contains("handshake")
90}
91
92/// Run a single HTTP check and persist the result. Maps SSL errors to 526
93/// (Cloudflare convention) and timeouts to 408 so the dashboard can show
94/// failure reasons without piping arbitrary error messages.
95pub async fn run_check(pool: &SqlitePool, prop: &PropertyRow) -> sqlx::Result<i64> {
96 let started = Instant::now();
97 let outcome = match probe_with_redirects(&prop.url).await {
98 Ok(o) => o,
99 Err(e) => {
100 let code = if looks_like_ssl_error(&e) { 526 } else { 408 };
101 // Record how long the failure actually took: a DNS NXDOMAIN or
102 // refused connect fails in milliseconds, and charting it as the
103 // full 10s timeout would poison the response-time averages.
104 ProbeOutcome {
105 status_code: code,
106 headers_json: "{}".to_string(),
107 timings: PhaseTimings {
108 total_ms: started.elapsed().as_millis() as i64,
109 ..PhaseTimings::default()
110 },
111 }
112 }
113 };
114
115 let id = prop.id.clone();
116 sqlx::query(
117 "INSERT INTO checks (property_id, status_code, response_ms, headers, dns_ms, tcp_ms, tls_ms, ttfb_ms, created_at) \
118 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
119 )
120 .bind(&id)
121 .bind(outcome.status_code)
122 .bind(outcome.timings.total_ms)
123 .bind(&outcome.headers_json)
124 .bind(outcome.timings.dns_ms)
125 .bind(outcome.timings.tcp_ms)
126 .bind(outcome.timings.tls_ms)
127 .bind(outcome.timings.ttfb_ms)
128 .bind(now_ms())
129 .execute(pool)
130 .await?;
131
132 Ok(outcome.status_code)
133}
134
135/// First-hop with full phase timings, then follow up to MAX_REDIRECTS
136/// 3xx hops to discover the final status code (so the alert state machine
137/// keeps working when a property uses an http→https or apex→www
138/// redirect). Phase timings always reflect the first hop only — that's
139/// the latency a fresh visitor pays before being redirected, and it's the
140/// only number that's meaningful when later hops live on different
141/// servers/domains.
142async fn probe_with_redirects(url_str: &str) -> anyhow::Result<ProbeOutcome> {
143 let url = Url::parse(url_str).context("invalid URL")?;
144
145 let outer = tokio::time::timeout(
146 Duration::from_secs(HTTP_TIMEOUT_SECS),
147 async {
148 let mut current = url.clone();
149 let first = phased_hop(¤t).await?;
150 let first_timings = first.timings;
151 let mut status = first.status_code;
152 let mut headers_json = first.raw_headers_json;
153 let mut headers = first.headers;
154 let mut hops = 0usize;
155
156 while is_redirect(status) && hops < MAX_REDIRECTS {
157 let Some(loc) = headers.get("location").cloned() else { break };
158 let Ok(next) = current.join(&loc) else { break };
159 current = next;
160 let hop = match phased_hop(¤t).await {
161 Ok(h) => h,
162 Err(_) => break,
163 };
164 status = hop.status_code;
165 headers_json = hop.raw_headers_json;
166 headers = hop.headers;
167 hops += 1;
168 }
169
170 Ok::<_, anyhow::Error>(ProbeOutcome {
171 status_code: status,
172 headers_json,
173 timings: first_timings,
174 })
175 },
176 )
177 .await;
178
179 match outer {
180 Ok(Ok(o)) => Ok(o),
181 Ok(Err(e)) => Err(e),
182 Err(_) => Err(anyhow!("timeout after {HTTP_TIMEOUT_SECS}s")),
183 }
184}
185
186fn is_redirect(code: i64) -> bool {
187 matches!(code, 301 | 302 | 303 | 307 | 308)
188}
189
190async fn phased_hop(url: &Url) -> anyhow::Result<HopResult> {
191 let host = url.host_str().context("URL missing host")?.to_string();
192 let port = url.port_or_known_default().context("URL missing port")?;
193 let path_q = match url.query() {
194 Some(q) if !q.is_empty() => format!("{}?{}", url.path(), q),
195 _ => url.path().to_string(),
196 };
197 let path_q = if path_q.is_empty() { "/".to_string() } else { path_q };
198 let is_https = url.scheme() == "https";
199
200 let total_start = Instant::now();
201
202 let dns_start = Instant::now();
203 let resolver =
204 TokioAsyncResolver::tokio_from_system_conf().context("creating dns resolver")?;
205 let lookup = resolver.lookup_ip(host.as_str()).await.context("dns lookup")?;
206 let ip = lookup
207 .iter()
208 .next()
209 .ok_or_else(|| anyhow!("no addresses for {host}"))?;
210 let dns_ms = elapsed_ms_atleast1(dns_start.elapsed());
211 let addr = SocketAddr::new(ip, port);
212
213 let tcp_start = Instant::now();
214 let tcp = TcpStream::connect(addr).await.context("tcp connect")?;
215 tcp.set_nodelay(true).ok();
216 let tcp_ms = elapsed_ms_atleast1(tcp_start.elapsed());
217
218 if !is_https {
219 // h2 over plain TCP (h2c with prior knowledge) is rare in the
220 // wild, and the project is HTTP/2-only, so reject http:// URLs
221 // explicitly rather than silently downgrading.
222 return Err(anyhow!("plain HTTP not supported; use https:// (HTTP/2 only)"));
223 }
224
225 let tls_start = Instant::now();
226 let server_name =
227 ServerName::try_from(host.clone()).context("invalid TLS server name")?;
228 let connector = TlsConnector::from(tls_config_h2());
229 let tls_stream = connector
230 .connect(server_name, tcp)
231 .await
232 .context("tls handshake")?;
233 let tls_ms = elapsed_ms_atleast1(tls_start.elapsed());
234
235 let (status, headers, raw_headers_json, ttfb_ms) =
236 h2_request(tls_stream, &host, &path_q).await?;
237
238 let total_ms = total_start.elapsed().as_millis() as i64;
239
240 Ok(HopResult {
241 status_code: status,
242 headers,
243 raw_headers_json,
244 timings: PhaseTimings {
245 dns_ms: Some(dns_ms),
246 tcp_ms: Some(tcp_ms),
247 tls_ms: Some(tls_ms),
248 ttfb_ms: Some(ttfb_ms),
249 total_ms,
250 },
251 })
252}
253
254/// Run an HTTP/2 GET over an established TLS stream and return
255/// (status_code, headers, headers_json, ttfb_ms). TTFB is measured from
256/// the start of the h2 client handshake (SETTINGS exchange) to the
257/// arrival of the response HEADERS frame, so it includes h2 protocol
258/// setup; the user-facing chart treats it as "everything between secure
259/// connection ready and first server byte", which matches curl's
260/// `time_starttransfer` minus `time_appconnect`.
261async fn h2_request(
262 tls_stream: tokio_rustls::client::TlsStream<TcpStream>,
263 host: &str,
264 path: &str,
265) -> anyhow::Result<(i64, BTreeMap<String, String>, String, i64)> {
266 let ttfb_start = Instant::now();
267 let (sr, connection) = h2::client::handshake(tls_stream)
268 .await
269 .context("h2 handshake")?;
270 // h2 needs someone to drive the connection's I/O loop. Spawn a task
271 // that lives just as long as this probe; we abort it on the way out.
272 let conn_task = tokio::spawn(async move {
273 let _ = connection.await;
274 });
275
276 let mut sr = sr.ready().await.context("h2 send-request ready")?;
277 let req = http::Request::builder()
278 .method("GET")
279 .uri(format!("https://{host}{path}"))
280 .header("user-agent", USER_AGENT)
281 .header("accept", "*/*")
282 .body(())
283 .context("h2 request build")?;
284 let (rsp_fut, _send_stream) = sr.send_request(req, true).context("h2 send_request")?;
285 let rsp = rsp_fut.await.context("h2 response")?;
286 let ttfb_ms = elapsed_ms_atleast1(ttfb_start.elapsed());
287
288 let status = rsp.status().as_u16() as i64;
289 let mut headers = BTreeMap::new();
290 for (k, v) in rsp.headers().iter() {
291 if let Ok(s) = v.to_str() {
292 headers.insert(k.as_str().to_lowercase(), s.to_string());
293 }
294 }
295 let raw_headers_json = serde_json::Value::Object(
296 headers.iter().map(|(k, v)| (k.clone(), json!(v))).collect(),
297 )
298 .to_string();
299
300 drop(sr);
301 conn_task.abort();
302 Ok((status, headers, raw_headers_json, ttfb_ms))
303}
304
305/// Run a check, persist it, then advance the alert state machine and fire
306/// notifications on transitions.
307pub async fn process_check(
308 pool: &SqlitePool,
309 config: &crate::Config,
310 prop: &PropertyRow,
311) -> anyhow::Result<()> {
312 let status_code = run_check(pool, prop).await?;
313 advance_alert_state(pool, config, prop, status_code).await?;
314 Ok(())
315}
316
317/// State transitions:
318/// UP -> DOWN: requires 2 consecutive non-200 checks (avoids false positives)
319/// DOWN -> UP: immediate on 200
320/// Commits state inside a transaction *before* firing notifications so a
321/// crash mid-alert can't cause duplicate sends.
322async fn advance_alert_state(
323 pool: &SqlitePool,
324 config: &crate::Config,
325 prop: &PropertyRow,
326 status_code: i64,
327) -> anyhow::Result<()> {
328 let is_up = status_code == 200;
329 let mut tx = pool.begin().await?;
330
331 let row: Option<(String,)> =
332 sqlx::query_as("SELECT alert_state FROM properties WHERE id = ?")
333 .bind(prop.id.clone())
334 .fetch_optional(&mut *tx)
335 .await?;
336 let Some((current_state,)) = row else {
337 return Ok(());
338 };
339
340 let mut transition: Option<&str> = None;
341 if is_up && current_state == "down" {
342 transition = Some("recovery");
343 } else if !is_up && current_state == "up" {
344 // Need 2 consecutive non-200s. We just inserted one; check the prior.
345 let recent: Vec<(i64,)> = sqlx::query_as(
346 "SELECT status_code FROM checks WHERE property_id = ? ORDER BY created_at DESC LIMIT 2",
347 )
348 .bind(prop.id.clone())
349 .fetch_all(&mut *tx)
350 .await?;
351 if recent.len() >= 2 && recent[0].0 != 200 && recent[1].0 != 200 {
352 transition = Some("down");
353 }
354 }
355
356 if let Some(kind) = transition {
357 let new_state = if kind == "recovery" { "up" } else { "down" };
358 sqlx::query(
359 "UPDATE properties SET alert_state = ?, last_alert_sent = ?, updated_at = ? WHERE id = ?",
360 )
361 .bind(new_state)
362 .bind(now_ms())
363 .bind(now_ms())
364 .bind(prop.id.clone())
365 .execute(&mut *tx)
366 .await?;
367 }
368 tx.commit().await?;
369
370 if let Some(kind) = transition {
371 let avg_response_time = recent_avg_response_ms(pool, &prop.id).await.unwrap_or(0);
372 let ctx = alerts::EmailContext {
373 id: prop.uuid(),
374 name: prop.name(),
375 url: prop.url.clone(),
376 current_status: status_code,
377 avg_response_time,
378 };
379 let alert_email = config.alert_email.clone();
380 let webhook = config.discord_webhook_url.clone();
381 let base_url = config.base_url.clone();
382 let url_for_log = prop.url.clone();
383 // Fire-and-forget: alerts don't block the scheduler tick.
384 tokio::spawn(async move {
385 if let Err(e) =
386 alerts::fire(kind, &ctx, &base_url, alert_email.as_deref(), webhook.as_deref()).await
387 {
388 tracing::warn!("alert dispatch failed for {url_for_log}: {e}");
389 }
390 });
391 }
392 Ok(())
393}
394
395/// Average response time across the most recent 31 checks. Mirrors the
396/// dashboard's "rolling avg" tile so the email matches what the user sees.
397async fn recent_avg_response_ms(pool: &SqlitePool, id: &[u8]) -> sqlx::Result<i64> {
398 let rows: Vec<(i64,)> = sqlx::query_as(
399 "SELECT response_ms FROM checks WHERE property_id = ? ORDER BY created_at DESC LIMIT 31",
400 )
401 .bind(id.to_vec())
402 .fetch_all(pool)
403 .await?;
404 if rows.is_empty() {
405 return Ok(0);
406 }
407 let sum: i64 = rows.iter().map(|r| r.0).sum();
408 Ok(sum / rows.len() as i64)
409}
410
411/// Compute the next due time aligned to a 3-minute boundary, matching Django.
412pub fn next_3min_boundary() -> i64 {
413 let now = chrono::Utc::now();
414 let minute = now.minute() as i64;
415 let aligned_min = (minute / 3) * 3;
416 let aligned = now
417 .with_minute(aligned_min as u32)
418 .and_then(|d| d.with_second(0))
419 .and_then(|d| d.with_nanosecond(0))
420 .unwrap_or(now);
421 (aligned + chrono::Duration::minutes(3)).timestamp_millis()
422}
423
424#[allow(dead_code)]
425pub async fn property_id_to_uuid(id_blob: &[u8]) -> Uuid {
426 Uuid::from_slice(id_blob).unwrap_or(Uuid::nil())
427}