holger-server-lib 0.6.9

//! Repository REPLICATION — copy a repo's artifacts from a SOURCE holger store
//! into a DESTINATION store, content-id idempotent (parity §I,
//! `.nornir/replication-design.md`).
//!
//! The engine is a **pure function** over two backend sets: hand it a list of
//! `(source_backend, dest_backend)` repo PAIRS and a [`ReplicationOptions`], and
//! it walks each source's `list()`, `fetch()`es the bytes, and — for every
//! artifact the destination does not already hold *by content id* — `put()`s it
//! into the writable destination. Its only I/O is the backend calls it is given,
//! so it is unit-testable with in-memory backends and reused verbatim by both
//! surfaces: the `holger-server replicate --from --to` CLI verb (two separate
//! file-backed stores) and the `ReplicationService.Replicate` gRPC RPC (two repos
//! within one running server).
//!
//! **Read-from-source / write-to-dest, never the reverse.** The source is only
//! ever `list()`ed + `fetch()`ed; the destination is only ever `fetch()`ed (to
//! test presence) + `put()` (to copy). A source is never mutated.
//!
//! **Idempotent by content id.** The dedup key is the artifact's content digest
//! (lowercase-hex sha256 of the bytes, the same digest the search engine matches).
//! Before copying, the engine fetches the destination's bytes for the same
//! coordinate: if they are already present AND hash to the same content id, the
//! artifact is *skipped*. So a second replicate run over an unchanged source is a
//! pure no-op (every artifact skipped, nothing written) — the property the
//! executor test asserts. A coordinate whose destination content DIFFERS is not
//! "present by content id", so it is (re-)copied, converging the destination onto
//! the source.
//!
//! **Gated.** `execute` defaults to `false` — a dry run reports what *would* copy
//! and writes nothing; an actual copy only happens under
//! [`ReplicationOptions::executing`]. A destination backend that is not writable
//! fails that pair's artifacts *closed* (reported, never silently dropped).

use std::sync::Arc;

use serde::{Deserialize, Serialize};

use traits::{ArtifactId, RepositoryBackendTrait};

/// Lowercase-hex sha256 — the content id replication dedups on. Kept local (the
/// same choice `search.rs` made) so the engine is self-contained + testable in
/// isolation.
fn content_id(data: &[u8]) -> String {
    // LAW #5 dedup: consume the shared `nornir-hash` leaf (edda) so the replication
    // content id is byte-identical to search.rs / the backends' recompute.
    nornir_hash::sha256_hex(data)
}

/// Knobs for a replication run.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct ReplicationOptions {
    /// When `false` (the DEFAULT), a dry run: report what *would* copy, write
    /// nothing. An actual `put()` into the destination only happens when `true`.
    pub execute: bool,
    /// Per-pair cap on artifacts examined; `0` ⇒ unlimited. Bounds a huge source.
    pub limit: usize,
}

impl Default for ReplicationOptions {
    fn default() -> Self {
        // Safe default: preview only, unbounded.
        Self { execute: false, limit: 0 }
    }
}

impl ReplicationOptions {
    /// A dry-run preview (writes nothing), unbounded.
    pub fn preview() -> Self {
        Self::default()
    }
    /// An executing run (actually copies), unbounded.
    pub fn executing() -> Self {
        Self { execute: true, limit: 0 }
    }
}

/// What happened to a single source artifact under replication.
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum Outcome {
    /// Copied into the destination (an executing run wrote the bytes).
    Copied,
    /// Already present in the destination by content id — skipped (idempotent).
    Skipped,
    /// A dry run that would copy this artifact (nothing was written).
    WouldCopy,
    /// Could not replicate (source unreadable, destination read-only, put failed).
    Failed,
}

/// One source artifact's replication result, tagged with both repo names + its
/// content id.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct ReplicatedArtifact {
    pub source_repo: String,
    pub dest_repo: String,
    pub id: ArtifactId,
    /// Lowercase-hex sha256 of the source bytes (empty when the source could not
    /// be read).
    pub content_id: String,
    pub size_bytes: i64,
    pub outcome: Outcome,
    /// Human detail — the failure reason for [`Outcome::Failed`], else empty.
    pub detail: String,
}

/// The result of a [`replicate_pairs`] run: one row per source artifact examined,
/// plus derived tallies.
#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]
pub struct ReplicationReport {
    pub items: Vec<ReplicatedArtifact>,
    /// True when a pair hit its `limit` cap and more source artifacts remained.
    pub truncated: bool,
}

impl ReplicationReport {
    fn count(&self, o: Outcome) -> usize {
        self.items.iter().filter(|i| i.outcome == o).count()
    }
    /// Artifacts actually written into the destination.
    pub fn copied(&self) -> usize {
        self.count(Outcome::Copied)
    }
    /// Artifacts already present by content id (skipped).
    pub fn skipped(&self) -> usize {
        self.count(Outcome::Skipped)
    }
    /// Artifacts a dry run would copy.
    pub fn would_copy(&self) -> usize {
        self.count(Outcome::WouldCopy)
    }
    /// Artifacts that could not be replicated.
    pub fn failed(&self) -> usize {
        self.count(Outcome::Failed)
    }
    /// Total source artifacts examined.
    pub fn examined(&self) -> usize {
        self.items.len()
    }
    /// True when nothing failed — the replication is fully accounted for.
    pub fn is_complete(&self) -> bool {
        self.failed() == 0
    }
    /// Sum of bytes actually copied (0 on a dry run).
    pub fn copied_bytes(&self) -> i64 {
        self.items
            .iter()
            .filter(|i| i.outcome == Outcome::Copied)
            .map(|i| i.size_bytes)
            .sum()
    }
}

/// One source→destination repository pairing for a replication run. The source is
/// read-only here; the destination is where artifacts land.
pub struct RepoPair {
    pub source_name: String,
    pub source: Arc<dyn RepositoryBackendTrait>,
    pub dest_name: String,
    pub dest: Arc<dyn RepositoryBackendTrait>,
}

/// Replicate every pairing in `pairs`, source→dest, content-id idempotent. Pure
/// over the backends it is given (the only I/O is their `list`/`fetch`/`put`), so
/// both the CLI and the gRPC RPC drive it unchanged. A source that cannot be
/// listed contributes no rows for that pair (best-effort, logged); every artifact
/// that *is* listed yields exactly one [`ReplicatedArtifact`] row.
pub fn replicate_pairs(pairs: &[RepoPair], opts: &ReplicationOptions) -> ReplicationReport {
    let mut report = ReplicationReport::default();
    for pair in pairs {
        replicate_one_pair(pair, opts, &mut report);
    }
    report
}

/// Replicate a single source→dest pair, appending its rows to `report`.
fn replicate_one_pair(
    pair: &RepoPair,
    opts: &ReplicationOptions,
    report: &mut ReplicationReport,
) {
    let entries = match pair.source.list(None, 0) {
        Ok(e) => e,
        Err(e) => {
            log::warn!(
                "replicate: list failed in source repo '{}': {e}",
                pair.source_name
            );
            return;
        }
    };

    let cap = if opts.limit == 0 { usize::MAX } else { opts.limit };
    let dest_writable = pair.dest.is_writable();

    for (i, entry) in entries.into_iter().enumerate() {
        if i >= cap {
            report.truncated = true;
            break;
        }
        let id = entry.id;

        // Read the SOURCE bytes (the source is only ever read).
        let bytes = match pair.source.fetch(&id) {
            Ok(Some(b)) => b,
            Ok(None) => {
                // Listed but unreadable — record a failure, never a silent drop.
                report.items.push(row(
                    pair,
                    &id,
                    "",
                    0,
                    Outcome::Failed,
                    "source listed the artifact but fetch returned no bytes",
                ));
                continue;
            }
            Err(e) => {
                report.items.push(row(
                    pair,
                    &id,
                    "",
                    0,
                    Outcome::Failed,
                    &format!("source fetch failed: {e}"),
                ));
                continue;
            }
        };
        let cid = content_id(&bytes);
        let size = bytes.len() as i64;

        // Idempotency: already present in the DESTINATION by content id?
        let already_present = match pair.dest.fetch(&id) {
            Ok(Some(existing)) => content_id(&existing) == cid,
            // Absent, or destination read error → treat as "needs copy" (a read
            // error on the presence probe must not skip the copy).
            Ok(None) => false,
            Err(e) => {
                log::warn!(
                    "replicate: dest presence probe failed in '{}' for {}@{}: {e}",
                    pair.dest_name,
                    id.name,
                    id.version
                );
                false
            }
        };

        if already_present {
            report
                .items
                .push(row(pair, &id, &cid, size, Outcome::Skipped, ""));
            continue;
        }

        // Needs copy. A read-only destination fails CLOSED (reported, not dropped).
        if !dest_writable {
            report.items.push(row(
                pair,
                &id,
                &cid,
                size,
                Outcome::Failed,
                &format!("destination repo '{}' is read-only", pair.dest_name),
            ));
            continue;
        }

        // Dry run: report the intent, write nothing.
        if !opts.execute {
            report
                .items
                .push(row(pair, &id, &cid, size, Outcome::WouldCopy, ""));
            continue;
        }

        // Executing: copy the bytes into the destination.
        match pair.dest.put(&id, &bytes) {
            Ok(()) => report
                .items
                .push(row(pair, &id, &cid, size, Outcome::Copied, "")),
            Err(e) => report.items.push(row(
                pair,
                &id,
                &cid,
                size,
                Outcome::Failed,
                &format!("destination put failed: {e}"),
            )),
        }
    }
}

/// Build one report row (keeps `replicate_one_pair` readable).
fn row(
    pair: &RepoPair,
    id: &ArtifactId,
    cid: &str,
    size: i64,
    outcome: Outcome,
    detail: &str,
) -> ReplicatedArtifact {
    ReplicatedArtifact {
        source_repo: pair.source_name.clone(),
        dest_repo: pair.dest_name.clone(),
        id: id.clone(),
        content_id: cid.to_string(),
        size_bytes: size,
        outcome,
        detail: detail.to_string(),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use async_trait::async_trait;
    use std::sync::Mutex;
    use traits::{ArchiveInfo, ArtifactEntry, ArtifactFormat};

    /// A mutable in-memory backend: `put` really stores (so a copied artifact is
    /// afterwards `fetch`-able, and the round-trip is byte-checkable). `writable`
    /// gates writes. `list`/`fetch`/`put` all go through the shared map.
    struct MemRepo {
        name: String,
        writable: bool,
        store: Mutex<Vec<(ArtifactId, Vec<u8>)>>,
    }

    impl MemRepo {
        fn new(name: &str, writable: bool) -> Self {
            Self { name: name.into(), writable, store: Mutex::new(Vec::new()) }
        }
        fn seeded(name: &str, entries: Vec<(ArtifactId, Vec<u8>)>) -> Self {
            Self { name: name.into(), writable: false, store: Mutex::new(entries) }
        }
        fn arc(self) -> Arc<dyn RepositoryBackendTrait> {
            Arc::new(self)
        }
    }

    #[async_trait]
    impl RepositoryBackendTrait for MemRepo {
        fn name(&self) -> &str {
            &self.name
        }
        fn format(&self) -> ArtifactFormat {
            ArtifactFormat::Rust
        }
        fn is_writable(&self) -> bool {
            self.writable
        }
        fn fetch(&self, id: &ArtifactId) -> anyhow::Result<Option<Vec<u8>>> {
            Ok(self
                .store
                .lock()
                .unwrap()
                .iter()
                .find(|(e, _)| e == id)
                .map(|(_, b)| b.clone()))
        }
        fn put(&self, id: &ArtifactId, data: &[u8]) -> anyhow::Result<()> {
            if !self.writable {
                anyhow::bail!("read-only");
            }
            let mut s = self.store.lock().unwrap();
            if let Some(slot) = s.iter_mut().find(|(e, _)| e == id) {
                slot.1 = data.to_vec(); // overwrite (converge on source)
            } else {
                s.push((id.clone(), data.to_vec()));
            }
            Ok(())
        }
        fn list(&self, _name: Option<&str>, _limit: usize) -> anyhow::Result<Vec<ArtifactEntry>> {
            Ok(self
                .store
                .lock()
                .unwrap()
                .iter()
                .map(|(id, b)| ArtifactEntry {
                    id: id.clone(),
                    size_bytes: b.len() as i64,
                    content_type: "application/octet-stream".into(),
                })
                .collect())
        }
        fn archive_info(&self) -> anyhow::Result<ArchiveInfo> {
            Ok(ArchiveInfo::default())
        }
        fn handle_http2_request(
            &self,
            _m: &str,
            _s: &str,
            _b: &[u8],
        ) -> anyhow::Result<(u16, Vec<(String, String)>, Vec<u8>)> {
            Ok((404, vec![], Vec::new()))
        }
    }

    fn id(name: &str, ver: &str) -> ArtifactId {
        ArtifactId { namespace: None, name: name.into(), version: ver.into() }
    }

    /// A source seeded with three crates; a fresh writable destination.
    fn pair(dest_writable: bool) -> (Arc<dyn RepositoryBackendTrait>, Arc<dyn RepositoryBackendTrait>) {
        let src = MemRepo::seeded(
            "rust-dev",
            vec![
                (id("serde", "1.0.0"), b"serde-bytes".to_vec()),
                (id("serde", "1.0.1"), b"serde-newer".to_vec()),
                (id("tokio", "1.40.0"), b"tokio-bytes".to_vec()),
            ],
        )
        .arc();
        let dst = MemRepo::new("rust-mirror", dest_writable).arc();
        (src, dst)
    }

    fn one_pair(
        src: Arc<dyn RepositoryBackendTrait>,
        dst: Arc<dyn RepositoryBackendTrait>,
    ) -> Vec<RepoPair> {
        vec![RepoPair {
            source_name: "rust-dev".into(),
            source: src,
            dest_name: "rust-mirror".into(),
            dest: dst,
        }]
    }

    #[test]
    fn dry_run_reports_every_artifact_and_writes_nothing() {
        let (src, dst) = pair(true);
        let pairs = one_pair(src, dst.clone());
        let r = replicate_pairs(&pairs, &ReplicationOptions::preview());
        assert_eq!(r.would_copy(), 3, "all 3 source artifacts would copy");
        assert_eq!(r.copied(), 0, "a dry run copies nothing");
        // RED-when-broken: a dry run that actually wrote would leave the dest
        // non-empty.
        assert!(
            dst.list(None, 0).unwrap().is_empty(),
            "dry run must NOT write to the destination"
        );
    }

    #[test]
    fn execute_copies_every_source_artifact_byte_identical() {
        let (src, dst) = pair(true);
        let pairs = one_pair(src.clone(), dst.clone());
        let r = replicate_pairs(&pairs, &ReplicationOptions::executing());
        assert_eq!(r.copied(), 3, "all 3 copied");
        assert!(r.is_complete(), "no failures");

        // RED-when-broken CORE assertion: the destination holds EVERY source
        // artifact, byte-identical. A dropped/garbled copy fails right here.
        for entry in src.list(None, 0).unwrap() {
            let want = src.fetch(&entry.id).unwrap().expect("source has it");
            let got = dst
                .fetch(&entry.id)
                .unwrap()
                .unwrap_or_else(|| panic!("destination MISSING {:?} after replicate", entry.id));
            assert_eq!(got, want, "destination bytes must equal source bytes for {:?}", entry.id);
        }
    }

    #[test]
    fn rerun_is_a_no_op_everything_already_present_by_content_id() {
        let (src, dst) = pair(true);
        // First run copies everything.
        let first = replicate_pairs(&one_pair(src.clone(), dst.clone()), &ReplicationOptions::executing());
        assert_eq!(first.copied(), 3);

        // Second run over the unchanged source: NOTHING copies, all skipped.
        let second = replicate_pairs(&one_pair(src, dst), &ReplicationOptions::executing());
        assert_eq!(second.copied(), 0, "re-run must copy nothing (idempotent)");
        assert_eq!(second.skipped(), 3, "every artifact already present by content id");
        assert!(second.is_complete());
    }

    #[test]
    fn changed_content_is_recopied_not_skipped() {
        let (src, dst) = pair(true);
        // Pre-seed the destination coordinate with DIFFERENT bytes than the source.
        dst.put(&id("serde", "1.0.0"), b"stale-different-bytes").unwrap();

        let r = replicate_pairs(&one_pair(src.clone(), dst.clone()), &ReplicationOptions::executing());
        // serde 1.0.0 differs → recopied; the other two are new → copied.
        assert_eq!(r.copied(), 3, "the divergent coordinate is re-copied, not skipped");
        assert_eq!(r.skipped(), 0);
        // Destination converged onto the source content.
        assert_eq!(
            dst.fetch(&id("serde", "1.0.0")).unwrap().unwrap(),
            b"serde-bytes",
            "destination now matches source bytes"
        );
    }

    #[test]
    fn read_only_destination_fails_closed_never_silently_drops() {
        let (src, dst) = pair(false); // destination NOT writable
        let r = replicate_pairs(&one_pair(src, dst.clone()), &ReplicationOptions::executing());
        // Every artifact is accounted for as a FAILURE, not silently skipped.
        assert_eq!(r.failed(), 3, "a read-only destination fails every copy closed");
        assert_eq!(r.copied(), 0);
        assert!(!r.is_complete(), "failures ⇒ not complete");
        assert!(dst.list(None, 0).unwrap().is_empty(), "nothing written to a read-only dest");
    }

    #[test]
    fn partial_destination_copies_only_the_missing_remainder() {
        let (src, dst) = pair(true);
        // Destination already holds serde 1.0.0 identically (a prior partial run).
        dst.put(&id("serde", "1.0.0"), b"serde-bytes").unwrap();

        let r = replicate_pairs(&one_pair(src.clone(), dst.clone()), &ReplicationOptions::executing());
        assert_eq!(r.skipped(), 1, "the already-present artifact is skipped");
        assert_eq!(r.copied(), 2, "only the two missing artifacts copy");
        // Full mirror regardless of the partial starting point.
        for entry in src.list(None, 0).unwrap() {
            assert!(dst.fetch(&entry.id).unwrap().is_some(), "dest complete after replicate");
        }
    }

    #[test]
    fn limit_caps_examined_artifacts_and_flags_truncation() {
        let (src, dst) = pair(true);
        let r = replicate_pairs(
            &one_pair(src, dst),
            &ReplicationOptions { execute: true, limit: 2 },
        );
        assert_eq!(r.examined(), 2, "only two artifacts examined under the cap");
        assert!(r.truncated, "more source artifacts remained ⇒ truncated");
    }
}