//! Chunk Store - Persistent storage for erasure-coded shards
//!
//! Uses the filesystem (or RocksDB when 'node' feature is enabled)
//! to store shards organized by deal ID.

use crate::erasure::Shard;
use crate::error::{Error, Result};
use std::path::{Path, PathBuf};
use tokio::fs;
use tokio::io::{AsyncReadExt, AsyncWriteExt};

/// Persistent storage for shards
pub struct ChunkStore {
    /// Base directory for storage
    base_dir: PathBuf,
    /// Bytes currently used
    used_bytes: std::sync::atomic::AtomicU64,
}

impl ChunkStore {
    /// Create a new chunk store
    pub fn new(base_dir: &str) -> Result<Self> {
        let path = PathBuf::from(base_dir);

        // Create directory synchronously for initialization
        std::fs::create_dir_all(&path)?;

        // Calculate used capacity from existing files
        let used = Self::calculate_used_capacity(&path);

        Ok(Self {
            base_dir: path,
            used_bytes: std::sync::atomic::AtomicU64::new(used),
        })
    }

    fn calculate_used_capacity(path: &Path) -> u64 {
        let mut total = 0u64;
        if let Ok(entries) = std::fs::read_dir(path) {
            for entry in entries.flatten() {
                if let Ok(metadata) = entry.metadata() {
                    if metadata.is_file() {
                        total += metadata.len();
                    } else if metadata.is_dir() {
                        total += Self::calculate_used_capacity(&entry.path());
                    }
                }
            }
        }
        total
    }

    /// Get bytes currently used
    pub fn used_capacity(&self) -> u64 {
        self.used_bytes.load(std::sync::atomic::Ordering::Relaxed)
    }

    /// Store a shard
    pub async fn store_shard(&self, deal_id: &[u8; 32], shard: &Shard) -> Result<()> {
        // Create deal directory
        let deal_dir = self.deal_dir(deal_id);
        fs::create_dir_all(&deal_dir).await?;

        // Shard file path
        let shard_path = deal_dir.join(format!("shard_{:03}.bin", shard.index));

        // Serialize shard
        let encoded = serde_json::to_vec(shard)?;

        // Write atomically (write to temp, then rename)
        let temp_path = deal_dir.join(format!("shard_{:03}.tmp", shard.index));
        let mut file = fs::File::create(&temp_path).await?;
        file.write_all(&encoded).await?;
        file.sync_all().await?;
        fs::rename(&temp_path, &shard_path).await?;

        // Update used capacity
        self.used_bytes.fetch_add(
            encoded.len() as u64,
            std::sync::atomic::Ordering::Relaxed,
        );

        Ok(())
    }

    /// Retrieve a shard
    pub async fn get_shard(&self, deal_id: &[u8; 32], shard_index: u8) -> Result<Shard> {
        let shard_path = self.deal_dir(deal_id).join(format!("shard_{:03}.bin", shard_index));

        if !shard_path.exists() {
            return Err(Error::NotFound(format!(
                "Shard {} not found for deal",
                shard_index
            )));
        }

        let mut file = fs::File::open(&shard_path).await?;
        let mut data = Vec::new();
        file.read_to_end(&mut data).await?;

        let shard: Shard = serde_json::from_slice(&data)?;
        Ok(shard)
    }

    /// Delete all shards for a deal
    pub async fn delete_deal(&self, deal_id: &[u8; 32]) -> Result<()> {
        let deal_dir = self.deal_dir(deal_id);

        if deal_dir.exists() {
            // Calculate size before deletion
            let size = Self::calculate_used_capacity(&deal_dir);

            // Remove directory
            fs::remove_dir_all(&deal_dir).await?;

            // Update used capacity
            self.used_bytes.fetch_sub(size, std::sync::atomic::Ordering::Relaxed);
        }

        Ok(())
    }

    /// List all shard indices for a deal
    pub async fn list_shards(&self, deal_id: &[u8; 32]) -> Result<Vec<u8>> {
        let deal_dir = self.deal_dir(deal_id);

        if !deal_dir.exists() {
            return Ok(Vec::new());
        }

        let mut indices = Vec::new();
        let mut entries = fs::read_dir(&deal_dir).await?;

        while let Some(entry) = entries.next_entry().await? {
            let name = entry.file_name();
            let name_str = name.to_string_lossy();

            if name_str.starts_with("shard_") && name_str.ends_with(".bin") {
                // Parse index from "shard_XXX.bin"
                if let Some(idx_str) = name_str.strip_prefix("shard_").and_then(|s| s.strip_suffix(".bin")) {
                    if let Ok(idx) = idx_str.parse::<u8>() {
                        indices.push(idx);
                    }
                }
            }
        }

        indices.sort();
        Ok(indices)
    }

    /// Check if a deal exists
    pub async fn has_deal(&self, deal_id: &[u8; 32]) -> bool {
        self.deal_dir(deal_id).exists()
    }

    /// Get the directory for a deal
    fn deal_dir(&self, deal_id: &[u8; 32]) -> PathBuf {
        // Use first 4 bytes as subdirectory for sharding
        let subdir = hex::encode(&deal_id[..2]);
        let deal_hex = hex::encode(deal_id);
        self.base_dir.join(subdir).join(deal_hex)
    }

    /// Verify shard integrity
    pub async fn verify_shard(&self, deal_id: &[u8; 32], shard_index: u8) -> Result<bool> {
        let shard = self.get_shard(deal_id, shard_index).await?;

        // Verify hash
        let computed_hash = *blake3::hash(&shard.data).as_bytes();
        Ok(computed_hash == shard.hash)
    }

    /// Get storage statistics
    pub fn stats(&self) -> StoreStats {
        StoreStats {
            used_bytes: self.used_capacity(),
            base_dir: self.base_dir.to_string_lossy().to_string(),
        }
    }
}

/// Storage statistics
#[derive(Debug, Clone)]
pub struct StoreStats {
    /// Bytes used
    pub used_bytes: u64,
    /// Base directory path
    pub base_dir: String,
}

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

    #[tokio::test]
    async fn test_store_and_retrieve_shard() {
        let temp_dir = tempfile::tempdir().unwrap();
        let store = ChunkStore::new(temp_dir.path().to_str().unwrap()).unwrap();

        let deal_id = [1u8; 32];
        let shard = Shard {
            index: 0,
            is_data: true,
            data: vec![1, 2, 3, 4, 5],
            hash: *blake3::hash(&[1, 2, 3, 4, 5]).as_bytes(),
        };

        // Store
        store.store_shard(&deal_id, &shard).await.unwrap();

        // Retrieve
        let retrieved = store.get_shard(&deal_id, 0).await.unwrap();
        assert_eq!(retrieved.data, shard.data);
        assert_eq!(retrieved.hash, shard.hash);
    }

    #[tokio::test]
    async fn test_list_shards() {
        let temp_dir = tempfile::tempdir().unwrap();
        let store = ChunkStore::new(temp_dir.path().to_str().unwrap()).unwrap();

        let deal_id = [2u8; 32];

        for i in 0u8..5 {
            let shard = Shard {
                index: i as usize,
                is_data: i < 3,
                data: vec![i; 10],
                hash: *blake3::hash(&vec![i; 10]).as_bytes(),
            };
            store.store_shard(&deal_id, &shard).await.unwrap();
        }

        let indices = store.list_shards(&deal_id).await.unwrap();
        assert_eq!(indices, vec![0, 1, 2, 3, 4]);
    }
}
