Initial Meterbox commit with boot disk usage metric.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
JamesandCursor committed 2026-09-19 02:45:52 +00:00
commit 4e44204d9b
55 files changed
+12104

No files matched your search

+789
View File
@@ -0,0 +1,789 @@
use axum::extract::{Path, Query, State};
use axum::http::{HeaderMap, StatusCode};
use axum::routing::{get, post};
use axum::{Json, Router};
use serde::{Deserialize, Serialize};
use crate::auth::UserPublic;
use crate::health::{HealthSite, SiteHistory, UpsertSite};
use crate::metrics::{HostInfo, Sample, Sampler};
use crate::probe::{probe_url, ProbeResult};
use crate::settings::{AppSettings, PublicAccess};
use crate::AppState;
const SESSION_TTL_SECS: i64 = 30 * 24 * 60 * 60;
pub fn router(state: AppState) -> Router {
Router::new()
.route("/health", get(health))
.route("/host", get(host))
.route("/snapshot", get(snapshot))
.route("/series", get(series))
.route("/compliance", get(compliance))
.route("/bwflow", get(bwflow_stats))
.route("/auth/status", get(auth_status))
.route("/auth/setup", post(auth_setup))
.route("/auth/login", post(auth_login))
.route("/auth/logout", post(auth_logout))
.route("/auth/me", get(auth_me))
.route("/sites", get(list_sites).post(create_site))
.route("/sites/probe", post(probe_site))
.route("/sites/{id}", get(get_site).put(update_site).delete(delete_site))
.route("/sites/{id}/check", post(check_site_now))
.route("/sites/{id}/history", get(site_history))
.route("/settings", get(get_settings).put(put_settings))
.with_state(state)
}
#[derive(Serialize)]
struct Health {
ok: bool,
service: &'static str,
samples: i64,
oldest_ts: Option<i64>,
db_bytes: u64,
max_db_bytes: u64,
retention_days: i64,
cleanup_interval_secs: u64,
}
async fn health(State(state): State<AppState>) -> Json<Health> {
Json(Health {
ok: true,
service: "meterbox",
samples: state.db.count().unwrap_or(0),
oldest_ts: state.db.oldest_ts().ok().flatten(),
db_bytes: state.db.on_disk_bytes().unwrap_or(0),
max_db_bytes: state.config.max_db_bytes,
retention_days: state.config.retention_days,
cleanup_interval_secs: state.config.cleanup_interval_secs,
})
}
async fn host(
State(state): State<AppState>,
headers: HeaderMap,
) -> Result<Json<HostInfo>, StatusCode> {
require_dashboard_read(&state, &headers)?;
let mut sampler = Sampler::new();
Ok(Json(sampler.host_info()))
}
async fn snapshot(
State(state): State<AppState>,
headers: HeaderMap,
) -> Result<Json<Option<Sample>>, StatusCode> {
require_dashboard_read(&state, &headers)?;
Ok(Json(state.db.latest().ok().flatten()))
}
async fn compliance(
State(state): State<AppState>,
headers: HeaderMap,
) -> Result<Json<crate::compliance::ComplianceReport>, StatusCode> {
require_dashboard_read(&state, &headers)?;
let mut sampler = Sampler::new();
let host = sampler.host_info();
let bw = crate::bwflow::fetch(&crate::bwflow::socket_from_env()).await;
let bw_tx = if bw.available { bw.tx_bytes } else { None };
Ok(Json(crate::compliance::evaluate(
&state.db,
&host,
state.config.interval_secs,
bw_tx,
)))
}
async fn bwflow_stats(
State(state): State<AppState>,
headers: HeaderMap,
) -> Result<Json<crate::bwflow::BwflowStats>, StatusCode> {
require_dashboard_read(&state, &headers)?;
let sock = crate::bwflow::socket_from_env();
Ok(Json(crate::bwflow::fetch(&sock).await))
}
#[derive(Debug, Deserialize)]
struct SeriesQuery {
/// 24h | 7d | 4w | 3m
range: String,
}
#[derive(Serialize)]
struct SeriesPoint {
ts: i64,
cpu_pct: f32,
mem_pct: f32,
mem_used_bytes: u64,
mem_total_bytes: u64,
disk_pct: f32,
disk_used_bytes: u64,
disk_total_bytes: u64,
net_rx_bps: f64,
net_tx_bps: f64,
load1: f64,
}
#[derive(Serialize)]
struct SeriesResponse {
range: String,
from_ts: i64,
to_ts: i64,
bucket_secs: i64,
points: Vec<SeriesPoint>,
averages: Averages,
}
#[derive(Serialize)]
struct Averages {
cpu_pct: f32,
mem_pct: f32,
disk_pct: f32,
net_rx_bps: f64,
net_tx_bps: f64,
load1: f64,
}
fn range_window(range: &str) -> Option<(i64, i64)> {
let secs = match range {
"24h" => 24 * 3600,
"7d" => 7 * 24 * 3600,
"4w" => 28 * 24 * 3600,
"3m" => 90 * 24 * 3600,
_ => return None,
};
let bucket = match range {
"24h" => 60, // 1 min
"7d" => 5 * 60, // 5 min
"4w" => 30 * 60, // 30 min
"3m" => 2 * 3600, // 2 hours
_ => 60,
};
Some((secs, bucket))
}
fn downsample(samples: &[Sample], bucket_secs: i64) -> Vec<SeriesPoint> {
if samples.is_empty() {
return Vec::new();
}
let mut out = Vec::new();
let mut bucket_start = samples[0].ts - (samples[0].ts % bucket_secs);
let mut cpu = 0.0f64;
let mut mem = 0.0f64;
let mut mem_used = 0u64;
let mut mem_total = 0u64;
let mut disk = 0.0f64;
let mut disk_used = 0u64;
let mut disk_total = 0u64;
let mut rx = 0.0f64;
let mut tx = 0.0f64;
let mut load = 0.0f64;
let mut n = 0u64;
let flush = |out: &mut Vec<SeriesPoint>,
start: i64,
cpu: f64,
mem: f64,
mem_used: u64,
mem_total: u64,
disk: f64,
disk_used: u64,
disk_total: u64,
rx: f64,
tx: f64,
load: f64,
n: u64| {
if n == 0 {
return;
}
let nf = n as f64;
out.push(SeriesPoint {
ts: start,
cpu_pct: (cpu / nf) as f32,
mem_pct: (mem / nf) as f32,
mem_used_bytes: mem_used / n,
mem_total_bytes: mem_total / n,
disk_pct: (disk / nf) as f32,
disk_used_bytes: disk_used / n,
disk_total_bytes: disk_total / n,
net_rx_bps: rx / nf,
net_tx_bps: tx / nf,
load1: load / nf,
});
};
for s in samples {
let b = s.ts - (s.ts % bucket_secs);
if b != bucket_start {
flush(
&mut out,
bucket_start,
cpu,
mem,
mem_used,
mem_total,
disk,
disk_used,
disk_total,
rx,
tx,
load,
n,
);
bucket_start = b;
cpu = 0.0;
mem = 0.0;
mem_used = 0;
mem_total = 0;
disk = 0.0;
disk_used = 0;
disk_total = 0;
rx = 0.0;
tx = 0.0;
load = 0.0;
n = 0;
}
cpu += s.cpu_pct as f64;
mem += s.mem_pct as f64;
mem_used += s.mem_used_bytes;
mem_total += s.mem_total_bytes;
disk += s.disk_pct as f64;
disk_used += s.disk_used_bytes;
disk_total += s.disk_total_bytes;
rx += s.net_rx_bps;
tx += s.net_tx_bps;
load += s.load1;
n += 1;
}
flush(
&mut out,
bucket_start,
cpu,
mem,
mem_used,
mem_total,
disk,
disk_used,
disk_total,
rx,
tx,
load,
n,
);
out
}
fn compute_averages(points: &[SeriesPoint]) -> Averages {
if points.is_empty() {
return Averages {
cpu_pct: 0.0,
mem_pct: 0.0,
disk_pct: 0.0,
net_rx_bps: 0.0,
net_tx_bps: 0.0,
load1: 0.0,
};
}
let n = points.len() as f64;
Averages {
cpu_pct: (points.iter().map(|p| p.cpu_pct as f64).sum::<f64>() / n) as f32,
mem_pct: (points.iter().map(|p| p.mem_pct as f64).sum::<f64>() / n) as f32,
disk_pct: (points.iter().map(|p| p.disk_pct as f64).sum::<f64>() / n) as f32,
net_rx_bps: points.iter().map(|p| p.net_rx_bps).sum::<f64>() / n,
net_tx_bps: points.iter().map(|p| p.net_tx_bps).sum::<f64>() / n,
load1: points.iter().map(|p| p.load1).sum::<f64>() / n,
}
}
async fn series(
State(state): State<AppState>,
headers: HeaderMap,
Query(q): Query<SeriesQuery>,
) -> Result<Json<SeriesResponse>, StatusCode> {
require_dashboard_read(&state, &headers)?;
let (window, bucket) = range_window(q.range.as_str()).ok_or(StatusCode::BAD_REQUEST)?;
let to_ts = chrono::Utc::now().timestamp();
let from_ts = to_ts - window;
let samples = state
.db
.query_range(from_ts, to_ts)
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
let points = downsample(&samples, bucket);
let averages = compute_averages(&points);
Ok(Json(SeriesResponse {
range: q.range,
from_ts,
to_ts,
bucket_secs: bucket,
points,
averages,
}))
}
// --- Auth (one-time admin setup + login for interactive dashboard) ----------
#[derive(Serialize)]
struct AuthStatus {
needs_setup: bool,
user: Option<UserPublic>,
public: PublicAccess,
}
#[derive(Deserialize)]
struct Credentials {
email: String,
password: String,
}
#[derive(Serialize)]
struct AuthSession {
token: String,
user: UserPublic,
}
#[derive(Serialize)]
struct ApiError {
error: String,
}
fn bearer_token(headers: &HeaderMap) -> Option<String> {
let raw = headers.get(axum::http::header::AUTHORIZATION)?.to_str().ok()?;
raw.strip_prefix("Bearer ")
.or_else(|| raw.strip_prefix("bearer "))
.map(|s| s.trim().to_string())
.filter(|s| !s.is_empty())
}
fn optional_user(state: &AppState, headers: &HeaderMap) -> Result<Option<UserPublic>, StatusCode> {
let Some(token) = bearer_token(headers) else {
return Ok(None);
};
state
.db
.user_from_token(&token)
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)
}
fn require_user(state: &AppState, headers: &HeaderMap) -> Result<UserPublic, StatusCode> {
optional_user(state, headers)?.ok_or(StatusCode::UNAUTHORIZED)
}
fn load_settings(state: &AppState) -> Result<AppSettings, StatusCode> {
state
.db
.get_settings()
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)
}
fn require_dashboard_read(state: &AppState, headers: &HeaderMap) -> Result<(), StatusCode> {
if optional_user(state, headers)?.is_some() {
return Ok(());
}
if load_settings(state)?.public_dashboard {
return Ok(());
}
Err(StatusCode::UNAUTHORIZED)
}
fn require_site_read(state: &AppState, headers: &HeaderMap, site_id: i64) -> Result<(), StatusCode> {
if optional_user(state, headers)?.is_some() {
return Ok(());
}
if load_settings(state)?.site_is_public(site_id) {
return Ok(());
}
Err(StatusCode::UNAUTHORIZED)
}
async fn auth_status(
State(state): State<AppState>,
headers: HeaderMap,
) -> Result<Json<AuthStatus>, StatusCode> {
let needs_setup = state
.db
.user_count()
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?
== 0;
let user = optional_user(&state, &headers)?;
let public = load_settings(&state)?.public_access();
Ok(Json(AuthStatus {
needs_setup,
user,
public,
}))
}
async fn auth_setup(
State(state): State<AppState>,
Json(body): Json<Credentials>,
) -> Result<Json<AuthSession>, (StatusCode, Json<ApiError>)> {
let user = state
.db
.create_admin(&body.email, &body.password)
.map_err(|e| {
let msg = e.to_string();
if msg.contains("already exists") {
(
StatusCode::CONFLICT,
Json(ApiError {
error: "setup already completed".into(),
}),
)
} else {
bad_request(msg)
}
})?;
let token = state
.db
.create_session(user.id, SESSION_TTL_SECS)
.map_err(internal_err)?;
Ok(Json(AuthSession { token, user }))
}
async fn auth_login(
State(state): State<AppState>,
Json(body): Json<Credentials>,
) -> Result<Json<AuthSession>, (StatusCode, Json<ApiError>)> {
let user = state
.db
.authenticate(&body.email, &body.password)
.map_err(internal_err)?
.ok_or_else(|| {
(
StatusCode::UNAUTHORIZED,
Json(ApiError {
error: "invalid email or password".into(),
}),
)
})?;
let token = state
.db
.create_session(user.id, SESSION_TTL_SECS)
.map_err(internal_err)?;
Ok(Json(AuthSession { token, user }))
}
async fn auth_logout(
State(state): State<AppState>,
headers: HeaderMap,
) -> Result<StatusCode, StatusCode> {
if let Some(token) = bearer_token(&headers) {
let _ = state.db.delete_session(&token);
}
Ok(StatusCode::NO_CONTENT)
}
async fn auth_me(
State(state): State<AppState>,
headers: HeaderMap,
) -> Result<Json<UserPublic>, StatusCode> {
Ok(Json(require_user(&state, &headers)?))
}
async fn get_settings(
State(state): State<AppState>,
headers: HeaderMap,
) -> Result<Json<AppSettings>, StatusCode> {
let _user = require_user(&state, &headers)?;
Ok(Json(load_settings(&state)?))
}
async fn put_settings(
State(state): State<AppState>,
headers: HeaderMap,
Json(body): Json<AppSettings>,
) -> Result<Json<AppSettings>, (StatusCode, Json<ApiError>)> {
let _user = require_user(&state, &headers).map_err(|_| unauthorized())?;
let saved = state.db.save_settings(&body).map_err(internal_err)?;
if saved.cleanup_logs_older_than_3_months {
if let Err(e) = state
.db
.prune_health_results_older_than(crate::settings::HEALTH_LOG_RETENTION_DAYS)
{
tracing::warn!(error = %e, "immediate health log prune failed");
}
}
Ok(Json(saved))
}
fn bad_request(msg: String) -> (StatusCode, Json<ApiError>) {
(StatusCode::BAD_REQUEST, Json(ApiError { error: msg }))
}
fn internal_err(e: anyhow::Error) -> (StatusCode, Json<ApiError>) {
tracing::error!(error = %e, "internal error");
(
StatusCode::INTERNAL_SERVER_ERROR,
Json(ApiError {
error: "internal error".into(),
}),
)
}
// --- Site health checks -----------------------------------------------------
#[derive(Deserialize)]
struct ProbeRequest {
url: String,
expected_status: Option<u16>,
expect_body_contains: Option<String>,
timeout_ms: Option<u64>,
}
#[derive(Serialize)]
struct SiteCreateResponse {
site: HealthSite,
probe: ProbeResult,
}
fn unauthorized() -> (StatusCode, Json<ApiError>) {
(
StatusCode::UNAUTHORIZED,
Json(ApiError {
error: "unauthorized".into(),
}),
)
}
async fn list_sites(
State(state): State<AppState>,
headers: HeaderMap,
) -> Result<Json<Vec<HealthSite>>, StatusCode> {
let user = optional_user(&state, &headers)?;
let settings = load_settings(&state)?;
let sites = state
.db
.list_health_sites()
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
if user.is_some() {
return Ok(Json(sites));
}
if settings.public_site_ids.is_empty() {
return Err(StatusCode::UNAUTHORIZED);
}
let public: Vec<_> = sites
.into_iter()
.filter(|s| settings.site_is_public(s.id))
.collect();
Ok(Json(public))
}
async fn get_site(
State(state): State<AppState>,
headers: HeaderMap,
Path(id): Path<i64>,
) -> Result<Json<HealthSite>, StatusCode> {
require_site_read(&state, &headers, id)?;
state
.db
.get_health_site(id)
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?
.map(Json)
.ok_or(StatusCode::NOT_FOUND)
}
#[derive(Deserialize)]
struct HistoryQuery {
range: Option<String>,
}
async fn site_history(
State(state): State<AppState>,
headers: HeaderMap,
Path(id): Path<i64>,
Query(q): Query<HistoryQuery>,
) -> Result<Json<SiteHistory>, StatusCode> {
require_site_read(&state, &headers, id)?;
let range = q.range.as_deref().unwrap_or("24h");
state
.db
.site_history(id, range)
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?
.map(Json)
.ok_or(StatusCode::NOT_FOUND)
}
async fn probe_site(
State(state): State<AppState>,
headers: HeaderMap,
Json(body): Json<ProbeRequest>,
) -> Result<Json<ProbeResult>, StatusCode> {
let _user = require_user(&state, &headers)?;
let expected = body.expected_status.unwrap_or(200);
let timeout = body.timeout_ms.unwrap_or(10_000);
let expect = body
.expect_body_contains
.as_ref()
.map(|s| s.trim())
.filter(|s| !s.is_empty());
Ok(Json(
probe_url(&body.url, expected, expect, timeout).await,
))
}
async fn create_site(
State(state): State<AppState>,
headers: HeaderMap,
Json(body): Json<UpsertSite>,
) -> Result<Json<SiteCreateResponse>, (StatusCode, Json<ApiError>)> {
let _user = require_user(&state, &headers).map_err(|_| unauthorized())?;
let expected = body.expected_status.unwrap_or(200) as u16;
let timeout = body.timeout_ms.unwrap_or(10_000) as u64;
let expect = body
.expect_body_contains
.as_ref()
.map(|s| s.trim().to_string())
.filter(|s| !s.is_empty());
let probe = probe_url(&body.url, expected, expect.as_deref(), timeout).await;
if !probe.ok {
let detail = if probe.cloudflare_fallback || probe.cloudflare_challenge {
format!(
"URL failed Cloudflare/origin validation: {}",
probe.reasons.join("; ")
)
} else {
format!("URL health probe failed: {}", probe.reasons.join("; "))
};
return Err((StatusCode::UNPROCESSABLE_ENTITY, Json(ApiError { error: detail })));
}
let site = state
.db
.insert_health_site(&body)
.map_err(|e| {
let msg = e.to_string();
if msg.contains("UNIQUE") || msg.contains("unique") {
bad_request("a site with this URL already exists".into())
} else {
bad_request(msg)
}
})?;
if let Err(e) = state.db.record_health_result(site.id, &probe) {
tracing::warn!(error = %e, "failed to store initial probe result");
}
let site = state
.db
.get_health_site(site.id)
.map_err(internal_err)?
.unwrap_or(site);
Ok(Json(SiteCreateResponse { site, probe }))
}
async fn update_site(
State(state): State<AppState>,
headers: HeaderMap,
Path(id): Path<i64>,
Json(body): Json<UpsertSite>,
) -> Result<Json<SiteCreateResponse>, (StatusCode, Json<ApiError>)> {
let _user = require_user(&state, &headers).map_err(|_| unauthorized())?;
let expected = body.expected_status.unwrap_or(200) as u16;
let timeout = body.timeout_ms.unwrap_or(10_000) as u64;
let expect = body
.expect_body_contains
.as_ref()
.map(|s| s.trim().to_string())
.filter(|s| !s.is_empty());
let probe = probe_url(&body.url, expected, expect.as_deref(), timeout).await;
if !probe.ok {
let detail = if probe.cloudflare_fallback || probe.cloudflare_challenge {
format!(
"URL failed Cloudflare/origin validation: {}",
probe.reasons.join("; ")
)
} else {
format!("URL health probe failed: {}", probe.reasons.join("; "))
};
return Err((StatusCode::UNPROCESSABLE_ENTITY, Json(ApiError { error: detail })));
}
let site = state
.db
.update_health_site(id, &body)
.map_err(|e| {
let msg = e.to_string();
if msg.contains("not found") {
(
StatusCode::NOT_FOUND,
Json(ApiError {
error: "site not found".into(),
}),
)
} else if msg.contains("UNIQUE") || msg.contains("unique") {
bad_request("a site with this URL already exists".into())
} else {
bad_request(msg)
}
})?;
let _ = state.db.record_health_result(site.id, &probe);
let site = state
.db
.get_health_site(site.id)
.map_err(internal_err)?
.unwrap_or(site);
Ok(Json(SiteCreateResponse { site, probe }))
}
async fn delete_site(
State(state): State<AppState>,
headers: HeaderMap,
Path(id): Path<i64>,
) -> Result<StatusCode, StatusCode> {
let _user = require_user(&state, &headers)?;
let deleted = state
.db
.delete_health_site(id)
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
if deleted {
Ok(StatusCode::NO_CONTENT)
} else {
Err(StatusCode::NOT_FOUND)
}
}
async fn check_site_now(
State(state): State<AppState>,
headers: HeaderMap,
Path(id): Path<i64>,
) -> Result<Json<SiteCreateResponse>, (StatusCode, Json<ApiError>)> {
let _user = require_user(&state, &headers).map_err(|_| unauthorized())?;
let site = state
.db
.get_health_site(id)
.map_err(internal_err)?
.ok_or_else(|| {
(
StatusCode::NOT_FOUND,
Json(ApiError {
error: "site not found".into(),
}),
)
})?;
let expect = site.expect_body_contains.clone();
let probe = probe_url(
&site.url,
site.expected_status as u16,
expect.as_deref(),
site.timeout_ms as u64,
)
.await;
state
.db
.record_health_result(site.id, &probe)
.map_err(internal_err)?;
let site = state
.db
.get_health_site(site.id)
.map_err(internal_err)?
.unwrap_or(site);
Ok(Json(SiteCreateResponse { site, probe }))
}
+188
View File
@@ -0,0 +1,188 @@
//! Single local admin account + session tokens (SQLite-backed).
//!
//! First-run setup creates the only admin (email + password). After that,
//! interactive dashboard use requires login.
use anyhow::{bail, Context, Result};
use argon2::password_hash::{PasswordHash, PasswordHasher, PasswordVerifier, SaltString};
use argon2::Argon2;
use rand::rngs::OsRng;
use rand::RngCore;
use sha2::{Digest, Sha256};
use crate::db::Db;
#[derive(Debug, Clone, serde::Serialize)]
pub struct UserPublic {
pub id: i64,
pub email: String,
pub created_at: i64,
}
pub fn hash_password(password: &str) -> Result<String> {
let salt = SaltString::generate(&mut OsRng);
let hash = Argon2::default()
.hash_password(password.as_bytes(), &salt)
.map_err(|e| anyhow::anyhow!("hash password: {e}"))?
.to_string();
Ok(hash)
}
pub fn verify_password(password: &str, password_hash: &str) -> Result<bool> {
let parsed = PasswordHash::new(password_hash)
.map_err(|e| anyhow::anyhow!("parse hash: {e}"))?;
Ok(Argon2::default()
.verify_password(password.as_bytes(), &parsed)
.is_ok())
}
fn hash_token(token: &str) -> String {
let mut hasher = Sha256::new();
hasher.update(token.as_bytes());
hex::encode(hasher.finalize())
}
fn new_token() -> String {
let mut bytes = [0u8; 32];
OsRng.fill_bytes(&mut bytes);
hex::encode(bytes)
}
fn normalize_email(email: &str) -> Result<String> {
let email = email.trim().to_lowercase();
if email.len() < 5 || !email.contains('@') || !email.contains('.') {
bail!("enter a valid email address");
}
let (local, domain) = email
.split_once('@')
.ok_or_else(|| anyhow::anyhow!("enter a valid email address"))?;
if local.is_empty() || domain.is_empty() || !domain.contains('.') {
bail!("enter a valid email address");
}
Ok(email)
}
impl Db {
pub fn user_count(&self) -> Result<i64> {
let conn = self.conn.lock().unwrap();
let n: i64 = conn.query_row("SELECT COUNT(*) FROM users", [], |r| r.get(0))?;
Ok(n)
}
/// Create the single admin account. Email is stored in the `username` column.
pub fn create_admin(&self, email: &str, password: &str) -> Result<UserPublic> {
if self.user_count()? > 0 {
bail!("admin account already exists");
}
let email = normalize_email(email)?;
if password.len() < 8 {
bail!("password must be at least 8 characters");
}
let hash = hash_password(password)?;
let now = chrono::Utc::now().timestamp();
let conn = self.conn.lock().unwrap();
conn.execute(
"INSERT INTO users (username, password_hash, created_at) VALUES (?1, ?2, ?3)",
rusqlite::params![email, hash, now],
)
.context("create admin")?;
let id = conn.last_insert_rowid();
Ok(UserPublic {
id,
email,
created_at: now,
})
}
pub fn authenticate(&self, email: &str, password: &str) -> Result<Option<UserPublic>> {
let email = normalize_email(email)?;
let conn = self.conn.lock().unwrap();
let row = conn
.query_row(
"SELECT id, username, password_hash, created_at FROM users WHERE username = ?1",
rusqlite::params![email],
|r| {
Ok((
r.get::<_, i64>(0)?,
r.get::<_, String>(1)?,
r.get::<_, String>(2)?,
r.get::<_, i64>(3)?,
))
},
)
.optional()?;
let Some((id, email, hash, created_at)) = row else {
return Ok(None);
};
drop(conn);
if !verify_password(password, &hash)? {
return Ok(None);
}
Ok(Some(UserPublic {
id,
email,
created_at,
}))
}
pub fn create_session(&self, user_id: i64, ttl_secs: i64) -> Result<String> {
let token = new_token();
let token_hash = hash_token(&token);
let now = chrono::Utc::now().timestamp();
let expires = now + ttl_secs;
let conn = self.conn.lock().unwrap();
conn.execute(
"INSERT INTO sessions (token_hash, user_id, created_at, expires_at) VALUES (?1, ?2, ?3, ?4)",
rusqlite::params![token_hash, user_id, now, expires],
)?;
Ok(token)
}
pub fn user_from_token(&self, token: &str) -> Result<Option<UserPublic>> {
if token.is_empty() {
return Ok(None);
}
let token_hash = hash_token(token);
let now = chrono::Utc::now().timestamp();
let conn = self.conn.lock().unwrap();
let row = conn
.query_row(
"SELECT u.id, u.username, u.created_at
FROM sessions s
JOIN users u ON u.id = s.user_id
WHERE s.token_hash = ?1 AND s.expires_at > ?2",
rusqlite::params![token_hash, now],
|r| {
Ok(UserPublic {
id: r.get(0)?,
email: r.get(1)?,
created_at: r.get(2)?,
})
},
)
.optional()?;
Ok(row)
}
pub fn delete_session(&self, token: &str) -> Result<()> {
let token_hash = hash_token(token);
let conn = self.conn.lock().unwrap();
conn.execute(
"DELETE FROM sessions WHERE token_hash = ?1",
rusqlite::params![token_hash],
)?;
Ok(())
}
pub fn prune_expired_sessions(&self) -> Result<usize> {
let now = chrono::Utc::now().timestamp();
let conn = self.conn.lock().unwrap();
let n = conn.execute(
"DELETE FROM sessions WHERE expires_at <= ?1",
rusqlite::params![now],
)?;
Ok(n)
}
}
use rusqlite::OptionalExtension;
+252
View File
@@ -0,0 +1,252 @@
//! Client for the bwflow daemon Unix socket (`/run/bwflow/bwflow.sock`).
//!
//! Protocol: send `status\n`, receive one NDJSON object.
use std::path::{Path, PathBuf};
use std::process::Command;
use std::time::Duration;
use anyhow::{Context, Result};
use serde::{Deserialize, Serialize};
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::net::UnixStream;
pub const DEFAULT_SOCKET: &str = "/run/bwflow/bwflow.sock";
#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct BwflowRawStatus {
pub state: String,
pub tx_bytes: u64,
pub tokens_bytes: i64,
pub throttle_kbps: u64,
pub monthly_budget_gb: u64,
pub pre_charge_gb: u64,
pub interface: String,
pub as_of: String,
}
#[derive(Debug, Clone, Serialize)]
pub struct MonthTraffic {
pub year: i32,
pub month: u32,
pub rx_bytes: u64,
pub tx_bytes: u64,
}
#[derive(Debug, Clone, Serialize)]
pub struct BwflowStats {
pub available: bool,
pub error: Option<String>,
pub socket: String,
pub state: Option<String>,
pub interface: Option<String>,
pub as_of: Option<String>,
pub tx_bytes: Option<u64>,
pub tokens_bytes: Option<i64>,
pub throttle_kbps: Option<u64>,
pub monthly_budget_gb: Option<u64>,
pub pre_charge_gb: Option<u64>,
/// 0–100 of monthly budget consumed
pub budget_used_pct: Option<f64>,
pub tokens_remaining_gb: Option<f64>,
pub tx_gb: Option<f64>,
pub rate_limit_on: Option<bool>,
/// Calendar-month period label, e.g. "September 2026"
pub calendar_month: Option<String>,
/// Inclusive start of the Oracle calendar-month window (UTC ISO date)
pub period_start: Option<String>,
/// Exclusive end / reset date of the window (1st of next month, UTC)
pub period_resets: Option<String>,
pub months: Vec<MonthTraffic>,
}
impl BwflowStats {
pub fn unavailable(socket: impl Into<String>, error: impl Into<String>) -> Self {
let (calendar_month, period_start, period_resets) = calendar_period_labels();
Self {
available: false,
error: Some(error.into()),
socket: socket.into(),
state: None,
interface: None,
as_of: None,
tx_bytes: None,
tokens_bytes: None,
throttle_kbps: None,
monthly_budget_gb: None,
pre_charge_gb: None,
budget_used_pct: None,
tokens_remaining_gb: None,
tx_gb: None,
rate_limit_on: None,
calendar_month: Some(calendar_month),
period_start: Some(period_start),
period_resets: Some(period_resets),
months: Vec::new(),
}
}
}
const GB: f64 = 1_000_000_000.0;
fn calendar_period_labels() -> (String, String, String) {
use chrono::{Datelike, Duration, TimeZone, Utc};
let now = Utc::now();
let start = Utc
.with_ymd_and_hms(now.year(), now.month(), 1, 0, 0, 0)
.single()
.unwrap_or(now);
let resets = if now.month() == 12 {
Utc.with_ymd_and_hms(now.year() + 1, 1, 1, 0, 0, 0)
.single()
.unwrap_or(start + Duration::days(31))
} else {
Utc.with_ymd_and_hms(now.year(), now.month() + 1, 1, 0, 0, 0)
.single()
.unwrap_or(start + Duration::days(31))
};
let label = start.format("%B %Y").to_string();
(
label,
start.format("%Y-%m-%d").to_string(),
resets.format("%Y-%m-%d").to_string(),
)
}
pub async fn fetch(socket: &Path) -> BwflowStats {
let sock = socket.display().to_string();
let (calendar_month, period_start, period_resets) = calendar_period_labels();
match query_status(socket).await {
Ok(raw) => {
let tx_gb = raw.tx_bytes as f64 / GB;
let budget_gb = raw.monthly_budget_gb as f64;
let used_pct = if budget_gb > 0.0 {
Some((tx_gb / budget_gb * 100.0).min(100.0))
} else {
None
};
let iface = raw.interface.clone();
let months = tokio::task::spawn_blocking(move || fetch_vnstat_months(&iface))
.await
.unwrap_or_else(|_| Vec::new());
BwflowStats {
available: true,
error: None,
socket: sock,
state: Some(raw.state.clone()),
interface: Some(raw.interface),
as_of: Some(raw.as_of),
tx_bytes: Some(raw.tx_bytes),
tokens_bytes: Some(raw.tokens_bytes),
throttle_kbps: Some(raw.throttle_kbps),
monthly_budget_gb: Some(raw.monthly_budget_gb),
pre_charge_gb: Some(raw.pre_charge_gb),
budget_used_pct: used_pct,
tokens_remaining_gb: Some(raw.tokens_bytes as f64 / GB),
tx_gb: Some(tx_gb),
rate_limit_on: Some(raw.state == "throttled"),
calendar_month: Some(calendar_month),
period_start: Some(period_start),
period_resets: Some(period_resets),
months,
}
}
Err(e) => BwflowStats::unavailable(sock, format!("{e:#}")),
}
}
async fn query_status(socket: &Path) -> Result<BwflowRawStatus> {
if !socket.exists() {
anyhow::bail!("socket not found at {}", socket.display());
}
let stream = tokio::time::timeout(Duration::from_secs(2), UnixStream::connect(socket))
.await
.context("connect timeout")?
.with_context(|| format!("connect {}", socket.display()))?;
let (reader, mut writer) = stream.into_split();
writer.write_all(b"status\n").await?;
writer.shutdown().await.ok();
let mut lines = BufReader::new(reader).lines();
let line = tokio::time::timeout(Duration::from_secs(2), lines.next_line())
.await
.context("read timeout")?
.context("read line")?
.context("empty response from bwflow")?;
let raw: BwflowRawStatus =
serde_json::from_str(&line).with_context(|| format!("parse bwflow json: {line}"))?;
Ok(raw)
}
fn fetch_vnstat_months(interface: &str) -> Vec<MonthTraffic> {
let output = Command::new("vnstat")
.args(["--json", "m", "-i", interface])
.output();
let Ok(out) = output else {
return Vec::new();
};
if !out.status.success() {
return Vec::new();
}
let Ok(v) = serde_json::from_slice::<serde_json::Value>(&out.stdout) else {
return Vec::new();
};
let mut months = Vec::new();
let Some(ifaces) = v.get("interfaces").and_then(|x| x.as_array()) else {
return months;
};
for iface in ifaces {
let name = iface.get("name").and_then(|n| n.as_str()).unwrap_or("");
if name != interface && !interface.is_empty() {
// Prefer matching interface; if only one, still take it below.
}
let Some(arr) = iface
.pointer("/traffic/month")
.and_then(|m| m.as_array())
else {
continue;
};
let take = name == interface || ifaces.len() == 1;
if !take {
continue;
}
for m in arr {
let year = m
.pointer("/date/year")
.and_then(|x| x.as_i64())
.unwrap_or(0) as i32;
let month = m
.pointer("/date/month")
.and_then(|x| x.as_u64())
.unwrap_or(0) as u32;
let rx = m.get("rx").and_then(|x| x.as_u64()).unwrap_or(0);
let tx = m.get("tx").and_then(|x| x.as_u64()).unwrap_or(0);
if year > 0 && month > 0 {
months.push(MonthTraffic {
year,
month,
rx_bytes: rx,
tx_bytes: tx,
});
}
}
if name == interface {
break;
}
}
// Keep last 6 months
if months.len() > 6 {
let skip = months.len() - 6;
months = months.into_iter().skip(skip).collect();
}
months
}
pub fn socket_from_env() -> PathBuf {
std::env::var("METERBOX_BWFLOW_SOCK")
.map(PathBuf::from)
.unwrap_or_else(|_| PathBuf::from(DEFAULT_SOCKET))
}
+64
View File
@@ -0,0 +1,64 @@
use std::time::Duration;
use crate::AppState;
/// Background job: age retention + hard SQLite size cap (default 10 GiB).
pub fn spawn(state: AppState) {
tokio::spawn(async move {
// Run once shortly after boot, then on the configured cadence.
tokio::time::sleep(Duration::from_secs(15)).await;
run_once(&state);
let secs = state.config.cleanup_interval_secs.max(60);
let mut ticker = tokio::time::interval(Duration::from_secs(secs));
ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
// Skip the immediate first tick (already ran above).
ticker.tick().await;
loop {
ticker.tick().await;
run_once(&state);
}
});
}
fn run_once(state: &AppState) {
let max = state.config.max_db_bytes;
// Aim for 90% of the cap so we don't thrash at the ceiling.
let target = (max as f64 * 0.90) as u64;
match state
.db
.run_cleanup(state.config.retention_days, max, target)
{
Ok(report) => {
tracing::info!(
deleted_by_age = report.deleted_by_age,
deleted_by_size = report.deleted_by_size,
vacuumed = report.vacuumed,
bytes_before = report.bytes_before,
bytes_after = report.bytes_after,
rows = report.rows_after,
max_bytes = max,
"sqlite cleanup finished"
);
}
Err(e) => tracing::warn!(error = %e, "sqlite cleanup failed"),
}
match state.db.get_settings() {
Ok(settings) if settings.cleanup_logs_older_than_3_months => {
match state
.db
.prune_health_results_older_than(crate::settings::HEALTH_LOG_RETENTION_DAYS)
{
Ok(n) if n > 0 => {
tracing::info!(deleted = n, "pruned health logs older than 3 months")
}
Ok(_) => {}
Err(e) => tracing::warn!(error = %e, "health log prune failed"),
}
}
Ok(_) => {}
Err(e) => tracing::warn!(error = %e, "load settings for cleanup failed"),
}
}
+43
View File
@@ -0,0 +1,43 @@
use std::time::Duration;
use crate::AppState;
use crate::metrics::Sampler;
pub fn spawn(state: AppState) {
tokio::spawn(async move {
let mut sampler = Sampler::new();
// Prime CPU & network counters
let _ = sampler.sample();
tokio::time::sleep(Duration::from_secs(1)).await;
let interval = Duration::from_secs(state.config.interval_secs.max(5));
let mut ticker = tokio::time::interval(interval);
ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
loop {
ticker.tick().await;
let sample = sampler.sample();
if let Err(e) = state.db.insert_sample(&sample) {
tracing::error!(error = %e, "failed to insert sample");
continue;
}
// Emergency size guard: if a write pushed us over the hard cap, trim ASAP.
match state.db.on_disk_bytes() {
Ok(bytes) if bytes > state.config.max_db_bytes => {
tracing::warn!(bytes, max = state.config.max_db_bytes, "db over size cap; emergency cleanup");
let target = (state.config.max_db_bytes as f64 * 0.90) as u64;
if let Err(e) = state.db.run_cleanup(
state.config.retention_days,
state.config.max_db_bytes,
target,
) {
tracing::error!(error = %e, "emergency cleanup failed");
}
}
Ok(_) => {}
Err(e) => tracing::debug!(error = %e, "could not stat db size"),
}
}
});
}
+355
View File
@@ -0,0 +1,355 @@
//! Host capacity / budget targets for this instance.
//!
//! Operational ceilings we track against local stats:
//! - ≤ 2 OCPUs and ≤ 12 GB RAM
//! - ≤ 200 GB root/block volume
//! - ≤ 10 TB outbound data transfer per calendar month
//!
//! Activity floor (7-day window) — warn when all are simultaneously low:
//! - CPU 95th percentile < 20%
//! - Network utilization < 20% of shape bandwidth
//! - Memory utilization < 20%
use serde::Serialize;
use crate::db::Db;
use crate::metrics::{boot_disk_usage, HostInfo, Sample};
pub const TARGET_OCPU_LIMIT: f64 = 2.0;
pub const TARGET_MEM_BYTES: u64 = 12 * 1024 * 1024 * 1024;
pub const TARGET_BLOCK_BYTES: u64 = 200 * 1024 * 1024 * 1024;
pub const TARGET_EGRESS_BYTES_MONTH: u64 = 10_000_000_000_000; // 10 TB / calendar month
pub const IDLE_CPU_P95_PCT: f32 = 20.0;
pub const IDLE_MEM_PCT: f32 = 20.0;
pub const IDLE_NET_UTIL_PCT: f32 = 20.0;
fn shape_bandwidth_bps(ocpus: f64) -> f64 {
ocpus.max(0.125) * 1_000_000_000.0 / 8.0
}
#[derive(Debug, Clone, Serialize)]
pub struct Check {
pub id: String,
pub label: String,
pub status: String, // "pass" | "fail" | "warn" | "info"
pub detail: String,
pub current: String,
pub limit: String,
}
#[derive(Debug, Clone, Serialize)]
pub struct ComplianceReport {
pub overall: String, // "compliant" | "at_risk" | "over_limit"
pub summary: String,
pub tier: String,
pub docs_url: String,
pub notes: Vec<String>,
pub checks: Vec<Check>,
pub host: HostSnapshot,
}
#[derive(Debug, Clone, Serialize)]
pub struct HostSnapshot {
pub ocpus: f64,
pub mem_total_bytes: u64,
pub disk_total_bytes: u64,
pub arch: String,
pub is_arm: bool,
}
pub fn disk_total_bytes() -> u64 {
boot_disk_usage().total_bytes
}
fn percentile_95(values: &mut [f32]) -> Option<f32> {
if values.is_empty() {
return None;
}
values.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal));
let idx = ((values.len() as f64 - 1.0) * 0.95).round() as usize;
Some(values[idx.min(values.len() - 1)])
}
fn avg_f32(values: &[f32]) -> Option<f32> {
if values.is_empty() {
return None;
}
Some(values.iter().map(|v| *v as f64).sum::<f64>() as f32 / values.len() as f32)
}
fn avg_f64(values: &[f64]) -> Option<f64> {
if values.is_empty() {
return None;
}
Some(values.iter().sum::<f64>() / values.len() as f64)
}
fn format_bytes(n: u64) -> String {
const KIB: f64 = 1024.0;
const MIB: f64 = KIB * 1024.0;
const GIB: f64 = MIB * 1024.0;
const TIB: f64 = GIB * 1024.0;
let x = n as f64;
if x >= TIB {
format!("{:.2} TiB", x / TIB)
} else if x >= GIB {
format!("{:.2} GiB", x / GIB)
} else if x >= MIB {
format!("{:.1} MiB", x / MIB)
} else {
format!("{n} B")
}
}
fn month_start_ts() -> i64 {
use chrono::{Datelike, TimeZone, Utc};
let now = Utc::now();
Utc.with_ymd_and_hms(now.year(), now.month(), 1, 0, 0, 0)
.single()
.map(|d| d.timestamp())
.unwrap_or(now.timestamp())
}
fn estimate_month_egress(samples: &[Sample], interval_hint_secs: f64) -> u64 {
if samples.len() < 2 {
return 0;
}
let mut total = 0.0f64;
for w in samples.windows(2) {
let dt = (w[1].ts - w[0].ts).max(1) as f64;
let dt = if dt > 3600.0 { interval_hint_secs } else { dt };
total += w[1].net_tx_bps * dt;
}
total.max(0.0) as u64
}
pub fn evaluate(
db: &Db,
host: &HostInfo,
interval_secs: u64,
bwflow_tx_bytes: Option<u64>,
) -> ComplianceReport {
let ocpus = host.cpus as f64;
let mem = host.mem_total_bytes;
let disk = disk_total_bytes();
let arch = std::env::consts::ARCH.to_string();
let is_arm = arch == "aarch64" || arch == "arm";
let mut checks = Vec::new();
let notes = vec![
"Targets apply to this host’s measured shape and traffic.".into(),
"Outbound budget is tracked per calendar month and resets on the 1st.".into(),
];
let ocpu_ok = ocpus <= TARGET_OCPU_LIMIT + 0.01;
checks.push(Check {
id: "ocpu".into(),
label: "CPU capacity".into(),
status: if ocpu_ok { "pass".into() } else { "fail".into() },
detail: if ocpu_ok {
"CPU count is within the configured target.".into()
} else {
"CPU count is above the configured target (2 OCPU).".into()
},
current: format!("{ocpus:.0} OCPU"),
limit: format!("≤ {TARGET_OCPU_LIMIT:.0} OCPU"),
});
let mem_ok = mem <= TARGET_MEM_BYTES + 512 * 1024 * 1024;
checks.push(Check {
id: "memory".into(),
label: "Memory capacity".into(),
status: if mem_ok { "pass".into() } else { "fail".into() },
detail: if mem_ok {
"Installed memory is within the configured target.".into()
} else {
"Installed memory is above the configured target (12 GB).".into()
},
current: format_bytes(mem),
limit: format!("≤ {}", format_bytes(TARGET_MEM_BYTES)),
});
let disk_ok = disk == 0 || disk <= TARGET_BLOCK_BYTES;
checks.push(Check {
id: "block".into(),
label: "Disk capacity".into(),
status: if disk == 0 {
"info".into()
} else if disk_ok {
"pass".into()
} else {
"fail".into()
},
detail: if disk == 0 {
"Could not read root filesystem size.".into()
} else if !disk_ok {
"Root volume is above the configured disk target (200 GB).".into()
} else {
format!(
"Root volume is within the disk target ({} of {}).",
format_bytes(disk),
format_bytes(TARGET_BLOCK_BYTES)
)
},
current: if disk == 0 {
"unknown".into()
} else {
format_bytes(disk)
},
limit: format!("≤ {}", format_bytes(TARGET_BLOCK_BYTES)),
});
checks.push(Check {
id: "arch".into(),
label: "Architecture".into(),
status: if is_arm { "pass".into() } else { "info".into() },
detail: if is_arm {
"aarch64 host — matches the expected ARM shape for these targets.".into()
} else {
"Non-ARM host — capacity targets still apply; verify shape separately.".into()
},
current: arch.clone(),
limit: "aarch64 preferred".into(),
});
let now = chrono::Utc::now().timestamp();
let week_ago = now - 7 * 24 * 3600;
let week_samples = db.query_range(week_ago, now).unwrap_or_default();
let mut cpu_vals: Vec<f32> = week_samples.iter().map(|s| s.cpu_pct).collect();
let mem_vals: Vec<f32> = week_samples.iter().map(|s| s.mem_pct).collect();
let bw = shape_bandwidth_bps(ocpus);
let net_utils: Vec<f64> = week_samples
.iter()
.map(|s| ((s.net_rx_bps + s.net_tx_bps) / bw * 100.0).min(100.0))
.collect();
let cpu_p95 = percentile_95(&mut cpu_vals);
let mem_avg = avg_f32(&mem_vals);
let net_avg = avg_f64(&net_utils);
let idle_cpu = cpu_p95.map(|v| v < IDLE_CPU_P95_PCT).unwrap_or(false);
let idle_mem = mem_avg.map(|v| v < IDLE_MEM_PCT).unwrap_or(false);
let idle_net = net_avg.map(|v| v < IDLE_NET_UTIL_PCT as f64).unwrap_or(false);
let low_activity = !week_samples.is_empty() && idle_cpu && idle_mem && idle_net;
let coverage_hours = if week_samples.len() < 2 {
0.0
} else {
(week_samples.last().unwrap().ts - week_samples.first().unwrap().ts) as f64 / 3600.0
};
let enough_history = coverage_hours >= 24.0;
let idle_status = if week_samples.is_empty() {
"info"
} else if low_activity && enough_history {
"warn"
} else if low_activity {
"info"
} else {
"pass"
};
checks.push(Check {
id: "idle_reclaim".into(),
label: "Activity floor (7-day)".into(),
status: idle_status.into(),
detail: {
let cpu_s = cpu_p95
.map(|v| format!("{v:.1}%"))
.unwrap_or_else(|| "n/a".into());
let mem_s = mem_avg
.map(|v| format!("{v:.1}%"))
.unwrap_or_else(|| "n/a".into());
let net_s = net_avg
.map(|v| format!("{v:.2}%"))
.unwrap_or_else(|| "n/a".into());
if week_samples.is_empty() {
"Not enough samples yet to judge activity.".into()
} else if low_activity {
format!(
"Sustained low activity on collected data (CPU p95 {cpu_s}, mem avg {mem_s}, net util avg {net_s}). History ≈ {coverage_hours:.0}h."
)
} else {
format!(
"Activity above floor. CPU p95 {cpu_s}, mem avg {mem_s}, net util avg {net_s}. History ≈ {coverage_hours:.0}h."
)
}
},
current: format!(
"cpu_p95={} mem_avg={} net_avg={}",
cpu_p95.map(|v| format!("{v:.1}%")).unwrap_or_else(|| "n/a".into()),
mem_avg.map(|v| format!("{v:.1}%")).unwrap_or_else(|| "n/a".into()),
net_avg.map(|v| format!("{v:.2}%")).unwrap_or_else(|| "n/a".into()),
),
limit: "keep some of: CPU p95≥20%, mem≥20%, or net≥20%".into(),
});
let month_samples = db
.query_range(month_start_ts(), now)
.unwrap_or_default();
let sample_egress = estimate_month_egress(&month_samples, interval_secs as f64);
let (egress, egress_source) = match bwflow_tx_bytes {
Some(tx) => (tx, "bwflow"),
None => (sample_egress, "meterbox samples"),
};
let egress_ok = egress <= TARGET_EGRESS_BYTES_MONTH;
let egress_pct = egress as f64 / TARGET_EGRESS_BYTES_MONTH as f64 * 100.0;
checks.push(Check {
id: "egress".into(),
label: "Outbound data (calendar month)".into(),
status: if bwflow_tx_bytes.is_none() && month_samples.len() < 10 {
"info".into()
} else if !egress_ok {
"fail".into()
} else if egress_pct > 80.0 {
"warn".into()
} else {
"pass".into()
},
detail: if bwflow_tx_bytes.is_none() && month_samples.len() < 10 {
"Needs bwflow or more samples for this calendar month.".into()
} else {
format!(
"Calendar-month egress from {egress_source} ({egress_pct:.2}% of 10 TB target). Resets on the 1st."
)
},
current: format_bytes(egress),
limit: "≤ 10 TB / calendar month".into(),
});
let has_fail = checks.iter().any(|c| c.status == "fail");
let has_warn = checks.iter().any(|c| c.status == "warn");
let (overall, summary) = if has_fail {
(
"over_limit".into(),
"One or more capacity targets are exceeded.".into(),
)
} else if has_warn {
(
"at_risk".into(),
"Within core targets, with warnings that need attention.".into(),
)
} else {
(
"compliant".into(),
"Host is within configured capacity and budget targets.".into(),
)
};
ComplianceReport {
overall,
summary,
tier: "Host capacity targets".into(),
docs_url: String::new(),
notes,
checks,
host: HostSnapshot {
ocpus,
mem_total_bytes: mem,
disk_total_bytes: disk,
arch,
is_arm,
},
}
}
+46
View File
@@ -0,0 +1,46 @@
use std::path::PathBuf;
use clap::Parser;
#[derive(Debug, Clone, Parser)]
#[command(name = "meterbox", about = "Host resource usage dashboard")]
pub struct Config {
/// Bind address — default localhost only (safe behind Caddy/Rathole)
#[arg(long, env = "METERBOX_LISTEN", default_value = "127.0.0.1:8787")]
pub listen: String,
/// SQLite database path
#[arg(long, env = "METERBOX_DB", default_value = "data/meterbox.db")]
pub db_path: PathBuf,
/// Directory with built React assets
#[arg(long, env = "METERBOX_STATIC", default_value = "frontend/dist")]
pub static_dir: PathBuf,
/// Sample interval in seconds
#[arg(long, env = "METERBOX_INTERVAL_SECS", default_value_t = 30)]
pub interval_secs: u64,
/// Retain samples for this many days (max ~3 months)
#[arg(long, env = "METERBOX_RETENTION_DAYS", default_value_t = 90)]
pub retention_days: i64,
/// Hard cap for SQLite on-disk size (main + WAL + SHM), bytes. Default 10 GiB.
#[arg(long, env = "METERBOX_MAX_DB_BYTES", default_value_t = 10 * 1024 * 1024 * 1024)]
pub max_db_bytes: u64,
/// How often the cleanup job runs (seconds). Default 1 hour.
#[arg(long, env = "METERBOX_CLEANUP_INTERVAL_SECS", default_value_t = 3600)]
pub cleanup_interval_secs: u64,
}
impl Config {
pub fn ensure_dirs(&self) -> anyhow::Result<()> {
if let Some(parent) = self.db_path.parent() {
if !parent.as_os_str().is_empty() {
std::fs::create_dir_all(parent)?;
}
}
Ok(())
}
}
+356
View File
@@ -0,0 +1,356 @@
use std::path::{Path, PathBuf};
use std::sync::Mutex;
use anyhow::{Context, Result};
use rusqlite::{params, Connection, OptionalExtension};
use crate::metrics::Sample;
pub struct Db {
path: PathBuf,
pub(crate) conn: Mutex<Connection>,
}
#[derive(Debug, Clone)]
pub struct CleanupReport {
pub deleted_by_age: usize,
pub deleted_by_size: usize,
pub vacuumed: bool,
pub bytes_before: u64,
pub bytes_after: u64,
pub rows_after: i64,
}
impl Db {
pub fn open(path: &Path) -> Result<Self> {
if let Some(parent) = path.parent() {
if !parent.as_os_str().is_empty() {
std::fs::create_dir_all(parent)?;
}
}
let conn = Connection::open(path).with_context(|| format!("open db {}", path.display()))?;
conn.execute_batch(
"
PRAGMA journal_mode = WAL;
PRAGMA synchronous = NORMAL;
PRAGMA temp_store = MEMORY;
PRAGMA foreign_keys = ON;
",
)?;
Ok(Self {
path: path.to_path_buf(),
conn: Mutex::new(conn),
})
}
pub fn path(&self) -> &Path {
&self.path
}
pub fn migrate(&self) -> Result<()> {
let conn = self.conn.lock().unwrap();
conn.execute_batch(
"
CREATE TABLE IF NOT EXISTS samples (
ts INTEGER PRIMARY KEY,
cpu_pct REAL NOT NULL,
mem_used_bytes INTEGER NOT NULL,
mem_total_bytes INTEGER NOT NULL,
mem_pct REAL NOT NULL,
disk_used_bytes INTEGER NOT NULL DEFAULT 0,
disk_total_bytes INTEGER NOT NULL DEFAULT 0,
disk_pct REAL NOT NULL DEFAULT 0,
net_rx_bps REAL NOT NULL,
net_tx_bps REAL NOT NULL,
load1 REAL NOT NULL,
load5 REAL NOT NULL,
load15 REAL NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_samples_ts ON samples(ts);
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
username TEXT NOT NULL UNIQUE COLLATE NOCASE,
password_hash TEXT NOT NULL,
created_at INTEGER NOT NULL
);
CREATE TABLE IF NOT EXISTS sessions (
token_hash TEXT PRIMARY KEY,
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
created_at INTEGER NOT NULL,
expires_at INTEGER NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_sessions_user ON sessions(user_id);
CREATE INDEX IF NOT EXISTS idx_sessions_expires ON sessions(expires_at);
CREATE TABLE IF NOT EXISTS health_sites (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
url TEXT NOT NULL,
enabled INTEGER NOT NULL DEFAULT 1,
interval_secs INTEGER NOT NULL DEFAULT 60,
timeout_ms INTEGER NOT NULL DEFAULT 10000,
expected_status INTEGER NOT NULL DEFAULT 200,
expect_body_contains TEXT,
created_at INTEGER NOT NULL,
updated_at INTEGER NOT NULL,
last_check_at INTEGER,
last_ok INTEGER,
last_status INTEGER,
last_latency_ms INTEGER,
last_error TEXT,
last_via_cloudflare INTEGER,
last_cf_fallback INTEGER,
last_cf_challenge INTEGER
);
CREATE UNIQUE INDEX IF NOT EXISTS idx_health_sites_url ON health_sites(url);
CREATE TABLE IF NOT EXISTS health_results (
id INTEGER PRIMARY KEY AUTOINCREMENT,
site_id INTEGER NOT NULL REFERENCES health_sites(id) ON DELETE CASCADE,
checked_at INTEGER NOT NULL,
ok INTEGER NOT NULL,
status INTEGER NOT NULL,
latency_ms INTEGER NOT NULL,
via_cloudflare INTEGER NOT NULL,
cf_fallback INTEGER NOT NULL,
cf_challenge INTEGER NOT NULL,
error TEXT,
reasons_json TEXT NOT NULL DEFAULT '[]'
);
CREATE INDEX IF NOT EXISTS idx_health_results_site_ts ON health_results(site_id, checked_at DESC);
CREATE TABLE IF NOT EXISTS settings (
key TEXT PRIMARY KEY,
value TEXT NOT NULL
);
",
)?;
// Additive migrations for samples disk columns (existing DBs).
let _ = conn.execute(
"ALTER TABLE samples ADD COLUMN disk_used_bytes INTEGER NOT NULL DEFAULT 0",
[],
);
let _ = conn.execute(
"ALTER TABLE samples ADD COLUMN disk_total_bytes INTEGER NOT NULL DEFAULT 0",
[],
);
let _ = conn.execute(
"ALTER TABLE samples ADD COLUMN disk_pct REAL NOT NULL DEFAULT 0",
[],
);
Ok(())
}
pub fn insert_sample(&self, s: &Sample) -> Result<()> {
let conn = self.conn.lock().unwrap();
conn.execute(
"INSERT OR REPLACE INTO samples
(ts, cpu_pct, mem_used_bytes, mem_total_bytes, mem_pct,
disk_used_bytes, disk_total_bytes, disk_pct,
net_rx_bps, net_tx_bps, load1, load5, load15)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13)",
params![
s.ts,
s.cpu_pct,
s.mem_used_bytes as i64,
s.mem_total_bytes as i64,
s.mem_pct,
s.disk_used_bytes as i64,
s.disk_total_bytes as i64,
s.disk_pct,
s.net_rx_bps,
s.net_tx_bps,
s.load1,
s.load5,
s.load15,
],
)?;
Ok(())
}
pub fn prune_older_than(&self, older_than_ts: i64) -> Result<usize> {
let conn = self.conn.lock().unwrap();
let n = conn.execute("DELETE FROM samples WHERE ts < ?1", params![older_than_ts])?;
Ok(n)
}
/// Delete the oldest `limit` rows. Returns how many were removed.
pub fn prune_oldest(&self, limit: i64) -> Result<usize> {
if limit <= 0 {
return Ok(0);
}
let conn = self.conn.lock().unwrap();
let n = conn.execute(
"DELETE FROM samples WHERE ts IN (
SELECT ts FROM samples ORDER BY ts ASC LIMIT ?1
)",
params![limit],
)?;
Ok(n)
}
pub fn checkpoint_truncate(&self) -> Result<()> {
let conn = self.conn.lock().unwrap();
conn.execute_batch("PRAGMA wal_checkpoint(TRUNCATE);")?;
Ok(())
}
pub fn vacuum(&self) -> Result<()> {
// VACUUM cannot run inside a transaction; unlock and run dedicated connection.
{
let conn = self.conn.lock().unwrap();
conn.execute_batch("PRAGMA wal_checkpoint(TRUNCATE);")?;
}
let conn = Connection::open(&self.path)?;
conn.execute_batch("VACUUM;")?;
// Re-apply WAL on the long-lived connection
let conn = self.conn.lock().unwrap();
conn.execute_batch("PRAGMA journal_mode = WAL;")?;
Ok(())
}
/// On-disk size of main DB + WAL + SHM.
pub fn on_disk_bytes(&self) -> Result<u64> {
Ok(file_size(&self.path)?
+ file_size(Path::new(&format!("{}-wal", self.path.display())))?
+ file_size(Path::new(&format!("{}-shm", self.path.display())))?)
}
pub fn latest(&self) -> Result<Option<Sample>> {
let conn = self.conn.lock().unwrap();
conn.query_row(
"SELECT ts, cpu_pct, mem_used_bytes, mem_total_bytes, mem_pct,
disk_used_bytes, disk_total_bytes, disk_pct,
net_rx_bps, net_tx_bps, load1, load5, load15
FROM samples ORDER BY ts DESC LIMIT 1",
[],
row_to_sample,
)
.optional()
.map_err(Into::into)
}
pub fn query_range(&self, from_ts: i64, to_ts: i64) -> Result<Vec<Sample>> {
let conn = self.conn.lock().unwrap();
let mut stmt = conn.prepare(
"SELECT ts, cpu_pct, mem_used_bytes, mem_total_bytes, mem_pct,
disk_used_bytes, disk_total_bytes, disk_pct,
net_rx_bps, net_tx_bps, load1, load5, load15
FROM samples WHERE ts >= ?1 AND ts <= ?2 ORDER BY ts ASC",
)?;
let rows = stmt.query_map(params![from_ts, to_ts], row_to_sample)?;
let mut out = Vec::new();
for r in rows {
out.push(r?);
}
Ok(out)
}
pub fn count(&self) -> Result<i64> {
let conn = self.conn.lock().unwrap();
let n: i64 = conn.query_row("SELECT COUNT(*) FROM samples", [], |r| r.get(0))?;
Ok(n)
}
pub fn oldest_ts(&self) -> Result<Option<i64>> {
let conn = self.conn.lock().unwrap();
conn.query_row("SELECT MIN(ts) FROM samples", [], |r| r.get(0))
.optional()
.map_err(Into::into)
}
/// Age retention + hard size cap. Shrinks to `target_bytes` when over `max_bytes`.
pub fn run_cleanup(
&self,
retention_days: i64,
max_bytes: u64,
target_bytes: u64,
) -> Result<CleanupReport> {
let bytes_before = self.on_disk_bytes()?;
let mut deleted_by_age = 0usize;
let mut deleted_by_size = 0usize;
let mut vacuumed = false;
let cutoff = chrono::Utc::now().timestamp() - retention_days.max(1) * 24 * 60 * 60;
deleted_by_age = self.prune_older_than(cutoff)?;
// Keep deleting oldest batches until under target when over the hard cap.
let mut guard = 0;
while self.on_disk_bytes()? > max_bytes.max(target_bytes) && guard < 64 {
guard += 1;
let rows = self.count()?.max(1);
// Remove ~5% of rows each pass (min 1k, max 100k) so we converge quickly.
let batch = ((rows as f64) * 0.05).ceil() as i64;
let batch = batch.clamp(1_000, 100_000);
let n = self.prune_oldest(batch)?;
if n == 0 {
break;
}
deleted_by_size += n;
let _ = self.checkpoint_truncate();
}
let needs_reclaim = deleted_by_age > 0
|| deleted_by_size > 0
|| self.on_disk_bytes()? > target_bytes;
if needs_reclaim && self.on_disk_bytes()? > target_bytes {
self.vacuum()?;
vacuumed = true;
} else if deleted_by_age > 0 || deleted_by_size > 0 {
let _ = self.checkpoint_truncate();
}
// If still over hard cap after vacuum, keep trimming (pathological case).
guard = 0;
while self.on_disk_bytes()? > max_bytes && guard < 32 {
guard += 1;
let rows = self.count()?.max(1);
let batch = ((rows as f64) * 0.10).ceil() as i64;
let batch = batch.clamp(5_000, 200_000);
let n = self.prune_oldest(batch)?;
if n == 0 {
break;
}
deleted_by_size += n;
self.vacuum()?;
vacuumed = true;
}
Ok(CleanupReport {
deleted_by_age,
deleted_by_size,
vacuumed,
bytes_before,
bytes_after: self.on_disk_bytes()?,
rows_after: self.count()?,
})
}
}
fn file_size(path: &Path) -> Result<u64> {
match std::fs::metadata(path) {
Ok(m) => Ok(m.len()),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(0),
Err(e) => Err(e.into()),
}
}
fn row_to_sample(row: &rusqlite::Row<'_>) -> rusqlite::Result<Sample> {
Ok(Sample {
ts: row.get(0)?,
cpu_pct: row.get(1)?,
mem_used_bytes: row.get::<_, i64>(2)? as u64,
mem_total_bytes: row.get::<_, i64>(3)? as u64,
mem_pct: row.get(4)?,
disk_used_bytes: row.get::<_, i64>(5)? as u64,
disk_total_bytes: row.get::<_, i64>(6)? as u64,
disk_pct: row.get(7)?,
net_rx_bps: row.get(8)?,
net_tx_bps: row.get(9)?,
load1: row.get(10)?,
load5: row.get(11)?,
load15: row.get(12)?,
})
}
+596
View File
@@ -0,0 +1,596 @@
//! Configurable site health checks stored in SQLite.
use anyhow::{bail, Context, Result};
use rusqlite::{params, OptionalExtension};
use serde::{Deserialize, Serialize};
use crate::db::Db;
use crate::probe::{probe_url, ProbeResult};
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct HealthSite {
pub id: i64,
pub name: String,
pub url: String,
pub enabled: bool,
pub interval_secs: i64,
pub timeout_ms: i64,
pub expected_status: i64,
pub expect_body_contains: Option<String>,
pub created_at: i64,
pub updated_at: i64,
pub last_check_at: Option<i64>,
pub last_ok: Option<bool>,
pub last_status: Option<i64>,
pub last_latency_ms: Option<i64>,
pub last_error: Option<String>,
pub last_via_cloudflare: Option<bool>,
pub last_cf_fallback: Option<bool>,
pub last_cf_challenge: Option<bool>,
}
#[derive(Debug, Clone, Deserialize)]
pub struct UpsertSite {
pub name: String,
pub url: String,
pub enabled: Option<bool>,
pub interval_secs: Option<i64>,
pub timeout_ms: Option<i64>,
pub expected_status: Option<i64>,
pub expect_body_contains: Option<String>,
}
#[derive(Debug, Clone, Serialize)]
pub struct HealthCheckResult {
pub id: i64,
pub site_id: i64,
pub checked_at: i64,
pub ok: bool,
pub status: i64,
pub latency_ms: i64,
pub via_cloudflare: bool,
pub cf_fallback: bool,
pub cf_challenge: bool,
pub error: Option<String>,
pub reasons: Vec<String>,
}
#[derive(Debug, Clone, Serialize)]
pub struct HealthIncident {
pub started_at: i64,
pub ended_at: Option<i64>,
pub duration_secs: Option<i64>,
pub ongoing: bool,
pub error: String,
pub status: Option<i64>,
pub checks: i64,
}
#[derive(Debug, Clone, Serialize)]
pub struct HistorySummary {
pub checks: i64,
pub ok: i64,
pub down: i64,
pub uptime_pct: Option<f64>,
}
#[derive(Debug, Clone, Serialize)]
pub struct SiteHistory {
pub site: HealthSite,
pub range: String,
pub from_ts: i64,
pub to_ts: i64,
pub bucket_secs: i64,
pub summary: HistorySummary,
pub incidents: Vec<HealthIncident>,
pub series: Vec<HealthSeriesPoint>,
pub results: Vec<HealthCheckResult>,
}
#[derive(Debug, Clone, Serialize)]
pub struct HealthSeriesPoint {
pub ts: i64,
/// Fraction healthy in bucket, 0–100.
pub uptime_pct: f64,
pub ok: i64,
pub down: i64,
pub avg_latency_ms: f64,
/// 1 if any check failed in the bucket, else 0 (for status chart).
pub had_down: i64,
}
fn parse_health_range(range: &str) -> (i64, i64) {
// (window_secs, bucket_secs)
match range {
"1h" => (3600, 60),
"12h" => (12 * 3600, 5 * 60),
"24h" => (24 * 3600, 10 * 60),
"7d" => (7 * 86400, 3600),
"30d" => (30 * 86400, 4 * 3600),
_ => (24 * 3600, 10 * 60),
}
}
/// Collapse consecutive failures into downtime incidents (newest first).
fn derive_incidents(results_newest_first: &[HealthCheckResult]) -> Vec<HealthIncident> {
let mut chronological: Vec<&HealthCheckResult> = results_newest_first.iter().collect();
chronological.reverse();
let mut incidents = Vec::new();
let mut current: Option<(i64, String, Option<i64>, i64)> = None; // start, error, status, checks
for r in &chronological {
if !r.ok {
match &mut current {
Some((_start, error, status, checks)) => {
*checks += 1;
if error.is_empty() {
if let Some(e) = &r.error {
*error = e.clone();
}
}
if status.is_none() && r.status > 0 {
*status = Some(r.status);
}
}
None => {
current = Some((
r.checked_at,
r.error.clone().unwrap_or_else(|| "health check failed".into()),
if r.status > 0 { Some(r.status) } else { None },
1,
));
}
}
} else if let Some((started_at, error, status, checks)) = current.take() {
incidents.push(HealthIncident {
started_at,
ended_at: Some(r.checked_at),
duration_secs: Some((r.checked_at - started_at).max(0)),
ongoing: false,
error,
status,
checks,
});
}
}
if let Some((started_at, error, status, checks)) = current {
let ended = chronological.last().map(|r| r.checked_at);
incidents.push(HealthIncident {
started_at,
ended_at: None,
duration_secs: ended.map(|e| (e - started_at).max(0)),
ongoing: true,
error,
status,
checks,
});
}
incidents.reverse(); // newest first
incidents
}
impl Db {
pub fn list_health_sites(&self) -> Result<Vec<HealthSite>> {
let conn = self.conn.lock().unwrap();
let mut stmt = conn.prepare(
"SELECT id, name, url, enabled, interval_secs, timeout_ms, expected_status,
expect_body_contains, created_at, updated_at, last_check_at, last_ok,
last_status, last_latency_ms, last_error, last_via_cloudflare,
last_cf_fallback, last_cf_challenge
FROM health_sites ORDER BY name COLLATE NOCASE ASC",
)?;
let rows = stmt.query_map([], row_to_site)?;
let mut out = Vec::new();
for r in rows {
out.push(r?);
}
Ok(out)
}
pub fn get_health_site(&self, id: i64) -> Result<Option<HealthSite>> {
let conn = self.conn.lock().unwrap();
conn.query_row(
"SELECT id, name, url, enabled, interval_secs, timeout_ms, expected_status,
expect_body_contains, created_at, updated_at, last_check_at, last_ok,
last_status, last_latency_ms, last_error, last_via_cloudflare,
last_cf_fallback, last_cf_challenge
FROM health_sites WHERE id = ?1",
params![id],
row_to_site,
)
.optional()
.map_err(Into::into)
}
pub fn insert_health_site(&self, input: &UpsertSite) -> Result<HealthSite> {
let name = input.name.trim();
let url = input.url.trim();
if name.is_empty() {
bail!("name is required");
}
if url.is_empty() {
bail!("url is required");
}
let now = chrono::Utc::now().timestamp();
let enabled = input.enabled.unwrap_or(true);
let interval_secs = input.interval_secs.unwrap_or(60).clamp(15, 3600);
let timeout_ms = input.timeout_ms.unwrap_or(10_000).clamp(1000, 60_000);
let expected_status = input.expected_status.unwrap_or(200).clamp(100, 599);
let expect_body = input
.expect_body_contains
.as_ref()
.map(|s| s.trim().to_string())
.filter(|s| !s.is_empty());
let conn = self.conn.lock().unwrap();
conn.execute(
"INSERT INTO health_sites
(name, url, enabled, interval_secs, timeout_ms, expected_status, expect_body_contains, created_at, updated_at)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)",
params![
name,
url,
enabled as i64,
interval_secs,
timeout_ms,
expected_status,
expect_body,
now,
now
],
)
.context("insert health site")?;
let id = conn.last_insert_rowid();
drop(conn);
self.get_health_site(id)?
.ok_or_else(|| anyhow::anyhow!("site missing after insert"))
}
pub fn update_health_site(&self, id: i64, input: &UpsertSite) -> Result<HealthSite> {
let existing = self
.get_health_site(id)?
.ok_or_else(|| anyhow::anyhow!("site not found"))?;
let name = input.name.trim();
let url = input.url.trim();
if name.is_empty() || url.is_empty() {
bail!("name and url are required");
}
let now = chrono::Utc::now().timestamp();
let enabled = input.enabled.unwrap_or(existing.enabled);
let interval_secs = input
.interval_secs
.unwrap_or(existing.interval_secs)
.clamp(15, 3600);
let timeout_ms = input
.timeout_ms
.unwrap_or(existing.timeout_ms)
.clamp(1000, 60_000);
let expected_status = input
.expected_status
.unwrap_or(existing.expected_status)
.clamp(100, 599);
let expect_body = input
.expect_body_contains
.as_ref()
.map(|s| s.trim().to_string())
.filter(|s| !s.is_empty());
let conn = self.conn.lock().unwrap();
conn.execute(
"UPDATE health_sites SET
name = ?1, url = ?2, enabled = ?3, interval_secs = ?4, timeout_ms = ?5,
expected_status = ?6, expect_body_contains = ?7, updated_at = ?8
WHERE id = ?9",
params![
name,
url,
enabled as i64,
interval_secs,
timeout_ms,
expected_status,
expect_body,
now,
id
],
)?;
drop(conn);
self.get_health_site(id)?
.ok_or_else(|| anyhow::anyhow!("site missing after update"))
}
pub fn delete_health_site(&self, id: i64) -> Result<bool> {
let conn = self.conn.lock().unwrap();
let n = conn.execute("DELETE FROM health_sites WHERE id = ?1", params![id])?;
Ok(n > 0)
}
pub fn record_health_result(&self, site_id: i64, probe: &ProbeResult) -> Result<()> {
let now = chrono::Utc::now().timestamp();
let error_text = if probe.ok {
None
} else {
Some(
probe
.error
.clone()
.filter(|s| !s.trim().is_empty())
.unwrap_or_else(|| probe.reasons.join("; ")),
)
.filter(|s| !s.is_empty())
.or_else(|| Some("health check failed".into()))
};
let conn = self.conn.lock().unwrap();
conn.execute(
"INSERT INTO health_results
(site_id, checked_at, ok, status, latency_ms, via_cloudflare, cf_fallback, cf_challenge, error, reasons_json)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)",
params![
site_id,
now,
probe.ok as i64,
probe.status as i64,
probe.latency_ms as i64,
probe.via_cloudflare as i64,
probe.cloudflare_fallback as i64,
probe.cloudflare_challenge as i64,
error_text,
serde_json::to_string(&probe.reasons).unwrap_or_else(|_| "[]".into()),
],
)?;
conn.execute(
"UPDATE health_sites SET
last_check_at = ?1,
last_ok = ?2,
last_status = ?3,
last_latency_ms = ?4,
last_error = ?5,
last_via_cloudflare = ?6,
last_cf_fallback = ?7,
last_cf_challenge = ?8,
updated_at = ?1
WHERE id = ?9",
params![
now,
probe.ok as i64,
probe.status as i64,
probe.latency_ms as i64,
error_text,
probe.via_cloudflare as i64,
probe.cloudflare_fallback as i64,
probe.cloudflare_challenge as i64,
site_id
],
)?;
// Age pruning of health logs is controlled by settings (3-month checkbox).
Ok(())
}
pub fn list_health_results_since(
&self,
site_id: i64,
since: i64,
limit: i64,
) -> Result<Vec<HealthCheckResult>> {
let limit = limit.clamp(1, 50_000);
let conn = self.conn.lock().unwrap();
let mut stmt = conn.prepare(
"SELECT id, site_id, checked_at, ok, status, latency_ms, via_cloudflare,
cf_fallback, cf_challenge, error, reasons_json
FROM health_results
WHERE site_id = ?1 AND checked_at >= ?2
ORDER BY checked_at DESC
LIMIT ?3",
)?;
let rows = stmt.query_map(params![site_id, since, limit], row_to_result)?;
let mut out = Vec::new();
for r in rows {
out.push(r?);
}
Ok(out)
}
pub fn site_history(&self, site_id: i64, range: &str) -> Result<Option<SiteHistory>> {
let site = match self.get_health_site(site_id)? {
Some(s) => s,
None => return Ok(None),
};
let (window, bucket) = parse_health_range(range);
let to_ts = chrono::Utc::now().timestamp();
let from_ts = to_ts - window;
let results = self.list_health_results_since(site_id, from_ts, 50_000)?;
let total = results.len() as i64;
let ok_count = results.iter().filter(|r| r.ok).count() as i64;
let down_count = total - ok_count;
let uptime_pct = if total == 0 {
None
} else {
Some((ok_count as f64) * 100.0 / (total as f64))
};
let incidents = derive_incidents(&results);
let series = bucket_results(&results, from_ts, to_ts, bucket);
Ok(Some(SiteHistory {
site,
range: normalize_range(range),
from_ts,
to_ts,
bucket_secs: bucket,
results,
incidents,
series,
summary: HistorySummary {
checks: total,
ok: ok_count,
down: down_count,
uptime_pct,
},
}))
}
pub fn sites_due_for_check(&self, now: i64) -> Result<Vec<HealthSite>> {
let sites = self.list_health_sites()?;
Ok(sites
.into_iter()
.filter(|s| s.enabled)
.filter(|s| match s.last_check_at {
None => true,
Some(ts) => now - ts >= s.interval_secs,
})
.collect())
}
}
fn row_to_result(row: &rusqlite::Row<'_>) -> rusqlite::Result<HealthCheckResult> {
let reasons_json: String = row.get(10)?;
let reasons: Vec<String> = serde_json::from_str(&reasons_json).unwrap_or_default();
Ok(HealthCheckResult {
id: row.get(0)?,
site_id: row.get(1)?,
checked_at: row.get(2)?,
ok: row.get::<_, i64>(3)? != 0,
status: row.get(4)?,
latency_ms: row.get(5)?,
via_cloudflare: row.get::<_, i64>(6)? != 0,
cf_fallback: row.get::<_, i64>(7)? != 0,
cf_challenge: row.get::<_, i64>(8)? != 0,
error: row.get(9)?,
reasons,
})
}
fn normalize_range(range: &str) -> String {
match range {
"1h" | "12h" | "24h" | "7d" | "30d" => range.to_string(),
_ => "24h".into(),
}
}
fn bucket_results(
results_newest_first: &[HealthCheckResult],
from_ts: i64,
to_ts: i64,
bucket: i64,
) -> Vec<HealthSeriesPoint> {
if bucket <= 0 || to_ts <= from_ts {
return Vec::new();
}
let n = ((to_ts - from_ts) / bucket) as usize + 1;
let mut ok = vec![0i64; n];
let mut down = vec![0i64; n];
let mut lat_sum = vec![0f64; n];
let mut lat_n = vec![0i64; n];
for r in results_newest_first {
if r.checked_at < from_ts || r.checked_at > to_ts {
continue;
}
let idx = ((r.checked_at - from_ts) / bucket) as usize;
let idx = idx.min(n - 1);
if r.ok {
ok[idx] += 1;
} else {
down[idx] += 1;
}
lat_sum[idx] += r.latency_ms as f64;
lat_n[idx] += 1;
}
(0..n)
.map(|i| {
let total = ok[i] + down[i];
let uptime_pct = if total == 0 {
// empty bucket: treat as unknown → show as 100 only if we leave gaps;
// use NaN-friendly 100 for continuity when no data, but mark had_down=0
f64::NAN
} else {
(ok[i] as f64) * 100.0 / (total as f64)
};
HealthSeriesPoint {
ts: from_ts + (i as i64) * bucket,
uptime_pct: if uptime_pct.is_nan() { 100.0 } else { uptime_pct },
ok: ok[i],
down: down[i],
avg_latency_ms: if lat_n[i] == 0 {
0.0
} else {
lat_sum[i] / lat_n[i] as f64
},
had_down: if down[i] > 0 { 1 } else { 0 },
}
})
.filter(|p| p.ok + p.down > 0) // only buckets with data
.collect()
}
fn row_to_site(row: &rusqlite::Row<'_>) -> rusqlite::Result<HealthSite> {
Ok(HealthSite {
id: row.get(0)?,
name: row.get(1)?,
url: row.get(2)?,
enabled: row.get::<_, i64>(3)? != 0,
interval_secs: row.get(4)?,
timeout_ms: row.get(5)?,
expected_status: row.get(6)?,
expect_body_contains: row.get(7)?,
created_at: row.get(8)?,
updated_at: row.get(9)?,
last_check_at: row.get(10)?,
last_ok: row
.get::<_, Option<i64>>(11)?
.map(|v| v != 0),
last_status: row.get(12)?,
last_latency_ms: row.get(13)?,
last_error: row.get(14)?,
last_via_cloudflare: row
.get::<_, Option<i64>>(15)?
.map(|v| v != 0),
last_cf_fallback: row
.get::<_, Option<i64>>(16)?
.map(|v| v != 0),
last_cf_challenge: row
.get::<_, Option<i64>>(17)?
.map(|v| v != 0),
})
}
pub fn spawn(state: crate::AppState) {
tokio::spawn(async move {
let mut ticker = tokio::time::interval(std::time::Duration::from_secs(15));
ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
loop {
ticker.tick().await;
let _ = state.db.prune_expired_sessions();
let now = chrono::Utc::now().timestamp();
let due = match state.db.sites_due_for_check(now) {
Ok(v) => v,
Err(e) => {
tracing::warn!(error = %e, "list due health sites failed");
continue;
}
};
for site in due {
let expect = site.expect_body_contains.clone();
let probe = probe_url(
&site.url,
site.expected_status as u16,
expect.as_deref(),
site.timeout_ms as u64,
)
.await;
if let Err(e) = state.db.record_health_result(site.id, &probe) {
tracing::warn!(site_id = site.id, error = %e, "record health result failed");
} else if !probe.ok {
tracing::info!(
site = %site.name,
status = probe.status,
cf_fallback = probe.cloudflare_fallback,
cf_challenge = probe.cloudflare_challenge,
"health check failed"
);
}
}
}
});
}
+86
View File
@@ -0,0 +1,86 @@
mod api;
mod auth;
mod bwflow;
mod cleanup;
mod collector;
mod compliance;
mod config;
mod db;
mod health;
mod metrics;
mod probe;
mod settings;
use std::net::SocketAddr;
use std::path::PathBuf;
use std::sync::Arc;
use axum::Router;
use clap::Parser;
use tower_http::compression::CompressionLayer;
use tower_http::cors::{Any, CorsLayer};
use tower_http::services::{ServeDir, ServeFile};
use tower_http::trace::TraceLayer;
use tracing_subscriber::EnvFilter;
use crate::config::Config;
use crate::db::Db;
#[derive(Clone)]
pub struct AppState {
pub db: Arc<Db>,
pub config: Arc<Config>,
}
#[tokio::main]
async fn main() -> anyhow::Result<()> {
tracing_subscriber::fmt()
.with_env_filter(EnvFilter::try_from_default_env().unwrap_or_else(|_| "meterbox=info".into()))
.init();
let config = Arc::new(Config::parse());
config.ensure_dirs()?;
let db = Arc::new(Db::open(&config.db_path)?);
db.migrate()?;
let state = AppState {
db: db.clone(),
config: config.clone(),
};
collector::spawn(state.clone());
cleanup::spawn(state.clone());
health::spawn(state.clone());
let static_dir = config.static_dir.clone();
let index = static_dir.join("index.html");
let api = api::router(state);
let app = Router::new()
.nest("/api", api)
.fallback_service(
ServeDir::new(&static_dir)
.not_found_service(ServeFile::new(index)),
)
.layer(CompressionLayer::new())
.layer(
CorsLayer::new()
.allow_origin(Any)
.allow_methods(Any)
.allow_headers(Any),
)
.layer(TraceLayer::new_for_http());
let addr: SocketAddr = config.listen.parse()?;
tracing::info!(%addr, db = %config.db_path.display(), "meterbox listening");
let listener = tokio::net::TcpListener::bind(addr).await?;
axum::serve(listener, app).await?;
Ok(())
}
#[allow(dead_code)]
fn default_static() -> PathBuf {
PathBuf::from("frontend/dist")
}
+203
View File
@@ -0,0 +1,203 @@
use serde::Serialize;
use sysinfo::{Networks, System};
#[derive(Debug, Clone, Serialize)]
pub struct Sample {
pub ts: i64,
pub cpu_pct: f32,
pub mem_used_bytes: u64,
pub mem_total_bytes: u64,
pub mem_pct: f32,
pub disk_used_bytes: u64,
pub disk_total_bytes: u64,
pub disk_pct: f32,
pub net_rx_bps: f64,
pub net_tx_bps: f64,
pub load1: f64,
pub load5: f64,
pub load15: f64,
}
#[derive(Debug, Clone, Serialize)]
pub struct HostInfo {
pub hostname: String,
pub os: String,
pub kernel: String,
pub cpus: usize,
pub mem_total_bytes: u64,
pub disk_used_bytes: u64,
pub disk_total_bytes: u64,
pub disk_pct: f32,
pub uptime_secs: u64,
}
#[derive(Debug, Clone, Copy)]
pub struct DiskUsage {
pub used_bytes: u64,
pub total_bytes: u64,
pub pct: f32,
}
/// Boot / root filesystem usage via `statvfs("/")`.
pub fn boot_disk_usage() -> DiskUsage {
#[cfg(target_os = "linux")]
{
use std::ffi::CString;
let Ok(path) = CString::new("/") else {
return DiskUsage {
used_bytes: 0,
total_bytes: 0,
pct: 0.0,
};
};
let mut buf: libc::statvfs = unsafe { std::mem::zeroed() };
let rc = unsafe { libc::statvfs(path.as_ptr(), &mut buf) };
if rc != 0 {
return DiskUsage {
used_bytes: 0,
total_bytes: 0,
pct: 0.0,
};
}
let frsize = buf.f_frsize as u64;
let total = buf.f_blocks as u64 * frsize;
let available = buf.f_bavail as u64 * frsize;
let used = total.saturating_sub(available);
let pct = if total > 0 {
(used as f64 / total as f64 * 100.0) as f32
} else {
0.0
};
return DiskUsage {
used_bytes: used,
total_bytes: total,
pct,
};
}
#[cfg(not(target_os = "linux"))]
{
DiskUsage {
used_bytes: 0,
total_bytes: 0,
pct: 0.0,
}
}
}
pub struct Sampler {
sys: System,
networks: Networks,
last_rx: u64,
last_tx: u64,
last_ts: Option<i64>,
primed: bool,
}
impl Sampler {
pub fn new() -> Self {
let mut sys = System::new();
sys.refresh_memory();
sys.refresh_cpu_all();
let networks = Networks::new_with_refreshed_list();
Self {
sys,
networks,
last_rx: 0,
last_tx: 0,
last_ts: None,
primed: false,
}
}
pub fn host_info(&mut self) -> HostInfo {
self.sys.refresh_memory();
let disk = boot_disk_usage();
HostInfo {
hostname: hostname::get()
.ok()
.and_then(|h| h.into_string().ok())
.unwrap_or_else(|| "unknown".into()),
os: System::long_os_version().unwrap_or_else(|| "unknown".into()),
kernel: System::kernel_version().unwrap_or_else(|| "unknown".into()),
cpus: self
.sys
.physical_core_count()
.unwrap_or_else(|| self.sys.cpus().len())
.max(1),
mem_total_bytes: self.sys.total_memory(),
disk_used_bytes: disk.used_bytes,
disk_total_bytes: disk.total_bytes,
disk_pct: disk.pct,
uptime_secs: System::uptime(),
}
}
pub fn sample(&mut self) -> Sample {
self.sys.refresh_cpu_all();
self.sys.refresh_memory();
self.networks.refresh(true);
let now = chrono::Utc::now().timestamp();
let cpu_pct = self.sys.global_cpu_usage();
let mem_total = self.sys.total_memory();
let mem_used = self.sys.used_memory();
let mem_pct = if mem_total > 0 {
(mem_used as f64 / mem_total as f64 * 100.0) as f32
} else {
0.0
};
let disk = boot_disk_usage();
let (rx, tx) = aggregate_networks(&self.networks);
let (rx_bps, tx_bps) = if let Some(prev) = self.last_ts {
let dt = (now - prev).max(1) as f64;
if self.primed && rx >= self.last_rx && tx >= self.last_tx {
(
(rx - self.last_rx) as f64 / dt,
(tx - self.last_tx) as f64 / dt,
)
} else {
(0.0, 0.0)
}
} else {
(0.0, 0.0)
};
self.last_rx = rx;
self.last_tx = tx;
self.last_ts = Some(now);
self.primed = true;
let load = System::load_average();
Sample {
ts: now,
cpu_pct,
mem_used_bytes: mem_used,
mem_total_bytes: mem_total,
mem_pct,
disk_used_bytes: disk.used_bytes,
disk_total_bytes: disk.total_bytes,
disk_pct: disk.pct,
net_rx_bps: rx_bps,
net_tx_bps: tx_bps,
load1: load.one,
load5: load.five,
load15: load.fifteen,
}
}
}
fn aggregate_networks(networks: &Networks) -> (u64, u64) {
let mut rx = 0u64;
let mut tx = 0u64;
for (name, data) in networks.iter() {
let n = name.to_lowercase();
if n == "lo" || n.starts_with("docker") || n.starts_with("br-") || n.starts_with("veth") {
continue;
}
rx = rx.saturating_add(data.total_received());
tx = tx.saturating_add(data.total_transmitted());
}
(rx, tx)
}
+230
View File
@@ -0,0 +1,230 @@
//! HTTP health probes with Cloudflare fallback / challenge detection.
use std::time::{Duration, Instant};
use serde::Serialize;
#[derive(Debug, Clone, Serialize)]
pub struct ProbeResult {
pub ok: bool,
pub url: String,
pub final_url: String,
pub status: u16,
pub latency_ms: u64,
pub server: Option<String>,
pub via_cloudflare: bool,
pub cloudflare_fallback: bool,
pub cloudflare_challenge: bool,
pub body_snippet: String,
pub error: Option<String>,
pub reasons: Vec<String>,
}
const CF_ORIGIN_ERRORS: &[u16] = &[520, 521, 522, 523, 524, 525, 526, 527, 530];
const CF_BODY_MARKERS: &[&str] = &[
"cloudflare ray id",
"cf-error-details",
"attention required! | cloudflare",
"just a moment...",
"enable javascript and cookies to continue",
"cdn-cgi/challenge",
"cdn-cgi/l/chk_jschl",
"always online",
"error 1000",
"error 1001",
"error 1014",
"error 1015",
"error 1016",
"origin is unreachable",
"web server is down",
"connection timed out",
"a timeout occurred",
"ssl handshake failed",
"invalid ssl certificate",
"please enable cookies",
"checking if the site connection is secure",
];
pub async fn probe_url(
url: &str,
expected_status: u16,
expect_body_contains: Option<&str>,
timeout_ms: u64,
) -> ProbeResult {
let mut reasons = Vec::new();
let parsed = match reqwest::Url::parse(url) {
Ok(u) => u,
Err(e) => {
return ProbeResult {
ok: false,
url: url.to_string(),
final_url: url.to_string(),
status: 0,
latency_ms: 0,
server: None,
via_cloudflare: false,
cloudflare_fallback: false,
cloudflare_challenge: false,
body_snippet: String::new(),
error: Some(format!("invalid url: {e}")),
reasons: vec!["URL failed to parse".into()],
};
}
};
if parsed.scheme() != "http" && parsed.scheme() != "https" {
return ProbeResult {
ok: false,
url: url.to_string(),
final_url: url.to_string(),
status: 0,
latency_ms: 0,
server: None,
via_cloudflare: false,
cloudflare_fallback: false,
cloudflare_challenge: false,
body_snippet: String::new(),
error: Some("only http/https URLs are supported".into()),
reasons: vec!["Unsupported URL scheme".into()],
};
}
let client = match reqwest::Client::builder()
.timeout(Duration::from_millis(timeout_ms.max(1000)))
.redirect(reqwest::redirect::Policy::limited(5))
.user_agent("MeterboxHealthCheck/0.1 (+local)")
.build()
{
Ok(c) => c,
Err(e) => {
return ProbeResult {
ok: false,
url: url.to_string(),
final_url: url.to_string(),
status: 0,
latency_ms: 0,
server: None,
via_cloudflare: false,
cloudflare_fallback: false,
cloudflare_challenge: false,
body_snippet: String::new(),
error: Some(format!("client build failed: {e}")),
reasons: vec!["HTTP client error".into()],
};
}
};
let started = Instant::now();
let response = match client.get(parsed.clone()).send().await {
Ok(r) => r,
Err(e) => {
return ProbeResult {
ok: false,
url: url.to_string(),
final_url: url.to_string(),
status: 0,
latency_ms: started.elapsed().as_millis() as u64,
server: None,
via_cloudflare: false,
cloudflare_fallback: false,
cloudflare_challenge: false,
body_snippet: String::new(),
error: Some(format!("request failed: {e}")),
reasons: vec!["Request failed (network/TLS/DNS)".into()],
};
}
};
let latency_ms = started.elapsed().as_millis() as u64;
let status = response.status().as_u16();
let final_url = response.url().to_string();
let server = response
.headers()
.get("server")
.and_then(|v| v.to_str().ok())
.map(|s| s.to_string());
let cf_ray = response.headers().get("cf-ray").is_some();
let cf_mitigated = response.headers().get("cf-mitigated").is_some();
let via_cloudflare = cf_ray
|| server
.as_deref()
.map(|s| s.to_ascii_lowercase().contains("cloudflare"))
.unwrap_or(false);
let body = response.text().await.unwrap_or_default();
let body_lower = body.to_ascii_lowercase();
let snippet: String = body.chars().take(280).collect();
let mut cloudflare_fallback = false;
let mut cloudflare_challenge = false;
if CF_ORIGIN_ERRORS.contains(&status) {
cloudflare_fallback = true;
reasons.push(format!(
"Cloudflare origin error status {status} (site not reachable behind CF)"
));
}
if cf_mitigated {
cloudflare_challenge = true;
reasons.push("cf-mitigated header present (Cloudflare challenge/bot fight)".into());
}
for marker in CF_BODY_MARKERS {
if body_lower.contains(marker) {
if marker.contains("just a moment")
|| marker.contains("challenge")
|| marker.contains("checking if the site")
|| marker.contains("enable javascript")
{
cloudflare_challenge = true;
reasons.push(format!("Response body looks like a Cloudflare challenge ({marker})"));
} else {
cloudflare_fallback = true;
reasons.push(format!("Response body matches Cloudflare fallback/error page ({marker})"));
}
break;
}
}
// Soft 200 that is clearly CF interstitial
if status == 200 && via_cloudflare && (cloudflare_challenge || cloudflare_fallback) {
reasons.push("Got HTTP 200 but content is Cloudflare interstitial/fallback, not origin".into());
}
let mut ok = true;
if status != expected_status {
ok = false;
reasons.push(format!(
"Expected HTTP {expected_status}, got {status}"
));
}
if cloudflare_fallback || cloudflare_challenge {
ok = false;
}
if let Some(needle) = expect_body_contains {
let needle = needle.trim();
if !needle.is_empty() && !body.contains(needle) {
ok = false;
reasons.push(format!("Body does not contain expected text: {needle:?}"));
}
}
if ok {
reasons.push("Probe passed origin health checks".into());
}
ProbeResult {
ok,
url: url.to_string(),
final_url,
status,
latency_ms,
server,
via_cloudflare,
cloudflare_fallback,
cloudflare_challenge,
body_snippet: snippet,
error: None,
reasons,
}
}
+153
View File
@@ -0,0 +1,153 @@
//! Persistent app settings (SQLite key/value).
use anyhow::{Context, Result};
use rusqlite::params;
use serde::{Deserialize, Serialize};
use crate::db::Db;
pub const HEALTH_LOG_RETENTION_DAYS: i64 = 90;
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AppSettings {
/// When true, prune health check logs older than 3 months.
pub cleanup_logs_older_than_3_months: bool,
/// When true, host telemetry dashboard is readable without sign-in.
pub public_dashboard: bool,
/// Site IDs that are readable without sign-in.
pub public_site_ids: Vec<i64>,
}
impl Default for AppSettings {
fn default() -> Self {
Self {
cleanup_logs_older_than_3_months: false,
public_dashboard: false,
public_site_ids: Vec::new(),
}
}
}
#[derive(Debug, Clone, Serialize)]
pub struct PublicAccess {
pub dashboard: bool,
pub site_ids: Vec<i64>,
}
impl AppSettings {
pub fn public_access(&self) -> PublicAccess {
PublicAccess {
dashboard: self.public_dashboard,
site_ids: self.public_site_ids.clone(),
}
}
pub fn has_any_public(&self) -> bool {
self.public_dashboard || !self.public_site_ids.is_empty()
}
pub fn site_is_public(&self, id: i64) -> bool {
self.public_site_ids.contains(&id)
}
}
impl Db {
pub fn ensure_settings_table(&self) -> Result<()> {
let conn = self.conn.lock().unwrap();
conn.execute_batch(
"
CREATE TABLE IF NOT EXISTS settings (
key TEXT PRIMARY KEY,
value TEXT NOT NULL
);
",
)?;
Ok(())
}
fn get_setting_raw(&self, key: &str) -> Result<Option<String>> {
let conn = self.conn.lock().unwrap();
let v = conn
.query_row(
"SELECT value FROM settings WHERE key = ?1",
params![key],
|r| r.get::<_, String>(0),
)
.optional()?;
Ok(v)
}
fn set_setting_raw(&self, key: &str, value: &str) -> Result<()> {
let conn = self.conn.lock().unwrap();
conn.execute(
"INSERT INTO settings (key, value) VALUES (?1, ?2)
ON CONFLICT(key) DO UPDATE SET value = excluded.value",
params![key, value],
)
.context("set setting")?;
Ok(())
}
pub fn get_settings(&self) -> Result<AppSettings> {
let mut s = AppSettings::default();
if let Some(v) = self.get_setting_raw("cleanup_logs_older_than_3_months")? {
s.cleanup_logs_older_than_3_months = v == "true" || v == "1";
}
if let Some(v) = self.get_setting_raw("public_dashboard")? {
s.public_dashboard = v == "true" || v == "1";
}
if let Some(v) = self.get_setting_raw("public_site_ids")? {
s.public_site_ids = serde_json::from_str(&v).unwrap_or_default();
}
// Drop IDs for sites that no longer exist
let known: std::collections::HashSet<i64> = self
.list_health_sites()?
.into_iter()
.map(|site| site.id)
.collect();
s.public_site_ids.retain(|id| known.contains(id));
Ok(s)
}
pub fn save_settings(&self, settings: &AppSettings) -> Result<AppSettings> {
let mut clean = settings.clone();
clean.public_site_ids.sort_unstable();
clean.public_site_ids.dedup();
let known: std::collections::HashSet<i64> = self
.list_health_sites()?
.into_iter()
.map(|site| site.id)
.collect();
clean.public_site_ids.retain(|id| known.contains(id));
self.set_setting_raw(
"cleanup_logs_older_than_3_months",
if clean.cleanup_logs_older_than_3_months {
"true"
} else {
"false"
},
)?;
self.set_setting_raw(
"public_dashboard",
if clean.public_dashboard { "true" } else { "false" },
)?;
self.set_setting_raw(
"public_site_ids",
&serde_json::to_string(&clean.public_site_ids).unwrap_or_else(|_| "[]".into()),
)?;
Ok(clean)
}
pub fn prune_health_results_older_than(&self, days: i64) -> Result<usize> {
let cutoff = chrono::Utc::now().timestamp() - days.max(1) * 86400;
let conn = self.conn.lock().unwrap();
let n = conn.execute(
"DELETE FROM health_results WHERE checked_at < ?1",
params![cutoff],
)?;
Ok(n)
}
}
use rusqlite::OptionalExtension;