//! The lines serve logs, kept to read back, for its operator:
//! `dev.knotless.getLogs`, `knotless admin logs` and MCP's `admin_logs`.
//!
//! Each JSON line serve writes to stderr ([`telemetry::keep_lines`]) is
//! kept here too, in a SQLite file beside the usage counts, written a
//! second at a time off the requests' way. Every serve running writes to
//! it, as two do for a moment in each deploy, so a deploy loses none:
//! the lines of a request a serve now gone answered are still here.
//!
//! What is kept is what the journal has (no query string, header, body
//! or address), and for no longer: lines at info and below, the most
//! numerous, only the last [`KEEP`] (`--logs-keep`); every line only
//! [`DAYS`] days. A read is itself logged, `logs read`, with who asked
//! and for what.

use crate::moderation::admin;
use crate::{App, AppError};
use axum::Json;
use axum::extract::{Query, State};
use axum::http::{HeaderMap, StatusCode};
use knotless::telemetry::{self, Fields, Level};
use rusqlite::{Connection, OptionalExtension, params};
use serde::Deserialize;
use serde_json::{Value, json};
use std::collections::VecDeque;
use std::path::Path;
use std::sync::{Arc, Mutex};
use std::time::{Duration, SystemTime};

/// How often the lines logged are written.
pub const FLUSH: Duration = Duration::from_secs(1);

/// How often the oldest are let go.
pub const PRUNE: Duration = Duration::from_secs(60);

/// How many lines at info and below are kept, unless `--logs-keep` says.
pub const KEEP: u64 = 500_000;

/// How many days any line is kept: as long as the journal keeps them.
pub const DAYS: u64 = 30;

/// The most lines held between writes: a write that cannot keep up lets
/// the oldest go rather than grow without end.
const PENDING_MAX: usize = 50_000;

/// The most lines a read answers.
const LIMIT_MAX: u32 = 500;

pub struct Logs {
    db: Mutex<Connection>,
    pending: Mutex<VecDeque<Vec<u8>>>,
    keep: u64,
}

impl Logs {
    /// The lines kept at `path`, keeping `keep` at info and below.
    pub fn open(path: &Path, keep: u64) -> rusqlite::Result<Logs> {
        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 line (
               id INTEGER PRIMARY KEY, at TEXT NOT NULL, level INTEGER NOT NULL,
               message TEXT NOT NULL, trace TEXT, ms INTEGER, line TEXT NOT NULL);
             CREATE INDEX IF NOT EXISTS line_trace ON line (trace) WHERE trace IS NOT NULL;
             CREATE INDEX IF NOT EXISTS line_level ON line (level, id);
             CREATE INDEX IF NOT EXISTS line_message ON line (message, id);
             CREATE INDEX IF NOT EXISTS line_at ON line (at);",
        );
        made?;
        Ok(Logs {
            db: Mutex::new(db),
            pending: Mutex::default(),
            keep,
        })
    }

    /// Holds `line`, one JSON line as logged, for the next write.
    pub fn keep(&self, line: &[u8]) {
        let mut pending = self.pending.lock().unwrap();
        if pending.len() >= PENDING_MAX {
            pending.pop_front();
        }
        pending.push_back(line.to_vec());
    }

    /// Writes the lines held: each with its time, level, message, trace
    /// and duration beside it, to find it by. A line that is not JSON is
    /// kept as a message of its own. Lines a write fails to keep, as when
    /// another serve holds the file past the busy timeout, are held again,
    /// ahead of those logged since, for the next.
    pub fn flush(&self) -> rusqlite::Result<()> {
        let batch = std::mem::take(&mut *self.pending.lock().unwrap());
        let written = self.write(&batch);
        if written.is_err() {
            let mut pending = self.pending.lock().unwrap();
            let since = std::mem::replace(&mut *pending, batch);
            pending.extend(since);
            let over = pending.len().saturating_sub(PENDING_MAX);
            pending.drain(..over);
        }
        written
    }

    /// Writes `pending` in one transaction: all of it or none.
    fn write(&self, pending: &VecDeque<Vec<u8>>) -> rusqlite::Result<()> {
        let mut db = self.db.lock().unwrap();
        let tx = db.transaction()?;
        let mut stmt =
            tx.prepare_cached("INSERT INTO line VALUES (NULL, ?1, ?2, ?3, ?4, ?5, ?6)")?;
        for line in pending {
            let text = String::from_utf8_lossy(line);
            let v: Value = serde_json::from_str(text.trim())
                .unwrap_or_else(|_| json!({"message": text.trim()}));
            let text = v.to_string();
            let at = v["timestamp"].as_str().map_or_else(now, str::to_string);
            let level = rank(v["level"].as_str().unwrap_or("INFO"));
            let message = v["message"].as_str().unwrap_or_default();
            let ms = v["duration_ms"].as_u64();
            stmt.execute(params![
                at,
                level,
                message,
                v["trace_id"].as_str(),
                ms,
                text
            ])?;
        }
        drop(stmt);
        tx.commit()
    }

    /// Lets go of what is past keeping: lines at info and below past the
    /// last [`Logs::keep`] of them, and any line older than [`DAYS`].
    pub fn prune(&self) -> rusqlite::Result<()> {
        let db = self.db.lock().unwrap();
        let since = SystemTime::now() - Duration::from_secs(86_400 * DAYS);
        let since = humantime::format_rfc3339_seconds(since).to_string();
        db.execute("DELETE FROM line WHERE at < ?1", [since])?;
        let last: Option<i64> = db
            .query_row(
                "SELECT id FROM line WHERE level <= ?1 ORDER BY id DESC LIMIT 1 OFFSET ?2",
                params![rank("INFO"), self.keep],
                |r| r.get(0),
            )
            .optional()?;
        if let Some(last) = last {
            db.execute(
                "DELETE FROM line WHERE level <= ?1 AND id <= ?2",
                params![rank("INFO"), last],
            )?;
        }
        Ok(())
    }

    /// The lines `asked` names, oldest first: all of one request's, by
    /// its trace, up to [`LIMIT_MAX`] unless asked for fewer, or the last
    /// of those of a message, a level or a duration.
    pub fn read(&self, asked: &Asked) -> rusqlite::Result<Vec<Value>> {
        let unasked = if asked.trace.is_some() {
            LIMIT_MAX
        } else {
            100
        };
        let limit = asked.limit.unwrap_or(unasked).clamp(1, LIMIT_MAX);
        let db = self.db.lock().unwrap();
        let level = asked.level.as_deref().map_or(0, rank);
        // A statement for each way of asking, each by its own index: one
        // asking `?2 IS NULL OR message = ?2` could use none, and read
        // every line kept, newest first, for a message seldom logged.
        let stmt = match (&asked.trace, &asked.message) {
            (Some(_), _) => db.prepare_cached(
                "SELECT line FROM line WHERE trace = ?1 AND level >= ?3 ORDER BY id DESC LIMIT ?5",
            ),
            (None, Some(_)) => db.prepare_cached(
                "SELECT line FROM line INDEXED BY line_message
                 WHERE message = ?2 AND level >= ?3 AND (?4 IS NULL OR ms >= ?4)
                 ORDER BY id DESC LIMIT ?5",
            ),
            (None, None) => db.prepare_cached(
                "SELECT line FROM line WHERE level >= ?3 AND (?4 IS NULL OR ms >= ?4)
                 ORDER BY id DESC LIMIT ?5",
            ),
        };
        let mut stmt = stmt?;
        let p = params![asked.trace, asked.message, level, asked.min_ms, limit];
        let rows = stmt.query_map(p, |r| r.get::<_, String>(0))?;
        let mut lines: Vec<Value> = rows
            .map(|r| r.map(|text| serde_json::from_str(&text).unwrap_or_default()))
            .collect::<rusqlite::Result<_>>()?;
        lines.reverse();
        Ok(lines)
    }
}

/// A level, as a number to compare: `ERROR` the highest.
fn rank(level: &str) -> i64 {
    match level.to_ascii_uppercase().as_str() {
        "ERROR" => 5,
        "WARN" => 4,
        "INFO" => 3,
        "DEBUG" => 2,
        _ => 1,
    }
}

fn now() -> String {
    humantime::format_rfc3339_micros(SystemTime::now()).to_string()
}

/// Writes the lines held every `every`, and lets go of the oldest every
/// [`PRUNE`], at once the first time.
pub async fn flushing(logs: Arc<Logs>, every: Duration) {
    let mut pruned: Option<std::time::Instant> = None;
    loop {
        wrote(logs.flush());
        if pruned.is_none_or(|t| t.elapsed() >= PRUNE) {
            wrote(logs.prune());
            pruned = Some(std::time::Instant::now());
        }
        tokio::time::sleep(every).await;
    }
}

/// Says, to stderr alone, why the lines were not written: they are
/// written with the next, unless too many wait.
fn wrote(done: rusqlite::Result<()>) {
    if let Err(err) = done {
        eprintln!("logs not kept: {err:#}");
    }
}

/// What a read asks for: at least one of these.
#[derive(Deserialize, Default, Debug)]
#[serde(rename_all = "camelCase")]
pub struct Asked {
    /// All of one request's lines.
    pub trace: Option<String>,
    /// Only lines with this message: `sql`, `outgoing`, `request`.
    pub message: Option<String>,
    /// Only lines at this level or above.
    pub level: Option<String>,
    /// Only lines that took this many milliseconds or more.
    pub min_ms: Option<u64>,
    /// How many, the last of them: at most 500, and unless asked, 100,
    /// or for a trace, 500.
    pub limit: Option<u32>,
}

impl Asked {
    /// What it asks, as the line telling of the read says it.
    fn said(&self) -> String {
        let parts = [
            self.trace.as_ref().map(|t| format!("trace={t}")),
            self.message.as_ref().map(|m| format!("message={m}")),
            self.level.as_ref().map(|l| format!("level={l}")),
            self.min_ms.map(|ms| format!("minMs={ms}")),
        ];
        parts.into_iter().flatten().collect::<Vec<_>>().join(" ")
    }
}

/// `dev.knotless.getLogs`, for an operator: the lines `asked` names.
pub async fn get_logs(
    State(app): State<Arc<App>>,
    headers: HeaderMap,
    Query(asked): Query<Asked>,
) -> Result<Json<Value>, AppError> {
    let did = admin(&app, &headers, knotless::moderation::GET_LOGS).await?;
    let Some(logs) = app.logs.as_deref() else {
        let why = "this server keeps no logs";
        return Err(AppError(StatusCode::NOT_FOUND, why.into()));
    };
    let said = asked.said();
    if said.is_empty() {
        let why = "ask for a trace, a message, a level or minMs";
        return Err(AppError(StatusCode::BAD_REQUEST, why.into()));
    }
    let f = Fields {
        did: Some(&did),
        reason: Some(&said),
        ..Fields::default()
    };
    telemetry::note(Level::INFO, "logs read", &f);
    // What was logged up to now, this read's own line with it.
    let read = logs.flush().and_then(|()| logs.read(&asked));
    let lines = read
        .map_err(anyhow::Error::from)
        .map_err(AppError::internal)?;
    let lines: Vec<Value> = lines
        .into_iter()
        .map(|line| json!({"line": line}))
        .collect();
    Ok(Json(json!({ "lines": lines })))
}

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

    fn line(at: &str, level: &str, message: &str, trace: Option<&str>, ms: Option<u64>) -> Vec<u8> {
        let v = json!({"timestamp": at, "level": level, "message": message,
            "trace_id": trace, "duration_ms": ms});
        format!("{v}\n").into_bytes()
    }

    #[test]
    fn keeps_reads_and_lets_go() {
        let dir = tempfile::tempdir().unwrap();
        let logs = Logs::open(&dir.path().join("logs.db"), 3).unwrap();
        let today = now();
        let long_ago = "2000-01-01T00:00:00.000000Z";
        logs.keep(&line(long_ago, "WARN", "outgoing", Some("t0"), Some(5000)));
        for i in 0..5u64 {
            logs.keep(&line(&today, "INFO", "request", Some("t1"), Some(i * 100)));
        }
        logs.keep(&line(&today, "WARN", "sql", Some("t1"), Some(400)));
        logs.keep(&line(&today, "DEBUG", "outgoing", Some("t1"), Some(20)));
        logs.keep(b"not json\n");
        logs.flush().unwrap();
        // Read twice: a write with nothing held writes nothing.
        logs.flush().unwrap();
        let read = |asked: Asked| logs.read(&asked).unwrap();
        let messages = |lines: &[Value]| -> Vec<String> {
            lines
                .iter()
                .map(|l| l["message"].as_str().unwrap_or("?").to_string())
                .collect()
        };
        let trace = read(Asked {
            trace: Some("t1".into()),
            ..Asked::default()
        });
        assert_eq!(trace.len(), 7);
        assert_eq!(messages(&trace)[5..], ["sql", "outgoing"]);
        let slow = read(Asked {
            min_ms: Some(400),
            ..Asked::default()
        });
        assert_eq!(messages(&slow), ["outgoing", "request", "sql"]);
        let warned = read(Asked {
            level: Some("warn".into()),
            limit: Some(1),
            ..Asked::default()
        });
        assert_eq!(messages(&warned), ["sql"]);
        let plain = read(Asked {
            message: Some("not json".into()),
            ..Asked::default()
        });
        assert_eq!(plain[0]["message"], "not json");

        // Past keeping: the line of long ago, and all but the last three at
        // info and below; what warned stays.
        logs.prune().unwrap();
        let left = read(Asked {
            level: Some("trace".into()),
            ..Asked::default()
        });
        assert_eq!(messages(&left), ["request", "sql", "outgoing", "not json"]);
        // With nothing past keeping, nothing goes.
        logs.prune().unwrap();
        assert_eq!(
            read(Asked {
                level: Some("trace".into()),
                ..Asked::default()
            })
            .len(),
            4
        );
    }

    /// A write that fails keeps what it held, ahead of what was logged
    /// since, for the next; and still holds only so many.
    #[test]
    fn a_failed_write_keeps_its_lines() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("logs.db");
        let logs = Logs::open(&path, KEEP).unwrap();
        let other = Connection::open(&path).unwrap();
        let today = now();
        logs.keep(&line(&today, "INFO", "first", Some("t"), None));
        other
            .execute_batch("ALTER TABLE line RENAME TO away")
            .unwrap();
        assert!(logs.flush().is_err());
        logs.keep(&line(&today, "INFO", "second", Some("t"), None));
        other
            .execute_batch("ALTER TABLE away RENAME TO line")
            .unwrap();
        logs.flush().unwrap();
        let trace = logs
            .read(&Asked {
                trace: Some("t".into()),
                ..Asked::default()
            })
            .unwrap();
        let messages: Vec<&str> = trace
            .iter()
            .map(|l| l["message"].as_str().unwrap())
            .collect();
        assert_eq!(messages, ["first", "second"]);

        other
            .execute_batch("ALTER TABLE line RENAME TO away")
            .unwrap();
        for _ in 0..PENDING_MAX {
            logs.keep(b"{}");
        }
        assert!(logs.flush().is_err());
        logs.keep(b"{}");
        assert_eq!(logs.pending.lock().unwrap().len(), PENDING_MAX);
    }

    /// A trace reads up to the most a read answers unless asked for
    /// fewer; anything else, the last 100.
    #[test]
    fn a_trace_reads_whole() {
        let dir = tempfile::tempdir().unwrap();
        let logs = Logs::open(&dir.path().join("logs.db"), KEEP).unwrap();
        let today = now();
        for _ in 0..150 {
            logs.keep(&line(&today, "INFO", "request", Some("t"), None));
        }
        logs.flush().unwrap();
        let read = |trace: Option<&str>, limit| {
            let asked = Asked {
                trace: trace.map(str::to_string),
                message: Some("request".into()),
                limit,
                ..Asked::default()
            };
            logs.read(&asked).unwrap().len()
        };
        assert_eq!(read(Some("t"), None), 150);
        assert_eq!(read(Some("t"), Some(10)), 10);
        assert_eq!(read(None, None), 100);
    }

    /// A write that cannot keep up lets the oldest go.
    #[test]
    fn holds_only_so_many() {
        let dir = tempfile::tempdir().unwrap();
        let logs = Logs::open(&dir.path().join("logs.db"), KEEP).unwrap();
        for _ in 0..PENDING_MAX + 1 {
            logs.keep(b"{}");
        }
        assert_eq!(logs.pending.lock().unwrap().len(), PENDING_MAX);
    }

    #[test]
    fn ranks_and_says() {
        let ranks: Vec<i64> = ["ERROR", "warn", "INFO", "DEBUG", "TRACE"]
            .into_iter()
            .map(rank)
            .collect();
        assert_eq!(ranks, [5, 4, 3, 2, 1]);
        let asked = Asked {
            trace: Some("t".into()),
            message: Some("sql".into()),
            level: Some("warn".into()),
            min_ms: Some(100),
            limit: None,
        };
        assert_eq!(asked.said(), "trace=t message=sql level=warn minMs=100");
        assert_eq!(Asked::default().said(), "");
        wrote(Ok(()));
        wrote(Err(rusqlite::Error::InvalidQuery));
    }

    #[tokio::test]
    async fn flushes_and_prunes_as_it_goes() {
        let dir = tempfile::tempdir().unwrap();
        let logs = Arc::new(Logs::open(&dir.path().join("logs.db"), KEEP).unwrap());
        logs.keep(&line(&now(), "INFO", "flushed", None, None));
        let going = tokio::spawn(flushing(logs.clone(), Duration::from_millis(10)));
        tokio::time::sleep(Duration::from_millis(100)).await;
        going.abort();
        let asked = Asked {
            message: Some("flushed".into()),
            ..Asked::default()
        };
        assert_eq!(logs.read(&asked).unwrap().len(), 1);
    }
}