//! Durable queue for S3 object deletions that must survive server crashes.

use crate::error::Result;
use sqlx::PgPool;
use uuid::Uuid;

/// A pending S3 deletion record.
#[derive(Debug, sqlx::FromRow)]
pub struct PendingS3Deletion {
    pub id: Uuid,
    pub s3_key: String,
    pub bucket: String,
    pub source: String,
    pub attempts: i32,
}

/// Enqueue S3 keys for deletion. Each key is (s3_key, bucket).
#[tracing::instrument(skip_all)]
pub async fn enqueue_deletions<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    keys: &[(String, String)],
    source: &str,
) -> Result<()> {
    if keys.is_empty() {
        return Ok(());
    }
    let s3_keys: Vec<&str> = keys.iter().map(|(k, _)| k.as_str()).collect();
    let buckets: Vec<&str> = keys.iter().map(|(_, b)| b.as_str()).collect();
    // Takes any `PgExecutor` so callers can enqueue inside a transaction,
    // letting a row delete, its storage refund, and the S3-delete enqueue commit
    // atomically (see `versions::delete_version`).
    sqlx::query!(
        "INSERT INTO pending_s3_deletions (s3_key, bucket, source) SELECT * FROM unnest($1::text[], $2::text[], $3::text[]) ON CONFLICT DO NOTHING",
        &s3_keys as &[&str],
        &buckets as &[&str],
        &vec![source; keys.len()] as &[&str],
    )
    .execute(executor)
    .await?;
    Ok(())
}

/// Remove completed deletions by ID.
#[tracing::instrument(skip_all)]
pub async fn remove_completed(pool: &PgPool, ids: &[Uuid]) -> Result<()> {
    if ids.is_empty() {
        return Ok(());
    }
    sqlx::query!("DELETE FROM pending_s3_deletions WHERE id = ANY($1)", ids)
        .execute(pool)
        .await?;
    Ok(())
}

/// Move permanently-failing deletions off the hot queue into the operator-visible
/// dead-letter table (see migration 129). Atomic per the single-statement CTE:
/// the rows are deleted from `pending_s3_deletions` and inserted into
/// `pending_s3_deletions_dead_letter` in one operation, preserving their
/// `created_at`/`attempts`/`last_attempted_at`. Returns the number moved.
///
/// Without this the worker only logged and DELETEd such rows, so a malformed
/// key / gone bucket / ACL gap left an S3 object orphaned with no durable
/// record, the exact leak the dead-letter table was added to prevent.
#[tracing::instrument(skip_all)]
pub async fn move_to_dead_letter(pool: &PgPool, ids: &[Uuid]) -> Result<u64> {
    if ids.is_empty() {
        return Ok(0);
    }
    let result = sqlx::query!(
        r#"
        WITH moved AS (
            DELETE FROM pending_s3_deletions
            WHERE id = ANY($1)
            RETURNING id, s3_key, bucket, source, created_at, attempts, last_attempted_at
        )
        INSERT INTO pending_s3_deletions_dead_letter
            (id, s3_key, bucket, source, created_at, attempts, last_attempted_at)
        SELECT id, s3_key, bucket, source, created_at, attempts, last_attempted_at
        FROM moved
        ON CONFLICT (id) DO NOTHING
        "#,
        ids,
    )
    .execute(pool)
    .await?;
    Ok(result.rows_affected())
}

/// One place a live S3 object key can still be referenced. [`is_s3_key_live`] is
/// generated from this registry, so a new s3_key-bearing table is added in
/// exactly one place AND declares its own bucket, it can't be checked against
/// the wrong bucket (the OTA-key-checked-in-the-main-branch bug this replaced)
/// or silently forgotten by one branch of a hand-written query. Every registered
/// column stores a bare s3_key, so the check is always `column = $1`, covers
/// gained `cover_s3_key` in migration 152, retiring the old URL-suffix match.
struct S3KeyRef {
    bucket: &'static str,
    table: &'static str,
    column: &'static str,
}

/// The registry. To add a table that stores an S3 key, add a row here.
const S3_KEY_REFS: &[S3KeyRef] = &[
    // ── main bucket: creator content ──
    S3KeyRef {
        bucket: "main",
        table: "media_files",
        column: "s3_key",
    },
    S3KeyRef {
        bucket: "main",
        table: "versions",
        column: "s3_key",
    },
    S3KeyRef {
        bucket: "main",
        table: "items",
        column: "audio_s3_key",
    },
    S3KeyRef {
        bucket: "main",
        table: "items",
        column: "video_s3_key",
    },
    // ── public bucket: CDN-served, unsigned image content ──
    // These four surfaces render straight from `cdn.makenot.work/{key}`, so after
    // promote their content object lives in the public bucket, NOT main (see
    // `ScanTargetKind::content_served_from_public_bucket`). Their staging object
    // is still in main, deleted under the `main` tag by the promote, only the
    // promoted content object is reaped against `public`. Covers store the bare
    // key directly (migration 152 added projects.cover_s3_key; items.cover_s3_key
    // predates it).
    S3KeyRef {
        bucket: "public",
        table: "items",
        column: "cover_s3_key",
    },
    S3KeyRef {
        bucket: "public",
        table: "projects",
        column: "cover_s3_key",
    },
    S3KeyRef {
        bucket: "public",
        table: "item_images",
        column: "s3_key",
    },
    S3KeyRef {
        bucket: "public",
        table: "project_images",
        column: "s3_key",
    },
    // content_insertions is served PRESIGNED from the private bucket, so it stays main.
    S3KeyRef {
        bucket: "main",
        table: "content_insertions",
        column: "storage_key",
    },
    // ── synckit bucket: SyncKit blobs + OTA artifacts (both deterministic keys) ──
    S3KeyRef {
        bucket: "synckit",
        table: "sync_blobs",
        column: "s3_key",
    },
    S3KeyRef {
        bucket: "synckit",
        table: "ota_artifacts",
        column: "s3_key",
    },
];

/// Returns true if any live row in `bucket` still references `s3_key`. Used by
/// the deletion worker to detect the delete-then-reupload race: a freshly
/// uploaded object reusing a queued key must not be torpedoed by the worker
/// draining the queue. The tables checked are [`S3_KEY_REFS`] filtered to
/// `bucket`, add a new key-bearing table there.
///
/// Note: this deliberately does NOT consult `pending_uploads`. An in-flight
/// upload whose S3 PUT finished but whose durable row hasn't committed is
/// invisible here. That window is safe in practice because the durable row is
/// written early for every key class (OTA writes `ota_artifacts` at presign
/// time; same-name item/version/media replaces are caught by their confirm
/// handlers' idempotency guards before any delete is queued), and a queued
/// delete only fires after a >=10-minute staleness check. Adding
/// `pending_uploads` would broaden the live-set to ephemeral rows the reaper is
/// meant to clean, so it is intentionally excluded.
#[tracing::instrument(skip_all)]
pub async fn is_s3_key_live(pool: &PgPool, bucket: &str, s3_key: &str) -> Result<bool> {
    let refs: Vec<&S3KeyRef> = S3_KEY_REFS.iter().filter(|r| r.bucket == bucket).collect();
    if refs.is_empty() {
        // Only "main"/"synckit" exist, so an unregistered bucket is unexpected.
        // Refuse to declare the key dead, don't let the worker delete an object
        // we have no way to verify.
        tracing::warn!(
            bucket,
            "is_s3_key_live: no registered tables for bucket; treating key as live"
        );
        return Ok(true);
    }

    // Table/column names are compile-time constants from S3_KEY_REFS (never user
    // input); the key value is always the bound parameter $1. Every class stores
    // a bare key, so liveness is a flat OR of equality checks.
    let clauses: Vec<String> = refs
        .iter()
        .map(|r| format!("EXISTS(SELECT 1 FROM {} WHERE {} = $1)", r.table, r.column))
        .collect();
    let sql = format!("SELECT {}", clauses.join(" OR "));

    Ok(sqlx::query_scalar::<_, bool>(&sql)
        .bind(s3_key)
        .fetch_one(pool)
        .await?)
}

/// Fetch stale pending deletions (older than min_age, up to limit).
/// Atomically increments attempt count.
#[tracing::instrument(skip_all)]
pub async fn get_stale_pending(
    pool: &PgPool,
    min_age: chrono::Duration,
    limit: i64,
) -> Result<Vec<PendingS3Deletion>> {
    let cutoff = chrono::Utc::now() - min_age;
    let rows = sqlx::query_as::<_, PendingS3Deletion>(
        r"
        UPDATE pending_s3_deletions
        SET attempts = attempts + 1, last_attempted_at = NOW()
        WHERE id IN (
            SELECT id FROM pending_s3_deletions
            WHERE created_at < $1
            ORDER BY created_at
            LIMIT $2
            FOR UPDATE SKIP LOCKED
        )
        RETURNING id, s3_key, bucket, source, attempts
        ",
    )
    .bind(cutoff)
    .bind(limit)
    .fetch_all(pool)
    .await?;
    Ok(rows)
}

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

    /// Pin the set of S3 key-bearing columns the deletion-worker liveness check
    /// consults. A new key-bearing column added anywhere must be registered in
    /// `S3_KEY_REFS` too, or a freshly re-uploaded object reusing a queued key
    /// could be torpedoed by the worker. This forces any change to be deliberate.
    #[test]
    fn s3_key_ref_registry_covers_every_known_class() {
        let actual: BTreeSet<(&str, &str, &str)> = S3_KEY_REFS
            .iter()
            .map(|r| (r.bucket, r.table, r.column))
            .collect();
        let expected: BTreeSet<(&str, &str, &str)> = [
            ("main", "media_files", "s3_key"),
            ("main", "versions", "s3_key"),
            ("main", "items", "audio_s3_key"),
            ("main", "items", "video_s3_key"),
            ("main", "content_insertions", "storage_key"),
            // CDN-served image content lives in the public bucket post-promote.
            ("public", "items", "cover_s3_key"),
            ("public", "projects", "cover_s3_key"),
            ("public", "item_images", "s3_key"),
            ("public", "project_images", "s3_key"),
            ("synckit", "sync_blobs", "s3_key"),
            ("synckit", "ota_artifacts", "s3_key"),
        ]
        .into_iter()
        .collect();
        assert_eq!(
            actual, expected,
            "S3_KEY_REFS changed: if you added or removed a key-bearing column, update this \
             test and confirm the deletion-worker liveness check still covers every key class"
        );
    }
}