AkurAI-CRM / crates/crm-api/src/handlers/client_logs.rs

Latest change b347e5072854c171398c23ae7bfcd3dce6647f9e - fmt: rustfmt after request-logging change by agent

use crate::router_setup::{bad_request, internal_error, json_response};
use akurai_http::{Request, Response};
use akurai_json::Value;
use reqwest::blocking::Client;
use std::time::{Duration, SystemTime, UNIX_EPOCH};

const APP: &str = "akurai-crm";
const HOST_ID: i64 = 1;
const APPLICATION_ID: i64 = 8;
const MAX_LOGS: usize = 500;
const MAX_LINE_CHARS: usize = 16_384;

pub fn client_logs_route() -> Box<dyn Fn(&Request) -> Response + Send + Sync> {
    let client = Client::builder()
        .timeout(Duration::from_secs(5))
        .build()
        .expect("reqwest client configuration is valid");

    Box::new(move |req: &Request| {
        let body = match std::str::from_utf8(&req.body)
            .map_err(|e| format!("body not UTF-8: {e}"))
            .and_then(|body| akurai_json::parse(body).map_err(|e| format!("invalid JSON: {e}")))
            .and_then(normalize_logs)
        {
            Ok(body) => body,
            Err(error) => return bad_request(&error),
        };

        let config = match monitor_config() {
            Some(config) => config,
            None => return internal_error("monitor ingest is not configured"),
        };

        match client
            .post(&config.0)
            .bearer_auth(&config.1)
            .header("Content-Type", "application/json")
            .body(body.to_json())
            .send()
        {
            Ok(response) if response.status().is_success() => json_response(Value::Object(vec![(
                "inserted".into(),
                Value::Int(body.get("logs").map_or(0, |logs| match logs {
                    Value::Array(logs) => logs.len() as i64,
                    _ => 0,
                })),
            )])),
            Ok(_) => internal_error("monitor ingest rejected client logs"),
            Err(_) => internal_error("monitor ingest unavailable"),
        }
    })
}

fn monitor_config() -> Option<(String, String)> {
    let url = std::env::var("MONITOR_INGEST_URL").ok()?;
    let token = std::env::var("MONITOR_INGEST_TOKEN").ok()?;
    if url.is_empty() || token.is_empty() {
        return None;
    }
    let url = url.trim_end_matches('/');
    let url = if url.ends_with("/api/ingest") {
        url.to_owned()
    } else {
        format!("{url}/api/ingest")
    };
    Some((url, token))
}

fn normalize_logs(body: Value) -> Result<Value, String> {
    let logs = match body.get("logs") {
        Some(Value::Array(logs)) => logs,
        _ => return Err("expected { logs: [...] }".into()),
    };

    let now = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default()
        .as_secs() as i64;
    let logs = logs
        .iter()
        .take(MAX_LOGS)
        .map(|log| {
            let line = log
                .get("line")
                .map(string_value)
                .unwrap_or_default()
                .chars()
                .take(MAX_LINE_CHARS)
                .collect();
            let level = match log.get("level").and_then(Value::as_str) {
                Some("trace" | "debug" | "info" | "warn" | "error") => {
                    log.get("level").cloned().unwrap()
                }
                _ => Value::Str("info".into()),
            };
            let target = match log.get("target") {
                Some(Value::Str(target)) => Value::Str(target.clone()),
                _ => Value::Null,
            };
            let fields = log.get("fields").cloned().unwrap_or(Value::Null);
            let ts = log.get("ts").and_then(Value::as_i64).unwrap_or(now);

            Value::Object(vec![
                ("source".into(), Value::Str(format!("{APP}-client"))),
                ("line".into(), Value::Str(line)),
                ("level".into(), level),
                ("target".into(), target),
                ("origin".into(), Value::Str("client".into())),
                ("fields".into(), fields),
                ("host_id".into(), Value::Int(HOST_ID)),
                ("application_id".into(), Value::Int(APPLICATION_ID)),
                ("ts".into(), Value::Int(ts)),
            ])
        })
        .collect();

    Ok(Value::Object(vec![("logs".into(), Value::Array(logs))]))
}

fn string_value(value: &Value) -> String {
    match value {
        Value::Str(value) => value.clone(),
        value => value.to_json(),
    }
}

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

    #[test]
    fn client_metadata_cannot_override_proxy_attribution() {
        let body = akurai_json::parse(
            r#"{"logs":[{"source":"spoofed","line":"broken","level":"error","origin":"server","host_id":99,"application_id":99,"target":"/people","fields":{"detail":"x"},"ts":123}]}"#,
        )
        .unwrap();
        let normalized = normalize_logs(body).unwrap();
        let Value::Array(logs) = normalized.get("logs").unwrap() else {
            panic!("logs should be an array");
        };
        let log = &logs[0];

        assert_eq!(
            log.get("source").and_then(Value::as_str),
            Some("akurai-crm-client")
        );
        assert_eq!(log.get("origin").and_then(Value::as_str), Some("client"));
        assert_eq!(log.get("host_id").and_then(Value::as_i64), Some(HOST_ID));
        assert_eq!(
            log.get("application_id").and_then(Value::as_i64),
            Some(APPLICATION_ID)
        );
        assert_eq!(log.get("line").and_then(Value::as_str), Some("broken"));
    }

    #[test]
    fn proxy_forwards_browser_logs_with_server_attribution() {
        use std::io::{Read, Write};
        use std::net::TcpListener;

        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let endpoint = format!("http://{}", listener.local_addr().unwrap());
        let monitor = std::thread::spawn(move || {
            let (mut stream, _) = listener.accept().unwrap();
            let mut request = [0; 4096];
            let size = stream.read(&mut request).unwrap();
            stream
                .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 14\r\n\r\n{\"inserted\":1}")
                .unwrap();
            String::from_utf8_lossy(&request[..size]).into_owned()
        });

        std::env::set_var("MONITOR_INGEST_URL", endpoint);
        std::env::set_var("MONITOR_INGEST_TOKEN", "test-token");
        let mut request =
            Request::parse_head(b"POST /api/client-logs HTTP/1.1\r\nContent-Length: 45\r\n\r\n")
                .unwrap();
        request.body = br#"{"logs":[{"line":"browser failure","level":"error"}]}"#.to_vec();
        let response = client_logs_route()(&request);
        std::env::remove_var("MONITOR_INGEST_URL");
        std::env::remove_var("MONITOR_INGEST_TOKEN");

        assert_eq!(response.status, 200);
        let forwarded = monitor.join().unwrap();
        assert!(forwarded.starts_with("POST /api/ingest HTTP/1.1"));
        assert!(forwarded.contains("authorization: Bearer test-token"));
        assert!(forwarded.contains(r#""host_id":1"#));
        assert!(forwarded.contains(r#""application_id":8"#));
        assert!(forwarded.contains(r#""origin":"client""#));
    }
}