use crate::api::envelope::Envelope;
use crate::error::{AppError, ErrorCode};
use crate::services::settings;
use crate::state::AppState;

use axum::body::Bytes;
use axum::extract::State;
use axum::http::HeaderMap;
use axum::routing::post;
use axum::{Json, Router};
use chrono::{TimeZone, Utc};
use hmac::{Hmac, Mac};
use percent_encoding::{utf8_percent_encode, NON_ALPHANUMERIC};
use serde::Deserialize;
use sha2::Sha256;
use sqlx::{Postgres, Transaction};

pub fn router() -> Router<AppState> {
    Router::new().route("/webhooks/stripe", post(stripe_webhook))
}

#[derive(Debug, Clone, Deserialize)]
struct StripeEvent {
    id: String,
    created: i64,
    #[serde(rename = "type")]
    type_: String,
    data: StripeEventData,
}

#[derive(Debug, Clone, Deserialize)]
struct StripeEventData {
    object: serde_json::Value,
}

async fn stripe_webhook(
    State(state): State<AppState>,
    headers: HeaderMap,
    body: Bytes,
) -> Result<Json<Envelope<serde_json::Value>>, AppError> {
    let secret = settings::get_stripe_webhook_secret(&state)
        .await
        .map_err(|err| err.with_source("stripe webhook secret not configured"))?;

    let sig = headers
        .get("Stripe-Signature")
        .and_then(|v| v.to_str().ok())
        .ok_or_else(|| AppError::new(ErrorCode::InvalidRequest, "缺少 Stripe-Signature"))?;

    verify_stripe_signature(&body, sig, &secret)?;

    let payload_str = std::str::from_utf8(&body)
        .map_err(|_| AppError::new(ErrorCode::InvalidRequest, "Webhook payload 非 UTF-8"))?;
    let event: StripeEvent = serde_json::from_str(payload_str).map_err(|err| {
        AppError::new(ErrorCode::InvalidRequest, "Webhook JSON 解析失败").with_source(err)
    })?;

    let claimed: Option<String> = sqlx::query_scalar(
        r#"
        INSERT INTO webhook_events (
          provider, provider_event_id, event_type, payload, status
        ) VALUES ('stripe', $1, $2, $3, 'processing')
        ON CONFLICT (provider, provider_event_id) DO UPDATE
        SET event_type = EXCLUDED.event_type,
            payload = EXCLUDED.payload,
            received_at = NOW(),
            processed_at = NULL,
            status = 'processing',
            error_message = NULL
        WHERE webhook_events.status IN ('received', 'failed')
           OR (
             webhook_events.status = 'processing'
             AND webhook_events.received_at < NOW() - INTERVAL '5 minutes'
           )
        RETURNING provider_event_id
        "#,
    )
    .bind(&event.id)
    .bind(&event.type_)
    .bind(&event.data.object)
    .fetch_optional(&state.db)
    .await
    .map_err(|err| AppError::new(ErrorCode::Internal, "Webhook 入库失败").with_source(err))?;

    if claimed.is_none() {
        let status: Option<String> = sqlx::query_scalar(
            "SELECT status FROM webhook_events WHERE provider = 'stripe' AND provider_event_id = $1",
        )
        .bind(&event.id)
        .fetch_optional(&state.db)
        .await
        .map_err(|err| AppError::new(ErrorCode::Internal, "Webhook 状态查询失败").with_source(err))?;
        if status.as_deref() == Some("processed") {
            return Ok(Json(Envelope {
                success: true,
                data: serde_json::json!({ "status": "duplicate" }),
            }));
        }
        return Err(AppError::new(
            ErrorCode::StorageUnavailable,
            "Webhook 事件正在处理，请稍后重试",
        ));
    }

    if let Err(err) = process_claimed_stripe_event(&state, &event).await {
        let _ = sqlx::query(
            "UPDATE webhook_events SET status = 'failed', error_message = $2, processed_at = NULL WHERE provider = 'stripe' AND provider_event_id = $1 AND status = 'processing'",
        )
        .bind(&event.id)
        .bind(err.to_string())
        .execute(&state.db)
        .await;

        return Err(err);
    }

    Ok(Json(Envelope {
        success: true,
        data: serde_json::json!({ "status": "ok" }),
    }))
}

async fn process_claimed_stripe_event(
    state: &AppState,
    event: &StripeEvent,
) -> Result<(), AppError> {
    let mut tx = state.db.begin().await.map_err(|err| {
        AppError::new(ErrorCode::Internal, "开启 Webhook 事务失败").with_source(err)
    })?;
    let status: Option<String> = sqlx::query_scalar(
        r#"
        SELECT status
        FROM webhook_events
        WHERE provider = 'stripe' AND provider_event_id = $1
        FOR UPDATE
        "#,
    )
    .bind(&event.id)
    .fetch_optional(&mut *tx)
    .await
    .map_err(|err| AppError::new(ErrorCode::Internal, "锁定 Webhook 失败").with_source(err))?;
    if status.as_deref() != Some("processing") {
        return Err(AppError::new(
            ErrorCode::IdempotencyConflict,
            "Webhook 事件未处于可处理状态",
        ));
    }

    process_stripe_event(state, &mut tx, event).await?;
    let updated = sqlx::query(
        r#"
        UPDATE webhook_events
        SET status = 'processed', processed_at = NOW(), error_message = NULL
        WHERE provider = 'stripe'
          AND provider_event_id = $1
          AND status = 'processing'
        "#,
    )
    .bind(&event.id)
    .execute(&mut *tx)
    .await
    .map_err(|err| AppError::new(ErrorCode::Internal, "更新 Webhook 状态失败").with_source(err))?;
    if updated.rows_affected() != 1 {
        return Err(AppError::new(
            ErrorCode::IdempotencyConflict,
            "Webhook 处理租约已失效",
        ));
    }
    tx.commit().await.map_err(|err| {
        AppError::new(ErrorCode::Internal, "提交 Webhook 事务失败").with_source(err)
    })?;
    Ok(())
}

fn verify_stripe_signature(payload: &[u8], sig_header: &str, secret: &str) -> Result<(), AppError> {
    let mut timestamp: Option<i64> = None;
    let mut signatures = Vec::<String>::new();

    for part in sig_header.split(',') {
        let part = part.trim();
        if let Some(v) = part.strip_prefix("t=") {
            timestamp = v.parse::<i64>().ok();
        } else if let Some(v) = part.strip_prefix("v1=") {
            signatures.push(v.to_string());
        }
    }

    let Some(ts) = timestamp else {
        return Err(AppError::new(
            ErrorCode::InvalidRequest,
            "Stripe-Signature 缺少 t",
        ));
    };
    if signatures.is_empty() {
        return Err(AppError::new(
            ErrorCode::InvalidRequest,
            "Stripe-Signature 缺少 v1",
        ));
    }

    // 5 minutes tolerance
    let now = Utc::now().timestamp();
    if (now - ts).abs() > 300 {
        return Err(AppError::new(
            ErrorCode::InvalidRequest,
            "Webhook 时间戳过期",
        ));
    }

    type HmacSha256 = Hmac<Sha256>;
    let mut mac = HmacSha256::new_from_slice(secret.as_bytes()).map_err(|err| {
        AppError::new(ErrorCode::Internal, "Webhook secret 错误").with_source(err)
    })?;
    mac.update(ts.to_string().as_bytes());
    mac.update(b".");
    mac.update(payload);
    let expected = hex::encode(mac.finalize().into_bytes());

    if signatures.iter().any(|sig| secure_eq(sig, &expected)) {
        Ok(())
    } else {
        Err(AppError::new(ErrorCode::InvalidRequest, "Webhook 验签失败"))
    }
}

fn secure_eq(a: &str, b: &str) -> bool {
    if a.len() != b.len() {
        return false;
    }
    let mut out = 0u8;
    for (x, y) in a.as_bytes().iter().zip(b.as_bytes().iter()) {
        out |= x ^ y;
    }
    out == 0
}

async fn process_stripe_event(
    state: &AppState,
    tx: &mut Transaction<'_, Postgres>,
    event: &StripeEvent,
) -> Result<(), AppError> {
    match event.type_.as_str() {
        "checkout.session.completed" => {
            map_checkout_session_completed(tx, &event.data.object).await
        }
        "customer.subscription.created" | "customer.subscription.updated" => {
            apply_subscription_event(state, tx, event, &event.data.object).await
        }
        "customer.subscription.deleted" => {
            apply_subscription_event(state, tx, event, &event.data.object).await
        }
        "invoice.paid" | "invoice.payment_failed" => upsert_invoice(tx, &event.data.object).await,
        _ => Ok(()),
    }
}

async fn map_checkout_session_completed(
    tx: &mut Transaction<'_, Postgres>,
    object: &serde_json::Value,
) -> Result<(), AppError> {
    let customer_id = object
        .get("customer")
        .and_then(|v| v.as_str())
        .filter(|v| !v.trim().is_empty())
        .ok_or_else(|| AppError::new(ErrorCode::InvalidRequest, "checkout.customer 缺失"))?;
    let session_id = object
        .get("id")
        .and_then(|value| value.as_str())
        .filter(|value| !value.trim().is_empty())
        .ok_or_else(|| AppError::new(ErrorCode::InvalidRequest, "checkout.id 缺失"))?;

    let user_id = object
        .get("client_reference_id")
        .and_then(|v| v.as_str())
        .and_then(|v| v.parse::<uuid::Uuid>().ok())
        .or_else(|| {
            object
                .pointer("/metadata/user_id")
                .and_then(|v| v.as_str())
                .and_then(|v| v.parse::<uuid::Uuid>().ok())
        })
        .ok_or_else(|| AppError::new(ErrorCode::InvalidRequest, "checkout.user_id 缺失"))?;
    let attempt_id = object
        .pointer("/metadata/checkout_attempt_id")
        .and_then(|value| value.as_str())
        .and_then(|value| value.parse::<uuid::Uuid>().ok());

    let current: Option<Option<String>> =
        sqlx::query_scalar("SELECT billing_customer_id FROM users WHERE id = $1 FOR UPDATE")
            .bind(user_id)
            .fetch_optional(&mut **tx)
            .await
            .map_err(|err| AppError::new(ErrorCode::Internal, "锁定用户失败").with_source(err))?;
    let current = current.ok_or_else(|| {
        AppError::new(
            ErrorCode::StorageUnavailable,
            "Checkout 对应用户不存在，事件将重试",
        )
    })?;
    if let Some(current) = current.filter(|value| !value.trim().is_empty()) {
        if current != customer_id {
            return Err(AppError::new(
                ErrorCode::StorageUnavailable,
                "Checkout Customer 与用户计费身份不一致，事件将重试",
            ));
        }
    } else {
        sqlx::query(
            "UPDATE users SET billing_customer_id = $2, updated_at = NOW() WHERE id = $1 AND (billing_customer_id IS NULL OR billing_customer_id = '')",
        )
        .bind(user_id)
        .bind(customer_id)
        .execute(&mut **tx)
        .await
        .map_err(|err| {
            AppError::new(ErrorCode::Internal, "保存 Stripe Customer 映射失败").with_source(err)
        })?;
    }

    sqlx::query(
        r#"
        UPDATE billing_checkout_sessions
        SET status = 'completed',
            stripe_customer_id = $3,
            stripe_session_id = $4,
            completed_at = NOW(),
            lease_owner = NULL,
            lease_until = NULL,
            error_message = NULL,
            updated_at = NOW()
        WHERE user_id = $1
          AND status = 'pending'
          AND (
            stripe_session_id = $4
            OR ($2::uuid IS NOT NULL AND id = $2)
          )
        "#,
    )
    .bind(user_id)
    .bind(attempt_id)
    .bind(customer_id)
    .bind(session_id)
    .execute(&mut **tx)
    .await
    .map_err(|err| AppError::new(ErrorCode::Internal, "完成 Checkout 状态失败").with_source(err))?;

    Ok(())
}

fn subscription_event_rank(event_type: &str) -> i16 {
    match event_type {
        "customer.subscription.deleted" => 2,
        "customer.subscription.updated" => 1,
        _ => 0,
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum WatermarkDecision {
    Apply,
    Ignore,
    Reconcile,
}

async fn decide_subscription_event(
    tx: &mut Transaction<'_, Postgres>,
    event: &StripeEvent,
    provider_subscription_id: &str,
    is_deleted: bool,
) -> Result<WatermarkDecision, AppError> {
    let rank = subscription_event_rank(&event.type_);
    let inserted: Option<String> = sqlx::query_scalar(
        r#"
        INSERT INTO provider_object_event_watermarks (
          provider, object_type, provider_object_id,
          last_event_created, last_event_rank, last_event_id, is_deleted
        ) VALUES (
          'stripe', 'subscription', $1,
          $2, $3, $4, $5
        )
        ON CONFLICT (provider, object_type, provider_object_id) DO NOTHING
        RETURNING provider_object_id
        "#,
    )
    .bind(provider_subscription_id)
    .bind(event.created)
    .bind(rank)
    .bind(&event.id)
    .bind(is_deleted)
    .fetch_optional(&mut **tx)
    .await
    .map_err(|err| AppError::new(ErrorCode::Internal, "创建 Stripe 水位失败").with_source(err))?;
    if inserted.is_some() {
        return Ok(WatermarkDecision::Apply);
    }

    let current: (i64, i16, String, bool) = sqlx::query_as(
        r#"
        SELECT last_event_created, last_event_rank, last_event_id, requires_reconciliation
        FROM provider_object_event_watermarks
        WHERE provider = 'stripe'
          AND object_type = 'subscription'
          AND provider_object_id = $1
        FOR UPDATE
        "#,
    )
    .bind(provider_subscription_id)
    .fetch_one(&mut **tx)
    .await
    .map_err(|err| AppError::new(ErrorCode::Internal, "锁定 Stripe 水位失败").with_source(err))?;
    if current.3 {
        return Ok(WatermarkDecision::Reconcile);
    }
    if current.1 == 100 && event.created == current.0 {
        return Ok(WatermarkDecision::Reconcile);
    }
    if event.created == current.0 && rank == current.1 {
        return if event.id == current.2 {
            Ok(WatermarkDecision::Ignore)
        } else {
            Ok(WatermarkDecision::Reconcile)
        };
    }
    if (event.created, rank) <= (current.0, current.1) {
        return Ok(WatermarkDecision::Ignore);
    }

    sqlx::query(
        r#"
        UPDATE provider_object_event_watermarks
        SET last_event_created = $2,
            last_event_rank = $3,
            last_event_id = $4,
            is_deleted = $5,
            requires_reconciliation = false,
            reconciliation_reason = NULL,
            updated_at = NOW()
        WHERE provider = 'stripe'
          AND object_type = 'subscription'
          AND provider_object_id = $1
        "#,
    )
    .bind(provider_subscription_id)
    .bind(event.created)
    .bind(rank)
    .bind(&event.id)
    .bind(is_deleted)
    .execute(&mut **tx)
    .await
    .map_err(|err| AppError::new(ErrorCode::Internal, "推进 Stripe 水位失败").with_source(err))?;
    Ok(WatermarkDecision::Apply)
}

#[derive(Debug)]
struct ResolvedSubscription {
    provider_subscription_id: String,
    provider_customer_id: String,
    user_id: uuid::Uuid,
    plan_id: uuid::Uuid,
    status: String,
    current_period_start: chrono::DateTime<Utc>,
    current_period_end: chrono::DateTime<Utc>,
    cancel_at_period_end: bool,
    canceled_at: Option<chrono::DateTime<Utc>>,
    checkout_attempt_id: Option<uuid::Uuid>,
}

struct StripeSubscriptionSnapshot {
    object: serde_json::Value,
    reconciled_through: i64,
}

async fn apply_subscription_event(
    state: &AppState,
    tx: &mut Transaction<'_, Postgres>,
    event: &StripeEvent,
    object: &serde_json::Value,
) -> Result<(), AppError> {
    let provider_subscription_id = object
        .get("id")
        .and_then(|value| value.as_str())
        .filter(|value| !value.trim().is_empty())
        .ok_or_else(|| AppError::new(ErrorCode::InvalidRequest, "subscription.id 缺失"))?;
    let is_deleted = event.type_ == "customer.subscription.deleted"
        || object.get("status").and_then(|value| value.as_str()) == Some("canceled");
    match decide_subscription_event(tx, event, provider_subscription_id, is_deleted).await? {
        WatermarkDecision::Ignore => Ok(()),
        WatermarkDecision::Apply => {
            let resolved = resolve_subscription(tx, object).await?;
            write_subscription(tx, &resolved).await
        }
        WatermarkDecision::Reconcile => {
            let snapshot =
                fetch_stripe_subscription_snapshot(state, provider_subscription_id).await?;
            let authoritative = resolve_subscription(tx, &snapshot.object).await?;
            if authoritative.provider_subscription_id != provider_subscription_id {
                return Err(AppError::new(
                    ErrorCode::Internal,
                    "Stripe 对账快照订阅 ID 不一致",
                ));
            }
            write_subscription(tx, &authoritative).await?;
            record_snapshot_watermark(
                tx,
                &authoritative,
                snapshot.reconciled_through.max(event.created),
                &event.id,
            )
            .await
        }
    }
}

async fn resolve_subscription(
    tx: &mut Transaction<'_, Postgres>,
    object: &serde_json::Value,
) -> Result<ResolvedSubscription, AppError> {
    let provider_subscription_id = object
        .get("id")
        .and_then(|value| value.as_str())
        .filter(|value| !value.trim().is_empty())
        .ok_or_else(|| AppError::new(ErrorCode::InvalidRequest, "subscription.id 缺失"))?;
    let provider_customer_id = object
        .get("customer")
        .and_then(|value| value.as_str())
        .filter(|value| !value.trim().is_empty())
        .ok_or_else(|| AppError::new(ErrorCode::InvalidRequest, "subscription.customer 缺失"))?;
    let price_id = object
        .pointer("/items/data/0/price/id")
        .and_then(|value| value.as_str())
        .or_else(|| {
            object
                .pointer("/items/data/0/plan/id")
                .and_then(|value| value.as_str())
        })
        .filter(|value| !value.trim().is_empty())
        .ok_or_else(|| AppError::new(ErrorCode::InvalidRequest, "subscription.price 缺失"))?;

    let user_id: Option<uuid::Uuid> =
        sqlx::query_scalar("SELECT id FROM users WHERE billing_customer_id = $1 FOR UPDATE")
            .bind(provider_customer_id)
            .fetch_optional(&mut **tx)
            .await
            .map_err(|err| {
                AppError::new(ErrorCode::Internal, "锁定订阅用户失败").with_source(err)
            })?;
    let user_id = user_id.ok_or_else(|| {
        AppError::new(
            ErrorCode::StorageUnavailable,
            "Stripe Customer 尚未映射到用户，订阅事件将重试",
        )
    })?;
    let plan_id: Option<uuid::Uuid> =
        sqlx::query_scalar("SELECT id FROM plans WHERE stripe_price_id = $1 LIMIT 1")
            .bind(price_id)
            .fetch_optional(&mut **tx)
            .await
            .map_err(|err| {
                AppError::new(ErrorCode::Internal, "查询订阅套餐失败").with_source(err)
            })?;
    let plan_id = plan_id.ok_or_else(|| {
        AppError::new(
            ErrorCode::StorageUnavailable,
            "Stripe Price 尚未映射到套餐，订阅事件将重试",
        )
    })?;

    let status = map_subscription_status(
        object
            .get("status")
            .and_then(|value| value.as_str())
            .unwrap_or("incomplete"),
    )
    .to_string();
    let now = Utc::now();
    let current_period_start = object
        .get("current_period_start")
        .and_then(|value| value.as_i64())
        .and_then(|value| Utc.timestamp_opt(value, 0).single())
        .unwrap_or(now);
    let current_period_end = object
        .get("current_period_end")
        .and_then(|value| value.as_i64())
        .and_then(|value| Utc.timestamp_opt(value, 0).single())
        .unwrap_or(now);
    let canceled_at = if status == "canceled" {
        object
            .get("canceled_at")
            .and_then(|value| value.as_i64())
            .and_then(|value| Utc.timestamp_opt(value, 0).single())
            .or(Some(now))
    } else {
        None
    };
    let checkout_attempt_id = object
        .pointer("/metadata/checkout_attempt_id")
        .and_then(|value| value.as_str())
        .and_then(|value| value.parse::<uuid::Uuid>().ok());

    Ok(ResolvedSubscription {
        provider_subscription_id: provider_subscription_id.to_string(),
        provider_customer_id: provider_customer_id.to_string(),
        user_id,
        plan_id,
        status,
        current_period_start,
        current_period_end,
        cancel_at_period_end: object
            .get("cancel_at_period_end")
            .and_then(|value| value.as_bool())
            .unwrap_or(false),
        canceled_at,
        checkout_attempt_id,
    })
}

async fn write_subscription(
    tx: &mut Transaction<'_, Postgres>,
    subscription: &ResolvedSubscription,
) -> Result<(), AppError> {
    if subscription.status != "canceled" {
        let conflicting: Option<String> = sqlx::query_scalar(
            r#"
            SELECT provider_subscription_id
            FROM subscriptions
            WHERE user_id = $1
              AND provider = 'stripe'
              AND status <> 'canceled'
              AND provider_subscription_id <> $2
            FOR UPDATE
            "#,
        )
        .bind(subscription.user_id)
        .bind(&subscription.provider_subscription_id)
        .fetch_optional(&mut **tx)
        .await
        .map_err(|err| {
            AppError::new(ErrorCode::Internal, "检查重复 Stripe 订阅失败").with_source(err)
        })?;
        if conflicting.is_some() {
            return Err(AppError::new(
                ErrorCode::StorageUnavailable,
                "用户已有其他未取消 Stripe 订阅，事件等待人工对账",
            ));
        }
    }

    sqlx::query(
        r#"
        INSERT INTO subscriptions (
          user_id, plan_id, status,
          current_period_start, current_period_end,
          cancel_at_period_end, canceled_at,
          provider, provider_customer_id, provider_subscription_id
        ) VALUES (
          $1, $2, $3::subscription_status,
          $4, $5, $6, $7,
          'stripe', $8, $9
        )
        ON CONFLICT (provider, provider_subscription_id) DO UPDATE
        SET user_id = EXCLUDED.user_id,
            plan_id = EXCLUDED.plan_id,
            status = EXCLUDED.status,
            current_period_start = EXCLUDED.current_period_start,
            current_period_end = EXCLUDED.current_period_end,
            cancel_at_period_end = EXCLUDED.cancel_at_period_end,
            canceled_at = EXCLUDED.canceled_at,
            provider_customer_id = EXCLUDED.provider_customer_id,
            updated_at = NOW()
        "#,
    )
    .bind(subscription.user_id)
    .bind(subscription.plan_id)
    .bind(&subscription.status)
    .bind(subscription.current_period_start)
    .bind(subscription.current_period_end)
    .bind(subscription.cancel_at_period_end)
    .bind(subscription.canceled_at)
    .bind(&subscription.provider_customer_id)
    .bind(&subscription.provider_subscription_id)
    .execute(&mut **tx)
    .await
    .map_err(|err| AppError::new(ErrorCode::Internal, "写入订阅失败").with_source(err))?;

    if subscription.status != "canceled" {
        sqlx::query(
            r#"
            UPDATE billing_checkout_sessions
            SET status = 'completed',
                completed_at = NOW(),
                lease_owner = NULL,
                lease_until = NULL,
                error_message = NULL,
                updated_at = NOW()
            WHERE user_id = $1
              AND status = 'pending'
              AND stripe_customer_id = $2
              AND ($3::uuid IS NULL OR id = $3)
            "#,
        )
        .bind(subscription.user_id)
        .bind(&subscription.provider_customer_id)
        .bind(subscription.checkout_attempt_id)
        .execute(&mut **tx)
        .await
        .map_err(|err| {
            AppError::new(ErrorCode::Internal, "完成订阅 Checkout 状态失败").with_source(err)
        })?;
    }
    Ok(())
}

async fn fetch_stripe_subscription_snapshot(
    state: &AppState,
    provider_subscription_id: &str,
) -> Result<StripeSubscriptionSnapshot, AppError> {
    let secret = settings::get_stripe_secret(state)
        .await
        .map_err(|err| err.with_source("stripe secret not configured"))?;
    let encoded_id = utf8_percent_encode(provider_subscription_id, NON_ALPHANUMERIC);
    let url = format!(
        "{}/v1/subscriptions/{encoded_id}",
        state.config.stripe_api_base_url
    );
    let response = reqwest::Client::new()
        .get(url)
        .bearer_auth(secret)
        .timeout(std::time::Duration::from_secs(15))
        .send()
        .await
        .map_err(|err| {
            AppError::new(ErrorCode::Internal, "Stripe 对账请求失败").with_source(err)
        })?;
    let status = response.status();
    let reconciled_through = response
        .headers()
        .get(reqwest::header::DATE)
        .and_then(|value| value.to_str().ok())
        .and_then(|value| chrono::DateTime::parse_from_rfc2822(value).ok())
        .map(|value| value.timestamp())
        .unwrap_or_else(|| Utc::now().timestamp());
    let body = response.text().await.map_err(|err| {
        AppError::new(ErrorCode::Internal, "读取 Stripe 对账响应失败").with_source(err)
    })?;
    if !status.is_success() {
        tracing::error!(%status, %body, provider_subscription_id, "Stripe subscription reconciliation failed");
        return Err(AppError::new(
            ErrorCode::StorageUnavailable,
            "Stripe 订阅对账失败，将自动重试",
        ));
    }
    let object = serde_json::from_str(&body).map_err(|err| {
        AppError::new(ErrorCode::Internal, "解析 Stripe 对账响应失败").with_source(err)
    })?;
    Ok(StripeSubscriptionSnapshot {
        object,
        reconciled_through,
    })
}

async fn record_snapshot_watermark(
    tx: &mut Transaction<'_, Postgres>,
    subscription: &ResolvedSubscription,
    reconciled_through: i64,
    trigger_id: &str,
) -> Result<(), AppError> {
    sqlx::query(
        r#"
        INSERT INTO provider_object_event_watermarks (
          provider, object_type, provider_object_id,
          last_event_created, last_event_rank, last_event_id,
          is_deleted, requires_reconciliation, reconciliation_reason,
          last_snapshot_at, updated_at
        ) VALUES (
          'stripe', 'subscription', $1,
          $2, 100, $3,
          $4, false, NULL,
          NOW(), NOW()
        )
        ON CONFLICT (provider, object_type, provider_object_id) DO UPDATE
        SET last_event_created = EXCLUDED.last_event_created,
            last_event_rank = 100,
            last_event_id = EXCLUDED.last_event_id,
            is_deleted = EXCLUDED.is_deleted,
            requires_reconciliation = false,
            reconciliation_reason = NULL,
            last_snapshot_at = NOW(),
            updated_at = NOW()
        "#,
    )
    .bind(&subscription.provider_subscription_id)
    .bind(reconciled_through)
    .bind(format!("snapshot:{trigger_id}"))
    .bind(subscription.status == "canceled")
    .execute(&mut **tx)
    .await
    .map_err(|err| {
        AppError::new(ErrorCode::Internal, "保存 Stripe 快照水位失败").with_source(err)
    })?;
    sqlx::query(
        r#"
        UPDATE stripe_subscription_reconciliations
        SET status = 'completed', completed_at = NOW(), updated_at = NOW(), last_error = NULL,
            lease_owner = NULL, lease_until = NULL
        WHERE provider_subscription_id = $1
        "#,
    )
    .bind(&subscription.provider_subscription_id)
    .execute(&mut **tx)
    .await
    .map_err(|err| {
        AppError::new(ErrorCode::Internal, "完成 Stripe 对账队列失败").with_source(err)
    })?;
    Ok(())
}

pub(crate) async fn reconciliation_loop(state: AppState) {
    loop {
        match reconcile_next_subscription(&state).await {
            Ok(true) => continue,
            Ok(false) => tokio::time::sleep(std::time::Duration::from_secs(60)).await,
            Err(err) => {
                tracing::error!(error = %err, "Stripe subscription reconciliation iteration failed");
                tokio::time::sleep(std::time::Duration::from_secs(10)).await;
            }
        }
    }
}

async fn reconcile_next_subscription(state: &AppState) -> Result<bool, AppError> {
    let lease_owner = uuid::Uuid::new_v4();
    let claim: Option<(String, i32)> = sqlx::query_as(
        r#"
        WITH candidate AS (
          SELECT provider_subscription_id
          FROM stripe_subscription_reconciliations
          WHERE (
              status IN ('pending', 'failed') AND next_attempt_at <= NOW()
            ) OR (
              status = 'processing' AND COALESCE(lease_until, updated_at) <= NOW()
            )
          ORDER BY next_attempt_at ASC, updated_at ASC
          FOR UPDATE SKIP LOCKED
          LIMIT 1
        )
        UPDATE stripe_subscription_reconciliations AS reconciliation
        SET status = 'processing',
            attempts = reconciliation.attempts + 1,
            lease_owner = $1,
            lease_until = NOW() + INTERVAL '1 minute',
            updated_at = NOW()
        FROM candidate
        WHERE reconciliation.provider_subscription_id = candidate.provider_subscription_id
        RETURNING reconciliation.provider_subscription_id, reconciliation.attempts
        "#,
    )
    .bind(lease_owner)
    .fetch_optional(&state.db)
    .await
    .map_err(|err| {
        AppError::new(ErrorCode::Internal, "领取 Stripe 对账任务失败").with_source(err)
    })?;
    let Some((provider_subscription_id, attempts)) = claim else {
        return Ok(false);
    };

    if let Err(err) =
        reconcile_claimed_subscription(state, &provider_subscription_id, lease_owner).await
    {
        let delay_seconds = (5_i64 * 2_i64.pow(attempts.clamp(0, 6) as u32)).min(300);
        let update_result = sqlx::query(
            r#"
            UPDATE stripe_subscription_reconciliations
            SET status = 'failed',
                next_attempt_at = NOW() + make_interval(secs => $3),
                last_error = $4,
                lease_owner = NULL,
                lease_until = NULL,
                updated_at = NOW()
            WHERE provider_subscription_id = $1
              AND status = 'processing'
              AND lease_owner = $2
            "#,
        )
        .bind(&provider_subscription_id)
        .bind(lease_owner)
        .bind(delay_seconds as f64)
        .bind(truncate(err.to_string(), 2_000))
        .execute(&state.db)
        .await;
        if let Err(update_err) = update_result {
            tracing::error!(error = %update_err, provider_subscription_id, "failed to persist Stripe reconciliation error");
        }
        return Err(err);
    }
    Ok(true)
}

async fn reconcile_claimed_subscription(
    state: &AppState,
    provider_subscription_id: &str,
    lease_owner: uuid::Uuid,
) -> Result<(), AppError> {
    let snapshot = fetch_stripe_subscription_snapshot(state, provider_subscription_id).await?;
    let mut tx = state.db.begin().await.map_err(|err| {
        AppError::new(ErrorCode::Internal, "开启 Stripe 对账事务失败").with_source(err)
    })?;
    let authoritative = resolve_subscription(&mut tx, &snapshot.object).await?;
    if authoritative.provider_subscription_id != provider_subscription_id {
        return Err(AppError::new(
            ErrorCode::Internal,
            "Stripe 对账快照订阅 ID 不一致",
        ));
    }

    sqlx::query(
        r#"
        SELECT provider_object_id
        FROM provider_object_event_watermarks
        WHERE provider = 'stripe'
          AND object_type = 'subscription'
          AND provider_object_id = $1
        FOR UPDATE
        "#,
    )
    .bind(provider_subscription_id)
    .fetch_optional(&mut *tx)
    .await
    .map_err(|err| {
        AppError::new(ErrorCode::Internal, "锁定 Stripe 对账水位失败").with_source(err)
    })?;

    let owned: Option<uuid::Uuid> = sqlx::query_scalar(
        r#"
        SELECT lease_owner
        FROM stripe_subscription_reconciliations
        WHERE provider_subscription_id = $1
          AND status = 'processing'
          AND lease_owner = $2
          AND lease_until > NOW()
        FOR UPDATE
        "#,
    )
    .bind(provider_subscription_id)
    .bind(lease_owner)
    .fetch_optional(&mut *tx)
    .await
    .map_err(|err| {
        AppError::new(ErrorCode::Internal, "校验 Stripe 对账租约失败").with_source(err)
    })?;
    if owned.is_none() {
        return Err(AppError::new(
            ErrorCode::IdempotencyConflict,
            "Stripe 对账租约已失效",
        ));
    }

    write_subscription(&mut tx, &authoritative).await?;
    record_snapshot_watermark(
        &mut tx,
        &authoritative,
        snapshot.reconciled_through,
        "queue",
    )
    .await?;
    tx.commit().await.map_err(|err| {
        AppError::new(ErrorCode::Internal, "提交 Stripe 对账失败").with_source(err)
    })?;
    Ok(())
}

async fn upsert_invoice(
    tx: &mut Transaction<'_, Postgres>,
    object: &serde_json::Value,
) -> Result<(), AppError> {
    let provider_invoice_id = object
        .get("id")
        .and_then(|v| v.as_str())
        .ok_or_else(|| AppError::new(ErrorCode::InvalidRequest, "invoice.id 缺失"))?;
    let provider_customer_id = object
        .get("customer")
        .and_then(|v| v.as_str())
        .ok_or_else(|| AppError::new(ErrorCode::InvalidRequest, "invoice.customer 缺失"))?;

    let user_id: Option<uuid::Uuid> =
        sqlx::query_scalar("SELECT id FROM users WHERE billing_customer_id = $1 LIMIT 1")
            .bind(provider_customer_id)
            .fetch_optional(&mut **tx)
            .await
            .map_err(|err| AppError::new(ErrorCode::Internal, "查询用户失败").with_source(err))?;

    let Some(user_id) = user_id else {
        return Ok(());
    };

    let stripe_status = object
        .get("status")
        .and_then(|v| v.as_str())
        .unwrap_or("open");
    let status = map_invoice_status(stripe_status);

    let invoice_number = object
        .get("number")
        .and_then(|v| v.as_str())
        .filter(|v| !v.trim().is_empty())
        .map(|v| v.to_string())
        .unwrap_or_else(|| format!("stripe_{provider_invoice_id}"));

    let currency = object
        .get("currency")
        .and_then(|v| v.as_str())
        .unwrap_or("cny")
        .to_uppercase();
    let total_amount_cents = object.get("total").and_then(|v| v.as_i64()).unwrap_or(0) as i32;

    let hosted_invoice_url = object
        .get("hosted_invoice_url")
        .and_then(|v| v.as_str())
        .map(|v| v.to_string());
    let pdf_url = object
        .get("invoice_pdf")
        .and_then(|v| v.as_str())
        .map(|v| v.to_string());

    let period_start = object
        .get("period_start")
        .and_then(|v| v.as_i64())
        .and_then(|ts| Utc.timestamp_opt(ts, 0).single());
    let period_end = object
        .get("period_end")
        .and_then(|v| v.as_i64())
        .and_then(|ts| Utc.timestamp_opt(ts, 0).single());

    let paid_at = object
        .pointer("/status_transitions/paid_at")
        .and_then(|v| v.as_i64())
        .and_then(|ts| Utc.timestamp_opt(ts, 0).single());

    let updated = sqlx::query(
        r#"
        UPDATE invoices
        SET status = $1::invoice_status,
            currency = $2,
            total_amount_cents = $3,
            hosted_invoice_url = $4,
            pdf_url = $5,
            period_start = $6,
            period_end = $7,
            paid_at = $8
        WHERE provider = 'stripe' AND provider_invoice_id = $9
        "#,
    )
    .bind(status)
    .bind(&currency)
    .bind(total_amount_cents)
    .bind(hosted_invoice_url.as_deref())
    .bind(pdf_url.as_deref())
    .bind(period_start)
    .bind(period_end)
    .bind(paid_at)
    .bind(provider_invoice_id)
    .execute(&mut **tx)
    .await
    .map_err(|err| AppError::new(ErrorCode::Internal, "更新发票失败").with_source(err))?;

    if updated.rows_affected() == 0 {
        let invoice_number = truncate(invoice_number, 50);
        let _ = sqlx::query(
            r#"
            INSERT INTO invoices (
              user_id, invoice_number, status, currency, total_amount_cents,
              period_start, period_end,
              provider, provider_invoice_id, hosted_invoice_url, pdf_url,
              paid_at
            ) VALUES (
              $1, $2, $3::invoice_status, $4, $5,
              $6, $7,
              'stripe', $8, $9, $10,
              $11
            )
            "#,
        )
        .bind(user_id)
        .bind(invoice_number)
        .bind(status)
        .bind(&currency)
        .bind(total_amount_cents)
        .bind(period_start)
        .bind(period_end)
        .bind(provider_invoice_id)
        .bind(hosted_invoice_url.as_deref())
        .bind(pdf_url.as_deref())
        .bind(paid_at)
        .execute(&mut **tx)
        .await
        .map_err(|err| AppError::new(ErrorCode::Internal, "创建发票失败").with_source(err))?;
    }

    Ok(())
}

fn truncate(mut s: String, max: usize) -> String {
    if s.len() > max {
        let mut end = max;
        while !s.is_char_boundary(end) {
            end -= 1;
        }
        s.truncate(end);
    }
    s
}

fn map_subscription_status(status: &str) -> &'static str {
    match status {
        "trialing" => "trialing",
        "active" => "active",
        "past_due" => "past_due",
        "canceled" => "canceled",
        _ => "incomplete",
    }
}

fn map_invoice_status(status: &str) -> &'static str {
    match status {
        "draft" => "draft",
        "paid" => "paid",
        "void" => "void",
        "uncollectible" => "uncollectible",
        _ => "open",
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::config::Config;
    use crate::services::mail::Mailer;
    use axum::extract::Path;
    use axum::http::{header, HeaderMap, HeaderValue, StatusCode};
    use axum::response::{IntoResponse, Response};
    use sqlx::postgres::PgPoolOptions;
    use std::collections::HashMap;
    use std::sync::atomic::{AtomicUsize, Ordering};
    use std::sync::Arc;
    use tokio::sync::{Barrier, RwLock, Semaphore};
    use uuid::Uuid;

    #[derive(Clone, Default)]
    struct StripeSnapshotMock {
        objects: Arc<RwLock<HashMap<String, serde_json::Value>>>,
        calls: Arc<AtomicUsize>,
    }

    async fn stripe_snapshot(
        State(mock): State<StripeSnapshotMock>,
        Path(subscription_id): Path<String>,
    ) -> Response {
        mock.calls.fetch_add(1, Ordering::SeqCst);
        let object = mock.objects.read().await.get(&subscription_id).cloned();
        let mut headers = HeaderMap::new();
        headers.insert(
            header::DATE,
            HeaderValue::from_static("Sun, 26 Jul 2026 00:00:00 GMT"),
        );
        match object {
            Some(object) => (StatusCode::OK, headers, Json(object)).into_response(),
            None => (
                StatusCode::NOT_FOUND,
                headers,
                Json(serde_json::json!({ "error": "subscription not found" })),
            )
                .into_response(),
        }
    }

    async fn spawn_stripe_snapshot_mock(
    ) -> (String, StripeSnapshotMock, tokio::task::JoinHandle<()>) {
        let mock = StripeSnapshotMock::default();
        let app = Router::new()
            .route(
                "/v1/subscriptions/{id}",
                axum::routing::get(stripe_snapshot),
            )
            .with_state(mock.clone());
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
            .await
            .expect("bind Stripe snapshot mock");
        let address = listener.local_addr().expect("read Stripe mock address");
        let task = tokio::spawn(async move {
            axum::serve(listener, app)
                .await
                .expect("serve Stripe snapshot mock");
        });
        (format!("http://{address}"), mock, task)
    }

    async fn build_test_state(
        pool: sqlx::PgPool,
        database_url: String,
        redis_url: String,
        stripe_api_base_url: String,
    ) -> AppState {
        let mut config = Config::from_env().expect("load test config");
        config.database_url = database_url;
        config.redis_url = redis_url;
        config.stripe_secret_key = Some("sk_test_webhook_ordering".to_string());
        config.stripe_api_base_url = stripe_api_base_url;
        config.mail_enabled = false;
        config.mail_log_links_when_disabled = false;
        let redis = redis::Client::open(config.redis_url.clone())
            .expect("create test redis client")
            .get_connection_manager()
            .await
            .expect("connect test redis");
        AppState {
            mailer: Arc::new(Mailer::new(&config).expect("create disabled test mailer")),
            image_processing_semaphore: Arc::new(Semaphore::new(2)),
            runtime_policy_cache: crate::services::settings::RuntimePolicyCache::new(),
            storage_cache: crate::services::storage::StorageCache::new(),
            config,
            db: pool,
            redis,
        }
    }

    #[test]
    fn truncate_preserves_utf8_boundaries() {
        assert_eq!(truncate("中文测试".to_string(), 5), "中");
        assert_eq!(truncate("abc中文".to_string(), 5), "abc");
        assert_eq!(truncate("short".to_string(), 20), "short");
    }

    fn subscription_event(
        event_id: &str,
        event_type: &str,
        created: i64,
        subscription_id: &str,
        customer_id: &str,
        price_id: &str,
    ) -> StripeEvent {
        let status = if event_type == "customer.subscription.deleted" {
            "canceled"
        } else {
            "active"
        };
        StripeEvent {
            id: event_id.to_string(),
            created,
            type_: event_type.to_string(),
            data: StripeEventData {
                object: serde_json::json!({
                    "id": subscription_id,
                    "customer": customer_id,
                    "status": status,
                    "current_period_start": 1_700_000_000_i64,
                    "current_period_end": 1_702_592_000_i64,
                    "cancel_at_period_end": false,
                    "canceled_at": if status == "canceled" { Some(created) } else { None },
                    "items": { "data": [{ "price": { "id": price_id } }] }
                }),
            },
        }
    }

    async fn apply_test_event(state: &AppState, event: &StripeEvent) -> Result<(), AppError> {
        let mut tx = state.db.begin().await.expect("begin event transaction");
        match process_stripe_event(state, &mut tx, event).await {
            Ok(()) => {
                tx.commit().await.expect("commit stripe event");
                Ok(())
            }
            Err(err) => {
                tx.rollback().await.expect("rollback stripe event");
                Err(err)
            }
        }
    }

    async fn assert_canceled_once(pool: &sqlx::PgPool, subscription_id: &str) {
        let rows: Vec<(String,)> = sqlx::query_as(
            "SELECT status::text FROM subscriptions WHERE provider = 'stripe' AND provider_subscription_id = $1",
        )
        .bind(subscription_id)
        .fetch_all(pool)
        .await
        .expect("query subscription");
        assert_eq!(rows, vec![("canceled".to_string(),)]);

        let watermark: (bool,) = sqlx::query_as(
            "SELECT is_deleted FROM provider_object_event_watermarks WHERE provider = 'stripe' AND object_type = 'subscription' AND provider_object_id = $1",
        )
        .bind(subscription_id)
        .fetch_one(pool)
        .await
        .expect("query subscription watermark");
        assert!(watermark.0);
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
    #[ignore = "requires isolated IMAGEFORGE_TEST_DATABASE_URL and IMAGEFORGE_TEST_REDIS_URL"]
    async fn subscription_events_are_monotonic_for_all_orders_and_concurrency() {
        let database_url = std::env::var("IMAGEFORGE_TEST_DATABASE_URL")
            .expect("IMAGEFORGE_TEST_DATABASE_URL must be set");
        assert!(
            database_url.to_ascii_lowercase().contains("test"),
            "refusing to run destructive integration test outside a test database"
        );
        let redis_url = std::env::var("IMAGEFORGE_TEST_REDIS_URL")
            .expect("IMAGEFORGE_TEST_REDIS_URL must be set");
        let pool = PgPoolOptions::new()
            .max_connections(16)
            .connect(&database_url)
            .await
            .expect("connect test database");
        sqlx::migrate!().run(&pool).await.expect("run migrations");
        let (stripe_api_base_url, stripe_mock, stripe_mock_task) =
            spawn_stripe_snapshot_mock().await;
        let state =
            build_test_state(pool.clone(), database_url, redis_url, stripe_api_base_url).await;

        let marker = Uuid::new_v4().simple().to_string();
        let customer_id = format!("cus_test_{marker}");
        let price_id = format!("price_test_{marker}");
        let user_id = Uuid::new_v4();
        let plan_id = Uuid::new_v4();
        sqlx::query(
            r#"
            INSERT INTO users (id, email, username, password_hash, billing_customer_id)
            VALUES ($1, $2, $3, 'test-only', $4)
            "#,
        )
        .bind(user_id)
        .bind(format!("stripe-{marker}@example.test"))
        .bind(format!("stripe_{marker}"))
        .bind(&customer_id)
        .execute(&pool)
        .await
        .expect("insert test user");
        sqlx::query(
            r#"
            INSERT INTO plans (
              id, code, name, stripe_price_id,
              included_units_per_period, max_file_size_mb,
              max_files_per_batch, concurrency_limit, retention_days
            ) VALUES ($1, $2, 'Stripe ordering test', $3, 10, 10, 10, 1, 1)
            "#,
        )
        .bind(plan_id)
        .bind(format!("stripe_test_{marker}"))
        .bind(&price_id)
        .execute(&pool)
        .await
        .expect("insert test plan");

        let permutations = [
            [0, 1, 2],
            [0, 2, 1],
            [1, 0, 2],
            [1, 2, 0],
            [2, 0, 1],
            [2, 1, 0],
        ];
        for (case, order) in permutations.into_iter().enumerate() {
            let subscription_id = format!("sub_{marker}_perm_{case}");
            let events = [
                subscription_event(
                    &format!("evt_{marker}_{case}_created"),
                    "customer.subscription.created",
                    1_700_000_100,
                    &subscription_id,
                    &customer_id,
                    &price_id,
                ),
                subscription_event(
                    &format!("evt_{marker}_{case}_updated"),
                    "customer.subscription.updated",
                    1_700_000_200,
                    &subscription_id,
                    &customer_id,
                    &price_id,
                ),
                subscription_event(
                    &format!("evt_{marker}_{case}_deleted"),
                    "customer.subscription.deleted",
                    1_700_000_300,
                    &subscription_id,
                    &customer_id,
                    &price_id,
                ),
            ];
            for index in order {
                apply_test_event(&state, &events[index])
                    .await
                    .expect("apply permuted Stripe event");
            }
            assert_canceled_once(&pool, &subscription_id).await;
        }

        let same_second_id = format!("sub_{marker}_same_second");
        for event in [
            subscription_event(
                &format!("evt_{marker}_same_deleted"),
                "customer.subscription.deleted",
                1_700_000_400,
                &same_second_id,
                &customer_id,
                &price_id,
            ),
            subscription_event(
                &format!("evt_{marker}_same_updated"),
                "customer.subscription.updated",
                1_700_000_400,
                &same_second_id,
                &customer_id,
                &price_id,
            ),
            subscription_event(
                &format!("evt_{marker}_same_created"),
                "customer.subscription.created",
                1_700_000_400,
                &same_second_id,
                &customer_id,
                &price_id,
            ),
        ] {
            apply_test_event(&state, &event)
                .await
                .expect("apply same-second ranked event");
        }
        assert_canceled_once(&pool, &same_second_id).await;

        for case in 0..12 {
            let subscription_id = format!("sub_{marker}_concurrent_{case}");
            let events = [
                subscription_event(
                    &format!("evt_{marker}_concurrent_{case}_created"),
                    "customer.subscription.created",
                    1_700_001_100,
                    &subscription_id,
                    &customer_id,
                    &price_id,
                ),
                subscription_event(
                    &format!("evt_{marker}_concurrent_{case}_updated"),
                    "customer.subscription.updated",
                    1_700_001_200,
                    &subscription_id,
                    &customer_id,
                    &price_id,
                ),
                subscription_event(
                    &format!("evt_{marker}_concurrent_{case}_deleted"),
                    "customer.subscription.deleted",
                    1_700_001_300,
                    &subscription_id,
                    &customer_id,
                    &price_id,
                ),
            ];
            let barrier = Arc::new(Barrier::new(events.len()));
            let mut joins = Vec::new();
            for event in events {
                let state = state.clone();
                let barrier = barrier.clone();
                joins.push(tokio::spawn(async move {
                    barrier.wait().await;
                    apply_test_event(&state, &event).await
                }));
            }
            for join in joins {
                join.await
                    .expect("concurrent event task")
                    .expect("apply concurrent event");
            }
            assert_canceled_once(&pool, &subscription_id).await;
        }

        let migration_subscription_id = format!("sub_{marker}_migration_100_200_150");
        let migration_created = subscription_event(
            &format!("evt_{marker}_migration_created"),
            "customer.subscription.created",
            100,
            &migration_subscription_id,
            &customer_id,
            &price_id,
        );
        apply_test_event(&state, &migration_created)
            .await
            .expect("apply pre-migration event at 100");
        sqlx::query(
            r#"
            UPDATE provider_object_event_watermarks
            SET last_event_created = 200,
                last_event_rank = 1,
                last_event_id = 'migration:test-local-clock',
                is_deleted = false,
                requires_reconciliation = false
            WHERE provider = 'stripe'
              AND object_type = 'subscription'
              AND provider_object_id = $1
            "#,
        )
        .bind(&migration_subscription_id)
        .execute(&pool)
        .await
        .expect("emulate migration 016 local timestamp 200");
        sqlx::query(
            r#"
            WITH corrected AS (
              UPDATE provider_object_event_watermarks
              SET last_event_created = 0,
                  last_event_rank = 0,
                  last_event_id = 'reconcile:migration',
                  requires_reconciliation = true,
                  reconciliation_reason = 'migration_016_non_causal_seed',
                  updated_at = NOW()
              WHERE provider = 'stripe'
                AND object_type = 'subscription'
                AND provider_object_id = $1
                AND is_deleted = false
                AND last_event_id LIKE 'migration:%'
              RETURNING provider_object_id
            )
            INSERT INTO stripe_subscription_reconciliations (
              provider_subscription_id, reason, status, next_attempt_at
            )
            SELECT provider_object_id, 'migration_016_non_causal_seed', 'pending', NOW()
            FROM corrected
            "#,
        )
        .bind(&migration_subscription_id)
        .execute(&pool)
        .await
        .expect("apply migration 018 correction");
        let migration_deleted = subscription_event(
            &format!("evt_{marker}_migration_deleted"),
            "customer.subscription.deleted",
            150,
            &migration_subscription_id,
            &customer_id,
            &price_id,
        );
        stripe_mock.objects.write().await.insert(
            migration_subscription_id.clone(),
            migration_deleted.data.object.clone(),
        );
        assert!(
            reconcile_next_subscription(&state)
                .await
                .expect("reconcile corrected migration watermark"),
            "migration reconciliation queue was not consumed"
        );
        apply_test_event(&state, &migration_deleted)
            .await
            .expect("apply delayed terminal event at 150");
        assert_canceled_once(&pool, &migration_subscription_id).await;
        let migration_reconciliation_status: String = sqlx::query_scalar(
            "SELECT status FROM stripe_subscription_reconciliations WHERE provider_subscription_id = $1",
        )
        .bind(&migration_subscription_id)
        .fetch_one(&pool)
        .await
        .expect("query migration reconciliation status");
        assert_eq!(migration_reconciliation_status, "completed");

        for case in 0..2 {
            let subscription_id = format!("sub_{marker}_ambiguous_order_{case}");
            let event_a = subscription_event(
                &format!("evt_{marker}_a_{case}"),
                "customer.subscription.updated",
                1_700_002_000,
                &subscription_id,
                &customer_id,
                &price_id,
            );
            let event_z = subscription_event(
                &format!("evt_{marker}_z_{case}"),
                "customer.subscription.updated",
                1_700_002_000,
                &subscription_id,
                &customer_id,
                &price_id,
            );
            let authoritative = subscription_event(
                &format!("evt_{marker}_snapshot_{case}"),
                "customer.subscription.deleted",
                1_700_002_100,
                &subscription_id,
                &customer_id,
                &price_id,
            );
            stripe_mock
                .objects
                .write()
                .await
                .insert(subscription_id.clone(), authoritative.data.object);
            let ordered = if case == 0 {
                [&event_a, &event_z]
            } else {
                [&event_z, &event_a]
            };
            for event in ordered {
                apply_test_event(&state, event)
                    .await
                    .expect("apply same-rank ambiguous event");
            }
            assert_canceled_once(&pool, &subscription_id).await;
        }

        let concurrent_ambiguous_id = format!("sub_{marker}_ambiguous_concurrent");
        let concurrent_events = [
            subscription_event(
                &format!("evt_{marker}_ambiguous_a"),
                "customer.subscription.updated",
                1_700_003_000,
                &concurrent_ambiguous_id,
                &customer_id,
                &price_id,
            ),
            subscription_event(
                &format!("evt_{marker}_ambiguous_z"),
                "customer.subscription.updated",
                1_700_003_000,
                &concurrent_ambiguous_id,
                &customer_id,
                &price_id,
            ),
        ];
        let authoritative = subscription_event(
            &format!("evt_{marker}_ambiguous_snapshot"),
            "customer.subscription.deleted",
            1_700_003_100,
            &concurrent_ambiguous_id,
            &customer_id,
            &price_id,
        );
        stripe_mock
            .objects
            .write()
            .await
            .insert(concurrent_ambiguous_id.clone(), authoritative.data.object);
        let barrier = Arc::new(Barrier::new(concurrent_events.len()));
        let mut joins = Vec::new();
        for event in concurrent_events {
            let state = state.clone();
            let barrier = barrier.clone();
            joins.push(tokio::spawn(async move {
                barrier.wait().await;
                apply_test_event(&state, &event).await
            }));
        }
        for join in joins {
            join.await
                .expect("join ambiguous concurrent event")
                .expect("apply ambiguous concurrent event");
        }
        assert_canceled_once(&pool, &concurrent_ambiguous_id).await;

        let unmapped_subscription_id = format!("sub_{marker}_unmapped");
        let unmapped = subscription_event(
            &format!("evt_{marker}_unmapped"),
            "customer.subscription.created",
            1_700_004_000,
            &unmapped_subscription_id,
            &format!("cus_{marker}_unmapped"),
            &price_id,
        );
        let unmapped_error = apply_test_event(&state, &unmapped)
            .await
            .expect_err("unmapped managed subscription was marked processed");
        assert_eq!(unmapped_error.code, ErrorCode::StorageUnavailable);
        let unmapped_watermarks: i64 = sqlx::query_scalar(
            "SELECT COUNT(*) FROM provider_object_event_watermarks WHERE provider_object_id = $1",
        )
        .bind(&unmapped_subscription_id)
        .fetch_one(&pool)
        .await
        .expect("count unmapped subscription watermarks");
        assert_eq!(unmapped_watermarks, 0);

        let primary_subscription_id = format!("sub_{marker}_single_primary");
        let secondary_subscription_id = format!("sub_{marker}_single_secondary");
        let primary = subscription_event(
            &format!("evt_{marker}_single_primary"),
            "customer.subscription.created",
            1_700_005_000,
            &primary_subscription_id,
            &customer_id,
            &price_id,
        );
        apply_test_event(&state, &primary)
            .await
            .expect("apply primary subscription");
        let secondary = subscription_event(
            &format!("evt_{marker}_single_secondary"),
            "customer.subscription.created",
            1_700_005_100,
            &secondary_subscription_id,
            &customer_id,
            &price_id,
        );
        let duplicate_error = apply_test_event(&state, &secondary)
            .await
            .expect_err("second open Stripe subscription was accepted");
        assert_eq!(duplicate_error.code, ErrorCode::StorageUnavailable);
        let open_subscriptions: i64 = sqlx::query_scalar(
            "SELECT COUNT(*) FROM subscriptions WHERE user_id = $1 AND provider = 'stripe' AND status <> 'canceled'",
        )
        .bind(user_id)
        .fetch_one(&pool)
        .await
        .expect("count open Stripe subscriptions");
        assert_eq!(open_subscriptions, 1);
        let primary_deleted = subscription_event(
            &format!("evt_{marker}_single_primary_deleted"),
            "customer.subscription.deleted",
            1_700_005_200,
            &primary_subscription_id,
            &customer_id,
            &price_id,
        );
        apply_test_event(&state, &primary_deleted)
            .await
            .expect("cancel primary subscription");

        assert_eq!(
            stripe_mock.calls.load(Ordering::SeqCst),
            4,
            "unexpected number of authoritative Stripe snapshots"
        );

        sqlx::query(
            "DELETE FROM stripe_subscription_reconciliations WHERE provider_subscription_id LIKE $1",
        )
        .bind(format!("sub_{marker}%"))
        .execute(&pool)
        .await
        .expect("delete test reconciliation rows");
        sqlx::query(
            "DELETE FROM provider_object_event_watermarks WHERE provider_object_id LIKE $1",
        )
        .bind(format!("sub_{marker}%"))
        .execute(&pool)
        .await
        .expect("delete test watermarks");
        sqlx::query("DELETE FROM users WHERE id = $1")
            .bind(user_id)
            .execute(&pool)
            .await
            .expect("delete test user");
        sqlx::query("DELETE FROM plans WHERE id = $1")
            .bind(plan_id)
            .execute(&pool)
            .await
            .expect("delete test plan");
        stripe_mock_task.abort();
    }
}