mecha10_core/state/

filesystem_impl.rs

//! `StateManager` implementation for [`super::FilesystemStateManager`].

use super::{FilesystemStateManager, StateManager, StateOptions, StateSnapshot, VersionedState};
use crate::error::{Mecha10Error, Result};
use crate::prelude::now_micros;
use async_trait::async_trait;
use serde::{de::DeserializeOwned, Serialize};
use std::path::Path;
use tokio::fs;

/// Lists filenames directly inside `dir` whose name ends with `suffix`.
///
/// Shared by `list_nodes` (scanning `base_path` for `*.state.json`) and `list_snapshots`
/// (scanning `snapshots/` for `*.meta.json`), which otherwise repeat the same
/// read-dir/next-entry/filename-suffix-filter loop. `dir_label` is folded into the I/O error
/// message so each call site keeps its own wording (e.g. "state directory" vs "snapshots
/// directory").
async fn filenames_with_suffix(dir: &Path, suffix: &str, dir_label: &str) -> Result<Vec<String>> {
    let mut names = Vec::new();
    let mut entries = fs::read_dir(dir).await.map_err(|e| Mecha10Error::IoError {
        message: format!("Failed to read {}: {}", dir_label, e),
        suggestion: "Check directory permissions".to_string(),
    })?;

    while let Some(entry) = entries.next_entry().await.map_err(|e| Mecha10Error::IoError {
        message: format!("Failed to read directory entry: {}", e),
        suggestion: "Check directory permissions".to_string(),
    })? {
        let path = entry.path();
        if let Some(filename) = path.file_name() {
            let filename = filename.to_string_lossy().into_owned();
            if filename.ends_with(suffix) {
                names.push(filename);
            }
        }
    }

    Ok(names)
}

#[async_trait]
impl StateManager for FilesystemStateManager {
    async fn save_with_options<T: Serialize + Send + Sync>(
        &self,
        node_id: &str,
        state: &T,
        options: StateOptions,
    ) -> Result<()> {
        let versioned = if let Some(ttl) = options.ttl {
            VersionedState::with_ttl(state, options.version, ttl)
        } else {
            VersionedState::new(state, options.version)
        };

        let path = self.state_path(node_id);

        if options.create_backup && path.exists() {
            let backup_path = path.with_extension("state.json.backup");
            fs::copy(&path, backup_path).await.ok();
        }

        let json = serde_json::to_string_pretty(&versioned).map_err(|e| Mecha10Error::SerializationError {
            message: format!("Failed to serialize state: {}", e),
            suggestion: "Ensure state type implements Serialize".to_string(),
        })?;

        fs::write(&path, json).await.map_err(|e| Mecha10Error::IoError {
            message: format!("Failed to write state file: {}", e),
            suggestion: "Check file permissions".to_string(),
        })?;

        Ok(())
    }

    async fn load_versioned<T: DeserializeOwned + Send + Sync>(
        &self,
        node_id: &str,
    ) -> Result<Option<VersionedState<T>>> {
        let path = self.state_path(node_id);

        if !path.exists() {
            return Ok(None);
        }

        let json = fs::read_to_string(&path).await.map_err(|e| Mecha10Error::IoError {
            message: format!("Failed to read state file: {}", e),
            suggestion: "Check file permissions".to_string(),
        })?;

        let versioned: VersionedState<T> =
            serde_json::from_str(&json).map_err(|e| Mecha10Error::SerializationError {
                message: format!("Failed to deserialize state: {}", e),
                suggestion: "State may be corrupted or type mismatch".to_string(),
            })?;

        if versioned.is_expired() {
            return Ok(None);
        }

        Ok(Some(versioned))
    }

    async fn delete(&self, node_id: &str) -> Result<()> {
        let path = self.state_path(node_id);

        if path.exists() {
            fs::remove_file(&path).await.map_err(|e| Mecha10Error::IoError {
                message: format!("Failed to delete state file: {}", e),
                suggestion: "Check file permissions".to_string(),
            })?;
        }

        Ok(())
    }

    async fn exists(&self, node_id: &str) -> Result<bool> {
        Ok(self.state_path(node_id).exists())
    }

    async fn list_nodes(&self) -> Result<Vec<String>> {
        let filenames = filenames_with_suffix(&self.base_path, ".state.json", "state directory").await?;

        Ok(filenames
            .into_iter()
            .map(|filename| filename.trim_end_matches(".state.json").to_string())
            .collect())
    }

    async fn size(&self, node_id: &str) -> Result<Option<usize>> {
        let path = self.state_path(node_id);

        if !path.exists() {
            return Ok(None);
        }

        let metadata = fs::metadata(&path).await.map_err(|e| Mecha10Error::IoError {
            message: format!("Failed to get file metadata: {}", e),
            suggestion: "Check file permissions".to_string(),
        })?;

        Ok(Some(metadata.len() as usize))
    }

    async fn create_snapshot(&self, node_id: &str, description: Option<String>) -> Result<StateSnapshot> {
        let state_path = self.state_path(node_id);

        if !state_path.exists() {
            return Err(Mecha10Error::Configuration(format!("No state for node: {}", node_id)));
        }

        let snapshot_id = uuid::Uuid::new_v4().to_string();

        let snapshots_dir = self.base_path.join("snapshots");
        fs::create_dir_all(&snapshots_dir)
            .await
            .map_err(|e| Mecha10Error::IoError {
                message: format!("Failed to create snapshots directory: {}", e),
                suggestion: "Check directory permissions".to_string(),
            })?;

        let snapshot_path = self.snapshot_path(&snapshot_id);
        fs::copy(&state_path, &snapshot_path)
            .await
            .map_err(|e| Mecha10Error::IoError {
                message: format!("Failed to create snapshot: {}", e),
                suggestion: "Check file permissions".to_string(),
            })?;

        let size_bytes = self.size(node_id).await?.unwrap_or(0);

        let snapshot = StateSnapshot {
            id: snapshot_id.clone(),
            node_id: node_id.to_string(),
            timestamp: now_micros(),
            description,
            size_bytes,
        };

        let meta_path = self.snapshot_meta_path(&snapshot_id);
        let meta_json = serde_json::to_string_pretty(&snapshot).unwrap();
        fs::write(&meta_path, meta_json)
            .await
            .map_err(|e| Mecha10Error::IoError {
                message: format!("Failed to write snapshot metadata: {}", e),
                suggestion: "Check file permissions".to_string(),
            })?;

        Ok(snapshot)
    }

    async fn restore_snapshot(&self, snapshot_id: &str) -> Result<()> {
        let snapshot_path = self.snapshot_path(snapshot_id);
        let meta_path = self.snapshot_meta_path(snapshot_id);

        if !snapshot_path.exists() || !meta_path.exists() {
            return Err(Mecha10Error::Configuration(format!(
                "Snapshot not found: {}",
                snapshot_id
            )));
        }

        let meta_json = fs::read_to_string(&meta_path)
            .await
            .map_err(|e| Mecha10Error::IoError {
                message: format!("Failed to read snapshot metadata: {}", e),
                suggestion: "Check file permissions".to_string(),
            })?;

        let snapshot: StateSnapshot =
            serde_json::from_str(&meta_json).map_err(|e| Mecha10Error::SerializationError {
                message: format!("Failed to deserialize snapshot metadata: {}", e),
                suggestion: "Snapshot metadata may be corrupted".to_string(),
            })?;

        let state_path = self.state_path(&snapshot.node_id);
        fs::copy(&snapshot_path, &state_path)
            .await
            .map_err(|e| Mecha10Error::IoError {
                message: format!("Failed to restore snapshot: {}", e),
                suggestion: "Check file permissions".to_string(),
            })?;

        Ok(())
    }

    async fn list_snapshots(&self, node_id: &str) -> Result<Vec<StateSnapshot>> {
        let snapshots_dir = self.base_path.join("snapshots");

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

        let mut snapshots = Vec::new();
        let filenames = filenames_with_suffix(&snapshots_dir, ".meta.json", "snapshots directory").await?;

        for filename in filenames {
            let json = fs::read_to_string(snapshots_dir.join(filename)).await.ok();
            if let Some(json) = json {
                if let Ok(snapshot) = serde_json::from_str::<StateSnapshot>(&json) {
                    if snapshot.node_id == node_id {
                        snapshots.push(snapshot);
                    }
                }
            }
        }

        snapshots.sort_by_key(|s| s.timestamp);
        snapshots.reverse();

        Ok(snapshots)
    }

    async fn delete_snapshot(&self, snapshot_id: &str) -> Result<()> {
        let snapshot_path = self.snapshot_path(snapshot_id);
        let meta_path = self.snapshot_meta_path(snapshot_id);

        if snapshot_path.exists() {
            fs::remove_file(&snapshot_path).await.ok();
        }

        if meta_path.exists() {
            fs::remove_file(&meta_path).await.ok();
        }

        Ok(())
    }

    async fn compare_and_swap<T: Serialize + DeserializeOwned + Send + Sync>(
        &self,
        node_id: &str,
        expected: &T,
        new: &T,
    ) -> Result<bool> {
        let path = self.state_path(node_id);

        if !path.exists() {
            return Ok(false);
        }

        let json = fs::read_to_string(&path).await.map_err(|e| Mecha10Error::IoError {
            message: format!("Failed to read state file: {}", e),
            suggestion: "Check file permissions".to_string(),
        })?;

        let current: VersionedState<T> = serde_json::from_str(&json).map_err(|e| Mecha10Error::SerializationError {
            message: format!("Failed to deserialize for CAS: {}", e),
            suggestion: "State may be corrupted".to_string(),
        })?;

        let expected_json = serde_json::to_string(expected).unwrap();
        let current_json = serde_json::to_string(&current.data).unwrap();

        if expected_json == current_json {
            let versioned = VersionedState::new(new, current.version + 1);
            let new_json = serde_json::to_string_pretty(&versioned).unwrap();
            fs::write(&path, new_json).await.map_err(|e| Mecha10Error::IoError {
                message: format!("Failed to write state file: {}", e),
                suggestion: "Check file permissions".to_string(),
            })?;
            Ok(true)
        } else {
            Ok(false)
        }
    }

    async fn clear_expired(&self) -> Result<usize> {
        let nodes = self.list_nodes().await?;
        let mut count = 0;

        for node_id in nodes {
            if let Ok(Some(versioned)) = self.load_versioned::<serde_json::Value>(&node_id).await {
                if versioned.is_expired() {
                    self.delete(&node_id).await?;
                    count += 1;
                }
            }
        }

        Ok(count)
    }
}

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

    #[tokio::test]
    async fn filenames_with_suffix_returns_only_matching_names() {
        let dir = tempfile::tempdir().expect("tempdir");
        fs::write(dir.path().join("a.state.json"), b"{}").await.unwrap();
        fs::write(dir.path().join("b.state.json"), b"{}").await.unwrap();
        fs::write(dir.path().join("c.meta.json"), b"{}").await.unwrap();
        fs::write(dir.path().join("d.txt"), b"ignored").await.unwrap();

        let mut names = filenames_with_suffix(dir.path(), ".state.json", "state directory")
            .await
            .unwrap();
        names.sort();

        assert_eq!(names, vec!["a.state.json".to_string(), "b.state.json".to_string()]);
    }

    #[tokio::test]
    async fn filenames_with_suffix_returns_empty_for_empty_dir() {
        let dir = tempfile::tempdir().expect("tempdir");

        let names = filenames_with_suffix(dir.path(), ".meta.json", "snapshots directory")
            .await
            .unwrap();

        assert!(names.is_empty());
    }

    #[tokio::test]
    async fn filenames_with_suffix_errors_on_missing_dir() {
        let dir = tempfile::tempdir().expect("tempdir");
        let missing = dir.path().join("does-not-exist");

        let err = filenames_with_suffix(&missing, ".state.json", "state directory")
            .await
            .unwrap_err();

        assert!(err.to_string().contains("state directory"));
    }
}