//! Handing a finished artifact to Sando.
//!
//! Design + rationale: maintainer wiki.
//! <!-- wiki: sando-bento-boundary -->
//!
//! The boundary is "Bento builds and packages, Sando decides whether a thing
//! advances a stage". Sando has an intake that proves an incoming bundle against
//! its record three ways and publishes it content-addressed — but `POST /intake`
//! takes a bundle already sitting under Sando's staging directory, and until now
//! nothing put one there. This is the half that moves the bytes.
//!
//! Two motions, in this order, per target:
//!
//! 1. rsync the target's collect directory into
//!    `<staging_root>/<app>-<version>-<target>/` on the Sando host.
//! 2. POST that path plus the artifact record to sandod, which verifies and
//!    publishes it, or refuses it and names the file that drifted.
//!
//! **The record does not travel with the bytes.** It names the digest of the
//! bundle, and the digest covers every file in the bundle, so a record copied in
//! among the artifacts would change the digest it names — Sando would recompute,
//! disagree, and refuse. It goes in the request body instead. That is the same
//! "evidence rides alongside, never inside" rule the contract crate is built on,
//! showing up here as an rsync exclude.
//!
//! **The staging directory is mirrored, not merged.** A retry after a partial
//! transfer would otherwise leave a file from the earlier attempt behind, and an
//! extra file is a manifest mismatch: the honest bundle would be refused for
//! carrying a leftover. `--delete` makes the directory hold exactly this attempt.
//!
//! Failure fails the target run. A build whose artifact never reached the deploy
//! controller has not finished the job the handoff exists to do, and the same
//! argument the archive deposit makes applies with more force: a silently
//! skipped handoff means the release nobody watched is the one Sando never heard
//! about.

use crate::config::{Config, Handoff};
use crate::domain::{AppId, Target, Version};
use ops_exec::{CapabilitySet, Executor, LocalExec, SshExec, SyncOpts};
use std::path::{Path, PathBuf};
use std::sync::Arc;

/// Where one target's bundle is staged on the Sando host.
///
/// Flat and fully qualified rather than nested `<app>/<version>/<target>/`,
/// because Sando's intake resolves this path against its own staging root and
/// publishes by renaming the directory out of it. One component means one
/// rename, and the name still says which build it is when an operator finds it
/// left behind after a refused intake.
pub fn staged_dir(staging_root: &Path, app: &AppId, version: &Version, target: Target) -> PathBuf {
    staging_root.join(format!(
        "{}-{}-{}",
        app.as_str(),
        version,
        crate::archive::target_slug(target)
    ))
}

/// The transport that writes into Sando's staging directory.
///
/// Granted nothing, like the archive's: it only ever calls `push_dir`, so no
/// capability in `ops_exec` is exercisable through this handle and the Sando
/// host cannot become a build host by holding one. It is not in the topology
/// either, so no recipe can name it.
fn transport(handoff: &Handoff) -> Arc<dyn Executor> {
    const NONE: [&str; 0] = [];
    let caps = CapabilitySet::from_tokens(NONE, NONE);
    if handoff.host == "local" || handoff.host.is_empty() {
        return Arc::new(LocalExec::new(caps));
    }
    Arc::new(SshExec::new(handoff.host.clone(), caps))
}

/// `POST` target for this handoff: the unprefixed mount is Sando's default
/// product, and a named one is nested under `/apps/<id>`.
fn intake_url(handoff: &Handoff) -> String {
    let base = handoff.url.trim_end_matches('/');
    match &handoff.sando_app {
        Some(id) => format!("{base}/apps/{id}/intake"),
        None => format!("{base}/intake"),
    }
}

/// Read the bearer token for this handoff out of the private layer.
///
/// Resolved under `secrets_root` the same way the `secret()` host function
/// resolves a recipe's credentials, so there is one place secrets live rather
/// than two. Never logged, and the trailing newline a file written by `echo`
/// carries is trimmed — it would otherwise become part of the header value.
async fn token(cfg: &Config, handoff: &Handoff) -> anyhow::Result<Option<String>> {
    let Some(rel) = &handoff.token_file else {
        return Ok(None);
    };
    let path = cfg.secrets_root.join(rel);
    let raw = tokio::fs::read_to_string(&path)
        .await
        .map_err(|e| anyhow::anyhow!("reading the sando token at {}: {e}", path.display()))?;
    let trimmed = raw.trim().to_string();
    anyhow::ensure!(
        !trimmed.is_empty(),
        "the sando token at {} is empty",
        path.display()
    );
    Ok(Some(trimmed))
}

/// Send one target's finished bundle to Sando and ask it to take the artifact
/// in. A no-op for an app with no handoff configured.
///
/// `local_dir` is the target's collect directory and `record_path` the artifact
/// record beside it — written by `artifact_record::emit` after the recipe, which
/// is why this runs there and not inside `collect`.
pub async fn send(
    cfg: &Config,
    local_dir: &Path,
    record_path: &Path,
    app: &AppId,
    version: &Version,
    target: Target,
) -> anyhow::Result<()> {
    let Some(handoff) = cfg.handoff.get(app.as_str()) else {
        return Ok(());
    };

    // Read the record first. It is what makes the transfer worth doing, and a
    // missing or unreadable one means shipping bytes Sando could only refuse.
    let record = tokio::fs::read_to_string(record_path).await.map_err(|e| {
        anyhow::anyhow!(
            "reading the artifact record at {}: {e}",
            record_path.display()
        )
    })?;

    let dest = staged_dir(&handoff.staging_root, app, version, target);
    transport(handoff)
        .push_dir(
            local_dir,
            &dest,
            &SyncOpts {
                // Exactly this attempt's bytes: see the module header.
                delete: true,
                exclude: vec![crate::artifact_record::RECORD_FILE.to_string()],
                ..SyncOpts::archive_deposit()
            },
        )
        .await
        .map_err(|e| {
            anyhow::anyhow!(
                "staging {} at {}:{}: {e}",
                local_dir.display(),
                handoff.host,
                dest.display()
            )
        })?;

    let url = intake_url(handoff);
    let mut req = crate::tls::builder()
        .build()?
        .post(&url)
        .json(&serde_json::json!({
            "staged": dest.to_string_lossy(),
            "record": record,
        }));
    if let Some(t) = token(cfg, handoff).await? {
        req = req.bearer_auth(t);
    }
    let resp = req
        .send()
        .await
        .map_err(|e| anyhow::anyhow!("posting the intake to {url}: {e}"))?;

    let status = resp.status();
    let body = resp.text().await.unwrap_or_default();
    anyhow::ensure!(
        status.is_success(),
        // Sando's refusals are the useful half — it names the file that drifted
        // — so the body is carried into the error rather than reduced to a code.
        "sando refused the intake at {url}: {status} {body}"
    );

    tracing::info!(
        %app, %target, %version, host = %handoff.host, dest = %dest.display(),
        "handed the artifact to sando"
    );
    Ok(())
}

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

    fn app() -> AppId {
        AppId::new("pom")
    }
    fn version() -> Version {
        "0.3.1".parse().unwrap()
    }
    fn handoff(root: &Path) -> Handoff {
        Handoff {
            host: "local".into(),
            staging_root: root.join("staging"),
            url: "http://127.0.0.1:1".into(),
            sando_app: None,
            token_file: None,
        }
    }

    /// One component, carrying every dimension that distinguishes a build. Two
    /// architectures of one version are two bundles, and staging them at one
    /// path would have the second overwrite the first.
    #[test]
    fn a_staged_dir_names_app_version_and_target_in_one_component() {
        let dir = staged_dir(
            Path::new("/srv/sando/staging"),
            &app(),
            &version(),
            "linux/aarch64".parse().unwrap(),
        );
        assert_eq!(dir, Path::new("/srv/sando/staging/pom-0.3.1-linux-aarch64"));
        assert_eq!(dir.parent(), Some(Path::new("/srv/sando/staging")));

        let other = staged_dir(
            Path::new("/srv/sando/staging"),
            &app(),
            &version(),
            "linux/x86_64".parse().unwrap(),
        );
        assert_ne!(dir, other);
    }

    /// The default product is the unprefixed mount, because that is what Sando's
    /// router does with a product it was not asked to nest.
    #[test]
    fn the_intake_url_nests_only_a_named_product() {
        let mut h = handoff(Path::new("/tmp"));
        h.url = "http://100.103.89.95:7766".into();
        assert_eq!(intake_url(&h), "http://100.103.89.95:7766/intake");

        h.sando_app = Some("pom".into());
        assert_eq!(intake_url(&h), "http://100.103.89.95:7766/apps/pom/intake");

        // A trailing slash is an operator's habit, not a second path segment.
        h.url = "http://100.103.89.95:7766/".into();
        assert_eq!(intake_url(&h), "http://100.103.89.95:7766/apps/pom/intake");
    }

    /// An app nobody configured a handoff for is untouched — the ordinary case
    /// for every app Bento ships that Sando does not deploy.
    #[tokio::test]
    async fn an_app_with_no_handoff_sends_nothing() {
        let tmp = tempfile::tempdir().unwrap();
        let cfg = Config::for_tests(tmp.path());
        assert!(cfg.handoff.is_empty());
        send(
            &cfg,
            tmp.path(),
            &tmp.path().join("record.json"),
            &app(),
            &version(),
            "linux/x86_64".parse().unwrap(),
        )
        .await
        .expect("a no-op cannot fail");
    }

    /// The staging transfer, up to the POST (which fails against a dead port).
    /// What it proves is the shape of what lands: the artifacts arrive, and the
    /// record does NOT — a record inside the bundle would change the digest it
    /// names and Sando would refuse the honest bytes.
    #[tokio::test]
    async fn the_bundle_is_staged_without_its_record() {
        let tmp = tempfile::tempdir().unwrap();
        let mut cfg = Config::for_tests(tmp.path());
        cfg.handoff.insert("pom".into(), handoff(tmp.path()));

        let collected = tmp.path().join("collected");
        std::fs::create_dir_all(&collected).unwrap();
        std::fs::write(collected.join("pom"), b"binary").unwrap();
        let record_path = collected.join(crate::artifact_record::RECORD_FILE);
        std::fs::write(&record_path, b"{\"producer\":\"bento\"}").unwrap();

        let target: Target = "linux/x86_64".parse().unwrap();
        // The POST cannot succeed here; the transfer before it still ran.
        let _ = send(&cfg, &collected, &record_path, &app(), &version(), target).await;

        let staged = staged_dir(&tmp.path().join("staging"), &app(), &version(), target);
        assert_eq!(std::fs::read(staged.join("pom")).unwrap(), b"binary");
        assert!(
            !staged.join(crate::artifact_record::RECORD_FILE).exists(),
            "the record must not travel inside the bundle it describes"
        );
    }

    /// A leftover from an earlier attempt is pruned. Without `--delete` it would
    /// survive as an extra file, and an extra file is a manifest mismatch — the
    /// retry of an honest build would be refused for carrying it.
    #[tokio::test]
    async fn a_retry_leaves_nothing_from_the_previous_attempt() {
        let tmp = tempfile::tempdir().unwrap();
        let mut cfg = Config::for_tests(tmp.path());
        cfg.handoff.insert("pom".into(), handoff(tmp.path()));

        let collected = tmp.path().join("collected");
        std::fs::create_dir_all(&collected).unwrap();
        std::fs::write(collected.join("pom"), b"binary").unwrap();
        let record_path = collected.join(crate::artifact_record::RECORD_FILE);
        std::fs::write(&record_path, b"{}").unwrap();

        let target: Target = "linux/x86_64".parse().unwrap();
        let staged = staged_dir(&tmp.path().join("staging"), &app(), &version(), target);
        std::fs::create_dir_all(&staged).unwrap();
        std::fs::write(staged.join("stale-from-last-time"), b"junk").unwrap();

        let _ = send(&cfg, &collected, &record_path, &app(), &version(), target).await;

        assert!(staged.join("pom").exists());
        assert!(
            !staged.join("stale-from-last-time").exists(),
            "the staging dir must hold exactly this attempt"
        );
    }

    /// A missing record is refused before any bytes move. Sando could only
    /// refuse the bundle anyway, and staging first would leave a directory
    /// behind for an intake that was never going to be requested.
    #[tokio::test]
    async fn a_missing_record_fails_before_anything_is_staged() {
        let tmp = tempfile::tempdir().unwrap();
        let mut cfg = Config::for_tests(tmp.path());
        cfg.handoff.insert("pom".into(), handoff(tmp.path()));

        let collected = tmp.path().join("collected");
        std::fs::create_dir_all(&collected).unwrap();
        std::fs::write(collected.join("pom"), b"binary").unwrap();

        let target: Target = "linux/x86_64".parse().unwrap();
        let err = send(
            &cfg,
            &collected,
            &collected.join("record.json"),
            &app(),
            &version(),
            target,
        )
        .await
        .expect_err("no record, no handoff");
        assert!(format!("{err:#}").contains("artifact record"), "{err:#}");

        let staged = staged_dir(&tmp.path().join("staging"), &app(), &version(), target);
        assert!(!staged.exists(), "nothing should have been staged");
    }

    /// An empty token file is a bootstrap that half-ran. Sending the header
    /// anyway would fail at sandod as a plain 401, which reads as a wrong token
    /// rather than a missing one.
    #[tokio::test]
    async fn an_empty_token_file_is_an_error_not_an_empty_header() {
        let tmp = tempfile::tempdir().unwrap();
        let cfg = Config::for_tests(tmp.path());
        std::fs::create_dir_all(&cfg.secrets_root).unwrap();
        std::fs::write(cfg.secrets_root.join("sando-token"), "\n").unwrap();

        let mut h = handoff(tmp.path());
        h.token_file = Some(PathBuf::from("sando-token"));
        let err = token(&cfg, &h).await.expect_err("empty is not a token");
        assert!(format!("{err:#}").contains("empty"), "{err:#}");
    }

    /// The whole motion against a sandod-shaped listener: the bytes are staged,
    /// and the request names the path they were staged at and carries the record
    /// as its own field. Those two together are the contract — Sando resolves
    /// `staged` on its own disk and proves it against `record`, so a handoff that
    /// staged one path and reported another would be refused for corruption when
    /// nothing had corrupted.
    #[tokio::test]
    async fn the_post_names_the_path_the_bytes_were_staged_at() {
        use axum::{Json, Router, routing::post};
        use std::sync::{Arc, Mutex};

        let seen: Arc<Mutex<Option<serde_json::Value>>> = Arc::new(Mutex::new(None));
        let auth: Arc<Mutex<Option<String>>> = Arc::new(Mutex::new(None));
        let (s, a) = (seen.clone(), auth.clone());
        let app_router = Router::new().route(
            "/apps/pom/intake",
            post(
                move |headers: axum::http::HeaderMap, Json(b): Json<serde_json::Value>| {
                    let (s, a) = (s.clone(), a.clone());
                    async move {
                        *s.lock().unwrap() = Some(b);
                        *a.lock().unwrap() = headers
                            .get("authorization")
                            .and_then(|v| v.to_str().ok())
                            .map(str::to_owned);
                        Json(serde_json::json!({ "accepted": true, "run_id": 1 }))
                    }
                },
            ),
        );
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        tokio::spawn(async move { axum::serve(listener, app_router).await.unwrap() });

        let tmp = tempfile::tempdir().unwrap();
        let mut cfg = Config::for_tests(tmp.path());
        std::fs::create_dir_all(&cfg.secrets_root).unwrap();
        std::fs::write(cfg.secrets_root.join("sando-token"), "s3cr3t\n").unwrap();
        let mut h = handoff(tmp.path());
        h.url = format!("http://{addr}");
        h.sando_app = Some("pom".into());
        h.token_file = Some(PathBuf::from("sando-token"));
        cfg.handoff.insert("pom".into(), h);

        let collected = tmp.path().join("collected");
        std::fs::create_dir_all(&collected).unwrap();
        std::fs::write(collected.join("pom"), b"binary").unwrap();
        let record_path = collected.join(crate::artifact_record::RECORD_FILE);
        std::fs::write(&record_path, b"{\"producer\":\"bento\"}").unwrap();

        let target: Target = "linux/x86_64".parse().unwrap();
        send(&cfg, &collected, &record_path, &app(), &version(), target)
            .await
            .expect("the intake was accepted");

        let body = seen.lock().unwrap().clone().expect("sandod was called");
        let staged = staged_dir(&tmp.path().join("staging"), &app(), &version(), target);
        assert_eq!(body["staged"], staged.to_string_lossy().as_ref());
        assert_eq!(body["record"], "{\"producer\":\"bento\"}");
        assert!(
            Path::new(body["staged"].as_str().unwrap())
                .join("pom")
                .exists()
        );
        assert_eq!(auth.lock().unwrap().as_deref(), Some("Bearer s3cr3t"));
    }

    /// A refusal is a failed handoff, and it carries Sando's own words. Its
    /// errors name the file that drifted, which is the whole return on a
    /// per-file manifest; reducing them to a status code would throw that away
    /// at the one moment somebody needs it.
    #[tokio::test]
    async fn a_refusal_fails_and_carries_what_sando_said() {
        use axum::{Router, http::StatusCode, routing::post};

        let app_router = Router::new().route(
            "/intake",
            post(|| async {
                (
                    StatusCode::BAD_REQUEST,
                    "the bundle is not what its record describes; `pom` differs",
                )
            }),
        );
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        tokio::spawn(async move { axum::serve(listener, app_router).await.unwrap() });

        let tmp = tempfile::tempdir().unwrap();
        let mut cfg = Config::for_tests(tmp.path());
        let mut h = handoff(tmp.path());
        h.url = format!("http://{addr}");
        cfg.handoff.insert("pom".into(), h);

        let collected = tmp.path().join("collected");
        std::fs::create_dir_all(&collected).unwrap();
        std::fs::write(collected.join("pom"), b"binary").unwrap();
        let record_path = collected.join(crate::artifact_record::RECORD_FILE);
        std::fs::write(&record_path, b"{}").unwrap();

        let err = send(
            &cfg,
            &collected,
            &record_path,
            &app(),
            &version(),
            "linux/x86_64".parse().unwrap(),
        )
        .await
        .expect_err("a refused intake is a failed handoff");
        let msg = format!("{err:#}");
        assert!(msg.contains("`pom` differs"), "{msg}");
    }

    /// The token is trimmed: a file written with `echo` ends in a newline, and a
    /// newline inside a header value is not a credential problem an operator
    /// would ever guess at.
    #[tokio::test]
    async fn a_token_is_read_trimmed() {
        let tmp = tempfile::tempdir().unwrap();
        let cfg = Config::for_tests(tmp.path());
        std::fs::create_dir_all(&cfg.secrets_root).unwrap();
        std::fs::write(cfg.secrets_root.join("sando-token"), "s3cr3t\n").unwrap();

        let mut h = handoff(tmp.path());
        h.token_file = Some(PathBuf::from("sando-token"));
        assert_eq!(token(&cfg, &h).await.unwrap().as_deref(), Some("s3cr3t"));
        // No file named means no header, not an empty one.
        h.token_file = None;
        assert_eq!(token(&cfg, &h).await.unwrap(), None);
    }
}