//! DB-layer contract tests for the webhook-driven lifecycle writes in
//! `db::subscriptions`.
//!
//! These are the writes Stripe drives, and Stripe redelivers, so what is pinned
//! is what a second delivery must not do:
//!
//!   - `create_subscription`'s single-live-row cleanup cancels exactly the
//!     lingering `past_due`/`trialing`/`incomplete` rows of the SAME subscriber
//!     and project, and nobody else's,
//!   - `cancel_subscription` under redelivery keeps the first `canceled_at`
//!     rather than restamping it, which is what `COALESCE(canceled_at, NOW())`
//!     is for,
//!   - `apply_stripe_update`'s period funnel writes a zero-length Stripe window
//!     and drops an inverted one while the status still lands.
//!
//! The access gate of the same module is `db_subscriptions_layer`.
//!
//! Reach: every function in `db::subscriptions` is `pub(crate)`, so these drive
//! the Stripe webhook route, the module's own production caller, and assert on
//! the rows the functions leave behind.
//!
//! Delete this file and a redelivered cancellation could slide the cancellation
//! date forward, and the cleanup's WHERE could widen to another fan's rows,
//! with nothing at this layer noticing.

use crate::harness::stripe::{TEST_WEBHOOK_SECRET, sign_webhook_payload};
use crate::harness::{TestHarness, seed_project, seed_user};
use chrono::{DateTime, Duration, Utc};
use makenotwork::db::{ProjectId, SubscriptionTierId, UserId};
use serde_json::Value;
use sqlx::PgPool;
use std::collections::HashMap;

// ── seeding ──

async fn seed_project_tier(pool: &PgPool, project: ProjectId) -> SubscriptionTierId {
    sqlx::query_scalar::<_, SubscriptionTierId>(
        "INSERT INTO subscription_tiers (project_id, name, price_cents)
         VALUES ($1, 'Project Tier', 1500) RETURNING id",
    )
    .bind(project)
    .fetch_one(pool)
    .await
    .expect("seed project tier")
}

/// A project-scoped subscription row (`item_id` NULL).
async fn seed_project_subscription(
    pool: &PgPool,
    subscriber: UserId,
    tier: SubscriptionTierId,
    project: ProjectId,
    stripe_id: &str,
    status: &str,
    period_end: Option<DateTime<Utc>>,
) {
    sqlx::query(
        "INSERT INTO subscriptions
           (subscriber_id, tier_id, project_id, stripe_subscription_id, stripe_customer_id,
            status, current_period_start, current_period_end)
         VALUES ($1, $2, $3, $4, 'cus_gate_seed', $5, NOW() - interval '1 day', $6)",
    )
    .bind(subscriber)
    .bind(tier)
    .bind(project)
    .bind(stripe_id)
    .bind(status)
    .bind(period_end)
    .execute(pool)
    .await
    .expect("seed project subscription");
}

/// Read one subscription row's status, canceled_at and period as text-free values.
async fn read_row(
    pool: &PgPool,
    stripe_id: &str,
) -> (
    String,
    Option<DateTime<Utc>>,
    Option<DateTime<Utc>>,
    Option<DateTime<Utc>>,
) {
    sqlx::query_as(
        "SELECT status, canceled_at, current_period_start, current_period_end
         FROM subscriptions WHERE stripe_subscription_id = $1",
    )
    .bind(stripe_id)
    .fetch_one(pool)
    .await
    .unwrap_or_else(|e| panic!("read subscription {stripe_id}: {e}"))
}

async fn status_of(pool: &PgPool, stripe_id: &str) -> String {
    read_row(pool, stripe_id).await.0
}
// ── webhook-driven lifecycle ──

/// Sign a Stripe event and POST it to the webhook endpoint.
async fn post_event(
    h: &mut TestHarness,
    event_id: &str,
    event_type: &str,
    object: 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 `customer.subscription.updated` / `.deleted` object with one item carrying
/// the given raw Stripe period.
fn subscription_object(stripe_sub_id: &str, status: &str, period: Option<(i64, i64)>) -> Value {
    let items = match period {
        Some((start, end)) => serde_json::json!([{
            "id": "si_dbsl",
            "object": "subscription_item",
            "subscription": stripe_sub_id,
            "current_period_start": start,
            "current_period_end": end,
            "metadata": {},
        }]),
        None => serde_json::json!([]),
    };
    serde_json::json!({
        "id": stripe_sub_id,
        "object": "subscription",
        "status": status,
        "cancel_at_period_end": false,
        "items": {"object": "list", "data": items},
    })
}

/// Creator with a project and a tier, plus a fan. Returns
/// `(fan_id, project_id, tier_id)`.
async fn webhook_fixture(
    h: &mut TestHarness,
    tag: &str,
) -> (UserId, ProjectId, SubscriptionTierId) {
    let creator = h
        .signup(
            &format!("wcreator_{tag}"),
            &format!("wcreator_{tag}@test.com"),
            "password123",
        )
        .await;
    h.grant_creator(creator).await;
    h.client.post_form("/logout", "").await;
    h.login(&format!("wcreator_{tag}"), "password123").await;

    let resp = h
        .client
        .post_form(
            "/api/projects",
            &format!("slug=whook-{tag}&title=Webhook+{tag}"),
        )
        .await;
    assert_eq!(resp.status, 200, "create project failed: {}", resp.text);
    let project: Value = resp.json();
    let project_uuid: ProjectId = project["id"]
        .as_str()
        .expect("project id")
        .parse()
        .expect("project id parses");
    let tier = seed_project_tier(&h.db, project_uuid).await;

    h.client.post_form("/logout", "").await;
    let fan = h
        .signup(
            &format!("wfan_{tag}"),
            &format!("wfan_{tag}@test.com"),
            "password123",
        )
        .await;
    h.client.post_form("/logout", "").await;

    (fan, project_uuid, tier)
}

#[tokio::test]
async fn a_new_subscription_cancels_only_the_same_fans_lingering_rows_for_that_project() {
    let mut h = TestHarness::with_stripe().await;
    let (fan, project, tier) = webhook_fixture(&mut h, "cleanup").await;

    // A second project and a second fan, so the cleanup's WHERE has something
    // to get wrong in each direction. Who owns the second project is irrelevant
    // here; only the (subscriber, project) pair is.
    let other_project = seed_project(&h.db, fan, "cleanup-other").await;
    let other_tier = seed_project_tier(&h.db, other_project).await;
    let other_fan = seed_user(&h.db, "cleanup_other_fan").await;

    let period = Some(Utc::now() + Duration::days(5));
    // Cleaned up: this fan, this project, a status in the cleanup set.
    seed_project_subscription(
        &h.db,
        fan,
        tier,
        project,
        "sub_stale_past_due",
        "past_due",
        period,
    )
    .await;
    seed_project_subscription(
        &h.db,
        fan,
        tier,
        project,
        "sub_stale_trialing",
        "trialing",
        period,
    )
    .await;
    // Left alone: 'unpaid' is deliberately NOT in the cleanup set, so this row
    // is what tells "cancel the three named statuses" apart from "cancel
    // everything that is not active".
    seed_project_subscription(
        &h.db,
        fan,
        tier,
        project,
        "sub_stale_unpaid",
        "unpaid",
        period,
    )
    .await;
    // Left alone: another fan, same project.
    seed_project_subscription(
        &h.db,
        other_fan,
        tier,
        project,
        "sub_other_fan",
        "past_due",
        period,
    )
    .await;
    // Left alone: same fan, another project.
    seed_project_subscription(
        &h.db,
        fan,
        other_tier,
        other_project,
        "sub_other_project",
        "past_due",
        period,
    )
    .await;

    let mut meta = HashMap::new();
    meta.insert("checkout_type".to_string(), "subscription".to_string());
    meta.insert("subscriber_id".to_string(), fan.to_string());
    meta.insert("project_id".to_string(), project.to_string());
    meta.insert("tier_id".to_string(), tier.to_string());
    let session = serde_json::json!({
        "id": "cs_dbsl_cleanup",
        "object": "checkout_session",
        "mode": "subscription",
        "metadata": meta,
        "subscription": "sub_fresh_cleanup",
        "customer": "cus_fresh_cleanup",
    });

    let resp = post_event(
        &mut h,
        "evt_dbsl_cleanup",
        "checkout.session.completed",
        session,
    )
    .await;
    assert_eq!(
        resp.status.as_u16(),
        200,
        "subscription checkout webhook failed: {}",
        resp.text
    );

    let (fresh_status, _, _, _) = read_row(&h.db, "sub_fresh_cleanup").await;
    assert_eq!(
        fresh_status, "active",
        "the new subscription is created active"
    );

    for stale in ["sub_stale_past_due", "sub_stale_trialing"] {
        let (status, canceled_at, _, _) = read_row(&h.db, stale).await;
        assert_eq!(
            status, "canceled",
            "{stale} is a lingering live row for the resubscribing fan and must be canceled"
        );
        assert!(
            canceled_at.is_some(),
            "{stale} was canceled, so canceled_at must be stamped, got {canceled_at:?}"
        );
    }

    assert_eq!(
        status_of(&h.db, "sub_stale_unpaid").await,
        "unpaid",
        "'unpaid' is outside the cleanup set and must survive untouched"
    );
    assert_eq!(
        status_of(&h.db, "sub_other_fan").await,
        "past_due",
        "another fan's row on the same project must not be canceled"
    );
    assert_eq!(
        status_of(&h.db, "sub_other_project").await,
        "past_due",
        "the same fan's row on a different project must not be canceled"
    );
}

#[tokio::test]
async fn a_redelivered_cancellation_keeps_the_first_cancellation_time() {
    let mut h = TestHarness::with_stripe().await;
    let (fan, project, tier) = webhook_fixture(&mut h, "cancel").await;

    // Already canceled, with a known cancellation instant: this is the row a
    // Stripe redelivery of `customer.subscription.deleted` lands on.
    let first_cancel = DateTime::parse_from_rfc3339("2026-01-05T06:07:08Z")
        .expect("fixed timestamp parses")
        .with_timezone(&Utc);
    seed_project_subscription(
        &h.db,
        fan,
        tier,
        project,
        "sub_cancel_replay",
        "canceled",
        Some(Utc::now() - Duration::days(10)),
    )
    .await;
    sqlx::query("UPDATE subscriptions SET canceled_at = $1 WHERE stripe_subscription_id = $2")
        .bind(first_cancel)
        .bind("sub_cancel_replay")
        .execute(&h.db)
        .await
        .expect("stamp the original cancellation time");

    // A live row, so the same event type is shown to have an effect at all.
    seed_project_subscription(
        &h.db,
        fan,
        tier,
        project,
        "sub_cancel_fresh",
        "active",
        Some(Utc::now() + Duration::days(11)),
    )
    .await;

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

    let (status, canceled_at, _, _) = read_row(&h.db, "sub_cancel_replay").await;
    assert_eq!(status, "canceled", "the row stays canceled on redelivery");
    assert_eq!(
        canceled_at,
        Some(first_cancel),
        "COALESCE(canceled_at, NOW()) must keep the FIRST cancellation time; a restamp would \
         move a fan's end-of-access date forward on every Stripe retry"
    );

    let before = Utc::now();
    let resp = post_event(
        &mut h,
        "evt_dbsl_cancel_fresh",
        "customer.subscription.deleted",
        subscription_object(
            "sub_cancel_fresh",
            "canceled",
            Some((1_700_000_000, 1_702_592_000)),
        ),
    )
    .await;
    assert_eq!(
        resp.status.as_u16(),
        200,
        "first cancellation webhook failed: {}",
        resp.text
    );
    let (status, canceled_at, _, _) = read_row(&h.db, "sub_cancel_fresh").await;
    assert_eq!(
        status, "canceled",
        "a live row is canceled by the same event"
    );
    let canceled_at = canceled_at.expect("a first cancellation stamps canceled_at");
    assert!(
        canceled_at >= before,
        "a row with no prior canceled_at is stamped now, got {canceled_at} (test began {before})"
    );
}

#[tokio::test]
async fn a_zero_length_stripe_period_is_written_and_an_inverted_one_is_dropped() {
    let mut h = TestHarness::with_stripe().await;
    let (fan, project, tier) = webhook_fixture(&mut h, "period").await;

    seed_project_subscription(
        &h.db,
        fan,
        tier,
        project,
        "sub_period_funnel",
        "active",
        Some(Utc::now() + Duration::days(4)),
    )
    .await;

    // start == end is the boundary the funnel accepts (`end > 0 && start <= end`).
    // Picking equal values is what separates `<=` from `<`.
    let boundary = 1_767_225_600_i64; // 2026-01-01T00:00:00Z
    let resp = post_event(
        &mut h,
        "evt_dbsl_period_equal",
        "customer.subscription.updated",
        subscription_object("sub_period_funnel", "active", Some((boundary, boundary))),
    )
    .await;
    assert_eq!(
        resp.status.as_u16(),
        200,
        "zero-length period webhook failed: {}",
        resp.text
    );

    let (status, _, start, end) = read_row(&h.db, "sub_period_funnel").await;
    assert_eq!(status, "active", "the status update lands");
    assert_eq!(
        start.map(|t| t.timestamp()),
        Some(boundary),
        "a zero-length window is a legal Stripe shape and must be written, got {start:?}"
    );
    assert_eq!(
        end.map(|t| t.timestamp()),
        Some(boundary),
        "a zero-length window is a legal Stripe shape and must be written, got {end:?}"
    );

    // Inverted: end one second BEFORE start. The funnel drops the period, and
    // COALESCE keeps what is already there, but the status half still applies.
    let resp = post_event(
        &mut h,
        "evt_dbsl_period_inverted",
        "customer.subscription.updated",
        subscription_object(
            "sub_period_funnel",
            "past_due",
            Some((boundary + 86_400, boundary + 86_399)),
        ),
    )
    .await;
    assert_eq!(
        resp.status.as_u16(),
        200,
        "inverted period webhook failed: {}",
        resp.text
    );

    let (status, _, start, end) = read_row(&h.db, "sub_period_funnel").await;
    assert_eq!(
        status, "past_due",
        "an inverted period drops only the period; the status still lands"
    );
    assert_eq!(
        start.map(|t| t.timestamp()),
        Some(boundary),
        "an inverted window must leave the existing period alone, got {start:?}"
    );
    assert_eq!(
        end.map(|t| t.timestamp()),
        Some(boundary),
        "an inverted window must leave the existing period alone, got {end:?}"
    );
}