//! Tracking for presigned uploads that have not yet been confirmed.

use sqlx::PgPool;

use crate::db::UserId;
use crate::error::Result;

/// Record that a presigned upload URL was issued.
pub(crate) async fn record_pending_upload(
    pool: &PgPool,
    user_id: UserId,
    s3_key: &str,
    bucket: &str,
) -> Result<()> {
    // Re-presigning the same key (idempotent retry, multi-part flow) must refresh
    // `created_at`, otherwise the stale-pending reaper can delete a freshly-pending
    // object that's actively being uploaded right now.
    //
    // Pin the refresh to the original owner: if a different user collides on the
    // same key, do NOT refresh, let the reaper age the original row out on its
    // own schedule. Otherwise a re-presign loop by another principal could keep
    // an orphan object alive indefinitely.
    sqlx::query!(
        "INSERT INTO pending_uploads (user_id, s3_key, bucket) VALUES ($1, $2, $3)
         ON CONFLICT (s3_key, bucket) DO UPDATE SET created_at = NOW()
         WHERE pending_uploads.user_id = EXCLUDED.user_id",
        user_id as UserId,
        s3_key,
        bucket,
    )
    .execute(pool)
    .await?;
    Ok(())
}

/// Record a pending multipart upload, persisting the size the caller declared at
/// `start`. `parts` reads it back (see [`declared_size`]) to bind the part
/// geometry to the tier-checked size instead of trusting a later request body.
pub(crate) async fn record_pending_multipart_upload(
    pool: &PgPool,
    user_id: UserId,
    s3_key: &str,
    bucket: &str,
    declared_size_bytes: i64,
) -> Result<()> {
    // Same owner-pinned created_at refresh as record_pending_upload; also refresh
    // the declared size so a re-issued `start` for the same key stays consistent.
    sqlx::query!(
        "INSERT INTO pending_uploads (user_id, s3_key, bucket, declared_size_bytes)
         VALUES ($1, $2, $3, $4)
         ON CONFLICT (s3_key, bucket) DO UPDATE
             SET created_at = NOW(), declared_size_bytes = EXCLUDED.declared_size_bytes
             WHERE pending_uploads.user_id = EXCLUDED.user_id",
        user_id as UserId,
        s3_key,
        bucket,
        declared_size_bytes,
    )
    .execute(pool)
    .await?;
    Ok(())
}

/// The size the caller declared when it opened this pending upload, or `None`
/// if the row does not exist or carries no declared size (a single-PUT presign).
/// Scoped to `(user_id, s3_key, bucket)`, the same ownership proof the rest of
/// this module uses.
pub(crate) async fn declared_size(
    pool: &PgPool,
    user_id: UserId,
    s3_key: &str,
    bucket: &str,
) -> Result<Option<i64>> {
    let row = sqlx::query!(
        "SELECT declared_size_bytes FROM pending_uploads
         WHERE s3_key = $1 AND user_id = $2 AND bucket = $3",
        s3_key,
        user_id as UserId,
        bucket,
    )
    .fetch_optional(pool)
    .await?;
    Ok(row.and_then(|r| r.declared_size_bytes))
}

/// Refresh a pending upload's `created_at` so an actively-progressing transfer
/// (e.g. a multipart session still requesting part URLs) is not aged out by the
/// stale-pending reaper mid-flight.
pub(crate) async fn touch_pending_upload(
    pool: &PgPool,
    user_id: UserId,
    s3_key: &str,
    bucket: &str,
) -> Result<()> {
    sqlx::query!(
        "UPDATE pending_uploads SET created_at = NOW()
         WHERE s3_key = $1 AND user_id = $2 AND bucket = $3",
        s3_key,
        user_id as UserId,
        bucket,
    )
    .execute(pool)
    .await?;
    Ok(())
}

/// Remove the pending upload record after a successful confirm. Scoped to
/// `user_id` so a future caller that accepts a partially user-supplied key
/// can't delete another user's pending row, today's per-handler prefix
/// validation makes cross-user collision unreachable, but the function
/// signature shouldn't be broader than the invariant it protects.
///
/// Also scoped to `bucket`: the unique key is `(s3_key, bucket)` (migration
/// 137), so a key present in both the main and synckit buckets has two distinct
/// rows. Matching on `s3_key` alone would let one bucket's confirm clear the
/// other bucket's pending record (Run #18 Storage B7).
pub(crate) async fn remove_pending_upload<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    user_id: UserId,
    s3_key: &str,
    bucket: &str,
) -> Result<()> {
    sqlx::query!(
        "DELETE FROM pending_uploads WHERE s3_key = $1 AND user_id = $2 AND bucket = $3",
        s3_key,
        user_id as UserId,
        bucket,
    )
    .execute(executor)
    .await?;
    Ok(())
}

/// Whether `s3_key` (in `bucket`) is a pending upload this user presigned.
///
/// The confirm handlers used to prove ownership by checking the client-supplied
/// key started with the user's `{user_id}/{item_id}/...` prefix. With scan-then-
/// promote the presigned key is an owner-less `staging/{uuid}.{ext}`, so that
/// structural check no longer binds the key to a user. Every presign records the
/// key here against its owner (`record_pending_upload`); confirm now proves
/// ownership by looking it up, a caller cannot confirm a staging key it did not
/// presign (and cannot guess another user's random staging uuid).
pub(crate) async fn is_owned(
    pool: &PgPool,
    user_id: UserId,
    s3_key: &str,
    bucket: &str,
) -> Result<bool> {
    let owned = sqlx::query_scalar::<_, bool>(
        "SELECT EXISTS (SELECT 1 FROM pending_uploads WHERE s3_key = $1 AND user_id = $2 AND bucket = $3)",
    )
    .bind(s3_key)
    .bind(user_id)
    .bind(bucket)
    .fetch_one(pool)
    .await?;
    Ok(owned)
}

/// Per-tick cap on the orphan-upload reaper. The reaper runs every scheduler
/// tick under the tick-wide advisory lock and deletes serially (one S3 round-
/// trip per row), so an unbounded result set lets a backlog wedge the tick
/// (PERF-S1, Run #23). Oldest-first + a bound drains across ticks instead.
pub(crate) const STALE_UPLOAD_BATCH: i64 = 200;

/// Fetch up to [`STALE_UPLOAD_BATCH`] presigned uploads older than `max_age`
/// that were never confirmed, oldest first.
pub(crate) async fn get_stale_pending_uploads(
    pool: &PgPool,
    max_age: chrono::Duration,
) -> Result<Vec<(String, String)>> {
    let cutoff = chrono::Utc::now() - max_age;
    // runtime-checked: binds a chrono `DateTime<Utc>` (`$1`). With sqlx's `time`
    // and `chrono` features unified (the session store pulls `time`), the macro
    // infers the TIMESTAMPTZ bind as `time::OffsetDateTime`, which a chrono value
    // won't satisfy, and a bind parameter's type can't be overridden in the macro
    // (only output columns can). Mirrors mt-db's one chrono-binding write path.
    let rows: Vec<(String, String)> = sqlx::query_as(
        "SELECT s3_key, bucket FROM pending_uploads WHERE created_at < $1 \
         ORDER BY created_at LIMIT $2",
    )
    .bind(cutoff)
    .bind(STALE_UPLOAD_BATCH)
    .fetch_all(pool)
    .await?;
    Ok(rows)
}

/// Bulk-delete pending upload records by `(s3_key, bucket)` pair. Bucket-scoped
/// to match the `(s3_key, bucket)` uniqueness (migration 137): the stale reaper
/// processes per-bucket, so it must clear only the row for the bucket it acted
/// on, not every bucket that happens to share the key (Run #18 Storage B7).
pub(crate) async fn delete_pending_uploads(pool: &PgPool, keys: &[(String, String)]) -> 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();
    // Match each (key, bucket) pair positionally via UNNEST so a key in two
    // buckets only clears the rows whose bucket was actually reaped.
    sqlx::query!(
        "DELETE FROM pending_uploads pu
         USING UNNEST($1::text[], $2::text[]) AS t(s3_key, bucket)
         WHERE pu.s3_key = t.s3_key AND pu.bucket = t.bucket",
        &s3_keys as &[&str],
        &buckets as &[&str],
    )
    .execute(pool)
    .await?;
    Ok(())
}