//! DB-layer contract tests for `db::synckit_billing`, the SyncKit v2
//! developer-billing writes.
//!
//! Audit Run 18 (Testing) flagged this module's guarded UPDATEs as asserted
//! only indirectly through the HTTP/webhook billing flow. These call the
//! `db::synckit_billing` functions directly against real Postgres so the
//! invariants the billing safety leans on are pinned at the layer they live in:
//!
//!   - `apply_billing_update` persists status + period in one guarded statement,
//!   - the terminal-`canceled` guard refuses to revive a canceled app (a stray
//!     `invoice.paid` after a `deleted` can't resurrect it or refresh its period),
//!   - the epoch-period guard drops a non-positive / inverted Stripe window
//!     rather than stamping a 1970 period onto a live app,
//!   - `activate_billing` is only valid from `draft` (replay / TOCTOU conflict),
//!   - `claim_key` enforces `key_cap` under the `FOR UPDATE` lock, a claim that
//!     would exceed the cap is refused atomically, never over-allocated,
//!   - `claim_key` / `release_key` are idempotent.

use crate::harness::db::TestDb;
use chrono::{DateTime, Utc};
use makenotwork::db::synckit_billing;
use makenotwork::db::{SyncAppId, SyncBillingStatus, SyncEnforcementMode, UserId};

/// Seed a verified user.
async fn seed_user(pool: &sqlx::PgPool, username: &str) -> UserId {
    let hash = makenotwork::auth::hash_password("password123").expect("hash");
    sqlx::query_scalar::<_, UserId>(
        "INSERT INTO users (username, email, password_hash, email_verified)
         VALUES ($1, $2, $3, true) RETURNING id",
    )
    .bind(username)
    .bind(format!("{username}@test.com"))
    .bind(&hash)
    .fetch_one(pool)
    .await
    .expect("seed user")
}

/// Seed a non-internal draft sync app plus its live-usage row (the row the
/// storage layer normally inserts on app create, and that `claim_key` locks
/// `FOR UPDATE`).
async fn seed_billable_app(pool: &sqlx::PgPool, user: UserId, name: &str) -> SyncAppId {
    let app: SyncAppId = sqlx::query_scalar::<_, SyncAppId>(
        "INSERT INTO sync_apps (creator_id, name, api_key_hash, api_key_prefix, is_internal, billing_status)
         VALUES ($1, $2, $3, $4, FALSE, 'draft') RETURNING id",
    )
    .bind(user)
    .bind(name)
    .bind(format!("hash_{name}"))
    .bind(&name[..name.len().min(8)])
    .fetch_one(pool)
    .await
    .expect("seed sync app");

    sqlx::query("INSERT INTO sync_app_usage_current (app_id) VALUES ($1)")
        .bind(app)
        .execute(pool)
        .await
        .expect("seed usage row");

    app
}

/// Read the current `keys_claimed` counter off `sync_app_usage_current`.
async fn keys_claimed(pool: &sqlx::PgPool, app: SyncAppId) -> i32 {
    sqlx::query_scalar::<_, i32>(
        "SELECT keys_claimed FROM sync_app_usage_current WHERE app_id = $1",
    )
    .bind(app)
    .fetch_one(pool)
    .await
    .expect("read keys_claimed")
}

/// Count of currently-active (un-released) key rows for the app.
async fn active_key_rows(pool: &sqlx::PgPool, app: SyncAppId) -> i64 {
    sqlx::query_scalar::<_, i64>(
        "SELECT COUNT(*) FROM sync_app_keys WHERE app_id = $1 AND released_at IS NULL",
    )
    .bind(app)
    .fetch_one(pool)
    .await
    .expect("count active keys")
}

fn ts(secs: i64) -> DateTime<Utc> {
    DateTime::<Utc>::from_timestamp(secs, 0).expect("valid timestamp")
}

/// Activate billing on a draft app in `bulk` mode with a 100 GB cap. Leaves the
/// app `active` with the shape constraint satisfied (storage_gb_cap set, per-key
/// knobs NULL).
async fn activate_bulk(pool: &sqlx::PgPool, app: SyncAppId, sub_id: &str, start: i64, end: i64) {
    synckit_billing::activate_billing(
        pool,
        app,
        SyncEnforcementMode::Bulk,
        Some(100),
        None,
        None,
        sub_id,
        ts(start),
        ts(end),
    )
    .await
    .expect("activate billing");
}

// ── apply_billing_update: persistence ────────────────────────────────────────

#[tokio::test]
async fn apply_billing_update_persists_status_and_period() {
    let db = TestDb::new().await;
    let user = seed_user(&db.pool, "sbb_persist").await;
    let app = seed_billable_app(&db.pool, user, "sbbpersist").await;
    activate_bulk(&db.pool, app, "sub_persist", 1_700_000_000, 1_700_100_000).await;

    // A Stripe-driven update flips status and refreshes the period atomically.
    let updated = synckit_billing::apply_billing_update(
        &db.pool,
        app,
        Some("suspended_unpaid"),
        Some((1_700_200_000, 1_700_300_000)),
    )
    .await
    .expect("apply update");
    assert!(updated, "a live app row is updated");

    let billing = synckit_billing::get_app_with_billing(&db.pool, app)
        .await
        .expect("load billing")
        .expect("app exists");
    assert_eq!(billing.billing_status, SyncBillingStatus::SuspendedUnpaid);
    assert_eq!(billing.current_period_start, Some(ts(1_700_200_000)));
    assert_eq!(billing.current_period_end, Some(ts(1_700_300_000)));
}

// ── apply_billing_update: terminal-canceled guard ────────────────────────────

#[tokio::test]
async fn apply_billing_update_cannot_revive_a_canceled_app() {
    let db = TestDb::new().await;
    let user = seed_user(&db.pool, "sbb_revive").await;
    let app = seed_billable_app(&db.pool, user, "sbbrevive").await;
    activate_bulk(&db.pool, app, "sub_revive", 1_700_000_000, 1_700_100_000).await;

    // Cancel is terminal. (A `customer.subscription.deleted` webhook.)
    let canceled = synckit_billing::apply_billing_update(&db.pool, app, Some("canceled"), None)
        .await
        .expect("cancel");
    assert!(canceled, "the cancel transition itself matches a live row");

    // A stray `invoice.paid` afterward tries to move canceled -> active AND
    // refresh the period. The guard must refuse both in one statement.
    let revived = synckit_billing::apply_billing_update(
        &db.pool,
        app,
        Some("active"),
        Some((1_700_900_000, 1_701_000_000)),
    )
    .await
    .expect("apply update");
    assert!(!revived, "no row updated: canceled is terminal");

    let billing = synckit_billing::get_app_with_billing(&db.pool, app)
        .await
        .expect("load billing")
        .expect("app exists");
    assert_eq!(
        billing.billing_status,
        SyncBillingStatus::Canceled,
        "status stays canceled"
    );
    // The period must not have been refreshed by the refused update.
    assert_eq!(
        billing.current_period_end,
        Some(ts(1_700_100_000)),
        "period is not refreshed on a canceled app"
    );
}

// ── apply_billing_update: epoch-period guard ─────────────────────────────────

#[tokio::test]
async fn apply_billing_update_ignores_a_nonpositive_period() {
    let db = TestDb::new().await;
    let user = seed_user(&db.pool, "sbb_epoch").await;
    let app = seed_billable_app(&db.pool, user, "sbbepoch").await;
    activate_bulk(&db.pool, app, "sub_epoch", 1_700_000_000, 1_700_100_000).await;

    // A thin/zero webhook: end <= 0. The row still matches (status unchanged),
    // but the COALESCE must keep the existing period rather than stamp 1970.
    let matched = synckit_billing::apply_billing_update(&db.pool, app, None, Some((0, 0)))
        .await
        .expect("apply update");
    assert!(
        matched,
        "the live row still matches even with no writable fields"
    );

    let billing = synckit_billing::get_app_with_billing(&db.pool, app)
        .await
        .expect("load billing")
        .expect("app exists");
    assert_eq!(
        billing.current_period_start,
        Some(ts(1_700_000_000)),
        "existing period start is preserved"
    );
    assert_eq!(
        billing.current_period_end,
        Some(ts(1_700_100_000)),
        "a non-positive window never stamps a 1970 period"
    );
}

#[tokio::test]
async fn apply_billing_update_ignores_an_inverted_period() {
    let db = TestDb::new().await;
    let user = seed_user(&db.pool, "sbb_inverted").await;
    let app = seed_billable_app(&db.pool, user, "sbbinvert").await;
    activate_bulk(&db.pool, app, "sub_invert", 1_700_000_000, 1_700_100_000).await;

    // end > 0 but start > end: an inverted range writes nothing.
    let matched = synckit_billing::apply_billing_update(
        &db.pool,
        app,
        None,
        Some((1_800_000_000, 1_700_000_000)),
    )
    .await
    .expect("apply update");
    assert!(matched);

    let billing = synckit_billing::get_app_with_billing(&db.pool, app)
        .await
        .expect("load billing")
        .expect("app exists");
    assert_eq!(billing.current_period_start, Some(ts(1_700_000_000)));
    assert_eq!(billing.current_period_end, Some(ts(1_700_100_000)));
}

// ── activate_billing: draft-only guard ───────────────────────────────────────

#[tokio::test]
async fn activate_billing_is_only_valid_from_draft() {
    let db = TestDb::new().await;
    let user = seed_user(&db.pool, "sbb_activate").await;
    let app = seed_billable_app(&db.pool, user, "sbbactiv").await;

    activate_bulk(&db.pool, app, "sub_first", 1_700_000_000, 1_700_100_000).await;

    let billing = synckit_billing::get_app_with_billing(&db.pool, app)
        .await
        .expect("load billing")
        .expect("app exists");
    assert_eq!(billing.billing_status, SyncBillingStatus::Active);
    assert_eq!(billing.stripe_subscription_id.as_deref(), Some("sub_first"));
    assert_eq!(billing.storage_gb_cap, Some(100));

    // A replay / TOCTOU race that reaches activation again against a now-active
    // app must conflict, not silently orphan the live subscription.
    let err = synckit_billing::activate_billing(
        &db.pool,
        app,
        SyncEnforcementMode::Bulk,
        Some(250),
        None,
        None,
        "sub_second",
        ts(1_700_500_000),
        ts(1_700_600_000),
    )
    .await
    .expect_err("re-activation of a non-draft app must fail");
    assert!(
        matches!(err, makenotwork::error::AppError::Conflict(_)),
        "expected a Conflict, got {err:?}"
    );

    // The first subscription id is untouched.
    let after = synckit_billing::get_app_with_billing(&db.pool, app)
        .await
        .expect("load billing")
        .expect("app exists");
    assert_eq!(after.stripe_subscription_id.as_deref(), Some("sub_first"));
    assert_eq!(
        after.storage_gb_cap,
        Some(100),
        "knobs are not overwritten by the refused activation"
    );
}

// ── claim_key: cap enforcement under the FOR UPDATE lock ──────────────────────

#[tokio::test]
async fn claim_key_enforces_cap_atomically() {
    let db = TestDb::new().await;
    let user = seed_user(&db.pool, "sbb_cap").await;
    let app = seed_billable_app(&db.pool, user, "sbbcap").await;

    // key_cap = 2. Two distinct claims fill it; the third is refused.
    let first = synckit_billing::claim_key(&db.pool, app, "k1", Some(2))
        .await
        .expect("claim k1");
    assert!(first.newly_claimed);
    assert!(!first.cap_reached);
    assert_eq!(first.total_claimed, 1);

    let second = synckit_billing::claim_key(&db.pool, app, "k2", Some(2))
        .await
        .expect("claim k2");
    assert!(second.newly_claimed);
    assert_eq!(second.total_claimed, 2);

    let third = synckit_billing::claim_key(&db.pool, app, "k3", Some(2))
        .await
        .expect("claim k3");
    assert!(!third.newly_claimed, "a new slot over the cap is refused");
    assert!(third.cap_reached);
    assert_eq!(third.total_claimed, 2, "the cap is not overshot");

    // The refused claim inserted no row and left the counter at the cap.
    assert_eq!(keys_claimed(&db.pool, app).await, 2);
    assert_eq!(active_key_rows(&db.pool, app).await, 2);
}

#[tokio::test]
async fn claim_key_is_idempotent_and_admits_reclaims_at_cap() {
    let db = TestDb::new().await;
    let user = seed_user(&db.pool, "sbb_reclaim").await;
    let app = seed_billable_app(&db.pool, user, "sbbreclaim").await;

    // key_cap = 1, filled by k1.
    let first = synckit_billing::claim_key(&db.pool, app, "k1", Some(1))
        .await
        .expect("claim k1");
    assert!(first.newly_claimed);
    assert_eq!(first.total_claimed, 1);

    // Re-claiming the already-active key consumes no new slot: not newly
    // claimed, and NOT reported as cap_reached (it's admitted).
    let reclaim = synckit_billing::claim_key(&db.pool, app, "k1", Some(1))
        .await
        .expect("re-claim k1");
    assert!(!reclaim.newly_claimed, "re-claim inserts nothing");
    assert!(
        !reclaim.cap_reached,
        "an admitted re-claim is not a cap refusal"
    );
    assert_eq!(reclaim.total_claimed, 1);

    // Exactly one active row and counter of 1, no double count.
    assert_eq!(keys_claimed(&db.pool, app).await, 1);
    assert_eq!(active_key_rows(&db.pool, app).await, 1);
}

// ── release_key: idempotency ─────────────────────────────────────────────────

#[tokio::test]
async fn release_key_is_idempotent() {
    let db = TestDb::new().await;
    let user = seed_user(&db.pool, "sbb_release").await;
    let app = seed_billable_app(&db.pool, user, "sbbrelease").await;

    synckit_billing::claim_key(&db.pool, app, "k1", Some(5))
        .await
        .expect("claim k1");

    let first = synckit_billing::release_key(&db.pool, app, "k1")
        .await
        .expect("release k1");
    assert!(first.newly_released);
    assert_eq!(first.total_claimed, 0);

    // Releasing an already-released key is a no-op, and the counter never goes
    // negative.
    let second = synckit_billing::release_key(&db.pool, app, "k1")
        .await
        .expect("re-release k1");
    assert!(!second.newly_released, "no active row to release");
    assert_eq!(second.total_claimed, 0);
    assert_eq!(keys_claimed(&db.pool, app).await, 0);
    assert_eq!(active_key_rows(&db.pool, app).await, 0);
}