//! How this server is used, for its operator: counted, and who only for
//! a month.
//!
//! - Every request but the health checks is counted by the hour, by its
//!   surface (the pages, the API, git, xrpc, MCP, the
//!   installer), the route axum matched (its pattern, not the path), its
//!   method, whether it failed, how long it took, and the kind of client
//!   that asked, from its `User-Agent`.
//! - The pages, once drawn in a browser, send `/_hit` the route that drew
//!   them and the site that sent the visitor, if another. Each visitor is
//!   counted once a day by a hash of their address, the last byte of it
//!   dropped, and their `User-Agent`, salted with a key made for the day
//!   and deleted once it is over: no address is kept, and no one can be
//!   followed from one day to the next. A browser asking not to be
//!   (`Sec-GPC: 1`) is not counted.
//! - What the handlers saw is counted by the hour, by its name and subject
//!   (a tool, a repository, a version), never by who. Where the asker is
//!   signed in (the pages' logins, MCP's sign-ins and tool calls, the
//!   gateway's delegations, pushes and commits), their DID is kept with
//!   the event's name and the day, to count how many accounts did each
//!   and the funnels. It is deleted once that day is no longer one of the
//!   last [`ACCOUNT_DAYS`] (30), today the last: at the first flush on the
//!   30th day after it (UTC), so no DID is kept longer than 30 days.
//!   Before they go, each funnel's day is kept as two counts with no one
//!   in them: how many accounts started it, and how many of those went on.
//! - What tells an operator why, by the hour, and for those same 30
//!   days: each of the index's statements slow enough to log (its SQL as
//!   written, no value it was given), why an event failed (what it
//!   answered, which can name a repository or an account), and the
//!   `User-Agent` of each crawler or script, never a browser's.
//!
//! Requests are counted in memory and written now and then
//! ([`Usage::flush`]); the salt is in the database, so every serve
//! running that day, as two do for a moment in each deploy, hashes alike.
//! `dev.knotless.getUsage`, for an operator, answers it with what the index says
//! the network did, as [`knotless::usage::Report`].

use crate::{App, AppError};
use axum::Json;
use axum::body::Bytes;
use axum::extract::{MatchedPath, Query, Request, State};
use axum::http::{HeaderMap, StatusCode, header};
use axum::middleware::Next;
use axum::response::{IntoResponse, Response};
use knotless::telemetry::{self, Fields, Level};
use knotless::usage::{
    Count, Day, Event, Failure, Funnel, Hour, Network, Report, Route, Statement, Stats,
};
use rand_core::{OsRng, RngCore};
use rusqlite::{Connection, params};
use serde::Deserialize;
use sha2::{Digest, Sha256};
use std::collections::{BTreeMap, BTreeSet, HashMap};
use std::net::IpAddr;
use std::path::Path;
use std::sync::{Arc, Mutex, RwLock, Weak};
use std::time::{Duration, Instant, SystemTime};

/// How often the counted requests are written.
pub const FLUSH: Duration = Duration::from_secs(15);

/// The most days `dev.knotless.getUsage` answers for.
const DAYS_MAX: u32 = 400;

/// How many days, today the last, the accounts in the events are kept:
/// the month `getUsage` counts each event's accounts over at most, as the
/// index counts the network's (`authorsMonth`).
pub const ACCOUNT_DAYS: u64 = 30;

/// The longest route or referrer a page may send.
const NAME_MAX: usize = 200;

/// The most of a failure's error, or a crawler's `User-Agent`, kept.
const ERROR_MAX: usize = 300;
const AGENT_MAX: usize = 200;

/// The most failures, or crawlers, kept apart in an hour; past it, the
/// rest are counted together as [`OVERFLOW`], so a crawler naming itself
/// anew each time, or a statement built to its input, cannot grow the
/// table without end.
const KEPT_PER_HOUR: i64 = 500;
const OVERFLOW: &str = "(others)";

pub struct Usage {
    db: Mutex<Connection>,
    /// Requests counted since the last flush.
    pending: Mutex<HashMap<Key, Tally>>,
    /// Events counted since the last flush.
    events: Mutex<HashMap<EventKey, Tally>>,
    /// The accounts seen in each, by day, in order: an account's first
    /// day of an event is written before its next.
    actors: Mutex<BTreeSet<(String, &'static str, String)>>,
    /// Slow statements counted since the last flush, by hour and SQL.
    statements: Mutex<HashMap<(String, String), Slow>>,
    /// Failures counted since the last flush.
    failures: Mutex<HashMap<FailureKey, u64>>,
    /// Crawlers' requests counted since the last flush, by hour and agent.
    agents: Mutex<HashMap<(String, String), u64>>,
}

/// A failure's hour, event name, subject and error.
type FailureKey = (String, &'static str, String, String);

/// A statement's slow runs: how many, their milliseconds, and the most.
#[derive(Default, Debug, PartialEq)]
struct Slow {
    n: u64,
    ms: u64,
    max: u64,
}

/// The usage slow statements are counted in: the index's statements are
/// watched by a function SQLite calls, which can take no state.
static SINK: RwLock<Weak<Usage>> = RwLock::new(Weak::new());

/// Has the index's slow statements counted in `usage`.
pub fn sink(usage: &Arc<Usage>) {
    *SINK.write().unwrap() = Arc::downgrade(usage);
}

/// Counts the slow statement `sql`, which took `took`, in the usage
/// [`sink`] named, if it named one.
pub fn slow_statement(sql: &str, took: Duration) {
    let usage = SINK.read().unwrap().upgrade();
    if let Some(usage) = usage {
        usage.statement(sql, took);
    }
}

/// The funnels the report follows: of the accounts that first did the one
/// in the days asked, how many went on to the other.
const FUNNELS: [(&str, &str); 2] = [
    ("gateway.delegated", "gateway.served"),
    ("mcp.signin", "mcp.tool"),
];

#[derive(Hash, PartialEq, Eq, Clone, Debug)]
struct Key {
    hour: String,
    surface: &'static str,
    route: String,
    method: String,
    /// The status's class: 2 for 2xx, and so on.
    class: u16,
    client: &'static str,
}

/// An event's hour, name, subject, and whether it worked.
type EventKey = (String, &'static str, String, bool);

#[derive(Default, Debug, PartialEq)]
struct Tally {
    n: u64,
    ms: u64,
    over_half: u64,
    over_two: u64,
}

impl Usage {
    pub fn open(path: &Path) -> rusqlite::Result<Usage> {
        let db = Connection::open(path)?;
        db.busy_timeout(Duration::from_secs(5))?;
        let made = db.execute_batch(
            "PRAGMA journal_mode = WAL;
             PRAGMA synchronous = NORMAL;
             CREATE TABLE IF NOT EXISTS salt (day TEXT PRIMARY KEY, value BLOB NOT NULL);
             CREATE TABLE IF NOT EXISTS visitor (
               day TEXT NOT NULL, hash BLOB NOT NULL, PRIMARY KEY (day, hash)) WITHOUT ROWID;
             CREATE TABLE IF NOT EXISTS visitors (day TEXT PRIMARY KEY, n INTEGER NOT NULL);
             CREATE TABLE IF NOT EXISTS request (
               hour TEXT NOT NULL, surface TEXT NOT NULL, route TEXT NOT NULL,
               method TEXT NOT NULL, class INTEGER NOT NULL, client TEXT NOT NULL,
               n INTEGER NOT NULL, ms INTEGER NOT NULL,
               over_half INTEGER NOT NULL, over_two INTEGER NOT NULL,
               PRIMARY KEY (hour, surface, route, method, class, client)) WITHOUT ROWID;
             CREATE TABLE IF NOT EXISTS view (
               hour TEXT NOT NULL, route TEXT NOT NULL, referrer TEXT NOT NULL,
               n INTEGER NOT NULL, PRIMARY KEY (hour, route, referrer)) WITHOUT ROWID;
             CREATE TABLE IF NOT EXISTS event (
               hour TEXT NOT NULL, name TEXT NOT NULL, subject TEXT NOT NULL,
               ok INTEGER NOT NULL, n INTEGER NOT NULL, ms INTEGER NOT NULL,
               PRIMARY KEY (hour, name, subject, ok)) WITHOUT ROWID;
             CREATE TABLE IF NOT EXISTS account (
               day TEXT NOT NULL, name TEXT NOT NULL, did TEXT NOT NULL,
               first INTEGER NOT NULL,
               PRIMARY KEY (day, name, did)) WITHOUT ROWID;
             CREATE INDEX IF NOT EXISTS account_did ON account (name, did, day);
             CREATE TABLE IF NOT EXISTS cohort (
               day TEXT NOT NULL, start TEXT NOT NULL, goal TEXT NOT NULL,
               started INTEGER NOT NULL, reached INTEGER NOT NULL,
               PRIMARY KEY (day, start, goal)) WITHOUT ROWID;
             -- Kept before as `actor`, for good: moved, and folded at the
             -- next flush.
             CREATE TABLE IF NOT EXISTS statement (
               hour TEXT NOT NULL, sql TEXT NOT NULL, n INTEGER NOT NULL,
               ms INTEGER NOT NULL, max INTEGER NOT NULL,
               PRIMARY KEY (hour, sql)) WITHOUT ROWID;
             CREATE TABLE IF NOT EXISTS failure (
               hour TEXT NOT NULL, name TEXT NOT NULL, subject TEXT NOT NULL,
               error TEXT NOT NULL, n INTEGER NOT NULL,
               PRIMARY KEY (hour, name, subject, error)) WITHOUT ROWID;
             CREATE TABLE IF NOT EXISTS agent (
               hour TEXT NOT NULL, agent TEXT NOT NULL, n INTEGER NOT NULL,
               PRIMARY KEY (hour, agent)) WITHOUT ROWID;
             CREATE TABLE IF NOT EXISTS actor (
               day TEXT NOT NULL, name TEXT NOT NULL, did TEXT NOT NULL,
               PRIMARY KEY (day, name, did)) WITHOUT ROWID;
             INSERT OR IGNORE INTO account SELECT day, name, did,
               NOT EXISTS (SELECT 1 FROM actor b
                 WHERE b.name = a.name AND b.did = a.did AND b.day < a.day)
               FROM actor a;
             DROP TABLE actor;",
        );
        made?;
        Ok(Usage {
            db: Mutex::new(db),
            pending: Mutex::default(),
            events: Mutex::default(),
            actors: Mutex::default(),
            statements: Mutex::default(),
            failures: Mutex::default(),
            agents: Mutex::default(),
        })
    }

    /// Counts the index's statement `sql`, which took `took`: one slow
    /// enough to log.
    pub fn statement(&self, sql: &str, took: Duration) {
        let key = (hour(SystemTime::now()), telemetry::sql_text(sql));
        let ms = took.as_millis() as u64;
        let mut statements = self.statements.lock().unwrap();
        let slow = statements.entry(key).or_default();
        slow.n += 1;
        slow.ms += ms;
        slow.max = slow.max.max(ms);
    }

    /// Counts why the event `name`, of `subject`, failed: `error`, what it
    /// answered.
    pub fn failed(&self, name: &'static str, subject: &str, error: &str) {
        let error: String = fingerprint(error).chars().take(ERROR_MAX).collect();
        let key = (hour(SystemTime::now()), name, subject.to_string(), error);
        *self.failures.lock().unwrap().entry(key).or_default() += 1;
    }

    /// Counts an event a handler saw: `name`, of `subject` (a tool, a
    /// repository, a version, or nothing), whether it worked, how long it
    /// took, and the account, if signed in.
    pub fn event(
        &self,
        name: &'static str,
        subject: &str,
        ok: bool,
        took: Duration,
        did: Option<&str>,
    ) {
        let now = SystemTime::now();
        let key = (hour(now), name, subject.to_string(), ok);
        let mut events = self.events.lock().unwrap();
        let tally = events.entry(key).or_default();
        tally.n += 1;
        tally.ms += took.as_millis() as u64;
        drop(events);
        let actor = did.map(|did| (day(now), name, did.to_string()));
        self.actors.lock().unwrap().extend(actor);
    }

    /// Counts a request from the CLI at `version`.
    fn cli(&self, version: Option<&str>) {
        if let Some(v) = version {
            self.event("cli.version", v, true, Duration::ZERO, None);
        }
    }

    /// Counts a request to `route` (or, matching none, `path`) that took
    /// `took` and answered `status`.
    fn tally(
        &self,
        route: Option<&str>,
        path: &str,
        method: &str,
        status: u16,
        agent: &str,
        took: Duration,
    ) {
        let Some((surface, route)) = surface(route, path) else {
            return;
        };
        let client = client(agent);
        if client == "bot" {
            let agent: String = agent.chars().take(AGENT_MAX).collect();
            let key = (hour(SystemTime::now()), agent);
            *self.agents.lock().unwrap().entry(key).or_default() += 1;
        }
        // A long poll's wait is what it is for, not what anyone waited on.
        let slow = u64::from(!crate::telemetry::waits(Some(route.as_str())));
        let key = Key {
            hour: hour(SystemTime::now()),
            surface,
            route,
            method: method.to_string(),
            class: status / 100,
            client,
        };
        let ms = took.as_millis() as u64;
        let mut pending = self.pending.lock().unwrap();
        let tally = pending.entry(key).or_default();
        tally.n += 1;
        tally.ms += ms;
        tally.over_half += u64::from(ms > 500) * slow;
        tally.over_two += u64::from(ms > 2000) * slow;
    }

    /// Writes the requests counted; folds the days gone by into their
    /// count of visitors, deleting their hashes and their salts; and folds
    /// the accounts' days past [`ACCOUNT_DAYS`] into the funnels' counts,
    /// deleting them.
    pub fn flush(&self) -> rusqlite::Result<()> {
        let pending = std::mem::take(&mut *self.pending.lock().unwrap());
        let mut db = self.db.lock().unwrap();
        let tx = db.transaction()?;
        let mut stmt = tx.prepare_cached(
            "INSERT INTO request VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)
             ON CONFLICT DO UPDATE SET n = n + excluded.n, ms = ms + excluded.ms,
               over_half = over_half + excluded.over_half, over_two = over_two + excluded.over_two",
        )?;
        for (k, t) in &pending {
            stmt.execute(params![
                k.hour,
                k.surface,
                k.route,
                k.method,
                k.class,
                k.client,
                t.n,
                t.ms,
                t.over_half,
                t.over_two
            ])?;
        }
        drop(stmt);
        let events = std::mem::take(&mut *self.events.lock().unwrap());
        let mut stmt = tx.prepare_cached(
            "INSERT INTO event VALUES (?1, ?2, ?3, ?4, ?5, ?6)
             ON CONFLICT DO UPDATE SET n = n + excluded.n, ms = ms + excluded.ms",
        )?;
        for ((hour, name, subject, ok), t) in &events {
            stmt.execute(params![hour, name, subject, ok, t.n, t.ms])?;
        }
        drop(stmt);
        let actors = std::mem::take(&mut *self.actors.lock().unwrap());
        // An account's first of an event in the days kept starts a funnel.
        let mut stmt = tx.prepare_cached(
            "INSERT OR IGNORE INTO account SELECT ?1, ?2, ?3, NOT EXISTS (SELECT 1 FROM account
               WHERE name = ?2 AND did = ?3 AND day < ?1)",
        )?;
        for (day, name, did) in &actors {
            stmt.execute(params![day, name, did])?;
        }
        drop(stmt);
        // Two serves may both fold: each adds what it finds, and the
        // second finds the hashes, and the accounts, already gone.
        let now = SystemTime::now();
        let today = day(now);
        let kept = day(now - Duration::from_secs(86_400 * (ACCOUNT_DAYS - 1)));
        let mut stmt = tx.prepare_cached(
            "INSERT INTO cohort SELECT day, ?1, ?2, count(*), sum(EXISTS (SELECT 1 FROM account b
               WHERE b.name = ?2 AND b.did = a.did AND b.day >= a.day))
             FROM account a WHERE name = ?1 AND first AND day < ?3 GROUP BY day
             ON CONFLICT DO UPDATE SET started = started + excluded.started,
               reached = reached + excluded.reached",
        )?;
        for (from, to) in FUNNELS {
            stmt.execute(params![from, to, kept])?;
        }
        drop(stmt);
        tx.execute("DELETE FROM account WHERE day < ?1", [&kept])?;
        self.flush_detail(&tx, &kept)?;
        tx.execute(
            "INSERT INTO visitors SELECT day, count(*) FROM visitor WHERE day < ?1 GROUP BY day
             ON CONFLICT DO UPDATE SET n = n + excluded.n",
            [&today],
        )?;
        tx.execute("DELETE FROM visitor WHERE day < ?1", [&today])?;
        tx.execute("DELETE FROM salt WHERE day < ?1", [&today])?;
        tx.commit()
    }

    /// Writes the statements, failures and crawlers counted, and deletes
    /// those of the days before `kept`, as the accounts are.
    fn flush_detail(&self, tx: &rusqlite::Transaction, kept: &str) -> rusqlite::Result<()> {
        // A new statement, failure or crawler past the hour's
        // [`KEPT_PER_HOUR`] is counted as [`OVERFLOW`]; one already kept
        // goes on being. Statements are as written, so few, but one built
        // to its input, as an `IN` list, is new at each size.
        let statements = std::mem::take(&mut *self.statements.lock().unwrap());
        let mut stmt = tx.prepare_cached(
            "INSERT INTO statement SELECT ?1,
               CASE WHEN EXISTS (SELECT 1 FROM statement WHERE hour = ?1 AND sql = ?2)
                 OR (SELECT count(*) FROM statement WHERE hour = ?1) < ?6
               THEN ?2 ELSE ?7 END, ?3, ?4, ?5
             ON CONFLICT DO UPDATE SET n = n + excluded.n, ms = ms + excluded.ms,
               max = max(max, excluded.max)",
        )?;
        for ((hour, sql), s) in &statements {
            let p = params![hour, sql, s.n, s.ms, s.max, KEPT_PER_HOUR, OVERFLOW];
            stmt.execute(p)?;
        }
        let failures = std::mem::take(&mut *self.failures.lock().unwrap());
        let mut stmt = tx.prepare_cached(
            "INSERT INTO failure SELECT ?1, ?2, ?3,
               CASE WHEN EXISTS (SELECT 1 FROM failure
                   WHERE hour = ?1 AND name = ?2 AND subject = ?3 AND error = ?4)
                 OR (SELECT count(*) FROM failure WHERE hour = ?1) < ?6
               THEN ?4 ELSE ?7 END, ?5
             ON CONFLICT DO UPDATE SET n = n + excluded.n",
        )?;
        for ((hour, name, subject, error), n) in &failures {
            let p = params![hour, name, subject, error, n, KEPT_PER_HOUR, OVERFLOW];
            stmt.execute(p)?;
        }
        let agents = std::mem::take(&mut *self.agents.lock().unwrap());
        let mut stmt = tx.prepare_cached(
            "INSERT INTO agent SELECT ?1,
               CASE WHEN EXISTS (SELECT 1 FROM agent WHERE hour = ?1 AND agent = ?2)
                 OR (SELECT count(*) FROM agent WHERE hour = ?1) < ?4
               THEN ?2 ELSE ?5 END, ?3
             ON CONFLICT DO UPDATE SET n = n + excluded.n",
        )?;
        for ((hour, agent), n) in &agents {
            stmt.execute(params![hour, agent, n, KEPT_PER_HOUR, OVERFLOW])?;
        }
        // An hour, `YYYY-MM-DDTHH`, sorts after its day and before the next.
        for table in ["statement", "failure", "agent"] {
            tx.execute(&format!("DELETE FROM {table} WHERE hour < ?1"), [kept])?;
        }
        Ok(())
    }

    /// Counts a page drawn in a browser: the route that drew it, the host
    /// that sent the visitor there, and the visitor, by `ip` and `agent`.
    fn view(
        &self,
        ip: Option<IpAddr>,
        agent: &str,
        route: &str,
        referrer: &str,
    ) -> rusqlite::Result<()> {
        let now = SystemTime::now();
        let today = day(now);
        let db = self.db.lock().unwrap();
        let mut fresh = [0u8; 16];
        OsRng.fill_bytes(&mut fresh);
        db.execute(
            "INSERT OR IGNORE INTO salt VALUES (?1, ?2)",
            params![today, fresh],
        )?;
        let salt = db.query_row("SELECT value FROM salt WHERE day = ?1", [&today], |r| {
            r.get::<_, Vec<u8>>(0)
        });
        let salt = salt?;
        let hash = visitor(&salt, ip, agent);
        db.execute(
            "INSERT OR IGNORE INTO visitor VALUES (?1, ?2)",
            params![today, hash],
        )?;
        db.execute(
            "INSERT INTO view VALUES (?1, ?2, ?3, 1) ON CONFLICT DO UPDATE SET n = n + 1",
            params![hour(now), route, referrer],
        )?;
        Ok(())
    }

    /// What was counted over the `days` days to today, into `report`.
    /// Fills `days` with each one's requests, failures, views and visitors.
    fn days(&self, days: &mut BTreeMap<String, Day>) -> rusqlite::Result<()> {
        self.flush()?;
        let since = days.keys().next().cloned().unwrap_or_default();
        let db = self.db.lock().unwrap();
        let mut stmt = db.prepare(
            "SELECT substr(hour, 1, 10), sum(n), sum(CASE WHEN class = 5 THEN n ELSE 0 END)
             FROM request WHERE hour >= ?1 GROUP BY 1",
        )?;
        for row in stmt.query_map([&since], counts)? {
            let (d, requests, errors) = row?;
            days.entry(d)
                .and_modify(|d| (d.requests, d.errors) = (requests, errors));
        }
        let mut stmt = db.prepare(
            "SELECT substr(hour, 1, 10), sum(n), 0 FROM view WHERE hour >= ?1 GROUP BY 1",
        )?;
        for row in stmt.query_map([&since], counts)? {
            let (d, views, _) = row?;
            days.entry(d).and_modify(|d| d.views = views);
        }
        // Today's, still hashes; the days before, folded.
        let mut stmt = db.prepare(
            "SELECT day, n, 0 FROM visitors WHERE day >= ?1
             UNION ALL SELECT day, count(*), 0 FROM visitor WHERE day >= ?1 GROUP BY day",
        )?;
        for row in stmt.query_map([&since], counts)? {
            let (d, visitors, _) = row?;
            days.entry(d).and_modify(|d| d.visitors += visitors);
        }
        Ok(())
    }

    /// Fills `days`, as [`Usage::days`] does, and the rest of `report`.
    fn read(&self, days: &mut BTreeMap<String, Day>, report: &mut Report) -> rusqlite::Result<()> {
        self.days(days)?;
        let since = days.keys().next().cloned().unwrap_or_default();
        let db = self.db.lock().unwrap();
        // The busiest hundred, and with them every write and every route
        // that was slow, however few its requests: what someone waited on.
        let mut stmt = db.prepare(
            "SELECT surface, method, route, n, errors, mean, over_half, over_two FROM (
               SELECT surface, method, route, sum(n) AS n,
                 sum(CASE WHEN class = 5 THEN n ELSE 0 END) AS errors,
                 sum(ms) / sum(n) AS mean, sum(over_half) AS over_half,
                 sum(over_two) AS over_two,
                 row_number() OVER (ORDER BY sum(n) DESC, route) AS rank
               FROM request WHERE hour >= ?1 GROUP BY 1, 2, 3)
             WHERE rank <= 100 OR method = 'POST' OR over_two > 0
             ORDER BY n DESC, route",
        )?;
        report.routes = stmt
            .query_map([&since], route)?
            .collect::<rusqlite::Result<_>>()?;
        let mut stmt = db.prepare(
            "SELECT hour, sum(n), sum(CASE WHEN class = 5 THEN n ELSE 0 END), sum(over_two)
             FROM request WHERE hour >= ?1 GROUP BY 1 ORDER BY 1",
        )?;
        report.hours = stmt
            .query_map([&since], hour_row)?
            .collect::<rusqlite::Result<_>>()?;
        let tops = [
            (
                "SELECT route, sum(n) FROM view WHERE hour >= ?1 GROUP BY 1 ORDER BY 2 DESC, 1 LIMIT 50",
                &mut report.pages,
            ),
            (
                "SELECT referrer, sum(n) FROM view WHERE hour >= ?1 AND referrer != ''
                 GROUP BY 1 ORDER BY 2 DESC, 1 LIMIT 50",
                &mut report.referrers,
            ),
            (
                "SELECT client, sum(n) FROM request WHERE hour >= ?1 GROUP BY 1 ORDER BY 2 DESC, 1",
                &mut report.clients,
            ),
        ];
        for (sql, into) in tops {
            let mut stmt = db.prepare(sql)?;
            *into = stmt
                .query_map([&since], count)?
                .collect::<rusqlite::Result<_>>()?;
        }
        let mut stmt = db.prepare(
            "SELECT e.name, e.subject, e.n, e.failed, e.ms / e.n,
               (SELECT count(DISTINCT did) FROM account a WHERE a.name = e.name AND a.day >= ?1)
             FROM (SELECT name, subject, sum(n) n, sum(CASE WHEN ok THEN 0 ELSE n END) failed,
                     sum(ms) ms FROM event WHERE hour >= ?1 GROUP BY 1, 2) e
             ORDER BY 1, 3 DESC, 2 LIMIT 200",
        )?;
        report.events = stmt
            .query_map([&since], event)?
            .collect::<rusqlite::Result<_>>()?;
        // An account's first day of `from` in the days kept, and whether
        // it did `to` then or after; and the days since forgotten, counted.
        let mut stmt = db.prepare(
            "SELECT ?1, ?2,
               count(*) + (SELECT coalesce(sum(started), 0) FROM cohort
                 WHERE start = ?1 AND goal = ?2 AND day >= ?3),
               coalesce(sum(EXISTS (SELECT 1 FROM account b
                 WHERE b.name = ?2 AND b.did = a.did AND b.day >= a.day)), 0)
                 + (SELECT coalesce(sum(reached), 0) FROM cohort
                   WHERE start = ?1 AND goal = ?2 AND day >= ?3)
             FROM account a WHERE name = ?1 AND first AND day >= ?3",
        )?;
        for (from, to) in FUNNELS {
            let funnel = stmt.query_row(params![from, to, since], funnel)?;
            report.funnels.push(funnel);
        }
        let mut stmt = db.prepare(
            "SELECT sql, sum(n), sum(ms) / sum(n), max(max) FROM statement WHERE hour >= ?1
             GROUP BY 1 ORDER BY sum(ms) DESC, 1 LIMIT 50",
        )?;
        report.statements = stmt
            .query_map([&since], statement)?
            .collect::<rusqlite::Result<_>>()?;
        let mut stmt = db.prepare(
            "SELECT name, subject, error, sum(n) FROM failure WHERE hour >= ?1
             GROUP BY 1, 2, 3 ORDER BY 4 DESC, 1, 2, 3 LIMIT 100",
        )?;
        report.failures = stmt
            .query_map([&since], failure)?
            .collect::<rusqlite::Result<_>>()?;
        let mut stmt = db.prepare(
            "SELECT agent, sum(n) FROM agent WHERE hour >= ?1 GROUP BY 1 ORDER BY 2 DESC, 1 LIMIT 20",
        )?;
        report.agents = stmt
            .query_map([&since], count)?
            .collect::<rusqlite::Result<_>>()?;
        Ok(())
    }
}

/// `error` with what differs from one failure to the next put as what
/// it is, so like failures count together: a DID as
/// `<did>`, an `at://` URI as `<uri>`, a hash or other hex of seven
/// digits or more as `<hex>`, and a number as `<n>`.
fn fingerprint(error: &str) -> String {
    let words = error.split_whitespace().map(|word| {
        let edge = |c: char| !c.is_alphanumeric() && !matches!(c, ':' | '/' | '.' | '_' | '-');
        let core = word.trim_matches(edge).trim_end_matches(['.', ':']);
        let is = |pred: fn(char) -> bool| !core.is_empty() && core.chars().all(pred);
        let kind = match () {
            _ if core.starts_with("did:") => "<did>",
            _ if core.starts_with("at://") => "<uri>",
            _ if core.len() >= 7 && is(|c| c.is_ascii_hexdigit()) && !is(char::is_alphabetic) => {
                "<hex>"
            }
            _ if is(|c| c.is_ascii_digit()) => "<n>",
            _ => return word.to_string(),
        };
        word.replacen(core, kind, 1)
    });
    words.collect::<Vec<_>>().join(" ")
}

fn statement(row: &rusqlite::Row) -> rusqlite::Result<Statement> {
    Ok(Statement {
        sql: row.get(0)?,
        calls: row.get(1)?,
        mean_ms: row.get(2)?,
        max_ms: row.get(3)?,
    })
}

fn failure(row: &rusqlite::Row) -> rusqlite::Result<Failure> {
    Ok(Failure {
        name: row.get(0)?,
        subject: row.get(1)?,
        error: row.get(2)?,
        n: row.get(3)?,
    })
}

fn counts(row: &rusqlite::Row) -> rusqlite::Result<(String, u64, u64)> {
    Ok((row.get(0)?, row.get(1)?, row.get(2)?))
}

fn count(row: &rusqlite::Row) -> rusqlite::Result<Count> {
    Ok(Count {
        name: row.get(0)?,
        n: row.get(1)?,
    })
}

fn event(row: &rusqlite::Row) -> rusqlite::Result<Event> {
    Ok(Event {
        name: row.get(0)?,
        subject: row.get(1)?,
        calls: row.get(2)?,
        failed: row.get(3)?,
        mean_ms: row.get(4)?,
        accounts: row.get(5)?,
    })
}

fn funnel(row: &rusqlite::Row) -> rusqlite::Result<Funnel> {
    Ok(Funnel {
        from: row.get(0)?,
        to: row.get(1)?,
        started: row.get(2)?,
        reached: row.get(3)?,
    })
}

fn route(row: &rusqlite::Row) -> rusqlite::Result<Route> {
    Ok(Route {
        surface: row.get(0)?,
        method: row.get(1)?,
        route: row.get(2)?,
        requests: row.get(3)?,
        errors: row.get(4)?,
        mean_ms: row.get(5)?,
        over_half: row.get(6)?,
        over_two: row.get(7)?,
    })
}

fn hour_row(row: &rusqlite::Row) -> rusqlite::Result<Hour> {
    Ok(Hour {
        hour: row.get(0)?,
        requests: row.get(1)?,
        errors: row.get(2)?,
        over_two: row.get(3)?,
    })
}

/// `YYYY-MM-DD` of `t`, in UTC.
fn day(t: SystemTime) -> String {
    humantime::format_rfc3339_seconds(t).to_string()[..10].to_string()
}

/// `YYYY-MM-DDTHH` of `t`, in UTC.
fn hour(t: SystemTime) -> String {
    humantime::format_rfc3339_seconds(t).to_string()[..13].to_string()
}

/// The surface a request was to, and the route it is counted under: the
/// pattern it matched. Matching none are the releases, a nested service,
/// and the pages, their assets apart from the rest. None for the health
/// checks and the counting itself.
fn surface(route: Option<&str>, path: &str) -> Option<(&'static str, String)> {
    let Some(route) = route else {
        let (surface, route) = match path {
            p if p.starts_with("/releases/") => ("install", "/releases/*"),
            p if p.starts_with("/_app/") => ("pages", "/_app/*"),
            _ => ("pages", "*"),
        };
        return Some((surface, route.into()));
    };
    let surface = match route {
        "/healthz" | "/_status" | "/_hit" | "/xrpc/dev.knotless.getUsage" => return None,
        r if r.starts_with("/api/") => "api",
        r if r.ends_with("/info/refs") || r.ends_with("/git-upload-pack") => "git",
        r if r.starts_with("/xrpc/") => "xrpc",
        r if r.starts_with("/mcp") || r.starts_with("/.well-known/oauth") => "mcp",
        "/install.sh" => "install",
        r if r.starts_with("/v2/") => "oci",
        // A repository's releases, counted as a directory's are.
        r if r.starts_with("/releases/") => return Some(("install", "/releases/*".into())),
        _ => "other",
    };
    Some((surface, route.to_string()))
}

/// The kind of client `agent` names. The pages' own queries, as they draw
/// a page someone asked for, are `pages`, counted apart from every client.
fn client(agent: &str) -> &'static str {
    let lower = agent.to_ascii_lowercase();
    if lower.starts_with("git/") {
        "git"
    } else if lower.starts_with("knotless/") {
        "cli"
    } else if lower == "knotless-pages" {
        "pages"
    } else if bot(&lower) {
        "bot"
    } else if lower.starts_with("mozilla/") {
        "browser"
    } else {
        "other"
    }
}

/// Whether `agent`, in lower case, says it is a crawler or a script, as
/// most do, or is a browser no one is at.
fn bot(agent: &str) -> bool {
    const WORDS: [&str; 17] = [
        "bot",
        // Google's crawler for all but search, which names itself so.
        "googleother",
        "crawl",
        "spider",
        "slurp",
        "fetch",
        "scrape",
        "preview",
        "headless",
        "lighthouse",
        "python",
        "curl",
        "wget",
        "go-http",
        "okhttp",
        "axios",
        "node",
    ];
    agent.is_empty() || WORDS.iter().any(|w| agent.contains(w))
}

/// The visitor's address, as the proxy in front saw it: the last address
/// in `X-Forwarded-For`, which it wrote, rather than any the client did.
fn address(headers: &HeaderMap) -> Option<IpAddr> {
    let forwarded = headers.get("x-forwarded-for")?.to_str().ok()?;
    forwarded.rsplit(',').next()?.trim().parse().ok()
}

/// A visitor's hash for the day: of their address without its last byte
/// (its last 80 bits, for IPv6), and their `User-Agent`, salted.
fn visitor(salt: &[u8], ip: Option<IpAddr>, agent: &str) -> Vec<u8> {
    let ip = match ip {
        Some(IpAddr::V4(v4)) => {
            let [a, b, c, _] = v4.octets();
            vec![a, b, c]
        }
        Some(IpAddr::V6(v6)) => v6.octets()[..6].to_vec(),
        None => vec![],
    };
    let mut hash = Sha256::new();
    hash.update(salt);
    hash.update(&ip);
    hash.update([0xff]);
    hash.update(agent.as_bytes());
    hash.finalize()[..16].to_vec()
}

fn agent(headers: &HeaderMap) -> &str {
    let agent = headers
        .get(header::USER_AGENT)
        .and_then(|v| v.to_str().ok());
    agent.unwrap_or_default()
}

/// The middleware counting each request.
pub async fn count_requests(usage: Arc<Usage>, req: Request, next: Next) -> Response {
    let route = req
        .extensions()
        .get::<MatchedPath>()
        .map(|m| m.as_str().to_string());
    let path = req.uri().path().to_string();
    let method = req.method().to_string();
    let agent = agent(req.headers()).to_string();
    // The CLI's version, which says how far each release has reached.
    let version = agent
        .strip_prefix("knotless/")
        .map(|v| v.split(' ').next().unwrap_or(v));
    usage.cli(version);
    let started = Instant::now();
    let res = next.run(req).await;
    let status = res.status().as_u16();
    usage.tally(
        route.as_deref(),
        &path,
        &method,
        status,
        &agent,
        started.elapsed(),
    );
    res
}

/// `routes`, each request to them counted.
pub fn counted(routes: axum::Router, usage: Arc<Usage>) -> axum::Router {
    routes.layer(axum::middleware::from_fn(move |req, next| {
        count_requests(usage.clone(), req, next)
    }))
}

/// Writes the counted requests every `every`, and the first time at once,
/// which folds the days a stopped server left.
pub async fn flushing(usage: Arc<Usage>, every: Duration) {
    loop {
        wrote(usage.flush());
        tokio::time::sleep(every).await;
    }
}

/// Logs, at warn, why the counts were not written, if they were not:
/// they are written with the next.
fn wrote(done: rusqlite::Result<()>) {
    if let Err(err) = done {
        let fields = Fields {
            error: Some(format!("{err:#}")),
            ..Fields::default()
        };
        telemetry::note(Level::WARN, "usage not written", &fields);
    }
}

impl App {
    /// Counts an event, if this server counts usage: see [`Usage::event`].
    pub fn event(
        &self,
        name: &'static str,
        subject: &str,
        ok: bool,
        took: Duration,
        did: Option<&str>,
    ) {
        if let Some(usage) = &self.usage {
            usage.event(name, subject, ok, took, did);
        }
    }
}

/// MCP's sign-ins and tool calls, counted, and why those that failed did.
pub fn mcp_events(usage: Arc<Usage>) -> knotless_mcp::http::Events {
    Arc::new(move |e| {
        usage.event(e.name, e.subject, e.ok, e.took, e.did);
        if let Some(error) = e.error {
            usage.failed(e.name, e.subject, error);
        }
    })
}

/// A page drawn, as the page tells of it.
#[derive(Deserialize)]
struct Hit {
    /// The route that drew it, as SvelteKit names it: `/[authority]/[repo]`.
    route: String,
    /// The page the visitor came from, if on another site.
    #[serde(default)]
    referrer: String,
}

fn usage(app: &App) -> Result<&Usage, AppError> {
    let usage = app.usage.as_deref();
    usage.ok_or_else(|| AppError(StatusCode::NOT_FOUND, "this server counts no usage".into()))
}

/// `POST /_hit`: a page drawn in a browser.
pub async fn hit(
    State(app): State<Arc<App>>,
    headers: HeaderMap,
    body: Bytes,
) -> Result<StatusCode, AppError> {
    let usage = usage(&app)?;
    let agent = agent(&headers);
    let private = headers.get("sec-gpc").is_some_and(|v| v == "1");
    if private || client(agent) != "browser" {
        return Ok(StatusCode::NO_CONTENT);
    }
    let bad = |why: &str| AppError(StatusCode::BAD_REQUEST, why.into());
    let hit: Hit = serde_json::from_slice(&body).map_err(|e| bad(&e.to_string()))?;
    if !hit.route.starts_with('/') || hit.route.len() > NAME_MAX || hit.referrer.len() > NAME_MAX {
        return Err(bad("a hit is a route, as /[authority], and a referrer"));
    }
    let referrer = host(&hit.referrer);
    let viewed = usage.view(address(&headers), agent, &hit.route, referrer);
    viewed
        .map_err(anyhow::Error::from)
        .map_err(AppError::internal)?;
    Ok(StatusCode::NO_CONTENT)
}

/// The host of `url`, or nothing.
fn host(url: &str) -> &str {
    let rest = url.split_once("://").map_or("", |(_, rest)| rest);
    rest.split(['/', '?', '#', ':']).next().unwrap_or_default()
}

#[derive(Deserialize)]
pub struct UsageParams {
    days: Option<u32>,
}

/// `dev.knotless.getUsage?days=<n>`, for an operator: the last
/// `n` days, 14 if not given.
pub async fn report(
    State(app): State<Arc<App>>,
    headers: HeaderMap,
    Query(p): Query<UsageParams>,
) -> Result<Json<Report>, AppError> {
    crate::moderation::admin(&app, &headers, knotless::moderation::GET_USAGE).await?;
    let n = p.days.unwrap_or(14).clamp(1, DAYS_MAX);
    let mut report = Report::default();
    let (days, network) = counted_days(&app, n, Some(&mut report))?;
    report.days = days;
    report.network = network;
    Ok(Json(report))
}

/// The last `n` days, today (UTC) last, each with what usage and the
/// index counted in it, and the network as the index has it now. Given
/// `report`, the rest of what usage keeps goes in it.
fn counted_days(
    app: &App,
    n: u32,
    report: Option<&mut Report>,
) -> Result<(Vec<Day>, Option<Network>), AppError> {
    let now = SystemTime::now();
    let ago = |k: u32| day(now - Duration::from_secs(86_400 * u64::from(k)));
    let mut days: BTreeMap<String, Day> = (0..n)
        .rev()
        .map(|k| {
            (
                ago(k),
                Day {
                    day: ago(k),
                    ..Day::default()
                },
            )
        })
        .collect();
    if let Some(usage) = app.usage.as_deref() {
        let read = match report {
            Some(report) => usage.read(&mut days, report),
            None => usage.days(&mut days),
        };
        read.map_err(anyhow::Error::from)
            .map_err(AppError::internal)?;
    }
    let mut network = None;
    if let Some(index) = app.index.as_deref() {
        let found = index.activity(&mut days, &ago(6), &ago(29));
        let found = found
            .map_err(anyhow::Error::from)
            .map_err(AppError::internal)?;
        network = Some(found);
    }
    Ok((days.into_values().collect(), network))
}

/// How many days `dev.knotless.getStats` shows.
const STATS_DAYS: u32 = 30;

/// `dev.knotless.getStats`, for anyone: the network, and the last 30 days
/// of what it wrote and of the pages' visitors, as the public `/stats`
/// page shows them. Kept for five minutes.
pub async fn stats(State(app): State<Arc<App>>) -> Result<Response, AppError> {
    let (days, network) = counted_days(&app, STATS_DAYS, None)?;
    let stats = Stats {
        days: days.into_iter().map(Into::into).collect(),
        network,
    };
    let cache = [(header::CACHE_CONTROL, "public, max-age=300")];
    Ok((cache, Json(stats)).into_response())
}

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

    #[test]
    fn surfaces() {
        let s = |route: Option<&str>, path: &str| surface(route, path);
        assert_eq!(s(None, "/a.test/r"), Some(("pages", "*".into())));
        assert_eq!(
            s(None, "/_app/immutable/x.js"),
            Some(("pages", "/_app/*".into()))
        );
        assert_eq!(s(Some("/healthz"), "/healthz"), None);
        for (route, surface) in [
            ("/api/{authority}/{repo}/tree", "api"),
            ("/{authority}/{repo}/info/refs", "git"),
            ("/{authority}/{repo}/git-upload-pack", "git"),
            ("/xrpc/dev.knotless.getFeed", "xrpc"),
            ("/mcp", "mcp"),
            ("/.well-known/oauth-authorization-server", "mcp"),
            ("/install.sh", "install"),
            ("/v2/{*rest}", "oci"),
            ("/card.png", "other"),
        ] {
            assert_eq!(s(Some(route), ""), Some((surface, route.into())), "{route}");
        }
        assert_eq!(
            s(Some("/releases/{version}/{file}"), "/releases/0.1.0/a"),
            Some(("install", "/releases/*".into()))
        );
        assert_eq!(
            s(None, "/releases/latest"),
            Some(("install", "/releases/*".into()))
        );
    }

    #[test]
    fn clients() {
        for (agent, kind) in [
            ("git/2.47.0", "git"),
            ("knotless/0.1.0", "cli"),
            ("knotless-pages", "pages"),
            ("node", "bot"),
            ("Mozilla/5.0 (compatible; Googlebot/2.1)", "bot"),
            ("Mozilla/5.0 (compatible; GoogleOther)", "bot"),
            ("Mozilla/5.0 (X11; Linux x86_64) HeadlessChrome/120", "bot"),
            ("curl/8.0", "bot"),
            ("", "bot"),
            ("Mozilla/5.0 (Macintosh) Safari/605", "browser"),
            ("Something/1", "other"),
        ] {
            assert_eq!(client(agent), kind, "{agent}");
        }
    }

    #[test]
    fn addresses_and_hashes() {
        let mut headers = HeaderMap::new();
        assert_eq!(address(&headers), None);
        headers.insert("x-forwarded-for", "6.6.6.6, 1.2.3.4".parse().unwrap());
        let ip = address(&headers);
        assert_eq!(ip, Some("1.2.3.4".parse().unwrap()));
        // The last byte, or 80 bits, does not tell visitors apart.
        let near = Some("1.2.3.200".parse().unwrap());
        assert_eq!(visitor(b"s", ip, "a"), visitor(b"s", near, "a"));
        assert_ne!(visitor(b"s", ip, "a"), visitor(b"t", ip, "a"));
        assert_ne!(visitor(b"s", ip, "a"), visitor(b"s", ip, "b"));
        let v6 = |s: &str| Some(s.parse().unwrap());
        assert_eq!(
            visitor(b"s", v6("2001:db8:1::1"), "a"),
            visitor(b"s", v6("2001:db8:1:ff::9"), "a")
        );
        assert_eq!(visitor(b"s", None, "a").len(), 16);
        assert_eq!(host("https://bsky.app/profile/x?y"), "bsky.app");
        assert_eq!(host("http://localhost:5173/"), "localhost");
        assert_eq!(host(""), "");
    }

    /// The index's slow statements reach the usage named, and none when
    /// it is gone.
    #[test]
    fn sinks_slow_statements() {
        let dir = tempfile::tempdir().unwrap();
        let usage = Arc::new(Usage::open(&dir.path().join("usage.db")).unwrap());
        sink(&usage);
        let sql = "SELECT 'sinks_slow_statements'";
        slow_statement(sql, Duration::from_millis(150));
        let counted = usage
            .statements
            .lock()
            .unwrap()
            .keys()
            .any(|(_, s)| s == sql);
        assert!(counted);
        drop(usage);
        slow_statement(sql, Duration::from_millis(150));
    }

    #[test]
    fn fingerprints() {
        for (error, print) in [
            (
                "did:plc:abc is not a member of at://did:plc:x/dev.knotless.repo/r.",
                "<did> is not a member of <uri>.",
            ),
            ("no commit 5bf7b54f503f (yet)", "no commit <hex> (yet)"),
            ("took 1500 ms, (3) times", "took <n> ms, (<n>) times"),
            ("deadbeef  and   facade", "deadbeef and facade"),
            ("", ""),
        ] {
            assert_eq!(fingerprint(error), print, "{error}");
        }
    }

    /// Past an hour's kept failures and crawlers, the new ones count
    /// together; one already kept goes on being.
    #[test]
    fn caps_each_hour() {
        let dir = tempfile::tempdir().unwrap();
        let usage = Usage::open(&dir.path().join("usage.db")).unwrap();
        let fast = Duration::ZERO;
        for i in 0..KEPT_PER_HOUR {
            let [a, b] = [i / 26, i % 26].map(|d| char::from(b'a' + d as u8));
            usage.failed("mcp.tool", "x", &format!("failed as {a}{b}"));
            usage.tally(None, "/", "GET", 200, &format!("bot{i}"), fast);
            usage.statement(&format!("SELECT {a}{b}"), Duration::from_millis(100));
        }
        usage.flush().unwrap();
        usage.failed("mcp.tool", "x", " failed as aa ");
        usage.failed("mcp.tool", "x", "new");
        usage.failed("mcp.tool", "x", "newer");
        usage.tally(None, "/", "GET", 200, "bot0", fast);
        usage.tally(None, "/", "GET", 200, "botnew", fast);
        usage.statement("SELECT aa", Duration::from_millis(300));
        usage.statement("SELECT new", Duration::from_millis(200));
        usage.flush().unwrap();
        let db = usage.db.lock().unwrap();
        let n = |sql: &str| db.query_row(sql, [], |r| r.get::<_, i64>(0)).unwrap();
        assert_eq!(n("SELECT count(*) FROM failure"), KEPT_PER_HOUR + 1);
        assert_eq!(n("SELECT n FROM failure WHERE error = '(others)'"), 2);
        assert_eq!(n("SELECT n FROM failure WHERE error = 'failed as aa'"), 2);
        assert_eq!(n("SELECT count(*) FROM agent"), KEPT_PER_HOUR + 1);
        assert_eq!(n("SELECT n FROM agent WHERE agent = '(others)'"), 1);
        assert_eq!(n("SELECT n FROM agent WHERE agent = 'bot0'"), 2);
        assert_eq!(n("SELECT count(*) FROM statement"), KEPT_PER_HOUR + 1);
        let row = |sql: &str| {
            let q = "SELECT n, ms, max FROM statement WHERE sql = ?1";
            db.query_row(q, [sql], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)))
                .unwrap()
        };
        assert_eq!(row("SELECT aa"), (2i64, 400i64, 300i64));
        assert_eq!(row("(others)"), (1, 200, 200));
    }

    #[test]
    fn says_what_was_not_written() {
        wrote(Ok(()));
        wrote(Err(rusqlite::Error::InvalidQuery));
    }

    #[test]
    fn forgets_accounts_after_a_month() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("usage.db");
        let now = SystemTime::now();
        let ago = |k: u64| day(now - Duration::from_secs(86_400 * k));
        // As an older serve left it: who, for good.
        {
            let db = Connection::open(&path).unwrap();
            db.execute_batch(
                "CREATE TABLE actor (day TEXT NOT NULL, name TEXT NOT NULL, did TEXT NOT NULL,
                   PRIMARY KEY (day, name, did)) WITHOUT ROWID;",
            )
            .unwrap();
            for (d, name, did) in [
                (ago(40), "gateway.delegated", "did:a"),
                (ago(40), "gateway.delegated", "did:b"),
                (ago(39), "gateway.served", "did:a"),
                (ago(35), "gateway.delegated", "did:a"),
                (ago(29), "gateway.delegated", "did:c"),
            ] {
                db.execute(
                    "INSERT INTO actor VALUES (?1, ?2, ?3)",
                    params![d, name, did],
                )
                .unwrap();
            }
        }
        let usage = Usage::open(&path).unwrap();
        // Opened again, as each deploy does: nothing moves twice.
        drop(usage);
        let usage = Usage::open(&path).unwrap();
        // What tells why, past the month and at its edge.
        {
            let db = usage.db.lock().unwrap();
            for d in [ago(30), ago(29)] {
                let h = format!("{d}T23");
                db.execute(
                    "INSERT INTO statement VALUES (?1, 'SELECT 1', 1, 100, 100)",
                    [&h],
                )
                .unwrap();
                db.execute(
                    "INSERT INTO failure VALUES (?1, 'mcp.tool', 'x', 'no', 1)",
                    [&h],
                )
                .unwrap();
                db.execute("INSERT INTO agent VALUES (?1, 'curl/8', 1)", [&h])
                    .unwrap();
            }
        }
        usage.event("gateway.served", "", true, Duration::ZERO, Some("did:c"));
        usage.event("gateway.delegated", "", true, Duration::ZERO, Some("did:c"));
        usage.event("gateway.delegated", "", true, Duration::ZERO, Some("did:d"));
        usage.flush().unwrap();
        usage.flush().unwrap();
        {
            let db = usage.db.lock().unwrap();
            let gone: bool = db
                .query_row(
                    "SELECT NOT EXISTS (SELECT 1 FROM sqlite_master WHERE name = 'actor')",
                    [],
                    |r| r.get(0),
                )
                .unwrap();
            assert!(gone);
            // No one past the month; the day at its edge still kept.
            let oldest: String = db
                .query_row("SELECT min(day) FROM account", [], |r| r.get(0))
                .unwrap();
            assert_eq!(oldest, ago(29));
            for table in ["statement", "failure", "agent"] {
                let oldest: String = db
                    .query_row(&format!("SELECT min(hour) FROM {table}"), [], |r| r.get(0))
                    .unwrap();
                assert_eq!(oldest, format!("{}T23", ago(29)), "{table}");
            }
            // Their funnels, counted: a and b started 40 days ago and a went
            // on; a again 35 days ago was not a start.
            let mut stmt = db
                .prepare("SELECT day, start, goal, started, reached FROM cohort ORDER BY day")
                .unwrap();
            let cohorts: Vec<(String, String, String, u64, u64)> = stmt
                .query_map([], |r| {
                    Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?, r.get(4)?))
                })
                .unwrap()
                .collect::<rusqlite::Result<_>>()
                .unwrap();
            assert_eq!(
                cohorts,
                [(
                    ago(40),
                    "gateway.delegated".into(),
                    "gateway.served".into(),
                    2,
                    1
                )]
            );
        }
        let mut days: BTreeMap<String, Day> = [ago(40), day(now)]
            .into_iter()
            .map(|d| {
                (
                    d.clone(),
                    Day {
                        day: d,
                        ..Day::default()
                    },
                )
            })
            .collect();
        let mut report = Report::default();
        usage.read(&mut days, &mut report).unwrap();
        // a and b, counted, c, who pushed since, and d, who has not; c
        // again today was not a start.
        let f = &report.funnels[0];
        assert_eq!(
            (f.from.as_str(), f.started, f.reached),
            ("gateway.delegated", 4, 2)
        );
        // The accounts are of the month only.
        let delegated = report.events.iter().find(|e| e.name == "gateway.delegated");
        assert_eq!(delegated.unwrap().accounts, 2);
    }

    /// Past the busiest hundred, a write and a slow route are kept, however
    /// few their requests; a route that is neither is not.
    #[test]
    fn writes_and_slow_routes_are_kept() {
        let dir = tempfile::tempdir().unwrap();
        let usage = Usage::open(&dir.path().join("usage.db")).unwrap();
        let today = day(SystemTime::now());
        let hour = format!("{today}T00");
        {
            let db = usage.db.lock().unwrap();
            let put = |route: &str, method: &str, n: i64, over_two: i64| {
                db.execute(
                    "INSERT INTO request VALUES (?1, 'xrpc', ?2, ?3, 2, 'browser', ?4, ?4, 0, ?5)",
                    rusqlite::params![hour, route, method, n, over_two],
                )
                .unwrap();
            };
            for i in 0..100 {
                put(&format!("/busy/{i}"), "GET", 1000 + i, 0);
            }
            put("/xrpc/dev.knotless.inviteToOrg", "POST", 1, 1);
            put("/slow", "GET", 1, 1);
            put("/quiet", "GET", 1, 0);
        }
        let mut days = BTreeMap::from([(
            today.clone(),
            Day {
                day: today,
                ..Day::default()
            },
        )]);
        let mut report = Report::default();
        usage.read(&mut days, &mut report).unwrap();
        let routes: Vec<_> = report.routes.iter().map(|r| r.route.as_str()).collect();
        assert_eq!(routes.len(), 102);
        assert_eq!(routes[0], "/busy/99");
        assert_eq!(&routes[100..], ["/slow", "/xrpc/dev.knotless.inviteToOrg"]);
    }

    /// A long poll is counted, but never as slow: it waits on purpose.
    #[test]
    fn long_polls_are_not_slow() {
        let dir = tempfile::tempdir().unwrap();
        let usage = Usage::open(&dir.path().join("usage.db")).unwrap();
        let route = "/xrpc/dev.knotless.awaitSpaceWrites";
        let waited = Duration::from_secs(25);
        usage.tally(Some(route), route, "GET", 200, "knotless/0.4.0", waited);
        let pending = usage.pending.lock().unwrap();
        let tally = pending.values().next().unwrap();
        assert_eq!((tally.n, tally.over_half, tally.over_two), (1, 0, 0));
    }

    #[test]
    fn counts_folds_and_reads() {
        let dir = tempfile::tempdir().unwrap();
        let usage = Usage::open(&dir.path().join("usage.db")).unwrap();
        let fast = Duration::from_millis(3);
        usage.tally(
            Some("/api/{authority}/{repo}/tree"),
            "/api/a/r/tree",
            "GET",
            200,
            "knotless/0.4.0",
            fast,
        );
        usage.tally(
            Some("/api/{authority}/{repo}/tree"),
            "/api/b/r/tree",
            "GET",
            502,
            "knotless/0.4.0",
            Duration::from_secs(3),
        );
        usage.tally(Some("/healthz"), "/healthz", "GET", 200, "curl/8.0", fast);
        usage.tally(
            None,
            "/a.test",
            "GET",
            200,
            "Mozilla/5.0",
            Duration::from_millis(600),
        );
        // A crawler, by its agent; the health check's was not counted.
        let crawler = format!("Googlebot/2.1 {}", "x".repeat(300));
        usage.tally(None, "/b.test", "GET", 200, &crawler, fast);
        usage.tally(None, "/b.test", "GET", 200, &crawler, fast);
        // Slow statements, as written; and why a tool failed.
        usage.statement("SELECT a\n  FROM b", Duration::from_millis(100));
        usage.statement("SELECT a FROM b", Duration::from_millis(500));
        usage.statement("SELECT c", Duration::from_millis(200));
        usage.failed(
            "mcp.tool",
            "read_file",
            &format!(" no such file {}", "y".repeat(400)),
        );
        usage.failed("mcp.tool", "rerun", "not the owner");
        usage.failed("mcp.tool", "rerun", "not the owner");
        {
            let pending = usage.pending.lock().unwrap();
            // Apart by the status's class, and by client.
            assert_eq!(pending.len(), 4);
            let failed = pending.iter().find(|(k, _)| k.class == 5).unwrap().1;
            assert_eq!(
                *failed,
                Tally {
                    n: 1,
                    ms: 3000,
                    over_half: 1,
                    over_two: 1
                }
            );
        }
        let ip = Some("1.2.3.4".parse().unwrap());
        usage
            .view(ip, "Mozilla/5.0", "/[authority]", "bsky.app")
            .unwrap();
        usage
            .view(ip, "Mozilla/5.0", "/[authority]/[repo]", "")
            .unwrap();
        usage
            .view(
                Some("5.6.7.8".parse().unwrap()),
                "Mozilla/5.0",
                "/[authority]",
                "",
            )
            .unwrap();
        usage.event("mcp.tool", "read_file", true, fast, Some("did:x"));
        usage.event("mcp.tool", "read_file", false, fast, None);
        // Yesterday's hashes, and its salt, as a serve left them.
        {
            let db = usage.db.lock().unwrap();
            db.execute_batch(
                "INSERT INTO visitor VALUES ('2000-01-01', x'01'), ('2000-01-01', x'02');
                 INSERT INTO salt VALUES ('2000-01-01', x'00');",
            )
            .unwrap();
        }
        usage.flush().unwrap();
        // Folded twice, as two serves would: counted once.
        usage.flush().unwrap();
        {
            let db = usage.db.lock().unwrap();
            let one = |sql: &str| db.query_row(sql, [], |r| r.get::<_, i64>(0)).unwrap();
            assert_eq!(one("SELECT n FROM visitors WHERE day = '2000-01-01'"), 2);
            assert_eq!(
                one("SELECT count(*) FROM visitor WHERE day = '2000-01-01'"),
                0
            );
            assert_eq!(one("SELECT count(*) FROM salt"), 1);
        }
        let now = SystemTime::now();
        let today = day(now);
        let mut days = BTreeMap::from([
            (
                "2000-01-01".to_string(),
                Day {
                    day: "2000-01-01".into(),
                    ..Day::default()
                },
            ),
            (
                today.clone(),
                Day {
                    day: today.clone(),
                    ..Day::default()
                },
            ),
        ]);
        let mut report = Report::default();
        usage.read(&mut days, &mut report).unwrap();
        let old = &days["2000-01-01"];
        assert_eq!((old.visitors, old.views, old.requests), (2, 0, 0));
        let now = &days[&today];
        assert_eq!(
            (now.visitors, now.views, now.requests, now.errors),
            (2, 3, 5, 1)
        );
        // The pages, a browser's and the crawler's, then the API.
        assert_eq!(report.routes[0].route, "*");
        let api = &report.routes[1];
        assert_eq!(api.route, "/api/{authority}/{repo}/tree");
        assert_eq!((api.requests, api.errors, api.mean_ms), (2, 1, 1501));
        // All in this hour: the one that failed took three seconds.
        let hours: Vec<_> = report
            .hours
            .iter()
            .map(|h| (h.requests, h.errors, h.over_two))
            .collect();
        assert_eq!(hours, [(5, 1, 1)]);
        assert!(report.hours[0].hour.starts_with(&today));
        let names = |c: &[Count]| c.iter().map(|c| (c.name.clone(), c.n)).collect::<Vec<_>>();
        assert_eq!(
            names(&report.pages),
            [
                ("/[authority]".into(), 2),
                ("/[authority]/[repo]".into(), 1)
            ]
        );
        assert_eq!(names(&report.referrers), [("bsky.app".into(), 1)]);
        assert_eq!(
            names(&report.clients),
            [("bot".into(), 2), ("cli".into(), 2), ("browser".into(), 1)]
        );
        let agent = &report.agents[0];
        assert_eq!((agent.name.len(), agent.n), (AGENT_MAX, 2));
        assert!(agent.name.starts_with("Googlebot/2.1 x"));
        assert_eq!(report.agents.len(), 1);
        let statements: Vec<_> = report
            .statements
            .iter()
            .map(|s| (s.sql.as_str(), s.calls, s.mean_ms, s.max_ms))
            .collect();
        assert_eq!(
            statements,
            [("SELECT a FROM b", 2, 300, 500), ("SELECT c", 1, 200, 200)]
        );
        let failures: Vec<_> = report
            .failures
            .iter()
            .map(|f| (f.subject.as_str(), f.error.len(), f.n))
            .collect();
        assert_eq!(failures, [("rerun", 13, 2), ("read_file", ERROR_MAX, 1)]);
        assert!(report.failures[1].error.starts_with("no such file y"));
        assert_eq!(hour(SystemTime::UNIX_EPOCH), "1970-01-01T00");
    }
}