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(
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 #[serde(default, skip_serializing_if = "Option::is_none")]
98 pub exit_reason: Option<String>,
99 #[serde(default, skip_serializing_if = "Option::is_none")]
102 pub memory_exhausted: Option<bool>,
103 #[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 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 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 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 pub fn to_json(&self) -> Value {
195 serde_json::to_value(self).unwrap_or(Value::Null)
196 }
197
198 pub fn from_json(value: &Value) -> Option<Self> {
200 serde_json::from_value(value.clone()).ok()
201 }
202}
203
204#[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
222pub struct LockManager {
224 lock_file_path: PathBuf,
225 lock_acquired: bool,
226}
227
228impl LockManager {
229 pub fn new(lock_file_path: PathBuf) -> Self {
231 LockManager {
232 lock_file_path,
233 lock_acquired: false,
234 }
235 }
236
237 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 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 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 thread::sleep(Duration::from_millis(100));
276 continue;
277 }
278 }
279 }
280
281 false
282 }
283
284 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 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 fn is_lock_stale(&self, lock_data: &Value) -> bool {
300 let timestamp = lock_data.get("timestamp").and_then(|t| t.as_u64());
301
302 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 #[cfg(unix)]
317 {
318 let pid = lock_data.get("pid").and_then(|p| p.as_u64());
319 if let Some(p) = pid {
320 let result = unsafe { libc::kill(p as i32, 0) };
322 if result != 0 {
323 return true; }
325 }
326 }
327
328 false
329 }
330}
331
332impl Drop for LockManager {
333 fn drop(&mut self) {
334 self.release();
335 }
336}
337
338pub 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
349pub 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#[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#[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 pub fn new() -> Self {
383 Self::with_options(ExecutionStoreOptions::default())
384 }
385
386 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 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 fn log(&self, message: &str) {
409 if self.verbose {
410 println!("[ExecutionStore] {}", message);
411 }
412 }
413
414 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 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 fn build_clink_create_query(&self, record: &ExecutionRecord) -> String {
464 let obj = record.to_json();
465 let mut links = Vec::new();
466
467 links.push(format!(
469 "({}: ExecutionRecord {})",
470 record.uuid, record.uuid
471 ));
472
473 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 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 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 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 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 let mut records = self.read_lino_records();
557
558 let existing_index = records.iter().position(|r| r.uuid == record.uuid);
560
561 if let Some(idx) = existing_index {
562 records[idx] = record.clone();
564 } else {
565 records.push(record.clone());
567 }
568
569 self.write_lino_records(&records)
571 .map_err(|e| e.to_string())?;
572
573 if self.use_links {
575 self.write_links_record(record);
576 }
577
578 Ok(())
579 }
580
581 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 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 pub fn get_all(&self) -> Vec<ExecutionRecord> {
613 self.read_lino_records()
614 }
615 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 pub fn get_executing(&self) -> Vec<ExecutionRecord> {
624 self.get_by_status(ExecutionStatus::Executing)
625 }
626 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 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 #[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 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 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(¤t) {
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 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); }
733
734 self.write_lino_records(&filtered)
735 .map_err(|e| e.to_string())?;
736
737 if self.use_links {
739 self.delete_links_record(uuid);
740 }
741
742 Ok(true)
743 }
744
745 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 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 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 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 match self.exec_clink("((($id: ExecutionRecord $uuid)) (($id: ExecutionRecord $uuid)))") {
785 Ok(output) => {
786 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 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 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 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 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#[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#[derive(Debug, Default)]
883pub struct CleanupOptions {
884 pub max_age_ms: Option<u64>,
885 pub dry_run: bool,
886}
887
888#[derive(Debug)]
890pub struct CleanupResult {
891 pub cleaned: usize,
892 pub records: Vec<ExecutionRecord>,
893 pub errors: Vec<String>,
894}
895
896#[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;