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 tokio::fs;
#[async_trait]
impl StateManager for FilesystemStateManager {
async fn save<T: Serialize + Send + Sync>(&self, node_id: &str, state: &T) -> Result<()> {
self.save_with_options(node_id, state, StateOptions::default()).await
}
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<T: DeserializeOwned + Send + Sync>(&self, node_id: &str) -> Result<Option<T>> {
if let Some(versioned) = self.load_versioned::<T>(node_id).await? {
Ok(Some(versioned.data))
} else {
Ok(None)
}
}
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 mut nodes = Vec::new();
let mut entries = fs::read_dir(&self.base_path).await.map_err(|e| Mecha10Error::IoError {
message: format!("Failed to read state directory: {}", 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();
if filename.ends_with(".state.json") {
let node_id = filename.trim_end_matches(".state.json").to_string();
nodes.push(node_id);
}
}
}
Ok(nodes)
}
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 mut entries = fs::read_dir(&snapshots_dir).await.map_err(|e| Mecha10Error::IoError {
message: format!("Failed to read snapshots directory: {}", 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();
if filename.ends_with(".meta.json") {
let json = fs::read_to_string(&path).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(¤t.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)
}
}