//! Paid-only first-party (internal) SyncKit: a logged-in user of an internal
//! app (GO/BB/AF) may sync only with an `active` end-user subscription, and the
//! per-user `storage_limit_bytes` is enforced atomically from the authoritative
//! S3 object size (not the client's declared size). Closes ultra-fuzz Run 3
//! CRITICAL #1 (caps never enforced for internal apps; no sub required) and
//! SERIOUS #2 (confirm trusted the client byte count).

use std::fmt::Write as _;
use std::sync::Arc;

use makenotwork::db::{SyncAppId, UserId};
use serde_json::json;
use sqlx::PgPool;

use crate::harness::{
    BuildOptions, TestHarness, client::TestResponse, storage::InMemoryStorage, stripe,
};

const GIB: i64 = 1024 * 1024 * 1024;

pub(crate) async fn harness_with_blobs() -> (TestHarness, Arc<InMemoryStorage>) {
    let synckit_mem = Arc::new(InMemoryStorage::new());
    let mock_stripe = Arc::new(stripe::MockPaymentProvider::new());
    let mut h = TestHarness::build(BuildOptions {
        synckit_storage: Some(synckit_mem.clone()),
        stripe_client: Some(mock_stripe.clone()),
        ..Default::default()
    })
    .await;
    h.mock_stripe = Some(mock_stripe);
    (h, synckit_mem)
}

/// Create an active first-party (internal) sync app.
pub(crate) async fn create_internal_app(pool: &PgPool, user_id: UserId) -> (SyncAppId, String) {
    let api_key = "test-api-key-paid-sync";
    let key_hash = crate::harness::hash_api_key(api_key);
    let key_prefix = &api_key[..8];
    let app_id: SyncAppId = sqlx::query_scalar(
        "INSERT INTO sync_apps (creator_id, name, api_key_hash, api_key_prefix, is_internal, billing_status)
         VALUES ($1, 'GoingsOn', $2, $3, TRUE, 'active')
         RETURNING id",
    )
    .bind(user_id)
    .bind(&key_hash)
    .bind(key_prefix)
    .fetch_one(pool)
    .await
    .expect("insert internal sync_app");
    (app_id, api_key.to_string())
}

/// Seed an end-user subscription row with the given status and cap.
pub(crate) async fn seed_subscription(
    pool: &PgPool,
    user_id: UserId,
    app_id: SyncAppId,
    status: &str,
    limit_bytes: i64,
) {
    sqlx::query(
        "INSERT INTO app_sync_subscriptions
             (user_id, app_id, stripe_subscription_id, stripe_customer_id, tier, status, storage_limit_bytes)
         VALUES ($1, $2, $3, 'cus_test', 'monthly', $4, $5)",
    )
    .bind(user_id)
    .bind(app_id)
    .bind(format!("sub_{user_id}_{app_id}"))
    .bind(status)
    .bind(limit_bytes)
    .execute(pool)
    .await
    .expect("seed subscription");
}

pub(crate) fn auth_as(h: &mut TestHarness, user_id: UserId, app_id: SyncAppId, key: &str) {
    let token = makenotwork::synckit_auth::create_sync_token(
        "test-synckit-jwt-secret",
        user_id,
        app_id,
        key,
    )
    .expect("mint test JWT");
    h.client.set_bearer_token(&token);
}

pub(crate) fn fake_hash(seed: u8) -> String {
    let mut s = String::with_capacity(64);
    for _ in 0..32 {
        write!(s, "{seed:02x}").unwrap();
    }
    s
}

/// Put a blob of `actual_bytes` directly into storage, then call confirm with
/// the (possibly different) `declared_bytes`. Returns the confirm response.
async fn put_and_confirm(
    h: &mut TestHarness,
    blobs: &Arc<InMemoryStorage>,
    app_id: SyncAppId,
    user_id: UserId,
    hash: &str,
    actual_bytes: usize,
    declared_bytes: i64,
) -> TestResponse {
    let s3_key = format!("{app_id}/{user_id}/{hash}");
    blobs.put(&s3_key, vec![0u8; actual_bytes]);
    h.client
        .post_json(
            "/api/sync/blobs/confirm",
            &json!({ "hash": hash, "size_bytes": declared_bytes }).to_string(),
        )
        .await
}

// ── Tests ──

#[tokio::test]
async fn upload_url_refused_without_subscription() {
    let (mut h, _blobs) = harness_with_blobs().await;
    let user_id = h
        .signup("paid_nosub", "paid_nosub@example.com", "Password1!")
        .await;
    let (app_id, _api_key) = create_internal_app(&h.db, user_id).await;
    auth_as(&mut h, user_id, app_id, "user-key");

    let resp = h
        .client
        .post_json(
            "/api/sync/blobs/upload",
            &json!({ "hash": fake_hash(0x01), "size_bytes": 100 }).to_string(),
        )
        .await;
    assert_eq!(
        resp.status, 402,
        "upload-url must be refused without a sub: {}",
        resp.text
    );
    let body: serde_json::Value = resp.json();
    assert_eq!(body["reason"], "no_subscription");
}

#[tokio::test]
async fn confirm_refused_without_subscription() {
    let (mut h, blobs) = harness_with_blobs().await;
    let user_id = h
        .signup("paid_nosub2", "paid_nosub2@example.com", "Password1!")
        .await;
    let (app_id, _api_key) = create_internal_app(&h.db, user_id).await;
    auth_as(&mut h, user_id, app_id, "user-key");

    // Bypass the upload-url gate and put the object directly to exercise the
    // confirm-side subscription check.
    let resp = put_and_confirm(&mut h, &blobs, app_id, user_id, &fake_hash(0x02), 100, 100).await;
    assert_eq!(
        resp.status, 402,
        "confirm must be refused without a sub: {}",
        resp.text
    );
    let body: serde_json::Value = resp.json();
    assert_eq!(body["reason"], "no_subscription");
}

#[tokio::test]
async fn push_refused_without_subscription() {
    let (mut h, _blobs) = harness_with_blobs().await;
    let user_id = h
        .signup("paid_push", "paid_push@example.com", "Password1!")
        .await;
    let (app_id, _api_key) = create_internal_app(&h.db, user_id).await;
    auth_as(&mut h, user_id, app_id, "user-key");

    let resp = h
        .client
        .post_json(
            "/api/sync/push",
            &json!({
                "device_id": uuid::Uuid::new_v4().to_string(),
                "batch_id": uuid::Uuid::new_v4().to_string(),
                "changes": [
                    { "table": "tasks", "op": "INSERT", "row_id": "a", "timestamp": "2025-01-01T00:00:00Z", "data": {"t": 1} }
                ]
            })
            .to_string(),
        )
        .await;
    assert_eq!(
        resp.status, 402,
        "push must be refused without a sub: {}",
        resp.text
    );
    let body: serde_json::Value = resp.json();
    assert_eq!(body["reason"], "no_subscription");
}

#[tokio::test]
async fn confirm_succeeds_within_cap_with_active_subscription() {
    let (mut h, blobs) = harness_with_blobs().await;
    let user_id = h
        .signup("paid_ok", "paid_ok@example.com", "Password1!")
        .await;
    let (app_id, _api_key) = create_internal_app(&h.db, user_id).await;
    seed_subscription(&h.db, user_id, app_id, "active", GIB).await;
    auth_as(&mut h, user_id, app_id, "user-key");

    let resp = put_and_confirm(
        &mut h,
        &blobs,
        app_id,
        user_id,
        &fake_hash(0x03),
        1000,
        1000,
    )
    .await;
    assert_eq!(
        resp.status, 204,
        "confirm within cap should succeed: {}",
        resp.text
    );

    // The blob is recorded with the authoritative size.
    let stored: i64 = sqlx::query_scalar(
        "SELECT size_bytes FROM sync_blobs WHERE app_id = $1 AND user_id = $2 AND hash = $3",
    )
    .bind(app_id)
    .bind(user_id)
    .bind(fake_hash(0x03))
    .fetch_one(&h.db)
    .await
    .unwrap();
    assert_eq!(stored, 1000);
}

#[tokio::test]
async fn confirm_rejected_over_cap() {
    let (mut h, blobs) = harness_with_blobs().await;
    let user_id = h
        .signup("paid_over", "paid_over@example.com", "Password1!")
        .await;
    let (app_id, _api_key) = create_internal_app(&h.db, user_id).await;
    seed_subscription(&h.db, user_id, app_id, "active", 50).await; // 50-byte cap
    auth_as(&mut h, user_id, app_id, "user-key");

    let resp = put_and_confirm(&mut h, &blobs, app_id, user_id, &fake_hash(0x04), 100, 100).await;
    assert_eq!(
        resp.status, 402,
        "confirm over cap should be refused: {}",
        resp.text
    );
    let body: serde_json::Value = resp.json();
    assert_eq!(body["reason"], "storage_limit_reached");
    assert_eq!(body["dimension"], "storage");
    assert_eq!(body["limit"], 50);
}

#[tokio::test]
async fn confirm_uses_authoritative_object_size_not_client_claim() {
    let (mut h, blobs) = harness_with_blobs().await;
    let user_id = h
        .signup("paid_spoof", "paid_spoof@example.com", "Password1!")
        .await;
    let (app_id, _api_key) = create_internal_app(&h.db, user_id).await;
    seed_subscription(&h.db, user_id, app_id, "active", 50).await; // 50-byte cap
    auth_as(&mut h, user_id, app_id, "user-key");

    // Client lies: declares 1 byte but the actual S3 object is 100 bytes. The
    // cap (50) must be enforced against the real 100, not the claimed 1.
    let resp = put_and_confirm(&mut h, &blobs, app_id, user_id, &fake_hash(0x05), 100, 1).await;
    assert_eq!(
        resp.status, 402,
        "size spoof (declare 1, upload 100) must be caught by object_size: {}",
        resp.text
    );
    let body: serde_json::Value = resp.json();
    assert_eq!(body["reason"], "storage_limit_reached");
}

#[tokio::test]
async fn delete_blob_removes_row_dead_letters_object_and_frees_space() {
    let (mut h, blobs) = harness_with_blobs().await;
    let user_id = h
        .signup("paid_del", "paid_del@example.com", "Password1!")
        .await;
    let (app_id, _api_key) = create_internal_app(&h.db, user_id).await;
    seed_subscription(&h.db, user_id, app_id, "active", 1500).await; // 1500-byte cap
    auth_as(&mut h, user_id, app_id, "user-key");

    // Fill most of the cap with one 1000-byte blob.
    let hash = fake_hash(0x06);
    let resp = put_and_confirm(&mut h, &blobs, app_id, user_id, &hash, 1000, 1000).await;
    assert_eq!(
        resp.status, 204,
        "first confirm should succeed: {}",
        resp.text
    );

    // A second 1000-byte blob would breach the 1500 cap (1000 + 1000 > 1500).
    let hash2 = fake_hash(0x07);
    let resp = put_and_confirm(&mut h, &blobs, app_id, user_id, &hash2, 1000, 1000).await;
    assert_eq!(
        resp.status, 402,
        "second confirm should breach the cap: {}",
        resp.text
    );

    // Delete the first blob.
    let s3_key = format!("{app_id}/{user_id}/{hash}");
    let resp = h.client.delete(&format!("/api/sync/blobs/{hash}")).await;
    assert_eq!(resp.status, 204, "delete should succeed: {}", resp.text);

    // Row is gone.
    let remaining: i64 = sqlx::query_scalar(
        "SELECT COUNT(*) FROM sync_blobs WHERE app_id = $1 AND user_id = $2 AND hash = $3",
    )
    .bind(app_id)
    .bind(user_id)
    .bind(&hash)
    .fetch_one(&h.db)
    .await
    .unwrap();
    assert_eq!(remaining, 0, "deleted blob row must be gone");

    // The S3 object was dead-lettered for the synckit bucket, not dropped.
    let dead_lettered: i64 = sqlx::query_scalar(
        "SELECT COUNT(*) FROM pending_s3_deletions
         WHERE s3_key = $1 AND bucket = 'synckit' AND source = 'synckit_blob_delete'",
    )
    .bind(&s3_key)
    .fetch_one(&h.db)
    .await
    .unwrap();
    assert_eq!(
        dead_lettered, 1,
        "deleted object must be enqueued to the dead-letter ladder"
    );

    // Space is freed: the previously-rejected second blob now fits (internal
    // usage is summed from sync_blobs, so removing the row shrinks usage).
    let resp = put_and_confirm(&mut h, &blobs, app_id, user_id, &hash2, 1000, 1000).await;
    assert_eq!(
        resp.status, 204,
        "after delete, the second blob should fit: {}",
        resp.text
    );
}

#[tokio::test]
async fn compaction_waits_for_a_freshly_registered_device() {
    // Regression: a new device sits at last_pulled_seq = 0 until its first pull.
    // Compaction must NOT delete log entries below the other devices' cursors
    // while a never-pulled device exists, or that device's first pull-from-0
    // would silently miss the compacted-away changes.
    let (mut h, _blobs) = harness_with_blobs().await;
    let user_id = h
        .signup("compact_dev", "compact_dev@example.com", "Password1!")
        .await;
    let (app_id, _api_key) = create_internal_app(&h.db, user_id).await;

    // Device A has pulled; device B is freshly registered (seq 0).
    let dev_a: uuid::Uuid = sqlx::query_scalar(
        "INSERT INTO sync_devices (app_id, user_id, device_name, platform)
         VALUES ($1, $2, 'A', 'test') RETURNING id",
    )
    .bind(app_id)
    .bind(user_id)
    .fetch_one(&h.db)
    .await
    .unwrap();
    let _dev_b: uuid::Uuid = sqlx::query_scalar(
        "INSERT INTO sync_devices (app_id, user_id, device_name, platform)
         VALUES ($1, $2, 'B', 'test') RETURNING id",
    )
    .bind(app_id)
    .bind(user_id)
    .fetch_one(&h.db)
    .await
    .unwrap();

    // One old log entry (backdated well past the 7-day compaction margin).
    let seq: i64 = sqlx::query_scalar(
        "INSERT INTO sync_log
            (app_id, user_id, device_id, table_name, operation, row_id, client_timestamp, created_at)
         VALUES ($1, $2, $3, 'tasks', 'INSERT', 'r1', NOW(), NOW() - INTERVAL '30 days')
         RETURNING seq",
    )
    .bind(app_id).bind(user_id).bind(dev_a).fetch_one(&h.db).await.unwrap();

    // Device A has pulled past the entry; device B has not pulled at all.
    sqlx::query("UPDATE sync_devices SET last_pulled_seq = $1 WHERE id = $2")
        .bind(seq)
        .bind(dev_a)
        .execute(&h.db)
        .await
        .unwrap();

    // With device B pinned at 0, compaction must delete nothing.
    let deleted = makenotwork::db::synckit::compact_all_sync_logs(&h.db, 7)
        .await
        .unwrap();
    assert_eq!(
        deleted, 0,
        "compaction must wait for the never-pulled device"
    );
    let present: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM sync_log WHERE seq = $1")
        .bind(seq)
        .fetch_one(&h.db)
        .await
        .unwrap();
    assert_eq!(
        present, 1,
        "log entry must survive while a new device sits at seq 0"
    );

    // Once device B catches up, the same entry is safely compactable.
    sqlx::query("UPDATE sync_devices SET last_pulled_seq = $1 WHERE device_name = 'B'")
        .bind(seq)
        .execute(&h.db)
        .await
        .unwrap();
    let deleted = makenotwork::db::synckit::compact_all_sync_logs(&h.db, 7)
        .await
        .unwrap();
    assert_eq!(
        deleted, 1,
        "after every device caught up, the old entry compacts"
    );
}

#[tokio::test]
async fn delete_absent_blob_is_idempotent_no_op() {
    let (mut h, _blobs) = harness_with_blobs().await;
    let user_id = h
        .signup("paid_del2", "paid_del2@example.com", "Password1!")
        .await;
    let (app_id, _api_key) = create_internal_app(&h.db, user_id).await;
    seed_subscription(&h.db, user_id, app_id, "active", GIB).await;
    auth_as(&mut h, user_id, app_id, "user-key");

    // Deleting a hash that was never stored is a 204 no-op, and enqueues nothing.
    let resp = h
        .client
        .delete(&format!("/api/sync/blobs/{}", fake_hash(0x08)))
        .await;
    assert_eq!(
        resp.status, 204,
        "deleting an absent blob should be a 204 no-op: {}",
        resp.text
    );

    let enqueued: i64 = sqlx::query_scalar(
        "SELECT COUNT(*) FROM pending_s3_deletions WHERE source = 'synckit_blob_delete'",
    )
    .fetch_one(&h.db)
    .await
    .unwrap();
    assert_eq!(enqueued, 0, "no-op delete must not dead-letter anything");
}