diff --git a/docs/billing.md b/docs/billing.md index 6bf69aa..453724c 100644 --- a/docs/billing.md +++ b/docs/billing.md @@ -130,14 +130,18 @@ RETURNING used_units; - `payments.provider_payment_id` ↔ Stripe `payment_intent.id`（或 charge id，按实现选） ### 4.2 Checkout / Portal -- Checkout：后端创建 Stripe Checkout Session，前端跳转 `checkout_url`。 +- Checkout：后端按用户行锁串行创建 Stripe Checkout Session，前端跳转 `checkout_url`。服务端使用稳定 Customer 幂等键和待支付记录 ID 对应的 Session 幂等键，不依赖客户端 `Idempotency-Key`。 +- 每个用户只能映射一个非空 Stripe Customer，同时只能存在一个未过期待支付记录和一个未取消 Stripe 订阅。并发请求复用同一 Session；Customer 映射未持久化时禁止返回或创建 Session。 +- 已有未取消 Stripe 订阅的用户不能再次进入订阅 Checkout，升级、降级、续费和取消统一走 Portal，避免多重周期扣费。 - Portal：后端创建 Stripe Billing Portal Session，前端跳转管理支付方式/取消订阅。 ### 4.3 Stripe Webhook（商用必须） 要求： - **验签**：使用 `STRIPE_WEBHOOK_SECRET` 校验 `Stripe-Signature`。 - **事件幂等**：按 `provider_event_id` 去重（落库 `webhook_events`）。 -- **乱序容忍**：订阅对象按 `(event.created, 事件优先级, event.id)` 保存独立水位；`deleted` 即使先到也会保留 tombstone，旧 `created/updated` 不得恢复已取消订阅。 +- **乱序容忍**：订阅对象按 `(event.created, 事件优先级)` 保存独立水位；`deleted` 即使先到也会保留 tombstone，旧 `created/updated` 不得恢复已取消订阅。 +- **同秒歧义**：两个不同事件具有相同 `(event.created, 事件优先级)` 时，不能用不透明的 Event ID 排序，必须从 Stripe 拉取当前订阅快照并以快照响应时间推进水位。 +- **迁移对账**：历史版本用本地 `subscriptions.updated_at` 播种的非终态水位会标记为待对账；API 后台任务持租约获取 Stripe 快照，成功后才清除标记。未映射 Customer 或 Price 的受管订阅事件返回失败并等待重试，不能标记为已处理。 - **并发一致性**：`subscriptions(provider, provider_subscription_id)` 唯一，订阅业务写入与 `webhook_events=processed` 在同一事务提交。 - **可重放**：保存原始 payload（脱敏）用于排查。 diff --git a/docs/database.md b/docs/database.md index e5e13d7..398ac85 100644 --- a/docs/database.md +++ b/docs/database.md @@ -310,6 +310,13 @@ CREATE UNIQUE INDEX idx_webhook_events_unique ON webhook_events(provider, provid CREATE INDEX idx_webhook_events_status ON webhook_events(status); ``` +Stripe 运行时还通过迁移维护三组一致性结构： +- `billing_checkout_sessions` 持久化每用户唯一的待支付 Session 及处理租约，防止并发创建多个 Customer/Session。 +- `provider_object_event_watermarks` 以 Stripe `event.created` 和事件等级保存对象水位；同秒同等级的不同事件标记为歧义并触发权威快照，不能按 Event ID 字典序决定先后。 +- `stripe_subscription_reconciliations` 保存历史非因果水位的租约化对账任务，允许多 API 实例用 `FOR UPDATE SKIP LOCKED` 安全消费。 + +数据库唯一索引同时保证非空 `users.billing_customer_id` 全局唯一、`subscriptions(provider, provider_subscription_id)` 唯一，以及每用户最多一条未取消 Stripe 订阅。部署这些索引前必须先清理存量冲突，具体检查见 `docs/deployment.md`。 + ### 4.8 tasks - 压缩任务 ```sql CREATE TABLE tasks ( diff --git a/docs/deployment.md b/docs/deployment.md index fd265fa..a1ebe05 100644 --- a/docs/deployment.md +++ b/docs/deployment.md @@ -55,15 +55,48 @@ curl --fail http://127.0.0.1:8080/metrics ### 更新与回滚 -更新代码后保留 `.env.production` 和命名卷，重新构建并滚动重建： +更新代码后保留 `.env.production` 和命名卷。包含迁移 `017`、`018` 的版本不能直接让旧、新 Worker 并行滚动：先备份数据库并停止旧 Worker，再构建新镜像。 + +迁移 `017` 会在发现重复 Customer 或同用户多条未取消 Stripe 订阅时主动失败。部署前先检查并人工对账，两个查询都必须返回 0 行： + +```sql +SELECT billing_customer_id, COUNT(*) +FROM users +WHERE billing_customer_id IS NOT NULL AND billing_customer_id <> '' +GROUP BY billing_customer_id +HAVING COUNT(*) > 1; + +SELECT user_id, COUNT(*) +FROM subscriptions +WHERE provider = 'stripe' AND status <> 'canceled' +GROUP BY user_id +HAVING COUNT(*) > 1; +``` + +推荐顺序： ```bash git pull --ff-only +docker compose --env-file .env.production -f docker/docker-compose.prod.yml stop worker docker compose --env-file .env.production -f docker/docker-compose.prod.yml build api -docker compose --env-file .env.production -f docker/docker-compose.prod.yml up -d +docker compose --env-file .env.production -f docker/docker-compose.prod.yml up -d postgres redis api +docker compose --env-file .env.production -f docker/docker-compose.prod.yml up -d worker ``` -生产镜像应使用不可变的 `IMAGEFORGE_TAG`。回滚时把该值改回上一镜像标签，然后再次运行 `up -d`。 +新 API 启动后会消费迁移 `018` 创建的 Stripe 对账队列。启动 Worker 前应确认 API 健康、`STRIPE_SECRET_KEY` 可用且服务器能访问 `STRIPE_API_BASE_URL`；对账可以后台继续，但必须监控失败项： + +```sql +SELECT status, COUNT(*) +FROM stripe_subscription_reconciliations +GROUP BY status; + +SELECT provider_object_id, reconciliation_reason, updated_at +FROM provider_object_event_watermarks +WHERE provider = 'stripe' AND requires_reconciliation = true +ORDER BY updated_at; +``` + +`failed` 会指数退避重试；持续失败通常表示 Stripe 凭据、网络、Customer/Price 映射不完整。上线验收要求 `pending/processing/failed` 最终归零，且 `requires_reconciliation=true` 为 0。生产镜像应使用不可变的 `IMAGEFORGE_TAG`。数据库迁移已应用后，不能只回滚旧二进制；应保留新 schema，并使用兼容该 schema 的修复镜像。 ### 反向代理 diff --git a/migrations/018_stripe_watermark_reconciliation.sql b/migrations/018_stripe_watermark_reconciliation.sql new file mode 100644 index 0000000..d0ea158 --- /dev/null +++ b/migrations/018_stripe_watermark_reconciliation.sql @@ -0,0 +1,57 @@ +ALTER TABLE provider_object_event_watermarks + ADD COLUMN IF NOT EXISTS requires_reconciliation BOOLEAN NOT NULL DEFAULT false, + ADD COLUMN IF NOT EXISTS reconciliation_reason TEXT, + ADD COLUMN IF NOT EXISTS last_snapshot_at TIMESTAMPTZ; + +CREATE TABLE IF NOT EXISTS stripe_subscription_reconciliations ( + provider_subscription_id VARCHAR(200) PRIMARY KEY, + reason TEXT NOT NULL, + status VARCHAR(20) NOT NULL DEFAULT 'pending', + attempts INTEGER NOT NULL DEFAULT 0, + next_attempt_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + last_error TEXT, + lease_owner UUID, + lease_until TIMESTAMPTZ, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + completed_at TIMESTAMPTZ, + CONSTRAINT stripe_subscription_reconciliations_status_check + CHECK (status IN ('pending', 'processing', 'completed', 'failed')) +); + +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 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 +ON CONFLICT (provider_subscription_id) DO UPDATE +SET reason = EXCLUDED.reason, + status = 'pending', + next_attempt_at = NOW(), + last_error = NULL, + lease_owner = NULL, + lease_until = NULL, + completed_at = NULL, + updated_at = NOW(); + +CREATE INDEX IF NOT EXISTS idx_stripe_subscription_reconciliations_ready + ON stripe_subscription_reconciliations(next_attempt_at, updated_at) + WHERE status IN ('pending', 'failed'); diff --git a/src/api/mod.rs b/src/api/mod.rs index 1411623..6746c01 100644 --- a/src/api/mod.rs +++ b/src/api/mod.rs @@ -44,7 +44,7 @@ pub async fn run(state: AppState) -> Result<(), AppError> { .nest("/api/v1", v1) .fallback_service(static_service) .layer(axum::middleware::from_fn(request_context::middleware)) - .with_state(state); + .with_state(state.clone()); let listener = tokio::net::TcpListener::bind(&addr) .await @@ -52,12 +52,15 @@ pub async fn run(state: AppState) -> Result<(), AppError> { tracing::info!(addr = %addr, "API server listening"); - axum::serve( + let reconciliation_task = tokio::spawn(webhooks::reconciliation_loop(state.clone())); + let serve_result = axum::serve( listener, app.into_make_service_with_connect_info::(), ) - .await - .map_err(|err| AppError::new(ErrorCode::Internal, "HTTP 服务异常退出").with_source(err)) + .await; + reconciliation_task.abort(); + serve_result + .map_err(|err| AppError::new(ErrorCode::Internal, "HTTP 服务异常退出").with_source(err)) } fn v1_router() -> Router { diff --git a/src/api/webhooks.rs b/src/api/webhooks.rs index ba647ab..4dae2c4 100644 --- a/src/api/webhooks.rs +++ b/src/api/webhooks.rs @@ -10,6 +10,7 @@ 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}; @@ -145,7 +146,7 @@ async fn process_claimed_stripe_event( )); } - process_stripe_event(&mut tx, event).await?; + process_stripe_event(state, &mut tx, event).await?; let updated = sqlx::query( r#" UPDATE webhook_events @@ -234,6 +235,7 @@ fn secure_eq(a: &str, b: &str) -> bool { } async fn process_stripe_event( + state: &AppState, tx: &mut Transaction<'_, Postgres>, event: &StripeEvent, ) -> Result<(), AppError> { @@ -242,9 +244,11 @@ async fn process_stripe_event( map_checkout_session_completed(tx, &event.data.object).await } "customer.subscription.created" | "customer.subscription.updated" => { - upsert_subscription(tx, event, &event.data.object).await + apply_subscription_event(state, tx, event, &event.data.object).await + } + "customer.subscription.deleted" => { + apply_subscription_event(state, 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(()), } @@ -257,7 +261,13 @@ async fn map_checkout_session_completed( let customer_id = object .get("customer") .and_then(|v| v.as_str()) - .filter(|v| !v.trim().is_empty()); + .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") @@ -268,57 +278,71 @@ async fn map_checkout_session_completed( .pointer("/metadata/user_id") .and_then(|v| v.as_str()) .and_then(|v| v.parse::().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::().ok()); - let Some(customer_id) = customer_id else { - tracing::warn!("checkout.session.completed missing customer"); - return Ok(()); - }; - let Some(user_id) = user_id else { - tracing::warn!(customer = %customer_id, "checkout.session.completed missing user_id"); - return Ok(()); - }; + let current: Option> = + 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) + })?; + } - let updated = sqlx::query( + sqlx::query( r#" - UPDATE users - SET billing_customer_id = $2, + 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 id = $1 - AND (billing_customer_id IS NULL OR billing_customer_id = '') + 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, "更新 Stripe Customer 映射失败").with_source(err) - })?; - - if updated.rows_affected() == 0 { - let existing: Option = sqlx::query_scalar::<_, Option>( - "SELECT billing_customer_id FROM users WHERE id = $1", - ) - .bind(user_id) - .fetch_optional(&mut **tx) - .await - .map_err(|err| AppError::new(ErrorCode::Internal, "查询用户失败").with_source(err))? - .flatten(); - - if let Some(existing) = existing.filter(|v| !v.trim().is_empty()) { - if existing != customer_id { - tracing::warn!( - user_id = %user_id, - existing_customer = %existing, - new_customer = %customer_id, - "user already mapped to different stripe customer" - ); - } - } else { - tracing::warn!(user_id = %user_id, "user not found for checkout.session.completed"); - } - } + .map_err(|err| AppError::new(ErrorCode::Internal, "完成 Checkout 状态失败").with_source(err))?; Ok(()) } @@ -331,13 +355,21 @@ fn subscription_event_rank(event_type: &str) -> i16 { } } -async fn claim_subscription_event( +#[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 { - let claimed: Option = sqlx::query_scalar( +) -> Result { + let rank = subscription_event_rank(&event.type_); + let inserted: Option = sqlx::query_scalar( r#" INSERT INTO provider_object_event_watermarks ( provider, object_type, provider_object_id, @@ -346,109 +378,272 @@ async fn claim_subscription_event( '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 - ) + ON CONFLICT (provider, object_type, provider_object_id) DO NOTHING RETURNING provider_object_id "#, ) .bind(provider_subscription_id) .bind(event.created) - .bind(subscription_event_rank(&event.type_)) + .bind(rank) .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()) + .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) } -async fn upsert_subscription( +#[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, + current_period_end: chrono::DateTime, + cancel_at_period_end: bool, + canceled_at: Option>, + checkout_attempt_id: Option, +} + +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(|v| v.as_str()) + .and_then(|value| value.as_str()) + .filter(|value| !value.trim().is_empty()) .ok_or_else(|| AppError::new(ErrorCode::InvalidRequest, "subscription.id 缺失"))?; - if !claim_subscription_event(tx, event, provider_subscription_id, false).await? { - return Ok(()); + 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 { + 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(|v| v.as_str()) + .and_then(|value| value.as_str()) + .filter(|value| !value.trim().is_empty()) .ok_or_else(|| AppError::new(ErrorCode::InvalidRequest, "subscription.customer 缺失"))?; - - let status = object - .get("status") - .and_then(|v| v.as_str()) - .unwrap_or("incomplete"); - let mapped_status = map_subscription_status(status); - - let cps = object - .get("current_period_start") - .and_then(|v| v.as_i64()) - .unwrap_or(0); - let cpe = object - .get("current_period_end") - .and_then(|v| v.as_i64()) - .unwrap_or(0); - let current_period_start = Utc.timestamp_opt(cps, 0).single().unwrap_or_else(Utc::now); - let current_period_end = Utc.timestamp_opt(cpe, 0).single().unwrap_or_else(Utc::now); - - let cancel_at_period_end = object - .get("cancel_at_period_end") - .and_then(|v| v.as_bool()) - .unwrap_or(false); - let price_id = object .pointer("/items/data/0/price/id") - .and_then(|v| v.as_str()) + .and_then(|value| value.as_str()) .or_else(|| { object .pointer("/items/data/0/plan/id") - .and_then(|v| v.as_str()) + .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 = - sqlx::query_scalar("SELECT id FROM users WHERE billing_customer_id = $1 LIMIT 1") + 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 Some(user_id) = user_id else { - tracing::warn!(customer = %provider_customer_id, "stripe customer not mapped to user"); - return Ok(()); - }; - + .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 = 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))?; + .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 Some(plan_id) = plan_id else { - tracing::warn!(price = %price_id, "stripe price not mapped to plan"); - return Ok(()); + 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::().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 = 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#" @@ -459,10 +654,8 @@ async fn upsert_subscription( 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 + $4, $5, $6, $7, + 'stripe', $8, $9 ) ON CONFLICT (provider, provider_subscription_id) DO UPDATE SET user_id = EXCLUDED.user_id, @@ -471,151 +664,308 @@ async fn upsert_subscription( 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, + canceled_at = EXCLUDED.canceled_at, provider_customer_id = EXCLUDED.provider_customer_id, updated_at = NOW() "#, ) - .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) + .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))?; - Ok(()) -} - -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 updated = sqlx::query( - r#" - UPDATE subscriptions - SET status = 'canceled', - cancel_at_period_end = false, - canceled_at = $2, - updated_at = NOW() - WHERE provider = 'stripe' AND provider_subscription_id = $1 - "#, - ) - .bind(provider_subscription_id) - .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); - + if subscription.status != "canceled" { 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, + 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(user_id) - .bind(plan_id) - .bind(period_start) - .bind(period_end) - .bind(canceled_at) - .bind(provider_customer_id) - .bind(provider_subscription_id) + .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, "创建已取消订阅 tombstone 失败").with_source(err) + AppError::new(ErrorCode::Internal, "完成订阅 Checkout 状态失败").with_source(err) })?; } + Ok(()) +} +async fn fetch_stripe_subscription_snapshot( + state: &AppState, + provider_subscription_id: &str, +) -> Result { + 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 { + 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 = 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(()) } @@ -783,11 +1133,96 @@ fn map_invoice_status(status: &str) -> &'static str { #[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; + use tokio::sync::{Barrier, RwLock, Semaphore}; use uuid::Uuid; + #[derive(Clone, Default)] + struct StripeSnapshotMock { + objects: Arc>>, + calls: Arc, + } + + async fn stripe_snapshot( + State(mock): State, + Path(subscription_id): Path, + ) -> 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), "中"); @@ -827,12 +1262,18 @@ mod tests { } } - 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 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) { @@ -856,7 +1297,7 @@ mod tests { } #[tokio::test(flavor = "multi_thread", worker_threads = 4)] - #[ignore = "requires an isolated IMAGEFORGE_TEST_DATABASE_URL containing 'test'"] + #[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"); @@ -864,12 +1305,18 @@ mod tests { 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}"); @@ -942,7 +1389,9 @@ mod tests { ), ]; for index in order { - apply_test_event(&pool, &events[index]).await; + apply_test_event(&state, &events[index]) + .await + .expect("apply permuted Stripe event"); } assert_canceled_once(&pool, &subscription_id).await; } @@ -974,7 +1423,9 @@ mod tests { &price_id, ), ] { - apply_test_event(&pool, &event).await; + apply_test_event(&state, &event) + .await + .expect("apply same-second ranked event"); } assert_canceled_once(&pool, &same_second_id).await; @@ -1009,19 +1460,282 @@ mod tests { let barrier = Arc::new(Barrier::new(events.len())); let mut joins = Vec::new(); for event in events { - let pool = pool.clone(); + let state = state.clone(); let barrier = barrier.clone(); joins.push(tokio::spawn(async move { barrier.wait().await; - apply_test_event(&pool, &event).await; + apply_test_event(&state, &event).await })); } for join in joins { - join.await.expect("concurrent event task"); + 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", ) @@ -1039,5 +1753,6 @@ mod tests { .execute(&pool) .await .expect("delete test plan"); + stripe_mock_task.abort(); } }