//! SyncKit binary blobs: resolve a blob by content hash, confirm an uploaded
//! blob (internal vs developer paths) once its S3 object lands, and delete a
//! blob. Content-addressed, so a repeat hash dedups rather than re-storing.

use sqlx::PgPool;

use crate::db::models::DbSyncBlob;
use crate::db::{SyncAppId, UserId};
use crate::error::Result;

// ── Sync Blobs ──

/// Get a blob by content hash for a user within an app.
#[tracing::instrument(skip_all)]
pub async fn get_sync_blob_by_hash(
    pool: &PgPool,
    app_id: SyncAppId,
    user_id: UserId,
    hash: &str,
) -> Result<Option<DbSyncBlob>> {
    let blob = sqlx::query_as::<_, DbSyncBlob>(
        "SELECT * FROM sync_blobs WHERE app_id = $1 AND user_id = $2 AND hash = $3",
    )
    .bind(app_id)
    .bind(user_id)
    .bind(hash)
    .fetch_optional(pool)
    .await?;

    Ok(blob)
}

/// Result of an atomic blob-confirm. Every variant is terminal for one confirm
/// call; the route handler maps it to a 204 or a 402 with the right reason.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum BlobConfirm {
    /// Newly recorded; usage counters were incremented.
    Stored,
    /// The blob (same hash) was already recorded, idempotent re-confirm, no
    /// double counting.
    AlreadyStored,
    /// First-party app and the user has no `active` subscription (paid-only).
    NoSubscription,
    /// Storing this blob would exceed the applicable cap. Nothing was written.
    QuotaExceeded {
        dimension: &'static str,
        used: i64,
        limit: i64,
        key: Option<String>,
    },
}

/// Insert a blob row inside an open transaction. `ON CONFLICT DO UPDATE` keeps
/// `size_bytes` consistent with the actual S3 object; the idempotency check in
/// the callers means we only reach this for a genuinely new `(app,user,hash)`.
async fn insert_blob_tx(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    app_id: SyncAppId,
    user_id: UserId,
    hash: &str,
    size_bytes: i64,
    s3_key: &str,
    key: &str,
) -> Result<()> {
    sqlx::query(
        r"
        INSERT INTO sync_blobs (app_id, user_id, hash, size_bytes, s3_key, key)
        VALUES ($1, $2, $3, $4, $5, $6)
        ON CONFLICT (app_id, user_id, hash)
        DO UPDATE SET size_bytes = EXCLUDED.size_bytes
        ",
    )
    .bind(app_id)
    .bind(user_id)
    .bind(hash)
    .bind(size_bytes)
    .bind(s3_key)
    .bind(key)
    .execute(&mut **tx)
    .await?;
    Ok(())
}

/// Confirm a blob for a first-party (`is_internal`) app under the paid-only
/// end-user model. Atomic: locks the user's subscription row, enforces an
/// `active` status and the per-user `storage_limit_bytes`, then inserts, all
/// in one transaction, so concurrent confirms for the same user can't overshoot
/// the cap (the prior read-then-add gate could). Usage is summed from the
/// authoritative `sync_blobs` table, so there is no counter to drift.
#[tracing::instrument(skip_all)]
pub async fn confirm_internal_blob(
    pool: &PgPool,
    app_id: SyncAppId,
    user_id: UserId,
    hash: &str,
    size_bytes: i64,
    s3_key: &str,
    key: &str,
) -> Result<BlobConfirm> {
    let mut tx = pool.begin().await?;

    // Lock the subscription row for this user+app. Serializes this user's
    // confirms; different users don't contend.
    let sub: Option<(String, Option<i64>)> = sqlx::query_as(
        "SELECT status, storage_limit_bytes FROM app_sync_subscriptions
         WHERE app_id = $1 AND user_id = $2 FOR UPDATE",
    )
    .bind(app_id)
    .bind(user_id)
    .fetch_optional(&mut *tx)
    .await?;
    let Some((status, limit)) = sub else {
        return Ok(BlobConfirm::NoSubscription);
    };
    if status != "active" {
        return Ok(BlobConfirm::NoSubscription);
    }

    // Idempotent re-confirm: already recorded, don't recount.
    let exists: Option<i32> = sqlx::query_scalar(
        "SELECT 1 FROM sync_blobs WHERE app_id = $1 AND user_id = $2 AND hash = $3",
    )
    .bind(app_id)
    .bind(user_id)
    .bind(hash)
    .fetch_optional(&mut *tx)
    .await?;
    if exists.is_some() {
        tx.commit().await?;
        return Ok(BlobConfirm::AlreadyStored);
    }

    let used: i64 = sqlx::query_scalar(
        "SELECT COALESCE(SUM(size_bytes), 0)::BIGINT FROM sync_blobs
         WHERE app_id = $1 AND user_id = $2",
    )
    .bind(app_id)
    .bind(user_id)
    .fetch_one(&mut *tx)
    .await?;
    let limit = limit.unwrap_or(0);
    if used.saturating_add(size_bytes) > limit {
        return Ok(BlobConfirm::QuotaExceeded {
            dimension: "storage",
            used,
            limit,
            key: None,
        });
    }

    insert_blob_tx(&mut tx, app_id, user_id, hash, size_bytes, s3_key, key).await?;
    tx.commit().await?;
    Ok(BlobConfirm::Stored)
}

/// Confirm a blob for a developer-billed (non-internal) app. Atomic: locks the
/// app usage row (and the per-key row in `per_key` mode), re-checks the cap
/// under the lock, inserts, and increments the counters, folding the old
/// `would_exceed_storage` (read) + `add_bytes_stored` (add) pair into one
/// transaction so concurrent uploads can't slip past the cap. Re-confirms are
/// idempotent and never double-count.
#[tracing::instrument(skip_all)]
#[allow(clippy::too_many_arguments)]
pub async fn confirm_developer_blob(
    pool: &PgPool,
    app_id: SyncAppId,
    user_id: UserId,
    hash: &str,
    size_bytes: i64,
    s3_key: &str,
    key: &str,
    enforcement_mode: crate::db::SyncEnforcementMode,
    storage_gb_cap: Option<i32>,
    key_cap: Option<i32>,
    gb_per_key: Option<i32>,
) -> Result<BlobConfirm> {
    let mut tx = pool.begin().await?;

    // Lock the app usage row; this is the serialization point for the app-wide
    // counter. Returns current app-level bytes_stored.
    let app_used: i64 = sqlx::query_scalar(
        "SELECT bytes_stored FROM sync_app_usage_current WHERE app_id = $1 FOR UPDATE",
    )
    .bind(app_id)
    .fetch_one(&mut *tx)
    .await?;

    // Idempotent re-confirm.
    let exists: Option<i32> = sqlx::query_scalar(
        "SELECT 1 FROM sync_blobs WHERE app_id = $1 AND user_id = $2 AND hash = $3",
    )
    .bind(app_id)
    .bind(user_id)
    .bind(hash)
    .fetch_optional(&mut *tx)
    .await?;
    if exists.is_some() {
        tx.commit().await?;
        return Ok(BlobConfirm::AlreadyStored);
    }

    match enforcement_mode {
        crate::db::SyncEnforcementMode::Bulk => {
            if let Some(gb) = storage_gb_cap {
                let limit = crate::synckit_billing::storage_cap_bytes(gb as u32);
                if app_used.saturating_add(size_bytes) > limit {
                    return Ok(BlobConfirm::QuotaExceeded {
                        dimension: "storage",
                        used: app_used,
                        limit,
                        key: None,
                    });
                }
            }
        }
        crate::db::SyncEnforcementMode::PerKey => {
            if let (Some(kc), Some(g)) = (key_cap, gb_per_key) {
                let per_key_limit = crate::synckit_billing::storage_cap_bytes(g as u32);
                let app_limit =
                    crate::synckit_billing::storage_cap_bytes(kc.saturating_mul(g) as u32);
                let key_used: i64 = sqlx::query_scalar(
                    "SELECT bytes_stored FROM sync_key_usage_current
                     WHERE app_id = $1 AND key = $2 FOR UPDATE",
                )
                .bind(app_id)
                .bind(key)
                .fetch_optional(&mut *tx)
                .await?
                .unwrap_or(0);
                if key_used.saturating_add(size_bytes) > per_key_limit {
                    return Ok(BlobConfirm::QuotaExceeded {
                        dimension: "storage_per_key",
                        used: key_used,
                        limit: per_key_limit,
                        key: Some(key.to_string()),
                    });
                }
                // Defensive app-aggregate ceiling (guards counter drift).
                if app_used.saturating_add(size_bytes) > app_limit {
                    return Ok(BlobConfirm::QuotaExceeded {
                        dimension: "storage",
                        used: app_used,
                        limit: app_limit,
                        key: None,
                    });
                }
            }
        }
    }

    insert_blob_tx(&mut tx, app_id, user_id, hash, size_bytes, s3_key, key).await?;

    // Increment the app-wide and per-key counters in the same transaction.
    sqlx::query(
        "UPDATE sync_app_usage_current
         SET bytes_stored = GREATEST(bytes_stored + $2, 0), updated_at = NOW()
         WHERE app_id = $1",
    )
    .bind(app_id)
    .bind(size_bytes)
    .execute(&mut *tx)
    .await?;
    sqlx::query(
        "INSERT INTO sync_key_usage_current (app_id, key, bytes_stored)
         VALUES ($1, $2, GREATEST($3, 0))
         ON CONFLICT (app_id, key)
         DO UPDATE SET bytes_stored = GREATEST(sync_key_usage_current.bytes_stored + $3, 0),
                       updated_at = NOW()",
    )
    .bind(app_id)
    .bind(key)
    .bind(size_bytes)
    .execute(&mut *tx)
    .await?;

    tx.commit().await?;
    Ok(BlobConfirm::Stored)
}

/// Outcome of a blob delete. Terminal for one call.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum BlobDelete {
    /// Row removed; S3 object dead-lettered; usage counters refunded.
    Deleted { size_bytes: i64 },
    /// No blob with that hash for this `(app, user)`, idempotent no-op.
    NotFound,
}

/// Delete a blob by hash for a user within an app. Atomic in one transaction:
/// removes the `sync_blobs` row, refunds the developer-billing counters, and
/// dead-letters the S3 object via `pending_s3_deletions`. This is the sole
/// shrink path for storage usage that doesn't wait on the weekly drift job,
/// and because the row delete, the counter refund, and the S3-delete enqueue
/// commit together, a delete can never leave `bytes_stored` overstating reality
/// or orphan the object on a mid-operation crash.
///
/// Internal (first-party) apps keep no counter, `confirm_internal_blob` sums
/// usage straight from `sync_blobs`, so for them the counter UPDATEs match no
/// row and are harmless no-ops; removing the row is the whole refund. Idempotent:
/// deleting an absent blob returns `NotFound` with nothing written.
#[tracing::instrument(skip_all)]
pub async fn delete_sync_blob(
    pool: &PgPool,
    app_id: SyncAppId,
    user_id: UserId,
    hash: &str,
) -> Result<BlobDelete> {
    let mut tx = pool.begin().await?;

    // Remove the row, capturing what we need to refund counters and dead-letter
    // the object. SyncKit blob keys are per-`(app, user, hash)` (no cross-user
    // dedup), so deleting this row's object frees only this user's copy.
    let row: Option<(i64, String, String)> = sqlx::query_as(
        "DELETE FROM sync_blobs WHERE app_id = $1 AND user_id = $2 AND hash = $3
         RETURNING size_bytes, s3_key, key",
    )
    .bind(app_id)
    .bind(user_id)
    .bind(hash)
    .fetch_optional(&mut *tx)
    .await?;

    let Some((size_bytes, s3_key, key)) = row else {
        tx.commit().await?;
        return Ok(BlobDelete::NotFound);
    };

    // Refund the developer-billing counters. `GREATEST(_, 0)` floors at zero so
    // a drifted counter can never go negative; internal apps have no usage row
    // and these UPDATEs touch nothing.
    sqlx::query(
        "UPDATE sync_app_usage_current
         SET bytes_stored = GREATEST(bytes_stored - $2, 0), updated_at = NOW()
         WHERE app_id = $1",
    )
    .bind(app_id)
    .bind(size_bytes)
    .execute(&mut *tx)
    .await?;
    sqlx::query(
        "UPDATE sync_key_usage_current
         SET bytes_stored = GREATEST(bytes_stored - $3, 0), updated_at = NOW()
         WHERE app_id = $1 AND key = $2",
    )
    .bind(app_id)
    .bind(&key)
    .bind(size_bytes)
    .execute(&mut *tx)
    .await?;

    // Dead-letter the object in the SAME tx as the row delete + refund (model:
    // `versions::delete_version`). After commit the row is gone, so the key is
    // non-live and the deletion worker (`retry_pending_s3_deletions`, which
    // routes `bucket = "synckit"` to `synckit_s3`) can act on it.
    crate::db::pending_s3_deletions::enqueue_deletions(
        &mut *tx,
        &[(s3_key, "synckit".to_string())],
        "synckit_blob_delete",
    )
    .await?;

    tx.commit().await?;
    Ok(BlobDelete::Deleted { size_bytes })
}

/// Count devices registered for a user/app pair.
#[tracing::instrument(skip_all)]
pub async fn count_sync_devices(pool: &PgPool, app_id: SyncAppId, user_id: UserId) -> Result<i64> {
    let count: i64 =
        sqlx::query_scalar("SELECT COUNT(*) FROM sync_devices WHERE app_id = $1 AND user_id = $2")
            .bind(app_id)
            .bind(user_id)
            .fetch_one(pool)
            .await?;

    Ok(count)
}