xz-memory-engine 0.2.0

use async_trait::async_trait;
use std::fs;
use std::path::{Path, PathBuf};

use serde::{Deserialize, Serialize};

use xz_memory_core::StoreError;
use xz_memory_core::traits::store::EntryStore;
use xz_memory_core::types::entry::*;

/// Internal YAML frontmatter representation for .md entry files.
#[derive(Serialize, Deserialize)]
struct FrontMatter {
    id: String,
    partition: String,
    recorded_at: u64,
}

/// Disk-backed [`EntryStore`] using markdown files in a directory tree.
///
/// Each partition is a subdirectory. Each entry is a .md file named `{entry_id}.md`
/// with YAML frontmatter containing `id`, `partition`, `recorded_at`, followed by
/// the entry body.
///
/// # Examples
///
/// ```
/// use std::path::PathBuf;
/// use xz_memory_engine::MarkdownEntryStore;
///
/// let store = MarkdownEntryStore::new(PathBuf::from("/tmp/xz-test-store"));
/// ```
pub struct MarkdownEntryStore {
    root: PathBuf,
}

impl MarkdownEntryStore {
    /// Create a new `MarkdownEntryStore` rooted at the given path.
    ///
    /// The directory will be created lazily on first write.
    pub fn new(root: PathBuf) -> Self {
        MarkdownEntryStore { root }
    }

    fn partition_dir(&self, partition: &str) -> PathBuf {
        self.root.join(partition)
    }

    fn serialize_entry(entry: &Entry) -> Result<String, StoreError> {
        let fm = FrontMatter {
            id: entry.id.clone(),
            partition: entry.partition.clone(),
            recorded_at: entry.recorded_at,
        };
        let yaml = serde_yaml::to_string(&fm)
            .map_err(|e| StoreError::Serialization(format!("Failed to serialize entry: {e}")))?;
        Ok(format!("---\n{}---\n{}", yaml, entry.body))
    }

    fn parse_file(path: &Path) -> Result<Entry, StoreError> {
        let content = fs::read_to_string(path)
            .map_err(|e| StoreError::Backend(format!("Failed to read {}: {e}", path.display())))?;

        let parts: Vec<&str> = content.splitn(3, "---\n").collect();
        if parts.len() < 3 {
            return Err(StoreError::Serialization(format!(
                "Invalid markdown entry in {}: missing YAML frontmatter",
                path.display()
            )));
        }

        let fm: FrontMatter = serde_yaml::from_str(parts[1]).map_err(|e| {
            StoreError::Serialization(format!("Failed to parse YAML in {}: {e}", path.display()))
        })?;

        Ok(Entry {
            id: fm.id,
            partition: fm.partition,
            body: parts[2].to_owned(),
            recorded_at: fm.recorded_at,
        })
    }
}

#[async_trait]
impl EntryStore for MarkdownEntryStore {
    async fn append(&self, entry: Entry) -> Result<(), StoreError> {
        let dir = self.partition_dir(&entry.partition);
        fs::create_dir_all(&dir).map_err(|e| {
            StoreError::Backend(format!("Failed to create partition directory: {e}"))
        })?;

        let content = Self::serialize_entry(&entry)?;
        let path = dir.join(format!("{}.md", entry.id));
        fs::write(&path, content)
            .map_err(|e| StoreError::Backend(format!("Failed to write entry file: {e}")))?;
        Ok(())
    }

    async fn query(
        &self,
        partition: &str,
        range: &TimeRange,
        opts: &QueryOptions,
    ) -> Result<Vec<Entry>, StoreError> {
        let dir = self.partition_dir(partition);
        if !dir.exists() {
            return Ok(vec![]);
        }

        let mut entries: Vec<Entry> = Vec::new();
        for entry_res in fs::read_dir(&dir)
            .map_err(|e| StoreError::Backend(format!("Failed to read directory: {e}")))?
        {
            let entry = entry_res
                .map_err(|e| StoreError::Backend(format!("Failed to read directory entry: {e}")))?;
            let path = entry.path();
            if path.extension().map_or(false, |ext| ext == "md") {
                if let Ok(e) = Self::parse_file(&path) {
                    let after_start = range.start.is_none_or(|s| e.recorded_at >= s);
                    let before_end = range.end.is_none_or(|t| e.recorded_at <= t);
                    if after_start && before_end {
                        entries.push(e);
                    }
                }
            }
        }

        match opts.sort {
            SortOrder::Ascending => entries.sort_by_key(|e| e.recorded_at),
            SortOrder::Descending => entries.sort_by_key(|b| std::cmp::Reverse(b.recorded_at)),
        }
        entries.truncate(opts.limit);
        Ok(entries)
    }

    async fn evict(&self, partition: &str, keep: usize) -> Result<usize, StoreError> {
        let dir = self.partition_dir(partition);
        if !dir.exists() {
            return Ok(0);
        }

        let mut entries: Vec<(PathBuf, Entry)> = Vec::new();
        for entry_res in fs::read_dir(&dir)
            .map_err(|e| StoreError::Backend(format!("Failed to read directory: {e}")))?
        {
            let entry = entry_res
                .map_err(|e| StoreError::Backend(format!("Failed to read directory entry: {e}")))?;
            let path = entry.path();
            if path.extension().map_or(false, |ext| ext == "md") {
                if let Ok(e) = Self::parse_file(&path) {
                    entries.push((path, e));
                }
            }
        }

        if entries.len() <= keep {
            return Ok(0);
        }

        entries.sort_by_key(|(_, e)| e.recorded_at);
        let remove_count = entries.len() - keep;
        for (path, _) in entries.iter().take(remove_count) {
            fs::remove_file(path)
                .map_err(|e| StoreError::Backend(format!("Failed to delete evicted entry: {e}")))?;
        }

        Ok(remove_count)
    }

    async fn delete(&self, id: &str) -> Result<(), StoreError> {
        if !self.root.exists() {
            return Ok(());
        }

        for entry_res in fs::read_dir(&self.root)
            .map_err(|e| StoreError::Backend(format!("Failed to read root: {e}")))?
        {
            let entry = entry_res
                .map_err(|e| StoreError::Backend(format!("Failed to read root entry: {e}")))?;
            let path = entry.path();
            if path.is_dir() {
                let file_path = path.join(format!("{id}.md"));
                if file_path.exists() {
                    fs::remove_file(&file_path).map_err(|e| {
                        StoreError::Backend(format!("Failed to delete entry file: {e}"))
                    })?;
                }
            }
        }

        Ok(())
    }

    async fn clear_partition(&self, partition: &str) -> Result<(), StoreError> {
        let dir = self.partition_dir(partition);
        if dir.exists() {
            fs::remove_dir_all(&dir)
                .map_err(|e| StoreError::Backend(format!("Failed to clear partition: {e}")))?;
        }
        Ok(())
    }
}

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

    fn entry(id: &str, partition: &str, body: &str, recorded_at: u64) -> Entry {
        Entry { id: id.into(), partition: partition.into(), body: body.into(), recorded_at }
    }

    fn unbounded() -> TimeRange {
        TimeRange { start: None, end: None }
    }

    fn opts(limit: usize, sort: SortOrder) -> QueryOptions {
        QueryOptions { limit, sort }
    }

    #[tokio::test]
    async fn test_append_creates_file() {
        let tmp = TempDir::new().unwrap();
        let store = MarkdownEntryStore::new(tmp.path().to_path_buf());

        store.append(entry("e1", "test", "hello world", 1000)).await.unwrap();

        let file_path = tmp.path().join("test").join("e1.md");
        assert!(file_path.exists());
        let content = fs::read_to_string(&file_path).unwrap();
        assert!(content.contains("id: e1"));
        assert!(content.contains("partition: test"));
        assert!(content.contains("recorded_at: 1000"));
        assert!(content.contains("hello world"));
    }

    #[tokio::test]
    async fn test_append_preserves_body_newlines() {
        let tmp = TempDir::new().unwrap();
        let store = MarkdownEntryStore::new(tmp.path().to_path_buf());

        store.append(entry("e1", "test", "line1\nline2\nline3", 1000)).await.unwrap();

        let file_path = tmp.path().join("test").join("e1.md");
        let content = fs::read_to_string(&file_path).unwrap();
        assert!(content.ends_with("line1\nline2\nline3"));
    }

    #[tokio::test]
    async fn test_query_filters_by_time_range() {
        let tmp = TempDir::new().unwrap();
        let store = MarkdownEntryStore::new(tmp.path().to_path_buf());

        store.append(entry("e1", "qtest", "a", 1000)).await.unwrap();
        store.append(entry("e2", "qtest", "b", 2000)).await.unwrap();
        store.append(entry("e3", "qtest", "c", 3000)).await.unwrap();

        let range = TimeRange { start: Some(1500), end: Some(2500) };
        let results = store.query("qtest", &range, &opts(10, SortOrder::Ascending)).await.unwrap();

        assert_eq!(results.len(), 1);
        assert_eq!(results[0].id, "e2");
    }

    #[tokio::test]
    async fn test_query_unbounded_range() {
        let tmp = TempDir::new().unwrap();
        let store = MarkdownEntryStore::new(tmp.path().to_path_buf());

        store.append(entry("e1", "unb", "a", 1000)).await.unwrap();
        store.append(entry("e2", "unb", "b", 2000)).await.unwrap();

        let results =
            store.query("unb", &unbounded(), &opts(10, SortOrder::Ascending)).await.unwrap();

        assert_eq!(results.len(), 2);
    }

    #[tokio::test]
    async fn test_query_sorts_descending() {
        let tmp = TempDir::new().unwrap();
        let store = MarkdownEntryStore::new(tmp.path().to_path_buf());

        store.append(entry("a", "sort", "x", 1000)).await.unwrap();
        store.append(entry("b", "sort", "y", 3000)).await.unwrap();
        store.append(entry("c", "sort", "z", 2000)).await.unwrap();

        let results =
            store.query("sort", &unbounded(), &opts(10, SortOrder::Descending)).await.unwrap();

        assert_eq!(results.len(), 3);
        assert_eq!(results[0].id, "b");
        assert_eq!(results[1].id, "c");
        assert_eq!(results[2].id, "a");
    }

    #[tokio::test]
    async fn test_query_respects_limit() {
        let tmp = TempDir::new().unwrap();
        let store = MarkdownEntryStore::new(tmp.path().to_path_buf());

        store.append(entry("a", "lim", "x", 1000)).await.unwrap();
        store.append(entry("b", "lim", "y", 2000)).await.unwrap();
        store.append(entry("c", "lim", "z", 3000)).await.unwrap();

        let results =
            store.query("lim", &unbounded(), &opts(2, SortOrder::Descending)).await.unwrap();

        assert_eq!(results.len(), 2);
        assert_eq!(results[0].recorded_at, 3000);
        assert_eq!(results[1].recorded_at, 2000);
    }

    #[tokio::test]
    async fn test_query_empty_partition() {
        let tmp = TempDir::new().unwrap();
        let store = MarkdownEntryStore::new(tmp.path().to_path_buf());

        let results = store
            .query("nonexistent", &unbounded(), &opts(10, SortOrder::Ascending))
            .await
            .unwrap();

        assert!(results.is_empty());
    }

    #[tokio::test]
    async fn test_evict_removes_oldest() {
        let tmp = TempDir::new().unwrap();
        let store = MarkdownEntryStore::new(tmp.path().to_path_buf());

        store.append(entry("e1", "evict", "a", 1000)).await.unwrap();
        store.append(entry("e2", "evict", "b", 3000)).await.unwrap();
        store.append(entry("e3", "evict", "c", 2000)).await.unwrap();

        let removed = store.evict("evict", 1).await.unwrap();
        assert_eq!(removed, 2);

        let results =
            store.query("evict", &unbounded(), &opts(10, SortOrder::Ascending)).await.unwrap();
        assert_eq!(results.len(), 1);
        assert_eq!(results[0].id, "e2");
    }

    #[tokio::test]
    async fn test_evict_noop_when_under_limit() {
        let tmp = TempDir::new().unwrap();
        let store = MarkdownEntryStore::new(tmp.path().to_path_buf());

        store.append(entry("e1", "noop", "a", 1000)).await.unwrap();

        let removed = store.evict("noop", 5).await.unwrap();
        assert_eq!(removed, 0);

        let results =
            store.query("noop", &unbounded(), &opts(10, SortOrder::Ascending)).await.unwrap();
        assert_eq!(results.len(), 1);
    }

    #[tokio::test]
    async fn test_evict_empty_partition() {
        let tmp = TempDir::new().unwrap();
        let store = MarkdownEntryStore::new(tmp.path().to_path_buf());

        let removed = store.evict("ghost", 5).await.unwrap();
        assert_eq!(removed, 0);
    }

    #[tokio::test]
    async fn test_delete_removes_by_id() {
        let tmp = TempDir::new().unwrap();
        let store = MarkdownEntryStore::new(tmp.path().to_path_buf());

        store.append(entry("delme", "p1", "x", 1000)).await.unwrap();
        store.append(entry("keep", "p1", "y", 2000)).await.unwrap();

        store.delete("delme").await.unwrap();

        let results =
            store.query("p1", &unbounded(), &opts(10, SortOrder::Ascending)).await.unwrap();
        assert_eq!(results.len(), 1);
        assert_eq!(results[0].id, "keep");
    }

    #[tokio::test]
    async fn test_delete_nonexistent_id() {
        let tmp = TempDir::new().unwrap();
        let store = MarkdownEntryStore::new(tmp.path().to_path_buf());

        store.append(entry("real", "p", "body", 1000)).await.unwrap();
        store.delete("ghost").await.unwrap();

        let results =
            store.query("p", &unbounded(), &opts(10, SortOrder::Ascending)).await.unwrap();
        assert_eq!(results.len(), 1);
    }

    #[tokio::test]
    async fn test_delete_from_empty_store() {
        let tmp = TempDir::new().unwrap();
        let store = MarkdownEntryStore::new(tmp.path().to_path_buf());

        store.delete("nonexistent").await.unwrap();
    }

    #[tokio::test]
    async fn test_clear_partition_removes_directory() {
        let tmp = TempDir::new().unwrap();
        let store = MarkdownEntryStore::new(tmp.path().to_path_buf());

        store.append(entry("e1", "clear", "x", 1000)).await.unwrap();
        assert!(tmp.path().join("clear").exists());

        store.clear_partition("clear").await.unwrap();
        assert!(!tmp.path().join("clear").exists());
    }

    #[tokio::test]
    async fn test_clear_partition_nonexistent() {
        let tmp = TempDir::new().unwrap();
        let store = MarkdownEntryStore::new(tmp.path().to_path_buf());

        store.clear_partition("ghost").await.unwrap();
    }
}