foxtive_supervisor/persistence/
fs.rs

1use super::{PersistedTaskState, TaskStateStore};
2use std::path::{Path, PathBuf};
3use tokio::fs;
4use tokio::io::{AsyncReadExt, AsyncWriteExt};
5
6/// A file system-based implementation of TaskStateStore.
7/// Each task's state is stored as a JSON file in a specified directory.
8pub struct FsStateStore {
9    base_path: PathBuf,
10}
11
12impl FsStateStore {
13    pub async fn new(base_path: impl AsRef<Path>) -> anyhow::Result<Self> {
14        let path = base_path.as_ref().to_path_buf();
15        fs::create_dir_all(&path).await?;
16        Ok(Self { base_path: path })
17    }
18
19    fn state_file_path(&self, task_id: &str) -> PathBuf {
20        self.base_path.join(format!("{}.json", task_id))
21    }
22}
23
24#[async_trait::async_trait]
25impl TaskStateStore for FsStateStore {
26    async fn save_state(&self, state: PersistedTaskState) -> anyhow::Result<()> {
27        let file_path = self.state_file_path(&state.task_id);
28        let json = serde_json::to_string_pretty(&state)?;
29        let mut file = fs::File::create(&file_path).await?;
30        file.write_all(json.as_bytes()).await?;
31        file.sync_all().await?; // Ensure data is written to disk
32        Ok(())
33    }
34
35    async fn load_state(&self, task_id: &str) -> anyhow::Result<Option<PersistedTaskState>> {
36        let file_path = self.state_file_path(task_id);
37        if !file_path.exists() {
38            return Ok(None);
39        }
40
41        let mut file = fs::File::open(&file_path).await?;
42        let mut json = String::new();
43        file.read_to_string(&mut json).await?;
44
45        if json.is_empty() {
46            return Ok(None);
47        }
48
49        let state: PersistedTaskState = serde_json::from_str(&json)?;
50        Ok(Some(state))
51    }
52
53    async fn load_all_states(&self) -> anyhow::Result<Vec<PersistedTaskState>> {
54        let mut states = Vec::new();
55        let mut entries = fs::read_dir(&self.base_path).await?;
56
57        while let Some(entry) = entries.next_entry().await? {
58            let path = entry.path();
59            if path.is_file() && path.extension().is_some_and(|ext| ext == "json") {
60                let mut file = fs::File::open(&path).await?;
61                let mut json = String::new();
62                file.read_to_string(&mut json).await?;
63
64                if json.is_empty() {
65                    continue;
66                }
67
68                if let Ok(state) = serde_json::from_str::<PersistedTaskState>(&json) {
69                    states.push(state);
70                } else {
71                    // Log error for malformed state files, but don't fail the whole operation
72                    eprintln!("Warning: Could not parse state file: {:?}", path);
73                }
74            }
75        }
76        Ok(states)
77    }
78
79    async fn delete_state(&self, task_id: &str) -> anyhow::Result<()> {
80        let file_path = self.state_file_path(task_id);
81        if file_path.exists() {
82            fs::remove_file(&file_path).await?;
83        }
84        Ok(())
85    }
86}
foxtive_supervisor/persistence/fs.rs

foxtive_supervisor/persistence/
fs.rs
