use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};

/// A blob identifier — BLAKE3 hash of the blob data
pub type BlobId = [u8; 32];

/// Compute the content address of a blob (BLAKE3 hash).
pub fn compute_blob_id(data: &[u8]) -> BlobId {
    *blake3::hash(data).as_bytes()
}

/// Verify that blob data matches a claimed CID.
pub fn verify_blob(cid: &BlobId, data: &[u8]) -> bool {
    &compute_blob_id(data) == cid
}

/// Filesystem-based blob store with 256-shard directory layout.
/// Blobs are stored at `{base_dir}/{hex[0..2]}/{hex}`.
pub struct BlobStore {
    base_dir: PathBuf,
    /// CDN delivery budget: bytes remaining we're willing to serve this hour.
    /// Shared between Node (for resets) and ConnectionManager (for enforcement).
    delivery_budget_remaining: AtomicU64,
    /// Hourly delivery limit for resets
    delivery_limit: AtomicU64,
    /// Last delivery budget reset timestamp (ms)
    delivery_last_reset_ms: AtomicU64,
}

impl BlobStore {
    /// Open or create a blob store at `{data_dir}/blobs/`.
    pub fn open(data_dir: &Path) -> anyhow::Result<Self> {
        let base_dir = data_dir.join("blobs");
        std::fs::create_dir_all(&base_dir)?;
        Ok(Self {
            base_dir,
            delivery_budget_remaining: AtomicU64::new(u64::MAX), // unlimited until set
            delivery_limit: AtomicU64::new(u64::MAX),
            delivery_last_reset_ms: AtomicU64::new(0),
        })
    }

    /// Initialize delivery budget from device role (called once by Node after construction).
    pub fn set_delivery_budget(&self, limit: u64) {
        let now = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap_or_default()
            .as_millis() as u64;
        self.delivery_limit.store(limit, Ordering::Relaxed);
        self.delivery_budget_remaining.store(limit, Ordering::Relaxed);
        self.delivery_last_reset_ms.store(now, Ordering::Relaxed);
    }

    /// Check and consume delivery budget. Returns true if within budget.
    /// Auto-resets if an hour has elapsed.
    pub fn consume_delivery_budget(&self, bytes: u64) -> bool {
        // Auto-reset check
        let now = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap_or_default()
            .as_millis() as u64;
        let last = self.delivery_last_reset_ms.load(Ordering::Relaxed);
        if now.saturating_sub(last) >= 3_600_000 {
            let limit = self.delivery_limit.load(Ordering::Relaxed);
            self.delivery_budget_remaining.store(limit, Ordering::Relaxed);
            self.delivery_last_reset_ms.store(now, Ordering::Relaxed);
        }

        let prev = self.delivery_budget_remaining.fetch_update(
            Ordering::Relaxed,
            Ordering::Relaxed,
            |current| {
                if current >= bytes { Some(current - bytes) } else { None }
            },
        );
        prev.is_ok()
    }

    /// Get remaining delivery budget bytes.
    pub fn delivery_budget_remaining(&self) -> u64 {
        self.delivery_budget_remaining.load(Ordering::Relaxed)
    }

    fn blob_path(&self, cid: &BlobId) -> PathBuf {
        let hex = hex::encode(cid);
        let shard = &hex[..2];
        self.base_dir.join(shard).join(&hex)
    }

    /// Store a blob. Returns false if it already exists (idempotent).
    pub fn store(&self, cid: &BlobId, data: &[u8]) -> anyhow::Result<bool> {
        let path = self.blob_path(cid);
        if path.exists() {
            return Ok(false);
        }
        if let Some(parent) = path.parent() {
            std::fs::create_dir_all(parent)?;
        }
        std::fs::write(&path, data)?;
        Ok(true)
    }

    /// Retrieve a blob by CID.
    pub fn get(&self, cid: &BlobId) -> anyhow::Result<Option<Vec<u8>>> {
        let path = self.blob_path(cid);
        if !path.exists() {
            return Ok(None);
        }
        Ok(Some(std::fs::read(&path)?))
    }

    /// Check if a blob exists locally.
    pub fn has(&self, cid: &BlobId) -> bool {
        self.blob_path(cid).exists()
    }

    /// Return the filesystem path if the blob exists locally (for streaming).
    pub fn file_path(&self, cid: &BlobId) -> Option<PathBuf> {
        let path = self.blob_path(cid);
        if path.exists() { Some(path) } else { None }
    }

    /// Delete a blob (for future eviction).
    pub fn delete(&self, cid: &BlobId) -> anyhow::Result<bool> {
        let path = self.blob_path(cid);
        if !path.exists() {
            return Ok(false);
        }
        std::fs::remove_file(&path)?;
        Ok(true)
    }
}

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

    #[test]
    fn compute_verify_roundtrip() {
        let data = b"hello world blob data";
        let cid = compute_blob_id(data);
        assert!(verify_blob(&cid, data));
        assert!(!verify_blob(&[0u8; 32], data));
    }

    #[test]
    fn store_retrieve() {
        let dir = tempfile::tempdir().unwrap();
        let store = BlobStore::open(dir.path()).unwrap();

        let data = b"test blob content";
        let cid = compute_blob_id(data);

        // Store
        assert!(store.store(&cid, data).unwrap()); // first time → true
        assert!(!store.store(&cid, data).unwrap()); // idempotent → false

        // Has
        assert!(store.has(&cid));
        assert!(!store.has(&[0u8; 32]));

        // Get
        let retrieved = store.get(&cid).unwrap().unwrap();
        assert_eq!(retrieved, data);

        // Missing
        assert!(store.get(&[0u8; 32]).unwrap().is_none());
    }

    #[test]
    fn delete_blob() {
        let dir = tempfile::tempdir().unwrap();
        let store = BlobStore::open(dir.path()).unwrap();

        let data = b"to be deleted";
        let cid = compute_blob_id(data);
        store.store(&cid, data).unwrap();

        assert!(store.delete(&cid).unwrap());
        assert!(!store.has(&cid));
        assert!(!store.delete(&cid).unwrap()); // already gone
    }
}
