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

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

/// 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(
    pool: &PgPool,
    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();
    sqlx::query(
        "INSERT INTO pending_s3_deletions (s3_key, bucket, source) SELECT * FROM unnest($1::text[], $2::text[], $3::text[]) ON CONFLICT DO NOTHING",
    )
    .bind(&s3_keys)
    .bind(&buckets)
    .bind(vec![source; keys.len()])
    .execute(pool)
    .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)")
        .bind(ids)
        .execute(pool)
        .await?;
    Ok(())
}

/// Returns true if any live row in the given bucket's storage tables 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.
///
/// `projects.cover_image_url` stores a full URL (not a bare s3_key) so we
/// check via `LIKE %s3_key`. Project image keys live under the
/// `projects/{id}/image/` prefix and don't collide with item/media keys, so
/// a positive match here is genuine. Without this clause a project cover
/// could be deleted out from under a still-live row.
#[tracing::instrument(skip_all)]
pub async fn is_s3_key_live(pool: &PgPool, bucket: &str, s3_key: &str) -> Result<bool> {
    let live = if bucket == "synckit" {
        sqlx::query_scalar::<_, bool>(
            "SELECT EXISTS(SELECT 1 FROM sync_blobs WHERE s3_key = $1)",
        )
        .bind(s3_key)
        .fetch_one(pool)
        .await?
    } else {
        // Escape the SQL LIKE wildcards (`_` matches one char, `%` matches
        // any) so an s3_key containing literal underscores from a user-
        // supplied media folder name (e.g. `cool_stuff/file.png`) doesn't
        // false-positive against neighbouring rows.
        let escaped = s3_key.replace('\\', "\\\\").replace('%', "\\%").replace('_', "\\_");
        let url_suffix = format!("%{escaped}");
        sqlx::query_scalar::<_, bool>(
            r#"
            SELECT
                EXISTS(SELECT 1 FROM media_files WHERE s3_key = $1)
             OR EXISTS(SELECT 1 FROM versions WHERE s3_key = $1)
             OR EXISTS(SELECT 1 FROM ota_artifacts WHERE s3_key = $1)
             OR EXISTS(SELECT 1 FROM items
                       WHERE audio_s3_key = $1 OR cover_s3_key = $1 OR video_s3_key = $1)
             OR EXISTS(SELECT 1 FROM items WHERE cover_image_url LIKE $2 ESCAPE '\')
             OR EXISTS(SELECT 1 FROM projects WHERE cover_image_url LIKE $2 ESCAPE '\')
             OR EXISTS(SELECT 1 FROM content_insertions WHERE storage_key = $1)
            "#,
        )
        .bind(s3_key)
        .bind(&url_suffix)
        .fetch_one(pool)
        .await?
    };
    Ok(live)
}

/// 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)
}