//! `/_status`: whether this server is doing its job, for a monitor off the
//! host to poll. Every page can answer 200 while the feeds that keep it
//! current are dead, or while a mirror's objects can no longer be read, so
//! this checks those:
//!
//! - each feed followed: connected, and an event lately. One feed bringing
//!   events is enough, since each brings all the index needs;
//! - the index answers a read;
//! - with `?repo=<handle or did>/<name>`, that repository's mirror reads as
//!   its code page does: its default branch's commit and tree;
//! - the disk its mirrors and caches are on has room: under 15% free is told
//!   of, under 5% fails. A full disk fails every push a mirror syncs, and
//!   answers 502 for every page that needs one.
//!
//! 200 when all of that holds, else 503, and JSON saying which part failed.
//! `/healthz` stays the liveness check: a dead feed upstream is no reason
//! to restart this server.

use crate::{App, objects, repo_name};
use axum::Json;
use axum::extract::{Query, State};
use axum::http::{StatusCode, header};
use axum::response::{IntoResponse, Response};
use knotless::status::{Check, Disk, Feed, MirrorCheck, Status};
use std::path::Path;
use std::sync::Arc;
use std::sync::atomic::Ordering;
use std::time::{Duration, Instant};

/// How long the firehose may stay silent: it carries every account's
/// commits, about 400 a second (October 2026), its longest gap a sixth of a
/// second, so two minutes without one is a feed gone dead. Two minutes is
/// also how long a watcher waits on a silent connection, through a ping,
/// before it reconnects.
pub const FIREHOSE_STALE: Duration = Duration::from_secs(120);

/// How long Jetstream may stay silent. Asked only for `dev.knotless.*`, it
/// still brings every account's identity, account and sync events: 0.3 a
/// second over 15 minutes (October 2026), its longest gap 11 seconds, so
/// two minutes is ten of the longest and leaves no doubt.
pub const JETSTREAM_STALE: Duration = Duration::from_secs(120);

/// How long a feed named `name` may stay silent before it counts as dead.
pub fn stale_after(name: &str) -> Duration {
    match name {
        "firehose" => FIREHOSE_STALE,
        _ => JETSTREAM_STALE,
    }
}

/// The share of the disk free below which it is low: told of among the
/// problems, still 200. Prod's 150 GB disk then has 22 GB left, a day or
/// more of deploys and runs, time to clear it before it fills.
pub const DISK_LOW: f64 = 0.15;

/// The share free below which the disk counts as full and `/_status`
/// fails: 7.5 GB of prod's, a single cargo build's worth.
pub const DISK_FULL: f64 = 0.05;

/// How a disk with `free` of its `total` bytes is: `ok`, `low` or `full`.
pub fn disk_state(free: u64, total: u64) -> &'static str {
    let share = free as f64 / total.max(1) as f64;
    match share {
        s if s < DISK_FULL => "full",
        s if s < DISK_LOW => "low",
        _ => "ok",
    }
}

/// The disk `path` is on, asked of its nearest directory that exists.
pub(crate) fn disk(path: &Path) -> Disk {
    let at = path.ancestors().find(|p| p.exists()).unwrap_or(path);
    let mut disk = Disk {
        path: path.display().to_string(),
        state: "unknown".into(),
        ok: false,
        free_bytes: 0,
        total_bytes: 0,
        error: None,
    };
    match rustix::fs::statvfs(at) {
        Ok(space) => {
            disk.free_bytes = space.f_bavail * space.f_frsize;
            disk.total_bytes = space.f_blocks * space.f_frsize;
            let state = disk_state(disk.free_bytes, disk.total_bytes);
            (disk.state, disk.ok) = (state.into(), state != "full");
        }
        Err(err) => disk.error = Some(format!("{}: {err}", at.display())),
    }
    disk
}

/// What `/_status` says of a disk that is not `ok`, among its problems.
fn disk_problem(disk: &Disk) -> Option<String> {
    let free = 100 * disk.free_bytes / disk.total_bytes.max(1);
    match (disk.state.as_str(), &disk.error) {
        ("ok", _) => None,
        (_, Some(err)) => Some(format!("disk: {err}")),
        (state, None) => Some(format!("disk {state}: {free}% free")),
    }
}

/// A feed as its watcher last told of it.
pub(crate) struct Followed {
    url: String,
    connected: bool,
    /// When it was first followed, which a feed that never brought an
    /// event is silent since.
    since: Instant,
    last: Option<Instant>,
    events: u64,
    seq: Option<i64>,
}

impl App {
    /// That the feed `name` is followed from `url`, not yet connected.
    pub fn follows(&self, name: &'static str, url: &str) {
        self.feeds.lock().unwrap().insert(
            name,
            Followed {
                url: url.to_string(),
                connected: false,
                since: Instant::now(),
                last: None,
                events: 0,
                seq: None,
            },
        );
    }

    /// That the feed `name` connected, or dropped. Mirrors trust the feeds
    /// to tell them of writes while any is connected.
    pub fn connected(&self, name: &str, connected: bool) {
        let mut feeds = self.feeds.lock().unwrap();
        let feed = feeds.get_mut(name).into_iter();
        feed.filter(|f| f.connected != connected).for_each(|f| {
            crate::telemetry::feed(name, &f.url, connected);
            f.connected = connected;
        });
        let any = feeds.values().any(|f| f.connected);
        self.watching.store(any, Ordering::Relaxed);
    }

    /// That the feed `name` brought an event, numbered `seq`.
    pub fn heard(&self, name: &str, seq: Option<i64>) {
        let mut feeds = self.feeds.lock().unwrap();
        feeds.get_mut(name).into_iter().for_each(|feed| {
            feed.last = Some(Instant::now());
            feed.events += 1;
            feed.seq = seq.or(feed.seq);
        });
    }

    /// The feeds, as `/_status` tells of them.
    pub(crate) fn feeds(&self) -> Vec<Feed> {
        let feeds = self.feeds.lock().unwrap();
        let mut out = Vec::new();
        for (name, f) in feeds.iter() {
            let stale = self.stale.unwrap_or(stale_after(name));
            let silent = f.last.unwrap_or(f.since).elapsed();
            out.push(Feed {
                name: name.to_string(),
                url: f.url.clone(),
                ok: f.connected && silent <= stale,
                connected: f.connected,
                silent_seconds: silent.as_secs(),
                stale_after_seconds: stale.as_secs(),
                events: f.events,
                seq: f.seq,
            });
        }
        out
    }

    /// How this server is, asking after `repo`'s mirror too if given.
    pub async fn status(&self, repo: Option<&str>) -> Status {
        let feeds = self.feeds();
        let mut problems = Vec::new();
        for f in feeds.iter().filter(|f| !f.ok) {
            let why = match f.connected {
                true => format!("no event for {}s", f.silent_seconds),
                false => "not connected".to_string(),
            };
            problems.push(format!("{}: {why}", f.name));
        }
        let fed = feeds.is_empty() || feeds.iter().any(|f| f.ok);
        if !fed {
            problems.push("no feed is bringing events".into());
        }
        let index = self.index.as_ref().map(|i| match i.check() {
            Ok(_) => Check {
                ok: true,
                error: None,
            },
            Err(err) => Check {
                ok: false,
                error: Some(err.to_string()),
            },
        });
        let mirror = match repo {
            Some(repo) => Some(probe(self, repo).await),
            None => None,
        };
        let failed = [
            index
                .as_ref()
                .and_then(|i| i.error.clone())
                .map(|e| format!("index: {e}")),
            mirror
                .as_ref()
                .and_then(|m| Some(format!("mirror {}: {}", m.repo, m.error.as_ref()?))),
        ];
        problems.extend(failed.into_iter().flatten());
        let disk = self.disk.as_deref().map(disk);
        problems.extend(disk.as_ref().and_then(disk_problem));
        let code = match self.code.as_deref() {
            Some(code) => Some(code.check().await),
            None => None,
        };
        let failing = code.as_ref().and_then(|c| c.error.as_deref());
        problems.extend(failing.map(|e| format!("code: {e}")));
        let parts = index.as_ref().is_none_or(|i| i.ok)
            && mirror.as_ref().is_none_or(|m| m.ok)
            && code.as_ref().is_none_or(|c| c.ok);
        Status {
            ok: fed && parts && disk.as_ref().is_none_or(|d| d.ok),
            problems,
            uptime_seconds: self.started.elapsed().as_secs(),
            feeds,
            index,
            mirror,
            disk,
            code,
        }
    }
}

/// What `/_status` is asked.
#[derive(serde::Deserialize)]
pub(crate) struct Ask {
    repo: Option<String>,
}

/// `GET /_status`.
pub(crate) async fn status(State(app): State<Arc<App>>, Query(ask): Query<Ask>) -> Response {
    let status = app.status(ask.repo.as_deref()).await;
    let code = match status.ok {
        true => StatusCode::OK,
        false => StatusCode::SERVICE_UNAVAILABLE,
    };
    (code, [(header::CACHE_CONTROL, "no-store")], Json(status)).into_response()
}

/// `repo`'s mirror, read as its code page reads it.
async fn probe(app: &App, repo: &str) -> MirrorCheck {
    let mut check = MirrorCheck {
        repo: repo.to_string(),
        ok: false,
        error: None,
        commit: None,
        packs: None,
    };
    match read(app, repo).await {
        Ok((commit, packs)) => {
            check.ok = true;
            check.commit = Some(commit);
            check.packs = Some(packs);
        }
        Err(err) => check.error = Some(err),
    }
    check
}

/// The commit `repo`'s default branch is at, its tree read, and how many
/// packs its mirror holds. A mirror with more packs than its open handle
/// has room for fails here, as its pages do.
async fn read(app: &App, repo: &str) -> Result<(String, usize), String> {
    let (authority, name) = repo
        .split_once('/')
        .ok_or("?repo= takes <handle or did>/<name>")?;
    let found = app.located(authority, repo_name(name)).await;
    let git = found.map_err(|e| e.1)?.git;
    let packs = std::fs::read_dir(git.dir.join("objects/pack"))
        .into_iter()
        .flatten()
        .flatten()
        .filter(|e| e.path().extension().is_some_and(|x| x == "pack"))
        .count();
    let objects = objects::open(&git.dir).map_err(text)?;
    let commit = objects.resolve("HEAD").map_err(text)?;
    objects.tree(&commit, "").map_err(text)?;
    Ok((commit, packs))
}

fn text(err: anyhow::Error) -> String {
    format!("{err:#}")
}

#[cfg(test)]
mod tests {
    use super::*;
    use knotless::env::{Ctx, Env};

    fn app(index: Option<crate::index::Index>) -> Arc<App> {
        let ctx = Ctx::new(Env::from_vars(|_| None));
        App::build(ctx, Duration::ZERO, None, index)
    }

    #[tokio::test]
    async fn feeds() {
        // Nothing followed: nothing to be stale.
        let mut alone = app(None);
        let s = alone.status(None).await;
        assert!(s.ok, "{s:?}");
        assert!(s.feeds.is_empty());
        assert_eq!((s.index, s.mirror), (None, None));
        let s = alone.status(Some("unnamed")).await;
        assert!(!s.ok);
        assert_eq!(
            s.problems,
            ["mirror unnamed: ?repo= takes <handle or did>/<name>"]
        );
        Arc::get_mut(&mut alone).unwrap().stale = Some(Duration::from_millis(200));

        // Followed but never connected: not ok, and mirrors fall back to the TTL.
        alone.follows("jetstream", "wss://jetstream");
        alone.watching.store(true, Ordering::Relaxed);
        alone.connected("jetstream", false);
        assert!(!alone.watching.load(Ordering::Relaxed));
        let s = alone.status(None).await;
        assert!(!s.ok);
        assert_eq!(
            s.problems,
            ["jetstream: not connected", "no feed is bringing events"]
        );

        // Connected and bringing events: ok.
        alone.connected("jetstream", true);
        alone.heard("jetstream", Some(7));
        alone.heard("jetstream", None);
        alone.heard("elsewhere", Some(1));
        alone.connected("elsewhere", false);
        assert!(alone.watching.load(Ordering::Relaxed));
        let s = alone.status(None).await;
        assert!(s.ok, "{s:?}");
        let f = &s.feeds[0];
        assert_eq!((f.events, f.seq, f.connected), (2, Some(7), true));
        assert_eq!(
            (f.url.as_str(), f.stale_after_seconds),
            ("wss://jetstream", 0)
        );

        // Connected, but silent past the threshold: stale.
        tokio::time::sleep(Duration::from_millis(300)).await;
        let s = alone.status(None).await;
        assert!(!s.ok);
        assert_eq!(s.problems[0], "jetstream: no event for 0s");

        // One feed bringing events is enough; each is still told of.
        alone.follows("firehose", "wss://relay");
        alone.connected("firehose", true);
        alone.heard("firehose", Some(1));
        let s = alone.status(None).await;
        assert!(s.ok, "{s:?}");
        assert_eq!(s.problems, ["jetstream: no event for 0s"]);
        assert_eq!(s.feeds.len(), 2);
    }

    #[test]
    fn thresholds() {
        assert_eq!(stale_after("firehose"), FIREHOSE_STALE);
        assert_eq!(stale_after("jetstream"), JETSTREAM_STALE);
        let app = app(None);
        app.follows("firehose", "wss://relay");
        assert_eq!(app.feeds()[0].stale_after_seconds, FIREHOSE_STALE.as_secs());
    }

    #[test]
    fn disk_thresholds() {
        assert_eq!(disk_state(100, 100), "ok");
        assert_eq!(disk_state(15, 100), "ok");
        assert_eq!(disk_state(14, 100), "low");
        assert_eq!(disk_state(5, 100), "low");
        assert_eq!(disk_state(4, 100), "full");
        assert_eq!(disk_state(0, 0), "full");
        let at = |state: &str, free: u64, error: Option<&str>| Disk {
            path: "/data".into(),
            state: state.into(),
            ok: state != "full",
            free_bytes: free,
            total_bytes: 100,
            error: error.map(String::from),
        };
        assert_eq!(disk_problem(&at("ok", 50, None)), None);
        assert_eq!(
            disk_problem(&at("low", 12, None)).unwrap(),
            "disk low: 12% free"
        );
        assert_eq!(
            disk_problem(&at("full", 3, None)).unwrap(),
            "disk full: 3% free"
        );
        assert_eq!(
            disk_problem(&at("unknown", 0, Some("gone"))).unwrap(),
            "disk: gone"
        );
    }

    #[tokio::test]
    async fn disk_asked() {
        // A directory not made yet is asked of the nearest that is.
        let dir = tempfile::tempdir().unwrap();
        let mut app = app(None);
        let path = dir.path().join("cache/serve");
        Arc::get_mut(&mut app).unwrap().disk = Some(path.clone());
        let d = app.status(None).await.disk.unwrap();
        assert_eq!(d.path, path.display().to_string());
        assert!(d.total_bytes > 0 && d.free_bytes <= d.total_bytes, "{d:?}");
        assert_eq!(d.state, disk_state(d.free_bytes, d.total_bytes));
        assert_eq!(d.error, None);

        // Nowhere to ask: unknown, and failing.
        let d = disk(Path::new("nowhere/at/all"));
        assert_eq!((d.state.as_str(), d.ok), ("unknown", false));
        assert!(d.error.unwrap().contains(": "));
    }

    #[tokio::test]
    async fn code() {
        // Zoekt gone: told of, and failing.
        let dir = tempfile::tempdir().unwrap();
        std::fs::create_dir(dir.path().join("meta")).unwrap();
        for shard in [
            "a_2fk_v16.00000.zoekt",
            "a_2fk_v16.00001.zoekt",
            "b_2fk_v16.00000.zoekt",
        ] {
            std::fs::write(dir.path().join(shard), "1234").unwrap();
        }
        let mut app = app(None);
        let code = crate::code::dead(dir.path());
        Arc::get_mut(&mut app).unwrap().code = Some(Arc::new(code));
        let s = app.status(None).await;
        assert!(!s.ok);
        let c = s.code.unwrap();
        assert_eq!((c.ok, c.repos, c.shards, c.bytes), (false, 2, 3, 12));
        assert_eq!(s.problems, [format!("code: {}", c.error.unwrap())]);
    }

    #[tokio::test]
    async fn index() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("index.db");
        let app = app(Some(crate::index::Index::open(&path).unwrap()));
        let s = app.status(None).await;
        assert!(s.ok, "{s:?}");
        assert_eq!(
            s.index.unwrap(),
            Check {
                ok: true,
                error: None
            }
        );

        // Its tables gone from under it: the index answers no more.
        let other = rusqlite::Connection::open(&path).unwrap();
        other.execute_batch("DROP TABLE stream").unwrap();
        let s = app.status(None).await;
        assert!(!s.ok);
        let error = s.index.unwrap().error.unwrap();
        assert_eq!(s.problems, [format!("index: {error}")]);
    }
}