//! What the runner tells the appview (`KNOTLESS_APPVIEW`) as it happens,
//! and what it hears from it.
//!
//! A check's record says nothing of a run until it ends, so each run sends
//! the appview what its steps print as they print it, every second while
//! someone follows it there and every few otherwise
//! (`dev.knotless.putRunOutput`). Only whole lines go, each with the run's
//! secrets masked as its log's are, so no secret is ever sent split where
//! masking would miss it, and with the steps that ended and the one going.
//! What fails to send is sent again next time, up to as much as the appview
//! keeps; a run never waits on the appview, and the log is what is kept.
//!
//! The appview is told of each write in a private repository's space,
//! which Jetstream never carries, and passes that on to the runners the
//! space's owner invited (`dev.knotless.awaitSpaceWrites`): a runner hears
//! of a push there at once, and asks the space itself only now and then.

use crate::check::{self, Step};
use anyhow::{Context, Result};
use knotless::env::Ctx;
use knotless::identity::Identity;
use knotless::session;
use knotless::xrpc::{Auth, Xrpc};
use serde::Deserialize;
use serde_json::{Value, json};
use std::collections::HashMap;
use std::io::{Read, Seek, SeekFrom};
use std::path::PathBuf;
use std::sync::Arc;
use std::time::{Duration, Instant, SystemTime};
use tokio::sync::{Mutex, OnceCell, watch};

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

/// How often output is sent while someone follows the run, and while
/// no one does.
pub const WATCHED: Duration = Duration::from_secs(1);
pub const UNWATCHED: Duration = Duration::from_secs(5);

/// How long a run's last output may take to send before it is filed.
pub const LAST: Duration = Duration::from_secs(5);

/// How often a private repository's space is asked itself whether it was
/// written to while the appview passes that on, for what it missed.
pub const WOKEN_POLL: Duration = Duration::from_secs(60);

/// How often a space is asked to tell the appview of its writes: the
/// authority keeps the ask for a day.
pub const REGISTER: Duration = Duration::from_secs(12 * 60 * 60);

/// How long after the appview could not be asked for writes it is asked
/// again.
pub const RETRY: Duration = Duration::from_secs(30);

/// How long a service token is asked for, and used.
const TOKEN_FOR: Duration = Duration::from_secs(30 * 60);
const TOKEN_USED: Duration = Duration::from_secs(25 * 60);

/// The most of a line held back for its end: past it, the line is let go
/// unsent rather than sent in part.
const LINE_MAX: usize = 1_000_000;

/// The most of a log read at once: the rest waits for the next send.
const READ_MAX: u64 = 256 << 10;

/// The most text sent at once, as `dev.knotless.putRunOutput` takes it.
const PIECE: usize = 16_000;

/// The most text held while the appview does not take it, as much as it
/// keeps of a run: past it, the oldest is let go.
const UNSENT_MAX: usize = 1_000_000;

/// The appview, as this runner's account calls it.
pub struct Appview {
    ctx: Ctx,
    identity: Identity,
    xrpc: Xrpc,
    url: String,
    did: OnceCell<String>,
    tokens: Mutex<HashMap<&'static str, (Auth, Instant)>>,
}

/// The spaces written to since a cursor, as the appview tells them.
#[derive(Deserialize, Debug)]
pub struct Writes {
    pub cursor: String,
    #[serde(default)]
    pub writes: Vec<Written>,
    /// Set when the cursor was of no place the appview remembers.
    #[serde(default)]
    pub missed: bool,
}

#[derive(Deserialize, Debug)]
pub struct Written {
    pub space: String,
}

impl Appview {
    /// The appview `ctx` names, as `identity` calls it, if it names one.
    pub fn new(ctx: &Ctx, identity: &Identity) -> Option<Arc<Appview>> {
        let url = ctx.env.appview.clone()?;
        Some(Arc::new(Appview {
            ctx: ctx.clone(),
            identity: identity.clone(),
            xrpc: Xrpc::new(ctx.http.clone(), &url),
            url,
            did: OnceCell::new(),
            tokens: Mutex::default(),
        }))
    }

    /// Its DID, as its document says, read once.
    pub async fn did(&self) -> Result<&str> {
        let did = self
            .did
            .get_or_try_init(|| knotless::moderation::appview_did(&self.ctx, &self.url));
        Ok(did.await?.as_str())
    }

    /// A token for calling `lxm`, asked for once in a while.
    async fn auth(&self, lxm: &'static str) -> Result<Auth> {
        let mut tokens = self.tokens.lock().await;
        if let Some((auth, _)) = tokens.get(lxm).filter(|(_, at)| at.elapsed() < TOKEN_USED) {
            return Ok(auth.clone());
        }
        let me = session::ensure(&self.ctx, &self.identity).await?;
        let now = SystemTime::now().duration_since(SystemTime::UNIX_EPOCH)?;
        let exp = (now + TOKEN_FOR).as_secs();
        let did = self.did().await?;
        let token = knotless::auth::lasting_token(&self.ctx, &me, did, lxm, exp).await?;
        let auth = Auth::Bearer(token);
        tokens.insert(lxm, (auth.clone(), Instant::now()));
        Ok(auth)
    }

    /// Sends `body`, a run's output since it last sent. Answers whether
    /// anyone follows the run.
    pub async fn put_output(&self, body: &Value) -> Result<bool> {
        #[derive(Deserialize)]
        struct Out {
            watched: bool,
        }
        let auth = self.auth(OUTPUT).await?;
        let out: Out = self.xrpc.procedure(OUTPUT, Some(body), Some(&auth)).await?;
        Ok(out.watched)
    }

    /// The spaces written to since `cursor`, waiting a while for one.
    pub async fn await_writes(&self, cursor: Option<&str>) -> Result<Writes> {
        let auth = self.auth(WRITES).await?;
        let params: Vec<(&str, &str)> = cursor.map(|c| ("cursor", c)).into_iter().collect();
        self.xrpc.query(WRITES, &params, Some(&auth)).await
    }

    /// The service a space's authority tells this appview of writes at:
    /// its DID, and the entry in its document.
    pub async fn service(&self) -> Result<String> {
        Ok(format!("{}{}", self.did().await?, knotless::space::APPVIEW))
    }
}

/// A run's log as it is read: where reading it got to, and the line
/// begun but not ended.
#[derive(Default)]
pub struct Tail {
    read: u64,
    line: Vec<u8>,
}

impl Tail {
    /// Takes in what the log has past where it was read to, and returns
    /// the lines ended, masked, to send; with `ended`, the rest too. A
    /// line longer than [`LINE_MAX`] is let go unsent.
    pub fn take(&mut self, more: &[u8], secrets: &[String], ended: bool) -> String {
        self.read += more.len() as u64;
        self.line.extend_from_slice(more);
        let end = match ended {
            true => self.line.len(),
            false => self
                .line
                .iter()
                .rposition(|b| *b == b'\n' || *b == b'\r')
                .map_or(0, |i| i + 1),
        };
        let whole: Vec<u8> = self.line.drain(..end).collect();
        if self.line.len() > LINE_MAX {
            self.line.clear();
        }
        String::from_utf8_lossy(&check::redact(&whole, secrets)).into_owned()
    }
}

/// How a run stands as its steps go: those that ended, and the one going.
#[derive(Default, Clone, PartialEq, Debug)]
pub struct Progress {
    pub steps: Vec<Step>,
    pub going: Option<String>,
}

/// What a run has for the appview, held until it is taken in.
#[derive(Default)]
struct Outbox {
    tail: Tail,
    /// Where `text` starts in all the run sent.
    at: u64,
    /// What was read and not yet taken in.
    text: String,
    /// How the run stands now, and as the appview last took it.
    now: Progress,
    told: Progress,
    over: bool,
    /// Set when a send of this round failed: the rest waits for the next.
    refused: bool,
    /// Set once the last send is taken in, or fails with the run over.
    done: bool,
}

impl Outbox {
    /// Starts a round: `more` of the log, how the run stands, and whether
    /// it is over.
    fn round(&mut self, more: &[u8], secrets: &[String], now: Progress, over: bool) {
        let text = self.tail.take(more, secrets, over);
        self.text.push_str(&text);
        let cut = ceil(&self.text, self.text.len().saturating_sub(UNSENT_MAX));
        self.text.drain(..cut);
        self.at += cut as u64;
        (self.now, self.over, self.refused) = (now, over, false);
    }

    /// The next send of this round, if it has one: a piece of the text,
    /// the last with how the run stands, and on the run's very last that
    /// it ended. How the run stands goes every time, for an appview that
    /// lost it in a restart.
    fn next(&self, check: &str, started: &str) -> Option<Value> {
        let due = !self.text.is_empty() || self.now != self.told || self.over;
        if self.refused || self.done || !due {
            return None;
        }
        let end = floor(&self.text, PIECE);
        let whole = end == self.text.len();
        let piece = &self.text[..end];
        let ended = self.over && whole;
        let mut body = json!({ "check": check, "startedAt": started, "offset": self.at, "text": piece, "ended": ended });
        if whole {
            body["steps"] = serde_json::to_value(&self.now.steps).unwrap_or_default();
            body["going"] = json!(self.now.going);
        }
        Some(body)
    }

    /// That `body`, from [`Outbox::next`], was `taken` in or not.
    fn answered(&mut self, body: &Value, taken: bool) {
        let sent = body["text"].as_str().map_or(0, str::len) * usize::from(taken);
        self.text.drain(..sent);
        self.at += sent as u64;
        if taken && body.get("steps").is_some() {
            self.told = self.now.clone();
        }
        self.refused = !taken;
        self.done = body["ended"] == true || (!taken && self.over);
    }
}

/// Sends what a run writes to `log`, its check `uri`'s, which started at
/// `started`, to `appview` as it writes it, with how its steps go, until
/// `ended` says it did: then the rest, and that it ended.
pub async fn send(
    appview: Arc<Appview>,
    log: PathBuf,
    uri: String,
    started: String,
    secrets: Vec<String>,
    mut progress: watch::Receiver<Progress>,
    mut ended: watch::Receiver<bool>,
) {
    let mut outbox = Outbox::default();
    let mut every = UNWATCHED;
    loop {
        // Its sender gone, as when the run failed to start, it ended too.
        let over = *ended.borrow_and_update() || ended.has_changed().is_err();
        let more = read_from(&log, outbox.tail.read).unwrap_or_default();
        let now = progress.borrow_and_update().clone();
        outbox.round(&more, &secrets, now, over);
        while let Some(body) = outbox.next(&uri, &started) {
            let answer = appview.put_output(&body).await;
            every = match answer.as_ref().is_ok_and(|watched| *watched) {
                true => WATCHED,
                false => UNWATCHED,
            };
            outbox.answered(&body, answer.is_ok());
        }
        if outbox.done {
            return;
        }
        tokio::select! {
            () = tokio::time::sleep(every) => {}
            _ = ended.changed() => {}
            _ = progress.changed() => {}
        }
    }
}

/// The nearest end of a character in `text` at `at` or before it.
fn floor(text: &str, at: usize) -> usize {
    let mut at = at.min(text.len());
    while !text.is_char_boundary(at) {
        at -= 1;
    }
    at
}

/// The nearest end of a character in `text` at `at` or after it.
fn ceil(text: &str, at: usize) -> usize {
    let mut at = at.min(text.len());
    while !text.is_char_boundary(at) {
        at += 1;
    }
    at
}

/// What the file `path` holds past `from`, [`READ_MAX`] at most.
fn read_from(path: &PathBuf, from: u64) -> Result<Vec<u8>> {
    let mut file = std::fs::File::open(path).context("the run's log")?;
    file.seek(SeekFrom::Start(from))?;
    let mut more = vec![];
    file.take(READ_MAX).read_to_end(&mut more)?;
    Ok(more)
}

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

    #[test]
    fn whole_lines_go_masked() {
        let secrets = vec!["s3cret".to_string()];
        let mut tail = Tail::default();
        assert_eq!(tail.take(b"one\ntw", &secrets, false), "one\n");
        assert_eq!(tail.take(b"o s3c", &secrets, false), "");
        // A secret split as it was written goes masked once its line ends.
        assert_eq!(tail.take(b"ret\rthree", &secrets, false), "two ***\r");
        assert_eq!(tail.take(b"", &secrets, true), "three");
        assert_eq!(tail.read, 20);
        // A line too long to hold is let go.
        let mut tail = Tail::default();
        let long = vec![b'x'; LINE_MAX + 1];
        assert_eq!(tail.take(&long, &secrets, false), "");
        assert_eq!(tail.take(b"next\n", &secrets, false), "next\n");
    }

    #[test]
    fn text_is_cut_at_whole_characters() {
        assert_eq!(floor("abécd", 3), 2);
        assert_eq!(floor("ab", 3), 2);
        assert_eq!(ceil("abécd", 3), 4);
    }

    #[test]
    fn what_is_not_taken_goes_again() {
        let step = |name: &str| Step {
            name: name.into(),
            status: "success".into(),
            exit_code: Some(0),
            duration_ms: 1,
        };
        let going = |name: &str, steps: Vec<Step>| Progress {
            steps,
            going: Some(name.into()),
        };
        let mut outbox = Outbox::default();
        // Nothing new is nothing to send.
        outbox.round(b"", &[], Progress::default(), false);
        assert_eq!(outbox.next("c", "t"), None);
        // A step going is sent, with what it printed; refused, it waits.
        outbox.round(b"==> a\nhi\n", &[], going("a", vec![]), false);
        let body = outbox.next("c", "t").unwrap();
        assert_eq!(body["text"], "==> a\nhi\n");
        assert_eq!(body["going"], "a");
        outbox.answered(&body, false);
        assert_eq!(outbox.next("c", "t"), None);
        assert!(!outbox.done);
        // Next round it goes again, from where it was, with what came since.
        outbox.round(b"more\n", &[], going("b", vec![step("a")]), false);
        let body = outbox.next("c", "t").unwrap();
        assert_eq!(
            (&body["offset"], &body["text"]),
            (&json!(0), &json!("==> a\nhi\nmore\n"))
        );
        assert_eq!(body["steps"][0]["name"], "a");
        outbox.answered(&body, true);
        assert_eq!(outbox.next("c", "t"), None);
        // More than a piece goes in pieces, how the run stands on the last.
        let long = "x".repeat(PIECE) + "\n";
        outbox.round(long.as_bytes(), &[], going("b", vec![step("a")]), false);
        let first = outbox.next("c", "t").unwrap();
        assert_eq!((&first["offset"], first.get("steps")), (&json!(14), None));
        outbox.answered(&first, true);
        let last = outbox.next("c", "t").unwrap();
        assert_eq!(last["text"], "\n");
        outbox.answered(&last, true);
        // Too much held is let go from its start, whole characters.
        let lines = format!("aé\n{}\n\n", "b\n".repeat(UNSENT_MAX / 2 - 2));
        assert_eq!(lines.len(), UNSENT_MAX + 2);
        outbox.round(lines.as_bytes(), &[], Progress::default(), false);
        assert!(outbox.text.starts_with("\nb\n"), "{:?}", &outbox.text[..4]);
        let ahead = (lines.len() - outbox.text.len()) as u64;
        assert_eq!(outbox.at, 14 + PIECE as u64 + 1 + ahead);
        outbox.text.clear();
        // Over, the last send says so; refused then, it is let go.
        outbox.round(b"", &[], Progress::default(), true);
        let body = outbox.next("c", "t").unwrap();
        assert_eq!(body["ended"], true);
        outbox.answered(&body, false);
        assert!(outbox.done);
        let mut taken = Outbox::default();
        taken.round(b"", &[], Progress::default(), true);
        let body = taken.next("c", "t").unwrap();
        taken.answered(&body, true);
        assert!(taken.done);
    }
}