//! The soft monthly ceiling on how much mail one creator sends.
//!
//! # What it protects
//!
//! The shared Postmark IP pool. One creator's fan-out degrades delivery for
//! every other creator on it, and nothing bounded that: [`super::lists::resolve_audience`]
//! caps one audience at 10,000 and `users.last_broadcast_at` allows one
//! broadcast per 24 hours, so a creator with several projects could mail without
//! any bound on the count over a month. The count is the number reputation
//! follows.
//!
//! # Soft, and the softness is the design
//!
//! Max, 2026-08-27: protect the commons through soft maximums with
//! application-based exceptions. Four things follow from that, and none of them
//! is polish:
//!
//! - The window is the creator's own **billing period**, not a calendar month,
//!   so the cap is legible next to what they pay for rather than being a second
//!   calendar to track. A creator with no subscription falls back to the
//!   calendar month, since they still send.
//! - A send that would cross the cap is **refused with a message**, never
//!   throttled. The alternative shapes were a draining queue and a silent
//!   swallow; a silent throttle reads as the platform losing mail, which is
//!   worse than a refusal, and a partial fan-out leaves half a list mailed.
//! - The count, the cap and the reset date are **visible before** either is met,
//!   which is the half that makes the cap acceptable.
//! - The number is set where a real creator never meets it, and the
//!   [application path](create_request) carries the rest. A cap that stops
//!   legitimate sends is a worse failure than one that lets a marginal send
//!   through.
//!
//! # Reserved up front, per send
//!
//! [`reserve`] takes the whole audience before any mail leaves. Counting each
//! mail as it went would be 8,000 writes for one announcement, and it would
//! decide the question halfway through a fan-out, which is exactly the
//! half-mailed list the refusal exists to avoid.
//!
//! Deliberately out of scope, filed rather than dropped: feeding a list's
//! complaint rate back into the cap. It predicts reputation damage better than
//! volume does, and it is a second mechanism with its own failure modes.

use chrono::{DateTime, Datelike as _, TimeZone as _, Utc};
use sqlx::PgPool;

use crate::db::UserId;
use crate::error::Result;

/// The billing month a count belongs to.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Window {
    pub start: DateTime<Utc>,
    pub end: DateTime<Utc>,
}

impl Window {
    /// The calendar month containing `now`, in UTC.
    ///
    /// The fallback for a creator with no subscription period to align to. Also
    /// the fallback when the stored period does not contain `now`, which is a
    /// Stripe update that has not landed yet rather than a period that really
    /// ran out: rolling the stale window forward would count this month's mail
    /// against last month's row.
    #[must_use]
    pub fn calendar_month(now: DateTime<Utc>) -> Self {
        let start = Utc
            .with_ymd_and_hms(now.year(), now.month(), 1, 0, 0, 0)
            .single()
            .unwrap_or(now);
        let (next_year, next_month) = if now.month() == 12 {
            (now.year() + 1, 1)
        } else {
            (now.year(), now.month() + 1)
        };
        let end = Utc
            .with_ymd_and_hms(next_year, next_month, 1, 0, 0, 0)
            .single()
            .unwrap_or(now);
        Self { start, end }
    }

    fn contains(&self, at: DateTime<Utc>) -> bool {
        at >= self.start && at < self.end
    }
}

/// What one creator has sent this window, against what they may.
#[derive(Debug, Clone, Copy)]
pub struct Usage {
    pub sent: i64,
    pub cap: i64,
    pub window: Window,
}

impl Usage {
    /// How much of the allowance is left. Never negative: a cap lowered under a
    /// creator who has already sent past it reads as nothing left rather than as
    /// a debt.
    #[must_use]
    pub fn remaining(&self) -> i64 {
        (self.cap - self.sent).max(0)
    }

    /// Percent of the allowance used, clamped to 100 for the gauge.
    #[must_use]
    pub fn percent(&self) -> i32 {
        if self.cap <= 0 {
            return 100;
        }
        #[expect(
            clippy::cast_possible_truncation,
            clippy::cast_precision_loss,
            reason = "clamped to 0..=100 before the cast"
        )]
        let pct = ((self.sent as f64 / self.cap as f64) * 100.0).clamp(0.0, 100.0) as i32;
        pct
    }

    /// Whether the creator is inside the warning band and should be told before
    /// they meet the cap rather than when they hit it.
    #[must_use]
    pub fn in_warning_band(&self) -> bool {
        let warn_at = crate::tier_prices::TierPrices::global().mail_cap_warn_at;
        #[expect(clippy::cast_precision_loss, reason = "counts, not currency")]
        let threshold = self.cap as f64 * warn_at;
        #[expect(clippy::cast_precision_loss, reason = "counts, not currency")]
        let sent = self.sent as f64;
        sent >= threshold
    }
}

/// What [`reserve`] decided.
#[derive(Debug, Clone, Copy)]
pub enum Verdict {
    /// The send may go. `usage` already counts it.
    Admitted { usage: Usage },
    /// The send may not go, and nothing was reserved. `usage` is the state that
    /// refused it, so the caller can say how much room is left rather than only
    /// that there is none.
    Refused { usage: Usage, requested: i64 },
}

impl Verdict {
    #[must_use]
    pub fn is_admitted(&self) -> bool {
        matches!(self, Self::Admitted { .. })
    }

    #[must_use]
    pub fn usage(&self) -> Usage {
        match *self {
            Self::Admitted { usage } | Self::Refused { usage, .. } => usage,
        }
    }

    /// The sentence a creator is owed when a send is refused. Says what the cap
    /// is, when it resets and what to do about it, because a refusal that only
    /// says no is the failure this whole mechanism is trying not to be.
    #[must_use]
    pub fn refusal_message(&self) -> Option<String> {
        let Self::Refused { usage, requested } = self else {
            return None;
        };
        Some(format!(
            "This send would reach {requested} recipients, and you have {remaining} of your \
             {cap} monthly emails left. The allowance resets on {reset}. Reply to \
             info@makenot.work with what you need and why, and we will raise it.",
            remaining = usage.remaining(),
            cap = usage.cap,
            reset = usage.window.end.format("%B %-d, %Y"),
        ))
    }
}

/// The window this creator's count belongs to.
///
/// Their Stripe billing period when there is one covering now, the calendar
/// month otherwise. See [`Window::calendar_month`] for why a stale period does
/// not roll forward.
#[tracing::instrument(skip_all)]
pub async fn window_for(pool: &PgPool, user_id: UserId) -> Result<Window> {
    let now = Utc::now();
    let period = sqlx::query_as::<_, (Option<DateTime<Utc>>, Option<DateTime<Utc>>)>(
        "SELECT current_period_start, current_period_end \
         FROM creator_subscriptions WHERE user_id = $1",
    )
    .bind(user_id)
    .fetch_optional(pool)
    .await?;

    if let Some((Some(start), Some(end))) = period {
        let window = Window { start, end };
        if window.contains(now) {
            return Ok(window);
        }
    }
    Ok(Window::calendar_month(now))
}

/// This creator's monthly allowance: their per-account override if an operator
/// has granted one, otherwise their tier's default from `assumptions.toml`.
///
/// The override is nullable rather than defaulted to the tier number on purpose.
/// A defaulted column would freeze today's number onto every row and stop a
/// config change from reaching an account that never asked for anything.
#[tracing::instrument(skip_all)]
pub async fn effective_cap(pool: &PgPool, user_id: UserId) -> Result<i64> {
    let override_cap = sqlx::query_scalar::<_, Option<i32>>(
        "SELECT monthly_mail_cap_override FROM users WHERE id = $1",
    )
    .bind(user_id)
    .fetch_optional(pool)
    .await?
    .flatten();

    if let Some(cap) = override_cap {
        return Ok(i64::from(cap));
    }

    let tier = super::creator_tiers::get_active_creator_tier(pool, user_id).await?;
    Ok(crate::tier_prices::TierPrices::global().monthly_mail_cap_for(tier))
}

/// What this creator has sent this window, without reserving anything. The read
/// behind the dashboard gauge.
#[tracing::instrument(skip_all)]
pub async fn usage(pool: &PgPool, user_id: UserId) -> Result<Usage> {
    let window = window_for(pool, user_id).await?;
    let cap = effective_cap(pool, user_id).await?;
    let sent = sent_in_window(pool, user_id, window).await?;
    Ok(Usage { sent, cap, window })
}

async fn sent_in_window(pool: &PgPool, user_id: UserId, window: Window) -> Result<i64> {
    let sent = sqlx::query_scalar::<_, Option<i64>>(
        "SELECT sent_count FROM creator_mail_usage WHERE user_id = $1 AND period_start = $2",
    )
    .bind(user_id)
    .bind(window.start)
    .fetch_optional(pool)
    .await?
    .flatten()
    .unwrap_or(0);
    Ok(sent)
}

/// Claim `mails` against this creator's allowance, all or nothing.
///
/// Atomic: the guard rides on the `ON CONFLICT` update, so two sends racing
/// cannot both see room that only one of them has. The single-send-larger-than-
/// the-whole-cap case is checked before the statement, because the insert branch
/// of an upsert has no `WHERE` to fail and would otherwise admit it.
#[tracing::instrument(skip_all, fields(mails))]
pub async fn reserve(pool: &PgPool, user_id: UserId, mails: i64) -> Result<Verdict> {
    let window = window_for(pool, user_id).await?;
    let cap = effective_cap(pool, user_id).await?;

    // Nothing to reserve, and no reason to make a row for a send with no
    // recipients.
    if mails <= 0 {
        let sent = sent_in_window(pool, user_id, window).await?;
        return Ok(Verdict::Admitted {
            usage: Usage { sent, cap, window },
        });
    }

    if mails > cap {
        let sent = sent_in_window(pool, user_id, window).await?;
        return Ok(Verdict::Refused {
            usage: Usage { sent, cap, window },
            requested: mails,
        });
    }

    let reserved = sqlx::query_scalar::<_, i64>(
        r"
        INSERT INTO creator_mail_usage (user_id, period_start, period_end, sent_count)
        VALUES ($1, $2, $3, $4)
        ON CONFLICT (user_id, period_start) DO UPDATE
           SET sent_count = creator_mail_usage.sent_count + EXCLUDED.sent_count,
               period_end = EXCLUDED.period_end,
               updated_at = now()
         WHERE creator_mail_usage.sent_count + EXCLUDED.sent_count <= $5
        RETURNING sent_count
        ",
    )
    .bind(user_id)
    .bind(window.start)
    .bind(window.end)
    .bind(mails)
    .bind(cap)
    .fetch_optional(pool)
    .await?;

    match reserved {
        Some(sent) => Ok(Verdict::Admitted {
            usage: Usage { sent, cap, window },
        }),
        None => {
            let sent = sent_in_window(pool, user_id, window).await?;
            Ok(Verdict::Refused {
                usage: Usage { sent, cap, window },
                requested: mails,
            })
        }
    }
}

/// Hand back a reservation that never turned into mail.
///
/// A caller that reserved and then failed to send owes the allowance back;
/// keeping it would charge a creator for a send that did not happen. Saturates
/// at zero rather than going negative.
#[tracing::instrument(skip_all)]
pub async fn release(pool: &PgPool, user_id: UserId, mails: i64) -> Result<()> {
    if mails <= 0 {
        return Ok(());
    }
    let window = window_for(pool, user_id).await?;
    sqlx::query(
        "UPDATE creator_mail_usage \
            SET sent_count = GREATEST(sent_count - $3, 0), updated_at = now() \
          WHERE user_id = $1 AND period_start = $2",
    )
    .bind(user_id)
    .bind(window.start)
    .bind(mails)
    .execute(pool)
    .await?;
    Ok(())
}

// --- The application path ---

/// One creator's ask for a bigger allowance.
#[derive(Debug, Clone, sqlx::FromRow)]
pub struct MailCapRequest {
    pub id: uuid::Uuid,
    pub user_id: UserId,
    pub requested_cap: i32,
    pub reason: String,
    pub status: String,
    pub granted_cap: Option<i32>,
    pub created_at: DateTime<Utc>,
    pub decided_at: Option<DateTime<Utc>>,
}

/// File an increase request. Returns `false` when one is already open: a second
/// ask is an amendment rather than a queue, and two identical pending rows means
/// an operator grants one and leaves the other to be granted again later.
#[tracing::instrument(skip_all)]
pub async fn create_request(
    pool: &PgPool,
    user_id: UserId,
    requested_cap: i32,
    reason: &str,
) -> Result<bool> {
    let inserted = sqlx::query(
        "INSERT INTO mail_cap_requests (user_id, requested_cap, reason) \
         VALUES ($1, $2, $3) ON CONFLICT DO NOTHING",
    )
    .bind(user_id)
    .bind(requested_cap)
    .bind(reason)
    .execute(pool)
    .await?;
    Ok(inserted.rows_affected() > 0)
}

/// This creator's most recent request, for the dashboard to show them where
/// their ask got to.
#[tracing::instrument(skip_all)]
pub async fn latest_request(pool: &PgPool, user_id: UserId) -> Result<Option<MailCapRequest>> {
    let row = sqlx::query_as::<_, MailCapRequest>(
        "SELECT id, user_id, requested_cap, reason, status, granted_cap, created_at, decided_at \
         FROM mail_cap_requests WHERE user_id = $1 ORDER BY created_at DESC LIMIT 1",
    )
    .bind(user_id)
    .fetch_optional(pool)
    .await?;
    Ok(row)
}

/// The operator queue: open requests, oldest first.
#[tracing::instrument(skip_all)]
pub async fn pending_requests(pool: &PgPool) -> Result<Vec<MailCapRequest>> {
    let rows = sqlx::query_as::<_, MailCapRequest>(
        "SELECT id, user_id, requested_cap, reason, status, granted_cap, created_at, decided_at \
         FROM mail_cap_requests WHERE status = 'pending' ORDER BY created_at LIMIT 200",
    )
    .fetch_all(pool)
    .await?;
    Ok(rows)
}

/// Grant a request at `granted_cap`, writing the per-account override in the
/// same transaction as the decision. The two are one act: a granted request
/// whose override never landed is a creator still being refused.
#[tracing::instrument(skip_all)]
pub async fn grant_request(
    pool: &PgPool,
    request_id: uuid::Uuid,
    granted_cap: i32,
    decided_by: UserId,
) -> Result<Option<UserId>> {
    let mut tx = pool.begin().await?;

    let user_id = sqlx::query_scalar::<_, UserId>(
        "UPDATE mail_cap_requests \
            SET status = 'granted', granted_cap = $2, decided_at = now(), decided_by = $3 \
          WHERE id = $1 AND status = 'pending' \
      RETURNING user_id",
    )
    .bind(request_id)
    .bind(granted_cap)
    .bind(decided_by)
    .fetch_optional(&mut *tx)
    .await?;

    let Some(user_id) = user_id else {
        tx.rollback().await?;
        return Ok(None);
    };

    sqlx::query("UPDATE users SET monthly_mail_cap_override = $2 WHERE id = $1")
        .bind(user_id)
        .bind(granted_cap)
        .execute(&mut *tx)
        .await?;

    tx.commit().await?;
    Ok(Some(user_id))
}

/// Deny a request. Leaves the override alone, so the tier default keeps
/// applying.
#[tracing::instrument(skip_all)]
pub async fn deny_request(
    pool: &PgPool,
    request_id: uuid::Uuid,
    decided_by: UserId,
) -> Result<Option<UserId>> {
    let user_id = sqlx::query_scalar::<_, UserId>(
        "UPDATE mail_cap_requests \
            SET status = 'denied', decided_at = now(), decided_by = $2 \
          WHERE id = $1 AND status = 'pending' \
      RETURNING user_id",
    )
    .bind(request_id)
    .bind(decided_by)
    .fetch_optional(pool)
    .await?;
    Ok(user_id)
}

/// The tier a cap belongs to, for the dashboard gauge. Same three bands the
/// storage gauge uses, read from the same helper so the two cannot drift.
#[must_use]
pub fn gauge_tier(usage: &Usage) -> &'static str {
    crate::types::gauge_tier(usage.percent())
}

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

    fn at(y: i32, m: u32, d: u32) -> DateTime<Utc> {
        Utc.with_ymd_and_hms(y, m, d, 12, 0, 0).unwrap()
    }

    #[test]
    fn a_calendar_month_runs_first_to_first() {
        let w = Window::calendar_month(at(2026, 8, 27));
        assert_eq!(w.start, Utc.with_ymd_and_hms(2026, 8, 1, 0, 0, 0).unwrap());
        assert_eq!(w.end, Utc.with_ymd_and_hms(2026, 9, 1, 0, 0, 0).unwrap());
        assert!(w.contains(at(2026, 8, 27)));
        assert!(!w.contains(at(2026, 9, 1)));
    }

    #[test]
    fn december_rolls_into_the_next_year() {
        // The one arithmetic in this module that can be wrong silently: a
        // December window ending on month 13 would put every December send in
        // the wrong row.
        let w = Window::calendar_month(at(2026, 12, 15));
        assert_eq!(w.end, Utc.with_ymd_and_hms(2027, 1, 1, 0, 0, 0).unwrap());
    }

    fn usage(sent: i64, cap: i64) -> Usage {
        Usage {
            sent,
            cap,
            window: Window::calendar_month(at(2026, 8, 27)),
        }
    }

    #[test]
    fn a_lowered_cap_leaves_nothing_rather_than_a_debt() {
        // An operator can lower an override under a creator who has already
        // sent past it. Negative remaining would read as owing mail back.
        let over = usage(9_000, 5_000);
        assert_eq!(over.remaining(), 0);
        assert_eq!(over.percent(), 100);
    }

    #[test]
    fn the_gauge_reads_the_same_bands_as_storage() {
        crate::tier_prices::TierPrices::install_test_default();
        assert_eq!(gauge_tier(&usage(0, 1_000)), "");
        assert_eq!(gauge_tier(&usage(800, 1_000)), "warn");
        assert_eq!(gauge_tier(&usage(950, 1_000)), "danger");
    }

    #[test]
    fn the_warning_band_arrives_before_the_cap() {
        crate::tier_prices::TierPrices::install_test_default();
        let warn_at = crate::tier_prices::TierPrices::global().mail_cap_warn_at;
        assert!(
            (0.0..1.0).contains(&warn_at),
            "a warning band at or above the cap warns nobody: {warn_at}"
        );

        #[expect(clippy::cast_possible_truncation, reason = "test arithmetic")]
        let threshold = (1_000.0 * warn_at) as i64;
        assert!(!usage(threshold - 1, 1_000).in_warning_band());
        assert!(usage(threshold, 1_000).in_warning_band());
        assert!(usage(1_000, 1_000).in_warning_band());
    }

    #[test]
    fn a_refusal_says_what_to_do_about_it() {
        crate::tier_prices::TierPrices::install_test_default();
        let verdict = Verdict::Refused {
            usage: usage(24_000, 25_000),
            requested: 4_000,
        };
        let message = verdict.refusal_message().expect("a refusal has a message");
        // The three facts a refused creator needs, and the address that lifts it.
        assert!(message.contains("4000"), "{message}");
        assert!(message.contains("1000"), "{message}");
        assert!(message.contains("September 1, 2026"), "{message}");
        assert!(message.contains("info@makenot.work"), "{message}");

        assert!(
            Verdict::Admitted {
                usage: usage(1, 25_000)
            }
            .refusal_message()
            .is_none()
        );
    }

    #[test]
    fn every_tier_has_an_allowance_a_real_creator_clears() {
        // The stated bias is against false positives, and the worked example in
        // the decision is a 2,000-person list mailed weekly. If any tier's
        // default sits under that, the cap stops legitimate work by default.
        crate::tier_prices::TierPrices::install_test_default();
        let prices = crate::tier_prices::TierPrices::global();
        const WEEKLY_TO_TWO_THOUSAND: i64 = 8_000;

        for tier in [
            Some(CreatorTier::Basic),
            Some(CreatorTier::SmallFiles),
            Some(CreatorTier::BigFiles),
            Some(CreatorTier::Everything),
        ] {
            let cap = prices.monthly_mail_cap_for(tier);
            assert!(
                cap > WEEKLY_TO_TWO_THOUSAND,
                "{tier:?} allows {cap}, under the worked example of {WEEKLY_TO_TWO_THOUSAND}"
            );
        }

        // A creator with no subscription gets less, and still gets something.
        let unsubscribed = prices.monthly_mail_cap_for(None);
        assert!(unsubscribed > 0);
        assert!(unsubscribed <= prices.monthly_mail_cap_for(Some(CreatorTier::Basic)));
    }
}