//! Accepting an artifact Sando did not build.
//!
//! Design: wiki [[sando-bento-boundary]], [[release-artifact-identity]].
//!
//! Everything downstream of `build.rs` assumes Sando compiled the thing it is
//! about to ship. That assumption is what the boundary removes: Bento builds,
//! Sando decides whether the result advances. Deciding requires being handed
//! bytes, and being handed bytes requires a reason to believe they are the ones
//! the evidence is about.
//!
//! That belief is not taken on trust here. An accepted artifact has been checked
//! three ways, each catching something the others cannot:
//!
//! 1. The record is internally consistent — its claimed digest is the digest of
//!    the manifest it carries. Catches a hand-edited or truncated document.
//! 2. The bytes on disk hash to that manifest, file for file. Catches a bundle
//!    that drifted in transit, or one swapped for another, and names the file.
//! 3. The digest of what arrived equals the digest the record claims. Implied by
//!    (2), and checked anyway because it is the value everything downstream
//!    keys on, and a mismatch here means the two computations disagree.
//!
//! Intake is about identity, not policy. A record whose builder gates failed is
//! accepted and says so: whether that artifact may advance is the gating layer's
//! call, and it cannot make that call about an artifact it was never handed.

use crate::bundle;
use ops_artifact::ArtifactRecord;
use std::path::{Path, PathBuf};

/// An artifact that arrived with credible paperwork, published into the local
/// release root.
#[derive(Debug)]
pub struct AcceptedArtifact {
    /// The builder's document, verified against the bytes.
    pub record: ArtifactRecord,
    /// The content-addressed directory the bundle now lives at.
    pub released: PathBuf,
}

impl AcceptedArtifact {
    /// True iff every gate the builder ran passed. Not a decision to advance:
    /// Sando's own gates have not run yet, and they are the ones about this
    /// artifact in an environment.
    pub fn builder_gates_passed(&self) -> bool {
        self.record.all_gates_passed()
    }
}

/// Why an artifact was turned away.
///
/// Typed rather than a string because the operator response differs per case
/// and only one of these is a transfer problem. A record that does not parse is
/// a producer bug; a manifest mismatch is corruption or substitution; a failure
/// to publish is a local disk problem on this machine.
#[derive(Debug)]
pub enum IntakeError {
    /// The record did not parse, or failed its own validation.
    BadRecord(ops_artifact::ContractError),
    /// The bundle could not be hashed (unreadable, or not a directory).
    Unreadable(String),
    /// The bytes are not the ones the record describes. `first_difference` names
    /// a file the two disagree about, when one can be identified.
    ManifestMismatch {
        first_difference: Option<String>,
        claimed_files: usize,
        found_files: usize,
    },
    /// The recomputed digest is not the one claimed.
    DigestMismatch { claimed: String, computed: String },
    /// Publishing the verified bundle into the release root failed.
    Publish(String),
}

impl std::fmt::Display for IntakeError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            IntakeError::BadRecord(e) => write!(f, "artifact record is not usable: {e}"),
            IntakeError::Unreadable(d) => write!(f, "could not read the incoming bundle: {d}"),
            IntakeError::ManifestMismatch {
                first_difference: Some(p),
                ..
            } => write!(
                f,
                "the bundle is not what its record describes; `{p}` differs. \
                 Refusing an artifact whose evidence is about other bytes"
            ),
            IntakeError::ManifestMismatch {
                first_difference: None,
                claimed_files,
                found_files,
            } => write!(
                f,
                "the bundle is not what its record describes: the record lists {claimed_files} \
                 file(s), the bundle holds {found_files}"
            ),
            IntakeError::DigestMismatch { claimed, computed } => write!(
                f,
                "bundle digest {computed} does not match the claimed {claimed}"
            ),
            IntakeError::Publish(d) => write!(f, "publishing the accepted bundle failed: {d}"),
        }
    }
}

impl std::error::Error for IntakeError {}

/// Verify an incoming bundle against its record and publish it
/// content-addressed into `release_root`.
///
/// `staged` must already be a directory under `release_root/staging/`, because
/// publishing is an atomic same-filesystem rename and a cross-device staging dir
/// would silently become a copy. The caller owns getting the bytes there; this
/// owns deciding whether they may stay.
///
/// `pinned` is passed straight through to publishing, which gc's the store: it
/// names the release dirs the deployed state still points at
/// ([`crate::retention::pinned_dirs`]). Passing an empty set is what a caller
/// with no deployed state to protect does, not a shortcut.
pub async fn accept(
    release_root: &Path,
    staged: &Path,
    record_json: &str,
    pinned: &crate::retention::PinnedReleases,
) -> Result<AcceptedArtifact, IntakeError> {
    let record = ArtifactRecord::parse(record_json).map_err(IntakeError::BadRecord)?;

    // Hash what actually arrived. Sando's own bundle walker, deliberately: the
    // artifact will later be verified on the node by the same code, so intake
    // proving the bundle with a different implementation than the deploy uses
    // would leave a gap exactly where the two disagree.
    let computed = bundle::digest_dir(staged)
        .await
        .map_err(|e| IntakeError::Unreadable(format!("{e:#}")))?;

    let claimed_manifest = record.manifest.to_text();
    if computed.manifest != claimed_manifest {
        return Err(manifest_mismatch(&claimed_manifest, &computed.manifest));
    }
    if computed.full != record.digest.as_str() {
        return Err(IntakeError::DigestMismatch {
            claimed: record.digest.to_string(),
            computed: computed.full,
        });
    }

    // Write the MANIFEST into the bundle so the node can verify per file after
    // the transfer. Excluded from its own hash, so this does not change what was
    // just proved.
    tokio::fs::write(
        staged.join(bundle::MANIFEST_NAME),
        computed.manifest.as_bytes(),
    )
    .await
    .map_err(|e| IntakeError::Publish(format!("writing MANIFEST: {e}")))?;

    let released =
        crate::deploy::finalize_local_release(release_root, staged, computed.short(), pinned)
            .await
            .map_err(|e| IntakeError::Publish(format!("{e:#}")))?;

    tracing::info!(
        app = %record.provenance.app,
        version = %record.provenance.version,
        target = %record.provenance.target,
        producer = %record.producer,
        digest = %record.digest.short(),
        "accepted an artifact built elsewhere"
    );
    Ok(AcceptedArtifact { record, released })
}

/// Name a file the two manifests disagree about, for the error.
///
/// A digest mismatch alone says only "not the same"; the useful half is which
/// file drifted, which is the whole reason the manifest is per-file rather than
/// one hash over a tarball.
fn manifest_mismatch(claimed: &str, computed: &str) -> IntakeError {
    let paths = |text: &str| -> Vec<(String, String)> {
        text.lines()
            .filter_map(|l| l.split_once("  "))
            .map(|(sha, path)| (path.to_string(), sha.to_string()))
            .collect()
    };
    let (a, b) = (paths(claimed), paths(computed));
    let first_difference = a
        .iter()
        .find(|(path, sha)| !b.iter().any(|(p, s)| p == path && s == sha))
        .or_else(|| b.iter().find(|(path, _)| !a.iter().any(|(p, _)| p == path)))
        .map(|(path, _)| path.clone());
    IntakeError::ManifestMismatch {
        first_difference,
        claimed_files: a.len(),
        found_files: b.len(),
    }
}

#[cfg(test)]
mod tests {
    /// No deployed state in a unit test, so nothing is pinned and gc is free to
    /// apply the count alone. Retention is exercised in `deploy`'s own tests.
    fn no_pins() -> crate::retention::PinnedReleases {
        crate::retention::PinnedReleases::none()
    }

    use super::*;
    use chrono::{DateTime, Utc};
    use ops_artifact::{GateRecord, Manifest, Provenance, Scope, Verdict};

    fn at() -> DateTime<Utc> {
        DateTime::<Utc>::from_timestamp(1_754_000_000, 0).unwrap()
    }

    async fn write(root: &Path, rel: &str, bytes: &[u8]) {
        let p = root.join(rel);
        tokio::fs::create_dir_all(p.parent().unwrap())
            .await
            .unwrap();
        tokio::fs::write(&p, bytes).await.unwrap();
    }

    /// A staging dir under `release_root`, holding a small bundle.
    async fn staged_bundle(release_root: &Path) -> PathBuf {
        let staged = release_root.join("staging").join("1");
        write(&staged, "pom", b"binary bytes").await;
        write(&staged, "static/app.css", b"body{}").await;
        staged
    }

    fn provenance() -> Provenance {
        Provenance {
            app: "pom".into(),
            version: "0.4.1".into(),
            tag: "pom-v0.4.1".into(),
            git_sha: "a".repeat(40),
            target: "linux/aarch64".into(),
            build_host: "astra".into(),
            toolchain: "rustc 1.97.0".into(),
            built_at: at(),
        }
    }

    /// The record Bento would have written for `staged`.
    async fn record_for(staged: &Path, gates: Vec<GateRecord>) -> String {
        let computed = bundle::digest_dir(staged).await.unwrap();
        let manifest = Manifest::parse(&computed.manifest).unwrap();
        ArtifactRecord::new("bento", manifest, provenance(), gates)
            .unwrap()
            .to_json()
    }

    fn passing_gate() -> GateRecord {
        GateRecord::new(
            "prebuild",
            Scope::Artifact,
            Verdict::Passed,
            "prebuild passed in 90s",
            at(),
        )
    }

    /// The load-bearing agreement: Sando's bundle walker and the contract's
    /// manifest must produce the same text, byte for byte. They are separate
    /// implementations on either side of the boundary, and a divergence would
    /// make every honest artifact look corrupt.
    #[tokio::test]
    async fn sandos_walker_and_the_contract_manifest_agree_exactly() {
        let dir = tempfile::tempdir().unwrap();
        let staged = staged_bundle(dir.path()).await;
        let computed = bundle::digest_dir(&staged).await.unwrap();
        let parsed = Manifest::parse(&computed.manifest).unwrap();
        assert_eq!(parsed.to_text(), computed.manifest);
        assert_eq!(parsed.digest().as_str(), computed.full);
    }

    #[tokio::test]
    async fn a_matching_bundle_is_accepted_and_published_content_addressed() {
        let dir = tempfile::tempdir().unwrap();
        let staged = staged_bundle(dir.path()).await;
        let json = record_for(&staged, vec![passing_gate()]).await;

        let accepted = accept(dir.path(), &staged, &json, &no_pins())
            .await
            .unwrap();
        assert!(accepted.builder_gates_passed());
        assert!(
            accepted.released.ends_with(accepted.record.digest.short()),
            "published at {}, expected a dir named for the digest",
            accepted.released.display()
        );
        // The MANIFEST rides along for node-side verification.
        assert!(accepted.released.join(bundle::MANIFEST_NAME).exists());
        // Staging is gone: the bundle was renamed, not copied.
        assert!(!staged.exists());
    }

    #[tokio::test]
    async fn a_bundle_that_drifted_is_refused_and_names_the_file() {
        // The failure this whole boundary exists to make impossible: evidence
        // that vouches for one set of bytes arriving with another.
        let dir = tempfile::tempdir().unwrap();
        let staged = staged_bundle(dir.path()).await;
        let json = record_for(&staged, vec![passing_gate()]).await;
        write(&staged, "static/app.css", b"tampered").await;

        let err = accept(dir.path(), &staged, &json, &no_pins())
            .await
            .unwrap_err();
        match err {
            IntakeError::ManifestMismatch {
                first_difference: Some(p),
                ..
            } => assert_eq!(p, "static/app.css"),
            other => panic!("expected a manifest mismatch naming the file, got {other}"),
        }
        // Nothing was published.
        assert!(!dir.path().join("releases").exists());
    }

    #[tokio::test]
    async fn an_extra_file_in_the_bundle_is_refused() {
        // Same bytes for everything the record lists, plus something it does not.
        // The digest would differ, but the useful error names the intruder.
        let dir = tempfile::tempdir().unwrap();
        let staged = staged_bundle(dir.path()).await;
        let json = record_for(&staged, vec![passing_gate()]).await;
        write(&staged, "companions/unexpected", b"who put this here").await;

        let err = accept(dir.path(), &staged, &json, &no_pins())
            .await
            .unwrap_err();
        match err {
            IntakeError::ManifestMismatch {
                first_difference: Some(p),
                ..
            } => assert_eq!(p, "companions/unexpected"),
            other => panic!("expected the extra file to be named, got {other}"),
        }
    }

    #[tokio::test]
    async fn a_record_that_does_not_parse_is_refused_before_anything_is_hashed() {
        let dir = tempfile::tempdir().unwrap();
        let staged = staged_bundle(dir.path()).await;
        let err = accept(dir.path(), &staged, "{\"not\": \"a record\"}", &no_pins())
            .await
            .unwrap_err();
        assert!(matches!(err, IntakeError::BadRecord(_)), "{err}");
        assert!(staged.exists(), "a refused intake leaves the bytes alone");
    }

    #[tokio::test]
    async fn a_record_whose_digest_was_edited_is_refused() {
        // Editing the digest to match tampered bytes does not help: the record
        // validates its own digest against its own manifest first.
        let dir = tempfile::tempdir().unwrap();
        let staged = staged_bundle(dir.path()).await;
        let json = record_for(&staged, vec![passing_gate()]).await;
        let forged = json.replace(&record_digest(&json), &"f".repeat(64));
        let err = accept(dir.path(), &staged, &forged, &no_pins())
            .await
            .unwrap_err();
        assert!(matches!(err, IntakeError::BadRecord(_)), "{err}");
    }

    #[tokio::test]
    async fn a_failed_builder_gate_is_accepted_and_reported_not_hidden() {
        // Intake decides identity, not whether the thing may ship. Refusing here
        // would leave the evidence of a failed build nowhere Sando can see it.
        let dir = tempfile::tempdir().unwrap();
        let staged = staged_bundle(dir.path()).await;
        let failed = GateRecord::new(
            "prebuild",
            Scope::Artifact,
            Verdict::Failed,
            "12 test(s) failed",
            at(),
        );
        let json = record_for(&staged, vec![failed]).await;
        let accepted = accept(dir.path(), &staged, &json, &no_pins())
            .await
            .unwrap();
        assert!(!accepted.builder_gates_passed());
    }

    fn record_digest(json: &str) -> String {
        let v: serde_json::Value = serde_json::from_str(json).unwrap();
        v["digest"].as_str().unwrap().to_string()
    }
}