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