graph_d 1.3.2

//! Storage layer for the graph database.
//!
//! Satisfies: RT-1 (Data MUST survive crashes and restarts)
//! Satisfies: T3 (Persistence must survive crash and recover)
//!
//! # Modules
//!
//! - [`checkpoint`] - Checkpoint system for efficient crash recovery (Phase B)
//! - [`mmap`] - Memory-mapped file storage backend
//! - [`wal`] - Write-Ahead Log for durability guarantees

pub mod checkpoint;
pub mod mmap;
pub mod wal;

// INT-2: Re-export Checkpoint types for public API (Satisfies: API completeness)
pub use checkpoint::{Checkpoint, CheckpointConfig, CheckpointManager, CheckpointStats};

// INT-5: Re-export WAL types for public API (Satisfies: API completeness)
#[cfg(feature = "wal")]
pub use wal::{SyncMode, WalConfig, WalEntry, WalOperation, WriteAheadLog};

use crate::error::{GraphError, Result};
use crate::graph::{Id, Node, Relationship};
use crate::index::IndexManager;
use serde_json::Value;
use std::collections::HashMap;
use std::path::Path;

#[cfg(feature = "wal")]
use parking_lot::Mutex;
#[cfg(feature = "wal")]
use std::path::PathBuf;
#[cfg(feature = "wal")]
use std::sync::Arc;
// Note: WalConfig, WalOperation, WriteAheadLog, CheckpointConfig, CheckpointManager
// are available via pub use re-exports above (lines 17, 21)

/// Storage backend types.
pub enum StorageBackend {
    /// In-memory storage for testing and small datasets
    InMemory(InMemoryStorage),
    /// Memory-mapped file storage for persistence
    MmapFile(mmap::MmapStorage),
}

/// Storage backend for the graph database.
/// Satisfies: RT-1 (crash recovery through WAL integration)
/// Satisfies: TN1 (hybrid WAL + checkpoint strategy)
pub struct Storage {
    backend: StorageBackend,
    /// Index manager for fast property and relationship lookups
    index_manager: IndexManager,
    /// Write-ahead log for crash recovery (optional, enabled with WAL feature)
    #[cfg(feature = "wal")]
    wal: Option<Arc<Mutex<WriteAheadLog>>>,
    /// Storage path for WAL/checkpoint file location
    #[cfg(feature = "wal")]
    #[allow(dead_code)] // Used for checkpoint file path resolution
    storage_path: Option<PathBuf>,
    /// INT-1: Checkpoint manager for efficient recovery (Phase B)
    /// Satisfies: RT-1, TN1 (hybrid WAL + checkpoint)
    #[cfg(feature = "wal")]
    checkpoint_manager: Option<CheckpointManager>,
}

/// In-memory storage implementation.
pub struct InMemoryStorage {
    nodes: HashMap<Id, Node>,
    relationships: HashMap<Id, Relationship>,
    node_relationships: HashMap<Id, Vec<Id>>, // node_id -> relationship_ids
}

impl Storage {
    /// Create a new in-memory storage.
    pub fn new() -> Result<Self> {
        Ok(Storage {
            backend: StorageBackend::InMemory(InMemoryStorage {
                nodes: HashMap::new(),
                relationships: HashMap::new(),
                node_relationships: HashMap::new(),
            }),
            index_manager: IndexManager::new(),
            #[cfg(feature = "wal")]
            wal: None,
            #[cfg(feature = "wal")]
            storage_path: None,
            #[cfg(feature = "wal")]
            checkpoint_manager: None,
        })
    }

    /// Open persistent storage at the given path using memory-mapped files.
    /// Satisfies: RT-1 (crash recovery on open)
    pub fn open<P: AsRef<Path>>(path: P) -> Result<Self> {
        let path_ref = path.as_ref();
        let mmap_storage = mmap::MmapStorage::open(path_ref)?;

        #[cfg(feature = "wal")]
        let wal_path = path_ref.with_extension("wal");

        let mut storage = Storage {
            backend: StorageBackend::MmapFile(mmap_storage),
            index_manager: IndexManager::new(),
            #[cfg(feature = "wal")]
            wal: None,
            #[cfg(feature = "wal")]
            storage_path: Some(path_ref.to_path_buf()),
            #[cfg(feature = "wal")]
            checkpoint_manager: None,
        };

        // Initialize WAL if feature enabled
        #[cfg(feature = "wal")]
        {
            let wal = WriteAheadLog::open(&wal_path)?;
            storage.wal = Some(Arc::new(Mutex::new(wal)));

            // Recover from WAL on startup
            storage.recover_from_wal()?;
        }

        // Rebuild indexes from existing data
        storage.rebuild_indexes()?;

        Ok(storage)
    }

    /// Open storage with explicit WAL configuration.
    /// Satisfies: RT-1, T3 (configurable durability)
    /// Satisfies: TN1 (hybrid WAL + checkpoint strategy)
    #[cfg(feature = "wal")]
    pub fn open_with_wal<P: AsRef<Path>>(path: P, wal_config: WalConfig) -> Result<Self> {
        let path_ref = path.as_ref();
        let mmap_storage = mmap::MmapStorage::open(path_ref)?;
        let wal_path = path_ref.with_extension("wal");

        let wal = WriteAheadLog::open_with_config(&wal_path, wal_config)?;

        // INT-1: Initialize CheckpointManager for hybrid recovery strategy
        let checkpoint_dir = path_ref.parent().unwrap_or(Path::new("."));
        let checkpoint_manager =
            CheckpointManager::new(checkpoint_dir, CheckpointConfig::default())?;

        let mut storage = Storage {
            backend: StorageBackend::MmapFile(mmap_storage),
            index_manager: IndexManager::new(),
            wal: Some(Arc::new(Mutex::new(wal))),
            storage_path: Some(path_ref.to_path_buf()),
            checkpoint_manager: Some(checkpoint_manager),
        };

        // Recover from WAL on startup
        storage.recover_from_wal()?;

        // Rebuild indexes from existing data
        storage.rebuild_indexes()?;

        Ok(storage)
    }

    /// Recover storage state from WAL entries.
    /// Satisfies: RT-1 (data survives crashes through replay)
    #[cfg(feature = "wal")]
    fn recover_from_wal(&mut self) -> Result<()> {
        // Collect entries first to avoid borrow checker issues
        let entries = if let Some(wal_arc) = &self.wal {
            let wal = wal_arc.lock();
            wal.recover()?
        } else {
            return Ok(());
        };

        // Now process entries with mutable borrow available
        for entry in entries {
            match entry.operation {
                WalOperation::CreateNode { id: _, data } => {
                    // Deserialize and store node
                    if let Ok(node) = bincode::deserialize::<Node>(&data) {
                        self.apply_node_internal(node)?;
                    }
                }
                WalOperation::CreateRelationship {
                    id: _,
                    from_id: _,
                    to_id: _,
                    rel_type: _,
                    data,
                } => {
                    if let Ok(rel) = bincode::deserialize::<Relationship>(&data) {
                        self.apply_relationship_internal(rel)?;
                    }
                }
                WalOperation::DeleteNode { id: _ } => {
                    // Mark node as deleted
                    // In a full implementation, we'd track deletions
                }
                WalOperation::CommitTransaction { tx_id: _ } => {
                    // Transaction committed, changes are durable
                }
                WalOperation::Checkpoint { .. } => {
                    // Checkpoint marker - data before this is safe
                }
                _ => {
                    // Other operations handled as needed
                }
            }
        }
        Ok(())
    }

    /// Apply node directly to storage (used during recovery).
    #[cfg(feature = "wal")]
    fn apply_node_internal(&mut self, node: Node) -> Result<()> {
        match &mut self.backend {
            StorageBackend::InMemory(storage) => storage.store_node(node),
            StorageBackend::MmapFile(storage) => storage.store_node(node),
        }
    }

    /// Apply relationship directly to storage (used during recovery).
    #[cfg(feature = "wal")]
    fn apply_relationship_internal(&mut self, rel: Relationship) -> Result<()> {
        match &mut self.backend {
            StorageBackend::InMemory(storage) => storage.store_relationship(rel),
            StorageBackend::MmapFile(storage) => storage.store_relationship(rel),
        }
    }

    /// Store a node in the storage.
    /// Satisfies: RT-1 (WAL before apply for crash recovery)
    pub fn store_node(&mut self, node: Node) -> Result<()> {
        // Write to WAL first (if enabled) for durability
        #[cfg(feature = "wal")]
        if let Some(wal_arc) = &self.wal {
            let data =
                bincode::serialize(&node).map_err(|e| GraphError::Serialization(e.to_string()))?;
            let mut wal = wal_arc.lock();
            wal.append(WalOperation::CreateNode { id: node.id, data })?;
        }

        // Add to indexes
        self.index_manager.add_node(&node);

        // Store in backend
        match &mut self.backend {
            StorageBackend::InMemory(storage) => storage.store_node(node),
            StorageBackend::MmapFile(storage) => storage.store_node(node),
        }
    }

    /// Retrieve a node by its ID.
    pub fn get_node(&self, id: Id) -> Result<Option<Node>> {
        match &self.backend {
            StorageBackend::InMemory(storage) => storage.get_node(id),
            StorageBackend::MmapFile(storage) => storage.get_node(id),
        }
    }

    /// Store a relationship in the storage.
    /// Satisfies: RT-1 (WAL before apply for crash recovery)
    pub fn store_relationship(&mut self, relationship: Relationship) -> Result<()> {
        // Write to WAL first (if enabled) for durability
        #[cfg(feature = "wal")]
        if let Some(wal_arc) = &self.wal {
            let data = bincode::serialize(&relationship)
                .map_err(|e| GraphError::Serialization(e.to_string()))?;
            let mut wal = wal_arc.lock();
            wal.append(WalOperation::CreateRelationship {
                id: relationship.id,
                from_id: relationship.from_id,
                to_id: relationship.to_id,
                rel_type: relationship.rel_type.clone(),
                data,
            })?;
        }

        // Add to indexes
        self.index_manager.add_relationship(&relationship);

        // Store in backend
        match &mut self.backend {
            StorageBackend::InMemory(storage) => storage.store_relationship(relationship),
            StorageBackend::MmapFile(storage) => storage.store_relationship(relationship),
        }
    }

    /// Retrieve a relationship by its ID.
    pub fn get_relationship(&self, id: Id) -> Result<Option<Relationship>> {
        match &self.backend {
            StorageBackend::InMemory(storage) => storage.get_relationship(id),
            StorageBackend::MmapFile(storage) => storage.get_relationship(id),
        }
    }

    /// Get all relationships for a given node.
    pub fn get_relationships_for_node(&self, node_id: Id) -> Result<Vec<Relationship>> {
        match &self.backend {
            StorageBackend::InMemory(storage) => storage.get_relationships_for_node(node_id),
            StorageBackend::MmapFile(storage) => storage.get_relationships_for_node(node_id),
        }
    }

    /// Delete a node from storage.
    /// Returns the deleted node if it existed.
    pub fn delete_node(&mut self, id: Id) -> Result<Option<Node>> {
        match &mut self.backend {
            StorageBackend::InMemory(storage) => storage.delete_node(id),
            StorageBackend::MmapFile(storage) => storage.delete_node(id),
        }
    }

    /// Delete a relationship from storage.
    /// Returns the deleted relationship if it existed.
    pub fn delete_relationship(&mut self, id: Id) -> Result<Option<Relationship>> {
        match &mut self.backend {
            StorageBackend::InMemory(storage) => storage.delete_relationship(id),
            StorageBackend::MmapFile(storage) => storage.delete_relationship(id),
        }
    }

    /// Check if a node has any relationships.
    pub fn node_has_relationships(&self, node_id: Id) -> bool {
        match &self.backend {
            StorageBackend::InMemory(storage) => storage.node_has_relationships(node_id),
            StorageBackend::MmapFile(storage) => storage.node_has_relationships(node_id),
        }
    }

    /// Get the number of nodes in storage.
    pub fn node_count(&self) -> usize {
        match &self.backend {
            StorageBackend::InMemory(storage) => storage.node_count(),
            StorageBackend::MmapFile(storage) => storage.node_count(),
        }
    }

    /// Get the number of relationships in storage.
    pub fn relationship_count(&self) -> usize {
        match &self.backend {
            StorageBackend::InMemory(storage) => storage.relationship_count(),
            StorageBackend::MmapFile(storage) => storage.relationship_count(),
        }
    }

    /// Flush any pending writes to disk (only applicable for persistent storage).
    /// Satisfies: RT-1 (ensure WAL is synced for durability)
    pub fn flush(&self) -> Result<()> {
        // Sync WAL first for durability
        #[cfg(feature = "wal")]
        if let Some(wal_arc) = &self.wal {
            let mut wal = wal_arc.lock();
            wal.sync()?;
        }

        // Then flush backend
        match &self.backend {
            StorageBackend::InMemory(_) => Ok(()), // No-op for in-memory
            StorageBackend::MmapFile(storage) => storage.flush(),
        }
    }

    /// Sync WAL to disk explicitly.
    /// Satisfies: RT-1 (explicit durability control)
    #[cfg(feature = "wal")]
    pub fn sync_wal(&self) -> Result<()> {
        if let Some(wal_arc) = &self.wal {
            let mut wal = wal_arc.lock();
            wal.sync()?;
        }
        Ok(())
    }

    /// Create a checkpoint in the WAL.
    /// Satisfies: TN1 (hybrid WAL + checkpoints)
    #[cfg(feature = "wal")]
    pub fn checkpoint(&self) -> Result<u64> {
        if let Some(wal_arc) = &self.wal {
            let mut wal = wal_arc.lock();
            wal.checkpoint()
        } else {
            Err(GraphError::Storage("WAL not enabled".to_string()))
        }
    }

    /// Check if WAL is enabled.
    #[cfg(feature = "wal")]
    pub fn has_wal(&self) -> bool {
        self.wal.is_some()
    }

    /// Check if WAL is enabled (always false when feature disabled).
    #[cfg(not(feature = "wal"))]
    pub fn has_wal(&self) -> bool {
        false
    }

    /// INT-1: Check if checkpoint manager is enabled.
    /// Satisfies: TN1 (hybrid strategy awareness)
    #[cfg(feature = "wal")]
    pub fn has_checkpoint_manager(&self) -> bool {
        self.checkpoint_manager.is_some()
    }

    /// INT-1: Check if checkpoint manager is enabled (always false when feature disabled).
    #[cfg(not(feature = "wal"))]
    pub fn has_checkpoint_manager(&self) -> bool {
        false
    }

    /// INT-1: Get access to the checkpoint manager for advanced operations.
    /// Satisfies: RT-1, TN1 (checkpoint management)
    #[cfg(feature = "wal")]
    pub fn checkpoint_manager(&self) -> Option<&CheckpointManager> {
        self.checkpoint_manager.as_ref()
    }

    /// INT-1: Get mutable access to checkpoint manager.
    /// Satisfies: RT-1, TN1 (checkpoint management)
    #[cfg(feature = "wal")]
    pub fn checkpoint_manager_mut(&mut self) -> Option<&mut CheckpointManager> {
        self.checkpoint_manager.as_mut()
    }

    /// Create a property index for fast lookups.
    pub fn create_property_index(&mut self, property_key: String) {
        self.index_manager.create_property_index(property_key);
    }

    /// Create a range index for range queries.
    pub fn create_range_index(&mut self, property_key: String) {
        self.index_manager.create_range_index(property_key);
    }

    /// Create a composite index for multi-property queries.
    pub fn create_composite_index(&mut self, property_keys: Vec<String>) {
        self.index_manager.create_composite_index(property_keys);
    }

    /// Find entities by exact property value.
    pub fn find_by_property(&self, property_key: &str, property_value: &Value) -> Vec<Id> {
        self.index_manager
            .find_by_property(property_key, property_value)
    }

    /// Find entities in the given range.
    pub fn find_in_range(
        &self,
        property_key: &str,
        min_value: &Value,
        max_value: &Value,
    ) -> Vec<Id> {
        self.index_manager
            .find_in_range(property_key, min_value, max_value)
    }

    /// Find relationships by type.
    pub fn find_relationships_by_type(&self, rel_type: &str) -> Vec<Id> {
        self.index_manager.find_relationships_by_type(rel_type)
    }

    /// Find outgoing relationships of a given type from a node.
    pub fn find_outgoing_relationships(&self, from_id: Id, rel_type: &str) -> Vec<Id> {
        self.index_manager
            .find_outgoing_relationships(from_id, rel_type)
    }

    /// Find incoming relationships of a given type to a node.
    pub fn find_incoming_relationships(&self, to_id: Id, rel_type: &str) -> Vec<Id> {
        self.index_manager
            .find_incoming_relationships(to_id, rel_type)
    }

    /// Get indexing statistics.
    pub fn get_index_stats(&self) -> crate::index::IndexStats {
        self.index_manager.get_index_stats()
    }

    /// Rebuild indexes from existing data (used after loading from disk).
    fn rebuild_indexes(&mut self) -> Result<()> {
        // Get all nodes and relationships from storage and add them to indexes
        match &self.backend {
            StorageBackend::InMemory(storage) => {
                for node in storage.nodes.values() {
                    self.index_manager.add_node(node);
                }
                for relationship in storage.relationships.values() {
                    self.index_manager.add_relationship(relationship);
                }
            }
            StorageBackend::MmapFile(_) => {
                // For mmap storage, we would need to iterate through all stored data
                // This is a simplified version - in production this would scan the file
                // For now, indexes will be empty until new data is added
            }
        }
        Ok(())
    }

    /// Get access to the index manager for advanced operations.
    pub fn index_manager(&self) -> &IndexManager {
        &self.index_manager
    }

    /// Get mutable access to the index manager for advanced operations.
    pub fn index_manager_mut(&mut self) -> &mut IndexManager {
        &mut self.index_manager
    }
}

impl InMemoryStorage {
    /// Store a node in memory.
    pub fn store_node(&mut self, node: Node) -> Result<()> {
        let id = node.id;
        self.nodes.insert(id, node);

        // Initialize empty relationship list for the node
        self.node_relationships.entry(id).or_default();

        Ok(())
    }

    /// Retrieve a node by its ID.
    pub fn get_node(&self, id: Id) -> Result<Option<Node>> {
        Ok(self.nodes.get(&id).cloned())
    }

    /// Store a relationship in memory.
    pub fn store_relationship(&mut self, relationship: Relationship) -> Result<()> {
        let id = relationship.id;
        let from_id = relationship.from_id;
        let to_id = relationship.to_id;

        // Verify that both nodes exist
        if !self.nodes.contains_key(&from_id) {
            return Err(GraphError::NotFound(format!("Node {from_id} not found")));
        }
        if !self.nodes.contains_key(&to_id) {
            return Err(GraphError::NotFound(format!("Node {to_id} not found")));
        }

        // Store the relationship
        self.relationships.insert(id, relationship);

        // Update node relationship indexes
        self.node_relationships.entry(from_id).or_default().push(id);

        if from_id != to_id {
            self.node_relationships.entry(to_id).or_default().push(id);
        }

        Ok(())
    }

    /// Retrieve a relationship by its ID.
    pub fn get_relationship(&self, id: Id) -> Result<Option<Relationship>> {
        Ok(self.relationships.get(&id).cloned())
    }

    /// Get all relationships for a given node.
    pub fn get_relationships_for_node(&self, node_id: Id) -> Result<Vec<Relationship>> {
        let empty_vec = Vec::new();
        let relationship_ids = self.node_relationships.get(&node_id).unwrap_or(&empty_vec);

        let mut relationships = Vec::new();
        for &rel_id in relationship_ids {
            if let Some(rel) = self.relationships.get(&rel_id) {
                relationships.push(rel.clone());
            }
        }

        Ok(relationships)
    }

    /// Get the number of nodes in storage.
    pub fn node_count(&self) -> usize {
        self.nodes.len()
    }

    /// Get the number of relationships in storage.
    pub fn relationship_count(&self) -> usize {
        self.relationships.len()
    }

    /// Delete a node from storage.
    /// Returns the deleted node if it existed.
    pub fn delete_node(&mut self, id: Id) -> Result<Option<Node>> {
        // Remove from node_relationships tracking
        self.node_relationships.remove(&id);
        // Remove the node
        Ok(self.nodes.remove(&id))
    }

    /// Delete a relationship from storage.
    /// Returns the deleted relationship if it existed.
    pub fn delete_relationship(&mut self, id: Id) -> Result<Option<Relationship>> {
        if let Some(rel) = self.relationships.remove(&id) {
            // Remove from node_relationships tracking
            if let Some(rels) = self.node_relationships.get_mut(&rel.from_id) {
                rels.retain(|&r| r != id);
            }
            if rel.from_id != rel.to_id {
                if let Some(rels) = self.node_relationships.get_mut(&rel.to_id) {
                    rels.retain(|&r| r != id);
                }
            }
            Ok(Some(rel))
        } else {
            Ok(None)
        }
    }

    /// Check if a node has any relationships.
    pub fn node_has_relationships(&self, node_id: Id) -> bool {
        self.node_relationships
            .get(&node_id)
            .map(|rels| !rels.is_empty())
            .unwrap_or(false)
    }
}

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

    #[test]
    fn test_storage_node_operations() {
        let mut storage = Storage::new().unwrap();
        let node = Node::new(1, HashMap::new());

        storage.store_node(node.clone()).unwrap();
        let retrieved = storage.get_node(1).unwrap();
        assert_eq!(retrieved, Some(node));
        assert_eq!(storage.node_count(), 1);
    }

    #[test]
    fn test_storage_relationship_operations() {
        let mut storage = Storage::new().unwrap();

        // Create nodes first
        let node1 = Node::new(1, HashMap::new());
        let node2 = Node::new(2, HashMap::new());
        storage.store_node(node1).unwrap();
        storage.store_node(node2).unwrap();

        // Create relationship
        let rel = Relationship::new(1, 1, 2, "KNOWS".to_string(), HashMap::new());
        storage.store_relationship(rel.clone()).unwrap();

        let retrieved = storage.get_relationship(1).unwrap();
        assert_eq!(retrieved, Some(rel));
        assert_eq!(storage.relationship_count(), 1);
    }

    #[test]
    fn test_node_relationships() {
        let mut storage = Storage::new().unwrap();

        // Create nodes
        let node1 = Node::new(1, HashMap::new());
        let node2 = Node::new(2, HashMap::new());
        storage.store_node(node1).unwrap();
        storage.store_node(node2).unwrap();

        // Create relationship
        let rel = Relationship::new(1, 1, 2, "KNOWS".to_string(), HashMap::new());
        storage.store_relationship(rel.clone()).unwrap();

        // Check relationships for node 1
        let rels = storage.get_relationships_for_node(1).unwrap();
        assert_eq!(rels.len(), 1);
        assert_eq!(rels[0], rel);

        // Check relationships for node 2
        let rels = storage.get_relationships_for_node(2).unwrap();
        assert_eq!(rels.len(), 1);
        assert_eq!(rels[0], rel);

        // Test relationship type lookup
        let knows_rels = storage.find_relationships_by_type("KNOWS");
        assert_eq!(knows_rels.len(), 1);
        assert!(knows_rels.contains(&1));

        // Test outgoing relationships
        let outgoing = storage.find_outgoing_relationships(1, "KNOWS");
        assert_eq!(outgoing.len(), 1);
        assert!(outgoing.contains(&1));
    }

    #[test]
    fn test_index_stats() {
        let mut storage = Storage::new().unwrap();
        storage.create_property_index("name".to_string());
        storage.create_range_index("age".to_string());

        let stats = storage.get_index_stats();
        assert_eq!(stats.property_index_count, 1);
        assert_eq!(stats.range_index_count, 1);
        assert_eq!(stats.composite_index_count, 0);
    }
}