Initial scaffold: web API (Bun/Elysia) + monitor (Rust/Tokio)

apps/monitor/Cargo.toml

@@ -0,0 +1,16 @@
+[package]
+name = "pingql-monitor"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+tokio        = { version = "1", features = ["full"] }
+reqwest      = { version = "0.12", features = ["json", "rustls-tls"], default-features = false }
+serde        = { version = "1", features = ["derive"] }
+serde_json   = "1"
+scraper      = "0.21"    # CSS selector / HTML parsing
+futures      = "0.3"
+regex        = "1"
+anyhow       = "1"
+tracing      = "0.1"
+tracing-subscriber = { version = "0.3", features = ["env-filter"] }

apps/monitor/src/main.rs

@@ -0,0 +1,35 @@
+mod query;
+mod runner;
+mod types;
+
+use anyhow::Result;
+use std::env;
+use tokio::time::{sleep, Duration};
+use tracing::{error, info};
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    tracing_subscriber::fmt()
+        .with_env_filter(env::var("RUST_LOG").unwrap_or_else(|_| "info".into()))
+        .init();
+
+    let coordinator_url = env::var("COORDINATOR_URL")
+        .unwrap_or_else(|_| "http://localhost:3000".into());
+    let monitor_token = env::var("MONITOR_TOKEN")
+        .expect("MONITOR_TOKEN must be set");
+
+    info!("PingQL monitor starting, coordinator: {coordinator_url}");
+
+    let client = reqwest::Client::builder()
+        .timeout(Duration::from_secs(30))
+        .user_agent("PingQL-Monitor/0.1")
+        .build()?;
+
+    loop {
+        match runner::fetch_and_run(&client, &coordinator_url, &monitor_token).await {
+            Ok(n) => info!("Ran {n} checks"),
+            Err(e) => error!("Check cycle failed: {e}"),
+        }
+        sleep(Duration::from_secs(10)).await;
+    }
+}

apps/monitor/src/query.rs

@@ -0,0 +1,141 @@
+/// PingQL query evaluation against a check response.
+///
+/// Query shape (MongoDB-inspired):
+///
+/// Simple equality:
+///   { "status": 200 }
+///
+/// Operators:
+///   { "status": { "$eq": 200 } }
+///   { "status": { "$ne": 500 } }
+///   { "status": { "$gte": 200, "$lt": 300 } }
+///   { "body": { "$contains": "healthy" } }
+///   { "body": { "$regex": "ok|healthy" } }
+///
+/// CSS selector (HTML parsing):
+///   { "$select": "span.status", "$eq": "operational" }
+///
+/// Logical:
+///   { "$and": [ { "status": 200 }, { "body": { "$contains": "ok" } } ] }
+///   { "$or":  [ { "status": 200 }, { "status": 204 } ] }
+
+use anyhow::{bail, Result};
+use regex::Regex;
+use scraper::{Html, Selector};
+use serde_json::Value;
+
+pub struct Response {
+    pub status: u16,
+    pub body: String,
+    pub headers: std::collections::HashMap<String, String>,
+}
+
+/// Returns true if `query` matches `response`. No query = always up.
+pub fn evaluate(query: &Value, response: &Response) -> Result<bool> {
+    match query {
+        Value::Object(map) => {
+            // $and / $or
+            if let Some(and) = map.get("$and") {
+                let Value::Array(clauses) = and else { bail!("$and expects array") };
+                return Ok(clauses.iter().all(|c| evaluate(c, response).unwrap_or(false)));
+            }
+            if let Some(or) = map.get("$or") {
+                let Value::Array(clauses) = or else { bail!("$or expects array") };
+                return Ok(clauses.iter().any(|c| evaluate(c, response).unwrap_or(false)));
+            }
+            // CSS selector shorthand: { "$select": "...", "$eq": "..." }
+            if let Some(sel) = map.get("$select") {
+                let sel_str = sel.as_str().unwrap_or("");
+                let selected = css_select(&response.body, sel_str);
+                if let Some(op_val) = map.get("$eq") {
+                    return Ok(selected.as_deref() == op_val.as_str());
+                }
+                if let Some(op_val) = map.get("$contains") {
+                    let needle = op_val.as_str().unwrap_or("");
+                    return Ok(selected.map(|s| s.contains(needle)).unwrap_or(false));
+                }
+                return Ok(selected.is_some());
+            }
+            // Field-level checks
+            for (field, condition) in map {
+                let field_val = resolve_field(field, response);
+                if !eval_condition(condition, &field_val, response)? {
+                    return Ok(false);
+                }
+            }
+            Ok(true)
+        }
+        _ => bail!("Query must be an object"),
+    }
+}
+
+fn resolve_field(field: &str, r: &Response) -> Value {
+    match field {
+        "status" | "status_code" => Value::Number(r.status.into()),
+        "body" => Value::String(r.body.clone()),
+        f if f.starts_with("headers.") => {
+            let key = f.trim_start_matches("headers.").to_lowercase();
+            r.headers.get(&key)
+                .map(|v| Value::String(v.clone()))
+                .unwrap_or(Value::Null)
+        }
+        _ => Value::Null,
+    }
+}
+
+fn eval_condition(condition: &Value, field_val: &Value, response: &Response) -> Result<bool> {
+    match condition {
+        // Shorthand: { "status": 200 }
+        Value::Number(n) => Ok(field_val.as_f64() == n.as_f64()),
+        Value::String(s) => Ok(field_val.as_str() == Some(s.as_str())),
+        Value::Bool(b) => Ok(field_val.as_bool() == Some(*b)),
+        Value::Object(ops) => {
+            for (op, val) in ops {
+                let ok = match op.as_str() {
+                    "$eq"  => field_val == val,
+                    "$ne"  => field_val != val,
+                    "$gt"  => cmp_num(field_val, val, |a,b| a > b),
+                    "$gte" => cmp_num(field_val, val, |a,b| a >= b),
+                    "$lt"  => cmp_num(field_val, val, |a,b| a < b),
+                    "$lte" => cmp_num(field_val, val, |a,b| a <= b),
+                    "$contains" => {
+                        let needle = val.as_str().unwrap_or("");
+                        field_val.as_str().map(|s| s.contains(needle)).unwrap_or(false)
+                    }
+                    "$regex" => {
+                        let pattern = val.as_str().unwrap_or("");
+                        let re = Regex::new(pattern).unwrap_or_else(|_| Regex::new("$^").unwrap());
+                        field_val.as_str().map(|s| re.is_match(s)).unwrap_or(false)
+                    }
+                    "$select" => {
+                        // Nested: { "body": { "$select": "css", "$eq": "val" } }
+                        let sel_str = val.as_str().unwrap_or("");
+                        let selected = css_select(&response.body, sel_str);
+                        if let Some(eq_val) = ops.get("$eq") {
+                            selected.as_deref() == eq_val.as_str()
+                        } else {
+                            selected.is_some()
+                        }
+                    }
+                    _ => true, // unknown op — skip
+                };
+                if !ok { return Ok(false); }
+            }
+            Ok(true)
+        }
+        _ => Ok(true),
+    }
+}
+
+fn cmp_num(a: &Value, b: &Value, f: impl Fn(f64, f64) -> bool) -> bool {
+    match (a.as_f64(), b.as_f64()) {
+        (Some(x), Some(y)) => f(x, y),
+        _ => false,
+    }
+}
+
+fn css_select(html: &str, selector: &str) -> Option<String> {
+    let doc = Html::parse_document(html);
+    let sel = Selector::parse(selector).ok()?;
+    doc.select(&sel).next().map(|el| el.text().collect::<String>().trim().to_string())
+}

apps/monitor/src/runner.rs

@@ -0,0 +1,115 @@
+use crate::query::{self, Response};
+use crate::types::{CheckResult, Monitor};
+use anyhow::Result;
+use serde_json::{json, Value};
+use std::collections::HashMap;
+use std::time::Instant;
+use tracing::{debug, warn};
+
+/// Fetch due monitors from coordinator, run them, post results back.
+pub async fn fetch_and_run(
+    client: &reqwest::Client,
+    coordinator_url: &str,
+    token: &str,
+) -> Result<usize> {
+    // Fetch due monitors
+    let monitors: Vec<Monitor> = client
+        .get(format!("{coordinator_url}/internal/due"))
+        .header("x-monitor-token", token)
+        .send()
+        .await?
+        .json()
+        .await?;
+
+    let n = monitors.len();
+    if n == 0 { return Ok(0); }
+
+    // Run all checks concurrently
+    let tasks: Vec<_> = monitors.into_iter().map(|monitor| {
+        let client = client.clone();
+        let coordinator_url = coordinator_url.to_string();
+        let token = token.to_string();
+        tokio::spawn(async move {
+            let result = run_check(&client, &monitor).await;
+            if let Err(e) = post_result(&client, &coordinator_url, &token, result).await {
+                warn!("Failed to post result for {}: {e}", monitor.id);
+            }
+        })
+    }).collect();
+
+    futures::future::join_all(tasks).await;
+    Ok(n)
+}
+
+async fn run_check(client: &reqwest::Client, monitor: &Monitor) -> CheckResult {
+    let start = Instant::now();
+
+    let result = client.get(&monitor.url).send().await;
+    let latency_ms = start.elapsed().as_millis() as u64;
+
+    match result {
+        Err(e) => CheckResult {
+            monitor_id: monitor.id.clone(),
+            status_code: None,
+            latency_ms: Some(latency_ms),
+            up: false,
+            error: Some(e.to_string()),
+            meta: None,
+        },
+        Ok(resp) => {
+            let status = resp.status().as_u16();
+            let headers: HashMap<String, String> = resp.headers().iter()
+                .filter_map(|(k, v)| Some((k.to_string(), v.to_str().ok()?.to_string())))
+                .collect();
+
+            let body = resp.text().await.unwrap_or_default();
+
+            // Evaluate query if present
+            let (up, query_error) = if let Some(q) = &monitor.query {
+                let response = Response { status, body: body.clone(), headers: headers.clone() };
+                match query::evaluate(q, &response) {
+                    Ok(result) => (result, None),
+                    Err(e) => {
+                        warn!("Query error for {}: {e}", monitor.id);
+                        // Fall back to status-based up/down
+                        (status < 400, Some(e.to_string()))
+                    }
+                }
+            } else {
+                // Default: up if 2xx/3xx
+                (status < 400, None)
+            };
+
+            let meta = json!({
+                "headers": headers,
+                "body_preview": &body[..body.len().min(500)],
+            });
+
+            debug!("{} → {status} {latency_ms}ms up={up}", monitor.url);
+
+            CheckResult {
+                monitor_id: monitor.id.clone(),
+                status_code: Some(status),
+                latency_ms: Some(latency_ms),
+                up,
+                error: query_error,
+                meta: Some(meta),
+            }
+        }
+    }
+}
+
+async fn post_result(
+    client: &reqwest::Client,
+    coordinator_url: &str,
+    token: &str,
+    result: CheckResult,
+) -> Result<()> {
+    client
+        .post(format!("{coordinator_url}/checks/ingest"))
+        .header("x-monitor-token", token)
+        .json(&result)
+        .send()
+        .await?;
+    Ok(())
+}

apps/monitor/src/types.rs

@@ -0,0 +1,20 @@
+use serde::{Deserialize, Serialize};
+use serde_json::Value;
+
+#[derive(Debug, Deserialize)]
+pub struct Monitor {
+    pub id: String,
+    pub url: String,
+    pub interval_s: i64,
+    pub query: Option<Value>,
+}
+
+#[derive(Debug, Serialize)]
+pub struct CheckResult {
+    pub monitor_id: String,
+    pub status_code: Option<u16>,
+    pub latency_ms: Option<u64>,
+    pub up: bool,
+    pub error: Option<String>,
+    pub meta: Option<Value>,
+}