holger-server-lib 0.6.9

//! Generic / **raw** writable backend: a disk-backed, path-addressed bucket for
//! arbitrary blobs — the Nexus/Artifactory "raw hosted repository" (the misc
//! bucket for tarballs, installers, reports, firmware, anything that has no
//! ecosystem protocol of its own). A client `PUT`s bytes at an arbitrary path
//! and `GET`s them back byte-identical; the path *is* the address.
//!
//! Deliberately the leanest writable backend: no index synthesis, no metadata
//! generation, no ecosystem parsing — just a safe, content-verifiable loose-file
//! store. Because it implements the full [`RepositoryBackendTrait`] it drops into
//! every existing seam unchanged: the checksum-deploy gate verifies uploads at
//! the HTTP door, cross-repo [`search`](crate::search) finds blobs by
//! name/path/checksum, the retention/GC executor reclaims them
//! ([`delete_artifact`](RawBackend::delete_artifact) is the SHA-256-verified safe
//! unlink, agreeing with `holger-server-lib::sha256_hex`), and
//! [`replication`](crate::replication) copies them content-id-idempotently.
//!
//! **Coordinate ↔ path bijection (the one rule both entry points obey).** The
//! gRPC / in-process side addresses a blob by an [`ArtifactId`]; the HTTP side
//! addresses it by a raw path. They meet at one deterministic mapping:
//!   - `id → relpath` = the `/`-join of the non-empty parts of
//!     `[namespace, name, version]`.
//!   - [`list`](RawBackend::list) reports every stored file as
//!     `ArtifactId { namespace: None, name: <relpath>, version: "" }`.
//! so `id_to_relpath(listed_id) == <relpath>` round-trips exactly — which is why
//! a listing fed back into `fetch`/`delete_artifact` (what retention, GC,
//! replication and search all do) resolves to the very same file.
//!
//! **Traversal-closed.** Every path segment is validated ([`safe_relpath`]): an
//! empty segment, `.`, `..`, a backslash or a NUL is refused, so no id and no
//! HTTP path can climb out of `base_dir`. (This mirrors the per-segment guard the
//! sibling loose-file backends keep privately — `sanitize_repo` in `promote`
//! collapses to a *single* segment, which would destroy the nested paths a raw
//! bucket exists to hold, so it is intentionally not reused here.)

use std::path::{Path, PathBuf};

use traits::{ArtifactEntry, ArtifactFormat, ArtifactId, RepositoryBackendTrait};

/// A writable, disk-backed, path-addressed blob store (the "raw hosted repo").
pub struct RawBackend {
    name: String,
    base_dir: PathBuf,
}

/// Lowercase-hex SHA-256 of `data`. The content id the retention/GC executor
/// verifies before an unlink — SHA-256 (not blake3) so it agrees with
/// `holger-server-lib::sha256_hex` and the other loose-file backends.
fn sha256_hex(data: &[u8]) -> String {
    // LAW #5 dedup: consume the shared `nornir-hash` leaf (edda) so this backend's
    // content id agrees with `holger-server-lib::sha256_hex` and the siblings.
    nornir_hash::sha256_hex(data)
}

/// Validate a `/`-delimited relative path and join it under `base_dir`.
///
/// Every segment must be a benign single directory level — a hostile segment
/// (empty, `.`, `..`, containing a backslash or a NUL byte) is refused, so the
/// resulting path can never escape `base_dir`. An empty path (no segments) is
/// itself invalid (a raw blob needs a name). Returns the joined absolute path.
fn safe_relpath(base_dir: &Path, relpath: &str) -> anyhow::Result<PathBuf> {
    let mut out = base_dir.to_path_buf();
    let mut any = false;
    for seg in relpath.split('/') {
        if seg.is_empty() || seg == "." || seg == ".." || seg.contains('\\') || seg.contains('\0') {
            anyhow::bail!("invalid raw path segment {seg:?} in {relpath:?}");
        }
        out.push(seg);
        any = true;
    }
    if !any {
        anyhow::bail!("empty raw artifact path");
    }
    Ok(out)
}

impl RawBackend {
    /// Build a raw repo named `name`, storing blobs under `base_dir`.
    pub fn new(name: String, base_dir: PathBuf) -> Self {
        Self { name, base_dir }
    }

    /// The one coordinate→path map: the `/`-join of the non-empty parts of
    /// `[namespace, name, version]`. A raw blob's address is its path, so a
    /// caller that leaves `namespace`/`version` empty simply addresses by `name`;
    /// a listing round-trips because `list` puts the whole relpath in `name`.
    fn id_to_relpath(id: &ArtifactId) -> String {
        let mut parts: Vec<&str> = Vec::with_capacity(3);
        if let Some(ns) = id.namespace.as_deref() {
            if !ns.is_empty() {
                parts.push(ns);
            }
        }
        if !id.name.is_empty() {
            parts.push(&id.name);
        }
        if !id.version.is_empty() {
            parts.push(&id.version);
        }
        parts.join("/")
    }

    /// Read the bytes stored at a validated relpath (`None` if absent).
    fn read_relpath(&self, relpath: &str) -> anyhow::Result<Option<Vec<u8>>> {
        let path = safe_relpath(&self.base_dir, relpath)?;
        match std::fs::read(&path) {
            Ok(bytes) => Ok(Some(bytes)),
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None),
            Err(e) => Err(e.into()),
        }
    }

    /// Atomically store `data` at a validated relpath (write-then-rename).
    fn write_relpath(&self, relpath: &str, data: &[u8]) -> anyhow::Result<()> {
        let path = safe_relpath(&self.base_dir, relpath)?;
        let parent = path
            .parent()
            .ok_or_else(|| anyhow::anyhow!("raw artifact path has no parent directory"))?;
        std::fs::create_dir_all(parent)?;
        // Unique temp name per concurrent write (PID + a process-wide counter) so
        // two simultaneous puts into one dir can't race the same `.tmp`.
        static TMP_SEQ: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
        let seq = TMP_SEQ.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        let tmp = parent.join(format!(".{}.{}.tmp", std::process::id(), seq));
        std::fs::write(&tmp, data)?;
        std::fs::rename(&tmp, &path)?;
        Ok(())
    }

    /// Unlink the file at a validated relpath. `Ok(true)` if a file was removed,
    /// `Ok(false)` if there was nothing there (operator DELETE is idempotent-ish:
    /// the caller maps absence to 404).
    fn remove_relpath(&self, relpath: &str) -> anyhow::Result<bool> {
        let path = safe_relpath(&self.base_dir, relpath)?;
        match std::fs::remove_file(&path) {
            Ok(()) => Ok(true),
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(false),
            Err(e) => Err(e.into()),
        }
    }

    /// The repo-name-stripped, query-stripped relpath of an HTTP suburl, or
    /// `None` if the first segment is not this repo (never our route).
    fn http_relpath<'a>(&self, suburl: &'a str) -> Option<String> {
        let path = suburl.split('?').next().unwrap_or(suburl);
        let mut segs = path.trim_start_matches('/').split('/').filter(|s| !s.is_empty());
        match segs.next() {
            Some(first) if first == self.name => {
                let rest: Vec<&str> = segs.collect();
                if rest.is_empty() {
                    None
                } else {
                    Some(rest.join("/"))
                }
            }
            _ => None,
        }
    }

    /// Recursively collect stored files as `(relpath, size)` under `base_dir`,
    /// capped at `cap`, optionally filtered to relpaths CONTAINING `filter`.
    fn walk(&self, filter: Option<&str>, cap: usize) -> anyhow::Result<Vec<(String, u64)>> {
        let mut out = Vec::new();
        // Stack of directories still to visit (relative-prefix, absolute-path).
        let mut stack: Vec<(String, PathBuf)> = vec![(String::new(), self.base_dir.clone())];
        while let Some((prefix, dir)) = stack.pop() {
            let iter = match std::fs::read_dir(&dir) {
                Ok(it) => it,
                // A missing base_dir (nothing stored yet) is an empty listing.
                Err(e) if e.kind() == std::io::ErrorKind::NotFound => continue,
                Err(e) => return Err(e.into()),
            };
            for entry in iter {
                let entry = entry?;
                let file_name = match entry.file_name().into_string() {
                    Ok(s) => s,
                    Err(_) => continue, // non-UTF-8 name: skip gracefully
                };
                // Skip in-flight `.<pid>.<seq>.tmp` staging files.
                if file_name.starts_with('.') && file_name.ends_with(".tmp") {
                    continue;
                }
                let rel = if prefix.is_empty() {
                    file_name.clone()
                } else {
                    format!("{prefix}/{file_name}")
                };
                let ft = entry.file_type()?;
                if ft.is_dir() {
                    stack.push((rel, entry.path()));
                } else if ft.is_file() {
                    if filter.map(|f| rel.contains(f)).unwrap_or(true) {
                        let size = entry.metadata().map(|m| m.len()).unwrap_or(0);
                        out.push((rel, size));
                        if out.len() >= cap {
                            return Ok(out);
                        }
                    }
                }
            }
        }
        Ok(out)
    }
}

impl RepositoryBackendTrait for RawBackend {
    fn name(&self) -> &str {
        &self.name
    }

    fn format(&self) -> ArtifactFormat {
        ArtifactFormat::Raw
    }

    fn is_writable(&self) -> bool {
        true
    }

    fn fetch(&self, id: &ArtifactId) -> anyhow::Result<Option<Vec<u8>>> {
        self.read_relpath(&Self::id_to_relpath(id))
    }

    fn put(&self, id: &ArtifactId, data: &[u8]) -> anyhow::Result<()> {
        self.write_relpath(&Self::id_to_relpath(id), data)
    }

    /// Reclaim one stored blob — the retention/GC executor's safe unlink. Read
    /// the on-disk bytes, recompute their SHA-256, and remove **only** if it
    /// matches `expected_content_id` (the digest the executor planned to drop); a
    /// mismatch means the store changed under the plan → refuse. A missing file
    /// is an error (nothing to reclaim). Mirrors the loose-file rust backend.
    fn delete_artifact(&self, id: &ArtifactId, expected_content_id: &str) -> anyhow::Result<()> {
        let relpath = Self::id_to_relpath(id);
        let bytes = match self.read_relpath(&relpath)? {
            Some(b) => b,
            None => anyhow::bail!("raw artifact {relpath:?} not present — nothing to reclaim"),
        };
        let actual = sha256_hex(&bytes);
        if actual != expected_content_id {
            anyhow::bail!(
                "refusing to delete raw {relpath:?}: on-disk content id {actual} != expected \
                 {expected_content_id} — the store changed under the retention plan"
            );
        }
        self.remove_relpath(&relpath)?;
        Ok(())
    }

    /// Every stored file, reported as `ArtifactId { namespace: None, name:
    /// <relpath>, version: "" }` so the id round-trips back to the same path.
    /// `name_filter` is a substring match on the relpath; `limit == 0` falls back
    /// to a sane cap (1024). A missing `base_dir` yields `[]`, not an error.
    fn list(&self, name_filter: Option<&str>, limit: usize) -> anyhow::Result<Vec<ArtifactEntry>> {
        const DEFAULT_CAP: usize = 1024;
        let cap = if limit == 0 { DEFAULT_CAP } else { limit };
        let files = self.walk(name_filter, cap)?;
        Ok(files
            .into_iter()
            .map(|(relpath, size)| ArtifactEntry {
                id: ArtifactId { namespace: None, name: relpath, version: String::new() },
                size_bytes: size as i64,
                content_type: "application/octet-stream".into(),
            })
            .collect())
    }

    /// Raw HTTP door: `GET`/`HEAD` reads, `PUT`/`POST` writes (the checksum-deploy
    /// gate has already verified the body at the gateway), `DELETE` removes.
    /// The path after the repo name is the blob address.
    fn handle_http2_request(
        &self,
        method: &str,
        suburl: &str,
        body: &[u8],
    ) -> anyhow::Result<(u16, Vec<(String, String)>, Vec<u8>)> {
        let relpath = match self.http_relpath(suburl) {
            Some(r) => r,
            None => return Ok((404, Vec::new(), b"Not found".to_vec())),
        };
        let octet = || vec![("Content-Type".to_string(), "application/octet-stream".to_string())];
        match method {
            "GET" | "HEAD" => match self.read_relpath(&relpath) {
                Ok(Some(bytes)) => {
                    let ok = true;
                    crate::grpc::functional_status("holger-raw/http", "raw_get", ok, &relpath);
                    let out = if method == "HEAD" { Vec::new() } else { bytes };
                    Ok((200, octet(), out))
                }
                Ok(None) => {
                    crate::grpc::functional_status("holger-raw/http", "raw_get_miss", true, &relpath);
                    Ok((404, Vec::new(), b"Not found".to_vec()))
                }
                Err(e) => Err(e),
            },
            "PUT" | "POST" => {
                self.write_relpath(&relpath, body)?;
                crate::grpc::functional_status("holger-raw/http", "raw_put", true, &relpath);
                Ok((201, octet(), Vec::new()))
            }
            "DELETE" => {
                let removed = self.remove_relpath(&relpath)?;
                crate::grpc::functional_status("holger-raw/http", "raw_delete", removed, &relpath);
                if removed {
                    Ok((204, Vec::new(), Vec::new()))
                } else {
                    Ok((404, Vec::new(), b"Not found".to_vec()))
                }
            }
            other => Ok((
                405,
                vec![("Allow".into(), "GET, HEAD, PUT, POST, DELETE".into())],
                format!("method {other} not allowed on a raw repository").into_bytes(),
            )),
        }
    }

    /// A raw repo's HTTP path IS its coordinate: `/<repo>/<relpath>` maps back to
    /// `ArtifactId { namespace: None, name: <relpath>, version: "" }` — exactly the
    /// coordinate [`list`](RawBackend::list) reports and [`fetch`](RawBackend::fetch)
    /// resolves. This lets the serve-time quarantine gate look up a raw blob's
    /// properties before serving it over HTTP.
    fn coordinate_for_path(&self, suburl: &str) -> Option<ArtifactId> {
        self.http_relpath(suburl).map(|relpath| ArtifactId {
            namespace: None,
            name: relpath,
            version: String::new(),
        })
    }
}

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

    fn tmpdir() -> PathBuf {
        let p = std::env::temp_dir().join(format!(
            "holger-raw-test-{}-{}",
            std::process::id(),
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap()
                .as_nanos()
        ));
        std::fs::create_dir_all(&p).unwrap();
        p
    }

    fn id(ns: Option<&str>, name: &str, version: &str) -> ArtifactId {
        ArtifactId {
            namespace: ns.map(|s| s.to_string()),
            name: name.into(),
            version: version.into(),
        }
    }

    #[test]
    fn put_then_fetch_round_trip() {
        let dir = tmpdir();
        let repo = RawBackend::new("raw".into(), dir.clone());
        let coord = id(Some("firmware"), "device.bin", "1.2.0");
        repo.put(&coord, b"\x00binary payload\xff").unwrap();
        // Byte-identical read-back.
        assert_eq!(
            repo.fetch(&coord).unwrap().as_deref(),
            Some(b"\x00binary payload\xff".as_slice())
        );
        // The path is the join of the non-empty coordinate parts.
        assert!(dir.join("firmware").join("device.bin").join("1.2.0").is_file());
        std::fs::remove_dir_all(&dir).ok();
    }

    #[test]
    fn fetch_missing_returns_none() {
        let dir = tmpdir();
        let repo = RawBackend::new("raw".into(), dir.clone());
        assert_eq!(repo.fetch(&id(None, "ghost", "")).unwrap(), None);
        std::fs::remove_dir_all(&dir).ok();
    }

    #[test]
    fn put_and_fetch_reject_path_traversal() {
        let dir = tmpdir();
        let repo = RawBackend::new("raw".into(), dir.clone());
        // A `..` segment must be refused on BOTH the write and read paths — never
        // silently written outside base_dir, never read from outside it.
        let hostile = id(Some(".."), "etc", "passwd");
        assert!(repo.put(&hostile, b"pwn").is_err(), "traversal put must be refused");
        assert!(repo.fetch(&hostile).is_err(), "traversal fetch must be refused");
        // An empty coordinate (no segments) is likewise rejected.
        assert!(repo.put(&id(None, "", ""), b"x").is_err(), "empty path refused");
        std::fs::remove_dir_all(&dir).ok();
    }

    #[test]
    fn http_put_get_delete_round_trip() {
        let dir = tmpdir();
        let repo = RawBackend::new("raw".into(), dir.clone());
        // PUT stores.
        let (s, _h, _b) = repo
            .handle_http2_request("PUT", "/raw/reports/2024/q1.pdf", b"PDF-BYTES")
            .unwrap();
        assert_eq!(s, 201, "PUT creates");
        // GET reads it back byte-identical.
        let (s, _h, body) = repo
            .handle_http2_request("GET", "/raw/reports/2024/q1.pdf", &[])
            .unwrap();
        assert_eq!(s, 200);
        assert_eq!(body, b"PDF-BYTES");
        // HEAD is 200 with no body.
        let (s, _h, body) = repo
            .handle_http2_request("HEAD", "/raw/reports/2024/q1.pdf", &[])
            .unwrap();
        assert_eq!(s, 200);
        assert!(body.is_empty(), "HEAD carries no body");
        // DELETE removes it; a second GET is 404.
        let (s, _h, _b) = repo
            .handle_http2_request("DELETE", "/raw/reports/2024/q1.pdf", &[])
            .unwrap();
        assert_eq!(s, 204);
        let (s, _h, _b) = repo
            .handle_http2_request("GET", "/raw/reports/2024/q1.pdf", &[])
            .unwrap();
        assert_eq!(s, 404, "gone after delete");
        std::fs::remove_dir_all(&dir).ok();
    }

    #[test]
    fn http_get_missing_is_404_and_wrong_repo_is_404() {
        let dir = tmpdir();
        let repo = RawBackend::new("raw".into(), dir.clone());
        let (s, _h, _b) = repo.handle_http2_request("GET", "/raw/nope.bin", &[]).unwrap();
        assert_eq!(s, 404, "absent blob → 404");
        // A path whose first segment is not this repo never resolves here.
        let (s, _h, _b) = repo
            .handle_http2_request("GET", "/other-repo/x.bin", &[])
            .unwrap();
        assert_eq!(s, 404, "not our route → 404");
        std::fs::remove_dir_all(&dir).ok();
    }

    #[test]
    fn http_rejects_traversal_in_path() {
        let dir = tmpdir();
        let repo = RawBackend::new("raw".into(), dir.clone());
        // `..` in the HTTP path must error out of safe_relpath, not escape.
        assert!(
            repo.handle_http2_request("PUT", "/raw/../escape.bin", b"pwn").is_err(),
            "HTTP traversal must be refused"
        );
        std::fs::remove_dir_all(&dir).ok();
    }

    #[test]
    fn list_enumerates_nested_and_round_trips_to_fetch() {
        let dir = tmpdir();
        let repo = RawBackend::new("raw".into(), dir.clone());
        repo.put(&id(Some("a/b"), "one.bin", ""), b"111").unwrap();
        repo.put(&id(None, "two.bin", ""), b"22").unwrap();
        repo.put(&id(Some("a"), "three.txt", ""), b"3333").unwrap();
        let listed = repo.list(None, 0).unwrap();
        assert_eq!(listed.len(), 3, "all three nested blobs enumerated");
        // Every listed id round-trips: fetch by the listed coordinate resolves to
        // the same bytes (this is what retention/search/replication rely on).
        for e in &listed {
            let bytes = repo.fetch(&e.id).unwrap();
            assert!(bytes.is_some(), "listed id {:?} must re-fetch", e.id.name);
            assert_eq!(bytes.unwrap().len() as i64, e.size_bytes, "size matches");
        }
        // The nested blob's relpath is carried whole in `name`.
        assert!(
            listed.iter().any(|e| e.id.name == "a/b/one.bin"),
            "nested relpath preserved in the listed coordinate"
        );
        std::fs::remove_dir_all(&dir).ok();
    }

    #[test]
    fn list_filters_by_substring_and_respects_limit() {
        let dir = tmpdir();
        let repo = RawBackend::new("raw".into(), dir.clone());
        repo.put(&id(None, "alpha.bin", ""), b"a").unwrap();
        repo.put(&id(None, "beta.bin", ""), b"b").unwrap();
        repo.put(&id(None, "alpine.txt", ""), b"c").unwrap();
        // Substring "alp" matches alpha + alpine, not beta.
        let matched = repo.list(Some("alp"), 0).unwrap();
        assert_eq!(matched.len(), 2, "substring filter selects alpha + alpine");
        assert!(matched.iter().all(|e| e.id.name.contains("alp")));
        // limit caps the count.
        let capped = repo.list(None, 1).unwrap();
        assert_eq!(capped.len(), 1, "limit=1 returns exactly one");
        std::fs::remove_dir_all(&dir).ok();
    }

    #[test]
    fn delete_artifact_verifies_content_id_before_unlink() {
        let dir = tmpdir();
        let repo = RawBackend::new("raw".into(), dir.clone());
        let coord = id(None, "keep.bin", "");
        repo.put(&coord, b"real-bytes").unwrap();
        let good = sha256_hex(b"real-bytes");
        // RED-when-broken: a WRONG content id must NOT unlink (safe GC contract).
        let wrong = sha256_hex(b"different");
        assert!(
            repo.delete_artifact(&coord, &wrong).is_err(),
            "mismatched content id must refuse to delete"
        );
        assert!(repo.fetch(&coord).unwrap().is_some(), "blob survives a refused delete");
        // The correct content id unlinks.
        repo.delete_artifact(&coord, &good).unwrap();
        assert_eq!(repo.fetch(&coord).unwrap(), None, "blob gone after verified delete");
        // Deleting an absent blob is an error (nothing to reclaim).
        assert!(repo.delete_artifact(&coord, &good).is_err(), "absent → error");
        std::fs::remove_dir_all(&dir).ok();
    }

    #[test]
    fn writable_and_raw_format() {
        let dir = tmpdir();
        let repo = RawBackend::new("raw".into(), dir.clone());
        assert!(repo.is_writable(), "raw is a writable hosted repo");
        assert_eq!(repo.format(), ArtifactFormat::Raw);
        assert_eq!(repo.name(), "raw");
        std::fs::remove_dir_all(&dir).ok();
    }

    #[test]
    fn method_not_allowed_for_unknown_verb() {
        let dir = tmpdir();
        let repo = RawBackend::new("raw".into(), dir.clone());
        let (s, headers, _b) = repo
            .handle_http2_request("PATCH", "/raw/x.bin", &[])
            .unwrap();
        assert_eq!(s, 405, "PATCH is not a raw verb");
        assert!(
            headers.iter().any(|(k, _)| k == "Allow"),
            "405 advertises the allowed verbs"
        );
        std::fs::remove_dir_all(&dir).ok();
    }
}