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")]
68 pub oom_killed: Option<bool>,
69 #[serde(default, skip_serializing_if = "Option::is_none")]
72 pub exit_reason: Option<String>,
73 #[serde(default, skip_serializing_if = "Option::is_none")]
76 pub memory_exhausted: Option<bool>,
77 #[serde(default, skip_serializing_if = "Option::is_none")]
79 pub memory_exhausted_reason: Option<String>,
80 pub working_directory: String,
81 pub shell: String,
82 pub platform: String,
83 #[serde(default)]
84 pub options: HashMap<String, Value>,
85}
86
87impl ExecutionRecord {
88 pub fn new(command: &str) -> Self {
90 let now = Utc::now();
91 ExecutionRecord {
92 uuid: Uuid::new_v4().to_string(),
93 pid: None,
94 status: ExecutionStatus::Executing,
95 exit_code: None,
96 command: command.to_string(),
97 log_path: String::new(),
98 start_time: now.to_rfc3339(),
99 end_time: None,
100 oom_killed: None,
101 exit_reason: None,
102 memory_exhausted: None,
103 memory_exhausted_reason: None,
104 working_directory: env::current_dir()
105 .map(|p| p.to_string_lossy().to_string())
106 .unwrap_or_default(),
107 shell: env::var("SHELL").unwrap_or_else(|_| "/bin/sh".to_string()),
108 platform: std::env::consts::OS.to_string(),
109 options: HashMap::new(),
110 }
111 }
112
113 pub fn with_options(options: ExecutionRecordOptions) -> Self {
115 let mut record = Self::new(&options.command);
116 if let Some(uuid) = options.uuid {
117 record.uuid = uuid;
118 }
119 if let Some(pid) = options.pid {
120 record.pid = Some(pid);
121 }
122 if let Some(status) = options.status {
123 record.status = status;
124 }
125 if let Some(exit_code) = options.exit_code {
126 record.exit_code = Some(exit_code);
127 }
128 if let Some(log_path) = options.log_path {
129 record.log_path = log_path;
130 }
131 if let Some(start_time) = options.start_time {
132 record.start_time = start_time;
133 }
134 if let Some(end_time) = options.end_time {
135 record.end_time = Some(end_time);
136 }
137 if let Some(oom_killed) = options.oom_killed {
138 record.oom_killed = Some(oom_killed);
139 }
140 if let Some(working_directory) = options.working_directory {
141 record.working_directory = working_directory;
142 }
143 if let Some(shell) = options.shell {
144 record.shell = shell;
145 }
146 if let Some(platform) = options.platform {
147 record.platform = platform;
148 }
149 if let Some(opts) = options.options {
150 record.options = opts;
151 }
152 record
153 }
154
155 pub fn complete(&mut self, exit_code: i32) {
157 self.status = ExecutionStatus::Executed;
158 self.exit_code = Some(exit_code);
159 self.end_time = Some(Utc::now().to_rfc3339());
160 }
161
162 pub fn to_json(&self) -> Value {
164 serde_json::to_value(self).unwrap_or(Value::Null)
165 }
166
167 pub fn from_json(value: &Value) -> Option<Self> {
169 serde_json::from_value(value.clone()).ok()
170 }
171}
172
173#[derive(Debug, Default)]
175pub struct ExecutionRecordOptions {
176 pub uuid: Option<String>,
177 pub command: String,
178 pub pid: Option<u32>,
179 pub status: Option<ExecutionStatus>,
180 pub exit_code: Option<i32>,
181 pub log_path: Option<String>,
182 pub start_time: Option<String>,
183 pub end_time: Option<String>,
184 pub oom_killed: Option<bool>,
185 pub working_directory: Option<String>,
186 pub shell: Option<String>,
187 pub platform: Option<String>,
188 pub options: Option<HashMap<String, Value>>,
189}
190
191pub struct LockManager {
193 lock_file_path: PathBuf,
194 lock_acquired: bool,
195}
196
197impl LockManager {
198 pub fn new(lock_file_path: PathBuf) -> Self {
200 LockManager {
201 lock_file_path,
202 lock_acquired: false,
203 }
204 }
205
206 pub fn acquire(&mut self, timeout_ms: u64) -> bool {
208 let start = std::time::Instant::now();
209 let timeout = Duration::from_millis(timeout_ms);
210
211 while start.elapsed() < timeout {
212 if self.lock_file_path.exists() {
214 if let Some(lock_data) = self.read_lock_file() {
215 if self.is_lock_stale(&lock_data) {
216 let _ = fs::remove_file(&self.lock_file_path);
217 }
218 }
219 }
220
221 match OpenOptions::new()
223 .write(true)
224 .create_new(true)
225 .open(&self.lock_file_path)
226 {
227 Ok(mut file) => {
228 let lock_data = json!({
229 "pid": std::process::id(),
230 "timestamp": std::time::SystemTime::now()
231 .duration_since(std::time::UNIX_EPOCH)
232 .map(|d| d.as_millis())
233 .unwrap_or(0),
234 "hostname": local_hostname::get()
235 .map(|h| h.to_string_lossy().to_string())
236 .unwrap_or_default()
237 });
238 let _ = file.write_all(lock_data.to_string().as_bytes());
239 self.lock_acquired = true;
240 return true;
241 }
242 Err(_) => {
243 thread::sleep(Duration::from_millis(100));
245 continue;
246 }
247 }
248 }
249
250 false
251 }
252
253 pub fn release(&mut self) {
255 if self.lock_acquired {
256 let _ = fs::remove_file(&self.lock_file_path);
257 self.lock_acquired = false;
258 }
259 }
260
261 fn read_lock_file(&self) -> Option<Value> {
263 let content = fs::read_to_string(&self.lock_file_path).ok()?;
264 serde_json::from_str(&content).ok()
265 }
266
267 fn is_lock_stale(&self, lock_data: &Value) -> bool {
269 let timestamp = lock_data.get("timestamp").and_then(|t| t.as_u64());
270
271 if let Some(ts) = timestamp {
273 let now = std::time::SystemTime::now()
274 .duration_since(std::time::UNIX_EPOCH)
275 .map(|d| d.as_millis() as u64)
276 .unwrap_or(0);
277 if now - ts > LOCK_STALE_MS {
278 return true;
279 }
280 } else {
281 return true;
282 }
283
284 #[cfg(unix)]
286 {
287 let pid = lock_data.get("pid").and_then(|p| p.as_u64());
288 if let Some(p) = pid {
289 let result = unsafe { libc::kill(p as i32, 0) };
291 if result != 0 {
292 return true; }
294 }
295 }
296
297 false
298 }
299}
300
301impl Drop for LockManager {
302 fn drop(&mut self) {
303 self.release();
304 }
305}
306
307pub fn is_clink_installed() -> bool {
309 Command::new("clink")
310 .arg("--version")
311 .stdout(Stdio::null())
312 .stderr(Stdio::null())
313 .status()
314 .map(|s| s.success())
315 .unwrap_or(false)
316}
317
318pub fn get_default_app_folder() -> PathBuf {
320 if let Ok(custom) = env::var("START_APP_FOLDER") {
321 return PathBuf::from(custom);
322 }
323
324 let home = env::var("HOME")
325 .or_else(|_| env::var("USERPROFILE"))
326 .unwrap_or_else(|_| ".".to_string());
327 PathBuf::from(home).join(DEFAULT_APP_FOLDER_NAME)
328}
329
330#[derive(Clone)]
332pub struct ExecutionStore {
333 app_folder: PathBuf,
334 lino_db_path: PathBuf,
335 links_db_path: PathBuf,
336 lock_file_path: PathBuf,
337 use_links: bool,
338 verbose: bool,
339}
340
341#[derive(Debug, Default)]
343pub struct ExecutionStoreOptions {
344 pub app_folder: Option<PathBuf>,
345 pub use_links: Option<bool>,
346 pub verbose: bool,
347}
348
349impl ExecutionStore {
350 pub fn new() -> Self {
352 Self::with_options(ExecutionStoreOptions::default())
353 }
354
355 pub fn with_options(options: ExecutionStoreOptions) -> Self {
357 let app_folder = options.app_folder.unwrap_or_else(get_default_app_folder);
358 let lino_db_path = app_folder.join(LINO_DB_FILE);
359 let links_db_path = app_folder.join(LINKS_DB_FILE);
360 let lock_file_path = app_folder.join(LOCK_FILE);
361 let use_links = options.use_links.unwrap_or_else(is_clink_installed);
362
363 let _ = fs::create_dir_all(&app_folder);
365
366 ExecutionStore {
367 app_folder,
368 lino_db_path,
369 links_db_path,
370 lock_file_path,
371 use_links,
372 verbose: options.verbose,
373 }
374 }
375
376 fn log(&self, message: &str) {
378 if self.verbose {
379 println!("[ExecutionStore] {}", message);
380 }
381 }
382
383 pub fn read_lino_records(&self) -> Vec<ExecutionRecord> {
385 if !self.lino_db_path.exists() {
386 return Vec::new();
387 }
388
389 match fs::read_to_string(&self.lino_db_path) {
390 Ok(content) => {
391 if content.trim().is_empty() {
392 return Vec::new();
393 }
394
395 match decode(&content) {
396 Ok(data) => {
397 if let LinoValue::Array(arr) = data {
398 arr.iter()
399 .map(lino_value_to_json)
400 .filter_map(|value| ExecutionRecord::from_json(&value))
401 .collect()
402 } else {
403 Vec::new()
404 }
405 }
406 Err(e) => {
407 self.log(&format!("Error decoding lino records: {}", e));
408 Vec::new()
409 }
410 }
411 }
412 Err(e) => {
413 self.log(&format!("Error reading lino records: {}", e));
414 Vec::new()
415 }
416 }
417 }
418
419 fn write_lino_records(&self, records: &[ExecutionRecord]) -> std::io::Result<()> {
421 let data: Vec<LinoValue> = records
422 .iter()
423 .map(|record| json_to_lino_value(&record.to_json()))
424 .collect();
425 let content = encode(&LinoValue::Array(data));
426 fs::write(&self.lino_db_path, content)?;
427 self.log(&format!("Wrote {} records to lino file", records.len()));
428 Ok(())
429 }
430
431 fn build_clink_create_query(&self, record: &ExecutionRecord) -> String {
433 let obj = record.to_json();
434 let mut links = Vec::new();
435
436 links.push(format!(
438 "({}: ExecutionRecord {})",
439 record.uuid, record.uuid
440 ));
441
442 if let Value::Object(map) = obj {
444 for (key, value) in map {
445 let escaped_value = match value {
446 Value::Object(_) | Value::Array(_) => {
447 serde_json::to_string(&value).unwrap_or_default()
448 }
449 Value::String(s) => s,
450 Value::Null => "null".to_string(),
451 other => other.to_string(),
452 };
453 links.push(format!(
454 "({}.{}: {} \"{}\")",
455 record.uuid,
456 key,
457 key,
458 escaped_value.replace('"', "\\\"")
459 ));
460 }
461 }
462
463 format!("() (({})))", links.join(") ("))
464 }
465
466 fn exec_clink(&self, query: &str) -> Result<String, String> {
468 match Command::new("clink")
469 .arg(query)
470 .arg("--db")
471 .arg(&self.links_db_path)
472 .output()
473 {
474 Ok(output) => {
475 if output.status.success() {
476 Ok(String::from_utf8_lossy(&output.stdout).to_string())
477 } else {
478 let stderr = String::from_utf8_lossy(&output.stderr).to_string();
479 self.log(&format!("Clink error: {}", stderr));
480 Err(stderr)
481 }
482 }
483 Err(e) => {
484 self.log(&format!("Clink execution error: {}", e));
485 Err(e.to_string())
486 }
487 }
488 }
489
490 fn write_links_record(&self, record: &ExecutionRecord) -> bool {
492 if !self.use_links {
493 return false;
494 }
495
496 let query = self.build_clink_create_query(record);
497 match self.exec_clink(&query) {
498 Ok(_) => {
499 self.log(&format!("Wrote record {} to links database", record.uuid));
500 true
501 }
502 Err(_) => false,
503 }
504 }
505
506 fn delete_links_record(&self, uuid: &str) -> bool {
508 if !self.use_links {
509 return false;
510 }
511
512 let query = format!("(($id: {} $any)) ()", uuid);
513 self.exec_clink(&query).is_ok()
514 }
515
516 pub fn save(&self, record: &ExecutionRecord) -> Result<(), String> {
518 let mut lock = LockManager::new(self.lock_file_path.clone());
519
520 if !lock.acquire(LOCK_TIMEOUT_MS) {
521 return Err("Failed to acquire lock for database write".to_string());
522 }
523
524 let mut records = self.read_lino_records();
526
527 let existing_index = records.iter().position(|r| r.uuid == record.uuid);
529
530 if let Some(idx) = existing_index {
531 records[idx] = record.clone();
533 } else {
534 records.push(record.clone());
536 }
537
538 self.write_lino_records(&records)
540 .map_err(|e| e.to_string())?;
541
542 if self.use_links {
544 self.write_links_record(record);
545 }
546
547 Ok(())
548 }
549
550 pub fn get(&self, identifier: &str) -> Option<ExecutionRecord> {
552 let records = self.read_lino_records();
553 records
554 .iter()
555 .find(|r| r.uuid == identifier)
556 .cloned()
557 .or_else(|| {
558 records
559 .iter()
560 .find(|r| {
561 r.options.get("sessionName").and_then(|v| v.as_str()) == Some(identifier)
562 })
563 .cloned()
564 })
565 .or_else(|| {
566 records.into_iter().find(|r| {
570 r.options
571 .get("sessionNameHistory")
572 .and_then(|v| v.as_array())
573 .is_some_and(|history| {
574 history.iter().any(|name| name.as_str() == Some(identifier))
575 })
576 })
577 })
578 }
579
580 pub fn get_all(&self) -> Vec<ExecutionRecord> {
582 self.read_lino_records()
583 }
584 pub fn get_by_status(&self, status: ExecutionStatus) -> Vec<ExecutionRecord> {
586 self.read_lino_records()
587 .into_iter()
588 .filter(|r| r.status == status)
589 .collect()
590 }
591 pub fn get_executing(&self) -> Vec<ExecutionRecord> {
593 self.get_by_status(ExecutionStatus::Executing)
594 }
595 pub fn get_recent(&self, limit: usize) -> Vec<ExecutionRecord> {
597 let mut records = self.read_lino_records();
598 records.sort_by(|a, b| b.start_time.cmp(&a.start_time));
599 records.truncate(limit);
600 records
601 }
602 pub fn cleanup_stale(&self, options: CleanupOptions) -> CleanupResult {
604 let max_age_ms = options.max_age_ms.unwrap_or(24 * 60 * 60 * 1000);
605 let dry_run = options.dry_run;
606 let mut result = CleanupResult {
607 cleaned: 0,
608 records: Vec::new(),
609 errors: Vec::new(),
610 };
611
612 let records = self.read_lino_records();
613 let mut stale_records: Vec<ExecutionRecord> = Vec::new();
614
615 for record in records
616 .iter()
617 .filter(|r| r.status == ExecutionStatus::Executing)
618 {
619 let mut is_stale = false;
620
621 #[cfg(unix)]
623 if let Some(pid) = record.pid {
624 if record.platform == std::env::consts::OS
625 && unsafe { libc::kill(pid as i32, 0) } != 0
626 {
627 is_stale = true;
628 self.log(&format!("Stale: {} (PID {} gone)", record.uuid, pid));
629 }
630 }
631
632 if !is_stale {
634 if let Ok(st) = chrono::DateTime::parse_from_rfc3339(&record.start_time) {
635 let age_ms =
636 (chrono::Utc::now() - st.with_timezone(&chrono::Utc)).num_milliseconds();
637 if age_ms > max_age_ms as i64 {
638 is_stale = true;
639 self.log(&format!(
640 "Stale: {} ({}min old)",
641 record.uuid,
642 age_ms / 60000
643 ));
644 }
645 }
646 }
647
648 if is_stale {
649 stale_records.push(record.clone());
650 }
651 }
652
653 result.records = stale_records.clone();
654
655 if !dry_run && !stale_records.is_empty() {
656 let mut lock = LockManager::new(self.lock_file_path.clone());
657 if !lock.acquire(LOCK_TIMEOUT_MS) {
658 result.errors.push("Failed to acquire lock".to_string());
659 return result;
660 }
661
662 let mut current = self.read_lino_records();
663 for stale in &stale_records {
664 if let Some(i) = current.iter().position(|r| r.uuid == stale.uuid) {
665 current[i].status = ExecutionStatus::Executed;
666 current[i].exit_code = Some(-1);
667 current[i].end_time = Some(chrono::Utc::now().to_rfc3339());
668 result.cleaned += 1;
669 }
670 }
671
672 if let Err(e) = self.write_lino_records(¤t) {
673 result.errors.push(format!("Cleanup error: {}", e));
674 } else {
675 self.log(&format!("Cleaned {} stale records", result.cleaned));
676 }
677 } else if dry_run {
678 result.cleaned = stale_records.len();
679 }
680
681 result
682 }
683
684 pub fn delete(&self, uuid: &str) -> Result<bool, String> {
686 let mut lock = LockManager::new(self.lock_file_path.clone());
687
688 if !lock.acquire(LOCK_TIMEOUT_MS) {
689 return Err("Failed to acquire lock for database write".to_string());
690 }
691
692 let records = self.read_lino_records();
693 let filtered: Vec<_> = records.iter().filter(|r| r.uuid != uuid).cloned().collect();
694
695 if filtered.len() == records.len() {
696 return Ok(false); }
698
699 self.write_lino_records(&filtered)
700 .map_err(|e| e.to_string())?;
701
702 if self.use_links {
704 self.delete_links_record(uuid);
705 }
706
707 Ok(true)
708 }
709
710 pub fn clear(&self) -> Result<(), String> {
712 let mut lock = LockManager::new(self.lock_file_path.clone());
713
714 if !lock.acquire(LOCK_TIMEOUT_MS) {
715 return Err("Failed to acquire lock for database write".to_string());
716 }
717
718 self.write_lino_records(&[]).map_err(|e| e.to_string())?;
719
720 if self.use_links && self.links_db_path.exists() {
722 let _ = fs::remove_file(&self.links_db_path);
723 }
724
725 Ok(())
726 }
727
728 pub fn verify_consistency(&self) -> ConsistencyResult {
730 let mut result = ConsistencyResult {
731 consistent: true,
732 lino_count: 0,
733 links_count: 0,
734 errors: Vec::new(),
735 };
736
737 let lino_records = self.read_lino_records();
739 result.lino_count = lino_records.len();
740
741 if !self.use_links {
742 result
743 .errors
744 .push("clink not installed - links database not available".to_string());
745 return result;
746 }
747
748 match self.exec_clink("((($id: ExecutionRecord $uuid)) (($id: ExecutionRecord $uuid)))") {
750 Ok(output) => {
751 let re = regex::Regex::new(r"ExecutionRecord\s+([a-f0-9-]{36})").unwrap();
753 let uuids: std::collections::HashSet<_> = re
754 .captures_iter(&output)
755 .filter_map(|c| c.get(1).map(|m| m.as_str().to_lowercase()))
756 .collect();
757 result.links_count = uuids.len();
758
759 if result.lino_count != result.links_count {
761 result.consistent = false;
762 result.errors.push(format!(
763 "Record count mismatch: lino={}, links={}",
764 result.lino_count, result.links_count
765 ));
766 }
767
768 for record in &lino_records {
770 if !uuids.contains(&record.uuid.to_lowercase()) {
771 result.consistent = false;
772 result.errors.push(format!(
773 "Record {} missing from links database",
774 record.uuid
775 ));
776 }
777 }
778 }
779 Err(e) => {
780 result
781 .errors
782 .push(format!("Failed to query links database: {}", e));
783 result.consistent = false;
784 }
785 }
786
787 result
788 }
789
790 pub fn get_stats(&self) -> ExecutionStats {
792 let records = self.read_lino_records();
793 let executing = records
794 .iter()
795 .filter(|r| r.status == ExecutionStatus::Executing)
796 .count();
797 let executed = records
798 .iter()
799 .filter(|r| r.status == ExecutionStatus::Executed)
800 .count();
801 let successful = records
802 .iter()
803 .filter(|r| r.status == ExecutionStatus::Executed && r.exit_code == Some(0))
804 .count();
805 let failed = records
806 .iter()
807 .filter(|r| {
808 r.status == ExecutionStatus::Executed
809 && r.exit_code.map(|c| c != 0).unwrap_or(false)
810 })
811 .count();
812
813 ExecutionStats {
814 total: records.len(),
815 executing,
816 executed,
817 successful,
818 failed,
819 clink_available: self.use_links,
820 lino_db_path: self.lino_db_path.to_string_lossy().to_string(),
821 links_db_path: self.links_db_path.to_string_lossy().to_string(),
822 }
823 }
824
825 pub fn app_folder(&self) -> &Path {
827 &self.app_folder
828 }
829}
830
831impl Default for ExecutionStore {
832 fn default() -> Self {
833 Self::new()
834 }
835}
836
837#[derive(Debug)]
839pub struct ConsistencyResult {
840 pub consistent: bool,
841 pub lino_count: usize,
842 pub links_count: usize,
843 pub errors: Vec<String>,
844}
845
846#[derive(Debug, Default)]
848pub struct CleanupOptions {
849 pub max_age_ms: Option<u64>,
850 pub dry_run: bool,
851}
852
853#[derive(Debug)]
855pub struct CleanupResult {
856 pub cleaned: usize,
857 pub records: Vec<ExecutionRecord>,
858 pub errors: Vec<String>,
859}
860
861#[derive(Debug)]
863pub struct ExecutionStats {
864 pub total: usize,
865 pub executing: usize,
866 pub executed: usize,
867 pub successful: usize,
868 pub failed: usize,
869 pub clink_available: bool,
870 pub lino_db_path: String,
871 pub links_db_path: String,
872}
873
874#[cfg(test)]
875#[path = "execution_store_cases.rs"]
876mod tests;