From 037e83e92f40eeb132dbf4a1c7a8ca5ca8b9204a Mon Sep 17 00:00:00 2001 From: 237899745 <237899745@users.noreply.git.workyai.cn> Date: Sun, 26 Jul 2026 04:53:21 +0800 Subject: [PATCH] fix: make Stripe invoice webhooks monotonic --- docs/billing.md | 2 + docs/database.md | 2 +- docs/deployment.md | 10 +- migrations/019_invoice_webhook_invariants.sql | 16 + src/api/webhooks.rs | 726 ++++++++++++++---- 5 files changed, 620 insertions(+), 136 deletions(-) create mode 100644 migrations/019_invoice_webhook_invariants.sql diff --git a/docs/billing.md b/docs/billing.md index 453724c..5824e02 100644 --- a/docs/billing.md +++ b/docs/billing.md @@ -100,6 +100,7 @@ RETURNING used_units; 说明： - 批量任务的计量仍以“成功文件数”为准；失败文件（含 `QUOTA_EXCEEDED`）不计费。 - 前端建议在上传前调用 `GET /billing/usage`（登录）或读取配额头（API）做本地提示/拦截。 +- 匿名批量任务先按文件数预留当日额度，终态结算只退还失败或未完成文件。未提供 `compression_rate` 属于正常压缩并计量；只有显式 `compression_rate=100`、同格式且无缩放的原样请求免计量。 --- @@ -142,6 +143,7 @@ RETURNING used_units; - **乱序容忍**：订阅对象按 `(event.created, 事件优先级)` 保存独立水位；`deleted` 即使先到也会保留 tombstone，旧 `created/updated` 不得恢复已取消订阅。 - **同秒歧义**：两个不同事件具有相同 `(event.created, 事件优先级)` 时，不能用不透明的 Event ID 排序，必须从 Stripe 拉取当前订阅快照并以快照响应时间推进水位。 - **迁移对账**：历史版本用本地 `subscriptions.updated_at` 播种的非终态水位会标记为待对账；API 后台任务持租约获取 Stripe 快照，成功后才清除标记。未映射 Customer 或 Price 的受管订阅事件返回失败并等待重试，不能标记为已处理。 +- **发票一致性**：`invoices(provider, provider_invoice_id)` 唯一，发票事件也使用对象水位；新 `invoice.paid` 不会被迟到的旧 `invoice.payment_failed` 回退。同秒同等级事件从 Stripe 获取权威发票快照，未映射 Customer 时返回失败重试。 - **并发一致性**：`subscriptions(provider, provider_subscription_id)` 唯一，订阅业务写入与 `webhook_events=processed` 在同一事务提交。 - **可重放**：保存原始 payload（脱敏）用于排查。 diff --git a/docs/database.md b/docs/database.md index 398ac85..6c328a2 100644 --- a/docs/database.md +++ b/docs/database.md @@ -315,7 +315,7 @@ Stripe 运行时还通过迁移维护三组一致性结构： - `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`。 +数据库唯一索引同时保证非空 `users.billing_customer_id` 全局唯一、`subscriptions(provider, provider_subscription_id)` 唯一、非空 `invoices(provider, provider_invoice_id)` 唯一，以及每用户最多一条未取消 Stripe 订阅。部署这些索引前必须先清理存量冲突，具体检查见 `docs/deployment.md`。 ### 4.8 tasks - 压缩任务 ```sql diff --git a/docs/deployment.md b/docs/deployment.md index a1ebe05..47e750a 100644 --- a/docs/deployment.md +++ b/docs/deployment.md @@ -55,9 +55,9 @@ curl --fail http://127.0.0.1:8080/metrics ### 更新与回滚 -更新代码后保留 `.env.production` 和命名卷。包含迁移 `017`、`018` 的版本不能直接让旧、新 Worker 并行滚动：先备份数据库并停止旧 Worker，再构建新镜像。 +更新代码后保留 `.env.production` 和命名卷。包含迁移 `017` 至 `019` 的版本不能直接让旧、新 Worker 并行滚动：先备份数据库并停止旧 Worker，再构建新镜像。 -迁移 `017` 会在发现重复 Customer 或同用户多条未取消 Stripe 订阅时主动失败。部署前先检查并人工对账，两个查询都必须返回 0 行： +迁移 `017` 会在发现重复 Customer 或同用户多条未取消 Stripe 订阅时主动失败，迁移 `019` 会在发现同一 Stripe 发票对应多行时主动失败。部署前先检查并人工对账，三个查询都必须返回 0 行： ```sql SELECT billing_customer_id, COUNT(*) @@ -71,6 +71,12 @@ FROM subscriptions WHERE provider = 'stripe' AND status <> 'canceled' GROUP BY user_id HAVING COUNT(*) > 1; + +SELECT provider, provider_invoice_id, COUNT(*) +FROM invoices +WHERE provider_invoice_id IS NOT NULL +GROUP BY provider, provider_invoice_id +HAVING COUNT(*) > 1; ``` 推荐顺序： diff --git a/migrations/019_invoice_webhook_invariants.sql b/migrations/019_invoice_webhook_invariants.sql new file mode 100644 index 0000000..3c1842a --- /dev/null +++ b/migrations/019_invoice_webhook_invariants.sql @@ -0,0 +1,16 @@ +DO $$ +BEGIN + IF EXISTS ( + SELECT 1 + FROM invoices + WHERE provider_invoice_id IS NOT NULL + GROUP BY provider, provider_invoice_id + HAVING COUNT(*) > 1 + ) THEN + RAISE EXCEPTION 'duplicate invoices(provider, provider_invoice_id) values require reconciliation before migration 019'; + END IF; +END $$; + +CREATE UNIQUE INDEX IF NOT EXISTS idx_invoices_provider_object_unique + ON invoices(provider, provider_invoice_id) + WHERE provider_invoice_id IS NOT NULL; diff --git a/src/api/webhooks.rs b/src/api/webhooks.rs index 4dae2c4..4e230b6 100644 --- a/src/api/webhooks.rs +++ b/src/api/webhooks.rs @@ -249,7 +249,9 @@ async fn process_stripe_event( "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, + "invoice.paid" | "invoice.payment_failed" => { + apply_invoice_event(state, tx, event, &event.data.object).await + } _ => Ok(()), } } @@ -355,6 +357,13 @@ fn subscription_event_rank(event_type: &str) -> i16 { } } +fn invoice_event_rank(event_type: &str) -> i16 { + match event_type { + "invoice.paid" => 1, + _ => 0, + } +} + #[derive(Debug, Clone, Copy, PartialEq, Eq)] enum WatermarkDecision { Apply, @@ -362,27 +371,29 @@ enum WatermarkDecision { Reconcile, } -async fn decide_subscription_event( +async fn decide_object_event( tx: &mut Transaction<'_, Postgres>, event: &StripeEvent, - provider_subscription_id: &str, + object_type: &str, + provider_object_id: &str, + rank: i16, is_deleted: bool, ) -> 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, last_event_created, last_event_rank, last_event_id, is_deleted ) VALUES ( - 'stripe', 'subscription', $1, - $2, $3, $4, $5 + 'stripe', $1, $2, + $3, $4, $5, $6 ) ON CONFLICT (provider, object_type, provider_object_id) DO NOTHING RETURNING provider_object_id "#, ) - .bind(provider_subscription_id) + .bind(object_type) + .bind(provider_object_id) .bind(event.created) .bind(rank) .bind(&event.id) @@ -399,12 +410,13 @@ async fn decide_subscription_event( 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 + AND object_type = $1 + AND provider_object_id = $2 FOR UPDATE "#, ) - .bind(provider_subscription_id) + .bind(object_type) + .bind(provider_object_id) .fetch_one(&mut **tx) .await .map_err(|err| AppError::new(ErrorCode::Internal, "锁定 Stripe 水位失败").with_source(err))?; @@ -436,15 +448,16 @@ async fn decide_subscription_event( reconciliation_reason = NULL, updated_at = NOW() WHERE provider = 'stripe' - AND object_type = 'subscription' + AND object_type = $6 AND provider_object_id = $1 "#, ) - .bind(provider_subscription_id) + .bind(provider_object_id) .bind(event.created) .bind(rank) .bind(&event.id) .bind(is_deleted) + .bind(object_type) .execute(&mut **tx) .await .map_err(|err| AppError::new(ErrorCode::Internal, "推进 Stripe 水位失败").with_source(err))?; @@ -465,7 +478,7 @@ struct ResolvedSubscription { checkout_attempt_id: Option, } -struct StripeSubscriptionSnapshot { +struct StripeObjectSnapshot { object: serde_json::Value, reconciled_through: i64, } @@ -481,17 +494,25 @@ async fn apply_subscription_event( .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? { + let resolved = resolve_subscription(tx, object).await?; + let is_deleted = + event.type_ == "customer.subscription.deleted" || resolved.status == "canceled"; + match decide_object_event( + tx, + event, + "subscription", + provider_subscription_id, + subscription_event_rank(&event.type_), + is_deleted, + ) + .await? + { WatermarkDecision::Ignore => Ok(()), - WatermarkDecision::Apply => { - let resolved = resolve_subscription(tx, object).await?; - write_subscription(tx, &resolved).await - } + WatermarkDecision::Apply => write_subscription(tx, &resolved).await, WatermarkDecision::Reconcile => { let snapshot = - fetch_stripe_subscription_snapshot(state, provider_subscription_id).await?; + fetch_stripe_object_snapshot(state, "subscriptions", provider_subscription_id) + .await?; let authoritative = resolve_subscription(tx, &snapshot.object).await?; if authoritative.provider_subscription_id != provider_subscription_id { return Err(AppError::new( @@ -500,7 +521,7 @@ async fn apply_subscription_event( )); } write_subscription(tx, &authoritative).await?; - record_snapshot_watermark( + record_subscription_snapshot_watermark( tx, &authoritative, snapshot.reconciled_through.max(event.created), @@ -710,17 +731,18 @@ async fn write_subscription( Ok(()) } -async fn fetch_stripe_subscription_snapshot( +async fn fetch_stripe_object_snapshot( state: &AppState, - provider_subscription_id: &str, -) -> Result { + resource: &str, + provider_object_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 encoded_id = utf8_percent_encode(provider_object_id, NON_ALPHANUMERIC); let url = format!( - "{}/v1/subscriptions/{encoded_id}", - state.config.stripe_api_base_url + "{}/v1/{resource}/{encoded_id}", + state.config.stripe_api_base_url, ); let response = reqwest::Client::new() .get(url) @@ -743,26 +765,28 @@ async fn fetch_stripe_subscription_snapshot( AppError::new(ErrorCode::Internal, "读取 Stripe 对账响应失败").with_source(err) })?; if !status.is_success() { - tracing::error!(%status, %body, provider_subscription_id, "Stripe subscription reconciliation failed"); + tracing::error!(%status, %body, resource, provider_object_id, "Stripe object reconciliation failed"); return Err(AppError::new( ErrorCode::StorageUnavailable, - "Stripe 订阅对账失败，将自动重试", + "Stripe 对象对账失败，将自动重试", )); } let object = serde_json::from_str(&body).map_err(|err| { AppError::new(ErrorCode::Internal, "解析 Stripe 对账响应失败").with_source(err) })?; - Ok(StripeSubscriptionSnapshot { + Ok(StripeObjectSnapshot { object, reconciled_through, }) } -async fn record_snapshot_watermark( +async fn write_snapshot_watermark( tx: &mut Transaction<'_, Postgres>, - subscription: &ResolvedSubscription, + object_type: &str, + provider_object_id: &str, reconciled_through: i64, trigger_id: &str, + is_deleted: bool, ) -> Result<(), AppError> { sqlx::query( r#" @@ -772,9 +796,9 @@ async fn record_snapshot_watermark( is_deleted, requires_reconciliation, reconciliation_reason, last_snapshot_at, updated_at ) VALUES ( - 'stripe', 'subscription', $1, - $2, 100, $3, - $4, false, NULL, + 'stripe', $1, $2, + $3, 100, $4, + $5, false, NULL, NOW(), NOW() ) ON CONFLICT (provider, object_type, provider_object_id) DO UPDATE @@ -788,15 +812,34 @@ async fn record_snapshot_watermark( updated_at = NOW() "#, ) - .bind(&subscription.provider_subscription_id) + .bind(object_type) + .bind(provider_object_id) .bind(reconciled_through) .bind(format!("snapshot:{trigger_id}")) - .bind(subscription.status == "canceled") + .bind(is_deleted) .execute(&mut **tx) .await .map_err(|err| { AppError::new(ErrorCode::Internal, "保存 Stripe 快照水位失败").with_source(err) })?; + Ok(()) +} + +async fn record_subscription_snapshot_watermark( + tx: &mut Transaction<'_, Postgres>, + subscription: &ResolvedSubscription, + reconciled_through: i64, + trigger_id: &str, +) -> Result<(), AppError> { + write_snapshot_watermark( + tx, + "subscription", + &subscription.provider_subscription_id, + reconciled_through, + trigger_id, + subscription.status == "canceled", + ) + .await?; sqlx::query( r#" UPDATE stripe_subscription_reconciliations @@ -901,7 +944,8 @@ async fn reconcile_claimed_subscription( provider_subscription_id: &str, lease_owner: uuid::Uuid, ) -> Result<(), AppError> { - let snapshot = fetch_stripe_subscription_snapshot(state, provider_subscription_id).await?; + let snapshot = + fetch_stripe_object_snapshot(state, "subscriptions", provider_subscription_id).await?; let mut tx = state.db.begin().await.map_err(|err| { AppError::new(ErrorCode::Internal, "开启 Stripe 对账事务失败").with_source(err) })?; @@ -956,7 +1000,7 @@ async fn reconcile_claimed_subscription( } write_subscription(&mut tx, &authoritative).await?; - record_snapshot_watermark( + record_subscription_snapshot_watermark( &mut tx, &authoritative, snapshot.reconciled_through, @@ -969,133 +1013,211 @@ async fn reconcile_claimed_subscription( Ok(()) } -async fn upsert_invoice( +#[derive(Debug)] +struct ResolvedInvoice { + provider_invoice_id: String, + user_id: uuid::Uuid, + invoice_number: Option, + fallback_invoice_number: String, + status: String, + currency: String, + total_amount_cents: i32, + period_start: Option>, + period_end: Option>, + hosted_invoice_url: Option, + pdf_url: Option, + paid_at: Option>, +} + +async fn apply_invoice_event( + state: &AppState, tx: &mut Transaction<'_, Postgres>, + event: &StripeEvent, object: &serde_json::Value, ) -> Result<(), AppError> { let provider_invoice_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, "invoice.id 缺失"))?; + let invoice = resolve_invoice(tx, object).await?; + match decide_object_event( + tx, + event, + "invoice", + provider_invoice_id, + invoice_event_rank(&event.type_), + false, + ) + .await? + { + WatermarkDecision::Ignore => Ok(()), + WatermarkDecision::Apply => write_invoice(tx, &invoice).await, + WatermarkDecision::Reconcile => { + let snapshot = + fetch_stripe_object_snapshot(state, "invoices", provider_invoice_id).await?; + let authoritative = resolve_invoice(tx, &snapshot.object).await?; + if authoritative.provider_invoice_id != provider_invoice_id { + return Err(AppError::new( + ErrorCode::Internal, + "Stripe 对账快照发票 ID 不一致", + )); + } + write_invoice(tx, &authoritative).await?; + write_snapshot_watermark( + tx, + "invoice", + &authoritative.provider_invoice_id, + snapshot.reconciled_through.max(event.created), + &event.id, + false, + ) + .await + } + } +} + +async fn resolve_invoice( + tx: &mut Transaction<'_, Postgres>, + object: &serde_json::Value, +) -> Result { + let provider_invoice_id = object + .get("id") + .and_then(|value| value.as_str()) + .filter(|value| !value.trim().is_empty()) .ok_or_else(|| AppError::new(ErrorCode::InvalidRequest, "invoice.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, "invoice.customer 缺失"))?; 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 { - return Ok(()); - }; - - let stripe_status = object - .get("status") - .and_then(|v| v.as_str()) - .unwrap_or("open"); - let status = map_invoice_status(stripe_status); + .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 status = map_invoice_status( + object + .get("status") + .and_then(|value| value.as_str()) + .unwrap_or("open"), + ) + .to_string(); 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}")); - + .and_then(|value| value.as_str()) + .filter(|value| !value.trim().is_empty()) + .map(|value| truncate(value.to_string(), 50)); + let fallback_invoice_number = truncate(format!("stripe_{provider_invoice_id}"), 50); let currency = object .get("currency") - .and_then(|v| v.as_str()) + .and_then(|value| value.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 total_amount_cents = i32::try_from( + object + .get("total") + .and_then(|value| value.as_i64()) + .unwrap_or(0), + ) + .map_err(|_| AppError::new(ErrorCode::InvalidRequest, "invoice.total 超出范围"))?; let hosted_invoice_url = object .get("hosted_invoice_url") - .and_then(|v| v.as_str()) - .map(|v| v.to_string()); + .and_then(|value| value.as_str()) + .filter(|value| !value.trim().is_empty()) + .map(str::to_string); let pdf_url = object .get("invoice_pdf") - .and_then(|v| v.as_str()) - .map(|v| v.to_string()); - + .and_then(|value| value.as_str()) + .filter(|value| !value.trim().is_empty()) + .map(str::to_string); let period_start = object .get("period_start") - .and_then(|v| v.as_i64()) - .and_then(|ts| Utc.timestamp_opt(ts, 0).single()); + .and_then(|value| value.as_i64()) + .and_then(|timestamp| Utc.timestamp_opt(timestamp, 0).single()); let period_end = object .get("period_end") - .and_then(|v| v.as_i64()) - .and_then(|ts| Utc.timestamp_opt(ts, 0).single()); - + .and_then(|value| value.as_i64()) + .and_then(|timestamp| Utc.timestamp_opt(timestamp, 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()); + .and_then(|value| value.as_i64()) + .and_then(|timestamp| Utc.timestamp_opt(timestamp, 0).single()); - let updated = sqlx::query( + Ok(ResolvedInvoice { + provider_invoice_id: provider_invoice_id.to_string(), + user_id, + invoice_number, + fallback_invoice_number, + status, + currency, + total_amount_cents, + period_start, + period_end, + hosted_invoice_url, + pdf_url, + paid_at, + }) +} + +async fn write_invoice( + tx: &mut Transaction<'_, Postgres>, + invoice: &ResolvedInvoice, +) -> Result<(), AppError> { + 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 + 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, COALESCE($2, $3), $4::invoice_status, $5, $6, + $7, $8, + 'stripe', $9, $10, $11, + $12 + ) + ON CONFLICT (provider, provider_invoice_id) + WHERE provider_invoice_id IS NOT NULL + DO UPDATE SET + user_id = EXCLUDED.user_id, + invoice_number = COALESCE($2, invoices.invoice_number), + status = EXCLUDED.status, + currency = EXCLUDED.currency, + total_amount_cents = EXCLUDED.total_amount_cents, + hosted_invoice_url = COALESCE(EXCLUDED.hosted_invoice_url, invoices.hosted_invoice_url), + pdf_url = COALESCE(EXCLUDED.pdf_url, invoices.pdf_url), + period_start = COALESCE(EXCLUDED.period_start, invoices.period_start), + period_end = COALESCE(EXCLUDED.period_end, invoices.period_end), + paid_at = COALESCE(EXCLUDED.paid_at, invoices.paid_at) "#, ) - .bind(status) - .bind(¤cy) - .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) + .bind(invoice.user_id) + .bind(invoice.invoice_number.as_deref()) + .bind(&invoice.fallback_invoice_number) + .bind(&invoice.status) + .bind(&invoice.currency) + .bind(invoice.total_amount_cents) + .bind(invoice.period_start) + .bind(invoice.period_end) + .bind(&invoice.provider_invoice_id) + .bind(invoice.hosted_invoice_url.as_deref()) + .bind(invoice.pdf_url.as_deref()) + .bind(invoice.paid_at) .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(¤cy) - .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))?; - } - + .map_err(|err| AppError::new(ErrorCode::Internal, "写入发票失败").with_source(err))?; Ok(()) } @@ -1153,10 +1275,10 @@ mod tests { async fn stripe_snapshot( State(mock): State, - Path(subscription_id): Path, + Path(object_id): Path, ) -> Response { mock.calls.fetch_add(1, Ordering::SeqCst); - let object = mock.objects.read().await.get(&subscription_id).cloned(); + let object = mock.objects.read().await.get(&object_id).cloned(); let mut headers = HeaderMap::new(); headers.insert( header::DATE, @@ -1181,6 +1303,7 @@ mod tests { "/v1/subscriptions/{id}", axum::routing::get(stripe_snapshot), ) + .route("/v1/invoices/{id}", axum::routing::get(stripe_snapshot)) .with_state(mock.clone()); let listener = tokio::net::TcpListener::bind("127.0.0.1:0") .await @@ -1262,6 +1385,39 @@ mod tests { } } + fn invoice_event( + event_id: &str, + event_type: &str, + created: i64, + invoice_id: &str, + customer_id: &str, + invoice_number: Option<&str>, + ) -> StripeEvent { + let paid = event_type == "invoice.paid"; + StripeEvent { + id: event_id.to_string(), + created, + type_: event_type.to_string(), + data: StripeEventData { + object: serde_json::json!({ + "id": invoice_id, + "customer": customer_id, + "number": invoice_number, + "status": if paid { "paid" } else { "open" }, + "currency": "cny", + "total": 1_900, + "period_start": 1_700_000_000_i64, + "period_end": 1_702_592_000_i64, + "hosted_invoice_url": format!("https://billing.example.test/{invoice_id}"), + "invoice_pdf": format!("https://billing.example.test/{invoice_id}.pdf"), + "status_transitions": { + "paid_at": if paid { Some(created) } else { None } + } + }), + }, + } + } + 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 { @@ -1296,6 +1452,29 @@ mod tests { assert!(watermark.0); } + async fn assert_invoice_once( + pool: &sqlx::PgPool, + invoice_id: &str, + expected_status: &str, + expected_number: &str, + ) { + let rows: Vec<(String, String, Option>)> = sqlx::query_as( + r#" + SELECT status::text, invoice_number, paid_at + FROM invoices + WHERE provider = 'stripe' AND provider_invoice_id = $1 + "#, + ) + .bind(invoice_id) + .fetch_all(pool) + .await + .expect("query Stripe invoice"); + assert_eq!(rows.len(), 1, "Stripe invoice was not unique"); + assert_eq!(rows[0].0, expected_status); + assert_eq!(rows[0].1, expected_number); + assert_eq!(rows[0].2.is_some(), expected_status == "paid"); + } + #[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() { @@ -1723,9 +1902,174 @@ mod tests { .await .expect("cancel primary subscription"); + let concurrent_invoice_id = format!("inv_{marker}_concurrent"); + let concurrent_invoice_number = format!("INV-{marker}-C"); + let concurrent_invoice_events = [ + invoice_event( + &format!("evt_{marker}_invoice_failed_concurrent"), + "invoice.payment_failed", + 1_700_006_000, + &concurrent_invoice_id, + &customer_id, + Some(&concurrent_invoice_number), + ), + invoice_event( + &format!("evt_{marker}_invoice_paid_concurrent"), + "invoice.paid", + 1_700_006_100, + &concurrent_invoice_id, + &customer_id, + Some(&concurrent_invoice_number), + ), + ]; + let barrier = Arc::new(Barrier::new(concurrent_invoice_events.len())); + let mut joins = Vec::new(); + for event in concurrent_invoice_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 concurrent invoice event") + .expect("apply concurrent invoice event"); + } + assert_invoice_once( + &pool, + &concurrent_invoice_id, + "paid", + &concurrent_invoice_number, + ) + .await; + + let ordered_invoice_id = format!("inv_{marker}_ordered"); + let ordered_invoice_number = format!("INV-{marker}-O"); + let paid = invoice_event( + &format!("evt_{marker}_invoice_paid_new"), + "invoice.paid", + 1_700_007_000, + &ordered_invoice_id, + &customer_id, + Some(&ordered_invoice_number), + ); + let stale_failed = invoice_event( + &format!("evt_{marker}_invoice_failed_old"), + "invoice.payment_failed", + 1_700_006_900, + &ordered_invoice_id, + &customer_id, + Some(&ordered_invoice_number), + ); + apply_test_event(&state, &paid) + .await + .expect("apply newer paid invoice"); + apply_test_event(&state, &stale_failed) + .await + .expect("ignore stale payment failure"); + assert_invoice_once(&pool, &ordered_invoice_id, "paid", &ordered_invoice_number).await; + + let numbered_invoice_id = format!("inv_{marker}_numbered"); + let fallback_number = truncate(format!("stripe_{numbered_invoice_id}"), 50); + let actual_number = format!("INV-{marker}-N"); + let missing_number = invoice_event( + &format!("evt_{marker}_invoice_missing_number"), + "invoice.payment_failed", + 1_700_008_000, + &numbered_invoice_id, + &customer_id, + None, + ); + let with_number = invoice_event( + &format!("evt_{marker}_invoice_with_number"), + "invoice.payment_failed", + 1_700_008_100, + &numbered_invoice_id, + &customer_id, + Some(&actual_number), + ); + apply_test_event(&state, &missing_number) + .await + .expect("apply invoice without number"); + assert_invoice_once(&pool, &numbered_invoice_id, "open", &fallback_number).await; + apply_test_event(&state, &with_number) + .await + .expect("apply invoice number update"); + assert_invoice_once(&pool, &numbered_invoice_id, "open", &actual_number).await; + + let ambiguous_invoice_id = format!("inv_{marker}_ambiguous"); + let ambiguous_number = format!("INV-{marker}-A"); + let ambiguous_a = invoice_event( + &format!("evt_{marker}_invoice_ambiguous_a"), + "invoice.payment_failed", + 1_700_009_000, + &ambiguous_invoice_id, + &customer_id, + Some(&ambiguous_number), + ); + let ambiguous_z = invoice_event( + &format!("evt_{marker}_invoice_ambiguous_z"), + "invoice.payment_failed", + 1_700_009_000, + &ambiguous_invoice_id, + &customer_id, + Some(&ambiguous_number), + ); + let authoritative_invoice = invoice_event( + &format!("evt_{marker}_invoice_snapshot"), + "invoice.paid", + 1_700_009_100, + &ambiguous_invoice_id, + &customer_id, + Some(&ambiguous_number), + ); + stripe_mock.objects.write().await.insert( + ambiguous_invoice_id.clone(), + authoritative_invoice.data.object, + ); + apply_test_event(&state, &ambiguous_a) + .await + .expect("apply first same-second invoice event"); + apply_test_event(&state, &ambiguous_z) + .await + .expect("reconcile same-second invoice ambiguity"); + assert_invoice_once(&pool, &ambiguous_invoice_id, "paid", &ambiguous_number).await; + + let unmapped_invoice_id = format!("inv_{marker}_unmapped"); + let unmapped_invoice = invoice_event( + &format!("evt_{marker}_invoice_unmapped"), + "invoice.paid", + 1_700_010_000, + &unmapped_invoice_id, + &format!("cus_{marker}_invoice_unmapped"), + Some(&format!("INV-{marker}-U")), + ); + let unmapped_invoice_error = apply_test_event(&state, &unmapped_invoice) + .await + .expect_err("unmapped invoice was marked processed"); + assert_eq!(unmapped_invoice_error.code, ErrorCode::StorageUnavailable); + let unmapped_invoice_rows: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM invoices WHERE provider = 'stripe' AND provider_invoice_id = $1", + ) + .bind(&unmapped_invoice_id) + .fetch_one(&pool) + .await + .expect("count unmapped invoices"); + assert_eq!(unmapped_invoice_rows, 0); + let unmapped_invoice_watermarks: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM provider_object_event_watermarks WHERE object_type = 'invoice' AND provider_object_id = $1", + ) + .bind(&unmapped_invoice_id) + .fetch_one(&pool) + .await + .expect("count unmapped invoice watermarks"); + assert_eq!(unmapped_invoice_watermarks, 0); + assert_eq!( stripe_mock.calls.load(Ordering::SeqCst), - 4, + 5, "unexpected number of authoritative Stripe snapshots" ); @@ -1743,6 +2087,13 @@ mod tests { .execute(&pool) .await .expect("delete test watermarks"); + sqlx::query( + "DELETE FROM provider_object_event_watermarks WHERE object_type = 'invoice' AND provider_object_id LIKE $1", + ) + .bind(format!("inv_{marker}%")) + .execute(&pool) + .await + .expect("delete test invoice watermarks"); sqlx::query("DELETE FROM users WHERE id = $1") .bind(user_id) .execute(&pool) @@ -1755,4 +2106,113 @@ mod tests { .expect("delete test plan"); stripe_mock_task.abort(); } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + #[ignore = "requires an isolated IMAGEFORGE_TEST_DATABASE_URL with CREATE DATABASE"] + async fn invoice_migration_rejects_ambiguous_duplicate_history() { + 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 mut admin_url = url::Url::parse(&database_url).expect("parse test database URL"); + admin_url.set_path("/postgres"); + let admin_pool = PgPoolOptions::new() + .max_connections(1) + .connect(admin_url.as_str()) + .await + .expect("connect PostgreSQL admin database"); + let marker = Uuid::new_v4().simple().to_string(); + let child_database = format!("imageforge_test_invoice_{marker}"); + sqlx::query(&format!("CREATE DATABASE {child_database}")) + .execute(&admin_pool) + .await + .expect("create invoice migration test database"); + + let mut child_url = url::Url::parse(&database_url).expect("parse child database URL"); + child_url.set_path(&format!("/{child_database}")); + let child_pool = PgPoolOptions::new() + .max_connections(2) + .connect(child_url.as_str()) + .await + .expect("connect invoice migration test database"); + let all_migrations = sqlx::migrate!(); + let legacy_migrator = sqlx::migrate::Migrator { + migrations: std::borrow::Cow::Owned( + all_migrations + .iter() + .filter(|migration| migration.version < 19) + .cloned() + .collect(), + ), + ignore_missing: false, + locking: true, + no_tx: false, + }; + legacy_migrator + .run(&child_pool) + .await + .expect("run migrations through 018"); + + let first_user = Uuid::new_v4(); + let second_user = Uuid::new_v4(); + sqlx::query( + r#" + INSERT INTO users (id, email, username, password_hash) + VALUES + ($1, $2, $3, 'test-only'), + ($4, $5, $6, 'test-only') + "#, + ) + .bind(first_user) + .bind(format!("invoice-migration-a-{marker}@example.test")) + .bind(format!("inv_a_{marker}")) + .bind(second_user) + .bind(format!("invoice-migration-b-{marker}@example.test")) + .bind(format!("inv_b_{marker}")) + .execute(&child_pool) + .await + .expect("insert duplicate-history users"); + let duplicate_invoice_id = format!("in_duplicate_{marker}"); + sqlx::query( + r#" + INSERT INTO invoices ( + user_id, invoice_number, status, provider, provider_invoice_id + ) VALUES + ($1, $2, 'open', 'stripe', $3), + ($4, $5, 'paid', 'stripe', $3) + "#, + ) + .bind(first_user) + .bind(format!("MIG-A-{marker}")) + .bind(&duplicate_invoice_id) + .bind(second_user) + .bind(format!("MIG-B-{marker}")) + .execute(&child_pool) + .await + .expect("insert duplicate invoice history"); + + let migration_result = all_migrations.run(&child_pool).await; + let index_exists: Option = + sqlx::query_scalar("SELECT to_regclass('idx_invoices_provider_object_unique')::text") + .fetch_one(&child_pool) + .await + .expect("query invoice unique index"); + child_pool.close().await; + sqlx::query(&format!("DROP DATABASE {child_database} WITH (FORCE)")) + .execute(&admin_pool) + .await + .expect("drop invoice migration test database"); + admin_pool.close().await; + + let migration_error = migration_result.expect_err("migration accepted duplicate invoices"); + assert!( + migration_error + .to_string() + .contains("duplicate invoices(provider, provider_invoice_id)"), + "unexpected migration error: {migration_error}" + ); + assert!(index_exists.is_none(), "unique index was partially applied"); + } }