Skip to main content

start_command/
execution_store.rs

1//! Execution Store - Dual storage (.lino text + .links binary) for command execution records
2
3use crate::lino_value_json::{json_to_lino_value, lino_value_to_json};
4use crate::local_hostname;
5use chrono::Utc;
6use lino_objects_codec::{decode, encode, LinoValue};
7use serde::{Deserialize, Serialize};
8use serde_json::{json, Value};
9use std::collections::HashMap;
10use std::env;
11use std::fs::{self, OpenOptions};
12use std::io::Write;
13use std::path::{Path, PathBuf};
14use std::process::{Command, Stdio};
15use std::thread;
16use std::time::Duration;
17use uuid::Uuid;
18
19/// Default application folder name
20const DEFAULT_APP_FOLDER_NAME: &str = ".start-command";
21/// Lino database file name
22const LINO_DB_FILE: &str = "executions.lino";
23/// Links database file name
24const LINKS_DB_FILE: &str = "executions.links";
25/// Lock file name
26const LOCK_FILE: &str = "executions.lock";
27/// Lock timeout in milliseconds
28const LOCK_TIMEOUT_MS: u64 = 30000;
29/// Consider lock stale after this many milliseconds
30const LOCK_STALE_MS: u64 = 60000;
31
32/// Execution status
33#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
34#[serde(rename_all = "lowercase")]
35pub enum ExecutionStatus {
36    Executing,
37    Executed,
38}
39
40impl ExecutionStatus {
41    pub fn as_str(&self) -> &'static str {
42        match self {
43            ExecutionStatus::Executing => "executing",
44            ExecutionStatus::Executed => "executed",
45        }
46    }
47}
48
49impl std::fmt::Display for ExecutionStatus {
50    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
51        write!(f, "{}", self.as_str())
52    }
53}
54
55/// Command Execution Record
56#[derive(Debug, Clone, Serialize, Deserialize)]
57#[serde(rename_all = "camelCase")]
58pub struct ExecutionRecord {
59    pub uuid: String,
60    pub pid: Option<u32>,
61    pub status: ExecutionStatus,
62    pub exit_code: Option<i32>,
63    pub command: String,
64    pub log_path: String,
65    pub start_time: String,
66    pub end_time: Option<String>,
67    /// Where `end_time` came from (issue #170.2): `docker-finished-at`,
68    /// `log-footer` or `observed-at`. Without it a real finish time and a
69    /// timestamp taken when someone happened to run `--status` are
70    /// indistinguishable.
71    #[serde(default, skip_serializing_if = "Option::is_none")]
72    pub end_time_source: Option<String>,
73    /// When the end of the execution was *observed*, for the cases where no
74    /// real finish time exists (issue #170.2).
75    #[serde(default, skip_serializing_if = "Option::is_none")]
76    pub observed_at: Option<String>,
77    /// When `cleanup_stale()` gave up on a record. Deliberately not an
78    /// `end_time`: cleanup time is not a finish time (issue #170.3).
79    #[serde(default, skip_serializing_if = "Option::is_none")]
80    pub stale_detected_at: Option<String>,
81    /// `docker inspect .State.StartedAt` of the container that ran the command.
82    #[serde(default, skip_serializing_if = "Option::is_none")]
83    pub container_started_at: Option<String>,
84    #[serde(default, skip_serializing_if = "Option::is_none")]
85    pub oom_killed: Option<bool>,
86    /// cgroup v2 memory counters the detached watcher sampled while the
87    /// container ran (issue #182). Unlike the sticky `oom_killed`, they are
88    /// counted for this run only.
89    #[serde(
90        default,
91        skip_serializing_if = "Option::is_none",
92        deserialize_with = "crate::cgroup_memory::deserialize_cgroup_memory"
93    )]
94    pub cgroup_memory: Option<crate::cgroup_memory::CgroupMemory>,
95    /// Hint explaining an otherwise opaque exit code (issue #162).
96    /// Derived from the log tail on read; never a stored verdict.
97    #[serde(default, skip_serializing_if = "Option::is_none")]
98    pub exit_reason: Option<String>,
99    /// Memory-exhaustion observation derived from the log tail (issue #165).
100    /// A runtime that aborts on its own heap limit never trips `oom_killed`.
101    #[serde(default, skip_serializing_if = "Option::is_none")]
102    pub memory_exhausted: Option<bool>,
103    /// The log line that carried the memory-exhaustion evidence (issue #165).
104    #[serde(default, skip_serializing_if = "Option::is_none")]
105    pub memory_exhausted_reason: Option<String>,
106    pub working_directory: String,
107    pub shell: String,
108    pub platform: String,
109    #[serde(default)]
110    pub options: HashMap<String, Value>,
111}
112
113impl ExecutionRecord {
114    /// Create a new execution record
115    pub fn new(command: &str) -> Self {
116        let now = Utc::now();
117        ExecutionRecord {
118            uuid: Uuid::new_v4().to_string(),
119            pid: None,
120            status: ExecutionStatus::Executing,
121            exit_code: None,
122            command: command.to_string(),
123            log_path: String::new(),
124            start_time: now.to_rfc3339(),
125            end_time: None,
126            end_time_source: None,
127            observed_at: None,
128            stale_detected_at: None,
129            container_started_at: None,
130            oom_killed: None,
131            cgroup_memory: None,
132            exit_reason: None,
133            memory_exhausted: None,
134            memory_exhausted_reason: None,
135            working_directory: env::current_dir()
136                .map(|p| p.to_string_lossy().to_string())
137                .unwrap_or_default(),
138            shell: env::var("SHELL").unwrap_or_else(|_| "/bin/sh".to_string()),
139            platform: std::env::consts::OS.to_string(),
140            options: HashMap::new(),
141        }
142    }
143
144    /// Create a new execution record with options
145    pub fn with_options(options: ExecutionRecordOptions) -> Self {
146        let mut record = Self::new(&options.command);
147        if let Some(uuid) = options.uuid {
148            record.uuid = uuid;
149        }
150        if let Some(pid) = options.pid {
151            record.pid = Some(pid);
152        }
153        if let Some(status) = options.status {
154            record.status = status;
155        }
156        if let Some(exit_code) = options.exit_code {
157            record.exit_code = Some(exit_code);
158        }
159        if let Some(log_path) = options.log_path {
160            record.log_path = log_path;
161        }
162        if let Some(start_time) = options.start_time {
163            record.start_time = start_time;
164        }
165        if let Some(end_time) = options.end_time {
166            record.end_time = Some(end_time);
167        }
168        if let Some(oom_killed) = options.oom_killed {
169            record.oom_killed = Some(oom_killed);
170        }
171        if let Some(working_directory) = options.working_directory {
172            record.working_directory = working_directory;
173        }
174        if let Some(shell) = options.shell {
175            record.shell = shell;
176        }
177        if let Some(platform) = options.platform {
178            record.platform = platform;
179        }
180        if let Some(opts) = options.options {
181            record.options = opts;
182        }
183        record
184    }
185
186    /// Mark execution as completed
187    pub fn complete(&mut self, exit_code: i32) {
188        self.status = ExecutionStatus::Executed;
189        self.exit_code = Some(exit_code);
190        self.end_time = Some(Utc::now().to_rfc3339());
191    }
192
193    /// Convert to JSON Value
194    pub fn to_json(&self) -> Value {
195        serde_json::to_value(self).unwrap_or(Value::Null)
196    }
197
198    /// Create from JSON Value
199    pub fn from_json(value: &Value) -> Option<Self> {
200        serde_json::from_value(value.clone()).ok()
201    }
202}
203
204/// Options for creating an execution record
205#[derive(Debug, Default)]
206pub struct ExecutionRecordOptions {
207    pub uuid: Option<String>,
208    pub command: String,
209    pub pid: Option<u32>,
210    pub status: Option<ExecutionStatus>,
211    pub exit_code: Option<i32>,
212    pub log_path: Option<String>,
213    pub start_time: Option<String>,
214    pub end_time: Option<String>,
215    pub oom_killed: Option<bool>,
216    pub working_directory: Option<String>,
217    pub shell: Option<String>,
218    pub platform: Option<String>,
219    pub options: Option<HashMap<String, Value>>,
220}
221
222/// File-based lock manager
223pub struct LockManager {
224    lock_file_path: PathBuf,
225    lock_acquired: bool,
226}
227
228impl LockManager {
229    /// Create a new lock manager
230    pub fn new(lock_file_path: PathBuf) -> Self {
231        LockManager {
232            lock_file_path,
233            lock_acquired: false,
234        }
235    }
236
237    /// Acquire an exclusive lock
238    pub fn acquire(&mut self, timeout_ms: u64) -> bool {
239        let start = std::time::Instant::now();
240        let timeout = Duration::from_millis(timeout_ms);
241
242        while start.elapsed() < timeout {
243            // Check if existing lock is stale
244            if self.lock_file_path.exists() {
245                if let Some(lock_data) = self.read_lock_file() {
246                    if self.is_lock_stale(&lock_data) {
247                        let _ = fs::remove_file(&self.lock_file_path);
248                    }
249                }
250            }
251
252            // Try to create lock file exclusively
253            match OpenOptions::new()
254                .write(true)
255                .create_new(true)
256                .open(&self.lock_file_path)
257            {
258                Ok(mut file) => {
259                    let lock_data = json!({
260                        "pid": std::process::id(),
261                        "timestamp": std::time::SystemTime::now()
262                            .duration_since(std::time::UNIX_EPOCH)
263                            .map(|d| d.as_millis())
264                            .unwrap_or(0),
265                        "hostname": local_hostname::get()
266                            .map(|h| h.to_string_lossy().to_string())
267                            .unwrap_or_default()
268                    });
269                    let _ = file.write_all(lock_data.to_string().as_bytes());
270                    self.lock_acquired = true;
271                    return true;
272                }
273                Err(_) => {
274                    // Lock file exists, wait and retry
275                    thread::sleep(Duration::from_millis(100));
276                    continue;
277                }
278            }
279        }
280
281        false
282    }
283
284    /// Release the lock
285    pub fn release(&mut self) {
286        if self.lock_acquired {
287            let _ = fs::remove_file(&self.lock_file_path);
288            self.lock_acquired = false;
289        }
290    }
291
292    /// Read lock file data
293    fn read_lock_file(&self) -> Option<Value> {
294        let content = fs::read_to_string(&self.lock_file_path).ok()?;
295        serde_json::from_str(&content).ok()
296    }
297
298    /// Check if lock is stale
299    fn is_lock_stale(&self, lock_data: &Value) -> bool {
300        let timestamp = lock_data.get("timestamp").and_then(|t| t.as_u64());
301
302        // Check if lock is too old
303        if let Some(ts) = timestamp {
304            let now = std::time::SystemTime::now()
305                .duration_since(std::time::UNIX_EPOCH)
306                .map(|d| d.as_millis() as u64)
307                .unwrap_or(0);
308            if now - ts > LOCK_STALE_MS {
309                return true;
310            }
311        } else {
312            return true;
313        }
314
315        // Check if the process that holds the lock is still running (Unix only)
316        #[cfg(unix)]
317        {
318            let pid = lock_data.get("pid").and_then(|p| p.as_u64());
319            if let Some(p) = pid {
320                // Check if process exists using kill(pid, 0)
321                let result = unsafe { libc::kill(p as i32, 0) };
322                if result != 0 {
323                    return true; // Process doesn't exist
324                }
325            }
326        }
327
328        false
329    }
330}
331
332impl Drop for LockManager {
333    fn drop(&mut self) {
334        self.release();
335    }
336}
337
338/// Check if clink is installed
339pub fn is_clink_installed() -> bool {
340    Command::new("clink")
341        .arg("--version")
342        .stdout(Stdio::null())
343        .stderr(Stdio::null())
344        .status()
345        .map(|s| s.success())
346        .unwrap_or(false)
347}
348
349/// Get the default application folder path
350pub fn get_default_app_folder() -> PathBuf {
351    if let Ok(custom) = env::var("START_APP_FOLDER") {
352        return PathBuf::from(custom);
353    }
354
355    let home = env::var("HOME")
356        .or_else(|_| env::var("USERPROFILE"))
357        .unwrap_or_else(|_| ".".to_string());
358    PathBuf::from(home).join(DEFAULT_APP_FOLDER_NAME)
359}
360
361/// ExecutionStore - Main store class for managing execution records
362#[derive(Clone)]
363pub struct ExecutionStore {
364    app_folder: PathBuf,
365    lino_db_path: PathBuf,
366    links_db_path: PathBuf,
367    lock_file_path: PathBuf,
368    use_links: bool,
369    verbose: bool,
370}
371
372/// Options for creating an ExecutionStore
373#[derive(Debug, Default)]
374pub struct ExecutionStoreOptions {
375    pub app_folder: Option<PathBuf>,
376    pub use_links: Option<bool>,
377    pub verbose: bool,
378}
379
380impl ExecutionStore {
381    /// Create a new ExecutionStore with default options
382    pub fn new() -> Self {
383        Self::with_options(ExecutionStoreOptions::default())
384    }
385
386    /// Create a new ExecutionStore with options
387    pub fn with_options(options: ExecutionStoreOptions) -> Self {
388        let app_folder = options.app_folder.unwrap_or_else(get_default_app_folder);
389        let lino_db_path = app_folder.join(LINO_DB_FILE);
390        let links_db_path = app_folder.join(LINKS_DB_FILE);
391        let lock_file_path = app_folder.join(LOCK_FILE);
392        let use_links = options.use_links.unwrap_or_else(is_clink_installed);
393
394        // Ensure app folder exists
395        let _ = fs::create_dir_all(&app_folder);
396
397        ExecutionStore {
398            app_folder,
399            lino_db_path,
400            links_db_path,
401            lock_file_path,
402            use_links,
403            verbose: options.verbose,
404        }
405    }
406
407    /// Log verbose message
408    fn log(&self, message: &str) {
409        if self.verbose {
410            println!("[ExecutionStore] {}", message);
411        }
412    }
413
414    /// Read all execution records from lino file
415    pub fn read_lino_records(&self) -> Vec<ExecutionRecord> {
416        if !self.lino_db_path.exists() {
417            return Vec::new();
418        }
419
420        match fs::read_to_string(&self.lino_db_path) {
421            Ok(content) => {
422                if content.trim().is_empty() {
423                    return Vec::new();
424                }
425
426                match decode(&content) {
427                    Ok(data) => {
428                        if let LinoValue::Array(arr) = data {
429                            arr.iter()
430                                .map(lino_value_to_json)
431                                .filter_map(|value| ExecutionRecord::from_json(&value))
432                                .collect()
433                        } else {
434                            Vec::new()
435                        }
436                    }
437                    Err(e) => {
438                        self.log(&format!("Error decoding lino records: {}", e));
439                        Vec::new()
440                    }
441                }
442            }
443            Err(e) => {
444                self.log(&format!("Error reading lino records: {}", e));
445                Vec::new()
446            }
447        }
448    }
449
450    /// Write execution records to lino file
451    fn write_lino_records(&self, records: &[ExecutionRecord]) -> std::io::Result<()> {
452        let data: Vec<LinoValue> = records
453            .iter()
454            .map(|record| json_to_lino_value(&record.to_json()))
455            .collect();
456        let content = encode(&LinoValue::Array(data));
457        fs::write(&self.lino_db_path, content)?;
458        self.log(&format!("Wrote {} records to lino file", records.len()));
459        Ok(())
460    }
461
462    /// Build clink query for creating/updating a record
463    fn build_clink_create_query(&self, record: &ExecutionRecord) -> String {
464        let obj = record.to_json();
465        let mut links = Vec::new();
466
467        // Create main record link
468        links.push(format!(
469            "({}: ExecutionRecord {})",
470            record.uuid, record.uuid
471        ));
472
473        // Create property links
474        if let Value::Object(map) = obj {
475            for (key, value) in map {
476                let escaped_value = match value {
477                    Value::Object(_) | Value::Array(_) => {
478                        serde_json::to_string(&value).unwrap_or_default()
479                    }
480                    Value::String(s) => s,
481                    Value::Null => "null".to_string(),
482                    other => other.to_string(),
483                };
484                links.push(format!(
485                    "({}.{}: {} \"{}\")",
486                    record.uuid,
487                    key,
488                    key,
489                    escaped_value.replace('"', "\\\"")
490                ));
491            }
492        }
493
494        format!("() (({})))", links.join(") ("))
495    }
496
497    /// Execute clink command
498    fn exec_clink(&self, query: &str) -> Result<String, String> {
499        match Command::new("clink")
500            .arg(query)
501            .arg("--db")
502            .arg(&self.links_db_path)
503            .output()
504        {
505            Ok(output) => {
506                if output.status.success() {
507                    Ok(String::from_utf8_lossy(&output.stdout).to_string())
508                } else {
509                    let stderr = String::from_utf8_lossy(&output.stderr).to_string();
510                    self.log(&format!("Clink error: {}", stderr));
511                    Err(stderr)
512                }
513            }
514            Err(e) => {
515                self.log(&format!("Clink execution error: {}", e));
516                Err(e.to_string())
517            }
518        }
519    }
520
521    /// Write a record to the links database using clink
522    fn write_links_record(&self, record: &ExecutionRecord) -> bool {
523        if !self.use_links {
524            return false;
525        }
526
527        let query = self.build_clink_create_query(record);
528        match self.exec_clink(&query) {
529            Ok(_) => {
530                self.log(&format!("Wrote record {} to links database", record.uuid));
531                true
532            }
533            Err(_) => false,
534        }
535    }
536
537    /// Delete a record from links database
538    fn delete_links_record(&self, uuid: &str) -> bool {
539        if !self.use_links {
540            return false;
541        }
542
543        let query = format!("(($id: {} $any)) ()", uuid);
544        self.exec_clink(&query).is_ok()
545    }
546
547    /// Save an execution record (creates or updates)
548    pub fn save(&self, record: &ExecutionRecord) -> Result<(), String> {
549        let mut lock = LockManager::new(self.lock_file_path.clone());
550
551        if !lock.acquire(LOCK_TIMEOUT_MS) {
552            return Err("Failed to acquire lock for database write".to_string());
553        }
554
555        // Read existing records
556        let mut records = self.read_lino_records();
557
558        // Find existing record index
559        let existing_index = records.iter().position(|r| r.uuid == record.uuid);
560
561        if let Some(idx) = existing_index {
562            // Update existing record
563            records[idx] = record.clone();
564        } else {
565            // Add new record
566            records.push(record.clone());
567        }
568
569        // Write to lino file
570        self.write_lino_records(&records)
571            .map_err(|e| e.to_string())?;
572
573        // Also write to links database if available
574        if self.use_links {
575            self.write_links_record(record);
576        }
577
578        Ok(())
579    }
580
581    /// Get an execution record by UUID or session name (falls back to options.sessionName)
582    pub fn get(&self, identifier: &str) -> Option<ExecutionRecord> {
583        let records = self.read_lino_records();
584        records
585            .iter()
586            .find(|r| r.uuid == identifier)
587            .cloned()
588            .or_else(|| {
589                records
590                    .iter()
591                    .find(|r| {
592                        r.options.get("sessionName").and_then(|v| v.as_str()) == Some(identifier)
593                    })
594                    .cloned()
595            })
596            .or_else(|| {
597                // Finally, honor session names this record used before it was
598                // resumed, so one logical session stays addressable across
599                // restarts (issue #162).
600                records.into_iter().find(|r| {
601                    r.options
602                        .get("sessionNameHistory")
603                        .and_then(|v| v.as_array())
604                        .is_some_and(|history| {
605                            history.iter().any(|name| name.as_str() == Some(identifier))
606                        })
607                })
608            })
609    }
610
611    /// Get all execution records
612    pub fn get_all(&self) -> Vec<ExecutionRecord> {
613        self.read_lino_records()
614    }
615    /// Get records filtered by status
616    pub fn get_by_status(&self, status: ExecutionStatus) -> Vec<ExecutionRecord> {
617        self.read_lino_records()
618            .into_iter()
619            .filter(|r| r.status == status)
620            .collect()
621    }
622    /// Get currently executing commands
623    pub fn get_executing(&self) -> Vec<ExecutionRecord> {
624        self.get_by_status(ExecutionStatus::Executing)
625    }
626    /// Get recently executed commands (sorted by start_time descending)
627    pub fn get_recent(&self, limit: usize) -> Vec<ExecutionRecord> {
628        let mut records = self.read_lino_records();
629        records.sort_by(|a, b| b.start_time.cmp(&a.start_time));
630        records.truncate(limit);
631        records
632    }
633    /// Clean up stale "executing" records (processes no longer running or aged out)
634    pub fn cleanup_stale(&self, options: CleanupOptions) -> CleanupResult {
635        let max_age_ms = options.max_age_ms.unwrap_or(24 * 60 * 60 * 1000);
636        let dry_run = options.dry_run;
637        let mut result = CleanupResult {
638            cleaned: 0,
639            records: Vec::new(),
640            errors: Vec::new(),
641        };
642
643        let records = self.read_lino_records();
644        let mut stale_records: Vec<ExecutionRecord> = Vec::new();
645
646        for record in records
647            .iter()
648            .filter(|r| r.status == ExecutionStatus::Executing)
649        {
650            let mut is_stale = false;
651
652            // Check if process is still running (Unix only, same platform)
653            #[cfg(unix)]
654            if let Some(pid) = record.pid {
655                if record.platform == std::env::consts::OS
656                    && unsafe { libc::kill(pid as i32, 0) } != 0
657                {
658                    is_stale = true;
659                    self.log(&format!("Stale: {} (PID {} gone)", record.uuid, pid));
660                }
661            }
662
663            // Check age
664            if !is_stale {
665                if let Ok(st) = chrono::DateTime::parse_from_rfc3339(&record.start_time) {
666                    let age_ms =
667                        (chrono::Utc::now() - st.with_timezone(&chrono::Utc)).num_milliseconds();
668                    if age_ms > max_age_ms as i64 {
669                        is_stale = true;
670                        self.log(&format!(
671                            "Stale: {} ({}min old)",
672                            record.uuid,
673                            age_ms / 60000
674                        ));
675                    }
676                }
677            }
678
679            if is_stale {
680                stale_records.push(record.clone());
681            }
682        }
683
684        result.records = stale_records.clone();
685
686        if !dry_run && !stale_records.is_empty() {
687            let mut lock = LockManager::new(self.lock_file_path.clone());
688            if !lock.acquire(LOCK_TIMEOUT_MS) {
689                result.errors.push("Failed to acquire lock".to_string());
690                return result;
691            }
692
693            let mut current = self.read_lino_records();
694            for stale in &stale_records {
695                if let Some(i) = current.iter().position(|r| r.uuid == stale.uuid) {
696                    current[i].status = ExecutionStatus::Executed;
697                    current[i].exit_code = Some(-1);
698                    // Cleanup time is not a finish time: the execution ended at
699                    // an unknown moment before this one. Record when it was
700                    // detected and leave `end_time` empty (issue #170.3).
701                    current[i].end_time = None;
702                    current[i].stale_detected_at = Some(chrono::Utc::now().to_rfc3339());
703                    result.cleaned += 1;
704                }
705            }
706
707            if let Err(e) = self.write_lino_records(&current) {
708                result.errors.push(format!("Cleanup error: {}", e));
709            } else {
710                self.log(&format!("Cleaned {} stale records", result.cleaned));
711            }
712        } else if dry_run {
713            result.cleaned = stale_records.len();
714        }
715
716        result
717    }
718
719    /// Delete an execution record
720    pub fn delete(&self, uuid: &str) -> Result<bool, String> {
721        let mut lock = LockManager::new(self.lock_file_path.clone());
722
723        if !lock.acquire(LOCK_TIMEOUT_MS) {
724            return Err("Failed to acquire lock for database write".to_string());
725        }
726
727        let records = self.read_lino_records();
728        let filtered: Vec<_> = records.iter().filter(|r| r.uuid != uuid).cloned().collect();
729
730        if filtered.len() == records.len() {
731            return Ok(false); // Record not found
732        }
733
734        self.write_lino_records(&filtered)
735            .map_err(|e| e.to_string())?;
736
737        // Also delete from links database
738        if self.use_links {
739            self.delete_links_record(uuid);
740        }
741
742        Ok(true)
743    }
744
745    /// Delete all records
746    pub fn clear(&self) -> Result<(), String> {
747        let mut lock = LockManager::new(self.lock_file_path.clone());
748
749        if !lock.acquire(LOCK_TIMEOUT_MS) {
750            return Err("Failed to acquire lock for database write".to_string());
751        }
752
753        self.write_lino_records(&[]).map_err(|e| e.to_string())?;
754
755        // Clear links database by removing the file
756        if self.use_links && self.links_db_path.exists() {
757            let _ = fs::remove_file(&self.links_db_path);
758        }
759
760        Ok(())
761    }
762
763    /// Verify that both databases have consistent data
764    pub fn verify_consistency(&self) -> ConsistencyResult {
765        let mut result = ConsistencyResult {
766            consistent: true,
767            lino_count: 0,
768            links_count: 0,
769            errors: Vec::new(),
770        };
771
772        // Read lino records
773        let lino_records = self.read_lino_records();
774        result.lino_count = lino_records.len();
775
776        if !self.use_links {
777            result
778                .errors
779                .push("clink not installed - links database not available".to_string());
780            return result;
781        }
782
783        // Query links database for all ExecutionRecord links
784        match self.exec_clink("((($id: ExecutionRecord $uuid)) (($id: ExecutionRecord $uuid)))") {
785            Ok(output) => {
786                // Count unique UUIDs in the output
787                let re = regex::Regex::new(r"ExecutionRecord\s+([a-f0-9-]{36})").unwrap();
788                let uuids: std::collections::HashSet<_> = re
789                    .captures_iter(&output)
790                    .filter_map(|c| c.get(1).map(|m| m.as_str().to_lowercase()))
791                    .collect();
792                result.links_count = uuids.len();
793
794                // Check if counts match
795                if result.lino_count != result.links_count {
796                    result.consistent = false;
797                    result.errors.push(format!(
798                        "Record count mismatch: lino={}, links={}",
799                        result.lino_count, result.links_count
800                    ));
801                }
802
803                // Verify each lino record exists in links
804                for record in &lino_records {
805                    if !uuids.contains(&record.uuid.to_lowercase()) {
806                        result.consistent = false;
807                        result.errors.push(format!(
808                            "Record {} missing from links database",
809                            record.uuid
810                        ));
811                    }
812                }
813            }
814            Err(e) => {
815                result
816                    .errors
817                    .push(format!("Failed to query links database: {}", e));
818                result.consistent = false;
819            }
820        }
821
822        result
823    }
824
825    /// Get database statistics
826    pub fn get_stats(&self) -> ExecutionStats {
827        let records = self.read_lino_records();
828        let executing = records
829            .iter()
830            .filter(|r| r.status == ExecutionStatus::Executing)
831            .count();
832        let executed = records
833            .iter()
834            .filter(|r| r.status == ExecutionStatus::Executed)
835            .count();
836        let successful = records
837            .iter()
838            .filter(|r| r.status == ExecutionStatus::Executed && r.exit_code == Some(0))
839            .count();
840        let failed = records
841            .iter()
842            .filter(|r| {
843                r.status == ExecutionStatus::Executed
844                    && r.exit_code.map(|c| c != 0).unwrap_or(false)
845            })
846            .count();
847
848        ExecutionStats {
849            total: records.len(),
850            executing,
851            executed,
852            successful,
853            failed,
854            clink_available: self.use_links,
855            lino_db_path: self.lino_db_path.to_string_lossy().to_string(),
856            links_db_path: self.links_db_path.to_string_lossy().to_string(),
857        }
858    }
859
860    /// Get the app folder path
861    pub fn app_folder(&self) -> &Path {
862        &self.app_folder
863    }
864}
865
866impl Default for ExecutionStore {
867    fn default() -> Self {
868        Self::new()
869    }
870}
871
872/// Result of consistency verification
873#[derive(Debug)]
874pub struct ConsistencyResult {
875    pub consistent: bool,
876    pub lino_count: usize,
877    pub links_count: usize,
878    pub errors: Vec<String>,
879}
880
881/// Options for cleanup_stale: max_age_ms (default 24h), dry_run
882#[derive(Debug, Default)]
883pub struct CleanupOptions {
884    pub max_age_ms: Option<u64>,
885    pub dry_run: bool,
886}
887
888/// Result of cleanup_stale: cleaned count, records, errors
889#[derive(Debug)]
890pub struct CleanupResult {
891    pub cleaned: usize,
892    pub records: Vec<ExecutionRecord>,
893    pub errors: Vec<String>,
894}
895
896/// Execution statistics
897#[derive(Debug)]
898pub struct ExecutionStats {
899    pub total: usize,
900    pub executing: usize,
901    pub executed: usize,
902    pub successful: usize,
903    pub failed: usize,
904    pub clink_available: bool,
905    pub lino_db_path: String,
906    pub links_db_path: String,
907}
908
909#[cfg(test)]
910#[path = "execution_store_cases.rs"]
911mod tests;