//! DB-layer contract tests for `db::synckit::subscriptions`, the end-user
//! entitlement for first-party sync.
//!
//! This module decides who is allowed to sync and what storage they hold, and
//! its writes all arrive from Stripe webhooks, which redeliver. The pieces
//! pinned here are the ones a wrong answer costs money or data on:
//!
//!   - `internal_write_allowed` is a status-only gate: exactly `active` opens
//!     writes, every other lifecycle status (`trialing`, `past_due`, `unpaid`,
//!     `incomplete`, `canceled`) denies, and the boundary is asserted on both
//!     sides. A lapsed `current_period_end` is deliberately NOT consulted, so
//!     that is pinned too rather than assumed.
//!   - the gate is scoped to the exact (app, user) pair: another user's
//!     subscription, or the same user's subscription on a different first-party
//!     app, opens nothing.
//!   - `create_app_sync_subscription` is the checkout webhook's write, so it is
//!     exercised under replay: an identical redelivery is a no-op that returns
//!     `false` and adds no second row, while each arm of the guard (a new Stripe
//!     subscription id, or a non-active row) does reactivate.
//!   - `update_app_sync_subscription_status`, `set_pending_storage_cap`,
//!     `set_storage_cap_now` and `apply_pending_storage_cap` are asserted to hit
//!     one row and leave every sibling row alone, which a missing key predicate
//!     would break silently.
//!
//! Deleting this file would leave the paid-sync gate and the cap-change ladder
//! asserted only through HTTP and webhook flows, where a wrong row or a second
//! credit is invisible.
//!
//! Not covered here: `get_user_app_subscription`'s per-user/per-app scoping and
//! the terminal-`canceled` and epoch-period guards on the status setter, which
//! `db_synckit_accounts_layer` already asserts.

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

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

/// Seed a first-party (internal) sync app owned by `user`. Only internal apps
/// use the end-user subscription model, so every gate test needs one.
async fn seed_internal_app(pool: &sqlx::PgPool, user: UserId, name: &str) -> SyncAppId {
    let app =
        synckit::create_sync_app(pool, user, name, &format!("key_{name}_padding"), None, None)
            .await
            .expect("seed sync app")
            .id;
    sqlx::query("UPDATE sync_apps SET is_internal = true WHERE id = $1")
        .bind(app)
        .execute(pool)
        .await
        .expect("mark app first-party");
    app
}

/// Insert a subscription through the production checkout write, asserting the
/// first delivery is a real insert.
async fn subscribe(
    pool: &sqlx::PgPool,
    user: UserId,
    app: SyncAppId,
    sub_id: &str,
    interval: &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_seed",
            interval,
            storage_limit_bytes: limit_bytes,
        },
    )
    .await
    .expect("seed subscription");
    assert!(created, "the first checkout delivery inserts a row");
}

/// How many subscription rows exist for the (user, app) pair. Used to prove a
/// replay adds nothing rather than merely returning `false`.
async fn row_count(pool: &sqlx::PgPool, user: UserId, app: SyncAppId) -> i64 {
    sqlx::query_scalar::<_, i64>(
        "SELECT COUNT(*) FROM app_sync_subscriptions WHERE user_id = $1 AND app_id = $2",
    )
    .bind(user)
    .bind(app)
    .fetch_one(pool)
    .await
    .expect("count subscription rows")
}

/// Read `canceled_at`, which the struct returned by `get_user_app_subscription`
/// does not carry but reactivation is contractually required to clear.
async fn canceled_at_is_set(pool: &sqlx::PgPool, sub_id: &str) -> bool {
    sqlx::query_scalar::<_, Option<chrono::DateTime<chrono::Utc>>>(
        "SELECT canceled_at FROM app_sync_subscriptions WHERE stripe_subscription_id = $1",
    )
    .bind(sub_id)
    .fetch_one(pool)
    .await
    .expect("read canceled_at")
    .is_some()
}

// ── internal_write_allowed: the entitlement boundary, both sides ─────────────

#[tokio::test]
async fn only_the_active_status_opens_writes() {
    let db = TestDb::new().await;
    let user = seed_user(&db.pool, "sks_status").await;
    let app = seed_internal_app(&db.pool, user, "sksstatus").await;
    subscribe(&db.pool, user, app, "sub_status", "monthly", 25_000_000_000).await;

    assert!(
        synckit::internal_write_allowed(&db.pool, app, user)
            .await
            .expect("gate query ok"),
        "an active subscription is the one status that syncs"
    );

    // Every other Stripe lifecycle status sits on the closed side of the gate.
    // `trialing` and `past_due` are the ones a plausible "not canceled" reading
    // of the rule would let through, so they are asserted individually rather
    // than as a family.
    for lapsed in ["trialing", "past_due", "unpaid", "incomplete", "canceled"] {
        synckit::update_app_sync_subscription_status(&db.pool, "sub_status", lapsed, None)
            .await
            .expect("status write ok");
        assert!(
            !synckit::internal_write_allowed(&db.pool, app, user)
                .await
                .expect("gate query ok"),
            "status {lapsed} must not open first-party sync"
        );
        // `canceled` is terminal, so it has to be the last status tried: nothing
        // after it could move the row back.
        if lapsed == "canceled" {
            break;
        }
        synckit::update_app_sync_subscription_status(&db.pool, "sub_status", "active", None)
            .await
            .expect("restore active ok");
        assert!(
            synckit::internal_write_allowed(&db.pool, app, user)
                .await
                .expect("gate query ok"),
            "returning to active reopens writes after {lapsed}"
        );
    }
}

#[tokio::test]
async fn a_lapsed_billing_period_does_not_close_the_gate_on_its_own() {
    let db = TestDb::new().await;
    let user = seed_user(&db.pool, "sks_period").await;
    let app = seed_internal_app(&db.pool, user, "sksperiod").await;
    subscribe(&db.pool, user, app, "sub_period", "annual", 7_500_000_000).await;

    // 2020-09-13, long past. The gate reads `status` alone, so entitlement
    // survives a stale period and it is Stripe flipping the status (past_due,
    // then canceled) that ends sync. Pinned because the alternative reading,
    // "deny once the period end is behind us", would lock out every user during
    // the ordinary gap between a renewal and its webhook.
    synckit::update_app_sync_subscription_status(
        &db.pool,
        "sub_period",
        "active",
        Some(1_600_000_000),
    )
    .await
    .expect("stamp a past period");

    let sub = synckit::get_user_app_subscription(&db.pool, user, app)
        .await
        .expect("read subscription")
        .expect("row exists");
    assert_eq!(
        sub.current_period_end,
        chrono::DateTime::from_timestamp(1_600_000_000, 0),
        "the raw Stripe seconds round-trip exactly"
    );
    assert!(
        synckit::internal_write_allowed(&db.pool, app, user)
            .await
            .expect("gate query ok"),
        "an active row with a lapsed period still syncs: status is the gate"
    );

    // And the far side of the boundary: the same row, one status change later.
    synckit::update_app_sync_subscription_status(&db.pool, "sub_period", "past_due", None)
        .await
        .expect("status write ok");
    assert!(
        !synckit::internal_write_allowed(&db.pool, app, user)
            .await
            .expect("gate query ok"),
        "the status is what closes it"
    );
}

#[tokio::test]
async fn one_subscription_entitles_exactly_one_user_on_exactly_one_app() {
    let db = TestDb::new().await;
    let payer = seed_user(&db.pool, "sks_payer").await;
    let freeloader = seed_user(&db.pool, "sks_freeloader").await;
    let paid_app = seed_internal_app(&db.pool, payer, "skspaid").await;
    let other_app = seed_internal_app(&db.pool, payer, "sksother").await;

    subscribe(
        &db.pool,
        payer,
        paid_app,
        "sub_scoped",
        "monthly",
        25_000_000_000,
    )
    .await;

    assert!(
        synckit::internal_write_allowed(&db.pool, paid_app, payer)
            .await
            .expect("gate query ok"),
        "the payer syncs the app they paid for"
    );
    // A gate that dropped the user predicate (EXISTS any active sub on the app)
    // would let this through, and a gate that dropped the app predicate would
    // let the next one through. Both are one deleted line away.
    assert!(
        !synckit::internal_write_allowed(&db.pool, paid_app, freeloader)
            .await
            .expect("gate query ok"),
        "another user's subscription entitles nobody else"
    );
    assert!(
        !synckit::internal_write_allowed(&db.pool, other_app, payer)
            .await
            .expect("gate query ok"),
        "paying for one first-party app entitles nothing on another"
    );

    // The freeloader subscribing opens only their own pair, and leaves the
    // payer's row untouched.
    subscribe(
        &db.pool,
        freeloader,
        paid_app,
        "sub_scoped_two",
        "monthly",
        7_500_000_000,
    )
    .await;
    assert!(
        synckit::internal_write_allowed(&db.pool, paid_app, freeloader)
            .await
            .expect("gate query ok")
    );
    assert!(
        !synckit::internal_write_allowed(&db.pool, other_app, freeloader)
            .await
            .expect("gate query ok"),
        "the second subscription is scoped the same way as the first"
    );
}

// ── create_app_sync_subscription: the checkout webhook, under replay ─────────

#[tokio::test]
async fn a_redelivered_checkout_webhook_writes_nothing_a_second_time() {
    let db = TestDb::new().await;
    let user = seed_user(&db.pool, "sks_replay").await;
    let app = seed_internal_app(&db.pool, user, "sksreplay").await;
    let params = NewAppSyncSubscription {
        user_id: user,
        app_id: app,
        stripe_subscription_id: "sub_replay",
        stripe_customer_id: "cus_replay",
        interval: "annual",
        storage_limit_bytes: 25_000_000_000,
    };

    assert!(
        synckit::create_app_sync_subscription(&db.pool, &params)
            .await
            .expect("first checkout ok"),
        "the first delivery inserts"
    );
    // Stripe redelivers; the guard WHERE makes an unchanged active row a no-op.
    assert!(
        !synckit::create_app_sync_subscription(&db.pool, &params)
            .await
            .expect("replay ok"),
        "a redelivered checkout for an unchanged active row writes nothing"
    );
    assert!(
        !synckit::create_app_sync_subscription(&db.pool, &params)
            .await
            .expect("second replay ok"),
        "and it stays a no-op however many times Stripe retries"
    );

    assert_eq!(
        row_count(&db.pool, user, app).await,
        1,
        "the replays added no second subscription row"
    );
    let sub = synckit::get_user_app_subscription(&db.pool, user, app)
        .await
        .expect("read subscription")
        .expect("row exists");
    assert_eq!(sub.stripe_subscription_id, "sub_replay");
    assert_eq!(sub.interval, "annual");
    assert_eq!(sub.status, "active");
    assert_eq!(
        sub.storage_limit_bytes,
        Some(25_000_000_000),
        "the cap the user paid for is unchanged by the replays"
    );
}

#[tokio::test]
async fn a_cap_change_delivered_as_a_checkout_replay_is_refused() {
    let db = TestDb::new().await;
    let user = seed_user(&db.pool, "sks_capreplay").await;
    let app = seed_internal_app(&db.pool, user, "skscapreplay").await;
    subscribe(
        &db.pool,
        user,
        app,
        "sub_capreplay",
        "monthly",
        25_000_000_000,
    )
    .await;

    // Same active row, same Stripe subscription, different cap and interval.
    // Neither arm of the guard fires, so the checkout path declines to move the
    // cap: cap changes are the `set_storage_cap_now` / `set_pending_storage_cap`
    // ladder's job, and letting a stale redelivery rewrite a cap would undo a
    // change the user already paid for.
    let changed = synckit::create_app_sync_subscription(
        &db.pool,
        &NewAppSyncSubscription {
            user_id: user,
            app_id: app,
            stripe_subscription_id: "sub_capreplay",
            stripe_customer_id: "cus_seed",
            interval: "annual",
            storage_limit_bytes: 7_500_000_000,
        },
    )
    .await
    .expect("checkout write ok");
    assert!(!changed, "an unchanged-subscription checkout is a no-op");

    let sub = synckit::get_user_app_subscription(&db.pool, user, app)
        .await
        .expect("read subscription")
        .expect("row exists");
    assert_eq!(
        sub.storage_limit_bytes,
        Some(25_000_000_000),
        "the cap on record is the one the ladder set, not the replayed one"
    );
    assert_eq!(sub.interval, "monthly", "and the interval is untouched too");
}

#[tokio::test]
async fn a_new_stripe_subscription_replaces_the_row_it_conflicts_with() {
    let db = TestDb::new().await;
    let user = seed_user(&db.pool, "sks_swap").await;
    let app = seed_internal_app(&db.pool, user, "sksswap").await;
    subscribe(
        &db.pool,
        user,
        app,
        "sub_swap_old",
        "monthly",
        7_500_000_000,
    )
    .await;

    // The user re-checks-out on a different Stripe subscription (the old one
    // ended at Stripe's side). The id differs, so the first arm of the guard
    // fires and every billing field moves together.
    let replaced = synckit::create_app_sync_subscription(
        &db.pool,
        &NewAppSyncSubscription {
            user_id: user,
            app_id: app,
            stripe_subscription_id: "sub_swap_new",
            stripe_customer_id: "cus_swap_new",
            interval: "annual",
            storage_limit_bytes: 25_000_000_000,
        },
    )
    .await
    .expect("re-checkout ok");
    assert!(
        replaced,
        "a different Stripe subscription id is a real write"
    );

    assert_eq!(
        row_count(&db.pool, user, app).await,
        1,
        "the (user, app) pair still holds exactly one subscription"
    );
    let sub = synckit::get_user_app_subscription(&db.pool, user, app)
        .await
        .expect("read subscription")
        .expect("row exists");
    assert_eq!(sub.stripe_subscription_id, "sub_swap_new");
    assert_eq!(sub.interval, "annual");
    assert_eq!(sub.storage_limit_bytes, Some(25_000_000_000));

    // The old id is no longer a route to this row, and the new one is.
    assert_eq!(
        synckit::get_subscription_by_stripe_id(&db.pool, "sub_swap_old")
            .await
            .expect("lookup ok"),
        None,
        "a webhook on the retired subscription finds nothing to update"
    );
    assert_eq!(
        synckit::get_subscription_by_stripe_id(&db.pool, "sub_swap_new")
            .await
            .expect("lookup ok"),
        Some((user, app))
    );
}

#[tokio::test]
async fn re_subscribing_after_a_cancellation_reactivates_at_checkout() {
    let db = TestDb::new().await;
    let user = seed_user(&db.pool, "sks_resub").await;
    let app = seed_internal_app(&db.pool, user, "sksresub").await;
    subscribe(&db.pool, user, app, "sub_resub", "monthly", 7_500_000_000).await;

    synckit::update_app_sync_subscription_status(&db.pool, "sub_resub", "canceled", None)
        .await
        .expect("cancel ok");
    assert!(
        canceled_at_is_set(&db.pool, "sub_resub").await,
        "cancellation stamps canceled_at"
    );
    assert!(
        !synckit::internal_write_allowed(&db.pool, app, user)
            .await
            .expect("gate query ok"),
        "a canceled subscription closes writes"
    );

    // The user pays again on the same Stripe subscription id. The second arm of
    // the guard (status != 'active') fires, so the row reactivates at checkout
    // rather than waiting for a later `customer.subscription.updated`.
    let revived = synckit::create_app_sync_subscription(
        &db.pool,
        &NewAppSyncSubscription {
            user_id: user,
            app_id: app,
            stripe_subscription_id: "sub_resub",
            stripe_customer_id: "cus_seed",
            interval: "annual",
            storage_limit_bytes: 25_000_000_000,
        },
    )
    .await
    .expect("re-subscribe ok");
    assert!(revived, "a paid re-subscribe is a write");

    let sub = synckit::get_user_app_subscription(&db.pool, user, app)
        .await
        .expect("read subscription")
        .expect("row exists");
    assert_eq!(sub.status, "active");
    assert_eq!(
        sub.interval, "annual",
        "the reactivating checkout carries the new interval"
    );
    assert_eq!(sub.storage_limit_bytes, Some(25_000_000_000));
    assert!(
        !canceled_at_is_set(&db.pool, "sub_resub").await,
        "reactivation clears canceled_at rather than leaving a canceled-looking row"
    );
    assert!(
        synckit::internal_write_allowed(&db.pool, app, user)
            .await
            .expect("gate query ok"),
        "and sync is open again"
    );

    // The very next redelivery of that same checkout is a no-op again.
    assert!(
        !synckit::create_app_sync_subscription(
            &db.pool,
            &NewAppSyncSubscription {
                user_id: user,
                app_id: app,
                stripe_subscription_id: "sub_resub",
                stripe_customer_id: "cus_seed",
                interval: "annual",
                storage_limit_bytes: 25_000_000_000,
            },
        )
        .await
        .expect("replay ok"),
        "the reactivated row is active, so the replay writes nothing"
    );
}

// ── keyed writes: one row moves, its siblings do not ─────────────────────────

#[tokio::test]
async fn a_status_webhook_moves_only_the_subscription_it_names() {
    let db = TestDb::new().await;
    let alice = seed_user(&db.pool, "sks_alice").await;
    let bob = seed_user(&db.pool, "sks_bob").await;
    let app = seed_internal_app(&db.pool, alice, "skssiblings").await;
    subscribe(&db.pool, alice, app, "sub_alice", "monthly", 25_000_000_000).await;
    subscribe(&db.pool, bob, app, "sub_bob", "annual", 7_500_000_000).await;

    synckit::update_app_sync_subscription_status(
        &db.pool,
        "sub_alice",
        "past_due",
        Some(1_800_000_000),
    )
    .await
    .expect("status write ok");

    let alice_sub = synckit::get_user_app_subscription(&db.pool, alice, app)
        .await
        .expect("read alice")
        .expect("row exists");
    assert_eq!(alice_sub.status, "past_due");
    assert_eq!(
        alice_sub.current_period_end,
        chrono::DateTime::from_timestamp(1_800_000_000, 0)
    );

    let bob_sub = synckit::get_user_app_subscription(&db.pool, bob, app)
        .await
        .expect("read bob")
        .expect("row exists");
    assert_eq!(
        bob_sub.status, "active",
        "a webhook keyed on one Stripe id must not touch another subscriber"
    );
    assert_eq!(
        bob_sub.current_period_end, None,
        "and it stamps no period on the sibling row"
    );
    assert!(
        synckit::internal_write_allowed(&db.pool, app, bob)
            .await
            .expect("gate query ok"),
        "bob keeps syncing while alice's payment is late"
    );
    assert!(
        !synckit::internal_write_allowed(&db.pool, app, alice)
            .await
            .expect("gate query ok")
    );

    // An id that names no row is a no-op, not an error and not a wildcard.
    synckit::update_app_sync_subscription_status(&db.pool, "sub_nobody", "canceled", None)
        .await
        .expect("unknown-id status write ok");
    assert_eq!(
        synckit::get_user_app_subscription(&db.pool, bob, app)
            .await
            .expect("read bob")
            .expect("row exists")
            .status,
        "active",
        "an unmatched webhook leaves every row alone"
    );
    assert_eq!(
        synckit::get_subscription_by_stripe_id(&db.pool, "sub_nobody")
            .await
            .expect("lookup ok"),
        None,
        "and an unknown Stripe id resolves to nothing"
    );
    assert_eq!(
        synckit::get_subscription_by_stripe_id(&db.pool, "sub_bob")
            .await
            .expect("lookup ok"),
        Some((bob, app)),
        "the lookup picks the row its id names, not the first row it meets"
    );
}

#[tokio::test]
async fn cap_writes_land_on_one_subscriber_at_a_time() {
    let db = TestDb::new().await;
    let alice = seed_user(&db.pool, "sks_capalice").await;
    let bob = seed_user(&db.pool, "sks_capbob").await;
    let app = seed_internal_app(&db.pool, alice, "skscaps").await;
    subscribe(
        &db.pool,
        alice,
        app,
        "sub_capalice",
        "monthly",
        25_000_000_000,
    )
    .await;
    subscribe(&db.pool, bob, app, "sub_capbob", "monthly", 7_500_000_000).await;

    // Alice queues a decrease for the next cycle; Bob's row must not move.
    synckit::set_pending_storage_cap(&db.pool, alice, app, 3_000_000_000)
        .await
        .expect("queue cap change");
    let bob_sub = synckit::get_user_app_subscription(&db.pool, bob, app)
        .await
        .expect("read bob")
        .expect("row exists");
    assert_eq!(bob_sub.storage_limit_bytes, Some(7_500_000_000));
    assert_eq!(
        bob_sub.pending_storage_limit_bytes, None,
        "one subscriber's queued change is not queued on another"
    );

    // Bob raises his cap now; the immediate write is likewise his alone.
    synckit::set_storage_cap_now(&db.pool, bob, app, 40_000_000_000)
        .await
        .expect("raise cap now");
    let alice_sub = synckit::get_user_app_subscription(&db.pool, alice, app)
        .await
        .expect("read alice")
        .expect("row exists");
    assert_eq!(
        alice_sub.storage_limit_bytes,
        Some(25_000_000_000),
        "alice keeps the cap she paid for while bob buys more"
    );
    assert_eq!(
        alice_sub.pending_storage_limit_bytes,
        Some(3_000_000_000),
        "and her queued decrease is still waiting"
    );

    // The renewal for Bob's subscription rolls only Bob's row, and since he has
    // nothing queued, the `IS NOT NULL` guard leaves his active cap alone rather
    // than nulling it out of the pending column.
    synckit::apply_pending_storage_cap(&db.pool, "sub_capbob")
        .await
        .expect("roll bob");
    assert_eq!(
        synckit::get_user_app_subscription(&db.pool, bob, app)
            .await
            .expect("read bob")
            .expect("row exists")
            .storage_limit_bytes,
        Some(40_000_000_000),
        "a renewal with nothing queued must not wipe the active cap"
    );
    assert_eq!(
        synckit::get_user_app_subscription(&db.pool, alice, app)
            .await
            .expect("read alice")
            .expect("row exists")
            .pending_storage_limit_bytes,
        Some(3_000_000_000),
        "and it did not promote alice's queued change early"
    );

    // Alice's own renewal promotes hers, once. A redelivered `invoice.paid`
    // finds nothing pending and changes nothing.
    synckit::apply_pending_storage_cap(&db.pool, "sub_capalice")
        .await
        .expect("roll alice");
    synckit::apply_pending_storage_cap(&db.pool, "sub_capalice")
        .await
        .expect("redelivered roll");
    let alice_sub = synckit::get_user_app_subscription(&db.pool, alice, app)
        .await
        .expect("read alice")
        .expect("row exists");
    assert_eq!(alice_sub.storage_limit_bytes, Some(3_000_000_000));
    assert_eq!(alice_sub.pending_storage_limit_bytes, None);
    assert_eq!(
        synckit::get_user_app_subscription(&db.pool, bob, app)
            .await
            .expect("read bob")
            .expect("row exists")
            .storage_limit_bytes,
        Some(40_000_000_000),
        "bob's cap survived both of alice's rolls"
    );
}