//! What this server logs of what it serves, and the numbers it keeps
//! (`knotless::telemetry`): a line for each request, with the trace its
//! calls hand on, and `/metrics` and `/log` on a port of their own.

use crate::App;
use axum::Router;
use axum::extract::{MatchedPath, Request, State};
use axum::http::{HeaderValue, StatusCode, header};
use axum::middleware::Next;
use axum::response::{IntoResponse, Response};
use axum::routing::{get, put};
use knotless::telemetry::{self, Fields, Level};
use rusqlite::Connection;
use rusqlite::trace::{TraceEvent, TraceEventCodes};
use std::collections::HashMap;
use std::sync::{Arc, LazyLock, Mutex, OnceLock};
use std::time::{Duration, Instant, SystemTime};

/// Why a response failed, for the line logged for it.
#[derive(Clone)]
pub struct Cause(pub String);

/// `status`, saying `message`, as the line logged for it says why.
pub fn failed(status: StatusCode, message: String) -> Response {
    let mut res = (status, message.clone()).into_response();
    res.extensions_mut().insert(Cause(message));
    res
}

/// The route a request matched, as the routing finds it.
#[derive(Clone, Default)]
struct Route(Arc<OnceLock<String>>);

/// `routes`, each request through them logged and counted, in a trace
/// of its own or the one its `traceparent` names, which the answer gives
/// back as `x-trace-id`.
pub fn observed(routes: Router) -> Router {
    routes
        .route_layer(axum::middleware::from_fn(matched))
        .layer(axum::middleware::from_fn(observe))
}

/// The routes that wait on purpose before they answer, for something to
/// happen or a while to pass: long polls, never slow however long they
/// take.
const WAITS: [&str; 2] = [
    "/xrpc/dev.knotless.awaitSpaceWrites",
    "/xrpc/dev.knotless.getDelegation",
];

/// Whether `route` is one of [`WAITS`].
pub fn waits(route: Option<&str>) -> bool {
    route.is_some_and(|r| WAITS.contains(&r))
}

async fn matched(req: Request, next: Next) -> Response {
    let path = req.extensions().get::<MatchedPath>();
    if let (Some(route), Some(path)) = (req.extensions().get::<Route>(), path) {
        let _ = route.0.set(path.as_str().to_string());
    }
    next.run(req).await
}

async fn observe(mut req: Request, next: Next) -> Response {
    let started = Instant::now();
    let route = Route::default();
    req.extensions_mut().insert(route.clone());
    let method = req.method().to_string();
    let path = telemetry::short(req.uri().path());
    let query = telemetry::query(&path, req.uri().query());
    let agent = req.headers().get(header::USER_AGENT);
    let agent = agent.map(HeaderValue::to_str).and_then(Result::ok);
    let agent = agent.map(str::to_string);
    let parent = req.headers().get("traceparent");
    let trace = telemetry::trace_from(parent.and_then(|v| v.to_str().ok()));
    req.extensions_mut().insert(telemetry::Trace(trace.clone()));
    let mut res = telemetry::traced(trace.clone(), next.run(req)).await;
    let id = HeaderValue::try_from(trace.as_str()).expect("hex is a header value");
    res.headers_mut().insert("x-trace-id", id);
    let cause = res.extensions().get::<Cause>();
    let bytes = axum::body::HttpBody::size_hint(res.body()).exact();
    let matched = route.0.get().map(String::as_str);
    telemetry::served(&telemetry::Served {
        route: matched,
        method: &method,
        path: &path,
        query: query.as_deref(),
        agent: agent.as_deref(),
        status: res.status().as_u16(),
        bytes,
        took: started.elapsed(),
        trace: &trace,
        cause: cause.map(|c| c.0.as_str()),
        waits: waits(matched),
    });
    res
}

/// Has `db` log each statement it runs that is slow ([`profiled`]), and
/// hand each to the planner, when they are planned ([`crate::plans`]).
pub fn watch(db: &Connection) {
    let mut codes = TraceEventCodes::SQLITE_TRACE_PROFILE;
    if crate::plans::planning() {
        codes |= TraceEventCodes::SQLITE_TRACE_STMT;
    }
    db.trace_v2(codes, Some(traced));
}

fn traced(event: TraceEvent<'_>) {
    if let TraceEvent::Stmt(stmt, _) = &event {
        crate::plans::check(&stmt.sql());
    }
    if let TraceEvent::Profile(stmt, took) = event {
        profiled(took, &stmt.sql());
    }
}

/// Logs the statement `sql`, and counts it in usage, which took `took`, if that is a tenth
/// of a slow request or more (`KNOTLESS_SLOW_MS`, a second unless set):
/// a request seldom runs one statement alone, so one that slow is what
/// made it slow, and the line, in its trace, says which.
fn profiled(took: Duration, sql: &str) {
    if telemetry::slow(took.saturating_mul(10)) {
        telemetry::sql(sql, took);
        crate::usage::slow_statement(sql, took);
    }
}

/// What is answered on `--metrics`, a port kept off the proxy: the
/// numbers at `/metrics`, and `PUT /log` setting what is logged, as
/// `RUST_LOG` does.
pub fn admin(app: Arc<App>) -> Router {
    Router::new()
        .route("/metrics", get(metrics))
        .route("/log", put(level))
        .with_state(app)
}

/// Answers [`admin`] on `addr`, if given, and prints where.
pub async fn serve_admin(addr: Option<&str>, app: Arc<App>) -> anyhow::Result<()> {
    let Some(addr) = addr else {
        return Ok(());
    };
    let listener = tokio::net::TcpListener::bind(addr).await?;
    let at = format!("http://{}", listener.local_addr()?);
    println!("metrics on {at}");
    said("metrics listening", Some(&at));
    tokio::spawn(axum::serve(listener, admin(app)).into_future());
    Ok(())
}

async fn metrics(State(app): State<Arc<App>>) -> Response {
    let mut text = telemetry::metrics();
    text.push_str(&feed_metrics(&app));
    let kind = "text/plain; version=0.0.4";
    ([(header::CONTENT_TYPE, kind)], text).into_response()
}

async fn level(body: String) -> Response {
    match telemetry::set_filter(body.trim()) {
        Ok(()) => format!("logging {}\n", body.trim()).into_response(),
        Err(err) => (StatusCode::BAD_REQUEST, err).into_response(),
    }
}

/// Each feed's state, as `/metrics` gives it.
fn feed_metrics(app: &App) -> String {
    let mut out = String::from("# TYPE knotless_feed_connected gauge\n");
    out.push_str("# TYPE knotless_feed_events_total counter\n");
    out.push_str("# TYPE knotless_feed_lag_seconds gauge\n");
    for f in app.feeds() {
        let name = &f.name;
        let up = u8::from(f.connected);
        out.push_str(&format!(
            "knotless_feed_connected{{feed=\"{name}\"}} {up}\n"
        ));
        out.push_str(&format!(
            "knotless_feed_events_total{{feed=\"{name}\"}} {}\n",
            f.events
        ));
        let lag = lag_ms(name).map(|ms| ms as f64 / 1000.0);
        let line = lag.map(|s| format!("knotless_feed_lag_seconds{{feed=\"{name}\"}} {s}\n"));
        out.push_str(&line.unwrap_or_default());
    }
    out
}

/// How far behind now each feed was at its last event, in milliseconds.
static LAGS: LazyLock<Mutex<HashMap<String, u64>>> = LazyLock::new(Mutex::default);

/// Keeps how far behind now the event in the Jetstream `frame` is, by
/// the time it was made: its sequence numbers are its own, not times.
pub fn lagged(feed: &str, frame: &str) {
    let frame: serde_json::Value = serde_json::from_str(frame).unwrap_or_default();
    let time = frame["payload"]["time"].as_str().unwrap_or_default();
    if let Ok(at) = humantime::parse_rfc3339_weak(time) {
        let lag = SystemTime::now().duration_since(at).unwrap_or_default();
        LAGS.lock()
            .unwrap()
            .insert(feed.to_string(), telemetry::millis(lag));
    }
}

/// How far behind now the feed `name` was at its last event.
fn lag_ms(name: &str) -> Option<u64> {
    LAGS.lock().unwrap().get(name).copied()
}

/// Logs a summary of what was counted every `every`, the first at once,
/// and how each feed is doing.
pub async fn summaries(app: Arc<App>, every: Duration) {
    let mut tick = tokio::time::interval(every);
    loop {
        tick.tick().await;
        telemetry::summarize();
        for f in app.feeds() {
            let fields = Fields {
                feed: Some(&f.name),
                connected: Some(f.connected),
                count: Some(f.events),
                lag_ms: lag_ms(&f.name),
                duration_ms: Some(f.silent_seconds * 1000),
                ..Fields::default()
            };
            telemetry::note(Level::INFO, "summary: feed", &fields);
        }
    }
}

/// Logs that the feed `name`, from `url`, connected or dropped.
pub fn feed(name: &str, url: &str, connected: bool) {
    const SAID: [(Level, &str); 2] = [
        (Level::WARN, "feed disconnected"),
        (Level::INFO, "feed connected"),
    ];
    let (level, said) = SAID[usize::from(connected)];
    let fields = Fields {
        feed: Some(name),
        source: Some(url),
        ..Fields::default()
    };
    telemetry::note(level, said, &fields);
}

/// Logs `said` at `level` of `did`, an account said to be deleted, as it
/// was asked after ([`crate::accounts::settle`]): the status its host
/// holds it at, the asks so far, and why none settled it.
pub fn account_asked(
    level: Level,
    said: &str,
    did: &str,
    status: Option<&str>,
    tried: Option<u32>,
    why: Option<String>,
) {
    let fields = Fields {
        did: Some(did),
        status,
        count: tried.map(u64::from),
        error: why,
        ..Fields::default()
    };
    telemetry::note(level, said, &fields);
}

/// Logs, at warn, an event of `did`'s the index did not take: one the
/// index will miss until a sweep fills it.
pub fn indexed<T, E: std::fmt::Display>(did: &str, done: Result<T, E>) {
    if let Err(err) = done {
        let fields = Fields {
            did: Some(did),
            error: Some(format!("{err:#}")),
            ..Fields::default()
        };
        telemetry::note(Level::WARN, "event dropped: not indexed", &fields);
    }
}

/// Logs, at warn, why a backfill could not read `did`: a later one tries
/// again.
pub fn backfill_failed(did: &str, err: &anyhow::Error) {
    let fields = Fields {
        did: Some(did),
        error: Some(format!("{err:#}")),
        ..Fields::default()
    };
    telemetry::note(Level::WARN, "account not backfilled", &fields);
}

/// Seconds since the epoch, now.
pub fn now_secs() -> u64 {
    let now = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH);
    now.unwrap_or_default().as_secs()
}

/// Logs, at warn, why a sync of the space `uri` failed, if it did: it is
/// tried again at the next.
pub fn space_synced(uri: &str, done: anyhow::Result<()>) {
    if let Err(err) = done {
        let fields = Fields {
            source: Some(uri),
            error: Some(format!("{err:#}")),
            ..Fields::default()
        };
        telemetry::note(Level::WARN, "space not synced", &fields);
    }
}

/// Logs, at warn, why the authority of the space `uri` could not be asked
/// to tell this server of writes there, if it could not: it is synced
/// every so often all the same.
pub fn space_unheard<T>(uri: &str, asked: anyhow::Result<T>) {
    if let Err(err) = asked {
        let fields = Fields {
            source: Some(uri),
            error: Some(format!("{err:#}")),
            ..Fields::default()
        };
        telemetry::note(Level::WARN, "space not listened to", &fields);
    }
}

/// Logs why `did` was not proven to read the space `uri`, if they were
/// not: at info when its authority refused them, which takes it out of
/// what they are listed as reading, or when their session's grant leaves
/// the space out, which only logging in again changes; at warn when it
/// could not be asked, which leaves that as it was, to ask again.
pub fn reader_checked<T>(uri: &str, did: &str, joined: &anyhow::Result<T>) {
    if let Err(err) = joined {
        let fields = Fields {
            source: Some(uri),
            did: Some(did),
            error: Some(format!("{err:#}")),
            ..Fields::default()
        };
        if knotless::space::denied(err) {
            telemetry::note(Level::INFO, "space reader refused", &fields);
        } else if knotless::oauth::out_of_scope(err) {
            telemetry::note(Level::INFO, "space reader out of scope", &fields);
        } else {
            telemetry::note(Level::WARN, "space reader not checked", &fields);
        }
    }
}

/// Logs, at warn, why `did`'s repo in the space `uri` could not be read on
/// through its oplog, if it could not: it is read whole instead.
pub fn read_whole(uri: &str, did: &str, read: &anyhow::Result<()>) {
    if let Err(err) = read {
        let fields = Fields {
            source: Some(uri),
            did: Some(did),
            error: Some(format!("{err:#}")),
            ..Fields::default()
        };
        telemetry::note(Level::WARN, "space repo read whole", &fields);
    }
}

/// Logs, at warn, why `did`'s repo in the space `uri` could not be read
/// whole, if it could not: it is read from the start of its oplog instead.
pub fn read_from_start(uri: &str, did: &str, read: &anyhow::Result<()>) {
    if let Err(err) = read {
        let fields = Fields {
            source: Some(uri),
            did: Some(did),
            error: Some(format!("{err:#}")),
            ..Fields::default()
        };
        telemetry::note(Level::WARN, "space repo read from its oplog", &fields);
    }
}

/// Logs, at warn, each record of `did`'s repo in the space `uri` that
/// could not be read, by its path, and why: each is left out of the
/// index, and the rest of the repo read all the same.
pub fn unreadable(uri: &str, did: &str, records: &[knotless::space_sync::Unreadable]) {
    for record in records {
        let path = format!("{}/{}", record.collection, record.rkey);
        let fields = Fields {
            source: Some(uri),
            did: Some(did),
            key: Some(&path),
            error: Some(record.why.clone()),
            ..Fields::default()
        };
        telemetry::note(Level::WARN, "space record unreadable", &fields);
    }
}

/// Logs, at warn, why following the labeler `did` stopped, if it did
/// for a reason: it is followed again at once.
pub fn labeler<T>(did: &str, done: anyhow::Result<T>) {
    if let Err(err) = done {
        let fields = Fields {
            did: Some(did),
            error: Some(format!("{err:#}")),
            ..Fields::default()
        };
        telemetry::note(Level::WARN, "labeler dropped", &fields);
    }
}

/// Logs that the index was opened at startup, taking `took`.
pub fn opened(index: Option<&crate::index::Index>, took: Duration) {
    if let Some(index) = index {
        let fields = Fields {
            rebuilt: Some(index.rebuilt),
            duration_ms: Some(telemetry::millis(took)),
            ..Fields::default()
        };
        telemetry::note(Level::INFO, "index opened", &fields);
    }
}

/// Fills the index from `service`, saying when it starts and how it
/// ended.
pub async fn sweep(
    ctx: &knotless::env::Ctx,
    index: &crate::index::Index,
    service: &str,
    sources: &crate::Sources,
) {
    let started = Instant::now();
    let source = Fields {
        source: Some(service),
        ..Fields::default()
    };
    telemetry::note(Level::INFO, "index sweep started", &source);
    let said = crate::backfilled(ctx, index, service, sources).await;
    let fields = Fields {
        reason: Some(&said),
        duration_ms: Some(telemetry::millis(started.elapsed())),
        ..source
    };
    telemetry::note(Level::INFO, "index sweep ended", &fields);
}

/// Logs, at info, `message` about this server: its address, or that it
/// is stopping.
pub fn said(message: &str, source: Option<&str>) {
    let fields = Fields {
        source,
        ..Fields::default()
    };
    telemetry::note(Level::INFO, message, &fields);
}

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

    #[test]
    fn slow_statements() {
        profiled(Duration::ZERO, "SELECT 1");
        profiled(Duration::from_secs(5), "SELECT 1");
        let db = Connection::open_in_memory().unwrap();
        watch(&db);
        db.execute_batch("SELECT 1").unwrap();
    }

    #[test]
    fn reader_checks() {
        let uri = "at://did:plc:o/space/dev.knotless.space/r";
        reader_checked(uri, "did:plc:m", &Ok(()));
        let no = knotless::xrpc::XrpcError {
            status: 400,
            error: "NotAuthorized".into(),
            message: String::new(),
            reset: None,
        };
        reader_checked::<()>(uri, "did:plc:m", &Err(no.into()));
        reader_checked::<()>(
            uri,
            "did:plc:m",
            &Err(anyhow::anyhow!("connection refused")),
        );
        let scope = anyhow::anyhow!("InvalidToken (403): Scope does not permit reading {uri}");
        reader_checked::<()>(uri, "did:plc:m", &Err(scope));
    }

    #[test]
    fn lags() {
        assert_eq!(lag_ms("firehose"), None);
        let ago = humantime::format_rfc3339_micros(SystemTime::now() - Duration::from_secs(3));
        let frame = serde_json::json!({"payload": {"seq": 5, "time": ago.to_string()}});
        lagged("jetstream-test", &frame.to_string());
        assert!(lag_ms("jetstream-test").unwrap() >= 3000);
        // A frame of no time leaves the last.
        lagged("jetstream-test", "junk");
        assert!(lag_ms("jetstream-test").unwrap() >= 3000);
    }

    #[test]
    fn notes() {
        // Nothing is set up to log here: each only has to run.
        feed("jetstream", "wss://jetstream.test", true);
        feed("jetstream", "wss://jetstream.test", false);
        indexed::<(), _>("did:plc:x", Err("no PDS"));
        indexed::<(), &str>("did:plc:x", Ok(()));
        said("stopping", None);
        labeler::<()>("did:web:labeler.test", Err(anyhow::anyhow!("gone")));
        space_synced("at://did:plc:x/space/t/k", Err(anyhow::anyhow!("refused")));
        space_unheard::<()>("at://did:plc:x/space/t/k", Err(anyhow::anyhow!("no")));
        read_whole(
            "at://did:plc:x/space/t/k",
            "did:plc:y",
            &Err(anyhow::anyhow!("no match")),
        );
        read_from_start(
            "at://did:plc:x/space/t/k",
            "did:plc:y",
            &Err(anyhow::anyhow!("no match")),
        );
        let cut = knotless::space_sync::Unreadable {
            collection: "a.b.c".into(),
            rkey: "1".into(),
            why: "a record that does not come to its CID".into(),
        };
        unreadable("at://did:plc:x/space/t/k", "did:plc:y", &[cut]);
        opened(None, Duration::ZERO);
        let res = failed(StatusCode::BAD_GATEWAY, "down".into());
        assert_eq!(res.extensions().get::<Cause>().unwrap().0, "down");
    }

    #[tokio::test]
    async fn admin_answers() {
        let ctx = knotless::env::Ctx::new(knotless::env::Env::from_vars(|_| None));
        let app = App::new(ctx, Duration::from_secs(5));
        app.follows("jetstream", "wss://jetstream.test");
        app.heard("jetstream", Some(1));
        let now = humantime::format_rfc3339_micros(SystemTime::now()).to_string();
        lagged(
            "jetstream",
            &serde_json::json!({"payload": {"time": now}}).to_string(),
        );
        let shown = metrics(State(app.clone())).await;
        let body = axum::body::to_bytes(shown.into_body(), usize::MAX)
            .await
            .unwrap();
        let body = String::from_utf8_lossy(&body);
        assert!(body.contains("knotless_feed_connected{feed=\"jetstream\"} 0"));
        assert!(body.contains("knotless_feed_lag_seconds{feed=\"jetstream\"}"));
        // Logging is not set up in tests, so a level is refused.
        let refused = level("debug".into()).await;
        assert_eq!(refused.status(), StatusCode::BAD_REQUEST);
        serve_admin(None, app.clone()).await.unwrap();
        serve_admin(Some("127.0.0.1:0"), app.clone()).await.unwrap();
        let summarizing = tokio::spawn(summaries(app, Duration::from_secs(60)));
        tokio::task::yield_now().await;
        summarizing.abort();
    }
}