//! Background maintenance tasks.

use std::collections::HashSet;
use std::sync::Arc;
use std::time::Duration;

use sqlx::PgPool;

use crate::storage::S3Storage;

/// One batch of removed-image objects to delete per round. S3's batch-delete
/// caps at 1000; staying well under keeps each request small.
const PURGE_BATCH: i64 = 500;

/// Periodically delete the S3 objects backing removed images whose object has
/// not been purged yet.
///
/// Closes two gaps left by the inline best-effort delete in
/// `remove_image_handler`: images removed before purge-tracking existed (the
/// backlog) and removals whose inline delete failed transiently. The sweep is
/// convergent: every successfully deleted object's image is marked
/// `s3_purged_at` and never revisited, so steady-state work is zero.
///
/// Runs once at startup, then every `interval`. Cancel by aborting the task.
pub async fn continuously_purge_removed_images(db: PgPool, s3: Arc<S3Storage>, interval: Duration) {
    loop {
        match purge_removed_image_objects(&db, &s3).await {
            Ok(0) => {}
            Ok(n) => tracing::info!(purged = n, "reconcile: purged orphaned image objects"),
            Err(e) => tracing::error!(error = %e, "reconcile: image purge sweep failed"),
        }
        tokio::time::sleep(interval).await;
    }
}

/// Delete the S3 objects for all removed-but-unpurged images, in batches.
/// Returns the number of image objects purged. A batch that fails to make any
/// progress (every key errored) stops the sweep so it retries next interval
/// rather than spinning on the same failures.
async fn purge_removed_image_objects(db: &PgPool, s3: &S3Storage) -> Result<usize, String> {
    let mut purged = 0usize;
    loop {
        let pending = mt_db::queries::list_images_pending_s3_purge(db, PURGE_BATCH)
            .await
            .map_err(|e| format!("db error listing images to purge: {e}"))?;
        if pending.is_empty() {
            break;
        }

        let keys: Vec<String> = pending.iter().map(|p| p.s3_key.clone()).collect();
        let failures = s3.delete_objects(&keys).await?;
        let failed: HashSet<&str> = failures.iter().map(|(k, _)| k.as_str()).collect();
        for (key, msg) in &failures {
            tracing::warn!(s3_key = %key, error = %msg, "reconcile: failed to delete image object");
        }

        let purged_ids: Vec<uuid::Uuid> = pending
            .iter()
            .filter(|p| !failed.contains(p.s3_key.as_str()))
            .map(|p| p.id)
            .collect();

        if purged_ids.is_empty() {
            // No progress this round. Leave the rest for the next interval.
            break;
        }

        purged += purged_ids.len();
        mt_db::mutations::mark_images_s3_purged(db, &purged_ids)
            .await
            .map_err(|e| format!("db error marking images purged: {e}"))?;

        if pending.len() < PURGE_BATCH as usize {
            break;
        }
    }
    Ok(purged)
}

// Chat

/// How often expired chat messages are swept.
///
/// Sets how long an expired message lingers, which is the only thing the
/// interval controls: expiry is stamped on the row at insert, so nothing is
/// kept alive by a late sweep. Fifteen minutes keeps each delete small and
/// bounds the lag well under the granularity anyone perceives in a window
/// measured in days.
pub const CHAT_SWEEP_INTERVAL: Duration = Duration::from_mins(15);

/// Periodically enforce both halves of chat retention.
///
/// Age and count are separate statements because they answer different
/// questions: age expires a quiet room, and the count cap bounds a busy one
/// that would reach its age limit holding far more than its owner allowed.
/// Whichever bites first wins, which is what `livechat::Retention` documents.
///
/// Both run every round even when the first finds nothing. Skipping the trim
/// when the age sweep is empty would be wrong in exactly the case that matters:
/// a room busy enough to be over its cap is usually one whose messages are all
/// too new to have expired.
///
/// Runs once at startup, then every `interval`. Cancel by aborting the task.
pub async fn continuously_sweep_chat(db: PgPool, interval: Duration) {
    loop {
        sweep_chat_once(&db).await;
        tokio::time::sleep(interval).await;
    }
}

/// One round of both retention statements. Split out of the loop so it can be
/// tested without waiting an interval.
///
/// Returns `(expired, trimmed)`. Errors are logged rather than returned: a
/// sweep is convergent and the next round retries, so a transient failure is
/// not worth propagating into a supervisor restart.
pub async fn sweep_chat_once(db: &PgPool) -> (u64, u64) {
    let expired = match mt_db::mutations::sweep_expired_chat_messages(db).await {
        Ok(n) => {
            if n > 0 {
                tracing::info!(expired = n, "chat: swept expired messages");
            }
            n
        }
        Err(e) => {
            tracing::error!(error = %e, "chat: expiry sweep failed");
            0
        }
    };

    // Runs regardless of the result above: a failed expiry sweep is no reason
    // to let a room grow past its cap as well.
    let trimmed = match mt_db::mutations::trim_chat_rooms_to_cap(db).await {
        Ok(n) => {
            if n > 0 {
                tracing::info!(trimmed = n, "chat: trimmed rooms to their message cap");
            }
            n
        }
        Err(e) => {
            tracing::error!(error = %e, "chat: room cap trim failed");
            0
        }
    };

    (expired, trimmed)
}

/// Periodically drop chat send-rate buckets nobody has touched recently.
///
/// Not optional and not the same job as the retention sweep. Bucket state is
/// keyed by (user, room), which is unbounded and attacker-influenced: anyone
/// who can reach a room can mint a bucket, and both units run under a hard 512M
/// cgroup cap where an OOM restarts the whole site rather than degrading chat.
///
/// The crate picks the interval, and the choice is a correctness one rather
/// than a tuning one: evicting a bucket resets its budget, so sweeping faster
/// than a bucket can refill would turn eviction into a way to skip the queue.
/// Hence `Chat::sweep_interval` rather than a number chosen here.
pub async fn continuously_sweep_chat_rate_limits(chat: Arc<livechat::Chat>) {
    let interval = chat.sweep_interval();
    loop {
        tokio::time::sleep(interval).await;
        let dropped = chat.sweep(std::time::Instant::now());
        if dropped > 0 {
            tracing::debug!(dropped, "chat: evicted idle rate-limit buckets");
        }
    }
}