//! Hourly SyncKit usage warning emails.
//!
//! Walks every `active`, non-internal sync app and computes the highest
//! warning band (75/90/100%) currently breached above the previously-stamped
//! `last_warning_pct`. For each breach, emails the app owner and stamps the
//! threshold so we don't re-fire on subsequent ticks.
//!
//! Thresholds reset at period rollover via `reset_period_usage` (called from
//! the `invoice.paid` webhook handler), so each billing cycle can fire warnings
//! again from scratch.
//!
//! Storage breaches do NOT reset at period rollover (the cap is absolute, not
//! per-period), but `last_warning_pct` is shared across both dimensions for
//! simplicity. A creator who silenced storage warnings at 75% and then crosses
//! egress 75% in the same cycle won't get re-notified for egress 75%, they'll
//! get the next band (90%/100%) instead. This is an acceptable v1 quirk; if
//! it bites we'll split `last_warning_pct` into two columns.

use std::sync::Arc;

use crate::AppState;
use crate::constants::SYNCKIT_WARNINGS_PER_TICK;
use crate::db::synckit_billing::{self, WarningCandidate};
use crate::email::EmailClient;

/// Run one pass: fetch a bounded slice of breach candidates and hand each
/// warning to the bounded background pool to send + stamp.
///
/// This runs inside the single-threaded
/// scheduler tick, so it must never `.await` a network send here: a slow mail
/// provider times N breaching apps would block every other periodic job. The
/// tick does only bounded DB work (a `LIMIT`ed fetch); the actual `email.send`
/// happens on `state.bg`, which is concurrency-capped and fire-and-forget.
/// Warned apps stamp `last_warning_pct` and drop out of the candidate set, so
/// any overflow beyond the per-tick limit drains on subsequent ticks.
#[tracing::instrument(skip_all)]
pub(super) async fn check_and_send_warnings(state: &AppState) {
    let candidates =
        match synckit_billing::get_apps_needing_warning(&state.db, SYNCKIT_WARNINGS_PER_TICK).await
        {
            Ok(c) => c,
            Err(e) => {
                tracing::error!(error = ?e, "synckit warnings: query failed");
                return;
            }
        };

    if candidates.is_empty() {
        return;
    }
    tracing::info!(count = candidates.len(), "synckit warnings: enqueuing");

    for c in candidates {
        let email = state.email.clone();
        let db = state.db.clone();
        let host_url = state.config.host_url.clone();
        // Enqueue the slow send (and its stamp) onto the bounded pool so the
        // tick returns immediately. Fire-and-forget: on queue overflow the
        // task is dropped and the band re-fires next tick.
        state.bg.spawn("synckit-usage-warning", async move {
            send_warning(email, db, host_url, c).await;
        });
    }
}

/// Send one usage-warning email, then stamp the band so it doesn't re-fire.
/// Runs on the background pool, never on the scheduler tick. A send failure
/// skips the stamp so the band re-fires next tick (preferred over losing the
/// notification); a stamp failure logs and also re-fires.
async fn send_warning(
    email: EmailClient,
    db: sqlx::PgPool,
    host_url: Arc<str>,
    c: WarningCandidate,
) {
    let url = format!("{host_url}/sync/apps/{}/billing", c.app_id);
    if let Err(e) = email
        .send_synckit_usage_warning(
            &c.creator_email,
            &c.app_name,
            c.dimension,
            c.key.as_deref(),
            c.threshold_pct,
            c.used,
            c.limit,
            &url,
        )
        .await
    {
        tracing::error!(
            error = ?e,
            app_id = %c.app_id,
            dimension = c.dimension,
            key = c.key.as_deref().unwrap_or(""),
            pct = c.threshold_pct,
            "synckit warnings: send failed",
        );
        return;
    }
    // Stamp the band on the right table: per-key warnings stamp the per-key
    // row, app-wide stamps the app row.
    let stamp = match c.key.as_deref() {
        Some(k) => synckit_billing::update_key_warning_pct(&db, c.app_id, k, c.threshold_pct).await,
        None => synckit_billing::update_warning_pct(&db, c.app_id, c.threshold_pct).await,
    };
    if let Err(e) = stamp {
        tracing::error!(
            error = ?e,
            app_id = %c.app_id,
            key = c.key.as_deref().unwrap_or(""),
            "synckit warnings: stamp failed (will re-fire next tick)",
        );
    }
}

#[cfg(test)]
mod tests {
    use crate::db::synckit_billing::highest_breached_threshold;

    /// Seal for the warning loop. The send must only ever happen on the bounded
    /// background pool, inside `send_warning`, never inline in the scheduler
    /// tick. This fails the build if `check_and_send_warnings` ever names the
    /// email send directly, so a serial send inside the tick cannot appear
    /// unnoticed.
    #[test]
    fn tick_never_sends_inline() {
        let path = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
            .join("src/scheduler/synckit_warnings.rs");
        let src = std::fs::read_to_string(&path).expect("read warnings source");

        let start = src
            .find("async fn check_and_send_warnings")
            .expect("tick fn present");
        let rest = &src[start..];
        // The tick body ends where the next top-level fn (send_warning) begins.
        let end = rest[1..]
            .find("\nasync fn ")
            .or_else(|| rest[1..].find("\nfn "))
            .map_or(rest.len(), |i| i + 1);
        let tick_body = &rest[..end];

        assert!(
            !tick_body.contains("send_synckit_usage_warning"),
            "CHRONIC seal violated: check_and_send_warnings must not perform an \
             email send inline, enqueue onto state.bg and send in send_warning."
        );
        assert!(
            tick_body.contains("state.bg.spawn"),
            "warning tick must dispatch sends onto the bounded background pool",
        );
    }

    #[test]
    fn no_breach_below_first_threshold() {
        // 50% used, no warning yet
        assert!(highest_breached_threshold(50, 100, 0).is_none());
    }

    #[test]
    fn first_breach_at_75() {
        assert_eq!(highest_breached_threshold(75, 100, 0), Some(75));
        assert_eq!(highest_breached_threshold(80, 100, 0), Some(75));
    }

    #[test]
    fn no_re_fire_at_same_band() {
        // 80% used, already warned at 75%, don't fire again at 75%.
        assert!(highest_breached_threshold(80, 100, 75).is_none());
    }

    #[test]
    fn next_band_fires_after_previous() {
        // 90% used, last fired at 75%, fire 90%.
        assert_eq!(highest_breached_threshold(90, 100, 75), Some(90));
    }

    #[test]
    fn jumps_to_highest_band_only() {
        // 100% used after only firing 75%, fire 100%, skip 90%.
        assert_eq!(highest_breached_threshold(100, 100, 75), Some(100));
    }

    #[test]
    fn overshoot_caps_at_100() {
        // 150% used, none fired, still goes straight to 100% (highest band).
        assert_eq!(highest_breached_threshold(150, 100, 0), Some(100));
    }

    #[test]
    fn nothing_above_100_fires() {
        // Already at 100%; further growth doesn't trigger anything new.
        assert!(highest_breached_threshold(200, 100, 100).is_none());
    }

    #[test]
    fn zero_limit_is_safe() {
        // Defensive: don't divide by zero.
        assert!(highest_breached_threshold(50, 0, 0).is_none());
    }

    #[test]
    fn exactly_75_percent() {
        // Floor of 75.0 is 75, should breach the 75 band.
        assert_eq!(highest_breached_threshold(75, 100, 0), Some(75));
    }

    #[test]
    fn just_under_75_does_not_fire() {
        // 74.99%, floor is 74, below the 75 threshold.
        assert!(highest_breached_threshold(74, 100, 0).is_none());
    }

    #[test]
    fn large_bytes_no_overflow() {
        // 800 GiB of 1 TiB, no warning yet.
        let used: i64 = 800 * 1024 * 1024 * 1024;
        let limit: i64 = 1024 * 1024 * 1024 * 1024;
        assert_eq!(highest_breached_threshold(used, limit, 0), Some(75));
    }
}