//! Usage, counted and read: the pages' hits, the requests, what the index
//! says the network did, and `knotless admin stats` showing it to an
//! operator, an account on the PDS. The index is filled by hand.

#[path = "../../knotless/tests/common/mod.rs"]
mod common;
use common::Sandbox;

use knotless::records::nsid;
use knotless::usage::{Report, Stats};
use knotless_mcp::http::Event;
use knotless_serve::gateway::Gateway;
use knotless_serve::index::Index;
use knotless_serve::logs::{self, Logs};
use knotless_serve::usage::{Usage, counted, mcp_events};
use knotless_serve::{App, router};
use serde_json::json;
use std::sync::Arc;
use std::time::{Duration, SystemTime};

/// `knotless <args>` as `who`, at the appview at `base`.
fn cli(who: &mut Sandbox, base: &str, args: &[&str]) -> (i32, String) {
    who.set("KNOTLESS_APPVIEW", base);
    who.cli(args, b"")
}

/// An appview, unserved yet, at a port of its own, as a gateway too: its
/// `did:web` is what its operator's service tokens are for. The account
/// `operator` signs in to runs it.
struct Appview {
    base: String,
    listener: tokio::net::TcpListener,
    operator: Sandbox,
}

impl Appview {
    fn new(rt: &tokio::runtime::Runtime) -> Appview {
        let listener = rt
            .block_on(tokio::net::TcpListener::bind("127.0.0.1:0"))
            .unwrap();
        let port = listener.local_addr().unwrap().port();
        let mut operator = Sandbox::new();
        operator.set("KNOTLESS_PASSWORD", "");
        let password = operator.account.password.clone();
        let handle = operator.account.handle.clone();
        let login = operator.cli(&["login", &handle, "--password-stdin"], password.as_bytes());
        assert_eq!(login.0, 0, "{}", login.1);
        Appview {
            base: format!("http://localhost:{port}"),
            listener,
            operator,
        }
    }

    fn app(&self, index: Option<Index>) -> Arc<App> {
        let mut server = Sandbox::with(self.operator.account.clone());
        server.set("KNOTLESS_DID_WEB_SCHEME", "http");
        server.set("KNOTLESS_ADMIN_DIDS", &self.operator.account.did);
        let ctx = common::ctx(&server);
        let gateway = Gateway::new(&self.base).unwrap();
        App::build(ctx, Duration::from_secs(60), Some(gateway), index)
    }

    /// Serves `app`, counted if it counts, and says where, and as whom.
    fn serve(self, rt: &tokio::runtime::Runtime, app: Arc<App>) -> (String, Sandbox) {
        let routes = match app.usage.clone() {
            Some(usage) => counted(router(app), usage),
            None => router(app),
        };
        let listener = self.listener;
        rt.spawn(async move { axum::serve(listener, routes).await.unwrap() });
        (self.base, self.operator)
    }
}

const BROWSER: &str = "Mozilla/5.0 (Macintosh) Safari/605";

#[test]
fn hits_requests_and_the_network() {
    let rt = tokio::runtime::Runtime::new().unwrap();
    let dir = tempfile::tempdir().unwrap();
    let now = humantime::format_rfc3339_seconds(SystemTime::now()).to_string();
    let index = Index::open(&dir.path().join("index.db")).unwrap();
    let repo = json!({"uri": "at://did:a/dev.knotless.repo/r", "key": "k1"});
    for (did, collection, rkey, value) in [
        (
            "did:a",
            nsid::REPO,
            "r",
            json!({"key": "k1", "createdAt": now}),
        ),
        (
            "did:a",
            nsid::CLAIM,
            "c",
            json!({"repo": "k1", "name": "refs/heads/main", "to": "abc", "createdAt": now}),
        ),
        (
            "did:b",
            nsid::PULL,
            "p",
            json!({"repo": repo, "createdAt": now}),
        ),
        (
            "did:b",
            nsid::ISSUE,
            "i",
            json!({"repo": repo, "createdAt": now}),
        ),
        (
            "did:b",
            nsid::COMMENT,
            "c",
            json!({"repo": repo, "createdAt": now}),
        ),
        (
            "did:b",
            nsid::STATE,
            "s",
            json!({"repo": repo, "createdAt": now}),
        ),
        // Dated past the days asked about: counted in the network alone.
        (
            "did:c",
            nsid::COMMENT,
            "later",
            json!({"repo": repo, "createdAt": "2999-01-01T00:00:00Z"}),
        ),
        (
            "did:run",
            nsid::CHECK,
            "x",
            json!({"repo": repo, "ref": "refs/heads/main", "commit": "abc", "startedAt": now}),
        ),
    ] {
        index.put(did, collection, rkey, &value).unwrap();
    }
    let appview = Appview::new(&rt);
    let mut app = appview.app(Some(index));
    let usage = Usage::open(&dir.path().join("usage.db")).unwrap();
    Arc::get_mut(&mut app).unwrap().usage = Some(Arc::new(usage));
    // What the handlers tell of: two accounts sign in to MCP, one goes on to
    // call a tool, twice; one delegates to the gateway and pushes.
    let mcp = mcp_events(app.usage.clone().unwrap());
    let ms = Duration::from_millis;
    for (name, subject, error, took, did) in [
        ("mcp.signin", "", None, ms(9), "did:x"),
        ("mcp.signin", "", None, ms(9), "did:y"),
        ("mcp.tool", "read_file", None, ms(100), "did:x"),
        (
            "mcp.tool",
            "read_file",
            Some("no such file"),
            ms(300),
            "did:x",
        ),
    ] {
        let did = Some(did);
        mcp(Event {
            name,
            subject,
            ok: error.is_none(),
            took,
            did,
            error,
        });
    }
    app.event("gateway.delegated", "", true, Duration::ZERO, Some("did:x"));
    app.event("gateway.push", "", true, ms(40), Some("did:x"));
    app.event("gateway.served", "", true, ms(40), Some("did:x"));
    let (base, mut operator) = appview.serve(&rt, app);

    let http = reqwest::Client::new();
    let hit_from = |from: Option<&str>, agent: &str, gpc: bool, body: &str| {
        let mut req = http
            .post(format!("{base}/_hit"))
            .header("user-agent", agent)
            .body(body.to_string());
        if let Some(from) = from {
            req = req.header("x-forwarded-for", from);
        }
        if gpc {
            req = req.header("sec-gpc", "1");
        }
        rt.block_on(async { req.send().await.unwrap().status().as_u16() })
    };
    let hit = |agent: &str, gpc: bool, body: &str| hit_from(Some("1.2.3.4"), agent, gpc, body);
    let page = r#"{"route": "/[authority]/[repo]", "referrer": "https://bsky.app/profile/x"}"#;
    assert_eq!(hit(BROWSER, false, page), 204);
    assert_eq!(hit(BROWSER, false, r#"{"route": "/"}"#), 204);
    // Not counted: one asking not to be, and a crawler.
    assert_eq!(hit(BROWSER, true, page), 204);
    assert_eq!(hit("Googlebot/2.1", false, page), 204);
    assert_eq!(hit("git/2.45.0", false, page), 204);
    assert_eq!(hit("curl/8.7.1", false, page), 204);
    // Visitors apart: one over IPv6, and one with no address at all.
    assert_eq!(hit_from(Some("2001:db8::1"), BROWSER, false, page), 204);
    assert_eq!(hit_from(None, BROWSER, false, page), 204);
    assert_eq!(hit(BROWSER, false, "nonsense"), 400);
    assert_eq!(hit(BROWSER, false, r#"{"route": "no slash"}"#), 400);
    let listed = rt.block_on(async {
        let url = format!("{base}/xrpc/dev.knotless.listRepos");
        let res = http
            .get(url)
            .header("user-agent", "knotless/0.1.0")
            .send()
            .await;
        res.unwrap().status().as_u16()
    });
    assert_eq!(listed, 200);
    // A crawler, counted by its agent.
    let crawled = rt.block_on(async {
        let url = format!("{base}/xrpc/dev.knotless.listRepos");
        let res = http
            .get(url)
            .header("user-agent", "curl/8.7.1")
            .send()
            .await;
        res.unwrap().status().as_u16()
    });
    assert_eq!(crawled, 200);
    let unauthorized = rt.block_on(async {
        http.get(format!("{base}/xrpc/dev.knotless.getUsage"))
            .send()
            .await
            .unwrap()
            .status()
            .as_u16()
    });
    assert_eq!(unauthorized, 401);

    let (code, json) = cli(
        &mut operator,
        &base,
        &["admin", "stats", "--days", "2", "--json"],
    );
    assert_eq!(code, 0, "{json}");
    let report: Report = serde_json::from_str(&json).unwrap();
    assert_eq!(report.days.len(), 2);
    let today = report.days.last().unwrap();
    assert_eq!(today.day, now[..10]);
    assert_eq!(
        (today.visitors, today.views, today.requests, today.errors),
        (3, 4, 3, 0)
    );
    assert_eq!(
        (
            today.repos,
            today.pushes,
            today.pulls,
            today.issues,
            today.comments,
            today.checks
        ),
        (1, 1, 1, 1, 1, 1)
    );
    assert_eq!(today.authors, 2);
    let network = report.network.unwrap();
    assert_eq!(
        (
            network.accounts,
            network.repos,
            network.authors_week,
            network.authors_month
        ),
        (3, 1, 3, 3)
    );
    // The listing, and the appview's DID document the CLI read to know
    // whom its token is for.
    let routes: Vec<(&str, &str, u64)> = report
        .routes
        .iter()
        .map(|r| (r.surface.as_str(), r.route.as_str(), r.requests))
        .collect();
    assert!(
        routes.contains(&("xrpc", "/xrpc/dev.knotless.listRepos", 2)),
        "{routes:?}"
    );
    assert!(
        routes.contains(&("other", "/.well-known/did.json", 1)),
        "{routes:?}"
    );
    assert_eq!(report.referrers[0].name, "bsky.app");
    assert_eq!(report.clients[0].name, "cli");
    let event = |name: &str, subject: &str| {
        let e = report
            .events
            .iter()
            .find(|e| e.name == name && e.subject == subject);
        e.unwrap_or_else(|| panic!("{name} {subject}: {:?}", report.events))
            .clone()
    };
    let tool = event("mcp.tool", "read_file");
    assert_eq!(
        (tool.calls, tool.failed, tool.mean_ms, tool.accounts),
        (2, 1, 200, 1)
    );
    assert_eq!(event("mcp.signin", "").accounts, 2);
    let failure = &report.failures[0];
    assert_eq!(
        (failure.subject.as_str(), failure.error.as_str(), failure.n),
        ("read_file", "no such file", 1)
    );
    assert_eq!(report.agents[0].name, "curl/8.7.1");
    assert!(event("cli.version", env!("CARGO_PKG_VERSION")).calls >= 1);
    let funnels: Vec<(&str, &str, u64, u64)> = report
        .funnels
        .iter()
        .map(|f| (f.from.as_str(), f.to.as_str(), f.started, f.reached))
        .collect();
    assert_eq!(
        funnels,
        [
            ("gateway.delegated", "gateway.served", 1, 1),
            ("mcp.signin", "mcp.tool", 2, 1)
        ]
    );

    // Anyone's: the days and the network, without what only an operator reads.
    let stats: Stats = rt.block_on(async {
        let res = http
            .get(format!("{base}/xrpc/dev.knotless.getStats"))
            .send()
            .await
            .unwrap();
        assert_eq!(
            res.headers()["cache-control"].to_str().unwrap(),
            "public, max-age=300"
        );
        res.json().await.unwrap()
    });
    assert_eq!(stats.days.len(), 30);
    let today = stats.days.last().unwrap();
    assert_eq!(today.day, now[..10]);
    assert_eq!(
        (today.visitors, today.views, today.pushes, today.authors),
        (3, 4, 1, 2)
    );
    assert_eq!(stats.network.unwrap().accounts, 3);

    let (code, text) = cli(&mut operator, &base, &["admin", "stats"]);
    assert_eq!(code, 0, "{text}");
    assert!(
        text.starts_with("network\t3 accounts, 1 repositories"),
        "{text}"
    );
    assert!(
        text.contains("route\txrpc\tGET /xrpc/dev.knotless.listRepos\t2 requests"),
        "{text}"
    );
    assert_ne!(
        cli(&mut operator, &base, &["admin", "stats", "--days", "x"]).0,
        0
    );
    assert_ne!(
        cli(&mut operator, &base, &["admin", "stats", "--weeks"]).0,
        0
    );
}

#[test]
fn a_server_counting_nothing() {
    let rt = tokio::runtime::Runtime::new().unwrap();
    let appview = Appview::new(&rt);
    let app = appview.app(None);
    let (base, mut operator) = appview.serve(&rt, app);
    let http = reqwest::Client::new();
    let hit = rt.block_on(async {
        let res = http
            .post(format!("{base}/_hit"))
            .body(r#"{"route": "/"}"#)
            .send()
            .await;
        res.unwrap().status().as_u16()
    });
    assert_eq!(hit, 404);
    // Still a report, of days with nothing in them.
    let (code, json) = cli(
        &mut operator,
        &base,
        &["admin", "stats", "--days", "1", "--json"],
    );
    assert_eq!(code, 0, "{json}");
    let report: Report = serde_json::from_str(&json).unwrap();
    assert_eq!((report.days.len(), report.network), (1, None));
    let stats: Stats = rt.block_on(async {
        let url = format!("{base}/xrpc/dev.knotless.getStats");
        http.get(url).send().await.unwrap().json().await.unwrap()
    });
    assert_eq!((stats.days.len(), stats.network), (30, None));
    // Nor are its logs kept: reading them fails.
    let (code, _) = cli(&mut operator, &base, &["admin", "logs", "--level", "warn"]);
    assert_ne!(code, 0);
    // And an answer that is not one says what it was.
    let (code, _) = cli(
        &mut operator,
        &format!("{base}/nowhere"),
        &["admin", "stats"],
    );
    assert_ne!(code, 0);
}

/// The lines serve logged, read back by an operator: a request's by its
/// trace, or the last of a level; asking for nothing is refused.
#[test]
fn logs_read_back() {
    let rt = tokio::runtime::Runtime::new().unwrap();
    let appview = Appview::new(&rt);
    let dir = tempfile::tempdir().unwrap();
    let logs = Arc::new(Logs::open(&dir.path().join("logs.db"), logs::KEEP).unwrap());
    for (level, message, trace) in [
        ("INFO", "request", "t1"),
        ("WARN", "outgoing", "t1"),
        ("INFO", "request", "t2"),
    ] {
        let line = json!({"timestamp": "2999-01-01T00:00:00Z", "level": level,
            "message": message, "trace_id": trace});
        logs.keep(format!("{line}\n").as_bytes());
    }
    let mut app = appview.app(None);
    Arc::get_mut(&mut app).unwrap().logs = Some(logs);
    let (base, mut operator) = appview.serve(&rt, app);
    let (code, text) = cli(&mut operator, &base, &["admin", "logs", "--trace", "t1"]);
    assert_eq!(code, 0, "{text}");
    let lines: Vec<serde_json::Value> = text
        .lines()
        .map(|l| serde_json::from_str(l).unwrap())
        .collect();
    let messages: Vec<&str> = lines
        .iter()
        .map(|l| l["message"].as_str().unwrap())
        .collect();
    assert_eq!(messages, ["request", "outgoing"]);
    let (code, text) = cli(
        &mut operator,
        &base,
        &["admin", "logs", "--level", "warn", "--limit", "5"],
    );
    assert_eq!(code, 0, "{text}");
    assert!(text.contains("\"outgoing\""), "{text}");
    assert!(!text.contains("\"t2\""), "{text}");
    // Asking for nothing is refused.
    let (code, _) = cli(&mut operator, &base, &["admin", "logs"]);
    assert_ne!(code, 0);
    assert_ne!(
        cli(&mut operator, &base, &["admin", "logs", "--since"]).0,
        0
    );
    assert_ne!(
        cli(&mut operator, &base, &["admin", "logs", "--trace"]).0,
        0
    );
}