diff --git a/docs/billing.md b/docs/billing.md index 781f79a..6bf69aa 100644 --- a/docs/billing.md +++ b/docs/billing.md @@ -137,7 +137,8 @@ RETURNING used_units; 要求： - **验签**：使用 `STRIPE_WEBHOOK_SECRET` 校验 `Stripe-Signature`。 - **事件幂等**：按 `provider_event_id` 去重（落库 `webhook_events`）。 -- **乱序容忍**：以 Stripe 事件时间 + 当前 DB 状态做“只前进不回退”更新。 +- **乱序容忍**：订阅对象按 `(event.created, 事件优先级, event.id)` 保存独立水位；`deleted` 即使先到也会保留 tombstone，旧 `created/updated` 不得恢复已取消订阅。 +- **并发一致性**：`subscriptions(provider, provider_subscription_id)` 唯一，订阅业务写入与 `webhook_events=processed` 在同一事务提交。 - **可重放**：保存原始 payload（脱敏）用于排查。 建议首期处理的事件（示例）： diff --git a/migrations/016_stripe_event_ordering.sql b/migrations/016_stripe_event_ordering.sql new file mode 100644 index 0000000..450362e --- /dev/null +++ b/migrations/016_stripe_event_ordering.sql @@ -0,0 +1,95 @@ +-- Repoint references before enforcing one row per provider subscription. +WITH ranked AS ( + SELECT + id, + FIRST_VALUE(id) OVER ( + PARTITION BY provider, provider_subscription_id + ORDER BY updated_at DESC, created_at DESC, id DESC + ) AS keep_id, + ROW_NUMBER() OVER ( + PARTITION BY provider, provider_subscription_id + ORDER BY updated_at DESC, created_at DESC, id DESC + ) AS row_number + FROM subscriptions + WHERE provider_subscription_id IS NOT NULL +) +UPDATE usage_periods AS usage +SET subscription_id = ranked.keep_id +FROM ranked +WHERE ranked.row_number > 1 + AND usage.subscription_id = ranked.id; + +WITH ranked AS ( + SELECT + id, + FIRST_VALUE(id) OVER ( + PARTITION BY provider, provider_subscription_id + ORDER BY updated_at DESC, created_at DESC, id DESC + ) AS keep_id, + ROW_NUMBER() OVER ( + PARTITION BY provider, provider_subscription_id + ORDER BY updated_at DESC, created_at DESC, id DESC + ) AS row_number + FROM subscriptions + WHERE provider_subscription_id IS NOT NULL +) +UPDATE invoices AS invoice +SET subscription_id = ranked.keep_id +FROM ranked +WHERE ranked.row_number > 1 + AND invoice.subscription_id = ranked.id; + +WITH ranked AS ( + SELECT + id, + ROW_NUMBER() OVER ( + PARTITION BY provider, provider_subscription_id + ORDER BY updated_at DESC, created_at DESC, id DESC + ) AS row_number + FROM subscriptions + WHERE provider_subscription_id IS NOT NULL +) +DELETE FROM subscriptions AS subscription +USING ranked +WHERE ranked.row_number > 1 + AND subscription.id = ranked.id; + +CREATE UNIQUE INDEX IF NOT EXISTS idx_subscriptions_provider_object_unique + ON subscriptions(provider, provider_subscription_id); + +CREATE TABLE IF NOT EXISTS provider_object_event_watermarks ( + provider VARCHAR(20) NOT NULL, + object_type VARCHAR(50) NOT NULL, + provider_object_id VARCHAR(200) NOT NULL, + last_event_created BIGINT NOT NULL, + last_event_rank SMALLINT NOT NULL DEFAULT 0, + last_event_id VARCHAR(200) NOT NULL, + is_deleted BOOLEAN NOT NULL DEFAULT false, + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + PRIMARY KEY (provider, object_type, provider_object_id), + CONSTRAINT provider_object_event_watermarks_rank_check + CHECK (last_event_rank BETWEEN 0 AND 100) +); + +CREATE INDEX IF NOT EXISTS idx_provider_object_event_watermarks_updated + ON provider_object_event_watermarks(updated_at); + +-- Seed rollout watermarks so a delayed pre-deployment event cannot revive an +-- already-canceled Stripe subscription before the first new event arrives. +INSERT INTO provider_object_event_watermarks ( + provider, object_type, provider_object_id, + last_event_created, last_event_rank, last_event_id, is_deleted, updated_at +) +SELECT + 'stripe', + 'subscription', + provider_subscription_id, + EXTRACT(EPOCH FROM updated_at)::bigint, + CASE WHEN status = 'canceled' THEN 2 ELSE 1 END, + 'migration:' || id::text, + status = 'canceled', + updated_at +FROM subscriptions +WHERE provider = 'stripe' + AND provider_subscription_id IS NOT NULL +ON CONFLICT (provider, object_type, provider_object_id) DO NOTHING; diff --git a/src/api/webhooks.rs b/src/api/webhooks.rs index 1ce69b1..ba647ab 100644 --- a/src/api/webhooks.rs +++ b/src/api/webhooks.rs @@ -12,20 +12,22 @@ use chrono::{TimeZone, Utc}; use hmac::{Hmac, Mac}; use serde::Deserialize; use sha2::Sha256; +use sqlx::{Postgres, Transaction}; pub fn router() -> Router { Router::new().route("/webhooks/stripe", post(stripe_webhook)) } -#[derive(Debug, Deserialize)] +#[derive(Debug, Clone, Deserialize)] struct StripeEvent { id: String, + created: i64, #[serde(rename = "type")] type_: String, data: StripeEventData, } -#[derive(Debug, Deserialize)] +#[derive(Debug, Clone, Deserialize)] struct StripeEventData { object: serde_json::Value, } @@ -99,7 +101,7 @@ async fn stripe_webhook( )); } - if let Err(err) = process_stripe_event(&state, &event).await { + 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'", ) @@ -111,19 +113,64 @@ async fn stripe_webhook( return Err(err); } - let _ = sqlx::query( - "UPDATE webhook_events SET status = 'processed', processed_at = NOW() WHERE provider = 'stripe' AND provider_event_id = $1", - ) - .bind(&event.id) - .execute(&state.db) - .await; - 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 = 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(&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 = None; let mut signatures = Vec::::new(); @@ -186,24 +233,25 @@ fn secure_eq(a: &str, b: &str) -> bool { out == 0 } -async fn process_stripe_event(state: &AppState, event: &StripeEvent) -> Result<(), AppError> { +async fn process_stripe_event( + tx: &mut Transaction<'_, Postgres>, + event: &StripeEvent, +) -> Result<(), AppError> { match event.type_.as_str() { "checkout.session.completed" => { - map_checkout_session_completed(state, &event.data.object).await + map_checkout_session_completed(tx, &event.data.object).await } "customer.subscription.created" | "customer.subscription.updated" => { - upsert_subscription(state, &event.data.object).await - } - "customer.subscription.deleted" => cancel_subscription(state, &event.data.object).await, - "invoice.paid" | "invoice.payment_failed" => { - upsert_invoice(state, &event.data.object).await + upsert_subscription(tx, event, &event.data.object).await } + "customer.subscription.deleted" => cancel_subscription(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( - state: &AppState, + tx: &mut Transaction<'_, Postgres>, object: &serde_json::Value, ) -> Result<(), AppError> { let customer_id = object @@ -242,7 +290,7 @@ async fn map_checkout_session_completed( ) .bind(user_id) .bind(customer_id) - .execute(&state.db) + .execute(&mut **tx) .await .map_err(|err| { AppError::new(ErrorCode::Internal, "更新 Stripe Customer 映射失败").with_source(err) @@ -253,7 +301,7 @@ async fn map_checkout_session_completed( "SELECT billing_customer_id FROM users WHERE id = $1", ) .bind(user_id) - .fetch_optional(&state.db) + .fetch_optional(&mut **tx) .await .map_err(|err| AppError::new(ErrorCode::Internal, "查询用户失败").with_source(err))? .flatten(); @@ -275,11 +323,72 @@ async fn map_checkout_session_completed( Ok(()) } -async fn upsert_subscription(state: &AppState, object: &serde_json::Value) -> Result<(), AppError> { +fn subscription_event_rank(event_type: &str) -> i16 { + match event_type { + "customer.subscription.deleted" => 2, + "customer.subscription.updated" => 1, + _ => 0, + } +} + +async fn claim_subscription_event( + tx: &mut Transaction<'_, Postgres>, + event: &StripeEvent, + provider_subscription_id: &str, + is_deleted: bool, +) -> Result { + let claimed: Option = 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 UPDATE + SET last_event_created = EXCLUDED.last_event_created, + last_event_rank = EXCLUDED.last_event_rank, + last_event_id = EXCLUDED.last_event_id, + is_deleted = EXCLUDED.is_deleted, + updated_at = NOW() + WHERE ( + EXCLUDED.last_event_created, + EXCLUDED.last_event_rank, + EXCLUDED.last_event_id + ) > ( + provider_object_event_watermarks.last_event_created, + provider_object_event_watermarks.last_event_rank, + provider_object_event_watermarks.last_event_id + ) + RETURNING provider_object_id + "#, + ) + .bind(provider_subscription_id) + .bind(event.created) + .bind(subscription_event_rank(&event.type_)) + .bind(&event.id) + .bind(is_deleted) + .fetch_optional(&mut **tx) + .await + .map_err(|err| { + AppError::new(ErrorCode::Internal, "更新 Stripe 订阅事件水位失败").with_source(err) + })?; + Ok(claimed.is_some()) +} + +async fn upsert_subscription( + tx: &mut Transaction<'_, Postgres>, + event: &StripeEvent, + object: &serde_json::Value, +) -> Result<(), AppError> { let provider_subscription_id = object .get("id") .and_then(|v| v.as_str()) .ok_or_else(|| AppError::new(ErrorCode::InvalidRequest, "subscription.id 缺失"))?; + if !claim_subscription_event(tx, event, provider_subscription_id, false).await? { + return Ok(()); + } let provider_customer_id = object .get("customer") .and_then(|v| v.as_str()) @@ -320,7 +429,7 @@ async fn upsert_subscription(state: &AppState, object: &serde_json::Value) -> Re let user_id: Option = sqlx::query_scalar("SELECT id FROM users WHERE billing_customer_id = $1 LIMIT 1") .bind(provider_customer_id) - .fetch_optional(&state.db) + .fetch_optional(&mut **tx) .await .map_err(|err| AppError::new(ErrorCode::Internal, "查询用户失败").with_source(err))?; @@ -332,7 +441,7 @@ async fn upsert_subscription(state: &AppState, object: &serde_json::Value) -> Re let plan_id: Option = sqlx::query_scalar("SELECT id FROM plans WHERE stripe_price_id = $1 LIMIT 1") .bind(price_id) - .fetch_optional(&state.db) + .fetch_optional(&mut **tx) .await .map_err(|err| AppError::new(ErrorCode::Internal, "查询套餐失败").with_source(err))?; @@ -341,20 +450,34 @@ async fn upsert_subscription(state: &AppState, object: &serde_json::Value) -> Re return Ok(()); }; - let updated: Option = sqlx::query_scalar( + sqlx::query( r#" - UPDATE subscriptions - SET user_id = $1, - plan_id = $2, - status = $3::subscription_status, - current_period_start = $4, - current_period_end = $5, - cancel_at_period_end = $6, - provider = 'stripe', - provider_customer_id = $7, + 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, + CASE WHEN $3 = 'canceled' THEN NOW() ELSE NULL END, + 'stripe', $7, $8 + ) + 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 = CASE + WHEN EXCLUDED.status = 'canceled'::subscription_status + THEN COALESCE(subscriptions.canceled_at, NOW()) + ELSE NULL + END, + provider_customer_id = EXCLUDED.provider_customer_id, updated_at = NOW() - WHERE provider = 'stripe' AND provider_subscription_id = $8 - RETURNING id "#, ) .bind(user_id) @@ -365,66 +488,141 @@ async fn upsert_subscription(state: &AppState, object: &serde_json::Value) -> Re .bind(cancel_at_period_end) .bind(provider_customer_id) .bind(provider_subscription_id) - .fetch_optional(&state.db) + .execute(&mut **tx) .await - .map_err(|err| AppError::new(ErrorCode::Internal, "更新订阅失败").with_source(err))?; - - if updated.is_none() { - let _ = sqlx::query( - r#" - INSERT INTO subscriptions ( - user_id, plan_id, status, - current_period_start, current_period_end, - cancel_at_period_end, - provider, provider_customer_id, provider_subscription_id - ) VALUES ( - $1, $2, $3::subscription_status, - $4, $5, - $6, - 'stripe', $7, $8 - ) - "#, - ) - .bind(user_id) - .bind(plan_id) - .bind(mapped_status) - .bind(current_period_start) - .bind(current_period_end) - .bind(cancel_at_period_end) - .bind(provider_customer_id) - .bind(provider_subscription_id) - .execute(&state.db) - .await - .map_err(|err| AppError::new(ErrorCode::Internal, "创建订阅失败").with_source(err))?; - } + .map_err(|err| AppError::new(ErrorCode::Internal, "写入订阅失败").with_source(err))?; Ok(()) } -async fn cancel_subscription(state: &AppState, object: &serde_json::Value) -> Result<(), AppError> { +async fn cancel_subscription( + tx: &mut Transaction<'_, Postgres>, + event: &StripeEvent, + object: &serde_json::Value, +) -> Result<(), AppError> { let provider_subscription_id = object .get("id") .and_then(|v| v.as_str()) .ok_or_else(|| AppError::new(ErrorCode::InvalidRequest, "subscription.id 缺失"))?; + if !claim_subscription_event(tx, event, provider_subscription_id, true).await? { + return Ok(()); + } + let canceled_at = object + .get("canceled_at") + .and_then(|v| v.as_i64()) + .unwrap_or(event.created); + let canceled_at = Utc + .timestamp_opt(canceled_at, 0) + .single() + .unwrap_or_else(Utc::now); - let _ = sqlx::query( + let updated = sqlx::query( r#" UPDATE subscriptions SET status = 'canceled', cancel_at_period_end = false, - canceled_at = NOW(), + canceled_at = $2, updated_at = NOW() WHERE provider = 'stripe' AND provider_subscription_id = $1 "#, ) .bind(provider_subscription_id) - .execute(&state.db) - .await; + .bind(canceled_at) + .execute(&mut **tx) + .await + .map_err(|err| AppError::new(ErrorCode::Internal, "取消订阅失败").with_source(err))?; + + if updated.rows_affected() == 0 { + let Some(provider_customer_id) = object + .get("customer") + .and_then(|value| value.as_str()) + .filter(|value| !value.trim().is_empty()) + else { + tracing::warn!(subscription = %provider_subscription_id, "deleted stripe subscription has no customer mapping; watermark retained"); + return Ok(()); + }; + let Some(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()) + }) + else { + tracing::warn!(subscription = %provider_subscription_id, "deleted stripe subscription has no price mapping; watermark retained"); + return Ok(()); + }; + let user_id: Option = + 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 plan_id: Option = + 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 (Some(user_id), Some(plan_id)) = (user_id, plan_id) else { + tracing::warn!(subscription = %provider_subscription_id, "deleted stripe subscription is not mapped locally; watermark retained"); + return Ok(()); + }; + let period_start = object + .get("current_period_start") + .and_then(|value| value.as_i64()) + .and_then(|value| Utc.timestamp_opt(value, 0).single()) + .unwrap_or(canceled_at); + let period_end = object + .get("current_period_end") + .and_then(|value| value.as_i64()) + .and_then(|value| Utc.timestamp_opt(value, 0).single()) + .unwrap_or(canceled_at); + + 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, 'canceled', $3, $4, false, $5, + 'stripe', $6, $7 + ) + ON CONFLICT (provider, provider_subscription_id) DO UPDATE + SET status = 'canceled', + cancel_at_period_end = false, + canceled_at = EXCLUDED.canceled_at, + updated_at = NOW() + "#, + ) + .bind(user_id) + .bind(plan_id) + .bind(period_start) + .bind(period_end) + .bind(canceled_at) + .bind(provider_customer_id) + .bind(provider_subscription_id) + .execute(&mut **tx) + .await + .map_err(|err| { + AppError::new(ErrorCode::Internal, "创建已取消订阅 tombstone 失败").with_source(err) + })?; + } Ok(()) } -async fn upsert_invoice(state: &AppState, object: &serde_json::Value) -> Result<(), AppError> { +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()) @@ -437,7 +635,7 @@ async fn upsert_invoice(state: &AppState, object: &serde_json::Value) -> Result< let user_id: Option = sqlx::query_scalar("SELECT id FROM users WHERE billing_customer_id = $1 LIMIT 1") .bind(provider_customer_id) - .fetch_optional(&state.db) + .fetch_optional(&mut **tx) .await .map_err(|err| AppError::new(ErrorCode::Internal, "查询用户失败").with_source(err))?; @@ -511,7 +709,7 @@ async fn upsert_invoice(state: &AppState, object: &serde_json::Value) -> Result< .bind(period_end) .bind(paid_at) .bind(provider_invoice_id) - .execute(&state.db) + .execute(&mut **tx) .await .map_err(|err| AppError::new(ErrorCode::Internal, "更新发票失败").with_source(err))?; @@ -543,7 +741,7 @@ async fn upsert_invoice(state: &AppState, object: &serde_json::Value) -> Result< .bind(hosted_invoice_url.as_deref()) .bind(pdf_url.as_deref()) .bind(paid_at) - .execute(&state.db) + .execute(&mut **tx) .await .map_err(|err| AppError::new(ErrorCode::Internal, "创建发票失败").with_source(err))?; } @@ -585,6 +783,10 @@ fn map_invoice_status(status: &str) -> &'static str { #[cfg(test)] mod tests { use super::*; + use sqlx::postgres::PgPoolOptions; + use std::sync::Arc; + use tokio::sync::Barrier; + use uuid::Uuid; #[test] fn truncate_preserves_utf8_boundaries() { @@ -592,4 +794,250 @@ mod tests { 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(pool: &sqlx::PgPool, event: &StripeEvent) { + let mut tx = pool.begin().await.expect("begin event transaction"); + process_stripe_event(&mut tx, event) + .await + .expect("apply stripe event"); + tx.commit().await.expect("commit stripe event"); + } + + 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 an isolated IMAGEFORGE_TEST_DATABASE_URL containing 'test'"] + 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 pool = PgPoolOptions::new() + .max_connections(16) + .connect(&database_url) + .await + .expect("connect test database"); + sqlx::migrate!().run(&pool).await.expect("run migrations"); + + 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(&pool, &events[index]).await; + } + 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(&pool, &event).await; + } + 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 pool = pool.clone(); + let barrier = barrier.clone(); + joins.push(tokio::spawn(async move { + barrier.wait().await; + apply_test_event(&pool, &event).await; + })); + } + for join in joins { + join.await.expect("concurrent event task"); + } + assert_canceled_once(&pool, &subscription_id).await; + } + + 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"); + } }