//! SyncKit blob storage: presigned upload/download URLs and upload confirmation.

use axum::{
    Json,
    extract::State,
    http::StatusCode,
    response::{IntoResponse, Response},
};
use serde_json::json;

use sqlx::PgPool;

use crate::{
    constants,
    db::{self, synckit_billing},
    error::{AppError, Result, ResultExt},
    synckit_auth::SyncUser,
    validation,
};

use super::{
    BlobConfirmRequest, BlobDownloadUrlRequest, BlobDownloadUrlResponse, BlobMultipartAbortRequest,
    BlobMultipartCompleteRequest, BlobMultipartPartUrl, BlobMultipartPartsRequest,
    BlobMultipartPartsResponse, BlobMultipartStartRequest, BlobMultipartStartResponse,
    BlobUploadUrlRequest, BlobUploadUrlResponse,
};

/// Request a pre-signed S3 upload URL for a blob.
///
/// Content-addressed by hash: if a blob with the same hash already exists
/// for this user/app, returns `already_exists: true` and an empty URL,
/// skipping the upload.
#[utoipa::path(post, path = "/api/v1/sync/blobs/upload", tag = "SyncKit",
    request_body = BlobUploadUrlRequest,
    responses((status = 200, description = "Pre-signed upload URL", body = BlobUploadUrlResponse)),
    security(("bearer" = [])),
)]
#[tracing::instrument(skip_all, name = "synckit::blob_upload_url")]
pub(super) async fn blob_upload_url(
    State(db): State<PgPool>,
    State(storage): State<crate::AppStorage>,
    sync_user: SyncUser,
    Json(req): Json<BlobUploadUrlRequest>,
) -> Result<Response> {
    let synckit_s3 = storage.synckit_s3.as_ref().ok_or_else(|| {
        AppError::ServiceUnavailable("SyncKit blob storage is not configured".to_string())
    })?;

    // The one-shot ceiling, not the multipart one: a single PUT is one
    // unresumable request, so anything larger belongs on the multipart session.
    if req.size_bytes <= 0 || req.size_bytes > constants::SYNCKIT_MAX_BLOB_SIZE_BYTES {
        return Err(AppError::BadRequest(format!(
            "Blob size must be between 1 and {} bytes; larger blobs upload through the multipart session",
            constants::SYNCKIT_MAX_BLOB_SIZE_BYTES
        )));
    }

    // Paid-only gate + content-address dedup. The gate refuses an unsubscribed
    // first-party user before any upload credential exists, so they can't stage
    // orphan S3 objects; storage-cap enforcement still happens atomically at
    // confirm time. Non-internal (developer-billed) apps always pass here.
    match blob_upload_gate(&db, &sync_user, &req.hash).await? {
        Some(BlobUploadStop::NoSubscription) => return Ok(no_subscription_response()),
        Some(BlobUploadStop::AlreadyExists) => {
            return Ok(Json(BlobUploadUrlResponse {
                upload_url: String::new(),
                already_exists: true,
            })
            .into_response());
        }
        None => {}
    }

    let s3_key = crate::storage::S3Client::generate_synckit_blob_key(
        sync_user.app_id,
        sync_user.user_id,
        &req.hash,
    );

    // Track the pending upload so the reaper can clean it up if never confirmed
    db::pending_uploads::record_pending_upload(&db, sync_user.user_id, &s3_key, "synckit").await?;

    let upload_url = synckit_s3
        .presign_upload(
            &s3_key,
            "application/octet-stream",
            Some(constants::SYNCKIT_BLOB_PRESIGN_EXPIRY_SECS),
            None,
            // Bind Content-Length at the S3 layer so the client can't upload
            // more than it declared. Confirm reads the actual object size as
            // the authoritative figure regardless.
            Some(req.size_bytes),
        )
        .await
        .context("presign upload for sync blob")?;

    Ok(Json(BlobUploadUrlResponse {
        upload_url,
        already_exists: false,
    })
    .into_response())
}

// --- Multipart blob session (large blobs) ---
//
// The chunked counterpart to `blob_upload_url`, and the only route past
// `SYNCKIT_MAX_BLOB_SIZE_BYTES`: a one-shot presigned PUT is a single
// unresumable request, so it keeps the modest ceiling while multipart carries
// the large blobs. These endpoints replace the *transport* only, the client
// still finishes at `/blobs/confirm`, which reads the authoritative object size
// from S3 and does all the quota/billing work unchanged.
//
// Authorization needs no key lookup here (unlike the creator-media multipart
// session, which signs owner-less `staging/{uuid}` keys): a synckit blob key is
// `{app_id}/{user_id}/{hash}`, derived server-side from the caller's JWT, so a
// caller can only ever address their own blob.

/// Largest window of presigned part URLs one `parts` call will mint. The client
/// pulls them as it progresses rather than holding hundreds of live
/// credentials for an upload that may never finish.
const BLOB_MULTIPART_PART_URL_WINDOW: u32 = 100;

/// Why a blob upload must not open. Both transports run the same gate and
/// render it in their own response shape.
enum BlobUploadStop {
    /// First-party app, unsubscribed user: don't hand out any upload
    /// credential, so they cannot stage orphan S3 objects.
    NoSubscription,
    /// This content address is already stored for this user/app.
    AlreadyExists,
}

/// Shared pre-upload checks: hash shape, the paid-only gate, and content-address
/// dedup. `Ok(None)` means the caller may open an upload. Size validation stays
/// at the call site, since the two transports have different ceilings.
async fn blob_upload_gate(
    db: &PgPool,
    sync_user: &SyncUser,
    hash: &str,
) -> Result<Option<BlobUploadStop>> {
    validation::validate_sync_blob_hash(hash)?;

    if !db::synckit::internal_write_allowed(db, sync_user.app_id, sync_user.user_id).await? {
        return Ok(Some(BlobUploadStop::NoSubscription));
    }

    if db::synckit::get_sync_blob_by_hash(db, sync_user.app_id, sync_user.user_id, hash)
        .await?
        .is_some()
    {
        return Ok(Some(BlobUploadStop::AlreadyExists));
    }

    Ok(None)
}

/// The 402 both transports return for an unsubscribed first-party user.
fn no_subscription_response() -> Response {
    (
        StatusCode::PAYMENT_REQUIRED,
        Json(json!({ "reason": "no_subscription" })),
    )
        .into_response()
}

/// Open a multipart upload session for a large blob.
///
/// `size_bytes` is the ciphertext length, which the client derives from the
/// plaintext length alone (`blob_encrypted_len`) before sealing anything. The
/// part geometry is pure arithmetic over it, so both sides compute identical
/// boundaries without a round trip.
#[utoipa::path(post, path = "/api/v1/sync/blobs/multipart/start", tag = "SyncKit",
    request_body = BlobMultipartStartRequest,
    responses((status = 200, description = "Multipart session opened", body = BlobMultipartStartResponse)),
    security(("bearer" = [])),
)]
#[tracing::instrument(skip_all, name = "synckit::blob_multipart_start")]
pub(super) async fn blob_multipart_start(
    State(db): State<PgPool>,
    State(storage): State<crate::AppStorage>,
    sync_user: SyncUser,
    Json(req): Json<BlobMultipartStartRequest>,
) -> Result<Response> {
    let synckit_s3 = storage.synckit_s3.as_ref().ok_or_else(|| {
        AppError::ServiceUnavailable("SyncKit blob storage is not configured".to_string())
    })?;

    if req.size_bytes <= 0 || req.size_bytes > constants::SYNCKIT_MAX_MULTIPART_BLOB_SIZE_BYTES {
        return Err(AppError::BadRequest(format!(
            "Blob size must be between 1 and {} bytes",
            constants::SYNCKIT_MAX_MULTIPART_BLOB_SIZE_BYTES
        )));
    }

    match blob_upload_gate(&db, &sync_user, &req.hash).await? {
        Some(BlobUploadStop::NoSubscription) => return Ok(no_subscription_response()),
        Some(BlobUploadStop::AlreadyExists) => {
            return Ok(Json(BlobMultipartStartResponse {
                upload_id: String::new(),
                part_size: 0,
                part_count: 0,
                already_exists: true,
            })
            .into_response());
        }
        None => {}
    }

    let plan =
        s3_storage::MultipartPlan::auto(req.size_bytes as u64).map_err(AppError::BadRequest)?;

    let s3_key = crate::storage::S3Client::generate_synckit_blob_key(
        sync_user.app_id,
        sync_user.user_id,
        &req.hash,
    );

    // Track the session so the orphan reaper aborts it if the client vanishes.
    // An abandoned multipart upload leaves no object at all, only billed parts,
    // which the reaper recovers by listing sessions for this key.
    db::pending_uploads::record_pending_upload(&db, sync_user.user_id, &s3_key, "synckit").await?;

    let upload_id = synckit_s3
        .create_multipart_upload(&s3_key, "application/octet-stream")
        .await?;

    tracing::info!(
        app = %sync_user.app_id, user = %sync_user.user_id,
        size = req.size_bytes, parts = plan.part_count,
        "SyncKit multipart blob upload started"
    );

    Ok(Json(BlobMultipartStartResponse {
        upload_id,
        part_size: plan.part_size,
        part_count: plan.part_count,
        already_exists: false,
    })
    .into_response())
}

/// Mint a bounded window of presigned `UploadPart` URLs, each carrying its exact
/// signed `Content-Length`, the same defense-in-depth the one-shot presign
/// applies.
#[utoipa::path(post, path = "/api/v1/sync/blobs/multipart/parts", tag = "SyncKit",
    request_body = BlobMultipartPartsRequest,
    responses((status = 200, description = "Presigned part URLs", body = BlobMultipartPartsResponse)),
    security(("bearer" = [])),
)]
#[tracing::instrument(skip_all, name = "synckit::blob_multipart_parts")]
pub(super) async fn blob_multipart_parts(
    State(storage): State<crate::AppStorage>,
    sync_user: SyncUser,
    Json(req): Json<BlobMultipartPartsRequest>,
) -> Result<Response> {
    let synckit_s3 = storage.synckit_s3.as_ref().ok_or_else(|| {
        AppError::ServiceUnavailable("SyncKit blob storage is not configured".to_string())
    })?;

    validation::validate_sync_blob_hash(&req.hash)?;
    if req.size_bytes <= 0 || req.size_bytes > constants::SYNCKIT_MAX_MULTIPART_BLOB_SIZE_BYTES {
        return Err(AppError::BadRequest(
            "Blob size is out of range".to_string(),
        ));
    }

    let plan =
        s3_storage::MultipartPlan::auto(req.size_bytes as u64).map_err(AppError::BadRequest)?;

    if req.count == 0 || req.count > BLOB_MULTIPART_PART_URL_WINDOW {
        return Err(AppError::BadRequest(format!(
            "count must be between 1 and {BLOB_MULTIPART_PART_URL_WINDOW}"
        )));
    }
    if req.first_part == 0 || req.first_part > plan.part_count {
        return Err(AppError::BadRequest(format!(
            "first_part must be between 1 and {}",
            plan.part_count
        )));
    }

    // Checksums are positional, so a short or long list would silently bind the
    // wrong digest to a part, reject it rather than guess the alignment.
    if let Some(checksums) = &req.checksums {
        if checksums.len() != req.count as usize {
            return Err(AppError::BadRequest(format!(
                "checksums must have exactly {} entries, one per requested part",
                req.count
            )));
        }
        for c in checksums {
            validation::validate_sha256_base64(c)?;
        }
    }

    let s3_key = crate::storage::S3Client::generate_synckit_blob_key(
        sync_user.app_id,
        sync_user.user_id,
        &req.hash,
    );

    let expires_in = constants::SYNCKIT_BLOB_PRESIGN_EXPIRY_SECS;
    let last = (req.first_part + req.count - 1).min(plan.part_count);
    let mut parts = Vec::with_capacity((last - req.first_part + 1) as usize);
    for part_number in req.first_part..=last {
        let content_length = plan.part_len(part_number);
        let checksum = req
            .checksums
            .as_ref()
            .and_then(|c| c.get((part_number - req.first_part) as usize))
            .map(String::as_str);
        let url = synckit_s3
            .presign_upload_part(
                &s3_key,
                &req.upload_id,
                part_number as i32,
                Some(expires_in),
                Some(content_length as i64),
                checksum,
            )
            .await
            .context("presign upload part for sync blob")?;
        parts.push(BlobMultipartPartUrl {
            part_number: part_number as i32,
            content_length,
            url,
        });
    }

    Ok(Json(BlobMultipartPartsResponse { parts, expires_in }).into_response())
}

/// Assemble the uploaded parts into the blob object.
///
/// Transport only: the client then calls `/blobs/confirm`, which reads the real
/// object size from S3 and applies every quota and billing rule.
#[utoipa::path(post, path = "/api/v1/sync/blobs/multipart/complete", tag = "SyncKit",
    request_body = BlobMultipartCompleteRequest,
    responses((status = 204, description = "Parts assembled")),
    security(("bearer" = [])),
)]
#[tracing::instrument(skip_all, name = "synckit::blob_multipart_complete")]
pub(super) async fn blob_multipart_complete(
    State(storage): State<crate::AppStorage>,
    sync_user: SyncUser,
    Json(req): Json<BlobMultipartCompleteRequest>,
) -> Result<Response> {
    let synckit_s3 = storage.synckit_s3.as_ref().ok_or_else(|| {
        AppError::ServiceUnavailable("SyncKit blob storage is not configured".to_string())
    })?;

    validation::validate_sync_blob_hash(&req.hash)?;
    if req.parts.is_empty() {
        return Err(AppError::BadRequest("No parts to complete".to_string()));
    }

    let s3_key = crate::storage::S3Client::generate_synckit_blob_key(
        sync_user.app_id,
        sync_user.user_id,
        &req.hash,
    );

    let parts: Vec<(i32, String)> = req
        .parts
        .into_iter()
        .map(|p| (p.part_number, p.etag))
        .collect();

    synckit_s3
        .complete_multipart_upload(&s3_key, &req.upload_id, &parts)
        .await?;

    tracing::info!(
        app = %sync_user.app_id, user = %sync_user.user_id, parts = parts.len(),
        "SyncKit multipart blob upload completed"
    );

    Ok(StatusCode::NO_CONTENT.into_response())
}

/// Release the parts of an abandoned session (client cancel).
///
/// Incomplete multipart uploads bill for their parts until aborted, so a client
/// that cleans up on cancel is the cheapest fix; the orphan reaper is the
/// backstop for clients that vanish.
#[utoipa::path(post, path = "/api/v1/sync/blobs/multipart/abort", tag = "SyncKit",
    request_body = BlobMultipartAbortRequest,
    responses((status = 204, description = "Session aborted")),
    security(("bearer" = [])),
)]
#[tracing::instrument(skip_all, name = "synckit::blob_multipart_abort")]
pub(super) async fn blob_multipart_abort(
    State(db): State<PgPool>,
    State(storage): State<crate::AppStorage>,
    sync_user: SyncUser,
    Json(req): Json<BlobMultipartAbortRequest>,
) -> Result<Response> {
    let synckit_s3 = storage.synckit_s3.as_ref().ok_or_else(|| {
        AppError::ServiceUnavailable("SyncKit blob storage is not configured".to_string())
    })?;

    validation::validate_sync_blob_hash(&req.hash)?;

    let s3_key = crate::storage::S3Client::generate_synckit_blob_key(
        sync_user.app_id,
        sync_user.user_id,
        &req.hash,
    );

    synckit_s3
        .abort_multipart_upload(&s3_key, &req.upload_id)
        .await?;
    // The session is gone, so the reaper has nothing left to find; drop the
    // tracking row rather than leaving it to age out.
    db::pending_uploads::remove_pending_upload(&db, sync_user.user_id, &s3_key, "synckit").await?;

    tracing::info!(
        app = %sync_user.app_id, user = %sync_user.user_id,
        "SyncKit multipart blob upload aborted"
    );

    Ok(StatusCode::NO_CONTENT.into_response())
}

/// Confirm that a blob upload to S3 completed successfully.
///
/// Verifies the object exists in S3, then records it in the database.
/// Idempotent: returns success without creating a duplicate.
///
/// Content-addressing trust model (ultra-fuzz Run 4 Storage NOTE, decision
/// 2026-06-23; revised 2026-07-21): the blob `hash` is treated as a
/// content-address LABEL, confirm reads the authoritative `object_size` from S3
/// but does not re-hash the bytes to prove they match `hash`. The blast radius
/// is per-user only: the key is `{app_id}/{user_id}/{hash}` and storage is
/// `UNIQUE(app_id, user_id, hash)`, so a client that stores mismatched bytes can
/// poison only its OWN dedup namespace, no cross-user effect, no data exposure.
///
/// This note used to say the A+ fix was binding `x-amz-checksum-sha256` into the
/// presigned PUT so S3 rejects a mismatched upload at write time. That reasoning
/// does not hold for these blobs, and the correction is worth keeping: the stored
/// object is E2E *ciphertext* sealed with random per-chunk nonces, while `hash`
/// is the SHA-256 of the *plaintext*. The server never sees plaintext, so it
/// cannot derive the expected ciphertext digest at presign time, any checksum it
/// binds has to come from the client, i.e. the party whose honesty was in
/// question. Checksum binding (which the multipart path now does per part) buys
/// transport integrity, not content-address enforcement.
///
/// What actually binds the bytes to the address is the AEAD: each chunk is sealed
/// with `(hash, chunk_index, chunk_count)` as associated data, so ciphertext that
/// opens under `hash` is cryptographically tied to it, and the client re-hashes
/// the plaintext after decrypting. A client storing mismatched bytes breaks only
/// its own blob. Server-side re-hashing would cost a full object download per
/// confirm to defend a client against itself, which is why it is not done.
#[utoipa::path(post, path = "/api/v1/sync/blobs/confirm", tag = "SyncKit",
    request_body = BlobConfirmRequest,
    responses((status = 204, description = "Upload confirmed")),
    security(("bearer" = [])),
)]
#[tracing::instrument(skip_all, name = "synckit::blob_confirm_upload")]
pub(super) async fn blob_confirm_upload(
    State(db): State<PgPool>,
    State(storage): State<crate::AppStorage>,
    sync_user: SyncUser,
    Json(req): Json<BlobConfirmRequest>,
) -> Result<Response> {
    let synckit_s3 = storage.synckit_s3.as_ref().ok_or_else(|| {
        AppError::ServiceUnavailable("SyncKit blob storage is not configured".to_string())
    })?;

    validation::validate_sync_blob_hash(&req.hash)?;

    let billing = synckit_billing::get_app_with_billing(&db, sync_user.app_id)
        .await?
        .ok_or(AppError::NotFound)?;

    let s3_key = crate::storage::S3Client::generate_synckit_blob_key(
        sync_user.app_id,
        sync_user.user_id,
        &req.hash,
    );

    // The authoritative size is the actual S3 object, never the client's
    // claim. `object_size` doubles as the existence check (None = not there).
    let actual_size = synckit_s3.object_size(&s3_key).await?.ok_or_else(|| {
        AppError::BadRequest("Blob not found in storage, upload before confirming".to_string())
    })?;
    // Bounded by the multipart ceiling, not the one-shot one: confirm cannot
    // tell which transport wrote the object, and the one-shot route is already
    // bounded to its own ceiling by the signed `Content-Length` at presign time.
    if actual_size <= 0 || actual_size > constants::SYNCKIT_MAX_MULTIPART_BLOB_SIZE_BYTES {
        return Err(AppError::BadRequest(format!(
            "Stored blob size {actual_size} is out of range"
        )));
    }

    // Record the blob and enforce the cap atomically. Internal apps gate on the
    // user's paid subscription + per-user cap; developer apps on the app/per-key
    // counters. Both are single-transaction (lock, check, insert, count).
    let outcome = if billing.is_internal {
        db::synckit::confirm_internal_blob(
            &db,
            sync_user.app_id,
            sync_user.user_id,
            &req.hash,
            actual_size,
            &s3_key,
            &sync_user.key,
        )
        .await?
    } else {
        if billing.billing_status != crate::db::SyncBillingStatus::Active {
            return Ok((
                StatusCode::PAYMENT_REQUIRED,
                Json(json!({ "reason": "billing_inactive" })),
            )
                .into_response());
        }
        db::synckit::confirm_developer_blob(
            &db,
            sync_user.app_id,
            sync_user.user_id,
            &req.hash,
            actual_size,
            &s3_key,
            &sync_user.key,
            billing.enforcement_mode,
            billing.storage_gb_cap,
            billing.key_cap,
            billing.gb_per_key,
        )
        .await?
    };

    match outcome {
        db::synckit::BlobConfirm::Stored | db::synckit::BlobConfirm::AlreadyStored => {
            // Only now that the blob is durably recorded do we drop the pending
            // row. Every refusal path below leaves it in place so the orphan reaper
            // (cleanup_orphaned_uploads) reclaims the unreferenced object; clearing
            // it before the quota gate stranded the object permanently and uncharged.
            db::pending_uploads::remove_pending_upload(&db, sync_user.user_id, &s3_key, "synckit")
                .await?;
            Ok(StatusCode::NO_CONTENT.into_response())
        }
        db::synckit::BlobConfirm::NoSubscription => Ok((
            StatusCode::PAYMENT_REQUIRED,
            Json(json!({ "reason": "no_subscription" })),
        )
            .into_response()),
        db::synckit::BlobConfirm::QuotaExceeded {
            dimension,
            used,
            limit,
            key,
        } => {
            let mut body = json!({
                "reason": "storage_limit_reached",
                "dimension": dimension,
                "used": used,
                "limit": limit,
            });
            if let Some(k) = key {
                body["key"] = json!(k);
            }
            Ok((StatusCode::PAYMENT_REQUIRED, Json(body)).into_response())
        }
    }
}

/// Request a pre-signed S3 download URL for a blob by hash.
#[utoipa::path(post, path = "/api/v1/sync/blobs/download", tag = "SyncKit",
    request_body = BlobDownloadUrlRequest,
    responses((status = 200, description = "Pre-signed download URL", body = BlobDownloadUrlResponse), (status = 404, description = "Blob not found")),
    security(("bearer" = [])),
)]
#[tracing::instrument(skip_all, name = "synckit::blob_download_url")]
pub(super) async fn blob_download_url(
    State(db): State<PgPool>,
    State(storage): State<crate::AppStorage>,
    State(bg): State<crate::background::BackgroundTx>,
    sync_user: SyncUser,
    Json(req): Json<BlobDownloadUrlRequest>,
) -> Result<Response> {
    let synckit_s3 = storage.synckit_s3.as_ref().ok_or_else(|| {
        AppError::ServiceUnavailable("SyncKit blob storage is not configured".to_string())
    })?;

    validation::validate_sync_blob_hash(&req.hash)?;

    let blob =
        db::synckit::get_sync_blob_by_hash(&db, sync_user.app_id, sync_user.user_id, &req.hash)
            .await?
            .ok_or(AppError::NotFound)?;

    // Billing check (internal apps bypass). Egress is NOT enforced, it's a
    // free metric for the developer's dashboard, absorbed in the storage rate
    // margin. We still count it at presign time so devs see the stat.
    let billing = synckit_billing::get_app_with_billing(&db, sync_user.app_id)
        .await?
        .ok_or(AppError::NotFound)?;
    if !billing.is_internal {
        if billing.billing_status != crate::db::SyncBillingStatus::Active {
            return Ok((
                StatusCode::PAYMENT_REQUIRED,
                Json(json!({ "reason": "billing_inactive" })),
            )
                .into_response());
        }
        // Count egress optimistically at presign time. The client may not
        // actually download (retries that hit dedup-cached content, for
        // example), so this overcounts slightly. Acceptable for a free
        // dashboard metric. Deferred onto the bounded background pool: it's a
        // single hot-row UPDATE that the download path shouldn't wait on or
        // contend its lock against (Perf P4).
        let db = db.clone();
        let app_id = sync_user.app_id;
        let egress_bytes = blob.size_bytes;
        bg.spawn("synckit-egress-bump", async move {
            if let Err(e) = synckit_billing::add_bytes_egress(&db, app_id, egress_bytes).await {
                tracing::error!(error = ?e, app_id = %app_id, "failed to bump bytes_egress_period");
            }
        });
    }

    let download_url = synckit_s3
        .presign_download(
            &crate::storage::S3Key::from_stored(&blob.s3_key),
            Some(constants::SYNCKIT_BLOB_PRESIGN_EXPIRY_SECS),
        )
        .await
        .context("presign download for sync blob")?;

    Ok(Json(BlobDownloadUrlResponse { download_url }).into_response())
}

/// Delete a blob by hash for the authenticated user.
///
/// Frees storage immediately: the row, the S3 object, and the usage counters
/// are released in one atomic step (see `db::synckit::delete_sync_blob`). This
/// is the live shrink path that the weekly drift job used to be the only source
/// of. Allowed regardless of billing status, a user must always be able to
/// reclaim space, even on a lapsed subscription. Idempotent: deleting a hash
/// that isn't stored returns 204.
#[utoipa::path(delete, path = "/api/v1/sync/blobs/{hash}", tag = "SyncKit",
    params(("hash" = String, Path, description = "Content hash of the blob to delete")),
    responses((status = 204, description = "Blob deleted (or already absent)")),
    security(("bearer" = [])),
)]
#[tracing::instrument(skip_all, name = "synckit::blob_delete")]
pub(super) async fn blob_delete(
    State(db): State<PgPool>,
    sync_user: SyncUser,
    axum::extract::Path(hash): axum::extract::Path<String>,
) -> Result<Response> {
    validation::validate_sync_blob_hash(&hash)?;
    db::synckit::delete_sync_blob(&db, sync_user.app_id, sync_user.user_id, &hash).await?;
    Ok(StatusCode::NO_CONTENT.into_response())
}

#[cfg(test)]
mod tests {
    //! The unsubscribed-user response. Both blob transports return it, and a
    //! client distinguishes "pay us" from "you are broken" by the status alone,
    //! so the code and the reason string are a wire contract.

    use super::*;

    #[test]
    fn an_unsubscribed_user_gets_402_and_not_403() {
        let resp = no_subscription_response();
        assert_eq!(
            resp.status(),
            StatusCode::PAYMENT_REQUIRED,
            "402 tells the client to offer a subscription; 403 tells it to give up"
        );
    }
}