af-workflow 0.4.0

//! Pure ingress planning helpers for durable workflow hosts.

use chrono::{DateTime, Utc};

use crate::{CatchUpPolicy, NodeRegistry, ScheduleCadence, SchedulePolicy, Spec};

/// Result of [`schedule_decision`]: whether an overdue schedule runs and where
/// its durable cursor moves.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ScheduleDecision {
    /// Occurrence to run now; `None` when the policy skips the missed window.
    pub tick_at: Option<DateTime<Utc>>,
    /// Next occurrence to persist as the schedule cursor.
    pub next_at: DateTime<Utc>,
    /// Catch-up ticks completed so far in the current `CatchUpAll` burst.
    pub catch_up_count: u32,
}

/// Next UTC occurrence of a cron expression after `after`.
pub fn next_cron_at(expression: &str, after: DateTime<Utc>) -> Result<DateTime<Utc>, String> {
    expression
        .parse::<cron::Schedule>()
        .map_err(|error| format!("invalid cron '{expression}': {error}"))?
        .after(&after)
        .next()
        .ok_or_else(|| format!("cron '{expression}' has no future occurrence"))
}

/// Next occurrence of a schedule after `after`; fixed-delay cadences measure from `last_completed_at`.
pub fn next_scheduled_at(
    policy: &SchedulePolicy,
    after: DateTime<Utc>,
    last_completed_at: Option<DateTime<Utc>>,
) -> Result<DateTime<Utc>, String> {
    policy.validate().map_err(|error| error.to_string())?;
    match &policy.cadence {
        ScheduleCadence::Cron { expression } => {
            let timezone = policy
                .timezone
                .parse::<chrono_tz::Tz>()
                .map_err(|_| format!("unknown IANA timezone '{}'", policy.timezone))?;
            expression
                .parse::<cron::Schedule>()
                .map_err(|error| format!("invalid cron '{expression}': {error}"))?
                .after(&after.with_timezone(&timezone))
                .next()
                .map(|next| next.with_timezone(&Utc))
                .ok_or_else(|| format!("cron '{expression}' has no future occurrence"))
        }
        ScheduleCadence::FixedRate { milliseconds } => after
            .checked_add_signed(chrono::Duration::milliseconds(
                i64::try_from(*milliseconds).map_err(|_| "fixed rate exceeds i64")?,
            ))
            .ok_or_else(|| "fixed rate overflows time".into()),
        ScheduleCadence::FixedDelay { milliseconds } => last_completed_at
            .unwrap_or(after)
            .checked_add_signed(chrono::Duration::milliseconds(
                i64::try_from(*milliseconds).map_err(|_| "fixed delay exceeds i64")?,
            ))
            .ok_or_else(|| "fixed delay overflows time".into()),
    }
}

fn first_after_now(
    policy: &SchedulePolicy,
    scheduled_at: DateTime<Utc>,
    now: DateTime<Utc>,
) -> Result<DateTime<Utc>, String> {
    match &policy.cadence {
        ScheduleCadence::FixedRate { milliseconds } => {
            let period = i64::try_from(*milliseconds).map_err(|_| "fixed rate exceeds i64")?;
            let overdue = now
                .signed_duration_since(scheduled_at)
                .num_milliseconds()
                .max(0);
            let periods = overdue / period + 1;
            scheduled_at
                .checked_add_signed(chrono::Duration::milliseconds(
                    period.saturating_mul(periods),
                ))
                .ok_or_else(|| "fixed rate overflows time".into())
        }
        ScheduleCadence::FixedDelay { .. } => next_scheduled_at(policy, now, Some(now)),
        ScheduleCadence::Cron { .. } => next_scheduled_at(policy, now, None),
    }
}

/// Decide whether an overdue schedule runs and where its durable cursor moves.
/// A run is considered missed only when at least one later occurrence is also
/// due, avoiding false skips from ordinary scheduler latency.
pub fn schedule_decision(
    policy: &SchedulePolicy,
    scheduled_at: DateTime<Utc>,
    now: DateTime<Utc>,
    catch_up_count: u32,
) -> Result<ScheduleDecision, String> {
    let next_due = next_scheduled_at(policy, scheduled_at, Some(now))?;
    let has_missed_window =
        !matches!(&policy.cadence, ScheduleCadence::FixedDelay { .. }) && next_due <= now;
    if !has_missed_window {
        return Ok(ScheduleDecision {
            tick_at: Some(scheduled_at),
            next_at: next_due,
            catch_up_count: 0,
        });
    }
    match policy.catch_up {
        CatchUpPolicy::Skip => Ok(ScheduleDecision {
            tick_at: None,
            next_at: first_after_now(policy, scheduled_at, now)?,
            catch_up_count: 0,
        }),
        CatchUpPolicy::CatchUpOnce => Ok(ScheduleDecision {
            tick_at: Some(scheduled_at),
            next_at: first_after_now(policy, scheduled_at, now)?,
            catch_up_count: 0,
        }),
        CatchUpPolicy::CatchUpAll { limit } => {
            let completed = catch_up_count.saturating_add(1);
            let keep_catching_up = completed < limit && next_due <= now;
            Ok(ScheduleDecision {
                tick_at: Some(scheduled_at),
                next_at: if keep_catching_up {
                    next_due
                } else {
                    first_after_now(policy, scheduled_at, now)?
                },
                catch_up_count: u32::from(keep_catching_up) * completed,
            })
        }
    }
}

/// Ingress node of one branch, for hosts that feed triggers.
#[derive(Debug, Clone, PartialEq)]
pub struct BranchIngressPlan {
    /// Workflow branch this record belongs to.
    pub branch_id: String,
    /// Ingress node type.
    pub ingress_type: String,
    /// Ingress node config.
    pub ingress_config: serde_json::Value,
}

/// Ingress plan per branch.
pub fn ingress_plans(spec: &Spec, registry: &NodeRegistry) -> Vec<BranchIngressPlan> {
    spec.branches
        .iter()
        .filter_map(|branch| {
            let ingress = branch
                .nodes
                .iter()
                .find(|node| registry.is_ingress(&node.node_type))?;
            Some(BranchIngressPlan {
                branch_id: branch.branch_id.clone(),
                ingress_type: ingress.node_type.clone(),
                ingress_config: ingress.config.clone(),
            })
        })
        .collect()
}

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

    #[test]
    fn timezone_and_fixed_delay_are_explicit() {
        let after = Utc.with_ymd_and_hms(2026, 1, 1, 0, 0, 0).unwrap();
        let cron = SchedulePolicy {
            cadence: ScheduleCadence::Cron {
                expression: "0 30 9 * * * *".into(),
            },
            timezone: "Asia/Shanghai".into(),
            catch_up: CatchUpPolicy::CatchUpOnce,
        };
        assert_eq!(
            next_scheduled_at(&cron, after, None).unwrap(),
            Utc.with_ymd_and_hms(2026, 1, 1, 1, 30, 0).unwrap()
        );

        let delay = SchedulePolicy {
            cadence: ScheduleCadence::FixedDelay {
                milliseconds: 1_000,
            },
            timezone: "UTC".into(),
            catch_up: CatchUpPolicy::CatchUpOnce,
        };
        assert_eq!(
            next_scheduled_at(&delay, after, Some(after + chrono::Duration::seconds(10))).unwrap(),
            after + chrono::Duration::seconds(11)
        );
        assert!(SchedulePolicy {
            timezone: "not/a-zone".into(),
            ..delay
        }
        .validate()
        .is_err());
    }

    #[test]
    fn catch_up_policies_advance_the_persisted_cursor_without_a_burst() {
        let scheduled = Utc.with_ymd_and_hms(2026, 1, 1, 0, 0, 0).unwrap();
        let now = scheduled + chrono::Duration::seconds(35);
        let policy = |catch_up| SchedulePolicy {
            cadence: ScheduleCadence::FixedRate {
                milliseconds: 10_000,
            },
            timezone: "UTC".into(),
            catch_up,
        };
        let skipped = schedule_decision(&policy(CatchUpPolicy::Skip), scheduled, now, 0).unwrap();
        assert_eq!(skipped.tick_at, None);
        assert_eq!(skipped.next_at, scheduled + chrono::Duration::seconds(40));

        let once =
            schedule_decision(&policy(CatchUpPolicy::CatchUpOnce), scheduled, now, 0).unwrap();
        assert_eq!(once.tick_at, Some(scheduled));
        assert_eq!(once.next_at, scheduled + chrono::Duration::seconds(40));

        let all = policy(CatchUpPolicy::CatchUpAll { limit: 2 });
        let first = schedule_decision(&all, scheduled, now, 0).unwrap();
        assert_eq!(first.next_at, scheduled + chrono::Duration::seconds(10));
        assert_eq!(first.catch_up_count, 1);
        let second = schedule_decision(&all, first.next_at, now, first.catch_up_count).unwrap();
        assert_eq!(second.next_at, scheduled + chrono::Duration::seconds(40));
        assert_eq!(second.catch_up_count, 0);
    }

    #[test]
    fn cron_uses_timezone_rules_across_dst() {
        let policy = SchedulePolicy {
            cadence: ScheduleCadence::Cron {
                expression: "0 30 2 * * * *".into(),
            },
            timezone: "America/New_York".into(),
            catch_up: CatchUpPolicy::CatchUpOnce,
        };
        let before_spring_forward = Utc.with_ymd_and_hms(2026, 3, 8, 6, 59, 0).unwrap();
        assert_eq!(
            next_scheduled_at(&policy, before_spring_forward, None).unwrap(),
            Utc.with_ymd_and_hms(2026, 3, 9, 6, 30, 0).unwrap(),
            "nonexistent local times are skipped by chrono-tz/cron"
        );
    }

    #[test]
    fn next_cron_at_reports_invalid_expressions() {
        let after = Utc.with_ymd_and_hms(2026, 1, 1, 0, 0, 0).unwrap();
        assert_eq!(
            next_cron_at("0 0 12 * * * *", after).unwrap(),
            Utc.with_ymd_and_hms(2026, 1, 1, 12, 0, 0).unwrap()
        );
        assert!(next_cron_at("not a cron", after)
            .unwrap_err()
            .contains("invalid cron"));
    }

    #[test]
    fn next_scheduled_at_rejects_unparseable_cron_and_fixed_rate_measures_from_after() {
        let after = Utc.with_ymd_and_hms(2026, 1, 1, 0, 0, 0).unwrap();
        let broken = SchedulePolicy {
            cadence: ScheduleCadence::Cron {
                expression: "every day".into(),
            },
            timezone: "UTC".into(),
            catch_up: CatchUpPolicy::Skip,
        };
        assert!(next_scheduled_at(&broken, after, None).is_err());
        let rate = SchedulePolicy {
            cadence: ScheduleCadence::FixedRate {
                milliseconds: 1_500,
            },
            timezone: "UTC".into(),
            catch_up: CatchUpPolicy::Skip,
        };
        assert_eq!(
            next_scheduled_at(&rate, after, Some(after + chrono::Duration::hours(1))).unwrap(),
            after + chrono::Duration::milliseconds(1_500)
        );
    }
}