//! Which release directories the deployed state still points at.
//!
//! <!-- wiki: release-artifact-identity -->
//!
//! Release dirs are content-addressed (`releases/<digest16>`) and trimmed after
//! every publish. Trimming used to be a count alone — keep the 5 newest by
//! mtime — which cannot express "this one is still in production". Three
//! rebuilds of a single version fill three of the five slots, and the artifacts
//! a tier is running and would roll back to fall off the end. That happened
//! twice (2026-08-19, 2026-08-25); the second time neither surviving directory
//! held a runnable binary and prod had nothing to roll back to.
//!
//! So the set below is computed first and set aside, and the count applies only
//! to what is left. The count is a floor on how much history to keep, not a
//! ceiling on what may be retained.
//!
//! # What counts as referenced
//!
//! Everything promote and rollback resolve through, because a dir that is
//! unreachable to them is exactly the one whose absence is discovered by an
//! rsync failing mid-promote:
//!
//! 1. `tier_state.current_build_id` / `previous_build_id` -> `build_runs.staged_path`.
//!    The identity path (migration 008), and the tier's own answer to what it is
//!    running.
//! 2. The newest green `build_runs` row per (version, platform) for every version
//!    a tier names. This is what [`crate::routes::promotion`] resolves a rollback
//!    to, and on a two-architecture product it is a different row per node.
//! 3. `versions.artifact_path` for those versions. The pre-identity path, still
//!    the first thing a canary rollback reads.
//!
//! The three overlap heavily and are unioned rather than ranked: being reachable
//! by any of them is enough to make a directory load-bearing.

use crate::domain::AppId;
use anyhow::Result;
use sqlx::{Row, SqlitePool};
use std::collections::{BTreeSet, HashSet};
use std::path::{Path, PathBuf};

/// The release directories that must survive a gc, by the name gc matches on.
///
/// A newtype rather than a bare `HashSet<String>` so the thing being passed
/// through publish and into gc says what it is at every hop, and so a caller
/// cannot hand it a set of paths, versions, or digests by accident.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct PinnedReleases(HashSet<String>);

impl PinnedReleases {
    /// Nothing to protect. What a caller with no deployed state passes — a unit
    /// test, or a store that has never served anything.
    pub fn none() -> Self {
        Self::default()
    }

    pub fn contains(&self, dir_name: &str) -> bool {
        self.0.contains(dir_name)
    }

    pub fn len(&self) -> usize {
        self.0.len()
    }

    pub fn is_empty(&self) -> bool {
        self.0.is_empty()
    }

    /// The pinned names in a stable order.
    ///
    /// Sorted rather than in hash order because the remote gc embeds these in a
    /// shell script: an unordered set would rewrite the script text on every
    /// deploy with no change of meaning, and a script that differs run to run is
    /// one nobody can diff against the last one that worked.
    pub fn sorted_names(&self) -> Vec<&str> {
        let mut names: Vec<&str> = self.0.iter().map(String::as_str).collect();
        names.sort_unstable();
        names
    }
}

impl FromIterator<String> for PinnedReleases {
    fn from_iter<T: IntoIterator<Item = String>>(iter: T) -> Self {
        Self(iter.into_iter().collect())
    }
}

/// Which of a tier's two artifacts a reference is.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Role {
    /// What the tier is running.
    Current,
    /// The one step of rollback history `tier_state` keeps.
    Previous,
}

impl Role {
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::Current => "current",
            Self::Previous => "previous",
        }
    }
}

impl std::fmt::Display for Role {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(self.as_str())
    }
}

/// One artifact a tier still names, and the bytes that have to be present for
/// the reference to be honoured.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ArtifactRef {
    pub tier: String,
    pub role: Role,
    /// The version label, when the tier records one.
    pub version: Option<String>,
    /// The release directory the reference resolves to.
    pub dir: PathBuf,
    /// The file whose absence makes the reference unusable — the primary binary
    /// inside `dir`. A directory holding only a MANIFEST is not a rollback
    /// target, and both stranded dirs in the 2026-08-25 measurement were exactly
    /// that, so existence of the directory is the wrong question to ask.
    pub binary: PathBuf,
}

/// The directory name gc sees under `releases/`.
///
/// `None` for a recorded path with no final component, which is a malformed row
/// rather than a reference to the root.
fn dir_name(dir: &Path) -> Option<String> {
    dir.file_name()
        .and_then(|n| n.to_str())
        .map(ToOwned::to_owned)
}

/// Every release directory the deployed state still points at, by the name gc
/// matches on.
///
/// Scoped to one product because `release_root` is: a second app's tiers name
/// directories in a store this gc never walks.
pub async fn pinned_dirs(pool: &SqlitePool, app: &AppId) -> Result<PinnedReleases> {
    let mut pinned = HashSet::new();

    // (1) + (2): every staged_path reachable from the tier's build ids, and
    // every newest-green build per (version, platform) for the versions the
    // tiers name. One query: the second set is what a rollback resolves to and
    // the first is what the tier is on, and a row can be in both.
    let rows = sqlx::query(
        "SELECT DISTINCT br.staged_path
           FROM build_runs br
          WHERE br.app = ?1
            AND br.staged_path IS NOT NULL
            AND (
                  br.id IN (SELECT current_build_id  FROM tier_state WHERE app = ?1
                            UNION ALL
                            SELECT previous_build_id FROM tier_state WHERE app = ?1)
               OR (br.result = 'passed'
                   AND br.version IN (SELECT current_version  FROM tier_state WHERE app = ?1
                                      UNION ALL
                                      SELECT previous_version FROM tier_state WHERE app = ?1)
                   AND br.id = (SELECT MAX(id) FROM build_runs
                                 WHERE app = ?1
                                   AND version = br.version
                                   AND platform IS br.platform
                                   AND result = 'passed'))
            )",
    )
    .bind(app)
    .fetch_all(pool)
    .await?;
    for r in rows {
        let path: String = r.get("staged_path");
        if let Some(name) = dir_name(Path::new(&path)) {
            pinned.insert(name);
        }
    }

    // (3) The pre-identity path. `artifact_path` names the primary binary, so
    // the directory is its parent.
    let rows = sqlx::query(
        "SELECT DISTINCT v.artifact_path
           FROM versions v
          WHERE v.app = ?1
            AND v.version IN (SELECT current_version  FROM tier_state WHERE app = ?1
                              UNION ALL
                              SELECT previous_version FROM tier_state WHERE app = ?1)",
    )
    .bind(app)
    .fetch_all(pool)
    .await?;
    for r in rows {
        let path: String = r.get("artifact_path");
        if let Some(name) = Path::new(&path).parent().and_then(dir_name) {
            pinned.insert(name);
        }
    }

    Ok(PinnedReleases(pinned))
}

/// What each tier is running and what it would roll back to, resolved the way
/// promote and rollback resolve it: the build row's `staged_path` when the tier
/// has an identity, `versions.artifact_path` when it predates one.
///
/// One entry per (tier, role) that resolves to a path at all. A tier with no
/// previous version contributes one entry, not two.
pub async fn tier_refs(
    pool: &SqlitePool,
    app: &AppId,
    primary_bin: &str,
) -> Result<Vec<ArtifactRef>> {
    let rows = sqlx::query(
        "SELECT ts.tier,
                ts.current_version,  cb.staged_path AS current_dir,  cv.artifact_path AS current_bin,
                ts.previous_version, pb.staged_path AS previous_dir, pv.artifact_path AS previous_bin
           FROM tier_state ts
           LEFT JOIN build_runs cb ON cb.id = ts.current_build_id
           LEFT JOIN build_runs pb ON pb.id = ts.previous_build_id
           LEFT JOIN versions   cv ON cv.app = ts.app AND cv.version = ts.current_version
           LEFT JOIN versions   pv ON pv.app = ts.app AND pv.version = ts.previous_version
          WHERE ts.app = ?
          ORDER BY ts.tier",
    )
    .bind(app)
    .fetch_all(pool)
    .await?;

    let mut refs = Vec::new();
    for r in rows {
        let tier: String = r.get("tier");
        for (role, version, dir_col, bin_col) in [
            (
                Role::Current,
                r.get::<Option<String>, _>("current_version"),
                r.get::<Option<String>, _>("current_dir"),
                r.get::<Option<String>, _>("current_bin"),
            ),
            (
                Role::Previous,
                r.get::<Option<String>, _>("previous_version"),
                r.get::<Option<String>, _>("previous_dir"),
                r.get::<Option<String>, _>("previous_bin"),
            ),
        ] {
            // The build row wins when there is one: it is the identity, and the
            // version label can be shared by several builds. `artifact_path` is
            // the fallback for a tier that predates migration 008 — the same
            // fallback the gate scope and the canary rollback take.
            let resolved = match (dir_col, bin_col) {
                (Some(dir), _) => {
                    let dir = PathBuf::from(dir);
                    let binary = dir.join(primary_bin);
                    Some((dir, binary))
                }
                (None, Some(bin)) => {
                    let binary = PathBuf::from(bin);
                    binary.parent().map(|d| (d.to_path_buf(), binary.clone()))
                }
                (None, None) => None,
            };
            if let Some((dir, binary)) = resolved {
                refs.push(ArtifactRef {
                    tier: tier.clone(),
                    role,
                    version,
                    dir,
                    binary,
                });
            }
        }
    }
    Ok(refs)
}

/// The references whose bytes are gone.
///
/// Deliberately a `stat` per reference rather than a directory walk: there are
/// at most two per tier, and asking about the exact file promote would rsync is
/// the only question whose answer means anything.
pub async fn missing(refs: &[ArtifactRef]) -> Vec<ArtifactRef> {
    let mut gone = Vec::new();
    for r in refs {
        if !tokio::fs::try_exists(&r.binary).await.unwrap_or(false) {
            gone.push(r.clone());
        }
    }
    gone
}

/// One line naming what is gone, for a log or a `/state` condition.
///
/// Sorted and deduplicated by tier so the sentence is stable across reads —
/// an operator surface that reworded itself every poll would read as churn.
pub fn describe(missing: &[ArtifactRef]) -> String {
    let lines: BTreeSet<String> = missing
        .iter()
        .map(|r| {
            let version = r.version.as_deref().unwrap_or("unknown version");
            format!("{} {} {version} ({})", r.tier, r.role, r.binary.display())
        })
        .collect();
    lines.into_iter().collect::<Vec<_>>().join("; ")
}

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

    async fn fresh_pool() -> SqlitePool {
        let pool = SqlitePoolOptions::new()
            .max_connections(1)
            .connect("sqlite::memory:")
            .await
            .unwrap();
        sqlx::migrate!("./migrations").run(&pool).await.unwrap();
        pool
    }

    fn app() -> AppId {
        AppId::default()
    }

    async fn tier(pool: &SqlitePool, name: &str, ord: i64) {
        sqlx::query("INSERT INTO tiers (app, name, ord, provisioned) VALUES ('mnw', ?, ?, 1)")
            .bind(name)
            .bind(ord)
            .execute(pool)
            .await
            .unwrap();
        sqlx::query("INSERT INTO tier_state (app, tier) VALUES ('mnw', ?)")
            .bind(name)
            .execute(pool)
            .await
            .unwrap();
    }

    /// A settled build row with an identity, as `runs::set_identity` leaves it.
    async fn build(pool: &SqlitePool, version: &str, dir: &str, platform: Option<&str>) -> i64 {
        sqlx::query(
            "INSERT INTO build_runs (app, sha, version, result, started_at, bundle_digest,
                                     staged_path, platform)
             VALUES ('mnw', 'abc123', ?, 'passed', '2026-08-25T00:00:00Z', ?, ?, ?)",
        )
        .bind(version)
        .bind(dir)
        .bind(format!("/srv/sando/releases/{dir}"))
        .bind(platform)
        .execute(pool)
        .await
        .unwrap()
        .last_insert_rowid()
    }

    async fn version_row(pool: &SqlitePool, version: &str, artifact_path: &str) {
        sqlx::query(
            "INSERT INTO versions (app, version, git_sha, built_at, artifact_path)
             VALUES ('mnw', ?, 'abc123', '2026-08-25T00:00:00Z', ?)",
        )
        .bind(version)
        .bind(artifact_path)
        .execute(pool)
        .await
        .unwrap();
    }

    /// Fixture-only: the SET clause varies per test and every value in it is a
    /// literal written here, so `raw_sql` is the honest tool rather than a
    /// bound query that cannot take a clause.
    async fn set_state(pool: &SqlitePool, tier: &str, sql: &str) {
        sqlx::raw_sql(sqlx::AssertSqlSafe(format!(
            "UPDATE tier_state SET {sql} WHERE app = 'mnw' AND tier = '{tier}'"
        )))
        .execute(pool)
        .await
        .unwrap();
    }

    #[tokio::test]
    async fn pins_what_every_tier_is_running_and_would_roll_back_to() {
        let pool = fresh_pool().await;
        tier(&pool, "host", 0).await;
        tier(&pool, "a", 1).await;

        let cur = build(&pool, "0.16.1", "aaaaaaaaaaaaaaaa", None).await;
        let prev = build(&pool, "0.11.20", "bbbbbbbbbbbbbbbb", None).await;
        set_state(
            &pool,
            "host",
            &format!(
                "current_version = '0.16.1', current_build_id = {cur},
                 previous_version = '0.11.20', previous_build_id = {prev}"
            ),
        )
        .await;

        let pinned = pinned_dirs(&pool, &app()).await.unwrap();
        assert!(pinned.contains("aaaaaaaaaaaaaaaa"));
        assert!(pinned.contains("bbbbbbbbbbbbbbbb"));
        assert_eq!(pinned.len(), 2, "{pinned:?}");
    }

    /// The exact shape that stranded prod: three builds of one version, the
    /// tier on the OLDEST of them. Pinning by version alone would keep the
    /// newest rebuild and lose the bytes actually deployed.
    #[tokio::test]
    async fn pins_the_build_the_tier_is_on_not_the_newest_rebuild_of_its_version() {
        let pool = fresh_pool().await;
        tier(&pool, "host", 0).await;

        let deployed = build(&pool, "0.16.1", "0000000000000001", None).await;
        build(&pool, "0.16.1", "0000000000000002", None).await;
        let newest = build(&pool, "0.16.1", "0000000000000003", None).await;
        set_state(
            &pool,
            "host",
            &format!("current_version = '0.16.1', current_build_id = {deployed}"),
        )
        .await;

        let pinned = pinned_dirs(&pool, &app()).await.unwrap();
        assert!(
            pinned.contains("0000000000000001"),
            "the deployed build must be pinned: {pinned:?}"
        );
        // And the newest green build of that version too: that is the row a
        // rollback to 0.16.1 resolves through, so evicting it breaks a path the
        // tier can still take.
        assert!(pinned.contains("0000000000000003"), "{pinned:?}");
        assert_ne!(deployed, newest);
        assert!(!pinned.contains("0000000000000002"), "{pinned:?}");
    }

    /// Two architectures of one version are two artifacts, and a rollback picks
    /// per node. Both have to survive.
    #[tokio::test]
    async fn pins_every_platform_of_a_referenced_version() {
        let pool = fresh_pool().await;
        tier(&pool, "host", 0).await;
        build(&pool, "0.4.0", "aaaa000000000000", Some("linux/x86_64")).await;
        build(&pool, "0.4.0", "bbbb000000000000", Some("linux/aarch64")).await;
        set_state(&pool, "host", "current_version = '0.4.0'").await;

        let pinned = pinned_dirs(&pool, &app()).await.unwrap();
        assert!(pinned.contains("aaaa000000000000"), "{pinned:?}");
        assert!(pinned.contains("bbbb000000000000"), "{pinned:?}");
    }

    /// A pre-identity tier has no build id at all. `versions.artifact_path` is
    /// the only handle, and it names the binary inside the dir.
    #[tokio::test]
    async fn pins_a_pre_identity_tier_through_versions() {
        let pool = fresh_pool().await;
        tier(&pool, "host", 0).await;
        version_row(
            &pool,
            "0.8.12",
            "/srv/sando/releases/cccccccccccccccc/makenotwork",
        )
        .await;
        set_state(&pool, "host", "current_version = '0.8.12'").await;

        let pinned = pinned_dirs(&pool, &app()).await.unwrap();
        assert!(pinned.contains("cccccccccccccccc"), "{pinned:?}");
    }

    /// `release_root` is per product, so another product's references must not
    /// leak into this store's keep-set.
    #[tokio::test]
    async fn another_products_references_are_not_pinned_here() {
        let pool = fresh_pool().await;
        tier(&pool, "host", 0).await;
        sqlx::query("INSERT INTO tiers (app, name, ord, provisioned) VALUES ('pom', 'host', 0, 1)")
            .execute(&pool)
            .await
            .unwrap();
        sqlx::query("INSERT INTO tier_state (app, tier) VALUES ('pom', 'host')")
            .execute(&pool)
            .await
            .unwrap();
        sqlx::query(
            "INSERT INTO build_runs (app, sha, version, result, started_at, staged_path)
             VALUES ('pom', 'abc', '1.0.0', 'passed', '2026-08-25T00:00:00Z',
                     '/srv/sando-pom/releases/dddddddddddddddd')",
        )
        .execute(&pool)
        .await
        .unwrap();
        sqlx::query(
            "UPDATE tier_state SET current_version = '1.0.0' WHERE app = 'pom' AND tier = 'host'",
        )
        .execute(&pool)
        .await
        .unwrap();

        let pinned = pinned_dirs(&pool, &app()).await.unwrap();
        assert!(pinned.is_empty(), "{pinned:?}");
    }

    #[tokio::test]
    async fn a_tier_that_has_never_deployed_pins_nothing() {
        let pool = fresh_pool().await;
        tier(&pool, "host", 0).await;
        assert!(pinned_dirs(&pool, &app()).await.unwrap().is_empty());
        assert!(
            tier_refs(&pool, &app(), "makenotwork")
                .await
                .unwrap()
                .is_empty()
        );
    }

    #[tokio::test]
    async fn a_reference_whose_binary_is_gone_is_reported_missing() {
        let tmp = tempfile::tempdir().unwrap();
        let releases = tmp.path().join("releases");
        let present = releases.join("aaaaaaaaaaaaaaaa");
        let hollow = releases.join("bbbbbbbbbbbbbbbb");
        tokio::fs::create_dir_all(&present).await.unwrap();
        tokio::fs::create_dir_all(&hollow).await.unwrap();
        tokio::fs::write(present.join("makenotwork"), b"bin")
            .await
            .unwrap();
        // The measured state: MANIFEST and tree present, no binary. A directory
        // check would call this healthy.
        tokio::fs::write(hollow.join("MANIFEST"), b"")
            .await
            .unwrap();

        let pool = fresh_pool().await;
        tier(&pool, "host", 0).await;
        let cur = sqlx::query(
            "INSERT INTO build_runs (app, sha, version, result, started_at, staged_path)
             VALUES ('mnw', 'abc', '0.16.1', 'passed', '2026-08-25T00:00:00Z', ?)",
        )
        .bind(present.to_string_lossy().as_ref())
        .execute(&pool)
        .await
        .unwrap()
        .last_insert_rowid();
        let prev = sqlx::query(
            "INSERT INTO build_runs (app, sha, version, result, started_at, staged_path)
             VALUES ('mnw', 'abc', '0.11.20', 'passed', '2026-08-25T00:00:00Z', ?)",
        )
        .bind(hollow.to_string_lossy().as_ref())
        .execute(&pool)
        .await
        .unwrap()
        .last_insert_rowid();
        set_state(
            &pool,
            "host",
            &format!(
                "current_version = '0.16.1', current_build_id = {cur},
                 previous_version = '0.11.20', previous_build_id = {prev}"
            ),
        )
        .await;

        let refs = tier_refs(&pool, &app(), "makenotwork").await.unwrap();
        assert_eq!(refs.len(), 2);
        let gone = missing(&refs).await;
        assert_eq!(gone.len(), 1, "{gone:?}");
        assert_eq!(gone[0].role, Role::Previous);
        assert_eq!(gone[0].version.as_deref(), Some("0.11.20"));

        let said = describe(&gone);
        assert!(said.contains("host"), "{said}");
        assert!(said.contains("previous"), "{said}");
        assert!(said.contains("0.11.20"), "{said}");
    }

    /// The rollback path reads `versions.artifact_path` directly, so the same
    /// question has to be asked of a pre-identity tier.
    #[tokio::test]
    async fn a_pre_identity_reference_is_checked_at_its_artifact_path() {
        let tmp = tempfile::tempdir().unwrap();
        let dir = tmp.path().join("releases").join("cccccccccccccccc");
        tokio::fs::create_dir_all(&dir).await.unwrap();
        let bin = dir.join("makenotwork");

        let pool = fresh_pool().await;
        tier(&pool, "host", 0).await;
        version_row(&pool, "0.8.12", bin.to_string_lossy().as_ref()).await;
        set_state(&pool, "host", "current_version = '0.8.12'").await;

        let refs = tier_refs(&pool, &app(), "makenotwork").await.unwrap();
        assert_eq!(refs.len(), 1);
        assert_eq!(refs[0].dir, dir);
        assert_eq!(missing(&refs).await.len(), 1);

        tokio::fs::write(&bin, b"bin").await.unwrap();
        assert!(missing(&refs).await.is_empty());
    }
}