//! Bounded background-task queue for fire-and-forget work.
//!
//! Replaces per-request `tokio::spawn(...)` for low-priority work that
//! competes with request handlers for the DB pool, email sends, mailing-list
//! subscriptions, etc. Run #4 fixed the same shape for page views via
//! `db::page_views::PageViewTx`; Run #8 surfaced it again on the webhook
//! hot path (`routes/stripe/webhook/checkout_helpers.rs`), so this module
//! generalizes the pattern.
//!
//! Two bounds applied:
//! - **Queue capacity**: `try_send` returns Err once `CAPACITY` tasks are
//!   queued; the task is dropped and a warning is logged. Caller is never
//!   blocked and never panics.
//! - **Concurrent execution**: a semaphore caps the number of background
//!   tasks running at once, so a burst of email sends can't exhaust the
//!   25-slot DB pool out from under request handlers.
//!
//! Use `state.bg.spawn(name, fut)` instead of `tokio::spawn(fut)` for any
//! work that (a) doesn't need its result and (b) acquires DB pool conns or
//! makes outbound network calls.

use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;
use tokio::sync::{Semaphore, mpsc, watch};
use tokio::task::JoinHandle;

/// Max queued tasks before `spawn` starts dropping with a warning.
const CAPACITY: usize = 1024;

/// Max background tasks running concurrently. Set well below
/// `DB_POOL_MAX_CONNECTIONS` (25) so background work can't starve requests.
const PARALLELISM: usize = 8;

type BoxFuture = Pin<Box<dyn Future<Output = ()> + Send + 'static>>;

#[derive(Clone)]
pub struct BackgroundTx {
    tx: mpsc::Sender<(&'static str, BoxFuture)>,
}

impl BackgroundTx {
    /// Submit a fire-and-forget task. On queue overflow the task is dropped
    /// and a warning is logged with `name` as the task identifier. Never
    /// blocks; never panics.
    pub fn spawn<F>(&self, name: &'static str, fut: F)
    where
        F: Future<Output = ()> + Send + 'static,
    {
        if self.tx.try_send((name, Box::pin(fut))).is_err() {
            // A dropped task is silent data loss (a missed email / cache purge /
            // MT thread), so surface it as a metric, not just a log line, so a
            // saturated queue is visible on the dashboard (audit Run 17 Resilience).
            metrics::counter!("background_tasks_dropped_total", "task" => name).increment(1);
            tracing::warn!(task = name, "background queue full, task dropped");
        }
    }
}

/// Spawn the drainer + return the sender to install on `AppState`, plus the
/// drainer's `JoinHandle` so shutdown can await an orderly drain. The drainer
/// pulls tasks off the channel and runs each under a semaphore permit so
/// concurrent execution is bounded.
///
/// On shutdown (the `shutdown` watch fires, a value change or all senders
/// dropped) the drainer stops taking new work, runs every already-queued task,
/// then waits for in-flight tasks to finish before exiting. Without this the
/// `bg` pool was the one undrained primitive: in-flight emails / cache purges /
/// MT thread creation were abandoned at the 10s hard exit on every restart
/// (audit Run 17 Resilience). The wait is best-effort and bounded by the
/// caller's shutdown timeout.
pub fn spawn_pool(mut shutdown: watch::Receiver<()>) -> (BackgroundTx, JoinHandle<()>) {
    let (tx, mut rx) = mpsc::channel::<(&'static str, BoxFuture)>(CAPACITY);
    let sem = Arc::new(Semaphore::new(PARALLELISM));
    let handle = tokio::spawn(async move {
        loop {
            tokio::select! {
                maybe = rx.recv() => match maybe {
                    Some((_, task)) => run_task(&sem, task).await,
                    None => break, // all senders dropped → nothing left to drain
                },
                // `changed()` resolves with Err once every sender is dropped,
                // which is exactly how main.rs signals shutdown (drop(shutdown_tx)).
                _ = shutdown.changed() => break,
            }
        }

        // Drain everything already enqueued, then wait for in-flight tasks by
        // reacquiring all permits (each running task holds one; once all
        // PARALLELISM are free, the pool is idle).
        while let Ok((_, task)) = rx.try_recv() {
            run_task(&sem, task).await;
        }
        let _ = sem.acquire_many(PARALLELISM as u32).await;
        tracing::info!("background-task drainer exited");
    });
    (BackgroundTx { tx }, handle)
}

/// Spawn a background pool with no external shutdown wiring, for test harnesses
/// and embeddings that manage lifecycle purely via `AppState` drop. The shutdown
/// sender is retained internally (its `changed()` never fires), so the pool
/// drains only when every `BackgroundTx` clone is dropped and the channel closes.
pub fn spawn_pool_detached() -> BackgroundTx {
    let (tx, rx) = watch::channel(());
    // Deliberately keep the sender alive so the drainer's shutdown arm never
    // fires; the pool then behaves exactly as it did before shutdown draining
    // existed (runs until the queue closes). Only reachable off the request path.
    std::mem::forget(tx);
    spawn_pool(rx).0
}

/// Acquire a permit and spawn `task`, releasing the permit when it finishes.
async fn run_task(sem: &Arc<Semaphore>, task: BoxFuture) {
    let Ok(permit) = sem.clone().acquire_owned().await else {
        return; // semaphore closed → shutting down
    };
    tokio::spawn(async move {
        task.await;
        drop(permit);
    });
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::atomic::{AtomicUsize, Ordering};

    #[tokio::test]
    async fn drains_queued_tasks_on_shutdown() {
        let (shutdown_tx, shutdown_rx) = watch::channel(());
        let (bg, handle) = spawn_pool(shutdown_rx);

        let counter = Arc::new(AtomicUsize::new(0));
        for _ in 0..50 {
            let c = counter.clone();
            bg.spawn("test", async move {
                c.fetch_add(1, Ordering::SeqCst);
            });
        }

        // Signal shutdown (drop the only sender) and await the orderly drain.
        drop(shutdown_tx);
        handle.await.expect("drainer joins cleanly");

        assert_eq!(
            counter.load(Ordering::SeqCst),
            50,
            "every queued task must run before the drainer exits"
        );
    }

    #[tokio::test]
    async fn drainer_exits_when_all_senders_drop() {
        let (_shutdown_tx, shutdown_rx) = watch::channel(());
        let (bg, handle) = spawn_pool(shutdown_rx);
        // Dropping every sender closes the channel; the drainer must exit on its
        // own (recv() -> None) without needing the shutdown signal.
        drop(bg);
        handle.await.expect("drainer exits when the queue closes");
    }
}