jan-cli 0.20.0

//! Background scheduler daemon for `jan cron`.
//!
//! The daemon ticks every 100 ms, evaluates `cron:` schedules from the preferred
//! YAML tree, and runs matching script `run` leaves. `jan cron` acts as a client
//! front end (similar to `./gradlew` and the Gradle daemon).

use std::collections::HashMap;
use std::fs;
use std::io::{BufRead, BufReader, Write};
use std::os::unix::net::{UnixListener, UnixStream};
use std::path::{Path, PathBuf};
use std::process::{Command, Stdio};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::{Duration, Instant};

use anyhow::{bail, Context, Result};
use serde::{Deserialize, Serialize};

use crate::config;
use crate::cron::{CronExpr, TickTime, TICK_MS};
use crate::inspect::{collect_scripts, ScriptEntry};
use crate::{load_spec, resolve_preferred_spec};

const SERVICE_NAME: &str = "jan-cron";
const SOCKET_FILE: &str = "cron.sock";
const PID_FILE: &str = "cron.pid";

static STOP_REQUESTED: AtomicBool = AtomicBool::new(false);

/// Directory for the control socket and pid file.
pub fn runtime_dir() -> PathBuf {
    if let Ok(dir) = std::env::var("XDG_RUNTIME_DIR") {
        let dir = dir.trim();
        if !dir.is_empty() {
            return PathBuf::from(dir).join("jan-cli");
        }
    }
    config::config_dir().join("run")
}

pub fn socket_path() -> PathBuf {
    runtime_dir().join(SOCKET_FILE)
}

pub fn pid_path() -> PathBuf {
    runtime_dir().join(PID_FILE)
}

fn systemd_unit_path() -> PathBuf {
    dirs::home_dir()
        .unwrap_or_else(|| PathBuf::from("."))
        .join(".config/systemd/user")
        .join(format!("{SERVICE_NAME}.service"))
}

fn jan_bin() -> Result<PathBuf> {
    let exe = std::env::current_exe().context("resolve jan binary path")?;
    exe.canonicalize()
        .with_context(|| format!("canonicalize {}", exe.display()))
}

fn write_pid_file() -> Result<()> {
    let dir = runtime_dir();
    fs::create_dir_all(&dir).with_context(|| format!("create {}", dir.display()))?;
    fs::write(pid_path(), format!("{}\n", std::process::id()))
        .with_context(|| format!("write {}", pid_path().display()))?;
    Ok(())
}

fn remove_runtime_files() {
    let _ = fs::remove_file(socket_path());
    let _ = fs::remove_file(pid_path());
}

/// Load schedules once from the preferred tree. Disk I/O happens only here
/// (startup / explicit `jan cron refresh`), never on the tick path.
fn load_schedule_cache() -> Result<ScheduleCache> {
    let (spec_path, identity) = resolve_preferred_spec()?;
    let spec = load_spec(&spec_path).with_context(|| format!("load {}", spec_path.display()))?;
    let scripts = collect_scripts(&spec);
    let mut schedules = Vec::new();
    for s in scripts.into_iter().filter(|s| !s.cron.is_empty()) {
        schedules.push(ParsedSchedule::try_from_entry(&s)?);
    }
    schedules.sort_by(|a, b| a.chain.cmp(&b.chain));
    Ok(ScheduleCache {
        schedules,
        spec_dir: identity.spec_dir,
        root_yaml: identity.root_yaml,
        loaded_at: Instant::now(),
    })
}

#[derive(Debug, Clone)]
struct ParsedSchedule {
    chain: Vec<String>,
    /// Original YAML expressions (for status / debugging).
    cron_raw: Vec<String>,
    /// Parsed once at refresh — tick matching never re-parses or hits disk.
    exprs: Vec<CronExpr>,
}

impl ParsedSchedule {
    fn try_from_entry(s: &ScriptEntry) -> Result<Self> {
        let mut exprs = Vec::with_capacity(s.cron.len());
        for raw in &s.cron {
            exprs.push(
                CronExpr::parse(raw)
                    .with_context(|| format!("{}: invalid cron `{raw}`", s.chain.join(" ")))?,
            );
        }
        Ok(Self {
            chain: s.chain.clone(),
            cron_raw: s.cron.clone(),
            exprs,
        })
    }

    fn matches_tick(&self, now: &TickTime) -> bool {
        self.exprs.iter().any(|e| e.matches_tick(now))
    }
}

/// One detected `cron:` script, as stored in the daemon cache.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct CachedCronEntry {
    pub chain: Vec<String>,
    pub cron: Vec<String>,
}

impl CachedCronEntry {
    pub fn chain_str(&self) -> String {
        self.chain.join(" ")
    }
}

#[derive(Debug)]
struct ScheduleCache {
    schedules: Vec<ParsedSchedule>,
    spec_dir: String,
    root_yaml: String,
    loaded_at: Instant,
}

#[derive(Debug)]
struct DaemonState {
    cache: ScheduleCache,
    last_fired: HashMap<String, TickTime>,
}

impl DaemonState {
    fn empty() -> Self {
        Self {
            cache: ScheduleCache {
                schedules: Vec::new(),
                spec_dir: String::new(),
                root_yaml: String::new(),
                loaded_at: Instant::now(),
            },
            last_fired: HashMap::new(),
        }
    }

    /// Replace the in-memory schedule cache from disk. Called only on startup
    /// and explicit refresh — never from the tick loop.
    fn refresh(&mut self) -> Result<usize> {
        self.cache = load_schedule_cache()?;
        self.last_fired.clear();
        Ok(self.cache.schedules.len())
    }

    fn tick(&mut self, now: &TickTime, jan: &Path, verbose: bool) {
        // Tick path: memory only. No YAML load, no CronExpr::parse.
        for sched in &self.cache.schedules {
            let chain_key = sched.chain.join(" ");
            if !sched.matches_tick(now) {
                continue;
            }
            if self
                .last_fired
                .get(&chain_key)
                .is_some_and(|prev| prev == now)
            {
                continue;
            }
            self.last_fired.insert(chain_key.clone(), *now);
            spawn_run(jan, &sched.chain, verbose);
        }
    }

    fn status_line(&self, started: Instant) -> String {
        let exprs: usize = self.cache.schedules.iter().map(|s| s.cron_raw.len()).sum();
        format!(
            "ok pid={} uptime_s={} schedules={} exprs={} cached=1 age_s={} spec={}/{}",
            std::process::id(),
            started.elapsed().as_secs(),
            self.cache.schedules.len(),
            exprs,
            self.cache.loaded_at.elapsed().as_secs(),
            self.cache.spec_dir,
            self.cache.root_yaml
        )
    }

    fn cached_entries(&self) -> Vec<CachedCronEntry> {
        self.cache
            .schedules
            .iter()
            .map(|s| CachedCronEntry {
                chain: s.chain.clone(),
                cron: s.cron_raw.clone(),
            })
            .collect()
    }
}

fn spawn_run(jan: &Path, chain: &[String], verbose: bool) {
    let mut args = vec!["--no-log".to_string()];
    args.extend(chain.iter().cloned());
    args.push("run".into());
    if verbose {
        eprintln!("jan cron daemon: spawn `{} {}`", jan.display(), args.join(" "));
    }
    match Command::new(jan)
        .args(&args)
        .stdin(Stdio::null())
        .stdout(Stdio::null())
        .stderr(Stdio::inherit())
        .spawn()
    {
        Ok(_child) => {}
        Err(e) => {
            eprintln!(
                "jan cron daemon: failed to spawn `{} {}`: {e:#}",
                jan.display(),
                args.join(" ")
            );
        }
    }
}

fn handle_client(mut stream: UnixStream, state: Arc<Mutex<DaemonState>>, started: Instant) {
    let reader = BufReader::new(
        stream
            .try_clone()
            .unwrap_or_else(|_| stream.try_clone().expect("clone daemon control socket")),
    );
    let line = match reader.lines().next() {
        Some(Ok(l)) => l,
        _ => return,
    };
    let reply = match line.trim().to_ascii_lowercase().as_str() {
        "ping" => "pong".to_string(),
        "status" => state.lock().unwrap().status_line(started),
        "reload" | "refresh" => match state.lock().unwrap().refresh() {
            Ok(n) => format!("ok schedules={n}"),
            Err(e) => format!("error {e:#}"),
        },
        "list" => match serde_json::to_string(&state.lock().unwrap().cached_entries()) {
            Ok(json) => format!("ok list {json}"),
            Err(e) => format!("error {e}"),
        },
        "stop" => {
            STOP_REQUESTED.store(true, Ordering::SeqCst);
            "ok stopping".to_string()
        }
        other => format!("error unknown command `{other}`"),
    };
    let _ = writeln!(stream, "{reply}");
}

fn accept_control(state: Arc<Mutex<DaemonState>>, started: Instant) {
    let listener = match UnixListener::bind(&socket_path()) {
        Ok(l) => l,
        Err(e) => {
            eprintln!(
                "jan cron daemon: bind {}: {e:#}",
                socket_path().display()
            );
            return;
        }
    };
    if let Err(e) = listener.set_nonblocking(true) {
        eprintln!("jan cron daemon: set_nonblocking: {e:#}");
        return;
    }
    while !STOP_REQUESTED.load(Ordering::SeqCst) {
        match listener.accept() {
            Ok((stream, _)) => {
                let state = Arc::clone(&state);
                thread::spawn(move || handle_client(stream, state, started));
            }
            Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => {
                thread::sleep(Duration::from_millis(50));
            }
            Err(e) => {
                eprintln!("jan cron daemon: accept: {e:#}");
                thread::sleep(Duration::from_millis(100));
            }
        }
    }
}

fn sleep_until_next_tick(start: Instant, tick_index: u64) {
    let target = start + Duration::from_millis(tick_index * TICK_MS);
    let now = Instant::now();
    if target > now {
        thread::sleep(target - now);
    }
}

/// Run the scheduler loop in the foreground (used by systemd and `jan cron daemon`).
pub fn run_foreground(verbose: bool) -> Result<i32> {
    STOP_REQUESTED.store(false, Ordering::SeqCst);
    fs::create_dir_all(runtime_dir()).with_context(|| format!("create {}", runtime_dir().display()))?;
    remove_runtime_files();
    write_pid_file()?;
    let jan = jan_bin()?;

    let state = Arc::new(Mutex::new(DaemonState::empty()));
    {
        let n = state.lock().unwrap().refresh().context("load cron schedules")?;
        if verbose {
            eprintln!("jan cron daemon: cached {n} scheduled script(s) (refresh to reload)");
        }
    }

    let started = Instant::now();
    let state_bg = Arc::clone(&state);
    let control = thread::spawn(move || accept_control(state_bg, started));

    let loop_start = Instant::now();
    let mut tick_index = 0u64;
    while !STOP_REQUESTED.load(Ordering::SeqCst) {
        sleep_until_next_tick(loop_start, tick_index);
        let now = TickTime::now_local();
        state.lock().unwrap().tick(&now, &jan, verbose);
        tick_index += 1;
    }

    remove_runtime_files();
    let _ = control.join();
    Ok(0)
}

pub fn send_command(cmd: &str) -> Result<String> {
    let path = socket_path();
    if !path.exists() {
        bail!("jan cron daemon is not running (no socket at {})", path.display());
    }
    let mut stream = UnixStream::connect(&path)
        .with_context(|| format!("connect to {}", path.display()))?;
    stream.set_read_timeout(Some(Duration::from_secs(5)))?;
    stream.set_write_timeout(Some(Duration::from_secs(5)))?;
    writeln!(stream, "{cmd}").context("write daemon command")?;
    let mut reader = BufReader::new(stream);
    let mut reply = String::new();
    reader.read_line(&mut reply).context("read daemon reply")?;
    Ok(reply.trim().to_string())
}

/// Cached cron entries from the running daemon (no YAML walk).
pub fn fetch_cached_entries() -> Result<Vec<CachedCronEntry>> {
    let reply = send_command("list").with_context(|| {
        "jan cron --list reads the daemon cache; run `jan cron start` first"
    })?;
    if let Some(rest) = reply.strip_prefix("error ") {
        if rest.contains("unknown command") && rest.contains("list") {
            bail!(
                "jan cron daemon is outdated (no `list` cache command); \
                 run `jan cron stop` then `jan cron start` with this jan binary"
            );
        }
        bail!("daemon list failed: {rest}");
    }
    let json = reply.strip_prefix("ok list ").ok_or_else(|| {
        anyhow::anyhow!("unexpected daemon list reply: {reply}")
    })?;
    serde_json::from_str(json).with_context(|| format!("parse daemon list JSON: {json}"))
}

pub fn daemon_running() -> bool {
    match send_command("ping") {
        Ok(r) => r == "pong",
        Err(_) => false,
    }
}

pub fn start_daemon_background(verbose: bool) -> Result<()> {
    if daemon_running() {
        return Ok(());
    }
    let jan = jan_bin()?;
    let mut cmd = Command::new(&jan);
    cmd.args(["--no-log", "cron", "daemon"]);
    if verbose {
        cmd.arg("-v");
    }
    cmd.stdin(Stdio::null())
        .stdout(Stdio::null())
        .stderr(Stdio::null());
    cmd.spawn()
        .with_context(|| format!("spawn `{} cron daemon`", jan.display()))?;
    for _ in 0..50 {
        if daemon_running() {
            return Ok(());
        }
        thread::sleep(Duration::from_millis(100));
    }
    bail!(
        "jan cron daemon failed to start (no response on {})",
        socket_path().display()
    );
}

pub fn stop_daemon() -> Result<()> {
    if !socket_path().exists() {
        println!("(jan cron daemon is not running)");
        return Ok(());
    }
    let reply = send_command("stop")?;
    for _ in 0..30 {
        if !socket_path().exists() {
            println!("stopped jan cron daemon");
            return Ok(());
        }
        thread::sleep(Duration::from_millis(100));
    }
    bail!("jan cron daemon did not stop: {reply}");
}

pub fn daemon_status() -> Result<i32> {
    match send_command("status") {
        Ok(reply) if reply.starts_with("ok ") => {
            println!("jan cron daemon running ({reply})");
            Ok(0)
        }
        Ok(reply) => {
            println!("jan cron daemon: {reply}");
            Ok(1)
        }
        Err(e) => {
            println!("jan cron daemon is not running ({e:#})");
            Ok(1)
        }
    }
}

pub fn reload_daemon() -> Result<()> {
    let reply = send_command("refresh")?;
    if reply.starts_with("ok ") {
        println!("refreshed jan cron schedule cache ({reply})");
        Ok(())
    } else {
        bail!("refresh failed: {reply}");
    }
}

fn systemd_unit_body(jan: &Path) -> String {
    format!(
        r#"[Unit]
Description=Jan cron scheduler daemon (100ms ticks)
After=default.target

[Service]
Type=simple
ExecStart={} --no-log cron daemon --foreground
Restart=on-failure
RestartSec=5

[Install]
WantedBy=default.target
"#,
        jan.display()
    )
}

pub fn install_systemd(dry_run: bool) -> Result<i32> {
    let jan = jan_bin()?;
    let unit_path = systemd_unit_path();
    let body = systemd_unit_body(&jan);
    if dry_run {
        println!("# dry-run: would write {}:", unit_path.display());
        print!("{body}");
        println!("# dry-run: would run systemctl --user daemon-reload");
        println!("# dry-run: would run systemctl --user enable --now {SERVICE_NAME}.service");
        return Ok(0);
    }
    if let Some(parent) = unit_path.parent() {
        fs::create_dir_all(parent)
            .with_context(|| format!("create {}", parent.display()))?;
    }
    fs::write(&unit_path, &body)
        .with_context(|| format!("write {}", unit_path.display()))?;
    run_systemctl(&["--user", "daemon-reload"])?;
    run_systemctl(&["--user", "enable", "--now", &format!("{SERVICE_NAME}.service")])?;
    println!(
        "installed and started {} (unit: {})",
        SERVICE_NAME,
        unit_path.display()
    );
    println!(
        "  ExecStart={} --no-log cron daemon --foreground",
        jan.display()
    );
    Ok(0)
}

pub fn uninstall_systemd(dry_run: bool) -> Result<i32> {
    let unit_path = systemd_unit_path();
    if dry_run {
        println!("# dry-run: would run systemctl --user disable --now {SERVICE_NAME}.service");
        if unit_path.is_file() {
            println!("# dry-run: would remove {}", unit_path.display());
        }
        return Ok(0);
    }
    let _ = run_systemctl(&["--user", "stop", &format!("{SERVICE_NAME}.service")]);
    let _ = run_systemctl(&["--user", "disable", &format!("{SERVICE_NAME}.service")]);
    if unit_path.is_file() {
        fs::remove_file(&unit_path)
            .with_context(|| format!("remove {}", unit_path.display()))?;
    }
    let _ = run_systemctl(&["--user", "daemon-reload"]);
    stop_daemon().ok();
    println!("removed {SERVICE_NAME} systemd user service");
    Ok(0)
}

fn run_systemctl(args: &[&str]) -> Result<()> {
    let out = Command::new("systemctl")
        .args(args)
        .output()
        .with_context(|| format!("spawn systemctl {}", args.join(" ")))?;
    if !out.status.success() {
        bail!(
            "systemctl {} failed ({}): {}",
            args.join(" "),
            out.status,
            String::from_utf8_lossy(&out.stderr).trim()
        );
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::cron::{CronExpr, TickTime};

    #[test]
    fn systemd_unit_contains_execstart() {
        let body = systemd_unit_body(Path::new("/usr/local/bin/jan"));
        assert!(body.contains("ExecStart=/usr/local/bin/jan --no-log cron daemon --foreground"));
        assert!(body.contains("WantedBy=default.target"));
    }

    #[test]
    fn cached_schedule_matches_without_reparsing() {
        let sched = ParsedSchedule {
            chain: vec!["scripts".into(), "misc".into(), "tick".into()],
            cron_raw: vec!["* * * * * *".into()],
            exprs: vec![CronExpr::parse("* * * * * *").unwrap()],
        };
        let now = TickTime {
            decisecond: 0,
            second: 12,
            minute: 30,
            hour: 10,
            day: 5,
            month: 8,
            dow: 2,
        };
        assert!(sched.matches_tick(&now));
        let mid = TickTime {
            decisecond: 5,
            ..now
        };
        // Six-field schedules fire only at decisecond 0.
        assert!(!sched.matches_tick(&mid));
    }

    #[test]
    fn cached_entries_serialize_roundtrip() {
        let entries = vec![CachedCronEntry {
            chain: vec!["scripts".into(), "misc".into(), "tick".into()],
            cron: vec!["30 10 * * *".into(), "* * * * * *".into()],
        }];
        let json = serde_json::to_string(&entries).unwrap();
        let back: Vec<CachedCronEntry> = serde_json::from_str(&json).unwrap();
        assert_eq!(back, entries);
        assert_eq!(back[0].chain_str(), "scripts misc tick");
    }
}