cfgd_core/state/
mod.rs

1use std::path::{Path, PathBuf};
2
3use rusqlite::Connection;
4
5use crate::errors::{Result, StateError};
6
7mod applies;
8mod backups;
9mod compliance;
10mod decisions;
11mod drift;
12mod journal;
13mod managed;
14mod modules;
15mod pending_config;
16mod sources;
17mod types;
18
19pub use pending_config::{
20    clear_pending_server_config, load_pending_server_config, save_pending_server_config,
21};
22pub use types::{
23    ApplyRecord, ApplyStatus, ComplianceHistoryRow, ConfigSourceRecord, DriftEvent,
24    FileBackupRecord, JournalEntry, ManagedResource, ModuleFileRecord, ModuleStateRecord,
25    PendingDecision, SourceConfigHash, SourceConflictRecord,
26};
27
28const MIGRATIONS: &[&str] = &[
29    "CREATE TABLE IF NOT EXISTS applies (
30        id INTEGER PRIMARY KEY AUTOINCREMENT,
31        timestamp TEXT NOT NULL,
32        profile TEXT NOT NULL,
33        plan_hash TEXT NOT NULL,
34        status TEXT NOT NULL,
35        summary TEXT
36    );
37
38    CREATE TABLE IF NOT EXISTS drift_events (
39        id INTEGER PRIMARY KEY AUTOINCREMENT,
40        timestamp TEXT NOT NULL,
41        resource_type TEXT NOT NULL,
42        resource_id TEXT NOT NULL,
43        expected TEXT,
44        actual TEXT,
45        source TEXT NOT NULL DEFAULT 'local',
46        resolved_by INTEGER,
47        FOREIGN KEY (resolved_by) REFERENCES applies(id)
48    );
49
50    CREATE TABLE IF NOT EXISTS managed_resources (
51        id INTEGER PRIMARY KEY AUTOINCREMENT,
52        resource_type TEXT NOT NULL,
53        resource_id TEXT NOT NULL,
54        source TEXT NOT NULL DEFAULT 'local',
55        last_hash TEXT,
56        last_applied INTEGER,
57        UNIQUE(resource_type, resource_id),
58        FOREIGN KEY (last_applied) REFERENCES applies(id)
59    );
60
61    CREATE TABLE IF NOT EXISTS config_sources (
62        id INTEGER PRIMARY KEY AUTOINCREMENT,
63        name TEXT NOT NULL UNIQUE,
64        origin_url TEXT NOT NULL,
65        origin_branch TEXT NOT NULL DEFAULT 'main',
66        last_fetched TEXT,
67        last_commit TEXT,
68        source_version TEXT,
69        pinned_version TEXT,
70        status TEXT NOT NULL DEFAULT 'active'
71    );
72
73    CREATE TABLE IF NOT EXISTS source_applies (
74        id INTEGER PRIMARY KEY AUTOINCREMENT,
75        source_id INTEGER NOT NULL,
76        apply_id INTEGER NOT NULL,
77        source_commit TEXT NOT NULL,
78        FOREIGN KEY (source_id) REFERENCES config_sources(id),
79        FOREIGN KEY (apply_id) REFERENCES applies(id)
80    );
81
82    CREATE TABLE IF NOT EXISTS source_conflicts (
83        id INTEGER PRIMARY KEY AUTOINCREMENT,
84        timestamp TEXT NOT NULL,
85        source_name TEXT NOT NULL,
86        resource_type TEXT NOT NULL,
87        resource_id TEXT NOT NULL,
88        resolution TEXT NOT NULL,
89        detail TEXT
90    );
91
92    CREATE TABLE IF NOT EXISTS pending_decisions (
93        id          INTEGER PRIMARY KEY AUTOINCREMENT,
94        source      TEXT NOT NULL,
95        resource    TEXT NOT NULL,
96        tier        TEXT NOT NULL,
97        action      TEXT NOT NULL,
98        summary     TEXT NOT NULL,
99        created_at  TEXT NOT NULL,
100        resolved_at TEXT,
101        resolution  TEXT
102    );
103
104    CREATE UNIQUE INDEX IF NOT EXISTS idx_pending_decisions_source_resource
105        ON pending_decisions (source, resource)
106        WHERE resolved_at IS NULL;
107
108    CREATE TABLE IF NOT EXISTS source_config_hashes (
109        source      TEXT PRIMARY KEY,
110        config_hash TEXT NOT NULL,
111        merged_at   TEXT NOT NULL
112    );
113
114    CREATE TABLE IF NOT EXISTS module_state (
115        id              INTEGER PRIMARY KEY AUTOINCREMENT,
116        module_name     TEXT NOT NULL UNIQUE,
117        installed_at    TEXT NOT NULL,
118        last_applied    INTEGER,
119        packages_hash   TEXT NOT NULL,
120        files_hash      TEXT NOT NULL,
121        git_sources     TEXT,
122        status          TEXT NOT NULL DEFAULT 'installed',
123        FOREIGN KEY (last_applied) REFERENCES applies(id)
124    );
125
126    CREATE TABLE IF NOT EXISTS schema_version (
127        version INTEGER NOT NULL
128    );
129
130    INSERT INTO schema_version (version) VALUES (0);",
131    // Migration 2: File safety — backup store, transaction journal, module file manifest
132    "CREATE TABLE IF NOT EXISTS file_backups (
133        id              INTEGER PRIMARY KEY AUTOINCREMENT,
134        apply_id        INTEGER NOT NULL,
135        file_path       TEXT NOT NULL,
136        content_hash    TEXT NOT NULL,
137        content         BLOB NOT NULL,
138        permissions     INTEGER,
139        was_symlink     INTEGER NOT NULL DEFAULT 0,
140        symlink_target  TEXT,
141        oversized       INTEGER NOT NULL DEFAULT 0,
142        backed_up_at    TEXT NOT NULL,
143        FOREIGN KEY (apply_id) REFERENCES applies(id)
144    );
145
146    CREATE INDEX IF NOT EXISTS idx_file_backups_apply ON file_backups (apply_id);
147    CREATE INDEX IF NOT EXISTS idx_file_backups_path ON file_backups (file_path);
148
149    CREATE TABLE IF NOT EXISTS apply_journal (
150        id              INTEGER PRIMARY KEY AUTOINCREMENT,
151        apply_id        INTEGER NOT NULL,
152        action_index    INTEGER NOT NULL,
153        phase           TEXT NOT NULL,
154        action_type     TEXT NOT NULL,
155        resource_id     TEXT NOT NULL,
156        pre_state       TEXT,
157        post_state      TEXT,
158        status          TEXT NOT NULL DEFAULT 'pending',
159        error           TEXT,
160        started_at      TEXT NOT NULL,
161        completed_at    TEXT,
162        FOREIGN KEY (apply_id) REFERENCES applies(id)
163    );
164
165    CREATE INDEX IF NOT EXISTS idx_apply_journal_apply ON apply_journal (apply_id);
166
167    CREATE TABLE IF NOT EXISTS module_file_manifest (
168        id              INTEGER PRIMARY KEY AUTOINCREMENT,
169        module_name     TEXT NOT NULL,
170        file_path       TEXT NOT NULL,
171        content_hash    TEXT NOT NULL,
172        strategy        TEXT NOT NULL,
173        last_applied    INTEGER,
174        UNIQUE(module_name, file_path),
175        FOREIGN KEY (last_applied) REFERENCES applies(id)
176    );
177
178    CREATE INDEX IF NOT EXISTS idx_module_file_manifest_module ON module_file_manifest (module_name);",
179    // Migration 3: Script output capture — store stdout/stderr from script actions
180    "ALTER TABLE apply_journal ADD COLUMN script_output TEXT;",
181    // Migration 4: Compliance snapshots — periodic machine state snapshots
182    "CREATE TABLE IF NOT EXISTS compliance_snapshots (
183        id INTEGER PRIMARY KEY AUTOINCREMENT,
184        timestamp TEXT NOT NULL,
185        content_hash TEXT NOT NULL,
186        snapshot_json TEXT NOT NULL,
187        summary_compliant INTEGER NOT NULL,
188        summary_warning INTEGER NOT NULL,
189        summary_violation INTEGER NOT NULL
190    );",
191];
192
193/// SQLite-backed state store for cfgd.
194pub struct StateStore {
195    pub(in crate::state) conn: Connection,
196}
197
198impl StateStore {
199    /// Open or create a state store at the default location.
200    /// Uses `~/.local/share/cfgd/state.db`.
201    pub fn open_default() -> Result<Self> {
202        let data_dir = default_state_dir()?;
203        std::fs::create_dir_all(&data_dir).map_err(|_| StateError::DirectoryNotWritable {
204            path: data_dir.clone(),
205        })?;
206        let db_path = data_dir.join("state.db");
207        Self::open(&db_path)
208    }
209
210    /// Open or create a state store at the given path.
211    pub fn open(path: &Path) -> Result<Self> {
212        let conn = Connection::open(path)?;
213        conn.execute_batch("PRAGMA journal_mode=WAL; PRAGMA foreign_keys=ON;")?;
214        conn.busy_timeout(std::time::Duration::from_secs(5))?;
215
216        let mut store = Self { conn };
217        store.run_migrations()?;
218        Ok(store)
219    }
220
221    /// Create an in-memory state store (for testing).
222    pub fn open_in_memory() -> Result<Self> {
223        let conn = Connection::open_in_memory()?;
224        conn.execute_batch("PRAGMA foreign_keys=ON;")?;
225
226        let mut store = Self { conn };
227        store.run_migrations()?;
228        Ok(store)
229    }
230
231    fn run_migrations(&mut self) -> Result<()> {
232        // Use EXCLUSIVE transaction to serialize concurrent migration attempts
233        // (e.g. parallel cargo test processes sharing the same state DB).
234        self.conn
235            .execute_batch("BEGIN EXCLUSIVE")
236            .map_err(|e| StateError::MigrationFailed {
237                message: format!("failed to acquire migration lock: {e}"),
238            })?;
239
240        let current_version = self.schema_version();
241
242        for (i, migration) in MIGRATIONS.iter().enumerate() {
243            if i >= current_version {
244                self.conn.execute_batch(migration).map_err(|e| {
245                    if let Err(rb) = self.conn.execute_batch("ROLLBACK") {
246                        tracing::error!("rollback after migration {i} failure also failed: {rb}");
247                    }
248                    StateError::MigrationFailed {
249                        message: format!("migration {}: {}", i, e),
250                    }
251                })?;
252                // Set version automatically — no hardcoded UPDATE in migration SQL
253                let new_version = (i + 1) as i64;
254                self.conn
255                    .execute(
256                        "UPDATE schema_version SET version = ?1",
257                        rusqlite::params![new_version],
258                    )
259                    .map_err(|e| {
260                        if let Err(rb) = self.conn.execute_batch("ROLLBACK") {
261                            tracing::error!(
262                                "rollback after schema_version update failure also failed: {rb}"
263                            );
264                        }
265                        StateError::MigrationFailed {
266                            message: format!("migration {}: failed to update version: {}", i, e),
267                        }
268                    })?;
269            }
270        }
271
272        self.conn
273            .execute_batch("COMMIT")
274            .map_err(|e| StateError::MigrationFailed {
275                message: format!("failed to commit migrations: {e}"),
276            })?;
277
278        Ok(())
279    }
280
281    fn schema_version(&self) -> usize {
282        self.conn
283            .query_row("SELECT version FROM schema_version", [], |row| {
284                row.get::<_, i64>(0)
285            })
286            .map(|v| v as usize)
287            .unwrap_or(0)
288    }
289}
290
291/// Compute SHA256 hash of a serializable plan for deduplication.
292pub fn plan_hash(data: &str) -> String {
293    crate::sha256_hex(data.as_bytes())
294}
295
296pub fn default_state_dir() -> Result<PathBuf> {
297    if let Ok(dir) = std::env::var("CFGD_STATE_DIR") {
298        return Ok(PathBuf::from(dir));
299    }
300    let base = directories::BaseDirs::new().ok_or_else(|| StateError::DirectoryNotWritable {
301        path: PathBuf::from("~/.local/share/cfgd"),
302    })?;
303    Ok(base.data_local_dir().join("cfgd"))
304}
305
306#[cfg(test)]
307mod tests;
cfgd_core/state/mod.rs

cfgd_core/state/
mod.rs
