use std::sync::{Mutex, OnceLock};
use crate::routine_storage::read_routine_crons;
use crate::routines::{load_agent_command, shell_quote, Routine, RoutineStore};
use crate::sync::{read_crontab, replace_block_with, to_os_schedule, write_crontab, SyncError};
use crate::utils::cron::{normalize_schedule, validate_cron};
use crate::utils::lock::LockRecover;
fn crontab_sync_lock() -> &'static Mutex<()> {
static LOCK: OnceLock<Mutex<()>> = OnceLock::new();
LOCK.get_or_init(|| Mutex::new(()))
}
pub(crate) const BLOCK_BEGIN: &str = "# BEGIN MOADIM-ROUTINES";
pub(crate) const BLOCK_END: &str = "# END MOADIM-ROUTINES";
const BLOCK_HEADER: &str = "# Managed by moadim — routines (agent tmux sessions)";
pub(crate) fn format_routine_line_for_schedule(routine: &Routine, schedule: &str) -> String {
#[allow(
clippy::expect_used,
reason = "the daemon is already running from this binary, so resolving its own path \
cannot realistically fail; a failure here means the process has no executable \
path at all, which is unrecoverable, and this fn has no `Result` to propagate \
through short of reshaping every crontab-formatting caller"
)]
let exe = std::env::current_exe().expect("daemon executable path is resolvable");
let schedule = to_os_schedule(schedule);
format!(
"{} {} schedule trigger {} # moadim-routine:{}",
schedule,
shell_quote(&exe.to_string_lossy()),
shell_quote(&routine.id),
routine.id
)
}
#[cfg(test)]
pub(crate) fn format_routine_line(routine: &Routine) -> String {
format_routine_line_for_schedule(routine, &routine.schedule)
}
fn pure_schedules_for_crontab(routine: &Routine) -> Vec<String> {
let rel_dir = crate::routine_storage::routine_rel_dir(routine);
let entries = read_routine_crons(&crate::paths::routine_dir(&rel_dir).join("schedule.cron"));
if entries.is_empty() {
vec![routine.schedule.clone()]
} else {
entries
}
}
fn compailed_schedules_for_crontab(routine: &Routine, pure_schedules: &[String]) -> Vec<String> {
let schedules: Vec<String> = pure_schedules
.iter()
.map(|entry| normalize_schedule(entry))
.filter(|schedule| match validate_cron(schedule) {
Ok(()) => true,
Err(err) => {
log::warn!(
"routine sync: invalid schedule.cron entry {:?} for routine {:?}: {}; skipping",
schedule,
routine.id,
err
);
false
}
})
.collect();
if schedules.is_empty() {
return vec![routine.schedule.clone()];
}
let refs: Vec<&str> = schedules.iter().map(String::as_str).collect();
match cron_union::union(refs) {
Ok(union) => union.iter().map(ToString::to_string).collect(),
Err(err) => {
log::warn!(
"routine sync: cron-union could not compile schedules for routine {:?}: {}; \
using validated schedules without dedupe",
routine.id,
err
);
schedules
}
}
}
fn write_compailed_cron_sidecar(routine: &Routine, schedules: &[String]) {
let rel_dir = crate::routine_storage::routine_rel_dir(routine);
let dir = crate::paths::routine_dir(&rel_dir);
let path = crate::paths::routine_compailed_cron_path(&rel_dir);
let legacy_path = dir.join(".compailed.cron");
let mut text = schedules.join("\n");
text.push('\n');
let _ = crate::utils::fs_perms::create_private_dir_all(&dir);
let _ = std::fs::write(&path, text);
if legacy_path != path && legacy_path.exists() {
let _ = std::fs::remove_file(legacy_path);
}
}
fn build_block(store: &RoutineStore) -> String {
if crate::global_lock::is_globally_locked() {
log::info!("routine sync: global lock active — clearing all routine crontab lines");
return format!("{BLOCK_BEGIN}\n{BLOCK_HEADER}\n{BLOCK_END}");
}
let me = crate::machine::current_machine();
let mut routines: Vec<Routine> = {
let lock = store.lock_recover();
lock.values()
.filter(|routine| routine.source == "managed" && routine.enabled)
.cloned()
.collect()
};
warn_dormant_routines(&routines);
routines.retain(|routine| crate::machine::targets(&routine.machines, &me));
routines.sort_by(|left, right| {
left.created_at
.cmp(&right.created_at)
.then_with(|| left.id.cmp(&right.id))
});
let lines: Vec<String> = routines
.iter()
.filter_map(|routine| match load_agent_command(&routine.agent) {
Ok(_) => Some({
let pure_schedules = pure_schedules_for_crontab(routine);
let compailed_schedules = compailed_schedules_for_crontab(routine, &pure_schedules);
write_compailed_cron_sidecar(routine, &compailed_schedules);
compailed_schedules
.iter()
.map(|schedule| format_routine_line_for_schedule(routine, schedule))
.collect::<Vec<_>>()
}),
Err(err) => {
log::warn!(
"routine sync: cannot load agent {:?} ({}) for routine {:?}; skipping",
routine.agent,
err,
routine.id
);
None
}
})
.flatten()
.collect();
if lines.is_empty() {
format!("{BLOCK_BEGIN}\n{BLOCK_HEADER}\n{BLOCK_END}")
} else {
format!(
"{BLOCK_BEGIN}\n{BLOCK_HEADER}\n{}\n{BLOCK_END}",
lines.join("\n")
)
}
}
fn warn_dormant_routines(routines: &[Routine]) {
let dormant: Vec<&str> = routines
.iter()
.filter(|routine| routine.machines.is_empty())
.map(|routine| routine.title.as_str())
.collect();
if !dormant.is_empty() {
log::warn!(
"{} enabled routine(s) have no machine assignment and will not be scheduled on any \
machine: {}; assign with `moadim routines update <id> --machines '[\"<name>\"]'`",
dormant.len(),
dormant.join(", ")
);
}
}
pub(crate) const ROUTINE_LINE_MARKER: &str = "# moadim-routine:";
pub fn sync_routines_to_crontab(store: &RoutineStore) -> Result<(), SyncError> {
let on_multi_thread_runtime = tokio::runtime::Handle::try_current()
.is_ok_and(|handle| handle.runtime_flavor() == tokio::runtime::RuntimeFlavor::MultiThread);
let result = if on_multi_thread_runtime {
tokio::task::block_in_place(|| sync_routines_to_crontab_blocking(store))
} else {
sync_routines_to_crontab_blocking(store)
};
match &result {
Ok(()) => crate::sync::record_crontab_sync_success(),
Err(err) => crate::sync::record_crontab_sync_failure(err),
}
result
}
fn sync_routines_to_crontab_blocking(store: &RoutineStore) -> Result<(), SyncError> {
let _crontab_guard = crontab_sync_lock().lock_recover();
let current = read_crontab()?;
if store.lock_recover().is_empty() && current.contains(ROUTINE_LINE_MARKER) {
log::warn!(
"routine sync: store is empty but the crontab still has routine lines; refusing to \
wipe the routines block (suspected load failure or a concurrent daemon)"
);
return Ok(());
}
let block = build_block(store);
let new_crontab = replace_block_with(¤t, &block, BLOCK_BEGIN, BLOCK_END);
if new_crontab == current {
return Ok(());
}
write_crontab(&new_crontab)
}
#[cfg(test)]
#[path = "routines_sync_tests.rs"]
mod routines_sync_tests;
#[cfg(test)]
#[path = "routines_sync_status_tests.rs"]
mod routines_sync_status_tests;
#[cfg(test)]
#[path = "routines_sync_multi_cron_tests.rs"]
mod routines_sync_multi_cron_tests;