//! **E3 — mirror del store content-addressed** ([SDD 13 §E3](../../docs/13-release-engineering.md)).
//!
//! El `/store` de hammer es un CAS: cada artefacto vive en un directorio `<hash>-<name>` y *el hash es
//! la dirección*. Un mirror, entonces, es ese CAS **replicado** entre dos máquinas + **resolución por
//! hash**. Esta capa replica los artefactos que le faltan al destino y **ancla la integridad en
//! `of_tree`** (el content-hash BLAKE3 del árbol, [`ArtifactHash::of_tree`]): el receptor **recomputa**
//! `of_tree` sobre lo que copió y **exige** que case el del índice **antes** de sellar. Una transferencia
//! corrupta (o un origen que sirvió bytes que no casan el hash anunciado) se **rechaza**, no se instala.
//!
//! El transporte de esta primera entrega es **sistema de ficheros**: el "remoto" es una ruta a otro
//! store (que puede ser un montaje de red, sshfs, etc.). El transporte sobre SSH/red —que el producto ya
//! expone vía sshd— es un envoltorio ortogonal posterior; la lógica de CAS (índice, diff, copia
//! verificada, sellado atómico) vive aquí y es agnóstica del medio.
//!
//! **Modelo de confianza:** la verificación `of_tree` detecta corrupción de transporte y que el origen
//! sirva un árbol que no case el hash que anuncia. Anclar ese hash a una raíz de confianza firmada
//! (el log de transparencia `bootstrap.json` / la atestación de Etapa D) es la capa de endurecimiento
//! posterior: hoy el invariante CAS que se *fuerza* es "el contenido recibido HASHEA a lo que el índice
//! dice".

use std::path::Path;

use hammer_core::{ArtifactHash, Store};
use serde::{Deserialize, Serialize};

#[derive(Debug, thiserror::Error)]
pub enum Error {
    #[error("io: {0}")]
    Io(#[from] std::io::Error),
    #[error("core: {0}")]
    Core(#[from] hammer_core::Error),
    #[error(
        "integridad: el artefacto '{dir}' recibido hashea a {got} pero el índice anuncia {want} — \
         transferencia corrupta o manipulada; NO se selló"
    )]
    Verify { dir: String, want: String, got: String },
    #[error("mirror: {0}")]
    Other(String),
}

pub type Result<T> = std::result::Result<T, Error>;

/// Una entrada del índice del store: el directorio del artefacto (`<hash>-<name>`) y su **content-hash**
/// (`of_tree`, `b3:…`). El `dir` es la dirección (input-addressed) y `content` la huella de los bytes:
/// dos artefactos con el mismo `dir` deben tener el mismo `content` (coherencia del CAS).
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ArtifactRef {
    pub dir: String,
    pub content: String,
}

/// Índice de un store: la lista de sus artefactos con su content-hash, ordenada por `dir` (estable).
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct StoreIndex {
    pub artifacts: Vec<ArtifactRef>,
}

impl StoreIndex {
    /// Busca un artefacto por su `dir`.
    pub fn get(&self, dir: &str) -> Option<&ArtifactRef> {
        self.artifacts.iter().find(|a| a.dir == dir)
    }
}

/// ¿El nombre de directorio tiene forma de artefacto del store? = 64 hex + `-` + `<name>` no vacío. Así
/// se ignoran `bootstrap.json`, `.bootstrap-tmp`, índices y otros ficheros del root del store.
fn is_artifact_dir(name: &str) -> bool {
    let bytes = name.as_bytes();
    bytes.len() > 65
        && bytes[64] == b'-'
        && bytes[..64].iter().all(|b| b.is_ascii_hexdigit())
}

/// Parte `<64hex>-<name>` en `(ArtifactHash, name)`. El `name` puede contener `-` (p.ej.
/// `product-attested-rootfs`); sólo el primer `-`, en la posición 64, separa el hash.
fn parse_dir(dir: &str) -> Option<(ArtifactHash, String)> {
    if !is_artifact_dir(dir) {
        return None;
    }
    Some((ArtifactHash::from_hex(&dir[..64]), dir[65..].to_string()))
}

/// Construye el índice de un store: para cada directorio-artefacto, su `of_tree`. Un store inexistente
/// se trata como vacío (el destino aún no creado en un primer `pull`). Determinista (ordenado por `dir`).
pub fn index(store_root: &Path) -> Result<StoreIndex> {
    let mut artifacts = Vec::new();
    if store_root.is_dir() {
        for entry in std::fs::read_dir(store_root)? {
            let entry = entry?;
            let name = entry.file_name().to_string_lossy().into_owned();
            if !is_artifact_dir(&name) || !entry.path().is_dir() {
                continue;
            }
            let content = ArtifactHash::of_tree(&entry.path())?.as_str().to_string();
            artifacts.push(ArtifactRef { dir: name, content });
        }
    }
    artifacts.sort_by(|a, b| a.dir.cmp(&b.dir));
    Ok(StoreIndex { artifacts })
}

/// Diferencia entre dos stores: `(faltan_en_dst, faltan_en_src, conflictos)`. Un **conflicto** es un
/// artefacto presente en ambos con el MISMO `dir` pero DISTINTO `content` (misma dirección, otros bytes:
/// rotura de la coherencia del CAS — no-reproducibilidad o corrupción). No copia nada.
pub fn diff(
    src_root: &Path,
    dst_root: &Path,
) -> Result<(Vec<ArtifactRef>, Vec<ArtifactRef>, Vec<String>)> {
    let si = index(src_root)?;
    let di = index(dst_root)?;
    let mut missing_in_dst = Vec::new();
    let mut conflicts = Vec::new();
    for a in &si.artifacts {
        match di.get(&a.dir) {
            None => missing_in_dst.push(a.clone()),
            Some(d) if d.content != a.content => conflicts.push(a.dir.clone()),
            Some(_) => {}
        }
    }
    let missing_in_src = di
        .artifacts
        .iter()
        .filter(|d| si.get(&d.dir).is_none())
        .cloned()
        .collect();
    Ok((missing_in_dst, missing_in_src, conflicts))
}

/// Reporte de un [`sync`]: qué se copió, cuántos ya estaban, conflictos y fallos de verificación.
#[derive(Debug, Clone, Default)]
pub struct SyncReport {
    /// `dir`s copiados+verificados+sellados en el destino.
    pub copied: Vec<String>,
    /// Ya presentes con el mismo content-hash (idempotencia).
    pub skipped_present: usize,
    /// Presentes en ambos con content-hash distinto (CAS incoherente; NO se sobrescriben).
    pub conflicts: Vec<String>,
    /// `(dir, motivo)` de los que fallaron la verificación de integridad (no se sellaron).
    pub failed: Vec<(String, String)>,
}

impl SyncReport {
    /// Éxito = nada falló la verificación y no hubo conflictos.
    pub fn ok(&self) -> bool {
        self.failed.is_empty() && self.conflicts.is_empty()
    }
}

/// Copia el árbol de un artefacto a un staging bajo el destino, **recomputa `of_tree`**, exige que case
/// `expected_content` y sólo entonces lo **sella** en `dst_store` (rename atómico + read-only, vía
/// [`Store::seal`]). El staging vive bajo el root del destino ⇒ mismo filesystem ⇒ el rename del seal es
/// atómico; ante un hash que no casa, se borra el staging y NO se instala nada.
fn install_verified(
    src_dir: &Path,
    dst_store: &Store,
    h: &ArtifactHash,
    name: &str,
    expected_content: &str,
) -> Result<()> {
    let dir_label = format!("{}-{name}", h.as_str().trim_start_matches("b3:"));
    let staging = dst_store
        .root()
        .join(".mirror-tmp")
        .join(format!("{dir_label}.incoming"));
    let _ = std::fs::remove_dir_all(&staging);
    if let Some(parent) = staging.parent() {
        std::fs::create_dir_all(parent)?;
    }

    let result = (|| -> Result<()> {
        copy_tree(src_dir, &staging)?;
        let got = ArtifactHash::of_tree(&staging)?;
        if got.as_str() != expected_content {
            return Err(Error::Verify {
                dir: dir_label.clone(),
                want: expected_content.to_string(),
                got: got.as_str().to_string(),
            });
        }
        dst_store.seal(&staging, h, name)?;
        Ok(())
    })();
    let _ = std::fs::remove_dir_all(&staging);
    result
}

/// **Sincroniza** `src`→`dst`: copia a `dst` cada artefacto que le falta (verificando `of_tree` antes de
/// sellar), salta los ya presentes con el mismo contenido y reporta conflictos (mismo `dir`, otro
/// contenido). Idempotente: re-correr sobre un destino ya al día copia cero. Es el motor de `push`
/// (`sync(local, remoto)`) y `pull` (`sync(remoto, local)`).
pub fn sync(src_root: &Path, dst_root: &Path) -> Result<SyncReport> {
    let si = index(src_root)?;
    let di = index(dst_root)?;
    let dst_store = Store::open(dst_root)?;
    let mut report = SyncReport::default();

    for a in &si.artifacts {
        match di.get(&a.dir) {
            Some(d) if d.content == a.content => {
                report.skipped_present += 1;
                continue;
            }
            Some(_) => {
                report.conflicts.push(a.dir.clone());
                continue;
            }
            None => {}
        }
        let (h, name) = match parse_dir(&a.dir) {
            Some(p) => p,
            None => continue, // index() ya filtra, defensivo
        };
        match install_verified(&src_root.join(&a.dir), &dst_store, &h, &name, &a.content) {
            Ok(()) => {
                tracing::info!(dir = %a.dir, "mirror: artefacto replicado y verificado");
                report.copied.push(a.dir.clone());
            }
            Err(e) => {
                tracing::warn!(dir = %a.dir, error = %e, "mirror: fallo de replicación");
                report.failed.push((a.dir.clone(), e.to_string()));
            }
        }
    }
    Ok(report)
}

/// Copia recursiva `src`→`dst` preservando symlinks (target literal) y modos (incl. bit de ejecución,
/// que `of_tree` mira). El destino se asume inexistente (staging recién limpiado).
fn copy_tree(src: &Path, dst: &Path) -> Result<()> {
    let meta = std::fs::symlink_metadata(src)?;
    let ft = meta.file_type();
    if ft.is_symlink() {
        let target = std::fs::read_link(src)?;
        std::os::unix::fs::symlink(target, dst)?;
    } else if ft.is_dir() {
        std::fs::create_dir_all(dst)?;
        for entry in std::fs::read_dir(src)? {
            let entry = entry?;
            copy_tree(&entry.path(), &dst.join(entry.file_name()))?;
        }
    } else {
        std::fs::copy(src, dst)?;
    }
    Ok(())
}

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

    /// Sella un artefacto sintético `<hex>-<name>` en `root` con los ficheros que pinte `build`.
    fn seal(root: &Path, hex: &str, name: &str, build: impl Fn(&Path)) -> String {
        let store = Store::open(root).unwrap();
        let h = ArtifactHash::from_hex(hex);
        let staging = root.join(".bootstrap-tmp").join(h.store_dir_name(name));
        let _ = std::fs::remove_dir_all(&staging);
        std::fs::create_dir_all(&staging).unwrap();
        build(&staging);
        store.seal(&staging, &h, name).unwrap();
        h.store_dir_name(name)
    }

    fn h64(prefix: &str) -> String {
        format!("{prefix}{}", "0".repeat(64 - prefix.len()))
    }

    #[test]
    fn artifact_dir_shape_filters_non_artifacts() {
        assert!(is_artifact_dir(&format!("{}-busybox", h64("ab"))));
        assert!(is_artifact_dir(&format!("{}-product-attested-rootfs", h64("ff"))));
        assert!(!is_artifact_dir("bootstrap.json"));
        assert!(!is_artifact_dir(".bootstrap-tmp"));
        assert!(!is_artifact_dir(&format!("{}-", h64("ab"))), "sin name");
        assert!(!is_artifact_dir(&format!("{}xy-z", h64("ab"))), "65 chars no-hex antes del -");
        // parse_dir respeta los '-' del name.
        let (hh, name) = parse_dir(&format!("{}-find-utils-xargs", h64("cd"))).unwrap();
        assert_eq!(name, "find-utils-xargs");
        assert!(hh.as_str().starts_with("b3:"));
    }

    #[test]
    fn index_lists_artifacts_with_content_hash_and_skips_junk() {
        let tmp = tempfile::tempdir().unwrap();
        let root = tmp.path();
        seal(root, &h64("aa"), "busybox", |w| {
            std::fs::write(w.join("bin"), b"\x7fELF").unwrap();
        });
        // ruido del root del store que NO debe entrar al índice.
        std::fs::write(root.join("bootstrap.json"), b"{}").unwrap();
        std::fs::create_dir_all(root.join(".bootstrap-tmp")).unwrap();

        let idx = index(root).unwrap();
        assert_eq!(idx.artifacts.len(), 1, "sólo el artefacto, no bootstrap.json ni .bootstrap-tmp");
        assert!(idx.artifacts[0].dir.ends_with("-busybox"));
        assert!(idx.artifacts[0].content.starts_with("b3:"));
        // content == of_tree real del dir.
        let ot = ArtifactHash::of_tree(&root.join(&idx.artifacts[0].dir)).unwrap();
        assert_eq!(idx.artifacts[0].content, ot.as_str());
    }

    #[test]
    fn sync_replicates_missing_and_is_idempotent() {
        let src = tempfile::tempdir().unwrap();
        let dst = tempfile::tempdir().unwrap();
        seal(src.path(), &h64("aa"), "busybox", |w| {
            std::fs::write(w.join("bin"), b"\x7fELF-bb").unwrap();
            std::os::unix::fs::symlink("bin", w.join("sh")).unwrap();
        });
        seal(src.path(), &h64("bb"), "uutils", |w| {
            std::fs::create_dir_all(w.join("usr/bin")).unwrap();
            std::fs::write(w.join("usr/bin/coreutils"), b"\x7fELF-uu").unwrap();
        });

        let r = sync(src.path(), dst.path()).unwrap();
        assert!(r.ok());
        assert_eq!(r.copied.len(), 2);
        // el destino quedó con los mismos artefactos y mismos content-hash.
        let (md, ms, conf) = diff(src.path(), dst.path()).unwrap();
        assert!(md.is_empty() && conf.is_empty(), "destino al día");
        assert!(ms.is_empty(), "el destino no tiene de más");
        // los bytes (y el symlink) viajaron.
        let bbdir = dst.path().join(format!("{}-busybox", h64("aa")));
        assert_eq!(std::fs::read(bbdir.join("bin")).unwrap(), b"\x7fELF-bb");
        assert_eq!(std::fs::read_link(bbdir.join("sh")).unwrap(), std::path::PathBuf::from("bin"));

        // 2ª corrida: idempotente, copia 0.
        let r2 = sync(src.path(), dst.path()).unwrap();
        assert!(r2.copied.is_empty() && r2.skipped_present == 2);
        assert!(!dst.path().join(".mirror-tmp").join(format!("{}-busybox.incoming", h64("aa"))).exists(),
            "el staging se limpió");
    }

    #[test]
    fn sync_rejects_corrupted_transfer_without_sealing() {
        // El guardrail de integridad: si lo que llega NO hashea a lo que el índice anuncia, se rechaza.
        let dst = tempfile::tempdir().unwrap();
        let src_dir = tempfile::tempdir().unwrap();
        std::fs::write(src_dir.path().join("bin"), b"bytes-reales").unwrap();
        let store = Store::open(dst.path()).unwrap();
        let h = ArtifactHash::from_hex(&h64("cc"));
        // expected_content MENTIROSO (no es el of_tree real del src) ⇒ debe fallar Verify y no sellar.
        let err = install_verified(src_dir.path(), &store, &h, "evil", "b3:deadbeef").unwrap_err();
        assert!(matches!(err, Error::Verify { .. }), "esperaba Verify, vino {err}");
        assert!(!store.has(&h, "evil"), "un artefacto que no verifica NO se sella");
        assert!(!dst.path().join(".mirror-tmp").join(format!("{}-evil.incoming", h64("cc"))).exists(),
            "staging limpiado tras el rechazo");
    }

    #[test]
    fn sync_flags_conflicts_and_does_not_overwrite() {
        // Mismo dir en src y dst pero contenido distinto = conflicto (CAS incoherente). No se pisa.
        let src = tempfile::tempdir().unwrap();
        let dst = tempfile::tempdir().unwrap();
        seal(src.path(), &h64("dd"), "x", |w| std::fs::write(w.join("f"), b"SRC").unwrap());
        seal(dst.path(), &h64("dd"), "x", |w| std::fs::write(w.join("f"), b"DST-distinto").unwrap());

        let r = sync(src.path(), dst.path()).unwrap();
        assert!(!r.ok(), "hay conflicto ⇒ no-ok");
        assert_eq!(r.conflicts.len(), 1);
        assert!(r.copied.is_empty());
        // el destino conserva SUS bytes (no se sobrescribió).
        assert_eq!(std::fs::read(dst.path().join(format!("{}-x", h64("dd"))).join("f")).unwrap(), b"DST-distinto");
    }
}