//! Alerting on security signals, so the logs are not the only place they land.
//!
//! A 5xx spike, a credential-stuffing run against login, a webhook arriving with
//! a bad Stripe signature: each is logged, and a log nobody reads is not a
//! control. This raises them where someone sees them.
//!
//! Deliberately not a monitoring product. No metrics stack, no time series, no
//! dashboard, no second thing to keep alive on the box. Counters live in this
//! process, in fixed windows, and cross a threshold at most once per window.
//! Delivery reuses what `POST /api/internal/alerts` already built: a row in
//! `admin_alerts` and mail to `ALERT_EMAIL`. That endpoint exists for external
//! agents (PoM, MT), so signals raised here call the same insert-and-mail path
//! in process rather than posting to ourselves over the loopback with a bearer
//! token. `ALERTS_INGEST_TOKEN` gates inbound requests from other machines and
//! is not a precondition for anything in this file.
//!
//! Thresholds are numbers, and every one of them is reversible: noisy, raise
//! it; quiet, lower it. Nothing here is structural.
//!
//! **Suppression is load-bearing.** Nothing dedups server-side --
//! `insert_alert` is a plain INSERT and `admin_alerts` has no unique constraint
//! on `dedup_key` (migration 169 says deduplication is the sending agent's job).
//! The once-per-window rule below is therefore the only thing standing between
//! one condition and a mailbox full of identical alerts, which is why
//! `crossing_twice_in_one_window_alerts_once` is a real test and not a
//! formality.
//!
//! <!-- wiki: mnw-server-overview -->

use std::collections::{HashMap, HashSet};
use std::sync::{Mutex, OnceLock};
use std::time::{Duration, Instant};

use crate::db::admin_alerts::{AlertKind, AlertSeverity, NewAlert};
use crate::email::EmailClient;

/// Counting window for everything except CSP. Long enough that a threshold
/// means a rate, short enough that the operator hears about it while it is
/// still happening.
const WINDOW: Duration = Duration::from_mins(5);

/// CSP violations are a page-level defect, not a rate: the same blocked URI
/// fires on every load until someone fixes the page. Once a day per distinct
/// URI says so without saying it four thousand times.
const CSP_WINDOW: Duration = Duration::from_hours(24);

/// Fraction of responses that must be 5xx before the error rate alerts.
const ERROR_RATE_THRESHOLD: f64 = 0.01;

/// Requests a window needs before its error rate is meaningful. Without this, a
/// quiet night with two requests and one error reads as a 50% error rate.
const ERROR_RATE_MIN_REQUESTS: u64 = 20;

/// Failed auth attempts from one address before it looks like stuffing rather
/// than a forgotten password.
const AUTH_FAILURES_PER_IP: u64 = 20;

/// Failed auth attempts across all addresses. Catches the distributed version,
/// which the per-IP threshold is blind to by construction.
const AUTH_FAILURES_SITE_WIDE: u64 = 100;

/// Rate-limit trips before the volume is worth hearing about. An individual
/// trip is the system working exactly as designed and must never alert.
const RATE_LIMIT_TRIPS: u64 = 50;

/// Everything needed to raise an alert, captured once at startup.
///
/// A global rather than state threaded through every call site: the CSP report
/// handler takes a request body and nothing else, and the response middleware
/// runs outside the state extractors. Uninitialised (every unit test, and any
/// build that never calls [`install`]) means counting still happens and nothing
/// is sent, which is the correct degraded behaviour for a best-effort channel.
struct Sink {
    db: sqlx::PgPool,
    email: EmailClient,
}

static SINK: OnceLock<Sink> = OnceLock::new();

/// Point the signal counters at a database and a mailer. Call once at startup.
/// A second call is ignored rather than treated as an error, so the per-test
/// `build_app` does not have to care.
pub fn install(db: sqlx::PgPool, email: EmailClient) {
    let _ = SINK.set(Sink { db, email });
}

/// A condition that has crossed its threshold and is worth one alert.
#[derive(Debug, Clone)]
struct Firing {
    severity: AlertSeverity,
    dedup_key: String,
    title: String,
    body: String,
}

/// Fixed-window counters. One mutex: every field is touched together on the
/// response path and the critical section is a handful of integer adds, so
/// splitting it would buy contention rather than remove it.
#[derive(Default)]
struct Counters {
    window_started: Option<Instant>,
    requests: u64,
    server_errors: u64,
    auth_failures: u64,
    auth_failures_by_ip: HashMap<String, u64>,
    rate_limit_trips: u64,
    /// Conditions already alerted on in this window. Cleared with the window.
    fired: HashSet<String>,

    csp_window_started: Option<Instant>,
    /// Blocked URIs already alerted on today. Bounded by [`CSP_URI_CAP`] so a
    /// page generating unique blocked URIs cannot grow this without limit.
    csp_fired: HashSet<String>,
}

/// Distinct blocked URIs tracked per CSP window. A violation carrying a
/// cache-busted or otherwise unique URI each time would otherwise make this set
/// a slow memory leak; past the cap, further distinct URIs are counted as
/// already-alerted rather than remembered.
const CSP_URI_CAP: usize = 256;

static COUNTERS: Mutex<Option<Counters>> = Mutex::new(None);

/// Run `f` against the counters, rolling the window first if it has expired.
///
/// A poisoned mutex is not propagated. Every caller is on a request path that
/// has real work to do, and failing a checkout because an alert counter panicked
/// in another thread would make this module the outage it exists to report.
fn with_counters<T>(now: Instant, f: impl FnOnce(&mut Counters) -> T) -> Option<T> {
    let mut guard = COUNTERS.lock().ok()?;
    let counters = guard.get_or_insert_with(Counters::default);

    match counters.window_started {
        Some(started) if now.duration_since(started) < WINDOW => {}
        _ => {
            counters.window_started = Some(now);
            counters.requests = 0;
            counters.server_errors = 0;
            counters.auth_failures = 0;
            counters.auth_failures_by_ip.clear();
            counters.rate_limit_trips = 0;
            counters.fired.clear();
        }
    }

    match counters.csp_window_started {
        Some(started) if now.duration_since(started) < CSP_WINDOW => {}
        _ => {
            counters.csp_window_started = Some(now);
            counters.csp_fired.clear();
        }
    }

    Some(f(counters))
}

impl Counters {
    /// Record that `key` has fired, returning false if it already had this
    /// window. The whole suppression rule, in one place.
    fn claim(&mut self, key: &str) -> bool {
        self.fired.insert(key.to_string())
    }
}

/// Observe one finished response.
///
/// Called from the outermost middleware, so it sees every request the server
/// answered, including the ones rejected by a layer before any handler ran.
/// `ip` is the Cloudflare-derived client address where one is available.
pub fn note_response(status: u16, ip: Option<&str>) {
    note_response_at(status, ip, Instant::now());
}

fn note_response_at(status: u16, ip: Option<&str>, now: Instant) {
    let firing = with_counters(now, |c| {
        c.requests += 1;

        if status >= 500 {
            c.server_errors += 1;
            let rate = c.server_errors as f64 / c.requests as f64;
            if c.requests >= ERROR_RATE_MIN_REQUESTS
                && rate > ERROR_RATE_THRESHOLD
                && c.claim("sec:5xx")
            {
                return Some(Firing {
                    severity: AlertSeverity::Critical,
                    dedup_key: "sec:5xx".to_string(),
                    title: "Server error rate is elevated".to_string(),
                    body: format!(
                        "{} of {} responses in the last {} minutes were 5xx ({:.1}%). \
                         Threshold is {:.0}% over at least {} requests.",
                        c.server_errors,
                        c.requests,
                        WINDOW.as_secs() / 60,
                        rate * 100.0,
                        ERROR_RATE_THRESHOLD * 100.0,
                        ERROR_RATE_MIN_REQUESTS,
                    ),
                });
            }
        }

        if status == 429 {
            c.rate_limit_trips += 1;
            if c.rate_limit_trips > RATE_LIMIT_TRIPS && c.claim("sec:ratelimit") {
                return Some(Firing {
                    severity: AlertSeverity::Warning,
                    dedup_key: "sec:ratelimit".to_string(),
                    title: "Rate limiting is tripping in volume".to_string(),
                    body: format!(
                        "{} requests were rate limited in the last {} minutes (threshold {}). \
                         Individual trips are the limiter working; this many suggests a scraper, \
                         a stuck client, or a limit set too tight.",
                        c.rate_limit_trips,
                        WINDOW.as_secs() / 60,
                        RATE_LIMIT_TRIPS,
                    ),
                });
            }
        }

        let _ = ip;
        None
    })
    .flatten();

    dispatch(firing);
}

/// Record one failed authentication attempt.
///
/// Called at the failure sites rather than inferred from status codes: a wrong
/// password re-renders the login form with a 200, so nothing about the response
/// says an attempt failed.
pub fn note_auth_failure(ip: Option<&str>) {
    note_auth_failure_at(ip, Instant::now());
}

fn note_auth_failure_at(ip: Option<&str>, now: Instant) {
    let firing = with_counters(now, |c| {
        c.auth_failures += 1;

        if let Some(ip) = ip {
            // One entry per address per window, and the window clears it. A
            // spray from many addresses is bounded by the same window rather
            // than by a cap, which is what the site-wide counter is for.
            let per_ip = c.auth_failures_by_ip.entry(ip.to_string()).or_insert(0);
            *per_ip += 1;
            let count = *per_ip;
            let key = format!("sec:authfail:{ip}");
            if count > AUTH_FAILURES_PER_IP && c.claim(&key) {
                return Some(Firing {
                    severity: AlertSeverity::Critical,
                    dedup_key: key,
                    title: format!("Repeated auth failures from {ip}"),
                    body: format!(
                        "{count} failed authentication attempts from {ip} in the last {} minutes \
                         (threshold {AUTH_FAILURES_PER_IP}). Looks like credential stuffing rather \
                         than a forgotten password.",
                        WINDOW.as_secs() / 60,
                    ),
                });
            }
        }

        if c.auth_failures > AUTH_FAILURES_SITE_WIDE && c.claim("sec:authfail") {
            return Some(Firing {
                severity: AlertSeverity::Critical,
                dedup_key: "sec:authfail".to_string(),
                title: "Auth failures are elevated site-wide".to_string(),
                body: format!(
                    "{} failed authentication attempts across all addresses in the last {} minutes \
                     (threshold {AUTH_FAILURES_SITE_WIDE}). A distributed attempt would look like \
                     this and would not trip the per-address threshold.",
                    c.auth_failures,
                    WINDOW.as_secs() / 60,
                ),
            });
        }

        None
    })
    .flatten();

    dispatch(firing);
}

/// Record a webhook that arrived with a signature that did not verify.
///
/// No threshold: there is no benign cause. Stripe is the caller and it signs
/// correctly, so one of these means either a misconfigured secret or someone
/// posting forged events at the billing path.
pub fn note_webhook_signature_failure(provider: &str) {
    note_webhook_signature_failure_at(provider, Instant::now());
}

fn note_webhook_signature_failure_at(provider: &str, now: Instant) {
    let provider = provider.to_string();
    let firing = with_counters(now, |c| {
        let key = format!("sec:webhook:{provider}");
        if !c.claim(&key) {
            return None;
        }
        Some(Firing {
            severity: AlertSeverity::Critical,
            dedup_key: key,
            title: format!("{provider} webhook signature failed to verify"),
            body: format!(
                "A request to the {provider} webhook path carried a signature that did not \
                 verify. There is no benign cause: either the signing secret is wrong (in which \
                 case real events are being dropped) or someone is posting forged events."
            ),
        })
    })
    .flatten();

    dispatch(firing);
}

/// Record a CSP violation report, alerting once per distinct blocked URI per
/// day.
pub fn note_csp_violation(blocked_uri: &str, directive: &str, document: &str) {
    note_csp_violation_at(blocked_uri, directive, document, Instant::now());
}

fn note_csp_violation_at(blocked_uri: &str, directive: &str, document: &str, now: Instant) {
    let firing = with_counters(now, |c| {
        if c.csp_fired.len() >= CSP_URI_CAP || !c.csp_fired.insert(blocked_uri.to_string()) {
            return None;
        }
        Some(Firing {
            severity: AlertSeverity::Warning,
            dedup_key: format!("sec:csp:{blocked_uri}"),
            title: format!("CSP violation: {directive}"),
            body: format!(
                "A page reported a Content-Security-Policy violation.\n\n\
                 directive: {directive}\nblocked: {blocked_uri}\ndocument: {document}\n\n\
                 Either something on the page broke, or someone is probing. Reported once per \
                 blocked URI per day."
            ),
        })
    })
    .flatten();

    dispatch(firing);
}

/// Persist and mail a firing, off the request path.
///
/// Best effort in the strict sense: a failure here logs and goes no further. It
/// never blocks the request, never returns an error to a caller, and never
/// panics. An alerting channel that can take the site down is worse than no
/// alerting channel.
fn dispatch(firing: Option<Firing>) {
    let Some(firing) = firing else { return };
    let Some(sink) = SINK.get() else {
        // No sink: unit tests and any build that skipped `install`. The
        // threshold logic still ran, which is what those tests assert.
        tracing::debug!(
            dedup_key = %firing.dedup_key,
            "security signal fired before the alert sink was installed"
        );
        return;
    };

    let db = sink.db.clone();
    let email = sink.email.clone();
    tokio::spawn(async move {
        tracing::warn!(
            target: "security_signal",
            dedup_key = %firing.dedup_key,
            severity = firing.severity.as_str(),
            "{}",
            firing.title
        );

        let id = match crate::db::admin_alerts::insert_alert(
            &db,
            &NewAlert {
                source: "mnw",
                kind: AlertKind::Security,
                severity: firing.severity,
                title: &firing.title,
                body: &firing.body,
                dedup_key: Some(&firing.dedup_key),
                details: None,
            },
        )
        .await
        {
            Ok(id) => id,
            Err(e) => {
                tracing::error!(error = ?e, "failed to persist security alert");
                return;
            }
        };

        crate::routes::api::internal::alerts::email_alert(
            &db,
            &email,
            id,
            "mnw",
            AlertKind::Security,
            firing.severity,
            &firing.title,
            &firing.body,
        )
        .await;
    });
}

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

    /// Every test drives the counters through the `_at` variants with an
    /// explicit clock, and they share one global. Serialise them rather than
    /// letting a stray count from a parallel test move a threshold.
    fn lock() -> std::sync::MutexGuard<'static, ()> {
        static SERIAL: Mutex<()> = Mutex::new(());
        SERIAL
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
    }

    /// Force a fresh window. The counters are global, so a test that assumes an
    /// empty window has to say so.
    fn reset() {
        if let Ok(mut guard) = COUNTERS.lock() {
            *guard = None;
        }
    }

    /// What `fired` holds, which is the observable form of "an alert was sent"
    /// without a database behind it.
    fn fired() -> HashSet<String> {
        COUNTERS
            .lock()
            .unwrap()
            .as_ref()
            .map(|c| c.fired.clone())
            .unwrap_or_default()
    }

    #[test]
    fn error_rate_needs_volume_before_it_alerts() {
        let _g = lock();
        reset();
        let t = Instant::now();

        // Two requests, one of them a 500. That is a 50% error rate and it must
        // not alert: below the minimum request count, the ratio is noise.
        note_response_at(500, None, t);
        note_response_at(200, None, t);
        assert!(!fired().contains("sec:5xx"), "alerted on two requests");
    }

    #[test]
    fn error_rate_alerts_once_past_the_threshold() {
        let _g = lock();
        reset();
        let t = Instant::now();

        for _ in 0..ERROR_RATE_MIN_REQUESTS {
            note_response_at(200, None, t);
        }
        assert!(!fired().contains("sec:5xx"), "clean traffic alerted");

        note_response_at(500, None, t);
        assert!(fired().contains("sec:5xx"), "1 in 21 is over 1%");
    }

    #[test]
    fn crossing_twice_in_one_window_alerts_once() {
        let _g = lock();
        reset();
        let t = Instant::now();

        for _ in 0..ERROR_RATE_MIN_REQUESTS {
            note_response_at(200, None, t);
        }
        note_response_at(500, None, t);
        assert!(fired().contains("sec:5xx"));

        // Nothing dedups server-side, so this is the only suppression there is.
        // Staying over the threshold must not produce a second alert.
        let before = fired().len();
        for _ in 0..50 {
            note_response_at(500, None, t);
        }
        assert_eq!(fired().len(), before, "a sustained condition realerted");
    }

    #[test]
    fn the_window_resets_and_can_alert_again() {
        let _g = lock();
        reset();
        let t = Instant::now();

        for _ in 0..ERROR_RATE_MIN_REQUESTS {
            note_response_at(200, None, t);
        }
        note_response_at(500, None, t);
        assert!(fired().contains("sec:5xx"));

        let later = t + WINDOW + Duration::from_secs(1);
        for _ in 0..ERROR_RATE_MIN_REQUESTS {
            note_response_at(200, None, later);
        }
        assert!(!fired().contains("sec:5xx"), "counters did not roll");

        note_response_at(500, None, later);
        assert!(fired().contains("sec:5xx"), "a new window cannot alert");
    }

    #[test]
    fn individual_rate_limit_trips_never_alert() {
        let _g = lock();
        reset();
        let t = Instant::now();

        for _ in 0..RATE_LIMIT_TRIPS {
            note_response_at(429, None, t);
        }
        assert!(
            !fired().contains("sec:ratelimit"),
            "the limiter working is not an incident"
        );

        note_response_at(429, None, t);
        assert!(fired().contains("sec:ratelimit"));
    }

    #[test]
    fn auth_failures_alert_per_address() {
        let _g = lock();
        reset();
        let t = Instant::now();

        for _ in 0..AUTH_FAILURES_PER_IP {
            note_auth_failure_at(Some("203.0.113.7"), t);
        }
        assert!(!fired().contains("sec:authfail:203.0.113.7"));

        note_auth_failure_at(Some("203.0.113.7"), t);
        assert!(fired().contains("sec:authfail:203.0.113.7"));

        // A different address is its own condition and its own alert.
        assert!(!fired().contains("sec:authfail:203.0.113.8"));
    }

    #[test]
    fn auth_failures_alert_site_wide_when_spread_thin() {
        let _g = lock();
        reset();
        let t = Instant::now();

        // One attempt per address, so no per-address threshold is ever crossed.
        // This is the distributed case the per-IP counter cannot see.
        for i in 0..=AUTH_FAILURES_SITE_WIDE {
            note_auth_failure_at(Some(&format!("198.51.100.{i}")), t);
        }
        assert!(
            fired().iter().all(|k| k != "sec:authfail:198.51.100.1"),
            "no single address should have tripped"
        );
        assert!(fired().contains("sec:authfail"));
    }

    #[test]
    fn one_webhook_signature_failure_is_enough() {
        let _g = lock();
        reset();
        let t = Instant::now();

        note_webhook_signature_failure_at("stripe", t);
        assert!(fired().contains("sec:webhook:stripe"));
    }

    #[test]
    fn csp_alerts_once_per_blocked_uri() {
        let _g = lock();
        reset();
        let t = Instant::now();

        note_csp_violation_at("https://evil.example/x.js", "script-src", "/", t);
        note_csp_violation_at("https://evil.example/x.js", "script-src", "/", t);
        note_csp_violation_at("https://other.example/y.js", "script-src", "/", t);

        let uris = COUNTERS.lock().unwrap().as_ref().unwrap().csp_fired.clone();
        assert_eq!(
            uris.len(),
            2,
            "one entry per distinct blocked URI: {uris:?}"
        );
    }

    #[test]
    fn csp_uri_tracking_is_bounded() {
        let _g = lock();
        reset();
        let t = Instant::now();

        // A page minting a unique blocked URI per load must not grow the set
        // for a whole day.
        for i in 0..(CSP_URI_CAP * 2) {
            note_csp_violation_at(
                &format!("https://evil.example/{i}.js"),
                "script-src",
                "/",
                t,
            );
        }
        let uris = COUNTERS.lock().unwrap().as_ref().unwrap().csp_fired.len();
        assert_eq!(uris, CSP_URI_CAP);
    }

    #[test]
    fn csp_survives_the_short_window_rolling() {
        let _g = lock();
        reset();
        let t = Instant::now();

        note_csp_violation_at("https://evil.example/x.js", "script-src", "/", t);

        // A CSP entry is kept for a day, not for five minutes. Rolling the
        // short window must not re-arm it.
        let later = t + WINDOW + Duration::from_secs(1);
        note_csp_violation_at("https://evil.example/x.js", "script-src", "/", later);

        let uris = COUNTERS.lock().unwrap().as_ref().unwrap().csp_fired.len();
        assert_eq!(uris, 1, "the daily CSP entry was cleared by the 5m window");
    }
}