areev_trigger/

render.rs

//! `areev trigger render` — emit scheduler config, never create it.
//!
//! These are **templates, not API clients**. Areev does not hold cloud
//! credentials and does not create cloud resources: provisioning a Cloud
//! Scheduler job or an EventBridge rule needs IAM permission to create
//! resources, which is a large privilege escalation for a memory engine, and it
//! buys a multi-cloud abstraction treadmill for very little differentiation.
//! You apply the output with whatever you already use for infrastructure.
//!
//! ## The heartbeat is deliberately coarse
//!
//! The rendered interval is the greatest common divisor of the declared
//! intervals, floored at 60s — not the shortest interval. The memory owns the
//! real cadence, so the heartbeat only has to be fine enough not to add
//! meaningful lag. Rendering a 30-second cron because one trigger asked for 30
//! seconds would put the schedule back in the crontab, which is the thing this
//! feature exists to stop.

use areev_core::types::{Trigger, TriggerKind};

use crate::error::{Result, TriggerError};

/// Never render a heartbeat finer than this. cron cannot express sub-minute
/// schedules at all, and a finer systemd timer mostly buys wakeups.
const MIN_HEARTBEAT_SECS: u64 = 60;

/// Pick the heartbeat for a set of declarations.
pub fn heartbeat_secs(triggers: &[Trigger]) -> u64 {
    let intervals: Vec<u64> = triggers
        .iter()
        .filter(|t| t.enabled)
        .filter_map(|t| match t.kind {
            TriggerKind::Interval | TriggerKind::Polling => t.interval_secs.map(u64::from),
            // A cron trigger's cadence is not a period, so it cannot contribute
            // a divisor. The floor covers it: cron has minute resolution.
            _ => None,
        })
        .filter(|s| *s > 0)
        .collect();

    let g = intervals.into_iter().reduce(gcd).unwrap_or(MIN_HEARTBEAT_SECS);
    g.max(MIN_HEARTBEAT_SECS)
}

fn gcd(a: u64, b: u64) -> u64 {
    if b == 0 {
        a
    } else {
        gcd(b, a % b)
    }
}

/// What the rendered config should invoke.
pub struct RenderContext<'a> {
    pub exe: &'a str,
    pub db: &'a str,
    pub ns: &'a str,
    pub heartbeat_secs: u64,
    /// Extra flags to pass through, e.g. `--connector-cmd ./gmail.sh`.
    pub extra_args: &'a str,
}

/// What `command[0]` must be inside a container image.
///
/// The host targets (`cron`, `launchd`, `systemd`) genuinely run on the machine
/// that produced the render, so the authoring binary's absolute path is right
/// there. A container runs the *image's* binary, so the only correct value is
/// the name on `PATH` inside it — `current_exe()` is the authoring host's path
/// and is guaranteed wrong.
const IMAGE_EXE: &str = "areev";

impl RenderContext<'_> {
    fn command(&self) -> String {
        self.command_as(self.exe)
    }

    /// [`command`](Self::command) for the container targets. See [`IMAGE_EXE`].
    fn container_command(&self) -> String {
        self.command_as(IMAGE_EXE)
    }

    fn command_as(&self, exe: &str) -> String {
        let mut c = format!("{} trigger run --db {} --ns {}", exe, self.db, self.ns);
        if !self.extra_args.trim().is_empty() {
            c.push(' ');
            c.push_str(self.extra_args.trim());
        }
        c
    }
}

/// Render config for `target`.
pub fn render(target: &str, ctx: &RenderContext<'_>) -> Result<String> {
    match target {
        "cron" => Ok(cron(ctx)),
        "launchd" => Ok(launchd(ctx)),
        "systemd" => Ok(systemd(ctx)),
        "k8s-cronjob" => Ok(k8s(ctx)),
        other => Err(TriggerError::UnknownTarget { target: other.to_string() }),
    }
}

/// Every target `render` accepts.
pub const TARGETS: &[&str] = &["cron", "launchd", "systemd", "k8s-cronjob"];

fn cron_expression(secs: u64) -> String {
    // cron has minute resolution and no "every N seconds".
    let mins = (secs / 60).max(1);
    if mins == 1 {
        "* * * * *".to_string()
    } else if mins < 60 && 60 % mins == 0 {
        format!("*/{mins} * * * *")
    } else if mins.is_multiple_of(60) && (mins / 60) < 24 {
        format!("0 */{} * * *", mins / 60)
    } else {
        // A period that does not divide an hour would drift or double-fire at
        // the wrap. Round down to a divisor rather than emit something subtly
        // wrong — the memory decides what is due anyway, so a finer heartbeat
        // is harmless and a wrong one is not.
        let divisor = [30u64, 20, 15, 12, 10, 6, 5, 4, 3, 2, 1]
            .into_iter()
            .find(|d| *d <= mins)
            .unwrap_or(1);
        format!("*/{divisor} * * * *")
    }
}

fn cron(ctx: &RenderContext<'_>) -> String {
    format!(
        "# areev triggers — the memory owns the real cadence; this is only a heartbeat.\n\
         # Generated by `areev trigger render --target cron`. Apply with `crontab -e`.\n\
         {} {}\n",
        cron_expression(ctx.heartbeat_secs),
        ctx.command()
    )
}

fn launchd(ctx: &RenderContext<'_>) -> String {
    let args: Vec<String> = ctx.command().split_whitespace().map(String::from).collect();
    let arg_xml: String = args
        .iter()
        .map(|a| format!("    <string>{}</string>\n", xml_escape(a)))
        .collect();
    format!(
        r#"<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
<!-- areev triggers heartbeat. The memory owns the real cadence. -->
<plist version="1.0">
<dict>
  <key>Label</key>
  <string>ai.areev.triggers</string>
  <key>ProgramArguments</key>
  <array>
{arg_xml}  </array>
  <key>StartInterval</key>
  <integer>{secs}</integer>
  <key>RunAtLoad</key>
  <true/>
</dict>
</plist>
"#,
        secs = ctx.heartbeat_secs
    )
}

fn systemd(ctx: &RenderContext<'_>) -> String {
    format!(
        r#"# areev triggers heartbeat. The memory owns the real cadence.
# Save as areev-triggers.service and areev-triggers.timer, then:
#   systemctl --user enable --now areev-triggers.timer

# ---- areev-triggers.service ----
[Unit]
Description=Areev trigger evaluation

[Service]
Type=oneshot
ExecStart={cmd}

# ---- areev-triggers.timer ----
[Unit]
Description=Areev trigger heartbeat

[Timer]
OnBootSec={secs}s
OnUnitActiveSec={secs}s
# Fire on resume if the machine was asleep through a window — the same
# catch-up posture the evaluator itself takes.
Persistent=true
# Spread wakeups deterministically rather than randomly, so behaviour stays
# reproducible.
FixedRandomDelay=true
RandomizedDelaySec={jitter}s

[Install]
WantedBy=timers.target
"#,
        cmd = ctx.command(),
        secs = ctx.heartbeat_secs,
        jitter = (ctx.heartbeat_secs / 10).max(1),
    )
}

fn k8s(ctx: &RenderContext<'_>) -> String {
    let args: Vec<String> = ctx.container_command().split_whitespace().map(String::from).collect();
    let arg_yaml: String =
        args.iter().map(|a| format!("            - {}\n", yaml_scalar(a))).collect();
    format!(
        r#"# areev triggers heartbeat. The memory owns the real cadence.
apiVersion: batch/v1
kind: CronJob
metadata:
  name: areev-triggers
spec:
  schedule: "{schedule}"
  # Forbid: a heartbeat that overlaps itself would have several evaluators
  # racing for the same claims. They would be safe — losers skip — but it is
  # wasted work, and the evaluator's own concurrency policy is the right place
  # to express overlap, per trigger.
  concurrencyPolicy: Forbid
  # A tick this late is not worth firing: the memory will decide what is due on
  # the next one anyway.
  startingDeadlineSeconds: {deadline}
  successfulJobsHistoryLimit: 1
  failedJobsHistoryLimit: 3
  jobTemplate:
    spec:
      template:
        spec:
          restartPolicy: OnFailure
          containers:
          - name: areev
            # `command` runs THIS image's binary, so it is the name on PATH in
            # the image — not a path from the machine that rendered this.
            image: areev:latest
            command:
{arg_yaml}            # --db above is the path as written when this was
            # rendered. Mount the memory and make that path resolve INSIDE the
            # container; a heartbeat pointed at a path that does not exist
            # fails every tick, which looks exactly like nothing being due.
"#,
        schedule = cron_expression(ctx.heartbeat_secs),
        deadline = (ctx.heartbeat_secs * 2).max(120),
    )
}

fn xml_escape(s: &str) -> String {
    s.replace('&', "&amp;").replace('<', "&lt;").replace('>', "&gt;")
}

/// Quote anything that is not an unambiguous YAML plain scalar.
fn yaml_scalar(s: &str) -> String {
    let plain = !s.is_empty()
        && s.chars().all(|c| c.is_ascii_alphanumeric() || matches!(c, '-' | '_' | '.' | '/'))
        && !s.starts_with('-');
    if plain {
        s.to_string()
    } else {
        format!("\"{}\"", s.replace('\\', "\\\\").replace('"', "\\\""))
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    const WF: &str = "a1b2c3d4e5f60718293a4b5c6d7e8f90a1b2c3d4e5f60718293a4b5c6d7e8f90";

    fn ctx(secs: u64) -> RenderContext<'static> {
        RenderContext {
            exe: "areev",
            db: "accounting.db",
            ns: "accounting",
            heartbeat_secs: secs,
            extra_args: "--connector-cmd ./gmail.sh",
        }
    }

    fn interval(secs: u32) -> Trigger {
        Trigger::new(TriggerKind::Interval, WF).interval_secs(secs)
    }

    #[test]
    fn the_heartbeat_is_the_gcd_not_the_shortest_interval() {
        // 120 and 300 share 60. Rendering 120 would be finer than needed;
        // rendering 300 would make the 120s trigger late.
        let h = heartbeat_secs(&[interval(120), interval(300)]);
        assert_eq!(h, 60);
    }

    #[test]
    fn the_heartbeat_never_goes_below_a_minute() {
        // Sub-minute is inexpressible in cron and mostly buys wakeups. The
        // memory decides what is due, so a coarse heartbeat is the point.
        assert_eq!(heartbeat_secs(&[interval(5), interval(10)]), 60);
    }

    #[test]
    fn a_disabled_trigger_does_not_pull_the_heartbeat_down() {
        let h = heartbeat_secs(&[
            interval(600),
            interval(120).enabled(false),
        ]);
        assert_eq!(h, 600, "a disabled trigger should not cost wakeups");
    }

    #[test]
    fn cron_triggers_do_not_contribute_a_divisor() {
        // A cron cadence is not a period, so it cannot be gcd'd. The floor
        // covers it, since cron has minute resolution anyway.
        let h = heartbeat_secs(&[Trigger::new(TriggerKind::Schedule, WF).cron("0 9 * * *")]);
        assert_eq!(h, 60);
    }

    #[test]
    fn no_triggers_still_renders_a_sane_heartbeat() {
        assert_eq!(heartbeat_secs(&[]), 60);
    }

    #[test]
    fn cron_expressions_stay_expressible() {
        assert_eq!(cron_expression(60), "* * * * *");
        assert_eq!(cron_expression(300), "*/5 * * * *");
        assert_eq!(cron_expression(3600), "0 */1 * * *");
        // 7 minutes does not divide an hour — it would drift at the wrap, so it
        // rounds DOWN to something that does rather than emitting a schedule
        // that is subtly wrong.
        let seven = cron_expression(420);
        assert!(seven.starts_with("*/"), "{seven}");
        let n: u64 = seven.trim_start_matches("*/").split(' ').next().unwrap().parse().unwrap();
        assert_eq!(60 % n, 0, "the rendered period must divide an hour: {seven}");
    }

    #[test]
    fn every_declared_target_renders() {
        for t in TARGETS {
            let out = render(t, &ctx(120)).unwrap_or_else(|e| panic!("{t}: {e}"));
            assert!(!out.trim().is_empty(), "{t} rendered nothing");
            // launchd splits argv into <string> elements, so the tokens are
            // asserted individually rather than as one substring.
            assert!(out.contains("trigger"), "{t} must invoke the evaluator");
            assert!(out.contains("run"), "{t} must invoke the evaluator");
            assert!(out.contains("accounting.db"), "{t} must name the memory");
        }
    }

    /// A container runs the image's binary, never the authoring host's.
    ///
    /// The regression (#69) shipped because the test context used
    /// `exe: "areev"` — the same string the fix produces — so a render that
    /// spliced in `current_exe()` looked identical to one that did not. This
    /// context uses a path that could only have come from the authoring
    /// machine, which is the whole defect.
    #[test]
    fn container_targets_never_emit_the_authoring_hosts_binary_path() {
        let local = "/Users/someone/.cargo/bin/areev";
        let c = RenderContext {
            exe: local,
            db: "accounting.db",
            ns: "accounting",
            heartbeat_secs: 300,
            extra_args: "",
        };

        let k8s = render("k8s-cronjob", &c).unwrap();
        assert!(
            !k8s.contains(local),
            "a local path in a container spec is wrong wherever it is applied:\n{k8s}"
        );
        assert!(
            k8s.contains("- areev\n"),
            "command[0] must be the name on PATH inside the image:\n{k8s}"
        );

        // …while the host targets genuinely run on the machine that rendered
        // them, so for them the absolute path is the correct answer.
        for t in ["cron", "launchd", "systemd"] {
            let out = render(t, &c).unwrap();
            assert!(out.contains(local), "{t} runs locally and must name the local binary");
        }
    }

    #[test]
    fn an_unknown_target_is_refused_with_its_code() {
        let e = render("eventbridge", &ctx(60)).unwrap_err();
        assert_eq!(e.code(), "TRG-E007");
    }

    #[test]
    fn launchd_output_is_well_formed_plist() {
        let out = render("launchd", &ctx(120)).unwrap();
        assert!(out.starts_with("<?xml"));
        assert!(out.contains("<key>StartInterval</key>"));
        assert!(out.contains("<integer>120</integer>"));
        assert_eq!(out.matches("<array>").count(), out.matches("</array>").count());
        assert_eq!(out.matches("<dict>").count(), out.matches("</dict>").count());
    }

    #[test]
    fn k8s_quotes_arguments_that_yaml_would_misread() {
        let out = render("k8s-cronjob", &ctx(120)).unwrap();
        // A leading dash starts a YAML sequence entry; unquoted `--db` inside a
        // list item is a parse hazard.
        assert!(out.contains("\"--db\""), "flags must be quoted:\n{out}");
        assert!(out.contains("concurrencyPolicy: Forbid"));
    }

    #[test]
    fn systemd_asks_for_catch_up_and_deterministic_jitter() {
        let out = render("systemd", &ctx(300)).unwrap();
        // Persistent=true is the systemd spelling of the catch-up posture the
        // evaluator already takes.
        assert!(out.contains("Persistent=true"));
        // Deterministic rather than random, so behaviour stays reproducible.
        assert!(out.contains("FixedRandomDelay=true"));
        assert!(out.contains("OnUnitActiveSec=300s"));
    }

    #[test]
    fn extra_args_reach_the_rendered_command() {
        for t in TARGETS {
            let out = render(t, &ctx(60)).unwrap();
            assert!(out.contains("gmail.sh"), "{t} dropped the connector command");
        }
    }
}