oxirs-star 0.2.4

//! Memory-efficient storage for large RDF-star graphs using scirs2-core
//!
//! This module provides disk-backed, memory-mapped storage for RDF-star triples
//! enabling processing of datasets larger than available RAM.

use crate::{StarError, StarResult, StarTerm, StarTriple};
use scirs2_core::memory_efficient::{create_mmap, AccessMode, MemoryMappedArray};
use scirs2_core::ndarray_ext::Array1;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::path::{Path, PathBuf};

/// Memory-efficient RDF-star store for large datasets
///
/// Uses memory-mapped files and lazy evaluation to handle graphs
/// larger than available RAM.
pub struct MemoryEfficientStore {
    /// Base directory for storage
    base_path: PathBuf,
    /// Memory-mapped triple storage
    triple_data: Option<MemoryMappedArray<u8>>,
    /// Chunked index for efficient queries
    index: ChunkedIndex,
    /// Statistics
    stats: StoreStatistics,
    /// Configuration
    config: StoreConfig,
}

/// Configuration for memory-efficient storage
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct StoreConfig {
    /// Chunk size for processing (in number of triples)
    pub chunk_size: usize,
    /// Maximum memory usage (in bytes)
    pub max_memory: usize,
    /// Enable compression
    pub enable_compression: bool,
    /// Access mode for memory mapping
    pub access_mode: MappedAccessMode,
}

/// Access mode for memory-mapped files
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum MappedAccessMode {
    /// Read-only access
    ReadOnly,
    /// Read-write access
    ReadWrite,
    /// Copy-on-write access
    CopyOnWrite,
}

impl Default for StoreConfig {
    fn default() -> Self {
        Self {
            chunk_size: 10000,
            max_memory: 1 << 30, // 1GB
            enable_compression: true,
            access_mode: MappedAccessMode::ReadWrite,
        }
    }
}

/// Statistics for memory-efficient store
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct StoreStatistics {
    /// Total triples stored
    pub total_triples: usize,
    /// Memory usage (bytes)
    pub memory_usage: usize,
    /// Disk usage (bytes)
    pub disk_usage: usize,
    /// Cache hits
    pub cache_hits: usize,
    /// Cache misses
    pub cache_misses: usize,
}

/// Chunked index for efficient queries
struct ChunkedIndex {
    /// Subject index chunks
    subject_chunks: Vec<HashMap<String, Vec<usize>>>,
    /// Predicate index chunks
    predicate_chunks: Vec<HashMap<String, Vec<usize>>>,
    /// Object index chunks
    object_chunks: Vec<HashMap<String, Vec<usize>>>,
}

impl ChunkedIndex {
    fn new() -> Self {
        Self {
            subject_chunks: Vec::new(),
            predicate_chunks: Vec::new(),
            object_chunks: Vec::new(),
        }
    }

    fn add_chunk(&mut self) {
        self.subject_chunks.push(HashMap::new());
        self.predicate_chunks.push(HashMap::new());
        self.object_chunks.push(HashMap::new());
    }
}

impl MemoryEfficientStore {
    /// Create a new memory-efficient store
    pub fn new<P: AsRef<Path>>(path: P) -> StarResult<Self> {
        Self::with_config(path, StoreConfig::default())
    }

    /// Create with custom configuration
    pub fn with_config<P: AsRef<Path>>(path: P, config: StoreConfig) -> StarResult<Self> {
        let base_path = path.as_ref().to_path_buf();

        // Create directory if it doesn't exist
        if !base_path.exists() {
            std::fs::create_dir_all(&base_path).map_err(|e| {
                StarError::resource_error(format!("Failed to create directory: {}", e))
            })?;
        }

        Ok(Self {
            base_path,
            triple_data: None,
            index: ChunkedIndex::new(),
            stats: StoreStatistics::default(),
            config,
        })
    }

    /// Insert triples in batches for efficient disk writes
    pub fn insert_batch(&mut self, triples: &[StarTriple]) -> StarResult<()> {
        // Serialize triples to bytes
        let serialized = self.serialize_triples(triples)?;

        // Write to memory-mapped file
        let data_path = self.base_path.join("triples.bin");

        if self.triple_data.is_none() {
            // Create initial memory map
            let array = Array1::from_vec(serialized.clone());
            let mmap = create_mmap(&array, &data_path, AccessMode::ReadWrite, 0).map_err(|e| {
                StarError::resource_error(format!("Failed to create memory map: {}", e))
            })?;
            self.triple_data = Some(mmap);
        } else {
            // Append to existing memory map
            self.append_data(&serialized)?;
        }

        // Update indices in chunks
        self.update_indices_chunked(triples)?;

        // Update statistics
        self.stats.total_triples += triples.len();
        self.stats.disk_usage += serialized.len();

        Ok(())
    }

    /// Query triples by subject with chunked processing
    pub fn query_by_subject_chunked<F>(
        &self,
        subject: &StarTerm,
        mut processor: F,
    ) -> StarResult<()>
    where
        F: FnMut(&StarTriple) -> StarResult<()>,
    {
        let subject_key = self.term_to_key(subject);

        // Query each chunk
        for (chunk_id, chunk_index) in self.index.subject_chunks.iter().enumerate() {
            if let Some(indices) = chunk_index.get(&subject_key) {
                for &idx in indices {
                    let triple = self.load_triple_from_chunk(chunk_id, idx)?;
                    processor(&triple)?;
                }
            }
        }

        Ok(())
    }

    /// Process entire store in chunks (for large operations)
    pub fn process_all_chunks<F>(&self, mut processor: F) -> StarResult<()>
    where
        F: FnMut(&[StarTriple]) -> StarResult<()>,
    {
        let total_chunks = self.index.subject_chunks.len();

        for chunk_id in 0..total_chunks {
            let triples = self.load_chunk(chunk_id)?;
            processor(&triples)?;
        }

        Ok(())
    }

    /// Optimize storage (compact, reindex, garbage collect)
    pub fn optimize(&mut self) -> StarResult<()> {
        // Compact fragmented data
        self.compact_storage()?;

        // Rebuild indices
        self.rebuild_indices()?;

        // Update statistics
        self.update_statistics()?;

        Ok(())
    }

    /// Get store statistics
    pub fn statistics(&self) -> &StoreStatistics {
        &self.stats
    }

    /// Serialize triples to bytes
    fn serialize_triples(&self, triples: &[StarTriple]) -> StarResult<Vec<u8>> {
        // Use bincode 2.0 for efficient serialization
        oxicode::serde::encode_to_vec(&triples, oxicode::config::standard())
            .map_err(|e| StarError::serialization_error(format!("Serialization failed: {}", e)))
    }

    /// Deserialize triples from bytes
    #[allow(dead_code)]
    fn deserialize_triples(&self, data: &[u8]) -> StarResult<Vec<StarTriple>> {
        oxicode::serde::decode_from_slice(data, oxicode::config::standard())
            .map(|(triples, _)| triples)
            .map_err(|e| StarError::parse_error(format!("Deserialization failed: {}", e)))
    }

    /// Append data to existing memory map
    fn append_data(&mut self, data: &[u8]) -> StarResult<()> {
        // For now, we'll recreate the memory map with appended data
        // In production, this should use a more sophisticated append mechanism
        let mut current_data = Vec::new();

        // Note: MemoryMappedArray doesn't expose raw data directly
        // In practice, we'd need to read from the file or use a different approach

        current_data.extend_from_slice(data);

        let data_path = self.base_path.join("triples.bin");
        let array = Array1::from_vec(current_data);
        let mmap = create_mmap(&array, &data_path, AccessMode::ReadWrite, 0).map_err(|e| {
            StarError::resource_error(format!("Failed to recreate memory map: {}", e))
        })?;

        self.triple_data = Some(mmap);

        Ok(())
    }

    /// Update indices for new triples in chunks
    fn update_indices_chunked(&mut self, triples: &[StarTriple]) -> StarResult<()> {
        let chunk_id = self.index.subject_chunks.len();

        if chunk_id == 0 || triples.len() >= self.config.chunk_size {
            self.index.add_chunk();
        }

        let current_chunk_id = self.index.subject_chunks.len() - 1;

        // Collect keys first to avoid borrow issues
        let keys: Vec<(String, String, String)> = triples
            .iter()
            .map(|triple| {
                (
                    Self::term_to_key_static(&triple.subject),
                    Self::term_to_key_static(&triple.predicate),
                    Self::term_to_key_static(&triple.object),
                )
            })
            .collect();

        // Now update indices
        let subject_chunk = &mut self.index.subject_chunks[current_chunk_id];
        let predicate_chunk = &mut self.index.predicate_chunks[current_chunk_id];
        let object_chunk = &mut self.index.object_chunks[current_chunk_id];

        for (idx, (subject_key, predicate_key, object_key)) in keys.into_iter().enumerate() {
            subject_chunk.entry(subject_key).or_default().push(idx);
            predicate_chunk.entry(predicate_key).or_default().push(idx);
            object_chunk.entry(object_key).or_default().push(idx);
        }

        Ok(())
    }

    /// Convert term to indexable key
    fn term_to_key(&self, term: &StarTerm) -> String {
        Self::term_to_key_static(term)
    }

    /// Static version for avoiding borrow issues
    fn term_to_key_static(term: &StarTerm) -> String {
        format!("{:?}", term)
    }

    /// Load a specific triple from a chunk
    fn load_triple_from_chunk(&self, _chunk_id: usize, _idx: usize) -> StarResult<StarTriple> {
        // Simplified implementation - would need actual chunk-based loading
        Err(StarError::processing_error("Not implemented"))
    }

    /// Load entire chunk
    fn load_chunk(&self, _chunk_id: usize) -> StarResult<Vec<StarTriple>> {
        // Simplified implementation
        Ok(Vec::new())
    }

    /// Compact storage to reduce fragmentation
    fn compact_storage(&mut self) -> StarResult<()> {
        // Implementation would rewrite data without gaps
        Ok(())
    }

    /// Rebuild all indices
    fn rebuild_indices(&mut self) -> StarResult<()> {
        self.index = ChunkedIndex::new();
        // Would reload and reindex all data
        Ok(())
    }

    /// Update storage statistics
    fn update_statistics(&mut self) -> StarResult<()> {
        if let Some(ref mmap) = self.triple_data {
            // Estimate memory usage from mmap size
            self.stats.memory_usage = mmap.size;
        }
        Ok(())
    }
}

/// Streaming iterator for memory-efficient traversal
pub struct ChunkedIterator<'a> {
    store: &'a MemoryEfficientStore,
    current_chunk: usize,
    chunk_offset: usize,
    total_chunks: usize,
}

impl<'a> ChunkedIterator<'a> {
    fn new(store: &'a MemoryEfficientStore) -> Self {
        let total_chunks = store.index.subject_chunks.len();
        Self {
            store,
            current_chunk: 0,
            chunk_offset: 0,
            total_chunks,
        }
    }
}

impl<'a> Iterator for ChunkedIterator<'a> {
    type Item = StarResult<StarTriple>;

    fn next(&mut self) -> Option<Self::Item> {
        if self.current_chunk >= self.total_chunks {
            return None;
        }

        // Load current triple
        match self
            .store
            .load_triple_from_chunk(self.current_chunk, self.chunk_offset)
        {
            Ok(triple) => {
                self.chunk_offset += 1;
                Some(Ok(triple))
            }
            Err(_) => {
                // Move to next chunk
                self.current_chunk += 1;
                self.chunk_offset = 0;
                self.next()
            }
        }
    }
}

impl MemoryEfficientStore {
    /// Get chunked iterator for streaming access
    pub fn iter_chunked(&self) -> ChunkedIterator<'_> {
        ChunkedIterator::new(self)
    }
}

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

    #[test]
    fn test_store_creation() {
        let temp_path = temp_dir().join("oxirs_star_test_store");
        let store = MemoryEfficientStore::new(&temp_path);
        assert!(store.is_ok());

        // Cleanup
        let _ = std::fs::remove_dir_all(&temp_path);
    }

    #[test]
    fn test_config_default() {
        let config = StoreConfig::default();
        assert_eq!(config.chunk_size, 10000);
        assert!(config.enable_compression);
    }

    #[test]
    fn test_statistics() {
        let temp_path = temp_dir().join("oxirs_star_test_stats");
        let store = MemoryEfficientStore::new(&temp_path).unwrap();
        let stats = store.statistics();
        assert_eq!(stats.total_triples, 0);

        // Cleanup
        let _ = std::fs::remove_dir_all(&temp_path);
    }
}