//! Content-addressed blob storage on the local filesystem.

use std::fmt;
use std::fs;
use std::io::Write as _;
use std::path::PathBuf;

use serde::{Deserialize, Serialize};
use tempfile::NamedTempFile;

use crate::backend::BlobBackend;
use crate::error::{Error, Result};

/// 32-byte blake3 content address of a stored blob.
#[derive(Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub struct BlobId(pub [u8; 32]);

impl BlobId {
    /// Compute the address of arbitrary bytes.
    pub fn from_bytes(bytes: &[u8]) -> Self {
        Self(*blake3::hash(bytes).as_bytes())
    }

    /// Lowercase hex representation (64 chars).
    pub fn to_hex(&self) -> String {
        let mut s = String::with_capacity(64);
        for byte in self.0 {
            use std::fmt::Write as _;
            let _ = write!(s, "{byte:02x}");
        }
        s
    }
}

impl fmt::Debug for BlobId {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "BlobId({})", self.to_hex())
    }
}

/// Filesystem-backed content-addressed blob store — the default
/// [`crate::BlobBackend`] impl.
///
/// Layout: `<root>/<aa>/<bb>/<full-hex>` where `aa`/`bb` are the first two
/// pairs of hex bytes of the blob id. The two-level fan-out keeps any single
/// directory from accumulating an unbounded number of entries.
///
/// Previously exported as `BlobStore`; the legacy alias is preserved in
/// `lib.rs` for backwards compatibility.
pub struct LocalFilesystemBlobs {
    root: PathBuf,
}

impl LocalFilesystemBlobs {
    /// Open or create a store at `root`. Creates the directory if absent.
    pub fn open(root: impl Into<PathBuf>) -> Result<Self> {
        let root = root.into();
        fs::create_dir_all(&root)?;
        Ok(Self { root })
    }

    /// Path at which the blob with `id` is (or would be) stored.
    pub fn path_for(&self, id: &BlobId) -> PathBuf {
        let hex = id.to_hex();
        let (aa, rest) = hex.split_at(2);
        let (bb, _) = rest.split_at(2);
        self.root.join(aa).join(bb).join(&hex)
    }

    /// Insert bytes into the store. Returns the content address. Idempotent:
    /// re-putting the same bytes is a no-op (returns the existing id without
    /// rewriting the file).
    pub fn put(&self, bytes: &[u8]) -> Result<BlobId> {
        let id = BlobId::from_bytes(bytes);
        let final_path = self.path_for(&id);
        if final_path.exists() {
            return Ok(id);
        }
        let parent = final_path
            .parent()
            .expect("path_for always produces a parent");
        fs::create_dir_all(parent)?;
        let mut tmp = NamedTempFile::new_in(parent)?;
        tmp.write_all(bytes)?;
        tmp.as_file().sync_all()?;
        // POSIX rename is atomic and replaces. If another writer raced us
        // and persisted the same blob first, the rename still succeeds and
        // the result is identical bytes.
        tmp.persist(&final_path)
            .map_err(|err| Error::Persist(err.to_string()))?;
        Ok(id)
    }

    /// Retrieve bytes by content address. Returns `NotFound` if absent;
    /// `Corrupt` if the on-disk bytes do not hash back to `id`.
    pub fn get(&self, id: &BlobId) -> Result<Vec<u8>> {
        let path = self.path_for(id);
        let bytes = match fs::read(&path) {
            Ok(bytes) => bytes,
            Err(err) if err.kind() == std::io::ErrorKind::NotFound => {
                return Err(Error::NotFound(*id));
            }
            Err(err) => return Err(Error::Io(err)),
        };
        let actual = BlobId::from_bytes(&bytes);
        if actual != *id {
            return Err(Error::Corrupt {
                expected: *id,
                actual,
            });
        }
        Ok(bytes)
    }

    /// Whether the blob with `id` exists on disk. Cheap — does NOT read
    /// the bytes back and does NOT verify integrity.
    pub fn exists(&self, id: &BlobId) -> bool {
        self.path_for(id).exists()
    }

    /// List every stored blob id whose hex representation starts with
    /// `prefix_hex`. An empty prefix lists every stored blob.
    pub fn list_prefix(&self, prefix_hex: &str) -> Result<Vec<BlobId>> {
        // Reject prefixes containing non-hex characters. The on-disk
        // layout only uses lowercase hex, so an uppercase byte never
        // matches and treating it as "no match" is correct.
        if !prefix_hex
            .bytes()
            .all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b))
        {
            return Ok(Vec::new());
        }

        let mut out: Vec<BlobId> = Vec::new();
        let aa_iter = match fs::read_dir(&self.root) {
            Ok(iter) => iter,
            Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(out),
            Err(err) => return Err(Error::Io(err)),
        };
        for aa_entry in aa_iter.flatten() {
            if aa_entry.path().is_dir() {
                collect_bb(&aa_entry.path(), prefix_hex, &mut out);
            }
        }
        Ok(out)
    }
}

fn collect_bb(aa_dir: &std::path::Path, prefix_hex: &str, out: &mut Vec<BlobId>) {
    let Ok(iter) = fs::read_dir(aa_dir) else {
        return;
    };
    for bb_entry in iter.flatten() {
        if bb_entry.path().is_dir() {
            collect_blobs(&bb_entry.path(), prefix_hex, out);
        }
    }
}

fn collect_blobs(bb_dir: &std::path::Path, prefix_hex: &str, out: &mut Vec<BlobId>) {
    let Ok(iter) = fs::read_dir(bb_dir) else {
        return;
    };
    for blob_entry in iter.flatten() {
        let Some(name) = blob_entry.file_name().into_string().ok() else {
            continue;
        };
        if name.starts_with(prefix_hex) && name.len() == 64 {
            if let Some(id) = parse_blob_id_hex(&name) {
                out.push(id);
            }
        }
    }
}

fn parse_blob_id_hex(hex: &str) -> Option<BlobId> {
    if hex.len() != 64 {
        return None;
    }
    let mut bytes = [0u8; 32];
    for i in 0..32 {
        let byte = u8::from_str_radix(&hex[i * 2..i * 2 + 2], 16).ok()?;
        bytes[i] = byte;
    }
    Some(BlobId(bytes))
}

impl BlobBackend for LocalFilesystemBlobs {
    fn put(&self, bytes: &[u8]) -> Result<BlobId> {
        LocalFilesystemBlobs::put(self, bytes)
    }

    fn get(&self, id: &BlobId) -> Result<Vec<u8>> {
        LocalFilesystemBlobs::get(self, id)
    }

    fn exists(&self, id: &BlobId) -> bool {
        LocalFilesystemBlobs::exists(self, id)
    }

    fn list_prefix(&self, prefix_hex: &str) -> Result<Vec<BlobId>> {
        LocalFilesystemBlobs::list_prefix(self, prefix_hex)
    }
}
