1use 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
19const DEFAULT_APP_FOLDER_NAME: &str = ".start-command";
21const LINO_DB_FILE: &str = "executions.lino";
23const LINKS_DB_FILE: &str = "executions.links";
25const LOCK_FILE: &str = "executions.lock";
27const LOCK_TIMEOUT_MS: u64 = 30000;
29const LOCK_STALE_MS: u64 = 60000;
31
32#[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#[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 #[serde(default, skip_serializing_if = "Option::is_none")]
72 pub end_time_source: Option<String>,
73 #[serde(default, skip_serializing_if = "Option::is_none")]
76 pub observed_at: Option<String>,
77 #[serde(default, skip_serializing_if = "Option::is_none")]
80 pub stale_detected_at: Option<String>,
81 #[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 #[serde(default, skip_serializing_if = "Option::is_none")]
89 pub exit_reason: Option<String>,
90 #[serde(default, skip_serializing_if = "Option::is_none")]
93 pub memory_exhausted: Option<bool>,
94 #[serde(default, skip_serializing_if = "Option::is_none")]
96 pub memory_exhausted_reason: Option<String>,
97 pub working_directory: String,
98 pub shell: String,
99 pub platform: String,
100 #[serde(default)]
101 pub options: HashMap<String, Value>,
102}
103
104impl ExecutionRecord {
105 pub fn new(command: &str) -> Self {
107 let now = Utc::now();
108 ExecutionRecord {
109 uuid: Uuid::new_v4().to_string(),
110 pid: None,
111 status: ExecutionStatus::Executing,
112 exit_code: None,
113 command: command.to_string(),
114 log_path: String::new(),
115 start_time: now.to_rfc3339(),
116 end_time: None,
117 end_time_source: None,
118 observed_at: None,
119 stale_detected_at: None,
120 container_started_at: None,
121 oom_killed: None,
122 exit_reason: None,
123 memory_exhausted: None,
124 memory_exhausted_reason: None,
125 working_directory: env::current_dir()
126 .map(|p| p.to_string_lossy().to_string())
127 .unwrap_or_default(),
128 shell: env::var("SHELL").unwrap_or_else(|_| "/bin/sh".to_string()),
129 platform: std::env::consts::OS.to_string(),
130 options: HashMap::new(),
131 }
132 }
133
134 pub fn with_options(options: ExecutionRecordOptions) -> Self {
136 let mut record = Self::new(&options.command);
137 if let Some(uuid) = options.uuid {
138 record.uuid = uuid;
139 }
140 if let Some(pid) = options.pid {
141 record.pid = Some(pid);
142 }
143 if let Some(status) = options.status {
144 record.status = status;
145 }
146 if let Some(exit_code) = options.exit_code {
147 record.exit_code = Some(exit_code);
148 }
149 if let Some(log_path) = options.log_path {
150 record.log_path = log_path;
151 }
152 if let Some(start_time) = options.start_time {
153 record.start_time = start_time;
154 }
155 if let Some(end_time) = options.end_time {
156 record.end_time = Some(end_time);
157 }
158 if let Some(oom_killed) = options.oom_killed {
159 record.oom_killed = Some(oom_killed);
160 }
161 if let Some(working_directory) = options.working_directory {
162 record.working_directory = working_directory;
163 }
164 if let Some(shell) = options.shell {
165 record.shell = shell;
166 }
167 if let Some(platform) = options.platform {
168 record.platform = platform;
169 }
170 if let Some(opts) = options.options {
171 record.options = opts;
172 }
173 record
174 }
175
176 pub fn complete(&mut self, exit_code: i32) {
178 self.status = ExecutionStatus::Executed;
179 self.exit_code = Some(exit_code);
180 self.end_time = Some(Utc::now().to_rfc3339());
181 }
182
183 pub fn to_json(&self) -> Value {
185 serde_json::to_value(self).unwrap_or(Value::Null)
186 }
187
188 pub fn from_json(value: &Value) -> Option<Self> {
190 serde_json::from_value(value.clone()).ok()
191 }
192}
193
194#[derive(Debug, Default)]
196pub struct ExecutionRecordOptions {
197 pub uuid: Option<String>,
198 pub command: String,
199 pub pid: Option<u32>,
200 pub status: Option<ExecutionStatus>,
201 pub exit_code: Option<i32>,
202 pub log_path: Option<String>,
203 pub start_time: Option<String>,
204 pub end_time: Option<String>,
205 pub oom_killed: Option<bool>,
206 pub working_directory: Option<String>,
207 pub shell: Option<String>,
208 pub platform: Option<String>,
209 pub options: Option<HashMap<String, Value>>,
210}
211
212pub struct LockManager {
214 lock_file_path: PathBuf,
215 lock_acquired: bool,
216}
217
218impl LockManager {
219 pub fn new(lock_file_path: PathBuf) -> Self {
221 LockManager {
222 lock_file_path,
223 lock_acquired: false,
224 }
225 }
226
227 pub fn acquire(&mut self, timeout_ms: u64) -> bool {
229 let start = std::time::Instant::now();
230 let timeout = Duration::from_millis(timeout_ms);
231
232 while start.elapsed() < timeout {
233 if self.lock_file_path.exists() {
235 if let Some(lock_data) = self.read_lock_file() {
236 if self.is_lock_stale(&lock_data) {
237 let _ = fs::remove_file(&self.lock_file_path);
238 }
239 }
240 }
241
242 match OpenOptions::new()
244 .write(true)
245 .create_new(true)
246 .open(&self.lock_file_path)
247 {
248 Ok(mut file) => {
249 let lock_data = json!({
250 "pid": std::process::id(),
251 "timestamp": std::time::SystemTime::now()
252 .duration_since(std::time::UNIX_EPOCH)
253 .map(|d| d.as_millis())
254 .unwrap_or(0),
255 "hostname": local_hostname::get()
256 .map(|h| h.to_string_lossy().to_string())
257 .unwrap_or_default()
258 });
259 let _ = file.write_all(lock_data.to_string().as_bytes());
260 self.lock_acquired = true;
261 return true;
262 }
263 Err(_) => {
264 thread::sleep(Duration::from_millis(100));
266 continue;
267 }
268 }
269 }
270
271 false
272 }
273
274 pub fn release(&mut self) {
276 if self.lock_acquired {
277 let _ = fs::remove_file(&self.lock_file_path);
278 self.lock_acquired = false;
279 }
280 }
281
282 fn read_lock_file(&self) -> Option<Value> {
284 let content = fs::read_to_string(&self.lock_file_path).ok()?;
285 serde_json::from_str(&content).ok()
286 }
287
288 fn is_lock_stale(&self, lock_data: &Value) -> bool {
290 let timestamp = lock_data.get("timestamp").and_then(|t| t.as_u64());
291
292 if let Some(ts) = timestamp {
294 let now = std::time::SystemTime::now()
295 .duration_since(std::time::UNIX_EPOCH)
296 .map(|d| d.as_millis() as u64)
297 .unwrap_or(0);
298 if now - ts > LOCK_STALE_MS {
299 return true;
300 }
301 } else {
302 return true;
303 }
304
305 #[cfg(unix)]
307 {
308 let pid = lock_data.get("pid").and_then(|p| p.as_u64());
309 if let Some(p) = pid {
310 let result = unsafe { libc::kill(p as i32, 0) };
312 if result != 0 {
313 return true; }
315 }
316 }
317
318 false
319 }
320}
321
322impl Drop for LockManager {
323 fn drop(&mut self) {
324 self.release();
325 }
326}
327
328pub fn is_clink_installed() -> bool {
330 Command::new("clink")
331 .arg("--version")
332 .stdout(Stdio::null())
333 .stderr(Stdio::null())
334 .status()
335 .map(|s| s.success())
336 .unwrap_or(false)
337}
338
339pub fn get_default_app_folder() -> PathBuf {
341 if let Ok(custom) = env::var("START_APP_FOLDER") {
342 return PathBuf::from(custom);
343 }
344
345 let home = env::var("HOME")
346 .or_else(|_| env::var("USERPROFILE"))
347 .unwrap_or_else(|_| ".".to_string());
348 PathBuf::from(home).join(DEFAULT_APP_FOLDER_NAME)
349}
350
351#[derive(Clone)]
353pub struct ExecutionStore {
354 app_folder: PathBuf,
355 lino_db_path: PathBuf,
356 links_db_path: PathBuf,
357 lock_file_path: PathBuf,
358 use_links: bool,
359 verbose: bool,
360}
361
362#[derive(Debug, Default)]
364pub struct ExecutionStoreOptions {
365 pub app_folder: Option<PathBuf>,
366 pub use_links: Option<bool>,
367 pub verbose: bool,
368}
369
370impl ExecutionStore {
371 pub fn new() -> Self {
373 Self::with_options(ExecutionStoreOptions::default())
374 }
375
376 pub fn with_options(options: ExecutionStoreOptions) -> Self {
378 let app_folder = options.app_folder.unwrap_or_else(get_default_app_folder);
379 let lino_db_path = app_folder.join(LINO_DB_FILE);
380 let links_db_path = app_folder.join(LINKS_DB_FILE);
381 let lock_file_path = app_folder.join(LOCK_FILE);
382 let use_links = options.use_links.unwrap_or_else(is_clink_installed);
383
384 let _ = fs::create_dir_all(&app_folder);
386
387 ExecutionStore {
388 app_folder,
389 lino_db_path,
390 links_db_path,
391 lock_file_path,
392 use_links,
393 verbose: options.verbose,
394 }
395 }
396
397 fn log(&self, message: &str) {
399 if self.verbose {
400 println!("[ExecutionStore] {}", message);
401 }
402 }
403
404 pub fn read_lino_records(&self) -> Vec<ExecutionRecord> {
406 if !self.lino_db_path.exists() {
407 return Vec::new();
408 }
409
410 match fs::read_to_string(&self.lino_db_path) {
411 Ok(content) => {
412 if content.trim().is_empty() {
413 return Vec::new();
414 }
415
416 match decode(&content) {
417 Ok(data) => {
418 if let LinoValue::Array(arr) = data {
419 arr.iter()
420 .map(lino_value_to_json)
421 .filter_map(|value| ExecutionRecord::from_json(&value))
422 .collect()
423 } else {
424 Vec::new()
425 }
426 }
427 Err(e) => {
428 self.log(&format!("Error decoding lino records: {}", e));
429 Vec::new()
430 }
431 }
432 }
433 Err(e) => {
434 self.log(&format!("Error reading lino records: {}", e));
435 Vec::new()
436 }
437 }
438 }
439
440 fn write_lino_records(&self, records: &[ExecutionRecord]) -> std::io::Result<()> {
442 let data: Vec<LinoValue> = records
443 .iter()
444 .map(|record| json_to_lino_value(&record.to_json()))
445 .collect();
446 let content = encode(&LinoValue::Array(data));
447 fs::write(&self.lino_db_path, content)?;
448 self.log(&format!("Wrote {} records to lino file", records.len()));
449 Ok(())
450 }
451
452 fn build_clink_create_query(&self, record: &ExecutionRecord) -> String {
454 let obj = record.to_json();
455 let mut links = Vec::new();
456
457 links.push(format!(
459 "({}: ExecutionRecord {})",
460 record.uuid, record.uuid
461 ));
462
463 if let Value::Object(map) = obj {
465 for (key, value) in map {
466 let escaped_value = match value {
467 Value::Object(_) | Value::Array(_) => {
468 serde_json::to_string(&value).unwrap_or_default()
469 }
470 Value::String(s) => s,
471 Value::Null => "null".to_string(),
472 other => other.to_string(),
473 };
474 links.push(format!(
475 "({}.{}: {} \"{}\")",
476 record.uuid,
477 key,
478 key,
479 escaped_value.replace('"', "\\\"")
480 ));
481 }
482 }
483
484 format!("() (({})))", links.join(") ("))
485 }
486
487 fn exec_clink(&self, query: &str) -> Result<String, String> {
489 match Command::new("clink")
490 .arg(query)
491 .arg("--db")
492 .arg(&self.links_db_path)
493 .output()
494 {
495 Ok(output) => {
496 if output.status.success() {
497 Ok(String::from_utf8_lossy(&output.stdout).to_string())
498 } else {
499 let stderr = String::from_utf8_lossy(&output.stderr).to_string();
500 self.log(&format!("Clink error: {}", stderr));
501 Err(stderr)
502 }
503 }
504 Err(e) => {
505 self.log(&format!("Clink execution error: {}", e));
506 Err(e.to_string())
507 }
508 }
509 }
510
511 fn write_links_record(&self, record: &ExecutionRecord) -> bool {
513 if !self.use_links {
514 return false;
515 }
516
517 let query = self.build_clink_create_query(record);
518 match self.exec_clink(&query) {
519 Ok(_) => {
520 self.log(&format!("Wrote record {} to links database", record.uuid));
521 true
522 }
523 Err(_) => false,
524 }
525 }
526
527 fn delete_links_record(&self, uuid: &str) -> bool {
529 if !self.use_links {
530 return false;
531 }
532
533 let query = format!("(($id: {} $any)) ()", uuid);
534 self.exec_clink(&query).is_ok()
535 }
536
537 pub fn save(&self, record: &ExecutionRecord) -> Result<(), String> {
539 let mut lock = LockManager::new(self.lock_file_path.clone());
540
541 if !lock.acquire(LOCK_TIMEOUT_MS) {
542 return Err("Failed to acquire lock for database write".to_string());
543 }
544
545 let mut records = self.read_lino_records();
547
548 let existing_index = records.iter().position(|r| r.uuid == record.uuid);
550
551 if let Some(idx) = existing_index {
552 records[idx] = record.clone();
554 } else {
555 records.push(record.clone());
557 }
558
559 self.write_lino_records(&records)
561 .map_err(|e| e.to_string())?;
562
563 if self.use_links {
565 self.write_links_record(record);
566 }
567
568 Ok(())
569 }
570
571 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 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 pub fn get_all(&self) -> Vec<ExecutionRecord> {
603 self.read_lino_records()
604 }
605 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 pub fn get_executing(&self) -> Vec<ExecutionRecord> {
614 self.get_by_status(ExecutionStatus::Executing)
615 }
616 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 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 let mut is_stale = false;
641
642 #[cfg(unix)]
644 if let Some(pid) = record.pid {
645 if record.platform == std::env::consts::OS
646 && unsafe { libc::kill(pid as i32, 0) } != 0
647 {
648 is_stale = true;
649 self.log(&format!("Stale: {} (PID {} gone)", record.uuid, pid));
650 }
651 }
652
653 if !is_stale {
655 if let Ok(st) = chrono::DateTime::parse_from_rfc3339(&record.start_time) {
656 let age_ms =
657 (chrono::Utc::now() - st.with_timezone(&chrono::Utc)).num_milliseconds();
658 if age_ms > max_age_ms as i64 {
659 is_stale = true;
660 self.log(&format!(
661 "Stale: {} ({}min old)",
662 record.uuid,
663 age_ms / 60000
664 ));
665 }
666 }
667 }
668
669 if is_stale {
670 stale_records.push(record.clone());
671 }
672 }
673
674 result.records = stale_records.clone();
675
676 if !dry_run && !stale_records.is_empty() {
677 let mut lock = LockManager::new(self.lock_file_path.clone());
678 if !lock.acquire(LOCK_TIMEOUT_MS) {
679 result.errors.push("Failed to acquire lock".to_string());
680 return result;
681 }
682
683 let mut current = self.read_lino_records();
684 for stale in &stale_records {
685 if let Some(i) = current.iter().position(|r| r.uuid == stale.uuid) {
686 current[i].status = ExecutionStatus::Executed;
687 current[i].exit_code = Some(-1);
688 current[i].end_time = None;
692 current[i].stale_detected_at = Some(chrono::Utc::now().to_rfc3339());
693 result.cleaned += 1;
694 }
695 }
696
697 if let Err(e) = self.write_lino_records(¤t) {
698 result.errors.push(format!("Cleanup error: {}", e));
699 } else {
700 self.log(&format!("Cleaned {} stale records", result.cleaned));
701 }
702 } else if dry_run {
703 result.cleaned = stale_records.len();
704 }
705
706 result
707 }
708
709 pub fn delete(&self, uuid: &str) -> Result<bool, String> {
711 let mut lock = LockManager::new(self.lock_file_path.clone());
712
713 if !lock.acquire(LOCK_TIMEOUT_MS) {
714 return Err("Failed to acquire lock for database write".to_string());
715 }
716
717 let records = self.read_lino_records();
718 let filtered: Vec<_> = records.iter().filter(|r| r.uuid != uuid).cloned().collect();
719
720 if filtered.len() == records.len() {
721 return Ok(false); }
723
724 self.write_lino_records(&filtered)
725 .map_err(|e| e.to_string())?;
726
727 if self.use_links {
729 self.delete_links_record(uuid);
730 }
731
732 Ok(true)
733 }
734
735 pub fn clear(&self) -> Result<(), String> {
737 let mut lock = LockManager::new(self.lock_file_path.clone());
738
739 if !lock.acquire(LOCK_TIMEOUT_MS) {
740 return Err("Failed to acquire lock for database write".to_string());
741 }
742
743 self.write_lino_records(&[]).map_err(|e| e.to_string())?;
744
745 if self.use_links && self.links_db_path.exists() {
747 let _ = fs::remove_file(&self.links_db_path);
748 }
749
750 Ok(())
751 }
752
753 pub fn verify_consistency(&self) -> ConsistencyResult {
755 let mut result = ConsistencyResult {
756 consistent: true,
757 lino_count: 0,
758 links_count: 0,
759 errors: Vec::new(),
760 };
761
762 let lino_records = self.read_lino_records();
764 result.lino_count = lino_records.len();
765
766 if !self.use_links {
767 result
768 .errors
769 .push("clink not installed - links database not available".to_string());
770 return result;
771 }
772
773 match self.exec_clink("((($id: ExecutionRecord $uuid)) (($id: ExecutionRecord $uuid)))") {
775 Ok(output) => {
776 let re = regex::Regex::new(r"ExecutionRecord\s+([a-f0-9-]{36})").unwrap();
778 let uuids: std::collections::HashSet<_> = re
779 .captures_iter(&output)
780 .filter_map(|c| c.get(1).map(|m| m.as_str().to_lowercase()))
781 .collect();
782 result.links_count = uuids.len();
783
784 if result.lino_count != result.links_count {
786 result.consistent = false;
787 result.errors.push(format!(
788 "Record count mismatch: lino={}, links={}",
789 result.lino_count, result.links_count
790 ));
791 }
792
793 for record in &lino_records {
795 if !uuids.contains(&record.uuid.to_lowercase()) {
796 result.consistent = false;
797 result.errors.push(format!(
798 "Record {} missing from links database",
799 record.uuid
800 ));
801 }
802 }
803 }
804 Err(e) => {
805 result
806 .errors
807 .push(format!("Failed to query links database: {}", e));
808 result.consistent = false;
809 }
810 }
811
812 result
813 }
814
815 pub fn get_stats(&self) -> ExecutionStats {
817 let records = self.read_lino_records();
818 let executing = records
819 .iter()
820 .filter(|r| r.status == ExecutionStatus::Executing)
821 .count();
822 let executed = records
823 .iter()
824 .filter(|r| r.status == ExecutionStatus::Executed)
825 .count();
826 let successful = records
827 .iter()
828 .filter(|r| r.status == ExecutionStatus::Executed && r.exit_code == Some(0))
829 .count();
830 let failed = records
831 .iter()
832 .filter(|r| {
833 r.status == ExecutionStatus::Executed
834 && r.exit_code.map(|c| c != 0).unwrap_or(false)
835 })
836 .count();
837
838 ExecutionStats {
839 total: records.len(),
840 executing,
841 executed,
842 successful,
843 failed,
844 clink_available: self.use_links,
845 lino_db_path: self.lino_db_path.to_string_lossy().to_string(),
846 links_db_path: self.links_db_path.to_string_lossy().to_string(),
847 }
848 }
849
850 pub fn app_folder(&self) -> &Path {
852 &self.app_folder
853 }
854}
855
856impl Default for ExecutionStore {
857 fn default() -> Self {
858 Self::new()
859 }
860}
861
862#[derive(Debug)]
864pub struct ConsistencyResult {
865 pub consistent: bool,
866 pub lino_count: usize,
867 pub links_count: usize,
868 pub errors: Vec<String>,
869}
870
871#[derive(Debug, Default)]
873pub struct CleanupOptions {
874 pub max_age_ms: Option<u64>,
875 pub dry_run: bool,
876}
877
878#[derive(Debug)]
880pub struct CleanupResult {
881 pub cleaned: usize,
882 pub records: Vec<ExecutionRecord>,
883 pub errors: Vec<String>,
884}
885
886#[derive(Debug)]
888pub struct ExecutionStats {
889 pub total: usize,
890 pub executing: usize,
891 pub executed: usize,
892 pub successful: usize,
893 pub failed: usize,
894 pub clink_available: bool,
895 pub lino_db_path: String,
896 pub links_db_path: String,
897}
898
899#[cfg(test)]
900#[path = "execution_store_cases.rs"]
901mod tests;