//! DB-layer contract tests for `db::fan_plus`, the Fan+ membership store.
//!
//! Every function in the module is `pub(crate)`, so it is driven here through
//! the Stripe webhook endpoint that is its only production caller, and the
//! assertions read the `fan_plus_subscriptions` rows the module writes. What is
//! pinned is the part Stripe redelivery can corrupt: a Fan+ checkout creates
//! exactly one subscription however many times it is delivered, a fresh
//! checkout revives a canceled membership rather than opening a second row, a
//! status update writes status, period and the cancel-at-period-end flag
//! together, and a repeated cancellation keeps the first `canceled_at` instead
//! of sliding it forward.
//!
//! Existing coverage of the canceled-is-terminal guard lives in
//! `stripe_webhooks.rs` and is not repeated here.
//!
//! Delete this file and nothing checks that a redelivered Fan+ webhook writes
//! one row, which is the failure Stripe's at-least-once delivery produces.

use crate::harness::TestHarness;
use crate::harness::stripe::{TEST_WEBHOOK_SECRET, sign_webhook_payload};
use makenotwork::db::UserId;
// ── db::fan_plus, driven through its only production caller ──

/// POST a signed Stripe event to the webhook endpoint.
async fn post_event(
    h: &mut TestHarness,
    event_id: &str,
    event_type: &str,
    object: serde_json::Value,
) -> crate::harness::client::TestResponse {
    let payload = serde_json::json!({
        "id": event_id,
        "type": event_type,
        "data": {"object": object},
    })
    .to_string();
    let signature = sign_webhook_payload(&payload, TEST_WEBHOOK_SECRET);
    h.client
        .request_with_headers(
            "POST",
            "/stripe/webhook",
            Some(&payload),
            &[
                ("stripe-signature", &signature),
                ("content-type", "application/json"),
            ],
        )
        .await
}

/// A completed Fan+ checkout session. `checkout_type=fan_plus` is what routes
/// it to the Fan+ handler; `user_id` is who gets the membership.
fn fan_plus_session(session_id: &str, user_id: UserId, sub_id: &str) -> serde_json::Value {
    serde_json::json!({
        "id": session_id,
        "object": "checkout.session",
        "mode": "subscription",
        "metadata": {"checkout_type": "fan_plus", "user_id": user_id.to_string()},
        "subscription": sub_id,
        "customer": format!("cus_{sub_id}"),
        "payment_status": "no_payment_required",
    })
}

/// A `customer.subscription.*` object carrying a period on its first item.
fn subscription_object(
    sub_id: &str,
    status: &str,
    cancel_at_period_end: bool,
    period: (i64, i64),
) -> serde_json::Value {
    serde_json::json!({
        "id": sub_id,
        "object": "subscription",
        "status": status,
        "cancel_at_period_end": cancel_at_period_end,
        "items": {"object": "list", "data": [{
            "id": "si_fp_layer",
            "current_period_start": period.0,
            "current_period_end": period.1,
        }]},
    })
}

async fn fan_plus_row_count(pool: &sqlx::PgPool, user_id: UserId) -> i64 {
    sqlx::query_scalar("SELECT COUNT(*) FROM fan_plus_subscriptions WHERE user_id = $1")
        .bind(user_id)
        .fetch_one(pool)
        .await
        .expect("count fan plus rows")
}

#[tokio::test]
async fn fan_plus_checkout_creates_exactly_one_subscription_under_redelivery() {
    let mut h = TestHarness::with_stripe().await;
    let user_id = h
        .signup("fplayer_new", "fplayer_new@test.com", "password123")
        .await;

    let session = fan_plus_session("cs_fp_layer_1", user_id, "sub_fp_layer_1");
    let resp = post_event(
        &mut h,
        "evt_fp_layer_create",
        "checkout.session.completed",
        session.clone(),
    )
    .await;
    assert_eq!(
        resp.status.as_u16(),
        200,
        "Fan+ checkout webhook failed: {}",
        resp.text
    );
    assert_eq!(
        fan_plus_row_count(&h.db, user_id).await,
        1,
        "one checkout creates one membership"
    );

    // Stripe's own retry: identical event id. The dedup layer must swallow it.
    let resp = post_event(
        &mut h,
        "evt_fp_layer_create",
        "checkout.session.completed",
        session.clone(),
    )
    .await;
    assert_eq!(
        resp.status.as_u16(),
        200,
        "retried delivery must be accepted, not errored: {}",
        resp.text
    );

    // A redelivery that escapes the event-id dedup (distinct id, same session)
    // reaches `create_fan_plus_subscription`, whose ON CONFLICT (user_id) is the
    // backstop: still one row, still the same Stripe ids.
    let resp = post_event(
        &mut h,
        "evt_fp_layer_create_dup",
        "checkout.session.completed",
        session,
    )
    .await;
    assert_eq!(
        resp.status.as_u16(),
        200,
        "duplicate Fan+ checkout must not error: {}",
        resp.text
    );
    assert_eq!(
        fan_plus_row_count(&h.db, user_id).await,
        1,
        "a redelivered Fan+ checkout must never mint a second membership"
    );

    let (sub_id, customer_id, status): (String, String, String) = sqlx::query_as(
        "SELECT stripe_subscription_id, stripe_customer_id, status \
         FROM fan_plus_subscriptions WHERE user_id = $1",
    )
    .bind(user_id)
    .fetch_one(&h.db)
    .await
    .unwrap();
    assert_eq!(sub_id, "sub_fp_layer_1");
    assert_eq!(customer_id, "cus_sub_fp_layer_1");
    assert_eq!(status, "active", "a new membership starts active");
}

#[tokio::test]
async fn a_new_fan_plus_checkout_reactivates_the_canceled_membership_row() {
    let mut h = TestHarness::with_stripe().await;
    let user_id = h
        .signup("fplayer_re", "fplayer_re@test.com", "password123")
        .await;

    // A previously canceled membership. The unique constraint is on user_id, so
    // re-subscribing has to reuse this row rather than insert beside it.
    sqlx::query(
        "INSERT INTO fan_plus_subscriptions \
           (user_id, stripe_subscription_id, stripe_customer_id, status, canceled_at) \
         VALUES ($1, 'sub_fp_layer_old', 'cus_fp_layer_old', 'canceled', NOW())",
    )
    .bind(user_id)
    .execute(&h.db)
    .await
    .unwrap();

    let resp = post_event(
        &mut h,
        "evt_fp_layer_resub",
        "checkout.session.completed",
        fan_plus_session("cs_fp_layer_re", user_id, "sub_fp_layer_new"),
    )
    .await;
    assert_eq!(
        resp.status.as_u16(),
        200,
        "re-subscribe webhook failed: {}",
        resp.text
    );

    assert_eq!(
        fan_plus_row_count(&h.db, user_id).await,
        1,
        "re-subscribing updates the existing row rather than adding one"
    );
    let (status, sub_id, canceled_at): (String, String, Option<chrono::DateTime<chrono::Utc>>) =
        sqlx::query_as(
            "SELECT status, stripe_subscription_id, canceled_at \
             FROM fan_plus_subscriptions WHERE user_id = $1",
        )
        .bind(user_id)
        .fetch_one(&h.db)
        .await
        .unwrap();
    assert_eq!(status, "active", "checkout is the reactivation path");
    assert_eq!(
        sub_id, "sub_fp_layer_new",
        "the row now points at the new Stripe subscription"
    );
    assert_eq!(
        canceled_at, None,
        "reactivation clears the old cancellation stamp"
    );
}

#[tokio::test]
async fn a_stripe_update_writes_fan_plus_status_period_and_cancel_flag_together() {
    let mut h = TestHarness::with_stripe().await;
    let user_id = h
        .signup("fplayer_upd", "fplayer_upd@test.com", "password123")
        .await;
    let sub_id = "sub_fp_layer_upd";
    sqlx::query(
        "INSERT INTO fan_plus_subscriptions \
           (user_id, stripe_subscription_id, stripe_customer_id, status, current_period_end) \
         VALUES ($1, $2, 'cus_fp_layer_upd', 'active', to_timestamp(1600000000))",
    )
    .bind(user_id)
    .bind(sub_id)
    .execute(&h.db)
    .await
    .unwrap();

    // Payment fell behind and the fan scheduled a cancellation: one event
    // carries status, period and the flag, and all three must land.
    let resp = post_event(
        &mut h,
        "evt_fp_layer_upd",
        "customer.subscription.updated",
        subscription_object(sub_id, "past_due", true, (1_700_000_000, 1_702_592_000)),
    )
    .await;
    assert_eq!(
        resp.status.as_u16(),
        200,
        "update webhook failed: {}",
        resp.text
    );

    let (status, period_end, cancel_pending): (String, i64, bool) = sqlx::query_as(
        "SELECT status, EXTRACT(EPOCH FROM current_period_end)::BIGINT, cancel_at_period_end \
         FROM fan_plus_subscriptions WHERE stripe_subscription_id = $1",
    )
    .bind(sub_id)
    .fetch_one(&h.db)
    .await
    .unwrap();
    assert_eq!(status, "past_due", "Stripe's status is written through");
    assert_eq!(
        period_end, 1_702_592_000,
        "the period advances to the one Stripe sent, not the seeded 1600000000"
    );
    assert!(
        cancel_pending,
        "cancel_at_period_end tracks Stripe, which owns that flag"
    );

    // And back: resuming in Stripe's portal clears the flag here.
    let resp = post_event(
        &mut h,
        "evt_fp_layer_upd_resume",
        "customer.subscription.updated",
        subscription_object(sub_id, "active", false, (1_700_000_000, 1_702_592_000)),
    )
    .await;
    assert_eq!(
        resp.status.as_u16(),
        200,
        "resume update failed: {}",
        resp.text
    );
    let (status, cancel_pending): (String, bool) = sqlx::query_as(
        "SELECT status, cancel_at_period_end FROM fan_plus_subscriptions \
         WHERE stripe_subscription_id = $1",
    )
    .bind(sub_id)
    .fetch_one(&h.db)
    .await
    .unwrap();
    assert_eq!(
        status, "active",
        "a recovered payment restores the membership"
    );
    assert!(!cancel_pending, "the scheduled cancellation is cleared");
}

#[tokio::test]
async fn a_repeated_cancellation_keeps_the_first_canceled_at() {
    let mut h = TestHarness::with_stripe().await;
    let user_id = h
        .signup("fplayer_del", "fplayer_del@test.com", "password123")
        .await;
    let sub_id = "sub_fp_layer_del";
    sqlx::query(
        "INSERT INTO fan_plus_subscriptions \
           (user_id, stripe_subscription_id, stripe_customer_id, status, current_period_end) \
         VALUES ($1, $2, 'cus_fp_layer_del', 'active', to_timestamp(1702592000))",
    )
    .bind(user_id)
    .bind(sub_id)
    .execute(&h.db)
    .await
    .unwrap();

    let resp = post_event(
        &mut h,
        "evt_fp_layer_del",
        "customer.subscription.deleted",
        subscription_object(sub_id, "canceled", false, (1_700_000_000, 1_702_592_000)),
    )
    .await;
    assert_eq!(
        resp.status.as_u16(),
        200,
        "cancellation webhook failed: {}",
        resp.text
    );

    let (status, first_canceled_at): (String, Option<chrono::DateTime<chrono::Utc>>) =
        sqlx::query_as(
            "SELECT status, canceled_at FROM fan_plus_subscriptions \
             WHERE stripe_subscription_id = $1",
        )
        .bind(sub_id)
        .fetch_one(&h.db)
        .await
        .unwrap();
    assert_eq!(status, "canceled");
    let first_canceled_at = first_canceled_at.expect("cancellation stamps canceled_at");

    // A second delivery of the cancellation (distinct event id, so it reaches
    // the handler) must be a no-op on the stamp: `COALESCE(canceled_at, NOW())`
    // is what stops the cancellation date sliding forward on every redelivery,
    // and that date is what the fan's remaining access window is judged from.
    let resp = post_event(
        &mut h,
        "evt_fp_layer_del_again",
        "customer.subscription.deleted",
        subscription_object(sub_id, "canceled", false, (1_700_000_000, 1_702_592_000)),
    )
    .await;
    assert_eq!(
        resp.status.as_u16(),
        200,
        "repeat cancellation must not error: {}",
        resp.text
    );

    let (status, second_canceled_at): (String, Option<chrono::DateTime<chrono::Utc>>) =
        sqlx::query_as(
            "SELECT status, canceled_at FROM fan_plus_subscriptions \
             WHERE stripe_subscription_id = $1",
        )
        .bind(sub_id)
        .fetch_one(&h.db)
        .await
        .unwrap();
    assert_eq!(status, "canceled", "still canceled");
    assert_eq!(
        second_canceled_at,
        Some(first_canceled_at),
        "a redelivered cancellation keeps the original canceled_at"
    );
    assert_eq!(
        fan_plus_row_count(&h.db, user_id).await,
        1,
        "cancellation never adds rows"
    );
}