dejavu/store/
db.rs

1//! SQLite connection wrapper: open, migrate, insert, and query.
2//!
3//! Concurrency: multiple `dejavu run` processes (e.g. a parallel agent or
4//! `make -j`) open the same DB. WAL + a 5s busy timeout absorb contention; the
5//! runtime treats any write failure as a reason to fall back to raw passthrough
6//! rather than change the command result.
7
8use super::models::{RunRecord, SessionRecord};
9use super::schema;
10use crate::error::StoreError;
11use rusqlite::{named_params, params, Connection, OptionalExtension};
12use std::path::Path;
13use std::time::Duration;
14
15pub struct Db {
16    pub conn: Connection,
17}
18
19/// Aggregate stats over a repo's runs (spec §17.6).
20#[derive(Debug, Clone, Default)]
21pub struct StatsAgg {
22    pub runs_captured: i64,
23    pub optimized: i64,
24    pub unchanged: i64,
25    pub small_delta: i64,
26    pub large_delta: i64,
27    pub passthrough: i64,
28    pub raw_tokens: i64,
29    pub emitted_tokens: i64,
30    pub saved_tokens: i64,
31    pub full_output_requested: i64,
32    pub internal_error: i64,
33    pub avg_overhead_ms: f64,
34}
35
36impl Db {
37    /// Read-only open for reporting over OTHER repos' databases (`stats --all`):
38    /// no migration, no WAL side files, works on read-only caches, and never
39    /// touches a DB that a live agent session is writing.
40    pub fn open_read_only(path: &Path) -> Result<Db, StoreError> {
41        let conn = Connection::open_with_flags(
42            path,
43            rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY | rusqlite::OpenFlags::SQLITE_OPEN_NO_MUTEX,
44        )?;
45        conn.busy_timeout(Duration::from_millis(5000))?;
46        Ok(Db { conn })
47    }
48
49    pub fn open(path: &Path) -> Result<Db, StoreError> {
50        if let Some(parent) = path.parent() {
51            std::fs::create_dir_all(parent)?;
52        }
53        let conn = Connection::open(path)?;
54        // execute_batch tolerates the result row that `PRAGMA journal_mode`
55        // returns; pragma_update would error on it.
56        conn.execute_batch(
57            "PRAGMA journal_mode=WAL;\
58             PRAGMA synchronous=NORMAL;\
59             PRAGMA foreign_keys=ON;",
60        )?;
61        conn.busy_timeout(Duration::from_millis(5000))?;
62        let db = Db { conn };
63        db.migrate()?;
64        Ok(db)
65    }
66
67    fn migrate(&self) -> Result<(), StoreError> {
68        let version: i64 = self
69            .conn
70            .query_row("PRAGMA user_version", [], |row| row.get(0))?;
71        if version < schema::SCHEMA_VERSION {
72            self.conn.execute_batch(schema::SCHEMA_V1)?;
73            self.conn
74                .execute_batch(&format!("PRAGMA user_version={};", schema::SCHEMA_VERSION))?;
75        }
76        Ok(())
77    }
78
79    pub fn insert_run(&self, r: &RunRecord) -> Result<(), StoreError> {
80        self.conn.execute(
81            "INSERT INTO runs (
82                id, session_id, created_at, repo_root, cwd, shim_name, argv_json,
83                command_original, command_family, command_key, classification,
84                exit_code, duration_ms, overhead_ms, stdout_path, stderr_path,
85                normalized_path, raw_stdout_bytes, raw_stderr_bytes, raw_total_bytes,
86                emitted_bytes, estimated_raw_tokens, estimated_emitted_tokens,
87                estimated_saved_tokens, normalized_hash, stdout_hash, stderr_hash,
88                git_head, git_worktree_hash, comparison_base_run_id, comparison_result,
89                summary, full_output_requested, internal_error
90            ) VALUES (
91                :id, :session_id, :created_at, :repo_root, :cwd, :shim_name, :argv_json,
92                :command_original, :command_family, :command_key, :classification,
93                :exit_code, :duration_ms, :overhead_ms, :stdout_path, :stderr_path,
94                :normalized_path, :raw_stdout_bytes, :raw_stderr_bytes, :raw_total_bytes,
95                :emitted_bytes, :estimated_raw_tokens, :estimated_emitted_tokens,
96                :estimated_saved_tokens, :normalized_hash, :stdout_hash, :stderr_hash,
97                :git_head, :git_worktree_hash, :comparison_base_run_id, :comparison_result,
98                :summary, :full_output_requested, :internal_error
99            )",
100            named_params! {
101                ":id": r.id,
102                ":session_id": r.session_id,
103                ":created_at": r.created_at,
104                ":repo_root": r.repo_root,
105                ":cwd": r.cwd,
106                ":shim_name": r.shim_name,
107                ":argv_json": r.argv_json,
108                ":command_original": r.command_original,
109                ":command_family": r.command_family,
110                ":command_key": r.command_key,
111                ":classification": r.classification,
112                ":exit_code": r.exit_code,
113                ":duration_ms": r.duration_ms,
114                ":overhead_ms": r.overhead_ms,
115                ":stdout_path": r.stdout_path,
116                ":stderr_path": r.stderr_path,
117                ":normalized_path": r.normalized_path,
118                ":raw_stdout_bytes": r.raw_stdout_bytes,
119                ":raw_stderr_bytes": r.raw_stderr_bytes,
120                ":raw_total_bytes": r.raw_total_bytes,
121                ":emitted_bytes": r.emitted_bytes,
122                ":estimated_raw_tokens": r.estimated_raw_tokens,
123                ":estimated_emitted_tokens": r.estimated_emitted_tokens,
124                ":estimated_saved_tokens": r.estimated_saved_tokens,
125                ":normalized_hash": r.normalized_hash,
126                ":stdout_hash": r.stdout_hash,
127                ":stderr_hash": r.stderr_hash,
128                ":git_head": r.git_head,
129                ":git_worktree_hash": r.git_worktree_hash,
130                ":comparison_base_run_id": r.comparison_base_run_id,
131                ":comparison_result": r.comparison_result,
132                ":summary": r.summary,
133                ":full_output_requested": r.full_output_requested,
134                ":internal_error": r.internal_error,
135            },
136        )?;
137        Ok(())
138    }
139
140    /// Resolve a run by exact id, then by unique short-id prefix (newest wins).
141    pub fn get_run(&self, target: &str) -> Result<Option<RunRecord>, StoreError> {
142        if let Some(run) = self
143            .conn
144            .query_row("SELECT * FROM runs WHERE id = ?1", [target], |row| {
145                RunRecord::from_row(row)
146            })
147            .optional()?
148        {
149            return Ok(Some(run));
150        }
151        let pattern = format!("{target}%");
152        let run = self
153            .conn
154            .query_row(
155                "SELECT * FROM runs WHERE id LIKE ?1 ORDER BY created_at DESC, rowid DESC LIMIT 1",
156                [pattern],
157                RunRecord::from_row,
158            )
159            .optional()?;
160        Ok(run)
161    }
162
163    /// How many runs share a short-id prefix (for ambiguity detection).
164    pub fn count_prefix(&self, prefix: &str) -> Result<i64, StoreError> {
165        let pattern = format!("{prefix}%");
166        let count: i64 = self.conn.query_row(
167            "SELECT COUNT(*) FROM runs WHERE id LIKE ?1",
168            [pattern],
169            |r| r.get(0),
170        )?;
171        Ok(count)
172    }
173
174    pub fn latest_run(&self, repo_root: &str) -> Result<Option<RunRecord>, StoreError> {
175        let run = self
176            .conn
177            .query_row(
178                "SELECT * FROM runs WHERE repo_root = ?1 ORDER BY created_at DESC, rowid DESC LIMIT 1",
179                [repo_root],
180                RunRecord::from_row,
181            )
182            .optional()?;
183        Ok(run)
184    }
185
186    /// The most recent comparable prior run (spec §12, hybrid decision #3):
187    /// matched on `(repo_root, cwd, command_family, command_key)`. Git state is
188    /// NOT part of the match.
189    pub fn find_comparable_prior(
190        &self,
191        repo_root: &str,
192        cwd: &str,
193        command_family: &str,
194        command_key: &str,
195        before_created_at: &str,
196    ) -> Result<Option<RunRecord>, StoreError> {
197        let run = self
198            .conn
199            .query_row(
200                "SELECT * FROM runs
201                 WHERE repo_root = :repo_root AND cwd = :cwd
202                   AND command_family = :family AND command_key = :key
203                   AND created_at < :before
204                   AND classification != 'internal_error'
205                 ORDER BY created_at DESC, rowid DESC LIMIT 1",
206                named_params! {
207                    ":repo_root": repo_root,
208                    ":cwd": cwd,
209                    ":family": command_family,
210                    ":key": command_key,
211                    ":before": before_created_at,
212                },
213                RunRecord::from_row,
214            )
215            .optional()?;
216        Ok(run)
217    }
218
219    pub fn upsert_session_start(&self, s: &SessionRecord) -> Result<(), StoreError> {
220        self.conn.execute(
221            "INSERT INTO sessions (id, created_at, ended_at, repo_root, agent_command,
222                 raw_tokens_total, emitted_tokens_total, saved_tokens_total)
223             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)
224             ON CONFLICT(id) DO NOTHING",
225            params![
226                s.id,
227                s.created_at,
228                s.ended_at,
229                s.repo_root,
230                s.agent_command,
231                s.raw_tokens_total,
232                s.emitted_tokens_total,
233                s.saved_tokens_total,
234            ],
235        )?;
236        Ok(())
237    }
238
239    pub fn accumulate_session_tokens(
240        &self,
241        session_id: &str,
242        raw: i64,
243        emitted: i64,
244        saved: i64,
245    ) -> Result<(), StoreError> {
246        self.conn.execute(
247            "UPDATE sessions SET
248                raw_tokens_total = raw_tokens_total + ?2,
249                emitted_tokens_total = emitted_tokens_total + ?3,
250                saved_tokens_total = saved_tokens_total + ?4
251             WHERE id = ?1",
252            params![session_id, raw, emitted, saved],
253        )?;
254        Ok(())
255    }
256
257    pub fn mark_full_output_requested(&self, run_id: &str) -> Result<(), StoreError> {
258        self.conn.execute(
259            "UPDATE runs SET full_output_requested = 1
260             WHERE id = ?1 AND full_output_requested = 0",
261            [run_id],
262        )?;
263        Ok(())
264    }
265
266    /// Runs older than `cutoff` (RFC3339), for log-file cleanup.
267    pub fn runs_before(&self, repo_root: &str, cutoff: &str) -> Result<Vec<RunRecord>, StoreError> {
268        let mut stmt = self.conn.prepare(
269            "SELECT * FROM runs WHERE repo_root = ?1 AND created_at < ?2 ORDER BY created_at",
270        )?;
271        let rows = stmt.query_map(params![repo_root, cutoff], RunRecord::from_row)?;
272        let mut out = Vec::new();
273        for row in rows {
274            out.push(row?);
275        }
276        Ok(out)
277    }
278
279    /// Delete run rows older than `cutoff`. Returns the number deleted.
280    pub fn delete_runs_before(&self, repo_root: &str, cutoff: &str) -> Result<usize, StoreError> {
281        let n = self.conn.execute(
282            "DELETE FROM runs WHERE repo_root = ?1 AND created_at < ?2",
283            params![repo_root, cutoff],
284        )?;
285        Ok(n)
286    }
287
288    /// Aggregate run stats; `repo_root: None` aggregates the whole database.
289    pub fn aggregate_stats(&self, repo_root: Option<&str>) -> Result<StatsAgg, StoreError> {
290        let agg = self.conn.query_row(
291            "SELECT
292                COUNT(*),
293                COALESCE(SUM(CASE WHEN classification IN
294                    ('first_seen','unchanged','small_delta','large_delta') THEN 1 ELSE 0 END), 0),
295                COALESCE(SUM(CASE WHEN classification='unchanged' THEN 1 ELSE 0 END), 0),
296                COALESCE(SUM(CASE WHEN classification='small_delta' THEN 1 ELSE 0 END), 0),
297                COALESCE(SUM(CASE WHEN classification='large_delta' THEN 1 ELSE 0 END), 0),
298                COALESCE(SUM(CASE WHEN classification='passthrough' THEN 1 ELSE 0 END), 0),
299                COALESCE(SUM(estimated_raw_tokens), 0),
300                COALESCE(SUM(estimated_emitted_tokens), 0),
301                COALESCE(SUM(estimated_saved_tokens), 0),
302                COALESCE(SUM(full_output_requested), 0),
303                COALESCE(SUM(CASE WHEN classification='internal_error' THEN 1 ELSE 0 END), 0),
304                COALESCE(AVG(overhead_ms), 0.0)
305             FROM runs WHERE (?1 IS NULL OR repo_root = ?1)",
306            params![repo_root],
307            |row| {
308                Ok(StatsAgg {
309                    runs_captured: row.get(0)?,
310                    optimized: row.get(1)?,
311                    unchanged: row.get(2)?,
312                    small_delta: row.get(3)?,
313                    large_delta: row.get(4)?,
314                    passthrough: row.get(5)?,
315                    raw_tokens: row.get(6)?,
316                    emitted_tokens: row.get(7)?,
317                    saved_tokens: row.get(8)?,
318                    full_output_requested: row.get(9)?,
319                    internal_error: row.get(10)?,
320                    avg_overhead_ms: row.get(11)?,
321                })
322            },
323        )?;
324        Ok(agg)
325    }
326
327    /// Top-N `(display, saved_tokens)` grouped by command, most savings first.
328    /// Top saving commands; `limit: None` returns them all (SQLite `LIMIT -1`).
329    pub fn top_savings(
330        &self,
331        repo_root: Option<&str>,
332        limit: Option<usize>,
333    ) -> Result<Vec<(String, i64)>, StoreError> {
334        let mut stmt = self.conn.prepare(
335            "SELECT command_original, SUM(estimated_saved_tokens) AS saved
336             FROM runs
337             WHERE (?1 IS NULL OR repo_root = ?1) AND estimated_saved_tokens > 0
338             GROUP BY command_key
339             ORDER BY saved DESC LIMIT ?2",
340        )?;
341        let limit = limit.map(|l| l as i64).unwrap_or(-1);
342        let rows = stmt.query_map(params![repo_root, limit], |row| {
343            Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)?))
344        })?;
345        let mut out = Vec::new();
346        for row in rows {
347            out.push(row?);
348        }
349        Ok(out)
350    }
351
352    /// Distinct repo roots recorded in this database (normally one per cache).
353    pub fn repo_roots(&self) -> Result<Vec<String>, StoreError> {
354        let mut stmt = self.conn.prepare(
355            "SELECT repo_root FROM (
356                SELECT DISTINCT repo_root FROM runs
357                UNION
358                SELECT DISTINCT repo_root FROM sessions
359             )
360             ORDER BY repo_root",
361        )?;
362        let rows = stmt.query_map([], |row| row.get::<_, String>(0))?;
363        let mut out = Vec::new();
364        for row in rows {
365            out.push(row?);
366        }
367        Ok(out)
368    }
369
370    pub fn session_count(&self, repo_root: &str) -> Result<i64, StoreError> {
371        let count = self.conn.query_row(
372            "SELECT COUNT(*) FROM sessions WHERE repo_root = ?1",
373            [repo_root],
374            |r| r.get(0),
375        )?;
376        Ok(count)
377    }
378
379    pub fn latest_activity(&self, repo_root: &str) -> Result<Option<String>, StoreError> {
380        let latest = self.conn.query_row(
381            "SELECT MAX(created_at) FROM (
382                SELECT created_at FROM runs WHERE repo_root = ?1
383                UNION ALL
384                SELECT created_at FROM sessions WHERE repo_root = ?1
385             )",
386            [repo_root],
387            |r| r.get(0),
388        )?;
389        Ok(latest)
390    }
391}
dejavu/store/db.rs

dejavu/store/
db.rs
