//! DB-layer contract tests for SyncKit devices, apps, end-user subscriptions
//! and the security audit log (`db::synckit::{devices, apps, subscriptions,
//! security}`), which had none of their own. `db_synckit_layer` covers the
//! change log, blobs and keys; see its header for what is deliberately left to
//! the HTTP workflows.
//!
//! `apps` is covered here beyond the API-key hashing its own `#[cfg(test)]`
//! module already pins, and sync-token revocation leaving the website session
//! alone stays in `synckit_security`, which asserts it end to end.

use crate::harness::db::TestDb;
use crate::harness::seed_user;

use makenotwork::db::synckit;
use makenotwork::db::synckit::NewAppSyncSubscription;
use makenotwork::db::{SyncAppId, SyncDeviceId, SyncPlatform, UserId};

/// Seed a sync app owned by `user`, with its usage row.
async fn seed_app(pool: &sqlx::PgPool, user: UserId, name: &str) -> SyncAppId {
    synckit::create_sync_app(pool, user, name, &format!("key_{name}_padding"), None, None)
        .await
        .expect("seed sync app")
        .id
}

/// Mark an app first-party, so the end-user subscription model applies.
async fn make_internal(pool: &sqlx::PgPool, app: SyncAppId) {
    sqlx::query("UPDATE sync_apps SET is_internal = true WHERE id = $1")
        .bind(app)
        .execute(pool)
        .await
        .expect("mark internal");
}

async fn seed_device(
    pool: &sqlx::PgPool,
    app: SyncAppId,
    user: UserId,
    name: &str,
) -> SyncDeviceId {
    synckit::upsert_sync_device(pool, app, user, name, SyncPlatform::Macos, None)
        .await
        .expect("seed device")
        .id
}

/// Give `user` an active subscription on `app` with a `limit_bytes` cap.
pub(crate) async fn seed_active_subscription(
    pool: &sqlx::PgPool,
    user: UserId,
    app: SyncAppId,
    sub_id: &str,
    limit_bytes: i64,
) {
    let created = synckit::create_app_sync_subscription(
        pool,
        &NewAppSyncSubscription {
            user_id: user,
            app_id: app,
            stripe_subscription_id: sub_id,
            stripe_customer_id: "cus_test",
            interval: "monthly",
            storage_limit_bytes: limit_bytes,
        },
    )
    .await
    .expect("seed subscription");
    assert!(created, "the first insert is a write");
}

// ── devices ─────────────────────────────────────────────────────────────────

#[tokio::test]
async fn re_registering_a_device_updates_the_row_it_already_has() {
    let db = TestDb::new().await;
    let user = seed_user(&db.pool, "skdev_upsert").await;
    let app = seed_app(&db.pool, user, "devupsert").await;

    let first = synckit::upsert_sync_device(
        &db.pool,
        app,
        user,
        "laptop",
        SyncPlatform::Macos,
        Some("1.2.0"),
    )
    .await
    .unwrap();
    let again = synckit::upsert_sync_device(
        &db.pool,
        app,
        user,
        "laptop",
        SyncPlatform::Linux,
        Some("1.3.0"),
    )
    .await
    .unwrap();
    assert_eq!(first.id, again.id, "the same device is not a second device");
    assert_eq!(again.platform, SyncPlatform::Linux);
    assert_eq!(again.client_version.as_deref(), Some("1.3.0"));

    // A client that stops sending a User-Agent has not downgraded to unknown.
    let quiet =
        synckit::upsert_sync_device(&db.pool, app, user, "laptop", SyncPlatform::Linux, None)
            .await
            .unwrap();
    assert_eq!(
        quiet.client_version.as_deref(),
        Some("1.3.0"),
        "None leaves the recorded version alone rather than blanking it"
    );
    assert_eq!(
        synckit::count_sync_devices(&db.pool, app, user)
            .await
            .unwrap(),
        1
    );
}

#[tokio::test]
async fn a_device_answers_only_to_its_own_owner() {
    let db = TestDb::new().await;
    let alice = seed_user(&db.pool, "skdev_alice").await;
    let bob = seed_user(&db.pool, "skdev_bob").await;
    let app = seed_app(&db.pool, alice, "devscope").await;
    let other_app = seed_app(&db.pool, alice, "devscope2").await;
    let device = seed_device(&db.pool, app, alice, "alice-laptop").await;

    assert!(
        synckit::sync_device_belongs(&db.pool, device, app, alice)
            .await
            .unwrap()
    );
    assert!(
        !synckit::sync_device_belongs(&db.pool, device, app, bob)
            .await
            .unwrap(),
        "a device id from another user's token must not verify"
    );
    assert!(
        !synckit::sync_device_belongs(&db.pool, device, other_app, alice)
            .await
            .unwrap(),
        "nor one from the same user in another app"
    );

    assert!(
        !synckit::delete_sync_device(&db.pool, device, app, bob)
            .await
            .unwrap(),
        "and a non-owner's delete removes nothing"
    );
    assert!(
        synckit::sync_device_belongs(&db.pool, device, app, alice)
            .await
            .unwrap()
    );
    assert!(
        synckit::delete_sync_device(&db.pool, device, app, alice)
            .await
            .unwrap()
    );
    assert!(
        !synckit::delete_sync_device(&db.pool, device, app, alice)
            .await
            .unwrap(),
        "deleting it twice reports no second removal"
    );

    let alice_devices = synckit::get_sync_devices(&db.pool, app, bob).await.unwrap();
    assert!(alice_devices.is_empty());
}

#[tokio::test]
async fn a_pull_cursor_only_ever_moves_forward() {
    let db = TestDb::new().await;
    let user = seed_user(&db.pool, "skdev_cursor").await;
    let app = seed_app(&db.pool, user, "devcursor").await;
    let device = seed_device(&db.pool, app, user, "laptop").await;

    synckit::touch_and_advance_cursor(&db.pool, device, 40, Some("1.4.0"))
        .await
        .unwrap();
    // An out-of-order pull response, or a client replaying an old one.
    synckit::touch_and_advance_cursor(&db.pool, device, 7, None)
        .await
        .unwrap();

    let row: (i64, Option<String>) =
        sqlx::query_as("SELECT last_pulled_seq, client_version FROM sync_devices WHERE id = $1")
            .bind(device)
            .fetch_one(&db.pool)
            .await
            .unwrap();
    assert_eq!(
        row.0, 40,
        "a lower cursor must not rewind the device, or compaction would \
         delete entries it has already seen"
    );
    assert_eq!(
        row.1.as_deref(),
        Some("1.4.0"),
        "and a version-less pull leaves the recorded version alone"
    );
}

// ── apps ────────────────────────────────────────────────────────────────────

#[tokio::test]
async fn regenerating_an_api_key_retires_the_old_one() {
    let db = TestDb::new().await;
    let user = seed_user(&db.pool, "skapp_regen").await;
    let app = synckit::create_sync_app(&db.pool, user, "regen", "sk_old_key_value", None, None)
        .await
        .unwrap();

    assert_eq!(
        synckit::get_sync_app_by_api_key(&db.pool, "sk_old_key_value")
            .await
            .unwrap()
            .map(|a| a.id),
        Some(app.id)
    );

    let updated = synckit::regenerate_sync_app_key(&db.pool, app.id, "sk_new_key_value")
        .await
        .unwrap();
    assert_eq!(
        updated.api_key_prefix, "sk_new_k",
        "the prefix is the first 8"
    );
    assert!(
        synckit::get_sync_app_by_api_key(&db.pool, "sk_old_key_value")
            .await
            .unwrap()
            .is_none(),
        "rotation is immediate: the old key stops working when this returns"
    );
    assert_eq!(
        synckit::get_sync_app_by_api_key(&db.pool, "sk_new_key_value")
            .await
            .unwrap()
            .map(|a| a.id),
        Some(app.id)
    );
}

#[tokio::test]
async fn the_keys_secret_is_a_separate_credential_from_the_api_key() {
    let db = TestDb::new().await;
    let user = seed_user(&db.pool, "skapp_secret").await;
    let app = synckit::create_sync_app(&db.pool, user, "secret", "sk_app_api_key", None, None)
        .await
        .unwrap();

    assert!(
        synckit::get_sync_app_by_keys_secret(&db.pool, "sk_app_api_key")
            .await
            .unwrap()
            .is_none(),
        "the api key ships in every client binary, so it may not open the \
         server-to-server routes"
    );

    synckit::set_sync_app_keys_secret(&db.pool, app.id, "ks_first_secret")
        .await
        .unwrap();
    assert_eq!(
        synckit::get_sync_app_by_keys_secret(&db.pool, "ks_first_secret")
            .await
            .unwrap()
            .map(|a| a.id),
        Some(app.id)
    );

    synckit::set_sync_app_keys_secret(&db.pool, app.id, "ks_second_secret")
        .await
        .unwrap();
    assert!(
        synckit::get_sync_app_by_keys_secret(&db.pool, "ks_first_secret")
            .await
            .unwrap()
            .is_none(),
        "rotation is immediate and unversioned"
    );

    sqlx::query("UPDATE sync_apps SET is_active = false WHERE id = $1")
        .bind(app.id)
        .execute(&db.pool)
        .await
        .unwrap();
    assert!(
        synckit::get_sync_app_by_keys_secret(&db.pool, "ks_second_secret")
            .await
            .unwrap()
            .is_none(),
        "a deactivated app authenticates nothing"
    );
}

// ── subscriptions ───────────────────────────────────────────────────────────

#[tokio::test]
async fn only_a_first_party_app_gates_writes_on_a_subscription() {
    let db = TestDb::new().await;
    let user = seed_user(&db.pool, "sksub_gate").await;
    let developer_app = seed_app(&db.pool, user, "gatedev").await;
    let internal_app = seed_app(&db.pool, user, "gateint").await;
    make_internal(&db.pool, internal_app).await;

    assert!(
        synckit::internal_write_allowed(&db.pool, developer_app, user)
            .await
            .unwrap(),
        "a developer-billed app's users never hold a subscription of their own"
    );
    assert!(
        !synckit::internal_write_allowed(&db.pool, internal_app, user)
            .await
            .unwrap(),
        "first-party sync is paid-only"
    );

    seed_active_subscription(&db.pool, user, internal_app, "sub_gate", 1_000).await;
    assert!(
        synckit::internal_write_allowed(&db.pool, internal_app, user)
            .await
            .unwrap()
    );

    synckit::update_app_sync_subscription_status(&db.pool, "sub_gate", "canceled", None)
        .await
        .unwrap();
    assert!(
        !synckit::internal_write_allowed(&db.pool, internal_app, user)
            .await
            .unwrap(),
        "a canceled subscription closes writes again"
    );

    assert!(
        !synckit::internal_write_allowed(&db.pool, SyncAppId::new(), user)
            .await
            .unwrap(),
        "an app that does not exist denies rather than defaults open"
    );
}

#[tokio::test]
async fn a_canceled_subscription_is_not_revived_by_a_late_webhook() {
    let db = TestDb::new().await;
    let user = seed_user(&db.pool, "sksub_terminal").await;
    let app = seed_app(&db.pool, user, "subterminal").await;
    seed_active_subscription(&db.pool, user, app, "sub_terminal", 1_000).await;

    synckit::update_app_sync_subscription_status(&db.pool, "sub_terminal", "canceled", None)
        .await
        .unwrap();
    // An `invoice.paid` that took the long way round arrives after the delete.
    synckit::update_app_sync_subscription_status(
        &db.pool,
        "sub_terminal",
        "active",
        Some(1_800_000_000),
    )
    .await
    .unwrap();

    let sub = synckit::get_user_app_subscription(&db.pool, user, app)
        .await
        .unwrap()
        .expect("row still there");
    assert_eq!(sub.status, "canceled", "canceled is terminal here");
    assert!(
        sub.current_period_end.is_none(),
        "and the refused update stamped no period either"
    );
}

#[tokio::test]
async fn a_thin_webhook_period_is_dropped_rather_than_stamped() {
    let db = TestDb::new().await;
    let user = seed_user(&db.pool, "sksub_epoch").await;
    let app = seed_app(&db.pool, user, "subepoch").await;
    seed_active_subscription(&db.pool, user, app, "sub_epoch", 1_000).await;

    synckit::update_app_sync_subscription_status(
        &db.pool,
        "sub_epoch",
        "active",
        Some(1_800_000_000),
    )
    .await
    .unwrap();
    let good = synckit::get_user_app_subscription(&db.pool, user, app)
        .await
        .unwrap()
        .unwrap()
        .current_period_end
        .expect("a real period lands");

    for thin in [None, Some(0), Some(-5)] {
        synckit::update_app_sync_subscription_status(&db.pool, "sub_epoch", "active", thin)
            .await
            .unwrap();
        assert_eq!(
            synckit::get_user_app_subscription(&db.pool, user, app)
                .await
                .unwrap()
                .unwrap()
                .current_period_end,
            Some(good),
            "a zero or missing period keeps the live one, never a 1970 stamp"
        );
    }
}

#[tokio::test]
async fn a_queued_cap_change_lands_only_at_the_period_roll() {
    let db = TestDb::new().await;
    let user = seed_user(&db.pool, "sksub_cap").await;
    let app = seed_app(&db.pool, user, "subcap").await;
    seed_active_subscription(&db.pool, user, app, "sub_cap", 1_000).await;

    synckit::set_pending_storage_cap(&db.pool, user, app, 500)
        .await
        .unwrap();
    let sub = synckit::get_user_app_subscription(&db.pool, user, app)
        .await
        .unwrap()
        .unwrap();
    assert_eq!(
        sub.storage_limit_bytes,
        Some(1_000),
        "the user paid for this period and keeps it"
    );
    assert_eq!(sub.pending_storage_limit_bytes, Some(500));

    synckit::apply_pending_storage_cap(&db.pool, "sub_cap")
        .await
        .unwrap();
    let sub = synckit::get_user_app_subscription(&db.pool, user, app)
        .await
        .unwrap()
        .unwrap();
    assert_eq!(sub.storage_limit_bytes, Some(500), "the roll promotes it");
    assert_eq!(sub.pending_storage_limit_bytes, None);

    // A second renewal must not re-apply anything.
    synckit::apply_pending_storage_cap(&db.pool, "sub_cap")
        .await
        .unwrap();
    let sub = synckit::get_user_app_subscription(&db.pool, user, app)
        .await
        .unwrap()
        .unwrap();
    assert_eq!(sub.storage_limit_bytes, Some(500));
    assert_eq!(sub.pending_storage_limit_bytes, None);
}

#[tokio::test]
async fn raising_the_cap_now_clears_the_queued_change() {
    let db = TestDb::new().await;
    let user = seed_user(&db.pool, "sksub_now").await;
    let app = seed_app(&db.pool, user, "subnow").await;
    seed_active_subscription(&db.pool, user, app, "sub_now", 1_000).await;
    synckit::set_pending_storage_cap(&db.pool, user, app, 500)
        .await
        .unwrap();

    // The user is blocked by a full cap and buys more; the price is re-quoted
    // against Stripe with prorations, so the storage arrives with the charge.
    synckit::set_storage_cap_now(&db.pool, user, app, 4_000)
        .await
        .unwrap();

    let sub = synckit::get_user_app_subscription(&db.pool, user, app)
        .await
        .unwrap()
        .unwrap();
    assert_eq!(sub.storage_limit_bytes, Some(4_000));
    assert_eq!(
        sub.pending_storage_limit_bytes, None,
        "the queued decrease is gone, not waiting to undo the purchase"
    );
    assert_eq!(
        synckit::get_subscription_by_stripe_id(&db.pool, "sub_now")
            .await
            .unwrap(),
        Some((user, app))
    );
}

#[tokio::test]
async fn a_subscription_is_read_back_per_user_and_per_app() {
    let db = TestDb::new().await;
    let alice = seed_user(&db.pool, "sksub_alice").await;
    let bob = seed_user(&db.pool, "sksub_bob").await;
    let app = seed_app(&db.pool, alice, "subscope").await;
    let other_app = seed_app(&db.pool, alice, "subscope2").await;
    seed_active_subscription(&db.pool, alice, app, "sub_scope", 1_000).await;

    assert!(
        synckit::get_user_app_subscription(&db.pool, bob, app)
            .await
            .unwrap()
            .is_none()
    );
    assert!(
        synckit::get_user_app_subscription(&db.pool, alice, other_app)
            .await
            .unwrap()
            .is_none(),
        "paying for one app is not paying for another"
    );
}

// ── security ────────────────────────────────────────────────────────────────

#[tokio::test]
async fn the_audit_log_appends_and_keeps_what_it_recorded() {
    let db = TestDb::new().await;
    let user = seed_user(&db.pool, "sksec_audit").await;
    let app = seed_app(&db.pool, user, "secaudit").await;

    synckit::record_security_event(
        &db.pool,
        app,
        Some(user),
        synckit::sync_security_event::DEVICE_REMOVED,
        Some(serde_json::json!({ "device_id": 7 })),
        Some("203.0.113.9"),
    )
    .await
    .unwrap();
    // A denied auth may not know who was knocking; the row still has to land.
    synckit::record_security_event(
        &db.pool,
        app,
        None,
        synckit::sync_security_event::AUTH_FAILURE,
        None,
        None,
    )
    .await
    .unwrap();

    let rows: Vec<(
        String,
        Option<UserId>,
        Option<serde_json::Value>,
        Option<String>,
    )> = sqlx::query_as(
        "SELECT event_type, user_id, detail, ip FROM sync_security_events
             WHERE app_id = $1 ORDER BY id",
    )
    .bind(app)
    .fetch_all(&db.pool)
    .await
    .unwrap();
    assert_eq!(rows.len(), 2, "an append, not an upsert: {rows:?}");
    assert_eq!(rows[0].0, "device_removed");
    assert_eq!(rows[0].1, Some(user));
    assert_eq!(rows[0].2, Some(serde_json::json!({ "device_id": 7 })));
    assert_eq!(rows[0].3.as_deref(), Some("203.0.113.9"));
    assert_eq!(rows[1].0, "auth_failure");
    assert_eq!(rows[1].1, None, "an unknown subject is recorded as unknown");
}

#[tokio::test]
async fn revoking_sync_tokens_touches_only_the_named_user() {
    let db = TestDb::new().await;
    let alice = seed_user(&db.pool, "sksec_alice").await;
    let bob = seed_user(&db.pool, "sksec_bob").await;

    synckit::invalidate_user_sync_tokens(&db.pool, alice)
        .await
        .unwrap();

    let stamped: Option<chrono::DateTime<chrono::Utc>> =
        sqlx::query_scalar("SELECT sync_jwt_invalidated_at FROM users WHERE id = $1")
            .bind(alice)
            .fetch_one(&db.pool)
            .await
            .unwrap();
    assert!(
        stamped.is_some(),
        "alice's sync sessions must re-authenticate"
    );
    let untouched: Option<chrono::DateTime<chrono::Utc>> =
        sqlx::query_scalar("SELECT sync_jwt_invalidated_at FROM users WHERE id = $1")
            .bind(bob)
            .fetch_one(&db.pool)
            .await
            .unwrap();
    assert!(untouched.is_none(), "and nobody else's");
}