//! What happens as it happens, for whoever is looking: the output of the
//! checks running now, and the spaces written to, for their runners.
//!
//! A check's record is written as it starts and as it ends, and its log is
//! a blob of the end, so a page showing a check that runs would show
//! nothing of it. Its runner sends this server what its steps print as they
//! print it (`dev.knotless.putRunOutput`), and the server keeps the tail of
//! each in memory for the pages that follow it ([`crate::web`]'s `/_live`),
//! never in the index: the record and its log stay what is kept. A runner
//! signs for itself with a service token, and sends only its own checks'.
//!
//! Nothing written in a space reaches Jetstream, so a runner of a private
//! repository would have to ask the space whether anything was. The
//! space's authority tells this server of each write instead
//! (`notifyWrite`), and the server passes that on to the runners the
//! space's owner invited (`dev.knotless.awaitSpaceWrites`), asking, as they
//! do, from where they last heard. A word passed on is only a reason to
//! look: the runner reads the space as a member before running anything.

use crate::gateway::caller;
use crate::moderation::{bad, index};
use crate::{App, AppError};
use axum::Json;
use axum::extract::{Query, State};
use axum::http::{HeaderMap, StatusCode, header};
use knotless::records::nsid;
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use std::collections::{HashMap, VecDeque};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use tokio::sync::{Notify, broadcast};
use tokio_util::sync::CancellationToken;

pub use knotless::invites::{OUTPUT, WRITES};

/// The most of a run's output kept: as much as its log holds.
pub const KEPT: usize = 1_000_000;

/// The largest output sent at once, escaped as JSON.
pub const BODY: usize = 256 << 10;

/// The most runs kept at once, and of one runner's.
const RUNS: usize = 1024;
const RUNS_EACH: usize = 128;

/// How long a run nothing was heard of is kept.
pub const IDLE: Duration = Duration::from_secs(600);

/// How many spaces' writes are remembered, for the runners asking from
/// where they last heard.
const WRITTEN: usize = 4096;

/// How long a runner's ask for writes waits for one.
pub const WAIT: Duration = Duration::from_secs(25);

/// How long a service token checked once is taken as checked.
const CHECKED: Duration = Duration::from_secs(60);

/// What this server holds of what happens now.
pub struct Live {
    runs: Mutex<HashMap<String, Arc<Run>>>,
    writes: Mutex<Writes>,
    written: Notify,
    /// The service tokens checked lately, and whose they are.
    tokens: Mutex<HashMap<String, (String, Instant)>>,
    /// Cancelled as the server stops, ending what is followed.
    pub stopping: CancellationToken,
}

impl Default for Live {
    fn default() -> Live {
        Live {
            runs: Mutex::default(),
            writes: Mutex::new(Writes {
                life: knotless::records::now(),
                ..Writes::default()
            }),
            written: Notify::new(),
            tokens: Mutex::default(),
            stopping: CancellationToken::new(),
        }
    }
}

/// The spaces written to lately, numbered from this server's start.
#[derive(Default)]
struct Writes {
    /// When this server started: a cursor of another start's is no
    /// place to go on from.
    life: String,
    last: u64,
    kept: VecDeque<(u64, String)>,
}

/// A check running, as its runner sent its output.
pub struct Run {
    kept: Mutex<Kept>,
    /// Rung as anything is new: each follower reads what is from what
    /// it had, so one that falls behind loses nothing kept.
    pub news: broadcast::Sender<()>,
}

/// The tail of a run's output, and where it starts in the whole.
#[derive(Default)]
pub struct Kept {
    pub start: u64,
    pub text: Vec<u8>,
    pub steps: Value,
    /// The step going, by name.
    pub going: Option<String>,
    pub ended: bool,
    runner: String,
    heard: Option<Instant>,
    /// When the run sending started, as its runner says.
    started: Option<String>,
}

impl Kept {
    /// Where the output ends in the whole.
    pub fn end(&self) -> u64 {
        self.start + self.text.len() as u64
    }

    /// Takes in `text`, from `at` in the whole: what was had already is
    /// passed over, and a gap starts the tail anew. Returns what is new.
    fn take<'a>(&mut self, at: u64, text: &'a [u8]) -> &'a [u8] {
        if at > self.end() {
            self.text.clear();
            self.start = at;
        }
        let had = usize::try_from(self.end() - at).unwrap_or(usize::MAX);
        let new = text.get(had..).unwrap_or_default();
        self.text.extend_from_slice(new);
        let over = self.text.len().saturating_sub(KEPT);
        self.text.drain(..over);
        self.start += over as u64;
        new
    }

    /// What was sent from `from` on, all of it if before what is kept,
    /// and where that starts.
    pub fn since(&self, from: Option<u64>) -> (u64, &[u8]) {
        let from = from.unwrap_or(0).clamp(self.start, self.end());
        let skip = usize::try_from(from - self.start).unwrap_or_default();
        (from, &self.text[skip..])
    }
}

impl Run {
    pub fn kept(&self) -> std::sync::MutexGuard<'_, Kept> {
        self.kept.lock().unwrap()
    }
}

impl Live {
    /// The run of the check `uri`, kept from now on if it was not: a page
    /// may follow a check before its runner sends anything.
    pub fn run(&self, uri: &str) -> Option<Arc<Run>> {
        let mut runs = self.runs.lock().unwrap();
        runs.retain(|_, r| r.news.receiver_count() > 0 || !r.kept().idle());
        if let Some(run) = runs.get(uri) {
            return Some(run.clone());
        }
        if runs.len() >= RUNS {
            return None;
        }
        let run = Arc::new(Run {
            kept: Mutex::default(),
            news: broadcast::channel(16).0,
        });
        runs.insert(uri.to_string(), run.clone());
        Some(run)
    }

    /// A run of the check `uri` anew, in place of `old`, another run of
    /// it: those following `old` are told it ended, and look again.
    fn anew(&self, uri: &str, old: &Run) -> Arc<Run> {
        old.kept().ended = true;
        let _ = old.news.send(());
        let run = Arc::new(Run {
            kept: Mutex::default(),
            news: broadcast::channel(16).0,
        });
        let mut runs = self.runs.lock().unwrap();
        runs.insert(uri.to_string(), run.clone());
        run
    }

    /// Tells the runners following them that the space `uri` was
    /// written to.
    pub fn written(&self, uri: &str) {
        let mut writes = self.writes.lock().unwrap();
        writes.last += 1;
        let last = writes.last;
        writes.kept.push_back((last, uri.to_string()));
        if writes.kept.len() > WRITTEN {
            writes.kept.pop_front();
        }
        drop(writes);
        self.written.notify_waiters();
    }

    /// The spaces written to after `cursor`, of those `wanted` says, and
    /// where they end; or `None` for a cursor of no place in what is
    /// remembered, which may have missed any.
    fn writes_since(
        &self,
        cursor: Option<&str>,
        wanted: &dyn Fn(&str) -> bool,
    ) -> (Option<Vec<Write>>, String) {
        let writes = self.writes.lock().unwrap();
        let now = format!("{}.{}", writes.life, writes.last);
        let at = cursor.and_then(|c| c.strip_prefix(&format!("{}.", writes.life)));
        let at = at.and_then(|n| n.parse::<u64>().ok());
        let first = writes.kept.front().map_or(writes.last + 1, |(n, _)| *n);
        let Some(at) = at.filter(|at| *at + 1 >= first && *at <= writes.last) else {
            return (None, now);
        };
        let spaces = writes.kept.iter().filter(|(n, _)| *n > at);
        let spaces = spaces
            .filter(|(_, s)| wanted(s))
            .map(|(_, s)| Write { space: s.clone() });
        let mut spaces: Vec<Write> = spaces.collect();
        spaces.sort();
        spaces.dedup();
        (Some(spaces), now)
    }

    /// The DID that signed the request's service token for `lxm`: checked
    /// once a minute at most, as a runner sends output every second.
    async fn caller(&self, app: &App, headers: &HeaderMap, lxm: &str) -> Result<String, AppError> {
        let token = headers.get(header::AUTHORIZATION);
        let token = token.and_then(|t| t.to_str().ok()).unwrap_or_default();
        let key = format!("{lxm} {token}");
        let cached = self.tokens.lock().unwrap().get(&key).cloned();
        if let Some((did, _)) = cached.filter(|(_, at)| at.elapsed() < CHECKED) {
            return Ok(did);
        }
        let did = caller(app, headers, lxm).await?;
        let mut tokens = self.tokens.lock().unwrap();
        tokens.retain(|_, (_, at)| at.elapsed() < CHECKED);
        tokens.insert(key, (did.clone(), Instant::now()));
        Ok(did)
    }
}

impl Kept {
    fn idle(&self) -> bool {
        self.heard.is_none_or(|at| at.elapsed() > IDLE)
    }
}

#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct Output {
    check: String,
    offset: u64,
    #[serde(default)]
    text: String,
    steps: Option<Value>,
    going: Option<String>,
    #[serde(default)]
    ended: bool,
    started_at: Option<String>,
}

/// `dev.knotless.putRunOutput`: what a run's steps printed since its
/// runner last sent, and its steps' ends, from the runner that filed its
/// check. Answers whether anyone follows it, so that a runner may send
/// less often when no one does.
pub async fn put_run_output(
    State(app): State<Arc<App>>,
    headers: HeaderMap,
    Json(out): Json<Output>,
) -> Result<Json<Value>, AppError> {
    let runner = app.live.caller(&app, &headers, OUTPUT).await?;
    match knotless::forge::parse_uri(&out.check) {
        Ok((did, nsid::CHECK, _)) if did == runner => {}
        _ => return Err(bad("a runner sends the output of its own checks")),
    }
    let held = app.live.runs.lock().unwrap();
    let theirs = held.values().filter(|r| r.kept().runner == runner).count();
    let full = theirs >= RUNS_EACH && !held.contains_key(&out.check);
    drop(held);
    let busy = AppError(
        StatusCode::TOO_MANY_REQUESTS,
        "too many runs at once".into(),
    );
    let mut run = app.live.run(&out.check).filter(|_| !full).ok_or(busy)?;
    // Another run of the same check, as a rerun at the same commit is,
    // keeps nothing of the last: its output starts again from nothing.
    let started = out.started_at.as_deref();
    let other = run
        .kept()
        .started
        .as_deref()
        .zip(started)
        .is_some_and(|(a, b)| a != b);
    if other {
        run = app.live.anew(&out.check, &run);
    }
    let mut kept = run.kept();
    (kept.runner, kept.heard) = (runner, Some(Instant::now()));
    if out.started_at.is_some() {
        kept.started = out.started_at;
    }
    let mut new = !kept.take(out.offset, out.text.as_bytes()).is_empty();
    // The step going is told with the steps that ended.
    let stood = |s: &Value| *s != kept.steps || out.going != kept.going;
    if let Some(steps) = out.steps.filter(stood) {
        (kept.steps, kept.going, new) = (steps, out.going, true);
    }
    new |= out.ended && !kept.ended;
    kept.ended |= out.ended;
    drop(kept);
    if new {
        let _ = run.news.send(());
    }
    let watched = run.news.receiver_count() > 0;
    Ok(Json(json!({ "watched": watched })))
}

/// A space written to, as `dev.knotless.awaitSpaceWrites` tells of it.
#[derive(Serialize, PartialEq, Eq, PartialOrd, Ord, Debug)]
pub struct Write {
    space: String,
}

#[derive(Deserialize)]
pub struct WritesParams {
    cursor: Option<String>,
}

/// `dev.knotless.awaitSpaceWrites`: the spaces written to since `cursor`,
/// of those whose owners invited the runner asking, as soon as there are
/// any, or none after a while. One whose cursor is of no place this server
/// remembers is told it may have missed any.
pub async fn await_space_writes(
    State(app): State<Arc<App>>,
    headers: HeaderMap,
    Query(p): Query<WritesParams>,
) -> Result<Json<Value>, AppError> {
    let index = index(&app)?;
    let runner = app.live.caller(&app, &headers, WRITES).await?;
    let wanted = |space: &str| {
        let owner = space.strip_prefix("at://").and_then(|s| s.split_once('/'));
        let name = space.rsplit('/').next().unwrap_or_default();
        let invited = owner.map(|(did, _)| index.invited(&runner, did, name));
        invited.is_some_and(|i| i.unwrap_or(false))
    };
    let deadline = tokio::time::Instant::now() + WAIT;
    let mut cursor = p.cursor;
    loop {
        let woken = app.live.written.notified();
        let (spaces, now) = app.live.writes_since(cursor.as_deref(), &wanted);
        let missed = spaces.is_none();
        let spaces = spaces.unwrap_or_default();
        let over = tokio::time::Instant::now() >= deadline;
        if missed || !spaces.is_empty() || over {
            let answer = json!({ "cursor": now, "writes": spaces, "missed": missed });
            return Ok(Json(answer));
        }
        cursor = Some(now);
        let _ = tokio::time::timeout_at(deadline, woken).await;
    }
}

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

    #[test]
    fn a_tail_takes_what_is_new() {
        let mut kept = Kept::default();
        assert_eq!(kept.take(0, b"one\n"), b"one\n");
        // Sent again, nothing is new; overlapping, only what is.
        assert_eq!(kept.take(0, b"one\n"), b"");
        assert_eq!(kept.take(2, b"e\ntwo\n"), b"two\n");
        assert_eq!(kept.since(None), (0, &b"one\ntwo\n"[..]));
        assert_eq!(kept.since(Some(4)), (4, &b"two\n"[..]));
        assert_eq!(kept.since(Some(99)), (8, &b""[..]));
        // A gap starts it anew.
        assert_eq!(kept.take(20, b"far\n"), b"far\n");
        assert_eq!((kept.start, kept.end()), (20, 24));
        assert_eq!(kept.since(Some(3)), (20, &b"far\n"[..]));
        // Only the tail is kept.
        let long = vec![b'x'; KEPT + 10];
        kept.take(24, &long);
        assert_eq!(kept.text.len(), KEPT);
        assert_eq!(kept.end(), 24 + KEPT as u64 + 10);
    }

    #[test]
    fn writes_go_on_from_a_cursor() {
        let live = Live::default();
        let all = |_: &str| true;
        let (none, start) = live.writes_since(None, &all);
        assert!(none.is_none());
        assert_eq!(live.writes_since(Some(&start), &all).0, Some(vec![]));
        live.written("b");
        live.written("a");
        live.written("b");
        let (spaces, now) = live.writes_since(Some(&start), &all);
        let space = |s: &str| Write { space: s.into() };
        assert_eq!(spaces, Some(vec![space("a"), space("b")]));
        let only = |s: &str| s == "a";
        assert_eq!(
            live.writes_since(Some(&start), &only).0,
            Some(vec![space("a")])
        );
        assert_eq!(live.writes_since(Some(&now), &all).0, Some(vec![]));
        // Another start's, or one past what is remembered, may have missed any.
        assert!(live.writes_since(Some("then.1"), &all).0.is_none());
        assert!(
            live.writes_since(Some(&format!("{now}0")), &all)
                .0
                .is_none()
        );
        for _ in 0..=WRITTEN {
            live.written("c");
        }
        assert!(live.writes_since(Some(&start), &all).0.is_none());
    }

    #[test]
    fn runs_are_kept_while_followed_or_heard_of() {
        let live = Live::default();
        let run = live.run("at://r/c/1").unwrap();
        // Never heard of and followed by no one, it goes as another comes.
        drop(run);
        live.run("at://r/c/2");
        assert!(!live.runs.lock().unwrap().contains_key("at://r/c/1"));
        let followed = live.run("at://r/c/2").unwrap().news.subscribe();
        // Followed, it is the same run asked for again.
        let again = live.run("at://r/c/2").unwrap();
        assert_eq!(again.news.receiver_count(), 1);
        for n in 3..RUNS + 2 {
            live.run(&format!("at://r/c/{n}")).unwrap().kept().heard = Some(Instant::now());
        }
        assert!(live.run("at://r/c/full").is_none());
        drop(followed);
    }
}