237899745/ystp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: make Stripe invoice webhooks monotonic
Some checks failed
CI / verify (push) Has been cancelled
Details

Browse Source
This commit is contained in:
237899745
2026-07-26 04:53:21 +08:00
parent 3aefacec6b
commit 037e83e92f
5 changed files with 620 additions and 136 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
docs/billing.md
View File

@@ -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（脱敏）用于排查。

2
docs/database.md
View File

@@ -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

10
docs/deployment.md
View File

@@ -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;
```
推荐顺序：

16
migrations/019_invoice_webhook_invariants.sql Normal file
View File

@@ -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;

726
src/api/webhooks.rs
View File

@@ -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<WatermarkDecision, AppError> {
let rank = subscription_event_rank(&event.type_);
let inserted: Option<String> = sqlx::query_scalar(
r#"
INSERT INTO provider_object_event_watermarks (
provider, object_type, provider_object_id,
last_event_created, last_event_rank, last_event_id, is_deleted
) VALUES (
'stripe', 'subscription', $1,
$2, $3, $4, $5
'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<uuid::Uuid>,
}
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<StripeSubscriptionSnapshot, AppError> {
resource: &str,
provider_object_id: &str,
) -> Result<StripeObjectSnapshot, AppError> {
let secret = settings::get_stripe_secret(state)
.await
.map_err(|err| err.with_source("stripe secret not configured"))?;
let encoded_id = utf8_percent_encode(provider_subscription_id, NON_ALPHANUMERIC);
let 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<String>,
fallback_invoice_number: String,
status: String,
currency: String,
total_amount_cents: i32,
period_start: Option<chrono::DateTime<Utc>>,
period_end: Option<chrono::DateTime<Utc>>,
hosted_invoice_url: Option<String>,
pdf_url: Option<String>,
paid_at: Option<chrono::DateTime<Utc>>,
}
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<ResolvedInvoice, AppError> {
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<uuid::Uuid> =
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(&currency)
.bind(total_amount_cents)
.bind(hosted_invoice_url.as_deref())
.bind(pdf_url.as_deref())
.bind(period_start)
.bind(period_end)
.bind(paid_at)
.bind(provider_invoice_id)
.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(&currency)
.bind(total_amount_cents)
.bind(period_start)
.bind(period_end)
.bind(provider_invoice_id)
.bind(hosted_invoice_url.as_deref())
.bind(pdf_url.as_deref())
.bind(paid_at)
.execute(&mut **tx)
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "创建发票失败").with_source(err))?;
}
.map_err(|err| AppError::new(ErrorCode::Internal, "写入发票失败").with_source(err))?;
Ok(())
}
@@ -1153,10 +1275,10 @@ mod tests {
async fn stripe_snapshot(
State(mock): State<StripeSnapshotMock>,
Path(subscription_id): Path<String>,
Path(object_id): Path<String>,
) -> 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<chrono::DateTime<Utc>>)> = 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<String> =
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");
}
}