tawasuyu/brahman

1//! Persistencia del estado del WorkspaceManager.
2//!
3//! v1: sólo `WorkspaceSpec`s vivos. Los comandos (PIDs) NO se persisten —
4//! el kernel los mata al cerrar el daemon. Sólo la *intención declarada*
5//! (Workspaces creados con su spec) sobrevive a un reboot del daemon.
6
7use crate::WorkspaceManager;
8use serde::{Deserialize, Serialize};
9use shipote_card::{PipelineSpec, WorkspaceId, WorkspaceSpec};
10use std::path::{Path, PathBuf};
11use tracing::{info, warn};
12
13/// v2 agregó `saved_pipelines`. v1 lee con campo ausente como vacío.
14pub const SNAPSHOT_VERSION: u16 = 2;
15
16#[derive(Debug, Clone, Serialize, Deserialize)]
17pub struct ShipoteSnapshot {
18    pub version: u16,
19    pub timestamp_ms: u64,
20    pub workspaces: Vec<WorkspaceEntry>,
21    #[serde(default)]
22    pub saved_pipelines: Vec<PipelineEntry>,
23}
24
25#[derive(Debug, Clone, Serialize, Deserialize)]
26pub struct WorkspaceEntry {
27    pub id: WorkspaceId,
28    pub spec: WorkspaceSpec,
29}
30
31#[derive(Debug, Clone, Serialize, Deserialize)]
32pub struct PipelineEntry {
33    pub name: String,
34    pub spec: PipelineSpec,
35}
36
37impl ShipoteSnapshot {
38    pub fn write(&self, path: &Path) -> anyhow::Result<()> {
39        let bytes = serde_json::to_vec_pretty(self)?;
40        if let Some(parent) = path.parent() {
41            std::fs::create_dir_all(parent).ok();
42        }
43        let tmp = path.with_extension("tmp");
44        std::fs::write(&tmp, &bytes)?;
45        std::fs::rename(&tmp, path)?;
46        Ok(())
47    }
48
49    pub fn read(path: &Path) -> anyhow::Result<Self> {
50        let bytes = std::fs::read(path)?;
51        let snap: ShipoteSnapshot = serde_json::from_slice(&bytes)?;
52        // v1 y v2 son compatibles forward (v1 sin saved_pipelines lee como vec vacío).
53        if snap.version > SNAPSHOT_VERSION {
54            anyhow::bail!(
55                "snapshot version {} no soportada (esperada ≤ {})",
56                snap.version,
57                SNAPSHOT_VERSION
58            );
59        }
60        Ok(snap)
61    }
62}
63
64/// Path canónico del snapshot: `$XDG_STATE_HOME/shipote/state.json`,
65/// fallback `$HOME/.local/state/shipote/state.json`,
66/// fallback `/tmp/shipote-state-$UID.json`.
67pub fn default_snapshot_path() -> PathBuf {
68    if let Ok(state) = std::env::var("XDG_STATE_HOME") {
69        return PathBuf::from(state).join("shipote/state.json");
70    }
71    if let Ok(home) = std::env::var("HOME") {
72        return PathBuf::from(home).join(".local/state/shipote/state.json");
73    }
74    let uid = nix::unistd::getuid().as_raw();
75    PathBuf::from(format!("/tmp/shipote-state-{uid}.json"))
76}
77
78fn now_ms() -> u64 {
79    std::time::SystemTime::now()
80        .duration_since(std::time::UNIX_EPOCH)
81        .map(|d| d.as_millis() as u64)
82        .unwrap_or(0)
83}
84
85impl WorkspaceManager {
86    /// Toma snapshot del estado actual.
87    pub async fn snapshot(&self) -> ShipoteSnapshot {
88        let g = self.inner.lock().await;
89        let workspaces = g
90            .workspaces
91            .iter()
92            .map(|(id, ws)| WorkspaceEntry {
93                id: *id,
94                spec: ws.spec.clone(),
95            })
96            .collect();
97        let saved_pipelines = g
98            .saved_pipelines
99            .iter()
100            .map(|(name, spec)| PipelineEntry {
101                name: name.clone(),
102                spec: spec.clone(),
103            })
104            .collect();
105        ShipoteSnapshot {
106            version: SNAPSHOT_VERSION,
107            timestamp_ms: now_ms(),
108            workspaces,
109            saved_pipelines,
110        }
111    }
112
113    /// Escribe snapshot a disco.
114    pub async fn save_snapshot(&self, path: &Path) -> anyhow::Result<()> {
115        let snap = self.snapshot().await;
116        snap.write(path)?;
117        info!(path = %path.display(), workspaces = snap.workspaces.len(), "snapshot saved");
118        Ok(())
119    }
120
121    /// Carga snapshot desde disco y restaura los Workspaces.
122    /// Errores no-fatales (workspaces inválidos) se loguean y se saltan.
123    pub async fn restore_snapshot(self: &std::sync::Arc<Self>, path: &Path) -> anyhow::Result<usize> {
124        let snap = match ShipoteSnapshot::read(path) {
125            Ok(s) => s,
126            Err(e) => {
127                warn!(?e, path = %path.display(), "no snapshot — start fresh");
128                return Ok(0);
129            }
130        };
131        let mut restored = 0usize;
132        for entry in snap.workspaces {
133            // v2+: reusamos el id original así clients que tracking
134            // workspace_id no se rompen al restart.
135            let label = entry.spec.label.clone();
136            match self.create_with_id(entry.id, entry.spec).await {
137                Ok(_) => restored += 1,
138                Err(e) => warn!(?e, %label, "skipped workspace en restore"),
139            }
140        }
141        for entry in snap.saved_pipelines {
142            self.save_pipeline(entry.name, entry.spec).await;
143        }
144        info!(restored, "snapshot restored");
145        Ok(restored)
146    }
147}
148
149#[cfg(test)]
150mod tests {
151    use super::*;
152    use crate::WorkspaceManager;
153    use ente_incarnate::IncarnatorConfig;
154    use shipote_card::{ExitPolicy, WorkspaceSpec};
155    use std::sync::Arc;
156
157    fn sample_ws(label: &str) -> WorkspaceSpec {
158        WorkspaceSpec {
159            label: label.into(),
160            soma: Default::default(),
161            permissions: Default::default(),
162            ttl: None,
163            flow_dirs: vec![],
164            on_exit: ExitPolicy::Reap,
165        }
166    }
167
168    #[tokio::test]
169    async fn roundtrip_snapshot_preserves_ulids() {
170        let tmp = tempfile::tempdir().unwrap();
171        let path = tmp.path().join("state.json");
172
173        let mgr1 = Arc::new(WorkspaceManager::new(IncarnatorConfig::default()));
174        let (id1, _) = mgr1.create(sample_ws("a")).await.unwrap();
175        let (id2, _) = mgr1.create(sample_ws("b")).await.unwrap();
176        mgr1.save_snapshot(&path).await.unwrap();
177
178        let mgr2 = Arc::new(WorkspaceManager::new(IncarnatorConfig::default()));
179        let n = mgr2.restore_snapshot(&path).await.unwrap();
180        assert_eq!(n, 2);
181        let listed = mgr2.list().await;
182        let restored_ids: std::collections::HashSet<_> = listed.iter().map(|s| s.id).collect();
183        assert!(restored_ids.contains(&id1));
184        assert!(restored_ids.contains(&id2));
185    }
186
187    #[tokio::test]
188    async fn snapshot_includes_saved_pipelines() {
189        use shipote_card::{CommandRef, DiscernPolicy, PipelineSpec};
190        let tmp = tempfile::tempdir().unwrap();
191        let path = tmp.path().join("state.json");
192
193        let mgr1 = Arc::new(WorkspaceManager::new(IncarnatorConfig::default()));
194        let (ws_id, _) = mgr1.create(sample_ws("ws")).await.unwrap();
195        let spec = PipelineSpec {
196            label: "echo-cat".into(),
197            workspace: ws_id,
198            nodes: vec![CommandRef {
199                label: "n1".into(),
200                payload: brahman_card::Payload::Native {
201                    exec: "/bin/echo".into(),
202                    argv: vec!["hi".into()],
203                    envp: vec![],
204                },
205                soma: Default::default(),
206                flows: Default::default(),
207                supervision: brahman_card::Supervision::OneShot,
208            }],
209            edges: vec![],
210            discern: DiscernPolicy::default(),
211        };
212        mgr1.save_pipeline("daily".into(), spec).await;
213        mgr1.save_snapshot(&path).await.unwrap();
214
215        let mgr2 = Arc::new(WorkspaceManager::new(IncarnatorConfig::default()));
216        mgr2.restore_snapshot(&path).await.unwrap();
217        let saved = mgr2.list_saved_pipelines().await;
218        assert_eq!(saved, vec!["daily".to_string()]);
219        let got = mgr2.get_saved_pipeline("daily").await.expect("saved");
220        assert_eq!(got.label, "echo-cat");
221    }
222
223    #[test]
224    fn default_path_ends_with_state_json() {
225        let p = default_snapshot_path();
226        assert!(p.to_string_lossy().ends_with("state.json"));
227    }
228}