hyperhive/swarmctl/src/schedule.rs

//! `swarmctl schedule` — the operator's scheduled prompts, through
//! swarm-controller's `/api/schedules` routes.
//!
//! Each verb waits for the controller's answer, which is the list or the
//! schedule as stored. Each agent the schedule names picks a change up on its
//! next read of its schedule key and fires in its own container.
//!
//! The shapes below mirror the controller's own `ScheduleView` and request
//! bodies, for the reason [`crate::agent`]'s module doc gives.

use std::path::Path;

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

use crate::agent::{parse_ident, send};

/// One schedule as the controller shows it.
#[derive(Debug, Deserialize)]
struct ScheduleView {
    id: String,
    name: String,
    agents: Vec<String>,
    body: String,
    anchor_unix: i64,
    interval_seconds: Option<u64>,
    paused: bool,
    next_fire_unix: Option<i64>,
}

/// Body of `POST /api/schedules`.
#[derive(Serialize)]
struct AddRequest<'a> {
    name: &'a str,
    agents: Vec<String>,
    body: &'a str,
    #[serde(skip_serializing_if = "Option::is_none")]
    anchor_unix: Option<i64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    interval_seconds: Option<u64>,
    paused: bool,
}

/// Body of `PATCH /api/schedules/{id}`. An absent field is left as it is.
#[derive(Default, Serialize)]
pub(crate) struct EditRequest {
    #[serde(skip_serializing_if = "Option::is_none")]
    name: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    body: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    anchor_unix: Option<i64>,
    #[serde(skip_serializing_if = "IntervalEdit::is_keep")]
    interval_seconds: IntervalEdit,
    #[serde(skip_serializing_if = "Option::is_none")]
    paused: Option<bool>,
    #[serde(skip_serializing_if = "Vec::is_empty")]
    add_agents: Vec<String>,
    #[serde(skip_serializing_if = "Vec::is_empty")]
    remove_agents: Vec<String>,
}

/// What an edit does to the interval. `Set(None)` is sent as `null`, which
/// makes a one-shot; `Keep` is not sent at all.
#[derive(Clone, Copy, Default)]
enum IntervalEdit {
    #[default]
    Keep,
    Set(Option<u64>),
}

impl IntervalEdit {
    fn is_keep(&self) -> bool {
        matches!(self, Self::Keep)
    }
}

impl Serialize for IntervalEdit {
    fn serialize<S: serde::Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
        match self {
            Self::Keep | Self::Set(None) => s.serialize_none(),
            Self::Set(Some(seconds)) => s.serialize_u64(*seconds),
        }
    }
}

/// What `schedule edit` was asked to change.
pub(crate) struct EditFlags<'a> {
    pub name: Option<String>,
    pub body: Option<String>,
    pub at: Option<&'a str>,
    pub every: Option<&'a str>,
    pub once: bool,
    pub add_agents: Vec<String>,
    pub remove_agents: Vec<String>,
}

impl EditRequest {
    /// The edit `schedule edit` asks for. `once` and `every` are exclusive
    /// in the CLI, so at most one of them is set here.
    pub(crate) fn new(flags: EditFlags<'_>) -> Result<Self> {
        let interval_seconds = match (flags.every, flags.once) {
            (Some(every), _) => IntervalEdit::Set(Some(parse_every(every)?)),
            (None, true) => IntervalEdit::Set(None),
            (None, false) => IntervalEdit::Keep,
        };
        let edit = Self {
            name: flags.name,
            body: flags.body,
            anchor_unix: flags.at.map(parse_at).transpose()?,
            interval_seconds,
            paused: None,
            add_agents: agents(&flags.add_agents)?,
            remove_agents: agents(&flags.remove_agents)?,
        };
        if edit.name.is_none()
            && edit.body.is_none()
            && edit.anchor_unix.is_none()
            && edit.interval_seconds.is_keep()
            && edit.add_agents.is_empty()
            && edit.remove_agents.is_empty()
        {
            bail!(
                "nothing to change: pass --name, --body, --at, --every, --once, --add-agent \
                 or --remove-agent"
            );
        }
        Ok(edit)
    }

    fn paused(paused: bool) -> Self {
        Self {
            paused: Some(paused),
            ..Self::default()
        }
    }
}

fn agents(names: &[String]) -> Result<Vec<String>> {
    names.iter().map(|n| parse_ident(n, "agent name")).collect()
}

fn runtime() -> Result<tokio::runtime::Runtime> {
    tokio::runtime::Builder::new_current_thread()
        .enable_io()
        .build()
        .context("starting a tokio runtime for the controller request")
}

const COLLECTION: &str = "/api/schedules";

/// The schedule's path. The id is checked like an agent name, which every id
/// the controller mints satisfies, so it cannot reshape the URI.
fn member(id: &str) -> Result<String> {
    let id = parse_ident(id, "schedule id")?;
    Ok(format!("{COLLECTION}/{id}"))
}

/// What `schedule add` was asked for.
pub(crate) struct AddFlags<'a> {
    pub name: &'a str,
    pub agents: &'a [String],
    pub body: &'a str,
    pub at: Option<&'a str>,
    pub every: Option<&'a str>,
    pub paused: bool,
}

/// Run `swarmctl schedule add`.
pub(crate) fn add(socket: &Path, flags: &AddFlags<'_>) -> Result<()> {
    let request = AddRequest {
        name: flags.name,
        agents: agents(flags.agents)?,
        body: flags.body,
        anchor_unix: flags.at.map(parse_at).transpose()?,
        interval_seconds: flags.every.map(parse_every).transpose()?,
        paused: flags.paused,
    };
    let added: ScheduleView = runtime()?.block_on(send(
        socket,
        hyper::Method::POST,
        COLLECTION,
        Some(&request),
        "schedule-add",
    ))?;
    println!("added schedule {}", added.id);
    println!("{}", render(&added));
    Ok(())
}

/// Run `swarmctl schedule list`, optionally only the schedules naming `agent`.
pub(crate) fn list(socket: &Path, agent: Option<&str>) -> Result<()> {
    let agent = agent.map(|a| parse_ident(a, "agent name")).transpose()?;
    let schedules: Vec<ScheduleView> = runtime()?.block_on(send(
        socket,
        hyper::Method::GET,
        COLLECTION,
        None::<&()>,
        "schedule-list",
    ))?;
    let shown: Vec<&ScheduleView> = schedules
        .iter()
        .filter(|s| agent.as_ref().is_none_or(|a| s.agents.contains(a)))
        .collect();
    if shown.is_empty() {
        println!("no schedules");
    }
    for s in shown {
        println!("{}", render(s));
    }
    Ok(())
}

/// Run `swarmctl schedule edit`.
pub(crate) fn edit(socket: &Path, id: &str, edit: &EditRequest) -> Result<()> {
    let edited: ScheduleView = runtime()?.block_on(send(
        socket,
        hyper::Method::PATCH,
        &member(id)?,
        Some(edit),
        "schedule-edit",
    ))?;
    println!("{}", render(&edited));
    Ok(())
}

/// Run `swarmctl schedule pause` or `resume`.
pub(crate) fn set_paused(socket: &Path, id: &str, paused: bool) -> Result<()> {
    edit(socket, id, &EditRequest::paused(paused))
}

/// Run `swarmctl schedule rm`.
pub(crate) fn remove(socket: &Path, id: &str) -> Result<()> {
    let uri = member(id)?;
    runtime()?.block_on(send::<(), ()>(
        socket,
        hyper::Method::DELETE,
        &uri,
        None,
        "schedule-rm",
    ))?;
    println!("removed schedule {id}");
    Ok(())
}

/// `--at`: unix seconds, or an RFC 3339 time with its offset.
fn parse_at(value: &str) -> Result<i64> {
    if let Ok(unix) = value.parse::<i64>() {
        return Ok(unix);
    }
    chrono::DateTime::parse_from_rfc3339(value)
        .map(|t| t.timestamp())
        .with_context(|| {
            format!(
                "--at {value:?} is neither unix seconds nor an RFC 3339 time such as \
                 2026-10-05T09:00:00+02:00"
            )
        })
}

/// `--every`: a positive count of seconds, minutes, hours or days, such as
/// `90s`, `15m`, `1h` or `7d`. A bare number is seconds.
fn parse_every(value: &str) -> Result<u64> {
    let (digits, unit) = match value.char_indices().last() {
        Some((i, c)) if c.is_ascii_alphabetic() => (&value[..i], c),
        _ => (value, 's'),
    };
    let scale = match unit {
        's' => 1,
        'm' => 60,
        'h' => 3600,
        'd' => 86_400,
        _ => bail!("--every {value:?}: the unit must be s, m, h or d"),
    };
    let count: u64 = digits
        .parse()
        .with_context(|| format!("--every {value:?} is not a count followed by s, m, h or d"))?;
    match count.checked_mul(scale) {
        Some(0) => bail!("--every {value:?} is zero; leave --every out for a one-shot"),
        Some(seconds) => Ok(seconds),
        None => bail!("--every {value:?} is too long"),
    }
}

fn utc(unix: i64) -> String {
    chrono::DateTime::from_timestamp(unix, 0).map_or_else(
        || unix.to_string(),
        |t| t.format("%Y-%m-%dT%H:%M:%SZ").to_string(),
    )
}

/// One row per schedule: id, name, cadence, next fire, agents, and the
/// body's first line.
fn render(s: &ScheduleView) -> String {
    let cadence = match s.interval_seconds {
        Some(i) => format!("every {}", every_text(i)),
        None => "once".to_owned(),
    };
    let next = match (s.paused, s.next_fire_unix) {
        (true, _) => "paused".to_owned(),
        (false, Some(t)) => format!("next {}", utc(t)),
        (false, None) => format!("fired at {}", utc(s.anchor_unix)),
    };
    let body = s.body.lines().next().unwrap_or_default();
    format!(
        "{}  {:?}  {cadence}  from {}  {next}  agents: {}  | {body}",
        s.id,
        s.name,
        utc(s.anchor_unix),
        s.agents.join(", ")
    )
}

fn every_text(seconds: u64) -> String {
    for (unit, scale) in [("d", 86_400), ("h", 3600), ("m", 60)] {
        if seconds.is_multiple_of(scale) {
            return format!("{}{unit}", seconds / scale);
        }
    }
    format!("{seconds}s")
}

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

    fn flags() -> EditFlags<'static> {
        EditFlags {
            name: None,
            body: None,
            at: None,
            every: None,
            once: false,
            add_agents: Vec::new(),
            remove_agents: Vec::new(),
        }
    }

    #[test]
    fn every_takes_a_count_and_a_unit() {
        assert_eq!(parse_every("90").unwrap(), 90);
        assert_eq!(parse_every("90s").unwrap(), 90);
        assert_eq!(parse_every("15m").unwrap(), 900);
        assert_eq!(parse_every("1h").unwrap(), 3600);
        assert_eq!(parse_every("7d").unwrap(), 604_800);
        assert!(parse_every("0h").is_err());
        assert!(parse_every("1w").is_err());
        assert!(parse_every("h").is_err());
        assert!(parse_every("").is_err());
    }

    #[test]
    fn at_takes_unix_seconds_or_rfc3339() {
        assert_eq!(parse_at("1700000000").unwrap(), 1_700_000_000);
        assert_eq!(parse_at("1970-01-01T01:00:00+01:00").unwrap(), 0);
        assert!(parse_at("tomorrow").is_err());
    }

    /// The edit body says exactly what the flags changed: a `--once` sends an
    /// explicit `null` interval, and an unset flag sends nothing.
    #[test]
    fn an_edit_sends_only_what_was_asked() {
        let body = |e: &EditRequest| serde_json::to_string(e).unwrap();
        let e = EditRequest::new(EditFlags {
            name: Some("hi".to_owned()),
            ..flags()
        })
        .unwrap();
        assert_eq!(body(&e), r#"{"name":"hi"}"#);
        let e = EditRequest::new(EditFlags {
            once: true,
            ..flags()
        })
        .unwrap();
        assert_eq!(body(&e), r#"{"interval_seconds":null}"#);
        let e = EditRequest::new(EditFlags {
            at: Some("5"),
            every: Some("1m"),
            add_agents: vec!["iris".to_owned()],
            remove_agents: vec!["atlas".to_owned()],
            ..flags()
        })
        .unwrap();
        assert_eq!(
            body(&e),
            r#"{"anchor_unix":5,"interval_seconds":60,"add_agents":["iris"],"remove_agents":["atlas"]}"#
        );
        assert_eq!(body(&EditRequest::paused(true)), r#"{"paused":true}"#);
        assert!(EditRequest::new(flags()).is_err());
        assert!(
            EditRequest::new(EditFlags {
                add_agents: vec!["Not An Ident".to_owned()],
                ..flags()
            })
            .is_err()
        );
    }

    /// The controller's list item, as the wire shape documents it, decodes
    /// and renders as one row naming every agent.
    #[test]
    fn a_listed_schedule_renders_as_one_row() {
        let s: ScheduleView = serde_json::from_str(
            r#"{"id":"abc","name":"standup","agents":["atlas","iris"],"body":"post it\nmore",
                "anchor_unix":0,"interval_seconds":86400,"paused":false,"next_fire_unix":86400}"#,
        )
        .unwrap();
        let row = render(&s);
        assert!(!row.contains('\n'), "{row}");
        assert!(row.contains("\"standup\""), "{row}");
        assert!(row.contains("every 1d"), "{row}");
        assert!(row.contains("next 1970-01-02T00:00:00Z"), "{row}");
        assert!(row.contains("agents: atlas, iris"), "{row}");
        assert!(row.ends_with("| post it"), "{row}");
    }
}