AkurAI-Framework / crates/cli/src/client_logs.rs

Latest change c6179dcf3abbf53272e4279e6740397ad2e75726 - AKURAI-FRAMEWORK-1..9 consolidate framework APIs by Ólafur Búi Ólafsson

//! Opt-in `POST /api/client-logs` proxy for browser telemetry.
//!
//! Apps enable it with `backend/client-logs.json`; Monitor credentials remain
//! server-only in `MONITOR_INGEST_URL` and `MONITOR_INGEST_TOKEN`.

use akurai_http::{Method, Request, Response};
use akurai_json::{parse, Value};
use std::fs;
use std::io::{Read as _, Write as _};
use std::net::{TcpStream, ToSocketAddrs};
use std::path::Path;
use std::time::Duration;

struct Config {
    app: String,
    host_id: i64,
    application_id: i64,
}

pub fn handle(root: &Path, req: &Request) -> Response {
    if req.method != Method::Post {
        return Response::new(405)
            .with_header("Allow", "POST")
            .with_text("POST only");
    }
    let Some(config) = load(root) else {
        return Response::not_found();
    };
    let logs = match payload(&req.body, &config) {
        Ok(logs) => logs,
        Err(message) => return Response::new(400).with_text(&message),
    };
    if logs.is_empty() {
        return super::json_api::response(
            200,
            &Value::Object(vec![("inserted".into(), Value::Int(0))]),
        );
    }

    let url = std::env::var("MONITOR_INGEST_URL").unwrap_or_default();
    let token = std::env::var("MONITOR_INGEST_TOKEN").unwrap_or_default();
    if url.is_empty() || token.is_empty() {
        return Response::new(503).with_text("Monitor ingest is not configured");
    }
    let body = Value::Object(vec![("logs".into(), Value::Array(logs.clone()))]).to_json();
    match post(&url, &token, &body) {
        Ok(()) => super::json_api::response(
            200,
            &Value::Object(vec![("inserted".into(), Value::Int(logs.len() as i64))]),
        ),
        Err(_) => Response::new(502).with_text("Monitor ingest failed"),
    }
}

fn load(root: &Path) -> Option<Config> {
    let text = fs::read_to_string(root.parent()?.join("backend/client-logs.json")).ok()?;
    let value = parse(&text).ok()?;
    Some(Config {
        app: value.get("app")?.as_str()?.to_string(),
        host_id: value.get("host_id")?.as_i64()?,
        application_id: value.get("application_id")?.as_i64()?,
    })
}

fn payload(body: &[u8], config: &Config) -> Result<Vec<Value>, String> {
    let value = super::json_api::parse_body(body).map_err(|message| {
        if message.starts_with("invalid JSON:") {
            "invalid JSON".to_string()
        } else {
            message
        }
    })?;
    let Some(Value::Array(items)) = value.get("logs") else {
        return Err("expected { logs: [...] }".to_string());
    };
    Ok(items
        .iter()
        .take(500)
        .filter_map(|item| client_log(item, config))
        .collect())
}

fn client_log(item: &Value, config: &Config) -> Option<Value> {
    let line = item.get("line")?.as_str()?;
    let level = item
        .get("level")
        .and_then(Value::as_str)
        .filter(|l| matches!(*l, "trace" | "debug" | "info" | "warn" | "error"))
        .unwrap_or("info");
    let target = item
        .get("target")
        .and_then(Value::as_str)
        .map(|target| Value::Str(target.chars().take(2048).collect()))
        .unwrap_or(Value::Null);
    let fields = item.get("fields").cloned().unwrap_or(Value::Null);
    let ts = item
        .get("ts")
        .and_then(Value::as_i64)
        .unwrap_or_else(now_secs);
    Some(Value::Object(vec![
        (
            "source".into(),
            Value::Str(format!("{}-client", config.app)),
        ),
        ("line".into(), Value::Str(line.chars().take(8192).collect())),
        ("level".into(), Value::Str(level.into())),
        ("target".into(), target),
        ("origin".into(), Value::Str("client".into())),
        ("fields".into(), fields),
        ("host_id".into(), Value::Int(config.host_id)),
        ("application_id".into(), Value::Int(config.application_id)),
        ("ts".into(), Value::Int(ts)),
    ]))
}

fn now_secs() -> i64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap_or_default()
        .as_secs() as i64
}

fn post(url: &str, token: &str, body: &str) -> Result<(), ()> {
    let authority_path = url
        .trim_end_matches('/')
        .strip_prefix("http://")
        .ok_or(())?;
    let (authority, base_path) = authority_path
        .split_once('/')
        .unwrap_or((authority_path, ""));
    let path = format!("{}/api/ingest", base_path.trim_end_matches('/'));
    let address = if authority.contains(':') {
        authority.to_string()
    } else {
        format!("{authority}:80")
    };
    let socket = address
        .to_socket_addrs()
        .map_err(|_| ())?
        .next()
        .ok_or(())?;
    let mut stream = TcpStream::connect_timeout(&socket, Duration::from_secs(2)).map_err(|_| ())?;
    stream
        .set_read_timeout(Some(Duration::from_secs(5)))
        .map_err(|_| ())?;
    stream
        .set_write_timeout(Some(Duration::from_secs(5)))
        .map_err(|_| ())?;
    if token.contains(['\r', '\n']) {
        return Err(());
    }
    write!(
        stream,
        "POST {path} HTTP/1.1\r\nHost: {authority}\r\nAuthorization: Bearer {token}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
        body.len()
    )
    .map_err(|_| ())?;
    stream.flush().map_err(|_| ())?;
    let mut response = [0; 64];
    let n = stream.read(&mut response).map_err(|_| ())?;
    let status = std::str::from_utf8(&response[..n]).map_err(|_| ())?;
    if status.starts_with("HTTP/1.1 2") || status.starts_with("HTTP/1.0 2") {
        Ok(())
    } else {
        Err(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::net::TcpListener;

    #[test]
    fn shapes_client_logs_and_overrides_identity() {
        let config = Config {
            app: "platform".into(),
            host_id: 1,
            application_id: 6,
        };
        let logs = payload(
            br#"{"logs":[{"line":"oops","level":"error","origin":"server"}]}"#,
            &config,
        )
        .unwrap();
        assert_eq!(logs.len(), 1);
        assert_eq!(
            logs[0].get("source").and_then(Value::as_str),
            Some("platform-client")
        );
        assert_eq!(
            logs[0].get("origin").and_then(Value::as_str),
            Some("client")
        );
        assert_eq!(logs[0].get("host_id").and_then(Value::as_i64), Some(1));
        assert_eq!(
            logs[0].get("application_id").and_then(Value::as_i64),
            Some(6)
        );
    }

    #[test]
    fn forwards_authorized_payload_to_monitor() {
        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let address = listener.local_addr().unwrap();
        let upstream = std::thread::spawn(move || {
            let (mut stream, _) = listener.accept().unwrap();
            let mut request = Vec::new();
            loop {
                let mut chunk = [0; 512];
                let n = stream.read(&mut chunk).unwrap();
                request.extend_from_slice(&chunk[..n]);
                if request
                    .windows(b"{\"logs\":[]}".len())
                    .any(|w| w == b"{\"logs\":[]}")
                {
                    break;
                }
            }
            let request = std::str::from_utf8(&request).unwrap();
            assert!(request.contains("Authorization: Bearer secret"));
            assert!(request.contains("{\"logs\":[]}"));
            stream
                .write_all(b"HTTP/1.1 201 Created\r\nContent-Length: 0\r\n\r\n")
                .unwrap();
        });
        post(&format!("http://{address}"), "secret", "{\"logs\":[]}").unwrap();
        upstream.join().unwrap();
    }
}