//! Background scheduler, runs periodic jobs on a fixed interval.
//!
//! Every tick: publish scheduled items/posts, send onboarding emails, dispatch builds.
//! Every 5 ticks: sandbox cleanup. Every tick: webhook retry, refund escalation, transaction cleanup,
//! acknowledgement alerts.
//! Daily: subscription checks, bounce monitoring, session pruning, IP scrubbing, account deletion.
//! Weekly: storage drift correction, sales count integrity.

mod acknowledgements;
mod announcements;
mod cleanup;
/// Re-exported for integration tests: abandoned multipart sessions hold billed
/// parts, so the reap path is worth exercising directly. The rest of `cleanup`
/// stays private.
pub use cleanup::abort_orphan_multipart_sessions;
#[doc(hidden)]
pub use cleanup::{cleanup_orphaned_uploads_for_test, drain_pending_s3_deletions_for_test};
mod integrity;
mod mt_threads;
mod synckit_warnings;
mod webhooks;

use tokio::sync::watch;
use tokio::task::JoinHandle;

use axum::extract::FromRef;

use crate::constants;
use crate::db;
use crate::{AppState, Integrations};

// Re-export public API used by route handlers
pub use announcements::{send_blog_post_announcements, send_release_announcements};
pub use mt_threads::{spawn_mt_thread_for_blog_post, spawn_mt_thread_for_item};

/// Advisory lock ID for single-instance scheduler coordination.
/// Prevents duplicate job execution during rolling deploys.
// Arbitrary fixed key, any stable, unique i64 works for pg_advisory_lock. (The
// value is not a real ASCII encoding of anything; an earlier comment claimed
// "MNW_SCH" but the literal has a stray nibble and decodes to no such string.
// Left as-is because changing it during a rolling deploy would briefly let old
// and new instances hold different keys and both run the scheduler.)
const SCHEDULER_ADVISORY_LOCK_ID: i64 = 0x04D4_E575_F534_3484;

/// Weekly drift correction interval in scheduler ticks (10,080 = 7 days at 60s).
const DRIFT_CORRECTION_INTERVAL: u64 = 10_080;

/// Daily interval in scheduler ticks (1,440 = 24h at 60s).
const DAILY_INTERVAL: u64 = 1440;

/// Sandbox cleanup interval in scheduler ticks (5 = 5min at 60s).
const SANDBOX_CLEANUP_INTERVAL: u64 = 5;

/// Hourly interval in scheduler ticks (60 = 1h at 60s). Used by the SyncKit
/// usage-warning job.
const HOURLY_INTERVAL: u64 = 60;

/// Soft tick-duration ceiling. A tick longer than this logs WARN; longer
/// than `TICK_DURATION_ALERT_SECS` also opens a WAM ticket (rate-limited).
/// Tuned conservatively, at 60s interval, anything over 30s means we're
/// burning more than half the budget and the next tick will skip.
const TICK_DURATION_WARN_SECS: u64 = 30;
const TICK_DURATION_ALERT_SECS: u64 = 50;
/// Hard ceiling on a single tick. The tick task is awaited while THIS loop task
/// holds the cross-instance advisory lock, so a job that hangs (e.g. a query
/// blocked on a row lock) would otherwise freeze all background maintenance on
/// every instance forever, the panic/overrun alerts only fire on completion or
/// panic, never on a hang. Past this bound we abort the tick, alert, and release
/// the lock so the next tick (or another instance) recovers. Far above the ~50s
/// healthy-tick ceiling, so it only trips on a true wedge.
const TICK_WATCHDOG_SECS: u64 = 300;

/// An import job whose liveness heartbeat is older than this is treated as
/// crashed and failed by the hourly reaper. Heartbeats bump after every 50-item
/// chunk, so 30 minutes of silence means the owning process is gone, not slow.
const STUCK_IMPORT_SECS: i64 = 1800;

/// Determine which scheduled job groups should run for a given tick.
///
/// Returns `(sandbox_cleanup, hourly_jobs, daily_jobs, weekly_jobs)`.
fn jobs_for_tick(tick: u64) -> (bool, bool, bool, bool) {
    let sandbox = tick.is_multiple_of(SANDBOX_CLEANUP_INTERVAL);
    let hourly = tick.is_multiple_of(HOURLY_INTERVAL);
    let daily = tick == 1 || tick.is_multiple_of(DAILY_INTERVAL);
    let weekly = tick.is_multiple_of(DRIFT_CORRECTION_INTERVAL);
    (sandbox, hourly, daily, weekly)
}

/// Spawn the background scheduler loop. Drop `shutdown_tx` to stop it.
pub fn spawn_scheduler(state: AppState, mut shutdown_rx: watch::Receiver<()>) -> JoinHandle<()> {
    tokio::spawn(async move {
        tracing::info!(
            "Scheduler started (interval={}s)",
            constants::SCHEDULER_INTERVAL_SECS
        );

        let mut interval = tokio::time::interval(std::time::Duration::from_secs(
            constants::SCHEDULER_INTERVAL_SECS,
        ));
        // Skip (not burst) missed ticks: a long-running pass (storage recalc,
        // drift checks under the advisory lock) must not trigger a catch-up
        // burst of back-to-back ticks on the next wake.
        interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
        interval.tick().await; // consume immediate first tick

        let mut tick_count: u64 = 0;
        let mut last_overrun_alert: Option<std::time::Instant> = None;
        let mut last_panic_alert: Option<std::time::Instant> = None;

        // Dedicated 1-connection pool for the advisory lock, created once. Holding
        // the lock across a tick on this pool no longer borrows from the 25-conn
        // request pool (was an accepted residual; Run 9). Falls back to the request
        // pool if the side-pool can't be created so the scheduler still runs.
        let lock_pool = sqlx::postgres::PgPoolOptions::new()
            .max_connections(1)
            .connect(&state.config.database_url)
            .await
            .map_err(|e| tracing::warn!(error = ?e, "scheduler: dedicated lock pool unavailable, using request pool"))
            .ok();

        loop {
            tokio::select! {
                _ = interval.tick() => {}
                _ = shutdown_rx.changed() => {
                    tracing::info!("Scheduler shutting down");
                    return;
                }
            }

            tick_count += 1;
            let tick_started = std::time::Instant::now();

            // Pin advisory lock to a dedicated connection held for the entire tick.
            // pg_try_advisory_lock is session-scoped, holding the connection prevents
            // another instance from acquiring the lock until this tick completes. The
            // connection comes from the dedicated `lock_pool` (1 conn), or the request
            // pool if that pool couldn't be created.
            let lock_source = lock_pool.as_ref().unwrap_or(&state.db);
            let mut lock_conn = match lock_source.acquire().await {
                Ok(conn) => conn,
                Err(e) => {
                    tracing::warn!(error = ?e, "scheduler: failed to acquire connection for advisory lock, skipping tick");
                    continue;
                }
            };
            let locked: bool = match sqlx::query_scalar("SELECT pg_try_advisory_lock($1)")
                .bind(SCHEDULER_ADVISORY_LOCK_ID)
                .fetch_one(&mut *lock_conn)
                .await
            {
                Ok(v) => v,
                Err(e) => {
                    tracing::warn!(error = ?e, "scheduler: failed to acquire advisory lock, skipping tick");
                    continue;
                }
            };
            if !locked {
                tracing::debug!("scheduler: advisory lock held by another instance, skipping tick");
                continue;
            }

            // Run every periodic job in a supervised child task. A panic in any
            // single job then aborts only this tick (the next tick re-runs the
            // full set); without supervision the panic unwinds the scheduler
            // loop and silently halts ALL background maintenance, including the
            // overrun/panic alerting below (Run #2 Performance CRITICAL). The
            // advisory lock stays held on `lock_conn` in this loop task for the
            // whole tick, so no other instance runs concurrently while the
            // child task works.
            let tick_jobs = {
                let state = state.clone();
                tokio::spawn(async move {
                    // Publish scheduled items. The announcement fan-out (4 DB
                    // roundtrips per item) gets spawned off the lock-held tick so a
                    // burst of releases doesn't extend the advisory-lock hold time.
                    match db::items::publish_scheduled_items(&state.db).await {
                        Ok(items) => {
                            for item in &items {
                                tracing::info!(
                                    item_id = %item.id,
                                    title = %item.title,
                                    "scheduler published item"
                                );
                                let db_pool = state.db.clone();
                                let mailer = state.email.clone();
                                let cfg = state.config.clone();
                                let item_for_announce = item.clone();
                                state.bg.spawn("release-announcements", async move {
                                    announcements::send_release_announcements(
                                        &db_pool,
                                        &mailer,
                                        &cfg,
                                        &item_for_announce,
                                    )
                                    .await;
                                });
                                if item.mt_thread_id.is_none() {
                                    mt_threads::spawn_mt_thread_for_item_by_lookup(
                                        &state.db,
                                        &state.bg,
                                        &Integrations::from_ref(&state),
                                        &state.config,
                                        item,
                                    );
                                }
                            }
                        }
                        Err(e) => {
                            tracing::error!(error = ?e, "scheduler failed to publish items");
                        }
                    }

                    // Send onboarding drip emails
                    announcements::send_onboarding_emails(&state.db, &state.email, &state.config)
                        .await;

                    // Dispatch pending builds. Project the runner's slice from the
                    // scheduler's `AppState` at the call site (same seam as the stripe
                    // dispatcher below) so the runner declares only what it needs.
                    crate::build_runner::dispatch_pending_build(&axum::extract::FromRef::from_ref(
                        &state,
                    ))
                    .await;

                    // Publish scheduled blog posts, same off-lock pattern as items.
                    match db::blog_posts::publish_scheduled_blog_posts(&state.db).await {
                        Ok(posts) => {
                            for post in &posts {
                                tracing::info!(
                                    post_id = %post.id,
                                    title = %post.title,
                                    "scheduler published blog post"
                                );
                                let db_pool = state.db.clone();
                                let mailer = state.email.clone();
                                let cfg = state.config.clone();
                                let post_for_announce = post.clone();
                                state.bg.spawn("blog-post-announcements", async move {
                                    announcements::send_blog_post_announcements(
                                        &db_pool,
                                        &mailer,
                                        &cfg,
                                        &post_for_announce,
                                    )
                                    .await;
                                });
                                if post.mt_thread_id.is_none() {
                                    mt_threads::spawn_mt_thread_for_blog_post_by_lookup(
                                        &state.db,
                                        &state.bg,
                                        &Integrations::from_ref(&state),
                                        &state.config,
                                        post,
                                    );
                                }
                            }
                        }
                        Err(e) => {
                            tracing::error!(error = ?e, "scheduler failed to publish blog posts");
                        }
                    }

                    // Clean up expired idempotency keys (every tick is fine, cheap DELETE)
                    if let Err(e) = db::idempotency::cleanup_expired(&state.db).await {
                        tracing::error!(error = ?e, "failed to clean up expired idempotency keys");
                    }

                    let (run_sandbox, run_hourly, run_daily, run_weekly) =
                        jobs_for_tick(tick_count);

                    // Clean up expired sandbox accounts (every 5 ticks = 5 min at 60s interval)
                    if run_sandbox {
                        cleanup::cleanup_sandbox_accounts(&state).await;
                        cleanup::retry_pending_s3_deletions(&state).await;
                        let report = crate::scanning::spool::reap_orphans(std::path::Path::new(
                            constants::SCAN_SPOOL_DIR,
                        ));
                        if report.deleted > 0 || report.errors > 0 {
                            tracing::info!(
                                deleted = report.deleted,
                                errors = report.errors,
                                "scan spool reaper swept orphans"
                            );
                        }
                    }

                    // Payments + storage outbound I/O: webhook retry, refund/credit
                    // escalation, platform-credit settle/reverse, stale-transaction and
                    // orphaned-upload cleanup. This block issues up to ~310 serial
                    // Stripe/S3 round-trips (30-90s worst case). Run it on the background
                    // pool rather than inline so the scheduler advisory lock is released as
                    // soon as the DB-only jobs finish instead of being held for the full
                    // outbound-I/O duration (fuzz-2026-07-06 F2, the hold tripped the 50s
                    // WAM tick-overrun alert and could eat the 60s interval). Every
                    // operation here is already safe to run without the single-instance
                    // tick lock, per-event `pg_try_advisory_xact_lock` dedup, `FOR UPDATE
                    // SKIP LOCKED` claims, and deterministic Stripe idempotency keys, so a
                    // run overlapping the next tick's just skips locked rows. Internal
                    // ordering (settle before escalate) is preserved by the sequence here.
                    {
                        let job_state = state.clone();
                        state.bg.spawn("scheduler-payments-io", async move {
                            webhooks::retry_failed_webhooks(&job_state).await;
                            webhooks::escalate_stale_refunds(&job_state).await;
                            webhooks::settle_platform_credits(&job_state).await;
                            webhooks::escalate_stale_platform_credits(&job_state).await;
                            webhooks::reverse_refunded_platform_credits(&job_state).await;
                            cleanup::cleanup_stale_pending_transactions(&job_state).await;
                            cleanup::cleanup_orphaned_uploads(&job_state).await;
                        });
                    }

                    // Alerts that repeat until a person confirms they read them.
                    // Every tick, because the weekly cadence lives in the query's
                    // WHERE and waking daily instead would put up to a day between
                    // something going wrong and the first message about it. One
                    // bounded indexed scan that returns nothing almost always; the
                    // sends fan out onto the background pool inside the job.
                    acknowledgements::send_due_acknowledgements(&state).await;

                    // Hourly: scan SyncKit apps for 75/90/100% cap breaches and email
                    // the app owner. Cheap query, single JOIN on a small table.
                    if run_hourly {
                        synckit_warnings::check_and_send_warnings(&state).await;
                        cleanup::purge_old_scan_jobs(&state).await;
                        // Fail imports stranded in `processing` by a crashed process.
                        // Heartbeats bump per chunk, so a 30-min-stale beat is dead.
                        match db::imports::reap_stuck_import_jobs(&state.db, STUCK_IMPORT_SECS)
                            .await
                        {
                            Ok(n) if n > 0 => tracing::warn!(
                                reaped = n,
                                "failed stuck import jobs (stale heartbeat)"
                            ),
                            Ok(_) => {}
                            Err(e) => {
                                tracing::error!(error = ?e, "failed to reap stuck import jobs");
                            }
                        }
                    }

                    // Weekly storage drift correction + integrity checks
                    if run_weekly {
                        integrity::recalculate_all_storage_used(&state).await;
                        integrity::check_sales_count_drift(&state).await;
                        match db::synckit_billing::recalculate_synckit_app_storage(&state.db).await
                        {
                            Ok(n) => {
                                if n > 0 {
                                    tracing::info!(
                                        corrected = n,
                                        "synckit app storage drift corrected"
                                    );
                                }
                            }
                            Err(e) => {
                                tracing::error!(error = ?e, "synckit storage drift correction failed");
                            }
                        }
                    }

                    // Daily checks (every 1440 ticks at 60s interval, plus first tick after startup)
                    if run_daily {
                        integrity::check_stale_subscriptions(&state).await;
                        integrity::check_email_bounce_spike(&state).await;

                        // Enforce post-grace item hiding (canceled 30+ days ago). Daily
                        // is ample for a 30-day grace window, and it's self-draining
                        // (mark_grace_enforced_batch), so running it every 60s tick only
                        // re-issued an empty query, moved here off the hot tick path.
                        integrity::enforce_post_grace_hiding(&state).await;

                        // Prune session records inactive for 90+ days
                        let session_threshold = chrono::Utc::now() - chrono::Duration::days(90);
                        match db::sessions::prune_expired_sessions(&state.db, session_threshold)
                            .await
                        {
                            Ok(n) => {
                                if n > 0 {
                                    tracing::info!(pruned = n, "pruned expired session records");
                                }
                                let _ = db::scheduler_jobs::record_job_run(
                                    &state.db,
                                    "session_prune",
                                    n as i64,
                                )
                                .await;
                            }
                            Err(e) => {
                                tracing::error!(error = ?e, "failed to prune expired sessions");
                            }
                        }

                        // Drop dead password-reset tokens (consumed/expired > 7 days).
                        match db::auth::prune_password_reset_tokens(&state.db).await {
                            Ok(n) => {
                                if n > 0 {
                                    tracing::info!(pruned = n, "pruned old password-reset tokens");
                                }
                            }
                            Err(e) => {
                                tracing::error!(error = ?e, "failed to prune password-reset tokens");
                            }
                        }

                        // Scrub IP addresses older than 30 days (privacy policy commitment)
                        cleanup::scrub_stale_ip_addresses(&state).await;

                        // Delete abandoned custom-page drafts older than 30 days.
                        match db::custom_pages::delete_drafts_older_than(&state.db, 30).await {
                            Ok(n) => {
                                if n > 0 {
                                    tracing::info!(deleted = n, "pruned old custom-page drafts");
                                }
                            }
                            Err(e) => {
                                tracing::error!(error = ?e, "failed to prune custom-page drafts");
                            }
                        }

                        // Delete terminated accounts whose 30-day export window has expired
                        cleanup::delete_expired_terminated_accounts(&state).await;

                        // Delete self-deleted creator accounts whose 90-day content grace period has expired
                        cleanup::delete_expired_content_removal_accounts(&state).await;

                        // Permanently delete soft-deleted items older than 7 days
                        cleanup::purge_expired_deleted_items(&state).await;

                        // Clean up stale and unavailable cart items
                        cleanup::cleanup_cart_items(&state).await;

                        // Prune page view aggregates older than 2 years
                        match db::page_views::prune_old_views(&state.db, 730).await {
                            Ok(n) => {
                                if n > 0 {
                                    tracing::info!(pruned = n, "pruned old page view records");
                                }
                            }
                            Err(e) => tracing::error!(error = ?e, "failed to prune page views"),
                        }

                        // Prune processed-webhook dedup markers older than 30 days. Stripe
                        // won't redeliver events that old, so they no longer prevent a
                        // duplicate; the table is otherwise append-only and grows one row
                        // per webhook forever (Run #21 Performance SERIOUS).
                        match db::webhook_events::prune_processed_events(&state.db, 30).await {
                            Ok(n) => {
                                if n > 0 {
                                    tracing::info!(
                                        pruned = n,
                                        "pruned old processed-webhook markers"
                                    );
                                }
                            }
                            Err(e) => {
                                tracing::error!(error = ?e, "failed to prune processed-webhook markers");
                            }
                        }

                        // Health/sync/OAuth prune+compaction. These ran in the monitor
                        // loop under a SECOND advisory lock (a parallel maintenance
                        // scheduler, Perf-S3 mandatory surprise); consolidated here so
                        // all periodic maintenance lives under the one scheduler lock and
                        // there is a single place to add a daily job.
                        match db::monitor::prune_health_history(
                            &state.db,
                            constants::HEALTH_HISTORY_RETAIN_DAYS,
                        )
                        .await
                        {
                            Ok(n) => {
                                if n > 0 {
                                    tracing::info!(
                                        deleted = n,
                                        "pruned old health history records"
                                    );
                                }
                            }
                            Err(e) => tracing::warn!(error = ?e, "failed to prune health history"),
                        }
                        match db::synckit::prune_sync_log(
                            &state.db,
                            constants::SYNC_LOG_RETAIN_DAYS,
                        )
                        .await
                        {
                            Ok(n) => {
                                if n > 0 {
                                    tracing::info!(deleted = n, "pruned old sync log records");
                                }
                            }
                            Err(e) => tracing::warn!(error = ?e, "failed to prune sync log"),
                        }
                        match db::synckit::compact_all_sync_logs(
                            &state.db,
                            constants::SYNC_LOG_COMPACT_MIN_AGE_DAYS,
                        )
                        .await
                        {
                            Ok(n) => {
                                if n > 0 {
                                    tracing::info!(
                                        deleted = n,
                                        "compacted sync log (cursor-based)"
                                    );
                                }
                            }
                            Err(e) => tracing::warn!(error = ?e, "failed to compact sync log"),
                        }
                        match db::oauth::cleanup_expired_oauth_codes(&state.db).await {
                            Ok(n) => {
                                if n > 0 {
                                    tracing::info!(deleted = n, "cleaned up expired OAuth codes");
                                }
                            }
                            Err(e) => tracing::warn!(error = ?e, "failed to clean up OAuth codes"),
                        }
                        match db::oauth::cleanup_expired_refresh_tokens(&state.db).await {
                            Ok(n) => {
                                if n > 0 {
                                    tracing::info!(
                                        deleted = n,
                                        "cleaned up expired OAuth refresh tokens"
                                    );
                                }
                            }
                            Err(e) => {
                                tracing::warn!(error = ?e, "failed to clean up OAuth refresh tokens");
                            }
                        }
                    }
                })
            };

            // Supervise the job task with BOTH a panic guard and a hang
            // watchdog. A panic surfaces as a JoinError; a hang trips the
            // watchdog. Either way we log, (rate-limited) alert, and fall
            // through to release the advisory lock so the freeze can't outlive
            // this tick. The watchdog is the load-bearing half: without it a
            // wedged query holds the lock and halts maintenance on every
            // instance with no signal (Run 21 Performance).
            let tick_abort = tick_jobs.abort_handle();
            match tokio::time::timeout(
                std::time::Duration::from_secs(TICK_WATCHDOG_SECS),
                tick_jobs,
            )
            .await
            {
                Ok(Ok(())) => {}
                Ok(Err(join_err)) => {
                    tracing::error!(
                        tick = tick_count, error = ?join_err,
                        "scheduler job task panicked; tick aborted, loop continues"
                    );
                    if let Some(ref wam) = state.wam {
                        let cooldown_ok = last_panic_alert.is_none_or(|t| {
                            t.elapsed().as_secs() >= constants::ALERT_COOLDOWN_SECS
                        });
                        if cooldown_ok {
                            let body = format!(
                                "tick #{tick_count}: a scheduled job panicked ({join_err}). \
                                 The tick was aborted; the scheduler loop survived and will \
                                 re-run all jobs on the next tick."
                            );
                            wam.create_ticket(
                                "Scheduler job panicked",
                                Some(&body),
                                "high",
                                "scheduler-job-panic",
                                None,
                            )
                            .await;
                            last_panic_alert = Some(std::time::Instant::now());
                        }
                    }
                }
                Err(_elapsed) => {
                    // The tick exceeded the watchdog, abort it so it stops
                    // holding DB connections, then release the lock below.
                    tick_abort.abort();
                    tracing::error!(
                        tick = tick_count,
                        watchdog_secs = TICK_WATCHDOG_SECS,
                        "scheduler tick exceeded watchdog; aborted and releasing advisory lock so maintenance can recover"
                    );
                    if let Some(ref wam) = state.wam {
                        let cooldown_ok = last_panic_alert.is_none_or(|t| {
                            t.elapsed().as_secs() >= constants::ALERT_COOLDOWN_SECS
                        });
                        if cooldown_ok {
                            let body = format!(
                                "tick #{tick_count}: a scheduled job hung past {TICK_WATCHDOG_SECS}s \
                                 (likely a query blocked on a lock). The tick was aborted and the \
                                 advisory lock released so the next tick recovers."
                            );
                            wam.create_ticket(
                                "Scheduler tick hung (watchdog)",
                                Some(&body),
                                "high",
                                "scheduler-tick-hang",
                                None,
                            )
                            .await;
                            last_panic_alert = Some(std::time::Instant::now());
                        }
                    }
                }
            }

            // Explicitly release the advisory lock (defense-in-depth: also released
            // when lock_conn is dropped, but explicit unlock survives refactors that
            // might move lock_conn into a shorter-lived scope).
            let _ = sqlx::query("SELECT pg_advisory_unlock($1)")
                .bind(SCHEDULER_ADVISORY_LOCK_ID)
                .execute(&mut *lock_conn)
                .await;

            // Tick-duration accounting. WARN at TICK_DURATION_WARN_SECS so the
            // log surfaces it; raise a WAM ticket past TICK_DURATION_ALERT_SECS
            // with a 1-hour cooldown so a chronic overrun doesn't flood tickets.
            let tick_duration = tick_started.elapsed();
            let tick_secs = tick_duration.as_secs();
            if tick_secs >= TICK_DURATION_WARN_SECS {
                tracing::warn!(
                    tick = tick_count,
                    duration_secs = tick_secs,
                    "scheduler tick exceeded soft duration ceiling"
                );
            }
            if tick_secs >= TICK_DURATION_ALERT_SECS
                && let Some(ref wam) = state.wam
            {
                let cooldown_ok = last_overrun_alert
                    .is_none_or(|t| t.elapsed().as_secs() >= constants::ALERT_COOLDOWN_SECS);
                if cooldown_ok {
                    let title = format!("Scheduler tick overran: {tick_secs}s");
                    let body = format!(
                        "tick #{tick_count} took {tick_secs}s (interval is {}s).",
                        constants::SCHEDULER_INTERVAL_SECS
                    );
                    wam.create_ticket(&title, Some(&body), "high", "scheduler-tick-overrun", None)
                        .await;
                    last_overrun_alert = Some(std::time::Instant::now());
                }
            }
        }
    })
}

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

    // ── Tick cadence tests ──

    #[test]
    fn tick_1_runs_daily_not_weekly() {
        let (sandbox, _hourly, daily, weekly) = jobs_for_tick(1);
        assert!(!sandbox, "tick 1 is not a multiple of 5");
        assert!(daily, "tick 1 should trigger daily jobs (first-tick rule)");
        assert!(!weekly, "tick 1 should not trigger weekly jobs");
    }

    #[test]
    fn tick_5_runs_sandbox_cleanup() {
        let (sandbox, _hourly, daily, weekly) = jobs_for_tick(5);
        assert!(sandbox);
        assert!(!daily);
        assert!(!weekly);
    }

    #[test]
    fn tick_1440_runs_daily_and_sandbox() {
        let (sandbox, _hourly, daily, weekly) = jobs_for_tick(1440);
        assert!(sandbox, "1440 is divisible by 5");
        assert!(daily, "1440 is the daily interval");
        assert!(!weekly);
    }

    #[test]
    fn tick_10080_runs_all_three() {
        let (sandbox, _hourly, daily, weekly) = jobs_for_tick(10_080);
        assert!(sandbox, "10080 is divisible by 5");
        assert!(daily, "10080 is divisible by 1440");
        assert!(weekly, "10080 is the weekly interval");
    }

    #[test]
    fn normal_tick_runs_nothing_special() {
        let (sandbox, _hourly, daily, weekly) = jobs_for_tick(7);
        assert!(!sandbox);
        assert!(!daily);
        assert!(!weekly);
    }

    #[test]
    fn second_daily_tick() {
        let (_, _, daily, _) = jobs_for_tick(2880);
        assert!(daily, "2880 = 2 * 1440");
    }

    #[test]
    fn second_weekly_tick() {
        let (_, _, _, weekly) = jobs_for_tick(20_160);
        assert!(weekly, "20160 = 2 * 10080");
    }

    // ── Interval constant sanity checks ──

    #[test]
    fn drift_correction_interval_is_7_days() {
        assert_eq!(DRIFT_CORRECTION_INTERVAL, 7 * 24 * 60);
    }

    #[test]
    fn daily_interval_is_24_hours() {
        assert_eq!(DAILY_INTERVAL, 24 * 60);
    }

    #[test]
    fn sandbox_cleanup_interval_is_5_minutes() {
        assert_eq!(SANDBOX_CLEANUP_INTERVAL, 5);
    }

    // ── Adversarial tests (test-fuzz) ──

    #[test]
    fn tick_0_runs_nothing() {
        // Tick 0 should not run anything: 0 % N == 0 for all N,
        // but tick 0 never happens in practice (counter starts at 0, increments before use).
        // Test what would happen if it did.
        let (sandbox, _hourly, daily, weekly) = jobs_for_tick(0);
        // 0.is_multiple_of(N) is true for all N, so these all fire
        assert!(sandbox, "0 is a multiple of 5");
        assert!(daily, "0 is a multiple of 1440");
        assert!(weekly, "0 is a multiple of 10080");
    }

    #[test]
    fn large_tick_values() {
        // 52 weeks of ticks, exact multiple of weekly interval
        let fifty_two_weeks = 52 * DRIFT_CORRECTION_INTERVAL;
        let (sandbox, _hourly, daily, weekly) = jobs_for_tick(fifty_two_weeks);
        assert!(sandbox);
        assert!(daily, "{fifty_two_weeks} should be divisible by 1440");
        assert!(weekly, "{fifty_two_weeks} should be divisible by 10080");

        // Large non-aligned tick
        let (_, _, daily, weekly) = jobs_for_tick(999_999);
        assert!(!daily, "999999 is not divisible by 1440");
        assert!(!weekly, "999999 is not divisible by 10080");
    }

    #[test]
    fn daily_not_on_partial_day() {
        // Tick 720 = half a day, should not trigger daily
        let (_, _, daily, _) = jobs_for_tick(720);
        assert!(!daily, "720 ticks is only half a day");
    }

    #[test]
    fn weekly_not_on_partial_week() {
        // 5040 = 3.5 days, should not trigger weekly
        let (_, _, _, weekly) = jobs_for_tick(5040);
        assert!(!weekly, "5040 is only 3.5 days");
    }

    #[test]
    fn sandbox_every_5_ticks_consecutively() {
        for tick in 1..=25 {
            let (sandbox, _, _, _) = jobs_for_tick(tick);
            if tick % 5 == 0 {
                assert!(sandbox, "tick {tick} should run sandbox cleanup");
            } else {
                assert!(!sandbox, "tick {tick} should NOT run sandbox cleanup");
            }
        }
    }

    #[test]
    fn intervals_are_coprime_aware() {
        // Verify weekly is an exact multiple of daily
        assert_eq!(
            DRIFT_CORRECTION_INTERVAL % DAILY_INTERVAL,
            0,
            "weekly interval must be exact multiple of daily"
        );
        // Verify sandbox fits evenly into daily
        assert_eq!(
            DAILY_INTERVAL % SANDBOX_CLEANUP_INTERVAL,
            0,
            "sandbox interval must divide evenly into daily"
        );
    }
}