//! `dev.knotless.check` records, in the runner's own account: one per
//! workflow, ref and commit, replaced as each run of it goes. A private
//! repository's are in the runner's repo in its space, as is all that
//! says anything of it.

use anyhow::Result;
use knotless::crypto::PublicKey;
use knotless::env::Ctx;
use knotless::identity::{self, Identity};
use knotless::records::{BlobRef, MinutesRecord, nsid};
use knotless::remote::{self, Remote};
use knotless::sealed;
use knotless::session;
use knotless::space;
use knotless::xrpc::{Auth, Xrpc, XrpcError};
use serde_json::json;
use std::time::Duration;

pub const NSID: &str = nsid::CHECK;

/// At most this much of a run's output is kept: its tail.
pub const LOG_MAX: usize = 1_000_000;

pub use knotless::keys::check_rkey as rkey;
pub use knotless::records::{CheckRecord as Check, CheckStep as Step, status};

/// Values shorter than this are left in a log: masking them would mask
/// ordinary output too.
pub const SECRET_MIN: usize = 4;

/// `log` with each of `secrets`, and each line of one that has several,
/// replaced by `***`, longest first so a secret inside another leaves no
/// part of it showing.
pub fn redact(log: &[u8], secrets: &[String]) -> Vec<u8> {
    let mut masks: Vec<&str> = secrets
        .iter()
        .flat_map(|s| std::iter::once(s.as_str()).chain(s.lines().map(str::trim)))
        .filter(|s| s.len() >= SECRET_MIN)
        .collect();
    masks.sort_by_key(|s| std::cmp::Reverse(s.len()));
    masks.dedup();
    let mut out = log.to_vec();
    for mask in masks {
        out = replace(&out, mask.as_bytes());
    }
    out
}

fn replace(haystack: &[u8], needle: &[u8]) -> Vec<u8> {
    let mut out = Vec::with_capacity(haystack.len());
    let mut rest = haystack;
    while let Some(at) = rest.windows(needle.len()).position(|w| w == needle) {
        out.extend_from_slice(&rest[..at]);
        out.extend_from_slice(b"***");
        rest = &rest[at + needle.len()..];
    }
    out.extend_from_slice(rest);
    out
}

/// The log of a run that takes another's, `earlier`, as its own: what it
/// `said` first, then as much of the end of `earlier` as fits within
/// [`LOG_MAX`] with it, so it stands alone once that run's check is gone.
pub fn copied(said: &str, earlier: &[u8]) -> Vec<u8> {
    let room = LOG_MAX.saturating_sub(said.len());
    let kept = &earlier[earlier.len().saturating_sub(room)..];
    [said.as_bytes(), kept].concat()
}

/// The most checks one write deletes: the reference PDS takes 200
/// writes in one `applyWrites`.
pub const DELETES: usize = 200;

/// The last [`LOG_MAX`] bytes of `log`.
pub fn tail(log: &[u8]) -> &[u8] {
    &log[log.len().saturating_sub(LOG_MAX)..]
}

/// The wait before a write the PDS failed is tried again, doubled each
/// time, while it is at most [`WRITE_RETRY_MAX`]: four tries in all.
const WRITE_RETRY: Duration = Duration::from_millis(500);
const WRITE_RETRY_MAX: Duration = Duration::from_secs(2);

/// The most a write waits, all told, for rate limits that refuse it to
/// let writes in again: a run holds its job meanwhile.
const LIMITED_MAX: Duration = Duration::from_secs(10 * 60);

/// Whether `err` may pass if tried again: the PDS failed or did not
/// answer, rather than refusing what was asked.
fn passing(err: &anyhow::Error) -> bool {
    XrpcError::of(err).is_none_or(|e| e.status >= 500 || e.status == 429)
}

/// How long a write refused by a rate limit, as `err` was, waits for it to
/// let writes in again, at `now` (seconds since the epoch), having waited
/// `limited` so far: `None` if `err` is not one that says when, or that is
/// past [`LIMITED_MAX`].
fn limited_for(err: &anyhow::Error, limited: Duration, now: u64) -> Option<Duration> {
    let reset = XrpcError::of(err).filter(|e| e.status == 429)?.reset?;
    let wait = Duration::from_secs(reset.saturating_sub(now)).max(WRITE_RETRY);
    (limited + wait <= LIMITED_MAX).then_some(wait)
}

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

/// Writes checks as the runner's account.
pub struct Reporter {
    ctx: Ctx,
    pub identity: Identity,
    xrpc: Xrpc,
    /// Where its writes go: its account's PDS, or a test's stand-in.
    writes: Xrpc,
}

impl Reporter {
    /// The account this process is logged in as.
    pub async fn new(ctx: &Ctx) -> Result<Reporter> {
        let me = session::me(ctx).await?;
        let identity = identity::resolve(ctx, &me.did).await?;
        let xrpc = Xrpc::new(ctx.http.clone(), &identity.pds);
        Ok(Reporter {
            writes: xrpc.clone(),
            xrpc,
            ctx: ctx.clone(),
            identity,
        })
    }

    /// Points its writes at `pds` rather than its account's PDS, as a
    /// test's stand-in in front of it does, over `ctx`'s connections: its
    /// reads go as ever.
    pub fn through(mut self, ctx: &Ctx, pds: &str) -> Reporter {
        self.writes = Xrpc::new(ctx.http.clone(), pds);
        self
    }

    /// The URI of this account's check `rkey` of the repository `of`
    /// reads: in full, in its space, for a private one.
    pub fn uri_of(&self, of: &Remote, rkey: &str) -> String {
        let at = knotless::aturi::AtUri::public(&self.identity.did, NSID, rkey);
        at.resolved(of.space_ref().as_ref()).to_string()
    }

    /// Files `check` under `rkey`, of the repository `of` reads.
    pub async fn put(&self, of: &Remote, rkey: &str, check: &Check) -> Result<()> {
        let mut record = serde_json::to_value(check)?;
        record["$type"] = NSID.into();
        let mut body = json!({
            "repo": self.identity.did,
            "collection": NSID,
            "rkey": rkey,
            "record": record,
        });
        let nsid = match &of.space {
            Some(space) => {
                body["space"] = space.uri.clone().into();
                "com.atproto.space.putRecord"
            }
            None => "com.atproto.repo.putRecord",
        };
        self.write(nsid, &body).await
    }

    /// Files `minutes`, of its owner's public repositories, under the
    /// owner's DID.
    pub async fn put_minutes(&self, owner: &str, minutes: &MinutesRecord) -> Result<()> {
        let mut record = serde_json::to_value(minutes)?;
        record["$type"] = nsid::MINUTES.into();
        let body = json!({
            "repo": self.identity.did,
            "collection": nsid::MINUTES,
            "rkey": owner,
            "record": record,
        });
        self.write("com.atproto.repo.putRecord", &body).await
    }

    /// Publishes `key` as the key owners seal its secrets to, unless its
    /// record already holds it.
    pub async fn put_key(&self, key: &PublicKey) -> Result<()> {
        let key = key.to_did_key();
        let params = [
            ("repo", self.identity.did.as_str()),
            ("collection", sealed::KEY_NSID),
            ("rkey", sealed::KEY_RKEY),
        ];
        let held =
            self.xrpc
                .query::<serde_json::Value>("com.atproto.repo.getRecord", &params, None);
        let held = held.await.ok().map(|r| r["value"]["key"].clone());
        if held.is_some_and(|k| k == key.as_str()) {
            return Ok(());
        }
        let body = json!({
            "repo": self.identity.did,
            "collection": sealed::KEY_NSID,
            "rkey": sealed::KEY_RKEY,
            "record": {
                "$type": sealed::KEY_NSID,
                "key": key,
                "createdAt": knotless::records::now(),
            },
        });
        self.write("com.atproto.repo.putRecord", &body).await
    }

    /// Calls the procedure `nsid` with `body`, as [`Reporter::retried`].
    async fn write(&self, nsid: &str, body: &serde_json::Value) -> Result<()> {
        self.retried(|auth| async move {
            let wrote =
                self.writes
                    .procedure::<serde::de::IgnoredAny>(nsid, Some(body), Some(&auth));
            wrote.await.map(drop)
        })
        .await
    }

    /// What `call` returns, as the account, tried again a few times while
    /// the PDS fails it or does not answer in time, as one busy for a
    /// minute does, and once a rate limit refusing it lets writes in again:
    /// a write given up on leaves a check that never ends, or a run that
    /// passed filed as an error.
    async fn retried<T, F: Future<Output = Result<T>>>(
        &self,
        call: impl Fn(Auth) -> F,
    ) -> Result<T> {
        let (mut wait, mut limited) = (WRITE_RETRY, Duration::ZERO);
        loop {
            let auth = session::ensure(&self.ctx, &self.identity).await?.auth();
            let err = match call(auth).await {
                Err(err) => err,
                done => return done,
            };
            let until = match limited_for(&err, limited, epoch_secs()) {
                Some(until) => {
                    limited += until;
                    until
                }
                None if wait <= WRITE_RETRY_MAX && passing(&err) => {
                    wait *= 2;
                    wait / 2
                }
                None => return Err(err),
            };
            crate::telemetry::retried(&err, until);
            tokio::time::sleep(until).await;
        }
    }

    /// The checks keyed under `prefix` that this account filed, by key,
    /// of the repository `of` reads: one that does not read as a check is
    /// left out.
    pub async fn checks(&self, of: &Remote, prefix: &str) -> Result<Vec<(String, Check)>> {
        let mine = self.mine(of).await?;
        let found = mine.scan_all::<serde_json::Value>(NSID, prefix).await?;
        Ok(found
            .into_iter()
            .filter_map(|s| Some((s.rkey, serde_json::from_value::<Check>(s.value).ok()?)))
            .collect())
    }

    /// The checks keyed under `prefix` that this account left queued or
    /// running, of the repository `of` reads.
    pub async fn running(&self, of: &Remote, prefix: &str) -> Result<Vec<(String, Check)>> {
        let all = self.checks(of, prefix).await?;
        let running = all.into_iter().filter(|(_, c)| status::going(&c.status));
        Ok(running.collect())
    }

    /// Deletes the checks this account filed under `rkeys`, of the
    /// repository `of` reads, [`DELETES`] to a write.
    pub async fn delete(&self, of: &Remote, rkeys: &[String]) -> Result<()> {
        let mut mine = self.mine(of).await?;
        mine.xrpc = self.writes.clone();
        for some in rkeys.chunks(DELETES) {
            let writes: Vec<_> = some.iter().map(|r| remote::delete(NSID, r)).collect();
            let (mine, writes) = (&mine, &writes);
            self.retried(|auth| async move { mine.apply(writes.clone(), &auth).await })
                .await?;
        }
        Ok(())
    }

    /// Whether this account filed any check keyed under `prefix`, of the
    /// repository `of` reads.
    pub async fn any(&self, of: &Remote, prefix: &str) -> Result<bool> {
        let mine = self.mine(of).await?;
        let page = mine.scan::<serde_json::Value>(NSID, prefix, None).await?;
        Ok(!page.items.is_empty())
    }

    /// The check this account filed under `rkey`, of the repository `of`
    /// reads, if there is one.
    pub async fn get(&self, of: &Remote, rkey: &str) -> Result<Option<Check>> {
        let mine = self.mine(of).await?;
        let found = mine.get_record::<Check>(NSID, rkey).await?;
        Ok(found.map(|s| s.value))
    }

    /// This account's records about the repository `of` reads: in its
    /// repo in the space, for a private one, read with a space credential
    /// as a member's are, even its own.
    async fn mine(&self, of: &Remote) -> Result<Remote> {
        let space = match &of.space {
            Some(space) => {
                let joined = space::joined(&self.ctx, &space.uri).await?;
                Some(joined.at(&self.identity.did))
            }
            None => None,
        };
        let mine = Remote {
            xrpc: self.xrpc.clone(),
            identity: self.identity.clone(),
            url: of.url.clone(),
            space,
        };
        Ok(mine)
    }

    /// The log `log` of a check this account filed of the repository `of`
    /// reads.
    pub async fn log_of(&self, of: &Remote, log: &BlobRef) -> Result<Vec<u8>> {
        let mine = self.mine(of).await?;
        mine.get_blob(&log.link.cid).await
    }

    pub async fn upload_log(&self, log: &[u8]) -> Result<BlobRef> {
        self.upload(tail(log).to_vec(), knotless::records::LOG_MIME)
            .await
    }

    pub async fn upload(&self, bytes: Vec<u8>, mime: &str) -> Result<BlobRef> {
        let bytes = &bytes;
        self.retried(
            |auth| async move { self.writes.upload_blob(bytes.clone(), mime, &auth).await },
        )
        .await
    }
}

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

    #[test]
    fn copies_keep_what_they_said_first() {
        assert_eq!(
            copied("==> passed\n", b"==> test\nok\n"),
            b"==> passed\n==> test\nok\n"
        );
        let long = vec![b'x'; LOG_MAX];
        let log = copied("==> passed\n", &long);
        assert_eq!(log.len(), LOG_MAX);
        assert!(log.starts_with(b"==> passed\nxx"));
    }

    #[test]
    fn keys_and_tails() {
        let k = rkey(
            "did:plc:a",
            "3l3qo2vuowo2a",
            "refs/heads/main",
            "abc",
            None,
            "ci",
        );
        assert_eq!(k.split('.').count(), 5);
        let pull = rkey(
            "did:plc:a",
            "3l3qo2vuowo2a",
            "at://p",
            "abc",
            Some("def"),
            "ci",
        );
        assert!(pull.ends_with(".abc.def.ci"), "{pull}");
        assert!(k.starts_with(&knotless::keys::check_prefix("did:plc:a", "3l3qo2vuowo2a")));
        assert!(k.ends_with(".ci"));
        assert_ne!(
            k,
            rkey(
                "did:plc:b",
                "3l3qo2vuowo2a",
                "refs/heads/main",
                "abc",
                None,
                "ci"
            )
        );
        assert_eq!(tail(b"abc"), b"abc");
        let secrets = [
            "s3cret".to_string(),
            "abc".to_string(),
            "-----BEGIN\n  keybody1\n-----END".to_string(),
        ];
        let log = b"s3crets3cret abc\n-----BEGIN\n  keybody1\n-----END\nkeybody1 again\n";
        assert_eq!(
            String::from_utf8(redact(log, &secrets)).unwrap(),
            "****** abc\n***\n*** again\n"
        );
        assert_eq!(redact(b"nothing", &[]), b"nothing");
        let big = vec![b'x'; LOG_MAX + 5];
        assert_eq!(tail(&big).len(), LOG_MAX);
    }

    #[test]
    fn what_is_tried_again() {
        let refused = |status| {
            anyhow::Error::from(XrpcError {
                status,
                error: "E".into(),
                message: "m".into(),
                reset: None,
            })
        };
        assert!(passing(&refused(500)));
        assert!(passing(&refused(429)));
        assert!(!passing(&refused(400)));
        assert!(passing(&anyhow::anyhow!("operation timed out")));
    }

    #[test]
    fn rate_limits_are_waited_out() {
        let limited = |status, reset| {
            anyhow::Error::from(XrpcError {
                status,
                error: "RateLimitExceeded".into(),
                message: "m".into(),
                reset,
            })
        };
        let none = Duration::ZERO;
        let wait = |err: &anyhow::Error, so_far| limited_for(err, so_far, 1_000);
        assert_eq!(
            wait(&limited(429, Some(1_030)), none),
            Some(Duration::from_secs(30))
        );
        // A reset already past is tried again soon, not at once.
        assert_eq!(wait(&limited(429, Some(990)), none), Some(WRITE_RETRY));
        // One that says nothing of when, or is not a limit, is backed off.
        assert_eq!(wait(&limited(429, None), none), None);
        assert_eq!(wait(&limited(503, Some(1_030)), none), None);
        assert_eq!(wait(&anyhow::anyhow!("timed out"), none), None);
        // Past what it waits all told, it gives up.
        assert_eq!(wait(&limited(429, Some(1_000 + 3_600)), none), None);
        let most = LIMITED_MAX - Duration::from_secs(10);
        assert_eq!(wait(&limited(429, Some(1_030)), most), None);
        assert!(epoch_secs() > 1_700_000_000);
    }
}