rullst-core 12.3.0

#![allow(clippy::expect_used, clippy::unwrap_used)]

use super::*;
use std::sync::atomic::{AtomicBool, Ordering};

fn errors() -> (ErrorReporter<SchedulerError>, ErrorBuffer<SchedulerError>) {
    error_buffer(ERROR_BUFFER_CAPACITY, "test")
}

fn task_for_test<F, Fut>(handler: F) -> ScheduledTask
where
    F: Fn() -> Fut + Send + Sync + 'static,
    Fut: Future<Output = ()> + Send + 'static,
{
    Scheduler::new()
        .task("* * * * *", handler)
        .expect("valid cron")
        .tasks
        .remove(0)
}

/// A scheduler with one task that fires every second, for lifecycle tests
/// that cannot wait for a whole cron minute.
pub(crate) fn every_second<F, Fut>(timeout: Duration, handler: F) -> Scheduler
where
    F: Fn() -> Fut + Send + Sync + 'static,
    Fut: Future<Output = ()> + Send + 'static,
{
    let handler: ScheduledHandler = Arc::new(Box::new(move || Box::pin(handler())));
    Scheduler {
        tasks: vec![ScheduledTask {
            label: "every second".to_string(),
            schedule: "* * * * * * *"
                .parse()
                .expect("valid every-second schedule"),
            handler,
        }],
        task_timeout: timeout,
        failure_policy: SchedulerFailurePolicy::Continue,
    }
}

fn handle_with(
    loops: Vec<(String, JoinHandle<()>)>,
    errors: ErrorBuffer<SchedulerError>,
) -> SchedulerHandle {
    let (shutdown, _) = watch::channel(false);
    SchedulerHandle {
        shutdown,
        loops,
        errors,
    }
}

#[tokio::test]
async fn completed_handler_and_closed_shutdown_channel_are_observable() {
    let task = task_for_test(|| async {});
    let (_shutdown_tx, mut shutdown) = watch::channel(false);
    assert!(matches!(
        execute_handler(&task, Duration::from_secs(1), &mut shutdown)
            .await
            .expect("completed handler"),
        ExecutionStatus::Completed
    ));

    let (shutdown_tx, mut shutdown_rx) = watch::channel(false);
    drop(shutdown_tx);
    tokio::time::timeout(
        Duration::from_millis(50),
        wait_for_shutdown(&mut shutdown_rx),
    )
    .await
    .expect("closed channel terminates wait");
}

#[tokio::test]
async fn handle_reports_queued_and_join_failures_but_ignores_cancellation() {
    let (errors_tx, errors_rx) = errors();
    errors_tx.report(SchedulerError::TaskTimedOut {
        label: "queued".to_string(),
        timeout_ms: 5,
    });
    drop(errors_tx);
    let queued = handle_with(vec![], errors_rx).shutdown().await;
    assert!(matches!(
        queued,
        Err(SchedulerError::TaskTimedOut { timeout_ms: 5, .. })
    ));

    let (_errors_tx, errors_rx) = errors();
    let panicking = tokio::spawn(async { panic!("isolated loop panic") });
    let joined = handle_with(vec![("panic-loop".to_string(), panicking)], errors_rx)
        .shutdown()
        .await;
    assert!(matches!(
        joined,
        Err(SchedulerError::LoopFailed { label, .. }) if label == "panic-loop"
    ));

    let (_errors_tx, errors_rx) = errors();
    let cancelled = tokio::spawn(std::future::pending::<()>());
    cancelled.abort();
    assert!(
        handle_with(vec![("cancelled".to_string(), cancelled)], errors_rx)
            .shutdown()
            .await
            .is_ok()
    );
}

#[tokio::test]
async fn next_error_and_drop_abort_have_deterministic_lifecycles() {
    let (errors_tx, errors_rx) = errors();
    errors_tx.report(SchedulerError::TaskPanicked {
        label: "task".to_string(),
    });
    drop(errors_tx);
    let mut handle = handle_with(vec![], errors_rx);
    assert!(matches!(
        handle.next_error().await,
        Some(SchedulerError::TaskPanicked { .. })
    ));
    assert!(handle.next_error().await.is_none());

    struct Dropped(Arc<AtomicBool>);
    impl Drop for Dropped {
        fn drop(&mut self) {
            self.0.store(true, Ordering::SeqCst);
        }
    }

    let started = Arc::new(AtomicBool::new(false));
    let dropped = Arc::new(AtomicBool::new(false));
    let started_in_task = Arc::clone(&started);
    let dropped_in_task = Arc::clone(&dropped);
    let task = tokio::spawn(async move {
        let _guard = Dropped(dropped_in_task);
        started_in_task.store(true, Ordering::SeqCst);
        std::future::pending::<()>().await;
    });
    while !started.load(Ordering::SeqCst) {
        tokio::task::yield_now().await;
    }
    let (_errors_tx, errors_rx) = errors();
    drop(handle_with(vec![("pending".to_string(), task)], errors_rx));
    tokio::time::timeout(Duration::from_millis(100), async {
        while !dropped.load(Ordering::SeqCst) {
            tokio::task::yield_now().await;
        }
    })
    .await
    .expect("dropping handle aborts its tasks");
}

#[tokio::test]
async fn undrained_failures_are_bounded_and_counted() {
    let (errors_tx, errors_rx) = errors();
    for _ in 0..ERROR_BUFFER_CAPACITY + 3 {
        errors_tx.report(SchedulerError::TaskPanicked {
            label: "task".to_string(),
        });
    }
    let mut handle = handle_with(vec![], errors_rx);
    assert_eq!(handle.dropped_errors(), 3);

    let mut buffered = 0;
    while handle.try_next_error().is_some() {
        buffered += 1;
    }
    assert_eq!(buffered, ERROR_BUFFER_CAPACITY);
}

#[test]
fn scheduler_defaults_and_duration_saturation_are_explicit() {
    let scheduler = Scheduler::default();
    assert!(scheduler.tasks.is_empty());
    assert_eq!(scheduler.task_timeout, Duration::from_secs(300));
    assert_eq!(scheduler.failure_policy, SchedulerFailurePolicy::Continue);
    assert_eq!(duration_millis_u64(Duration::MAX), u64::MAX);
}

/// Pins the documented `cron`-crate day-of-week semantics.
#[test]
fn day_of_week_follows_the_documented_cron_crate_numbering() {
    use chrono::{Datelike, TimeZone, Weekday};

    let next_weekday = |expression: &str| {
        let task = Scheduler::new()
            .task(expression, || async {})
            .unwrap()
            .tasks
            .remove(0);
        // 2026-09-30 is a Wednesday.
        let after = chrono::Utc.with_ymd_and_hms(2026, 9, 30, 12, 0, 0).unwrap();
        task.schedule.after(&after).next().unwrap().weekday()
    };

    assert_eq!(next_weekday("0 9 * * 1"), Weekday::Sun);
    assert_eq!(next_weekday("0 9 * * 2"), Weekday::Mon);
    assert_eq!(next_weekday("0 9 * * 7"), Weekday::Sat);
    assert_eq!(next_weekday("0 9 * * MON"), Weekday::Mon);
    assert!(matches!(
        Scheduler::new().task("0 9 * * 0", || async {}),
        Err(SchedulerError::InvalidCron(_, _))
    ));

    // Restricted day-of-month and day-of-week must both match.
    let task = Scheduler::new()
        .task("0 0 1 * MON", || async {})
        .unwrap()
        .tasks
        .remove(0);
    let after = chrono::Utc.with_ymd_and_hms(2026, 9, 30, 12, 0, 0).unwrap();
    let next = task.schedule.after(&after).next().unwrap();
    assert_eq!((next.day(), next.weekday()), (1, Weekday::Mon));
}