max / makenotwork

85.6 KB · 2528 lines History Blame Raw

1	//! Stripe webhook workflow tests, purchase, refund, account update,
2	//! invalid signature, subscription lifecycle.
3
4	use crate::harness::TestHarness;
5	use crate::harness::stripe::{TEST_WEBHOOK_SECRET, TEST_WEBHOOK_SECRET_V2, sign_webhook_payload};
6	use makenotwork::db::UserId;
7	use serde_json::Value;
8	use std::collections::HashMap;
9
10	/// Build a JSON event with the given type and object, sign it, and POST to /stripe/webhook.
11	async fn post_event_json(
12	h: &mut TestHarness,
13	event_type: &str,
14	object: serde_json::Value,
15	) -> crate::harness::client::TestResponse {
16	post_event_json_with_id(h, "evt_test_000", event_type, object).await
17	}
18
19	async fn post_event_json_with_id(
20	h: &mut TestHarness,
21	event_id: &str,
22	event_type: &str,
23	object: serde_json::Value,
24	) -> crate::harness::client::TestResponse {
25	let payload = serde_json::json!({
26	"id": event_id,
27	"type": event_type,
28	"data": {"object": object},
29	})
30	.to_string();
31	let signature = sign_webhook_payload(&payload, TEST_WEBHOOK_SECRET);
32
33	h.client
34	.request_with_headers(
35	"POST",
36	"/stripe/webhook",
37	Some(&payload),
38	&[
39	("stripe-signature", &signature),
40	("content-type", "application/json"),
41	],
42	)
43	.await
44	}
45
46	// Tests
47
48	#[tokio::test]
49	async fn webhook_invalid_signature() {
50	let mut h = TestHarness::with_stripe().await;
51
52	let payload = r#"{"id":"evt_bad","type":"account.updated","data":{"object":{}}}"#;
53	let bad_sig = "t=0,v1=00000000000000000000000000000000";
54
55	let resp = h
56	.client
57	.request_with_headers(
58	"POST",
59	"/stripe/webhook",
60	Some(payload),
61	&[
62	("stripe-signature", bad_sig),
63	("content-type", "application/json"),
64	],
65	)
66	.await;
67	assert_eq!(
68	resp.status.as_u16(),
69	400,
70	"Expected 400 for bad signature, got: {}",
71	resp.status
72	);
73	}
74
75	#[tokio::test]
76	async fn webhook_account_updated() {
77	let mut h = TestHarness::with_stripe().await;
78
79	// Create a user with a known stripe_account_id
80	let user_id = h
81	.signup("stripecreator", "sc@test.com", "password123")
82	.await;
83	let acct_id = "acct_test_wh_123";
84	sqlx::query("UPDATE users SET stripe_account_id = $1 WHERE id = $2")
85	.bind(acct_id)
86	.bind(user_id)
87	.execute(&h.db)
88	.await
89	.unwrap();
90
91	// Build account object with valid id prefix
92	let account = serde_json::json!({
93	"id": acct_id,
94	"object": "account",
95	"charges_enabled": true,
96	"payouts_enabled": true,
97	"details_submitted": true,
98	});
99
100	let resp = post_event_json(&mut h, "account.updated", account).await;
101	assert_eq!(resp.status.as_u16(), 200, "Webhook failed: {}", resp.text);
102
103	// Verify DB was updated
104	let (charges, payouts, onboarding): (bool, bool, bool) = sqlx::query_as(
105	"SELECT stripe_charges_enabled, stripe_payouts_enabled, stripe_onboarding_complete FROM users WHERE id = $1",
106	)
107	.bind(user_id)
108	.fetch_one(&h.db)
109	.await
110	.unwrap();
111
112	assert!(charges, "charges_enabled should be true");
113	assert!(payouts, "payouts_enabled should be true");
114	assert!(onboarding, "onboarding_complete should be true");
115	}
116
117	#[tokio::test]
118	async fn webhook_purchase_completed() {
119	let mut h = TestHarness::with_stripe().await;
120
121	// Create buyer + seller
122	let buyer_id = h.signup("buyer", "buyer@test.com", "password123").await;
123	h.client.post_form("/logout", "").await;
124	let seller_id = h.signup("seller", "seller@test.com", "password123").await;
125	h.grant_creator(seller_id).await;
126	h.client.post_form("/logout", "").await;
127	h.login("seller", "password123").await;
128
129	// Create project + item
130	let resp = h
131	.client
132	.post_form("/api/projects", "slug=stripeproj&title=Stripe+Project")
133	.await;
134	let project: Value = resp.json();
135	let project_id = project["id"].as_str().unwrap().to_string();
136	let resp = h
137	.client
138	.post_form(
139	&format!("/api/projects/{project_id}/items"),
140	"title=Paid+Track&price_cents=999&item_type=audio",
141	)
142	.await;
143	let item: Value = resp.json();
144	let item_id = item["id"].as_str().unwrap().to_string();
145
146	// Insert a pending transaction via direct SQL
147	let session_id = "cs_test_purchase_123";
148	sqlx::query(
149	r"INSERT INTO transactions
150	(buyer_id, seller_id, item_id, amount_cents, status,
151	stripe_checkout_session_id, item_title, seller_username)
152	VALUES ($1, $2, $3::uuid, 999, 'pending', $4, 'Paid Track', 'seller')",
153	)
154	.bind(buyer_id)
155	.bind(seller_id)
156	.bind(&item_id)
157	.bind(session_id)
158	.execute(&h.db)
159	.await
160	.unwrap();
161
162	// Build checkout session with valid IDs
163	let mut meta = HashMap::new();
164	meta.insert("buyer_id".to_string(), buyer_id.to_string());
165	meta.insert("seller_id".to_string(), seller_id.to_string());
166	meta.insert("item_id".to_string(), item_id.clone());
167	let session = serde_json::json!({
168	"id": session_id,
169	"object": "checkout_session",
170	"mode": "payment",
171	"metadata": meta,
172	"payment_intent": "pi_test_purchase_123",
173	});
174
175	let resp = post_event_json(&mut h, "checkout.session.completed", session).await;
176	assert_eq!(resp.status.as_u16(), 200, "Webhook failed: {}", resp.text);
177
178	// Verify transaction was completed
179	let status: String =
180	sqlx::query_scalar("SELECT status FROM transactions WHERE stripe_checkout_session_id = $1")
181	.bind(session_id)
182	.fetch_one(&h.db)
183	.await
184	.unwrap();
185	assert_eq!(status, "completed");
186
187	// Verify sales_count was incremented
188	let sales: i32 = sqlx::query_scalar("SELECT sales_count FROM items WHERE id = $1::uuid")
189	.bind(&item_id)
190	.fetch_one(&h.db)
191	.await
192	.unwrap();
193	assert_eq!(sales, 1);
194	}
195
196	#[tokio::test]
197	async fn webhook_charge_refunded() {
198	let mut h = TestHarness::with_stripe().await;
199
200	// Create buyer + seller + item
201	let buyer_id = h.signup("rbuyer", "rb@test.com", "password123").await;
202	h.client.post_form("/logout", "").await;
203	let seller_id = h.signup("rseller", "rs@test.com", "password123").await;
204	h.grant_creator(seller_id).await;
205	h.client.post_form("/logout", "").await;
206	h.login("rseller", "password123").await;
207
208	let resp = h
209	.client
210	.post_form("/api/projects", "slug=refundproj&title=Refund+Project")
211	.await;
212	let project: Value = resp.json();
213	let project_id = project["id"].as_str().unwrap().to_string();
214	let resp = h
215	.client
216	.post_form(
217	&format!("/api/projects/{project_id}/items"),
218	"title=Refund+Track&price_cents=500&item_type=audio",
219	)
220	.await;
221	let item: Value = resp.json();
222	let item_id = item["id"].as_str().unwrap().to_string();
223
224	// Set sales_count to 1 and insert a completed transaction
225	let pi_id = "pi_test_refund_123";
226	sqlx::query("UPDATE items SET sales_count = 1 WHERE id = $1::uuid")
227	.bind(&item_id)
228	.execute(&h.db)
229	.await
230	.unwrap();
231	sqlx::query(
232	r"INSERT INTO transactions
233	(buyer_id, seller_id, item_id, amount_cents, status,
234	stripe_payment_intent_id, stripe_checkout_session_id,
235	item_title, seller_username, completed_at)
236	VALUES ($1, $2, $3::uuid, 500, 'completed', $4, 'cs_refund', 'Refund Track', 'rseller', NOW())",
237	)
238	.bind(buyer_id)
239	.bind(seller_id)
240	.bind(&item_id)
241	.bind(pi_id)
242	.execute(&h.db)
243	.await
244	.unwrap();
245
246	// Build charge with valid id and payment_intent
247	let charge = serde_json::json!({
248	"id": "ch_test_refund",
249	"object": "charge",
250	"amount": 500,
251	"amount_refunded": 500,
252	"payment_intent": "pi_test_refund_123",
253	});
254
255	let resp = post_event_json(&mut h, "charge.refunded", charge).await;
256	assert_eq!(resp.status.as_u16(), 200, "Webhook failed: {}", resp.text);
257
258	// Verify transaction was refunded
259	let status: String =
260	sqlx::query_scalar("SELECT status FROM transactions WHERE stripe_payment_intent_id = $1")
261	.bind(pi_id)
262	.fetch_one(&h.db)
263	.await
264	.unwrap();
265	assert_eq!(status, "refunded");
266
267	// Verify sales_count was decremented
268	let sales: i32 = sqlx::query_scalar("SELECT sales_count FROM items WHERE id = $1::uuid")
269	.bind(&item_id)
270	.fetch_one(&h.db)
271	.await
272	.unwrap();
273	assert_eq!(sales, 0);
274	}
275
276	#[tokio::test]
277	async fn webhook_subscription_deleted() {
278	let mut h = TestHarness::with_stripe().await;
279
280	// Create subscriber and creator with project + tier
281	let sub_user_id = h.signup("subscriber", "sub@test.com", "password123").await;
282	h.client.post_form("/logout", "").await;
283	let creator_id = h.signup("tiercreator", "tc@test.com", "password123").await;
284	h.grant_creator(creator_id).await;
285	h.client.post_form("/logout", "").await;
286	h.login("tiercreator", "password123").await;
287
288	let resp = h
289	.client
290	.post_form("/api/projects", "slug=subproj&title=Sub+Project")
291	.await;
292	let project: Value = resp.json();
293	let project_id = project["id"].as_str().unwrap().to_string();
294
295	// Create subscription tier via direct SQL
296	let tier_id = uuid::Uuid::new_v4();
297	sqlx::query(
298	r"INSERT INTO subscription_tiers
299	(id, project_id, name, price_cents)
300	VALUES ($1, $2::uuid, 'Basic', 500)",
301	)
302	.bind(tier_id)
303	.bind(&project_id)
304	.execute(&h.db)
305	.await
306	.unwrap();
307
308	// Create subscription via direct SQL
309	let stripe_sub_id = "sub_test_delete_123";
310	sqlx::query(
311	r"INSERT INTO subscriptions
312	(subscriber_id, tier_id, project_id, stripe_subscription_id, stripe_customer_id, status)
313	VALUES ($1, $2, $3::uuid, $4, 'cus_test', 'active')",
314	)
315	.bind(sub_user_id)
316	.bind(tier_id)
317	.bind(&project_id)
318	.bind(stripe_sub_id)
319	.execute(&h.db)
320	.await
321	.unwrap();
322
323	let sub = serde_json::json!({
324	"id": stripe_sub_id,
325	"object": "subscription",
326	"status": "canceled",
327	"cancel_at_period_end": false,
328	"items": {
329	"object": "list",
330	"data": [{
331	"id": "si_test_del",
332	"object": "subscription_item",
333	"subscription": stripe_sub_id,
334	"current_period_start": 1_700_000_000,
335	"current_period_end": 1_702_592_000,
336	"metadata": {},
337	}],
338	},
339	});
340
341	let resp = post_event_json(&mut h, "customer.subscription.deleted", sub).await;
342	assert_eq!(resp.status.as_u16(), 200, "Webhook failed: {}", resp.text);
343
344	// Verify subscription was canceled
345	let status: String =
346	sqlx::query_scalar("SELECT status FROM subscriptions WHERE stripe_subscription_id = $1")
347	.bind(stripe_sub_id)
348	.fetch_one(&h.db)
349	.await
350	.unwrap();
351	assert_eq!(status, "canceled");
352	}
353
354	// Shared fixture for subscription webhook tests
355
356	#[allow(
357	clippy::struct_field_names,
358	reason = "every field names a distinct entity id; the shared _id suffix is the meaningful convention, not noise"
359	)]
360	struct SubscriptionFixture {
361	#[allow(dead_code)]
362	creator_id: UserId,
363	subscriber_id: UserId,
364	project_id: String,
365	tier_id: uuid::Uuid,
366	}
367
368	/// Creates a creator (with project + tier) and a subscriber user.
369	/// Leaves the harness logged out.
370	async fn setup_subscription_fixture(h: &mut TestHarness) -> SubscriptionFixture {
371	let subscriber_id = h.signup("subuser", "subuser@test.com", "password123").await;
372	h.client.post_form("/logout", "").await;
373	let creator_id = h.signup("creator", "creator@test.com", "password123").await;
374	h.grant_creator(creator_id).await;
375	h.client.post_form("/logout", "").await;
376	h.login("creator", "password123").await;
377
378	let resp = h
379	.client
380	.post_form("/api/projects", "slug=subfix&title=Sub+Fixture")
381	.await;
382	let project: Value = resp.json();
383	let project_id = project["id"].as_str().unwrap().to_string();
384
385	let tier_id = uuid::Uuid::new_v4();
386	sqlx::query(
387	r"INSERT INTO subscription_tiers (id, project_id, name, price_cents)
388	VALUES ($1, $2::uuid, 'Pro', 1000)",
389	)
390	.bind(tier_id)
391	.bind(&project_id)
392	.execute(&h.db)
393	.await
394	.unwrap();
395
396	h.client.post_form("/logout", "").await;
397
398	SubscriptionFixture {
399	creator_id,
400	subscriber_id,
401	project_id,
402	tier_id,
403	}
404	}
405
406	/// Insert an active subscription row into the DB. Returns the stripe subscription ID.
407	async fn insert_active_subscription(
408	h: &TestHarness,
409	fix: &SubscriptionFixture,
410	stripe_sub_id: &str,
411	) {
412	sqlx::query(
413	r"INSERT INTO subscriptions
414	(subscriber_id, tier_id, project_id, stripe_subscription_id, stripe_customer_id, status)
415	VALUES ($1, $2, $3::uuid, $4, 'cus_test_fixture', 'active')",
416	)
417	.bind(fix.subscriber_id)
418	.bind(fix.tier_id)
419	.bind(&fix.project_id)
420	.bind(stripe_sub_id)
421	.execute(&h.db)
422	.await
423	.unwrap();
424	}
425
426	/// Insert an active creator-tier subscription row and sync the denormalized
427	/// `users.creator_tier` column to match. Returns nothing; the caller owns the ids.
428	async fn insert_active_creator_sub(
429	h: &TestHarness,
430	user_id: UserId,
431	stripe_sub_id: &str,
432	tier: &str,
433	) {
434	sqlx::query(
435	r"INSERT INTO creator_subscriptions
436	(user_id, stripe_subscription_id, stripe_customer_id, tier, status)
437	VALUES ($1, $2, 'cus_ct_fixture', $3, 'active')",
438	)
439	.bind(user_id)
440	.bind(stripe_sub_id)
441	.bind(tier)
442	.execute(&h.db)
443	.await
444	.unwrap();
445	sqlx::query("UPDATE users SET creator_tier = $2 WHERE id = $1")
446	.bind(user_id)
447	.bind(tier)
448	.execute(&h.db)
449	.await
450	.unwrap();
451	}
452
453	// New tests
454
455	#[tokio::test]
456	async fn webhook_subscription_checkout_completed() {
457	let mut h = TestHarness::with_stripe().await;
458	let fix = setup_subscription_fixture(&mut h).await;
459
460	let stripe_sub_id = "sub_test_checkout_001";
461	let stripe_customer_id = "cus_test_checkout_001";
462
463	// Build checkout session with subscription metadata
464	let mut meta = HashMap::new();
465	meta.insert("checkout_type".to_string(), "subscription".to_string());
466	meta.insert("subscriber_id".to_string(), fix.subscriber_id.to_string());
467	meta.insert("project_id".to_string(), fix.project_id.clone());
468	meta.insert("tier_id".to_string(), fix.tier_id.to_string());
469	let session = serde_json::json!({
470	"id": "cs_test_sub_checkout_001",
471	"object": "checkout_session",
472	"mode": "subscription",
473	"metadata": meta,
474	"subscription": stripe_sub_id,
475	"customer": stripe_customer_id,
476	});
477
478	let resp = post_event_json(&mut h, "checkout.session.completed", session).await;
479	assert_eq!(resp.status.as_u16(), 200, "Webhook failed: {}", resp.text);
480
481	// Verify subscription row was created
482	let (status, sub_stripe_id, sub_customer_id): (String, String, String) = sqlx::query_as(
483	"SELECT status, stripe_subscription_id, stripe_customer_id FROM subscriptions WHERE stripe_subscription_id = $1",
484	)
485	.bind(stripe_sub_id)
486	.fetch_one(&h.db)
487	.await
488	.unwrap();
489
490	assert_eq!(status, "active");
491	assert_eq!(sub_stripe_id, stripe_sub_id);
492	assert_eq!(sub_customer_id, stripe_customer_id);
493	}
494
495	#[tokio::test]
496	async fn webhook_subscription_checkout_completed_idempotent() {
497	let mut h = TestHarness::with_stripe().await;
498	let fix = setup_subscription_fixture(&mut h).await;
499
500	let stripe_sub_id = "sub_test_checkout_idem";
501	let stripe_customer_id = "cus_test_checkout_idem";
502
503	let build_session = \|\| {
504	let mut meta = HashMap::new();
505	meta.insert("checkout_type".to_string(), "subscription".to_string());
506	meta.insert("subscriber_id".to_string(), fix.subscriber_id.to_string());
507	meta.insert("project_id".to_string(), fix.project_id.clone());
508	meta.insert("tier_id".to_string(), fix.tier_id.to_string());
509	serde_json::json!({
510	"id": "cs_test_sub_idem",
511	"object": "checkout_session",
512	"mode": "subscription",
513	"metadata": meta,
514	"subscription": stripe_sub_id,
515	"customer": stripe_customer_id,
516	})
517	};
518
519	// First event
520	let resp = post_event_json(&mut h, "checkout.session.completed", build_session()).await;
521	assert_eq!(
522	resp.status.as_u16(),
523	200,
524	"First webhook failed: {}",
525	resp.text
526	);
527
528	// Second event (duplicate), use a different event ID
529	let resp = post_event_json_with_id(
530	&mut h,
531	"evt_test_001",
532	"checkout.session.completed",
533	build_session(),
534	)
535	.await;
536	assert_eq!(
537	resp.status.as_u16(),
538	200,
539	"Duplicate webhook should succeed: {}",
540	resp.text
541	);
542
543	// Verify still only one subscription row
544	let count: i64 =
545	sqlx::query_scalar("SELECT COUNT(*) FROM subscriptions WHERE stripe_subscription_id = $1")
546	.bind(stripe_sub_id)
547	.fetch_one(&h.db)
548	.await
549	.unwrap();
550	assert_eq!(count, 1, "Should have exactly one subscription row");
551	}
552
553	#[tokio::test]
554	async fn webhook_subscription_updated() {
555	let mut h = TestHarness::with_stripe().await;
556	let fix = setup_subscription_fixture(&mut h).await;
557
558	let stripe_sub_id = "sub_test_updated_001";
559	insert_active_subscription(&h, &fix, stripe_sub_id).await;
560
561	let sub = serde_json::json!({
562	"id": stripe_sub_id,
563	"object": "subscription",
564	"status": "past_due",
565	"cancel_at_period_end": false,
566	"items": {
567	"object": "list",
568	"data": [{
569	"id": "si_test_upd",
570	"object": "subscription_item",
571	"subscription": stripe_sub_id,
572	"current_period_start": 1_702_592_000,
573	"current_period_end": 1_705_184_000,
574	"metadata": {},
575	}],
576	},
577	});
578
579	let resp = post_event_json(&mut h, "customer.subscription.updated", sub).await;
580	assert_eq!(resp.status.as_u16(), 200, "Webhook failed: {}", resp.text);
581
582	// Verify status changed
583	let status: String =
584	sqlx::query_scalar("SELECT status FROM subscriptions WHERE stripe_subscription_id = $1")
585	.bind(stripe_sub_id)
586	.fetch_one(&h.db)
587	.await
588	.unwrap();
589	assert_eq!(status, "past_due");
590
591	// Verify period was updated
592	let (period_start, period_end): (Option<chrono::DateTime<chrono::Utc>>, Option<chrono::DateTime<chrono::Utc>>) = sqlx::query_as(
593	"SELECT current_period_start, current_period_end FROM subscriptions WHERE stripe_subscription_id = $1",
594	)
595	.bind(stripe_sub_id)
596	.fetch_one(&h.db)
597	.await
598	.unwrap();
599	assert!(period_start.is_some(), "period_start should be set");
600	assert!(period_end.is_some(), "period_end should be set");
601	assert_eq!(period_start.unwrap().timestamp(), 1_702_592_000);
602	assert_eq!(period_end.unwrap().timestamp(), 1_705_184_000);
603	}
604
605	/// A thin `customer.subscription.updated` carrying no
606	/// period must NOT overwrite an active row's period with an epoch (1970) value.
607	/// The access gate requires `current_period_end > NOW()`, so an epoch write
608	/// would cut off a paying subscriber until the next invoice repaired the row.
609	/// The sealed writer drops a missing/zero period (COALESCE keeps the existing).
610	#[tokio::test]
611	async fn webhook_subscription_updated_thin_event_preserves_period() {
612	let mut h = TestHarness::with_stripe().await;
613	let fix = setup_subscription_fixture(&mut h).await;
614
615	let stripe_sub_id = "sub_test_thin_period";
616	let original_end: i64 = 1_893_456_000; // 2030-01-01, comfortably in the future
617	sqlx::query(
618	r"INSERT INTO subscriptions
619	(subscriber_id, tier_id, project_id, stripe_subscription_id, stripe_customer_id,
620	status, current_period_start, current_period_end)
621	VALUES ($1, $2, $3::uuid, $4, 'cus_test_fixture', 'active',
622	to_timestamp($5), to_timestamp($6))",
623	)
624	.bind(fix.subscriber_id)
625	.bind(fix.tier_id)
626	.bind(&fix.project_id)
627	.bind(stripe_sub_id)
628	.bind(1_700_000_000_i64)
629	.bind(original_end)
630	.execute(&h.db)
631	.await
632	.unwrap();
633
634	// Thin update: a status the row already has, and NO items/period at all.
635	let sub = serde_json::json!({
636	"id": stripe_sub_id,
637	"object": "subscription",
638	"status": "active",
639	"cancel_at_period_end": false,
640	"items": { "object": "list", "data": [] },
641	});
642	let resp = post_event_json(&mut h, "customer.subscription.updated", sub).await;
643	assert_eq!(resp.status.as_u16(), 200, "Webhook failed: {}", resp.text);
644
645	let period_end: Option<chrono::DateTime<chrono::Utc>> = sqlx::query_scalar(
646	"SELECT current_period_end FROM subscriptions WHERE stripe_subscription_id = $1",
647	)
648	.bind(stripe_sub_id)
649	.fetch_one(&h.db)
650	.await
651	.unwrap();
652	assert_eq!(
653	period_end.expect("period_end must remain set").timestamp(),
654	original_end,
655	"thin webhook must leave the period untouched, not stamp 1970"
656	);
657	}
658
659	#[tokio::test]
660	async fn webhook_invoice_payment_succeeded() {
661	let mut h = TestHarness::with_stripe().await;
662	let fix = setup_subscription_fixture(&mut h).await;
663
664	let stripe_sub_id = "sub_test_inv_success";
665	insert_active_subscription(&h, &fix, stripe_sub_id).await;
666
667	let invoice = serde_json::json!({
668	"id": "in_test_success_001",
669	"object": "invoice",
670	"subscription": stripe_sub_id,
671	"period_start": 1_702_592_000,
672	"period_end": 1_705_184_000,
673	"billing_reason": "subscription_cycle",
674	"livemode": false,
675	});
676
677	let resp = post_event_json(&mut h, "invoice.payment_succeeded", invoice).await;
678	assert_eq!(resp.status.as_u16(), 200, "Webhook failed: {}", resp.text);
679
680	// Verify subscription period was updated
681	let (period_start, period_end): (Option<chrono::DateTime<chrono::Utc>>, Option<chrono::DateTime<chrono::Utc>>) = sqlx::query_as(
682	"SELECT current_period_start, current_period_end FROM subscriptions WHERE stripe_subscription_id = $1",
683	)
684	.bind(stripe_sub_id)
685	.fetch_one(&h.db)
686	.await
687	.unwrap();
688	assert!(period_start.is_some(), "period_start should be set");
689	assert!(period_end.is_some(), "period_end should be set");
690	assert_eq!(period_start.unwrap().timestamp(), 1_702_592_000);
691	assert_eq!(period_end.unwrap().timestamp(), 1_705_184_000);
692	}
693
694	#[tokio::test]
695	async fn webhook_invoice_payment_failed() {
696	let mut h = TestHarness::with_stripe().await;
697	let fix = setup_subscription_fixture(&mut h).await;
698
699	let stripe_sub_id = "sub_test_inv_failed";
700	insert_active_subscription(&h, &fix, stripe_sub_id).await;
701
702	let invoice = serde_json::json!({
703	"id": "in_test_failed_001",
704	"object": "invoice",
705	"subscription": stripe_sub_id,
706	"period_start": 1_700_000_000,
707	"period_end": 1_702_592_000,
708	"livemode": false,
709	});
710
711	let resp = post_event_json(&mut h, "invoice.payment_failed", invoice).await;
712	assert_eq!(resp.status.as_u16(), 200, "Webhook failed: {}", resp.text);
713
714	// Verify status changed to past_due
715	let status: String =
716	sqlx::query_scalar("SELECT status FROM subscriptions WHERE stripe_subscription_id = $1")
717	.bind(stripe_sub_id)
718	.fetch_one(&h.db)
719	.await
720	.unwrap();
721	assert_eq!(status, "past_due");
722	}
723
724	#[tokio::test]
725	async fn webhook_account_updated_partial() {
726	let mut h = TestHarness::with_stripe().await;
727
728	let user_id = h
729	.signup("partialcreator", "pc@test.com", "password123")
730	.await;
731	let acct_id = "acct_test_partial_123";
732	sqlx::query("UPDATE users SET stripe_account_id = $1 WHERE id = $2")
733	.bind(acct_id)
734	.bind(user_id)
735	.execute(&h.db)
736	.await
737	.unwrap();
738
739	// Only details_submitted is true; charges and payouts still false
740	let account = serde_json::json!({
741	"id": acct_id,
742	"object": "account",
743	"charges_enabled": false,
744	"payouts_enabled": false,
745	"details_submitted": true,
746	});
747
748	let resp = post_event_json(&mut h, "account.updated", account).await;
749	assert_eq!(resp.status.as_u16(), 200, "Webhook failed: {}", resp.text);
750
751	let (charges, payouts, onboarding): (bool, bool, bool) = sqlx::query_as(
752	"SELECT stripe_charges_enabled, stripe_payouts_enabled, stripe_onboarding_complete FROM users WHERE id = $1",
753	)
754	.bind(user_id)
755	.fetch_one(&h.db)
756	.await
757	.unwrap();
758
759	assert!(!charges, "charges_enabled should be false");
760	assert!(!payouts, "payouts_enabled should be false");
761	assert!(onboarding, "onboarding_complete should be true");
762	}
763
764	#[tokio::test]
765	async fn webhook_account_updated_unknown_account() {
766	let mut h = TestHarness::with_stripe().await;
767
768	// No user has this stripe_account_id
769	let account = serde_json::json!({
770	"id": "acct_nonexistent",
771	"object": "account",
772	"charges_enabled": true,
773	"payouts_enabled": true,
774	"details_submitted": true,
775	});
776
777	let resp = post_event_json(&mut h, "account.updated", account).await;
778	assert_eq!(
779	resp.status.as_u16(),
780	200,
781	"Unknown account should still return 200: {}",
782	resp.text
783	);
784
785	// Verify no users were affected
786	let count: i64 = sqlx::query_scalar(
787	"SELECT COUNT(*) FROM users WHERE stripe_account_id = 'acct_nonexistent'",
788	)
789	.fetch_one(&h.db)
790	.await
791	.unwrap();
792	assert_eq!(count, 0);
793	}
794
795	#[tokio::test]
796	async fn webhook_purchase_completed_idempotent() {
797	let mut h = TestHarness::with_stripe().await;
798
799	// Create buyer + seller
800	let buyer_id = h.signup("idembuyer", "ib@test.com", "password123").await;
801	h.client.post_form("/logout", "").await;
802	let seller_id = h.signup("idemseller", "is@test.com", "password123").await;
803	h.grant_creator(seller_id).await;
804	h.client.post_form("/logout", "").await;
805	h.login("idemseller", "password123").await;
806
807	// Create project + item
808	let resp = h
809	.client
810	.post_form("/api/projects", "slug=idemproj&title=Idem+Project")
811	.await;
812	let project: Value = resp.json();
813	let project_id = project["id"].as_str().unwrap().to_string();
814	let resp = h
815	.client
816	.post_form(
817	&format!("/api/projects/{project_id}/items"),
818	"title=Idem+Track&price_cents=500&item_type=audio",
819	)
820	.await;
821	let item: Value = resp.json();
822	let item_id = item["id"].as_str().unwrap().to_string();
823
824	// Insert a pending transaction
825	let session_id = "cs_test_idem_001";
826	sqlx::query(
827	r"INSERT INTO transactions
828	(buyer_id, seller_id, item_id, amount_cents, status,
829	stripe_checkout_session_id, item_title, seller_username)
830	VALUES ($1, $2, $3::uuid, 500, 'pending', $4, 'Idem Track', 'idemseller')",
831	)
832	.bind(buyer_id)
833	.bind(seller_id)
834	.bind(&item_id)
835	.bind(session_id)
836	.execute(&h.db)
837	.await
838	.unwrap();
839
840	let build_session = \|\| {
841	let mut meta = HashMap::new();
842	meta.insert("buyer_id".to_string(), buyer_id.to_string());
843	meta.insert("seller_id".to_string(), seller_id.to_string());
844	meta.insert("item_id".to_string(), item_id.clone());
845	serde_json::json!({
846	"id": session_id,
847	"object": "checkout_session",
848	"mode": "payment",
849	"metadata": meta,
850	"payment_intent": "pi_test_idem_001",
851	})
852	};
853
854	// First event
855	let resp = post_event_json(&mut h, "checkout.session.completed", build_session()).await;
856	assert_eq!(
857	resp.status.as_u16(),
858	200,
859	"First webhook failed: {}",
860	resp.text
861	);
862
863	// Second event (duplicate)
864	let resp = post_event_json_with_id(
865	&mut h,
866	"evt_test_002",
867	"checkout.session.completed",
868	build_session(),
869	)
870	.await;
871	assert_eq!(
872	resp.status.as_u16(),
873	200,
874	"Duplicate webhook should succeed: {}",
875	resp.text
876	);
877
878	// Verify still one completed transaction
879	let count: i64 = sqlx::query_scalar(
880	"SELECT COUNT(*) FROM transactions WHERE stripe_checkout_session_id = $1 AND status = 'completed'",
881	)
882	.bind(session_id)
883	.fetch_one(&h.db)
884	.await
885	.unwrap();
886	assert_eq!(count, 1, "Should have exactly one completed transaction");
887
888	// Verify sales_count is still 1 (not incremented twice)
889	let sales: i32 = sqlx::query_scalar("SELECT sales_count FROM items WHERE id = $1::uuid")
890	.bind(&item_id)
891	.fetch_one(&h.db)
892	.await
893	.unwrap();
894	assert_eq!(sales, 1, "sales_count should be 1, not 2");
895	}
896
897	/// Dedup primitive regression: the "processed" row is written only AFTER a
898	/// handler succeeds, so the read/mark pair must compose correctly, an unseen
899	/// event reads false, a marked event reads true, and the mark is idempotent
900	/// (a redelivery or concurrent duplicate must not error). This pins the layer
901	/// the live handlers rely on (the `_idempotent` tests above use distinct event
902	/// IDs, so they only exercise the handler's SQL idempotency, not this one).
903	#[tokio::test]
904	async fn webhook_event_dedup_layering() {
905	use makenotwork::db::webhook_events;
906
907	let h = TestHarness::new().await;
908	let id = "evt_dedup_layer_001";
909
910	// Never seen → not processed.
911	assert!(
912	!webhook_events::is_event_processed(&h.db, id).await.unwrap(),
913	"an unseen event must read as not-processed"
914	);
915	// Mark after a (hypothetical) successful handler → now processed, so a
916	// redelivery short-circuits.
917	webhook_events::mark_event_processed(&h.db, id)
918	.await
919	.unwrap();
920	assert!(
921	webhook_events::is_event_processed(&h.db, id).await.unwrap(),
922	"a marked event must read as processed"
923	);
924	// Marking again (redelivery / concurrent duplicate) is idempotent, not an error.
925	webhook_events::mark_event_processed(&h.db, id)
926	.await
927	.unwrap();
928	assert!(
929	webhook_events::is_event_processed(&h.db, id).await.unwrap(),
930	"re-marking must stay processed and must not error"
931	);
932	}
933
934	// v2 thin event tests
935
936	/// Helper to POST a raw JSON payload to /stripe/webhook/v2 with a given signature.
937	async fn post_v2_raw(
938	h: &mut TestHarness,
939	payload: &str,
940	signature: &str,
941	) -> crate::harness::client::TestResponse {
942	h.client
943	.request_with_headers(
944	"POST",
945	"/stripe/webhook/v2",
946	Some(payload),
947	&[
948	("stripe-signature", signature),
949	("content-type", "application/json"),
950	],
951	)
952	.await
953	}
954
955	#[tokio::test]
956	async fn webhook_v2_invalid_signature() {
957	let mut h = TestHarness::with_stripe().await;
958
959	let payload = r#"{"id":"evt_v2_bad","type":"v2.core.account.updated","related_object":{"id":"acct_123","type":"account"}}"#;
960	let bad_sig = "t=0,v1=00000000000000000000000000000000";
961
962	let resp = post_v2_raw(&mut h, payload, bad_sig).await;
963	assert_eq!(
964	resp.status.as_u16(),
965	400,
966	"Expected 400 for bad v2 signature, got: {}",
967	resp.status
968	);
969	}
970
971	#[tokio::test]
972	async fn webhook_v2_account_event_accepted() {
973	let mut h = TestHarness::with_mocks().await;
974
975	// Uses mock Stripe so fetch_account returns success
976	let payload = serde_json::json!({
977	"id": "evt_v2_acct_001",
978	"type": "v2.core.account.updated",
979	"related_object": {
980	"id": "acct_test_v2_123",
981	"type": "account"
982	}
983	})
984	.to_string();
985
986	let signature = sign_webhook_payload(&payload, TEST_WEBHOOK_SECRET_V2);
987	let resp = post_v2_raw(&mut h, &payload, &signature).await;
988
989	assert_eq!(
990	resp.status.as_u16(),
991	200,
992	"v2 account event should return 200 even if API fetch fails: {}",
993	resp.text
994	);
995	}
996
997	#[tokio::test]
998	async fn webhook_v2_unknown_event_type_returns_200() {
999	let mut h = TestHarness::with_stripe().await;
1000
1001	let payload = serde_json::json!({
1002	"id": "evt_v2_unknown_001",
1003	"type": "v2.billing.meter.no_meter_found",
1004	"related_object": {
1005	"id": "mtr_123",
1006	"type": "billing.meter"
1007	}
1008	})
1009	.to_string();
1010
1011	let signature = sign_webhook_payload(&payload, TEST_WEBHOOK_SECRET_V2);
1012	let resp = post_v2_raw(&mut h, &payload, &signature).await;
1013
1014	assert_eq!(
1015	resp.status.as_u16(),
1016	200,
1017	"Unknown v2 event type should return 200: {}",
1018	resp.text
1019	);
1020	}
1021
1022	// Fan+ subscription webhook lifecycle
1023	//
1024	// Pins the earlier cascade branches in handle_subscription_updated /
1025	// handle_subscription_deleted / handle_invoice_payment_succeeded /
1026	// handle_invoice_payment_failed, previously only the generic creator-sub
1027	// fallback path was exercised.
1028
1029	fn make_subscription(stripe_sub_id: &str, status: &str) -> serde_json::Value {
1030	serde_json::json!({
1031	"id": stripe_sub_id,
1032	"object": "subscription",
1033	"status": status,
1034	"cancel_at_period_end": false,
1035	"items": {
1036	"object": "list",
1037	"data": [{
1038	"id": "si_fan_plus_test",
1039	"object": "subscription_item",
1040	"subscription": stripe_sub_id,
1041	"current_period_start": 1_700_000_000_i64,
1042	"current_period_end": 1_702_592_000_i64,
1043	"metadata": {},
1044	}],
1045	},
1046	})
1047	}
1048
1049	#[tokio::test]
1050	async fn webhook_subscription_updated_fan_plus_path() {
1051	let mut h = TestHarness::with_stripe().await;
1052	let user_id = h
1053	.signup("fpupdate", "fpupdate@test.com", "password123")
1054	.await;
1055
1056	let stripe_sub_id = "sub_fp_update_1";
1057	sqlx::query(
1058	r"INSERT INTO fan_plus_subscriptions
1059	(user_id, stripe_subscription_id, stripe_customer_id, status, current_period_end)
1060	VALUES ($1, $2, 'cus_fp_update', 'active', NOW() + interval '30 days')",
1061	)
1062	.bind(user_id)
1063	.bind(stripe_sub_id)
1064	.execute(&h.db)
1065	.await
1066	.unwrap();
1067
1068	let sub = make_subscription(stripe_sub_id, "past_due");
1069	let resp = post_event_json(&mut h, "customer.subscription.updated", sub).await;
1070	assert_eq!(resp.status.as_u16(), 200, "Webhook failed: {}", resp.text);
1071
1072	// fan_plus row reflects the new status, pins that the fan_plus branch
1073	// ran and reached `update_fan_plus_status`, not the generic fallback.
1074	let status: String = sqlx::query_scalar(
1075	"SELECT status::text FROM fan_plus_subscriptions WHERE stripe_subscription_id = $1",
1076	)
1077	.bind(stripe_sub_id)
1078	.fetch_one(&h.db)
1079	.await
1080	.unwrap();
1081	assert_eq!(status, "past_due");
1082	}
1083
1084	#[tokio::test]
1085	async fn webhook_subscription_deleted_fan_plus_path() {
1086	let mut h = TestHarness::with_stripe().await;
1087	let user_id = h
1088	.signup("fpdelete", "fpdelete@test.com", "password123")
1089	.await;
1090
1091	let stripe_sub_id = "sub_fp_delete_1";
1092	sqlx::query(
1093	r"INSERT INTO fan_plus_subscriptions
1094	(user_id, stripe_subscription_id, stripe_customer_id, status, current_period_end)
1095	VALUES ($1, $2, 'cus_fp_delete', 'active', NOW() + interval '30 days')",
1096	)
1097	.bind(user_id)
1098	.bind(stripe_sub_id)
1099	.execute(&h.db)
1100	.await
1101	.unwrap();
1102
1103	let sub = make_subscription(stripe_sub_id, "canceled");
1104	let resp = post_event_json(&mut h, "customer.subscription.deleted", sub).await;
1105	assert_eq!(resp.status.as_u16(), 200, "Webhook failed: {}", resp.text);
1106
1107	// Pins that `cancel_fan_plus` ran. We don't pin the exact column the
1108	// cancellation writes to (status vs separate canceled_at); just that
1109	// a downstream lookup classifies this user as NOT active.
1110	let active: bool = sqlx::query_scalar(
1111	"SELECT EXISTS(SELECT 1 FROM fan_plus_subscriptions \
1112	WHERE user_id = $1 AND status = 'active' AND canceled_at IS NULL)",
1113	)
1114	.bind(user_id)
1115	.fetch_one(&h.db)
1116	.await
1117	.unwrap();
1118	assert!(
1119	!active,
1120	"Fan+ subscription must not be active after cancellation"
1121	);
1122	}
1123
1124	fn make_invoice(stripe_sub_id: &str, billing_reason: &str) -> serde_json::Value {
1125	serde_json::json!({
1126	"id": "in_test_fp",
1127	"object": "invoice",
1128	"subscription": stripe_sub_id,
1129	"billing_reason": billing_reason,
1130	"period_start": 1_700_000_000_i64,
1131	"period_end": 1_702_592_000_i64,
1132	"currency": "usd",
1133	"livemode": false,
1134	})
1135	}
1136
1137	#[tokio::test]
1138	async fn webhook_invoice_payment_succeeded_updates_fan_plus_period() {
1139	let mut h = TestHarness::with_stripe().await;
1140	let user_id = h
1141	.signup("fpinvoice", "fpinvoice@test.com", "password123")
1142	.await;
1143
1144	let stripe_sub_id = "sub_fp_invoice_1";
1145	sqlx::query(
1146	r"INSERT INTO fan_plus_subscriptions
1147	(user_id, stripe_subscription_id, stripe_customer_id, status, current_period_end)
1148	VALUES ($1, $2, 'cus_fp_invoice', 'active', NOW())",
1149	)
1150	.bind(user_id)
1151	.bind(stripe_sub_id)
1152	.execute(&h.db)
1153	.await
1154	.unwrap();
1155
1156	// billing_reason != "subscription_cycle" → not a renewal, just updates period.
1157	let invoice = make_invoice(stripe_sub_id, "subscription_create");
1158	let resp = post_event_json(&mut h, "invoice.payment_succeeded", invoice).await;
1159	assert_eq!(resp.status.as_u16(), 200, "Webhook failed: {}", resp.text);
1160
1161	// Period_end should now match the invoice's period_end (2024-12-14T22:13:20Z = 1_702_592_000).
1162	let period_end: chrono::DateTime<chrono::Utc> = sqlx::query_scalar(
1163	"SELECT current_period_end FROM fan_plus_subscriptions WHERE stripe_subscription_id = $1",
1164	)
1165	.bind(stripe_sub_id)
1166	.fetch_one(&h.db)
1167	.await
1168	.unwrap();
1169	assert_eq!(period_end.timestamp(), 1_702_592_000);
1170	}
1171
1172	#[tokio::test]
1173	async fn webhook_invoice_payment_failed_sets_fan_plus_past_due() {
1174	let mut h = TestHarness::with_stripe().await;
1175	let user_id = h.signup("fpfail", "fpfail@test.com", "password123").await;
1176
1177	let stripe_sub_id = "sub_fp_fail_1";
1178	sqlx::query(
1179	r"INSERT INTO fan_plus_subscriptions
1180	(user_id, stripe_subscription_id, stripe_customer_id, status, current_period_end)
1181	VALUES ($1, $2, 'cus_fp_fail', 'active', NOW() + interval '30 days')",
1182	)
1183	.bind(user_id)
1184	.bind(stripe_sub_id)
1185	.execute(&h.db)
1186	.await
1187	.unwrap();
1188
1189	let invoice = make_invoice(stripe_sub_id, "subscription_cycle");
1190	let resp = post_event_json(&mut h, "invoice.payment_failed", invoice).await;
1191	assert_eq!(resp.status.as_u16(), 200, "Webhook failed: {}", resp.text);
1192
1193	let status: String = sqlx::query_scalar(
1194	"SELECT status::text FROM fan_plus_subscriptions WHERE stripe_subscription_id = $1",
1195	)
1196	.bind(stripe_sub_id)
1197	.fetch_one(&h.db)
1198	.await
1199	.unwrap();
1200	assert_eq!(
1201	status, "past_due",
1202	"Fan+ must be flipped to past_due on payment failure"
1203	);
1204	}
1205
1206	#[tokio::test]
1207	async fn webhook_subscription_updated_unknown_id_returns_200() {
1208	// Pins the fall-through: an event for a stripe_sub_id that has no fan_plus,
1209	// creator_tier, or app_sync row should still return 200 (no-op).
1210	let mut h = TestHarness::with_stripe().await;
1211	let sub = make_subscription("sub_does_not_exist", "active");
1212	let resp = post_event_json(&mut h, "customer.subscription.updated", sub).await;
1213	// Generic path runs `update_subscription_status` which finds nothing,
1214	// current behavior is to return 200 (idempotent / unknown-sub tolerance).
1215	assert_eq!(
1216	resp.status.as_u16(),
1217	200,
1218	"Unknown sub_id should not error: {}",
1219	resp.text
1220	);
1221	}
1222
1223	// Creator-tier subscription webhook lifecycle (test-fuzz Phase 2.1)
1224	//
1225	// The platform's own revenue ($16-60/mo). Before this block, the creator_tier
1226	// branch in every billing/subscription handler was untested, only fan_plus and
1227	// the generic creator-sub fallback were exercised. These pin that the
1228	// creator_tier branch runs AND keeps the denormalized `users.creator_tier`
1229	// column in sync (sync_user_creator_tier sets it to the active sub's tier, or
1230	// NULL when no active sub remains).
1231
1232	#[tokio::test]
1233	async fn webhook_creator_tier_checkout_completed() {
1234	let mut h = TestHarness::with_stripe().await;
1235	let user_id = h
1236	.signup("ctcheckout", "ctcheckout@test.com", "password123")
1237	.await;
1238
1239	let stripe_sub_id = "sub_ct_checkout_1";
1240	let mut meta = HashMap::new();
1241	meta.insert("checkout_type".to_string(), "creator_tier".to_string());
1242	meta.insert("user_id".to_string(), user_id.to_string());
1243	meta.insert("tier".to_string(), "small_files".to_string());
1244	let session = serde_json::json!({
1245	"id": "cs_ct_checkout_1",
1246	"object": "checkout_session",
1247	"mode": "subscription",
1248	"metadata": meta,
1249	"subscription": stripe_sub_id,
1250	"customer": "cus_ct_checkout_1",
1251	});
1252
1253	let resp = post_event_json(&mut h, "checkout.session.completed", session).await;
1254	assert_eq!(resp.status.as_u16(), 200, "Webhook failed: {}", resp.text);
1255
1256	// Subscription row created with the right tier + active status.
1257	let (status, tier): (String, String) = sqlx::query_as(
1258	"SELECT status, tier FROM creator_subscriptions WHERE stripe_subscription_id = $1",
1259	)
1260	.bind(stripe_sub_id)
1261	.fetch_one(&h.db)
1262	.await
1263	.unwrap();
1264	assert_eq!(status, "active");
1265	assert_eq!(tier, "small_files");
1266
1267	// Denormalized column synced on the user.
1268	let user_tier: Option<String> =
1269	sqlx::query_scalar("SELECT creator_tier FROM users WHERE id = $1")
1270	.bind(user_id)
1271	.fetch_one(&h.db)
1272	.await
1273	.unwrap();
1274	assert_eq!(
1275	user_tier.as_deref(),
1276	Some("small_files"),
1277	"users.creator_tier must be synced"
1278	);
1279	}
1280
1281	#[tokio::test]
1282	async fn webhook_invoice_payment_succeeded_updates_creator_tier_period() {
1283	let mut h = TestHarness::with_stripe().await;
1284	let user_id = h
1285	.signup("ctinvoice", "ctinvoice@test.com", "password123")
1286	.await;
1287
1288	let stripe_sub_id = "sub_ct_invoice_1";
1289	insert_active_creator_sub(&h, user_id, stripe_sub_id, "big_files").await;
1290
1291	let invoice = make_invoice(stripe_sub_id, "subscription_cycle");
1292	let resp = post_event_json(&mut h, "invoice.payment_succeeded", invoice).await;
1293	assert_eq!(resp.status.as_u16(), 200, "Webhook failed: {}", resp.text);
1294
1295	let (period_start, period_end): (
1296	Option<chrono::DateTime<chrono::Utc>>,
1297	Option<chrono::DateTime<chrono::Utc>>,
1298	) = sqlx::query_as(
1299	"SELECT current_period_start, current_period_end FROM creator_subscriptions WHERE stripe_subscription_id = $1",
1300	)
1301	.bind(stripe_sub_id)
1302	.fetch_one(&h.db)
1303	.await
1304	.unwrap();
1305	assert_eq!(period_start.unwrap().timestamp(), 1_700_000_000);
1306	assert_eq!(period_end.unwrap().timestamp(), 1_702_592_000);
1307
1308	// Tier remains live after a successful renewal.
1309	let user_tier: Option<String> =
1310	sqlx::query_scalar("SELECT creator_tier FROM users WHERE id = $1")
1311	.bind(user_id)
1312	.fetch_one(&h.db)
1313	.await
1314	.unwrap();
1315	assert_eq!(user_tier.as_deref(), Some("big_files"));
1316	}
1317
1318	#[tokio::test]
1319	async fn webhook_invoice_payment_failed_sets_creator_tier_past_due() {
1320	let mut h = TestHarness::with_stripe().await;
1321	let user_id = h.signup("ctfail", "ctfail@test.com", "password123").await;
1322
1323	let stripe_sub_id = "sub_ct_fail_1";
1324	insert_active_creator_sub(&h, user_id, stripe_sub_id, "everything").await;
1325
1326	let invoice = make_invoice(stripe_sub_id, "subscription_cycle");
1327	let resp = post_event_json(&mut h, "invoice.payment_failed", invoice).await;
1328	assert_eq!(resp.status.as_u16(), 200, "Webhook failed: {}", resp.text);
1329
1330	let status: String = sqlx::query_scalar(
1331	"SELECT status FROM creator_subscriptions WHERE stripe_subscription_id = $1",
1332	)
1333	.bind(stripe_sub_id)
1334	.fetch_one(&h.db)
1335	.await
1336	.unwrap();
1337	assert_eq!(status, "past_due");
1338
1339	// A past_due creator sub is no longer active, so the denormalized tier is cleared,
1340	// this is the gate that downstream enforcement reads to start the grace clock.
1341	let user_tier: Option<String> =
1342	sqlx::query_scalar("SELECT creator_tier FROM users WHERE id = $1")
1343	.bind(user_id)
1344	.fetch_one(&h.db)
1345	.await
1346	.unwrap();
1347	assert_eq!(
1348	user_tier, None,
1349	"users.creator_tier must clear when the sub goes past_due"
1350	);
1351	}
1352
1353	#[tokio::test]
1354	async fn webhook_subscription_updated_creator_tier_path() {
1355	let mut h = TestHarness::with_stripe().await;
1356	let user_id = h.signup("ctupd", "ctupd@test.com", "password123").await;
1357
1358	let stripe_sub_id = "sub_ct_upd_1";
1359	insert_active_creator_sub(&h, user_id, stripe_sub_id, "small_files").await;
1360
1361	let sub = make_subscription(stripe_sub_id, "past_due");
1362	let resp = post_event_json(&mut h, "customer.subscription.updated", sub).await;
1363	assert_eq!(resp.status.as_u16(), 200, "Webhook failed: {}", resp.text);
1364
1365	let (status, period_end): (String, Option<chrono::DateTime<chrono::Utc>>) = sqlx::query_as(
1366	"SELECT status, current_period_end FROM creator_subscriptions WHERE stripe_subscription_id = $1",
1367	)
1368	.bind(stripe_sub_id)
1369	.fetch_one(&h.db)
1370	.await
1371	.unwrap();
1372	assert_eq!(status, "past_due");
1373	assert_eq!(
1374	period_end.unwrap().timestamp(),
1375	1_702_592_000,
1376	"period must be updated from the event"
1377	);
1378
1379	let user_tier: Option<String> =
1380	sqlx::query_scalar("SELECT creator_tier FROM users WHERE id = $1")
1381	.bind(user_id)
1382	.fetch_one(&h.db)
1383	.await
1384	.unwrap();
1385	assert_eq!(
1386	user_tier, None,
1387	"tier sync must run on the creator_tier update branch"
1388	);
1389	}
1390
1391	#[tokio::test]
1392	async fn webhook_subscription_deleted_creator_tier_path() {
1393	let mut h = TestHarness::with_stripe().await;
1394	let user_id = h.signup("ctdel", "ctdel@test.com", "password123").await;
1395
1396	let stripe_sub_id = "sub_ct_del_1";
1397	insert_active_creator_sub(&h, user_id, stripe_sub_id, "everything").await;
1398
1399	let sub = make_subscription(stripe_sub_id, "canceled");
1400	let resp = post_event_json(&mut h, "customer.subscription.deleted", sub).await;
1401	assert_eq!(resp.status.as_u16(), 200, "Webhook failed: {}", resp.text);
1402
1403	let status: String = sqlx::query_scalar(
1404	"SELECT status FROM creator_subscriptions WHERE stripe_subscription_id = $1",
1405	)
1406	.bind(stripe_sub_id)
1407	.fetch_one(&h.db)
1408	.await
1409	.unwrap();
1410	assert_eq!(status, "canceled");
1411
1412	let user_tier: Option<String> =
1413	sqlx::query_scalar("SELECT creator_tier FROM users WHERE id = $1")
1414	.bind(user_id)
1415	.fetch_one(&h.db)
1416	.await
1417	.unwrap();
1418	assert_eq!(
1419	user_tier, None,
1420	"users.creator_tier must clear on cancellation"
1421	);
1422	}
1423
1424	// Fan+ renewal credit code (test-fuzz Phase 2.1)
1425	//
1426	// The earlier fan+ invoice test deliberately uses billing_reason=subscription_create
1427	// to avoid the renewal branch. THIS pins the renewal branch: a subscription_cycle
1428	// invoice generates the $5 single-use platform promo code that funds the Fan+
1429	// monthly credit. That code is a real DB write nothing previously exercised.
1430
1431	#[tokio::test]
1432	async fn webhook_invoice_payment_succeeded_fan_plus_renewal_generates_credit() {
1433	let mut h = TestHarness::with_stripe().await;
1434	let user_id = h
1435	.signup("fpcredit", "fpcredit@test.com", "password123")
1436	.await;
1437
1438	let stripe_sub_id = "sub_fp_credit_1";
1439	sqlx::query(
1440	r"INSERT INTO fan_plus_subscriptions
1441	(user_id, stripe_subscription_id, stripe_customer_id, status, current_period_end)
1442	VALUES ($1, $2, 'cus_fp_credit', 'active', NOW())",
1443	)
1444	.bind(user_id)
1445	.bind(stripe_sub_id)
1446	.execute(&h.db)
1447	.await
1448	.unwrap();
1449
1450	// subscription_cycle == a renewal, which triggers the credit-code path.
1451	let invoice = make_invoice(stripe_sub_id, "subscription_cycle");
1452	let resp = post_event_json(&mut h, "invoice.payment_succeeded", invoice).await;
1453	assert_eq!(resp.status.as_u16(), 200, "Webhook failed: {}", resp.text);
1454
1455	// Exactly one $5 single-use fixed-discount promo code now exists for this user.
1456	let (count, discount_type, discount_value, max_uses): (
1457	i64,
1458	Option<String>,
1459	Option<i32>,
1460	Option<i32>,
1461	) = sqlx::query_as(
1462	"SELECT COUNT(*), MIN(discount_type), MIN(discount_value), MIN(max_uses) \
1463	FROM promo_codes WHERE creator_id = $1 AND code_purpose = 'discount'",
1464	)
1465	.bind(user_id)
1466	.fetch_one(&h.db)
1467	.await
1468	.unwrap();
1469	assert_eq!(count, 1, "renewal must mint exactly one Fan+ credit code");
1470	assert_eq!(discount_type.as_deref(), Some("fixed"));
1471	assert_eq!(discount_value, Some(500), "$5 credit");
1472	assert_eq!(max_uses, Some(1), "single use");
1473	}
1474
1475	// Generic creator-sub renewal email gating (test-fuzz Phase 2.1)
1476	//
1477	// The generic invoice.payment_succeeded test only asserts the period write. The
1478	// renewal-email side effect, sent on a renewal, suppressed on the first invoice
1479	//, was untested. These use with_mocks() to capture the email.
1480
1481	#[tokio::test]
1482	async fn webhook_invoice_payment_succeeded_renewal_sends_email() {
1483	let mut h = TestHarness::with_mocks().await;
1484	let fix = setup_subscription_fixture(&mut h).await;
1485	let stripe_sub_id = "sub_renewal_email_1";
1486	insert_active_subscription(&h, &fix, stripe_sub_id).await;
1487
1488	let invoice = make_invoice(stripe_sub_id, "subscription_cycle");
1489	let resp = post_event_json(&mut h, "invoice.payment_succeeded", invoice).await;
1490	assert_eq!(resp.status.as_u16(), 200, "Webhook failed: {}", resp.text);
1491
1492	// Fire-and-forget email task, give it a beat to land.
1493	tokio::time::sleep(std::time::Duration::from_millis(200)).await;
1494
1495	let mock_email = h.mock_email.as_ref().unwrap();
1496	let mails = mock_email.sent_to("subuser@test.com");
1497	assert!(
1498	mails.iter().any(\|e\| e.subject.contains("renewed")),
1499	"renewal must send a 'renewed' email, got: {:?}",
1500	mails.iter().map(\|e\| &e.subject).collect::<Vec<_>>()
1501	);
1502	}
1503
1504	#[tokio::test]
1505	async fn webhook_invoice_payment_succeeded_first_invoice_sends_no_renewal_email() {
1506	let mut h = TestHarness::with_mocks().await;
1507	let fix = setup_subscription_fixture(&mut h).await;
1508	let stripe_sub_id = "sub_first_invoice_1";
1509	insert_active_subscription(&h, &fix, stripe_sub_id).await;
1510
1511	// subscription_create is the FIRST invoice, not a renewal.
1512	let invoice = make_invoice(stripe_sub_id, "subscription_create");
1513	let resp = post_event_json(&mut h, "invoice.payment_succeeded", invoice).await;
1514	assert_eq!(resp.status.as_u16(), 200, "Webhook failed: {}", resp.text);
1515
1516	tokio::time::sleep(std::time::Duration::from_millis(200)).await;
1517
1518	let mock_email = h.mock_email.as_ref().unwrap();
1519	let mails = mock_email.sent_to("subuser@test.com");
1520	assert!(
1521	!mails.iter().any(\|e\| e.subject.contains("renewed")),
1522	"first invoice must NOT send a renewal email, got: {:?}",
1523	mails.iter().map(\|e\| &e.subject).collect::<Vec<_>>()
1524	);
1525	}
1526
1527	// charge.refunded edge paths (test-fuzz Phase 2.1)
1528	//
1529	// Only the full-refund-with-matching-transaction path was tested. The partial
1530	// refund (preserve access) and no-match (queue a pending refund) branches,
1531	// both money-critical, were not.
1532
1533	#[tokio::test]
1534	async fn webhook_charge_refunded_partial_preserves_access() {
1535	let mut h = TestHarness::with_stripe().await;
1536
1537	let buyer_id = h.signup("prbuyer", "prb@test.com", "password123").await;
1538	h.client.post_form("/logout", "").await;
1539	let seller_id = h.signup("prseller", "prs@test.com", "password123").await;
1540	h.grant_creator(seller_id).await;
1541	h.client.post_form("/logout", "").await;
1542	h.login("prseller", "password123").await;
1543
1544	let resp = h
1545	.client
1546	.post_form("/api/projects", "slug=partialproj&title=Partial+Project")
1547	.await;
1548	let project: Value = resp.json();
1549	let project_id = project["id"].as_str().unwrap().to_string();
1550	let resp = h
1551	.client
1552	.post_form(
1553	&format!("/api/projects/{project_id}/items"),
1554	"title=Partial+Track&price_cents=1000&item_type=audio",
1555	)
1556	.await;
1557	let item: Value = resp.json();
1558	let item_id = item["id"].as_str().unwrap().to_string();
1559
1560	let pi_id = "pi_partial_refund_1";
1561	sqlx::query("UPDATE items SET sales_count = 1 WHERE id = $1::uuid")
1562	.bind(&item_id)
1563	.execute(&h.db)
1564	.await
1565	.unwrap();
1566	sqlx::query(
1567	r"INSERT INTO transactions
1568	(buyer_id, seller_id, item_id, amount_cents, status,
1569	stripe_payment_intent_id, stripe_checkout_session_id,
1570	item_title, seller_username, completed_at)
1571	VALUES ($1, $2, $3::uuid, 1000, 'completed', $4, 'cs_partial', 'Partial Track', 'prseller', NOW())",
1572	)
1573	.bind(buyer_id)
1574	.bind(seller_id)
1575	.bind(&item_id)
1576	.bind(pi_id)
1577	.execute(&h.db)
1578	.await
1579	.unwrap();
1580
1581	// Refund only 400 of 1000 cents, a partial refund.
1582	let charge = serde_json::json!({
1583	"id": "ch_partial_1",
1584	"object": "charge",
1585	"amount": 1000,
1586	"amount_refunded": 400,
1587	"payment_intent": pi_id,
1588	});
1589	let resp = post_event_json(&mut h, "charge.refunded", charge).await;
1590	assert_eq!(resp.status.as_u16(), 200, "Webhook failed: {}", resp.text);
1591
1592	// Access preserved: the transaction is NOT marked refunded and the sale stands.
1593	let status: String =
1594	sqlx::query_scalar("SELECT status FROM transactions WHERE stripe_payment_intent_id = $1")
1595	.bind(pi_id)
1596	.fetch_one(&h.db)
1597	.await
1598	.unwrap();
1599	assert_eq!(status, "completed", "partial refund must NOT revoke access");
1600
1601	let sales: i32 = sqlx::query_scalar("SELECT sales_count FROM items WHERE id = $1::uuid")
1602	.bind(&item_id)
1603	.fetch_one(&h.db)
1604	.await
1605	.unwrap();
1606	assert_eq!(sales, 1, "partial refund must NOT decrement the sale");
1607	}
1608
1609	#[tokio::test]
1610	async fn webhook_charge_refunded_no_transaction_queues_pending() {
1611	let mut h = TestHarness::with_stripe().await;
1612
1613	// A full refund whose payment_intent matches no transaction or tip,
1614	// the payment webhook likely hasn't arrived yet. Must be queued, not dropped.
1615	let pi_id = "pi_orphan_refund_1";
1616	let charge = serde_json::json!({
1617	"id": "ch_orphan_1",
1618	"object": "charge",
1619	"amount": 1500,
1620	"amount_refunded": 1500,
1621	"payment_intent": pi_id,
1622	});
1623	let resp = post_event_json(&mut h, "charge.refunded", charge).await;
1624	assert_eq!(resp.status.as_u16(), 200, "Webhook failed: {}", resp.text);
1625
1626	let (amount, amount_refunded): (i64, i64) = sqlx::query_as(
1627	"SELECT amount, amount_refunded FROM pending_refunds WHERE payment_intent_id = $1",
1628	)
1629	.bind(pi_id)
1630	.fetch_one(&h.db)
1631	.await
1632	.expect("an unmatched full refund must be queued as a pending refund");
1633	assert_eq!(amount, 1500);
1634	assert_eq!(amount_refunded, 1500);
1635	}
1636
1637	#[tokio::test]
1638	async fn webhook_refund_created_line_scoped_does_not_reverse_cart() {
1639	// Run #2 Payments SERIOUS: a cart puts every line of an order under ONE
1640	// payment_intent. A self-service refund of one line tags the Stripe refund
1641	// with mnw_transaction_id; the refund.created webhook must mark/revoke ONLY
1642	// that transaction and leave the order's other lines untouched.
1643	let mut h = TestHarness::with_stripe().await;
1644
1645	let buyer_id = h.signup("clbuyer", "clb@test.com", "password123").await;
1646	h.client.post_form("/logout", "").await;
1647	let seller_id = h.signup("clseller", "cls@test.com", "password123").await;
1648	h.grant_creator(seller_id).await;
1649	h.client.post_form("/logout", "").await;
1650	h.login("clseller", "password123").await;
1651
1652	let resp = h
1653	.client
1654	.post_form("/api/projects", "slug=cartproj&title=Cart+Project")
1655	.await;
1656	let project: Value = resp.json();
1657	let project_id = project["id"].as_str().unwrap().to_string();
1658
1659	// Two items, both sold to the same buyer in one cart (shared payment_intent).
1660	let mut item_ids = Vec::new();
1661	for (n, title) in [("Line+One", "a"), ("Line+Two", "b")] {
1662	let _ = title;
1663	let resp = h
1664	.client
1665	.post_form(
1666	&format!("/api/projects/{project_id}/items"),
1667	&format!("title={n}&price_cents=500&item_type=audio"),
1668	)
1669	.await;
1670	let item: Value = resp.json();
1671	item_ids.push(item["id"].as_str().unwrap().to_string());
1672	}
1673
1674	let pi_id = "pi_cart_line_refund";
1675	let mut tx_ids = Vec::new();
1676	for item_id in &item_ids {
1677	sqlx::query("UPDATE items SET sales_count = 1 WHERE id = $1::uuid")
1678	.bind(item_id)
1679	.execute(&h.db)
1680	.await
1681	.unwrap();
1682	let tx_id: uuid::Uuid = sqlx::query_scalar(
1683	r"INSERT INTO transactions
1684	(buyer_id, seller_id, item_id, amount_cents, status,
1685	stripe_payment_intent_id, stripe_checkout_session_id,
1686	item_title, seller_username, completed_at)
1687	VALUES ($1, $2, $3::uuid, 500, 'completed', $4, 'cs_cart', 'Line', 'clseller', NOW())
1688	RETURNING id",
1689	)
1690	.bind(buyer_id)
1691	.bind(seller_id)
1692	.bind(item_id)
1693	.bind(pi_id)
1694	.fetch_one(&h.db)
1695	.await
1696	.unwrap();
1697	tx_ids.push(tx_id);
1698	}
1699
1700	// Refund ONLY the first line via refund.created carrying its transaction id.
1701	let refund = serde_json::json!({
1702	"id": "re_cart_line_1",
1703	"object": "refund",
1704	"amount": 500,
1705	"status": "succeeded",
1706	"payment_intent": pi_id,
1707	"metadata": { "mnw_transaction_id": tx_ids[0].to_string() },
1708	});
1709	let resp = post_event_json(&mut h, "refund.created", refund).await;
1710	assert_eq!(resp.status.as_u16(), 200, "Webhook failed: {}", resp.text);
1711
1712	// Line one is refunded; line two is untouched (the bug would refund both).
1713	let status_one: String = sqlx::query_scalar("SELECT status FROM transactions WHERE id = $1")
1714	.bind(tx_ids[0])
1715	.fetch_one(&h.db)
1716	.await
1717	.unwrap();
1718	let status_two: String = sqlx::query_scalar("SELECT status FROM transactions WHERE id = $1")
1719	.bind(tx_ids[1])
1720	.fetch_one(&h.db)
1721	.await
1722	.unwrap();
1723	assert_eq!(
1724	status_one, "refunded",
1725	"refunded line must be marked refunded"
1726	);
1727	assert_eq!(
1728	status_two, "completed",
1729	"sibling cart line must NOT be reversed"
1730	);
1731
1732	// Sales count: only the refunded line's item decremented.
1733	let sales_one: i32 = sqlx::query_scalar("SELECT sales_count FROM items WHERE id = $1::uuid")
1734	.bind(&item_ids[0])
1735	.fetch_one(&h.db)
1736	.await
1737	.unwrap();
1738	let sales_two: i32 = sqlx::query_scalar("SELECT sales_count FROM items WHERE id = $1::uuid")
1739	.bind(&item_ids[1])
1740	.fetch_one(&h.db)
1741	.await
1742	.unwrap();
1743	assert_eq!(sales_one, 0, "refunded line's sale decremented");
1744	assert_eq!(sales_two, 1, "sibling line's sale stands");
1745	}
1746
1747	// Subscription / invoice edge cases (test-fuzz Phase 2.1)
1748
1749	#[tokio::test]
1750	async fn webhook_subscription_updated_unknown_status_is_noop() {
1751	// Stripe periodically adds statuses (e.g. "paused"). The handler must treat
1752	// an unknown status as a no-op, returning Err would pin Stripe in an
1753	// infinite retry storm. The existing subscription must keep its prior status.
1754	let mut h = TestHarness::with_stripe().await;
1755	let fix = setup_subscription_fixture(&mut h).await;
1756	let stripe_sub_id = "sub_unknown_status_1";
1757	insert_active_subscription(&h, &fix, stripe_sub_id).await;
1758
1759	let sub = make_subscription(stripe_sub_id, "paused"); // not a known SubscriptionStatus
1760	let resp = post_event_json(&mut h, "customer.subscription.updated", sub).await;
1761	assert_eq!(
1762	resp.status.as_u16(),
1763	200,
1764	"unknown status must not error: {}",
1765	resp.text
1766	);
1767
1768	let status: String =
1769	sqlx::query_scalar("SELECT status FROM subscriptions WHERE stripe_subscription_id = $1")
1770	.bind(stripe_sub_id)
1771	.fetch_one(&h.db)
1772	.await
1773	.unwrap();
1774	assert_eq!(
1775	status, "active",
1776	"unknown status must be a no-op, leaving status untouched"
1777	);
1778	}
1779
1780	#[tokio::test]
1781	async fn webhook_invoice_without_subscription_is_noop() {
1782	// A non-subscription invoice (no subscription id anywhere) must short-circuit
1783	// to Ok without touching any subscription table.
1784	let mut h = TestHarness::with_stripe().await;
1785	let invoice = serde_json::json!({
1786	"id": "in_no_sub_1",
1787	"object": "invoice",
1788	"period_start": 1_700_000_000_i64,
1789	"period_end": 1_702_592_000_i64,
1790	"billing_reason": "manual",
1791	"livemode": false,
1792	});
1793	let resp = post_event_json(&mut h, "invoice.payment_succeeded", invoice).await;
1794	assert_eq!(
1795	resp.status.as_u16(),
1796	200,
1797	"invoice without subscription must be a 200 no-op: {}",
1798	resp.text
1799	);
1800	}
1801
1802	// Adversarial: malformed fulfillment metadata (test-fuzz Phase 2.1)
1803	//
1804	// A checkout.session.completed whose metadata is missing the required buyer_id
1805	// makes the handler error. The dispatcher must queue it for retry (a row in
1806	// webhook_events) and still 200 Stripe, never drop it, never 500.
1807
1808	#[tokio::test]
1809	async fn webhook_purchase_missing_buyer_id_metadata_is_queued() {
1810	let mut h = TestHarness::with_stripe().await;
1811	let seller_id = h.signup("mmseller", "mms@test.com", "password123").await;
1812
1813	// No checkout_type → routed to the purchase handler, whose
1814	// CheckoutMetadata::from_metadata requires buyer_id and errors without it.
1815	let mut meta = HashMap::new();
1816	meta.insert("seller_id".to_string(), seller_id.to_string());
1817	let session = serde_json::json!({
1818	"id": "cs_missing_buyer_1",
1819	"object": "checkout_session",
1820	"mode": "payment",
1821	"metadata": meta,
1822	"payment_intent": "pi_missing_buyer_1",
1823	});
1824
1825	let resp = post_event_json(&mut h, "checkout.session.completed", session).await;
1826	assert_eq!(
1827	resp.status.as_u16(),
1828	200,
1829	"malformed event must still 200 after queueing: {}",
1830	resp.text
1831	);
1832
1833	// The failed handler queued the event for retry rather than dropping it.
1834	let queued: i64 = sqlx::query_scalar(
1835	"SELECT COUNT(*) FROM webhook_events WHERE source = 'stripe' AND event_type = 'checkout.session.completed'",
1836	)
1837	.fetch_one(&h.db)
1838	.await
1839	.unwrap();
1840	assert_eq!(
1841	queued, 1,
1842	"a handler error must enqueue exactly one retry row"
1843	);
1844	}
1845
1846	// Canceled is terminal: an out-of-order update must not revive it (Run #11 fix)
1847
1848	#[tokio::test]
1849	async fn webhook_subscription_updated_cannot_revive_canceled_sub() {
1850	let mut h = TestHarness::with_stripe().await;
1851	let fix = setup_subscription_fixture(&mut h).await;
1852	let stripe_sub_id = "sub_revival_guard_1";
1853	insert_active_subscription(&h, &fix, stripe_sub_id).await;
1854
1855	// Cancel it (terminal state).
1856	let resp = post_event_json_with_id(
1857	&mut h,
1858	"evt_revive_del",
1859	"customer.subscription.deleted",
1860	make_subscription(stripe_sub_id, "canceled"),
1861	)
1862	.await;
1863	assert_eq!(resp.status.as_u16(), 200, "delete failed: {}", resp.text);
1864	let status: String =
1865	sqlx::query_scalar("SELECT status FROM subscriptions WHERE stripe_subscription_id = $1")
1866	.bind(stripe_sub_id)
1867	.fetch_one(&h.db)
1868	.await
1869	.unwrap();
1870	assert_eq!(status, "canceled");
1871
1872	// An out-of-order `updated`(active) arrives AFTER the cancellation, distinct
1873	// event id, so the dedup layer does NOT short-circuit it. The DB guard must
1874	// refuse to revive the canceled subscription.
1875	let resp = post_event_json_with_id(
1876	&mut h,
1877	"evt_revive_upd",
1878	"customer.subscription.updated",
1879	make_subscription(stripe_sub_id, "active"),
1880	)
1881	.await;
1882	assert_eq!(
1883	resp.status.as_u16(),
1884	200,
1885	"update must not error: {}",
1886	resp.text
1887	);
1888	let status: String =
1889	sqlx::query_scalar("SELECT status FROM subscriptions WHERE stripe_subscription_id = $1")
1890	.bind(stripe_sub_id)
1891	.fetch_one(&h.db)
1892	.await
1893	.unwrap();
1894	assert_eq!(
1895	status, "canceled",
1896	"canceled is terminal, an out-of-order update must not revive it"
1897	);
1898	}
1899
1900	#[tokio::test]
1901	async fn webhook_subscription_updated_cannot_revive_canceled_creator_tier() {
1902	let mut h = TestHarness::with_stripe().await;
1903	let user_id = h
1904	.signup("ctrevive", "ctrevive@test.com", "password123")
1905	.await;
1906	let stripe_sub_id = "sub_ct_revive_1";
1907	insert_active_creator_sub(&h, user_id, stripe_sub_id, "small_files").await;
1908
1909	// Cancel, clears the denormalized tier.
1910	let resp = post_event_json_with_id(
1911	&mut h,
1912	"evt_ctrevive_del",
1913	"customer.subscription.deleted",
1914	make_subscription(stripe_sub_id, "canceled"),
1915	)
1916	.await;
1917	assert_eq!(resp.status.as_u16(), 200, "delete failed: {}", resp.text);
1918
1919	// Out-of-order active update must not revive the canceled creator sub.
1920	let resp = post_event_json_with_id(
1921	&mut h,
1922	"evt_ctrevive_upd",
1923	"customer.subscription.updated",
1924	make_subscription(stripe_sub_id, "active"),
1925	)
1926	.await;
1927	assert_eq!(
1928	resp.status.as_u16(),
1929	200,
1930	"update must not error: {}",
1931	resp.text
1932	);
1933
1934	let status: String = sqlx::query_scalar(
1935	"SELECT status FROM creator_subscriptions WHERE stripe_subscription_id = $1",
1936	)
1937	.bind(stripe_sub_id)
1938	.fetch_one(&h.db)
1939	.await
1940	.unwrap();
1941	assert_eq!(
1942	status, "canceled",
1943	"canceled creator sub must not be revived"
1944	);
1945	let tier: Option<String> = sqlx::query_scalar("SELECT creator_tier FROM users WHERE id = $1")
1946	.bind(user_id)
1947	.fetch_one(&h.db)
1948	.await
1949	.unwrap();
1950	assert_eq!(
1951	tier, None,
1952	"denormalized tier must stay cleared after a refused revival"
1953	);
1954	}
1955
1956	#[tokio::test]
1957	async fn webhook_subscription_updated_cannot_revive_canceled_fan_plus() {
1958	// Fan+ was the sibling the Run #11 revival guard missed (Run #12 SERIOUS).
1959	let mut h = TestHarness::with_stripe().await;
1960	let user_id = h
1961	.signup("fprevive", "fprevive@test.com", "password123")
1962	.await;
1963
1964	let stripe_sub_id = "sub_fp_revive_1";
1965	sqlx::query(
1966	r"INSERT INTO fan_plus_subscriptions
1967	(user_id, stripe_subscription_id, stripe_customer_id, status, current_period_end)
1968	VALUES ($1, $2, 'cus_fp_revive', 'active', NOW() + interval '30 days')",
1969	)
1970	.bind(user_id)
1971	.bind(stripe_sub_id)
1972	.execute(&h.db)
1973	.await
1974	.unwrap();
1975
1976	// Cancel (terminal).
1977	let resp = post_event_json_with_id(
1978	&mut h,
1979	"evt_fp_revive_del",
1980	"customer.subscription.deleted",
1981	make_subscription(stripe_sub_id, "canceled"),
1982	)
1983	.await;
1984	assert_eq!(resp.status.as_u16(), 200, "delete failed: {}", resp.text);
1985
1986	// Out-of-order active update must NOT revive it.
1987	let resp = post_event_json_with_id(
1988	&mut h,
1989	"evt_fp_revive_upd",
1990	"customer.subscription.updated",
1991	make_subscription(stripe_sub_id, "active"),
1992	)
1993	.await;
1994	assert_eq!(
1995	resp.status.as_u16(),
1996	200,
1997	"update must not error: {}",
1998	resp.text
1999	);
2000
2001	// The downstream "is this user Fan+?" check must classify them as NOT active.
2002	let active: bool = sqlx::query_scalar(
2003	"SELECT EXISTS(SELECT 1 FROM fan_plus_subscriptions \
2004	WHERE user_id = $1 AND status = 'active' AND canceled_at IS NULL)",
2005	)
2006	.bind(user_id)
2007	.fetch_one(&h.db)
2008	.await
2009	.unwrap();
2010	assert!(
2011	!active,
2012	"a canceled Fan+ sub must not be revived by an out-of-order update"
2013	);
2014	}
2015
2016	// Canceled is terminal for the PERIOD too: an out-of-order invoice.paid must
2017	// not refresh current_period_* on a canceled row. This is the sibling the
2018	// status guard used to miss, now status + period are written together under
2019	// one guard, so the period write can't bypass it. (Run #13 chronic fix.)
2020
2021	#[tokio::test]
2022	async fn webhook_invoice_paid_cannot_refresh_period_on_canceled_sub() {
2023	let mut h = TestHarness::with_stripe().await;
2024	let fix = setup_subscription_fixture(&mut h).await;
2025	let stripe_sub_id = "sub_period_guard_1";
2026	insert_active_subscription(&h, &fix, stripe_sub_id).await;
2027
2028	// Cancel (terminal).
2029	let resp = post_event_json_with_id(
2030	&mut h,
2031	"evt_periodguard_del",
2032	"customer.subscription.deleted",
2033	make_subscription(stripe_sub_id, "canceled"),
2034	)
2035	.await;
2036	assert_eq!(resp.status.as_u16(), 200, "delete failed: {}", resp.text);
2037
2038	let before: Option<chrono::DateTime<chrono::Utc>> = sqlx::query_scalar(
2039	"SELECT current_period_end FROM subscriptions WHERE stripe_subscription_id = $1",
2040	)
2041	.bind(stripe_sub_id)
2042	.fetch_one(&h.db)
2043	.await
2044	.unwrap();
2045
2046	// A stray invoice.payment_succeeded (period_end 1_702_592_000) arrives after cancel.
2047	let resp = post_event_json(
2048	&mut h,
2049	"invoice.payment_succeeded",
2050	make_invoice(stripe_sub_id, "subscription_cycle"),
2051	)
2052	.await;
2053	assert_eq!(
2054	resp.status.as_u16(),
2055	200,
2056	"invoice must not error: {}",
2057	resp.text
2058	);
2059
2060	let after: Option<chrono::DateTime<chrono::Utc>> = sqlx::query_scalar(
2061	"SELECT current_period_end FROM subscriptions WHERE stripe_subscription_id = $1",
2062	)
2063	.bind(stripe_sub_id)
2064	.fetch_one(&h.db)
2065	.await
2066	.unwrap();
2067	assert_eq!(
2068	after, before,
2069	"period must not be refreshed on a canceled subscription"
2070	);
2071	assert_ne!(
2072	after.map(\|d\| d.timestamp()),
2073	Some(1_702_592_000),
2074	"invoice period must not land on a canceled row"
2075	);
2076	}
2077
2078	#[tokio::test]
2079	async fn webhook_invoice_paid_cannot_refresh_period_on_canceled_creator_tier() {
2080	let mut h = TestHarness::with_stripe().await;
2081	let user_id = h
2082	.signup("ctperiod", "ctperiod@test.com", "password123")
2083	.await;
2084	let stripe_sub_id = "sub_ct_period_1";
2085	insert_active_creator_sub(&h, user_id, stripe_sub_id, "small_files").await;
2086
2087	let resp = post_event_json_with_id(
2088	&mut h,
2089	"evt_ctperiod_del",
2090	"customer.subscription.deleted",
2091	make_subscription(stripe_sub_id, "canceled"),
2092	)
2093	.await;
2094	assert_eq!(resp.status.as_u16(), 200, "delete failed: {}", resp.text);
2095
2096	let before: Option<chrono::DateTime<chrono::Utc>> = sqlx::query_scalar(
2097	"SELECT current_period_end FROM creator_subscriptions WHERE stripe_subscription_id = $1",
2098	)
2099	.bind(stripe_sub_id)
2100	.fetch_one(&h.db)
2101	.await
2102	.unwrap();
2103
2104	let resp = post_event_json(
2105	&mut h,
2106	"invoice.payment_succeeded",
2107	make_invoice(stripe_sub_id, "subscription_cycle"),
2108	)
2109	.await;
2110	assert_eq!(
2111	resp.status.as_u16(),
2112	200,
2113	"invoice must not error: {}",
2114	resp.text
2115	);
2116
2117	let after: Option<chrono::DateTime<chrono::Utc>> = sqlx::query_scalar(
2118	"SELECT current_period_end FROM creator_subscriptions WHERE stripe_subscription_id = $1",
2119	)
2120	.bind(stripe_sub_id)
2121	.fetch_one(&h.db)
2122	.await
2123	.unwrap();
2124	assert_eq!(
2125	after, before,
2126	"period must not be refreshed on a canceled creator sub"
2127	);
2128	}
2129
2130	#[tokio::test]
2131	async fn webhook_invoice_paid_cannot_refresh_period_on_canceled_fan_plus() {
2132	let mut h = TestHarness::with_stripe().await;
2133	let user_id = h
2134	.signup("fpperiod", "fpperiod@test.com", "password123")
2135	.await;
2136	let stripe_sub_id = "sub_fp_period_1";
2137	sqlx::query(
2138	r"INSERT INTO fan_plus_subscriptions
2139	(user_id, stripe_subscription_id, stripe_customer_id, status, current_period_end)
2140	VALUES ($1, $2, 'cus_fp_period', 'active', to_timestamp(1_000_000_000))",
2141	)
2142	.bind(user_id)
2143	.bind(stripe_sub_id)
2144	.execute(&h.db)
2145	.await
2146	.unwrap();
2147
2148	let resp = post_event_json_with_id(
2149	&mut h,
2150	"evt_fpperiod_del",
2151	"customer.subscription.deleted",
2152	make_subscription(stripe_sub_id, "canceled"),
2153	)
2154	.await;
2155	assert_eq!(resp.status.as_u16(), 200, "delete failed: {}", resp.text);
2156
2157	// Stray invoice.paid with a DIFFERENT period_end (1_702_592_000) must be refused.
2158	let resp = post_event_json(
2159	&mut h,
2160	"invoice.payment_succeeded",
2161	make_invoice(stripe_sub_id, "subscription_cycle"),
2162	)
2163	.await;
2164	assert_eq!(
2165	resp.status.as_u16(),
2166	200,
2167	"invoice must not error: {}",
2168	resp.text
2169	);
2170
2171	let after: chrono::DateTime<chrono::Utc> = sqlx::query_scalar(
2172	"SELECT current_period_end FROM fan_plus_subscriptions WHERE stripe_subscription_id = $1",
2173	)
2174	.bind(stripe_sub_id)
2175	.fetch_one(&h.db)
2176	.await
2177	.unwrap();
2178	assert_eq!(
2179	after.timestamp(),
2180	1_000_000_000,
2181	"period must stay frozen on a canceled Fan+ sub, not jump to the invoice's period_end"
2182	);
2183	}
2184
2185	#[tokio::test]
2186	async fn fan_plus_renewal_credit_issued_once_across_duplicate_deliveries() {
2187	// Regression for the Run #22 SERIOUS: webhook event-dedup is a check-then-act
2188	// read keyed on event id; a duplicate `invoice.payment_succeeded` delivery
2189	// that escapes it (distinct event id, same invoice/period) must NOT mint a
2190	// second $5 Fan+ credit. The per-(stripe_sub_id, period_end) idempotency
2191	// guard makes the renewal credit a single DB-level write.
2192	let mut h = TestHarness::with_stripe().await;
2193	let user_id = h
2194	.signup("fpcredit", "fpcredit@test.com", "password123")
2195	.await;
2196	let stripe_sub_id = "sub_fp_credit_idem";
2197	sqlx::query(
2198	r"INSERT INTO fan_plus_subscriptions
2199	(user_id, stripe_subscription_id, stripe_customer_id, status, current_period_end)
2200	VALUES ($1, $2, 'cus_fp_credit', 'active', to_timestamp(1_700_000_000))",
2201	)
2202	.bind(user_id)
2203	.bind(stripe_sub_id)
2204	.execute(&h.db)
2205	.await
2206	.unwrap();
2207
2208	// The platform-wide $5 credit codes minted for this user.
2209	async fn credit_count(h: &TestHarness, user_id: UserId) -> i64 {
2210	sqlx::query_scalar(
2211	"SELECT COUNT(*) FROM promo_codes \
2212	WHERE creator_id = $1 AND is_platform_wide = true AND discount_value = 500",
2213	)
2214	.bind(user_id)
2215	.fetch_one(&h.db)
2216	.await
2217	.unwrap()
2218	}
2219
2220	// First renewal delivery mints exactly one credit.
2221	let resp = post_event_json_with_id(
2222	&mut h,
2223	"evt_fpcredit_1",
2224	"invoice.payment_succeeded",
2225	make_invoice(stripe_sub_id, "subscription_cycle"),
2226	)
2227	.await;
2228	assert_eq!(
2229	resp.status.as_u16(),
2230	200,
2231	"first renewal failed: {}",
2232	resp.text
2233	);
2234	assert_eq!(
2235	credit_count(&h, user_id).await,
2236	1,
2237	"first renewal must mint one credit"
2238	);
2239
2240	// A duplicate delivery of the SAME renewal period under a different event id
2241	// (so event-dedup does not catch it) must not mint a second credit.
2242	let resp = post_event_json_with_id(
2243	&mut h,
2244	"evt_fpcredit_2",
2245	"invoice.payment_succeeded",
2246	make_invoice(stripe_sub_id, "subscription_cycle"),
2247	)
2248	.await;
2249	assert_eq!(
2250	resp.status.as_u16(),
2251	200,
2252	"duplicate renewal failed: {}",
2253	resp.text
2254	);
2255	assert_eq!(
2256	credit_count(&h, user_id).await,
2257	1,
2258	"a duplicate delivery of the same renewal must not double-issue the credit"
2259	);
2260	}
2261
2262	// PAY-S1: a failed pending-refund process must RELEASE its claim
2263	//
2264	// check_pending_refund claims the pending refund (matched_at = NOW()) before the
2265	// fallible handle_charge_refunded. If that work fails, the claim must be released
2266	// (matched_at -> NULL) so the stale-refund sweep escalates it and a later
2267	// delivery can retry. Without the release the row is lost forever: the sweep
2268	// filters `matched_at IS NULL`, and the original charge.refunded event was
2269	// already marked processed so Stripe never redelivers. We force the failure with
2270	// a trigger that raises when the refund flips the transaction to 'refunded'.
2271
2272	#[tokio::test]
2273	async fn pending_refund_releases_claim_when_processing_fails() {
2274	let mut h = TestHarness::with_stripe().await;
2275
2276	let buyer_id = h
2277	.signup("prfbuyer", "prfbuyer@test.com", "password123")
2278	.await;
2279	h.client.post_form("/logout", "").await;
2280	let seller_id = h
2281	.signup("prfseller", "prfseller@test.com", "password123")
2282	.await;
2283	h.grant_creator(seller_id).await;
2284	h.client.post_form("/logout", "").await;
2285	h.login("prfseller", "password123").await;
2286
2287	let resp = h
2288	.client
2289	.post_form("/api/projects", "slug=prfproj&title=PRF+Project")
2290	.await;
2291	let project: Value = resp.json();
2292	let project_id = project["id"].as_str().unwrap().to_string();
2293	let resp = h
2294	.client
2295	.post_form(
2296	&format!("/api/projects/{project_id}/items"),
2297	"title=PRF+Track&price_cents=999&item_type=audio",
2298	)
2299	.await;
2300	let item: Value = resp.json();
2301	let item_id = item["id"].as_str().unwrap().to_string();
2302
2303	let session_id = "cs_prf_001";
2304	let pi = "pi_faultrefund_001";
2305
2306	// Pending transaction for this checkout session, carrying the sentinel PI.
2307	sqlx::query(
2308	r"INSERT INTO transactions
2309	(buyer_id, seller_id, item_id, amount_cents, status,
2310	stripe_checkout_session_id, stripe_payment_intent_id, item_title, seller_username)
2311	VALUES ($1, $2, $3::uuid, 999, 'pending', $4, $5, 'PRF Track', 'prfseller')",
2312	)
2313	.bind(buyer_id)
2314	.bind(seller_id)
2315	.bind(&item_id)
2316	.bind(session_id)
2317	.bind(pi)
2318	.execute(&h.db)
2319	.await
2320	.unwrap();
2321
2322	// A full refund for this PI arrived first and was queued.
2323	sqlx::query("INSERT INTO pending_refunds (payment_intent_id, amount, amount_refunded) VALUES ($1, 999, 999)")
2324	.bind(pi).execute(&h.db).await.unwrap();
2325
2326	// Force handle_charge_refunded to fail: raise when the refund flips this PI's
2327	// transaction to 'refunded' (the completion -> 'completed' update is untouched).
2328	sqlx::query(
2329	r"CREATE OR REPLACE FUNCTION test_refund_fault() RETURNS trigger AS $$
2330	BEGIN
2331	IF NEW.status = 'refunded' AND COALESCE(NEW.stripe_payment_intent_id,'') LIKE '%faultrefund%' THEN
2332	RAISE EXCEPTION 'injected refund fault';
2333	END IF;
2334	RETURN NEW;
2335	END; $$ LANGUAGE plpgsql",
2336	).execute(&h.db).await.unwrap();
2337	sqlx::query("CREATE TRIGGER test_refund_fault BEFORE UPDATE ON transactions FOR EACH ROW EXECUTE FUNCTION test_refund_fault()")
2338	.execute(&h.db).await.unwrap();
2339
2340	// Complete the checkout: fulfillment completes the tx, then check_pending_refund
2341	// claims the refund and tries to process it -> trigger raises -> claim released.
2342	let mut meta = HashMap::new();
2343	meta.insert("buyer_id".to_string(), buyer_id.to_string());
2344	meta.insert("seller_id".to_string(), seller_id.to_string());
2345	meta.insert("item_id".to_string(), item_id.clone());
2346	let session = serde_json::json!({
2347	"id": session_id,
2348	"object": "checkout_session",
2349	"mode": "payment",
2350	"metadata": meta,
2351	"payment_intent": pi,
2352	});
2353	let resp = post_event_json(&mut h, "checkout.session.completed", session).await;
2354	assert_eq!(
2355	resp.status.as_u16(),
2356	200,
2357	"checkout webhook should still 200: {}",
2358	resp.text
2359	);
2360
2361	// The claim must be RELEASED so the stale-refund sweep can pick it up.
2362	let matched_at: Option<chrono::DateTime<chrono::Utc>> =
2363	sqlx::query_scalar("SELECT matched_at FROM pending_refunds WHERE payment_intent_id = $1")
2364	.bind(pi)
2365	.fetch_one(&h.db)
2366	.await
2367	.unwrap();
2368	assert!(
2369	matched_at.is_none(),
2370	"a failed pending-refund process must release its claim (matched_at NULL), else the refund is lost"
2371	);
2372
2373	// The refund rolled back atomically: the transaction is NOT marked refunded.
2374	let status: String =
2375	sqlx::query_scalar("SELECT status FROM transactions WHERE stripe_payment_intent_id = $1")
2376	.bind(pi)
2377	.fetch_one(&h.db)
2378	.await
2379	.unwrap();
2380	assert_ne!(
2381	status, "refunded",
2382	"the failed refund must not have committed"
2383	);
2384	}
2385
2386	// Complement: a pending refund that processes SUCCESSFULLY on completion is
2387	// consumed (matched_at set, transaction refunded), pins that the happy path
2388	// still claims-and-keeps, so the release above is failure-only.
2389	#[tokio::test]
2390	async fn pending_refund_processed_on_completion_marks_matched() {
2391	let mut h = TestHarness::with_stripe().await;
2392
2393	let buyer_id = h
2394	.signup("prfok_buyer", "prfokb@test.com", "password123")
2395	.await;
2396	h.client.post_form("/logout", "").await;
2397	let seller_id = h
2398	.signup("prfok_seller", "prfoks@test.com", "password123")
2399	.await;
2400	h.grant_creator(seller_id).await;
2401	h.client.post_form("/logout", "").await;
2402	h.login("prfok_seller", "password123").await;
2403
2404	let resp = h
2405	.client
2406	.post_form("/api/projects", "slug=prfokproj&title=PRFOK")
2407	.await;
2408	let project: Value = resp.json();
2409	let project_id = project["id"].as_str().unwrap().to_string();
2410	let resp = h
2411	.client
2412	.post_form(
2413	&format!("/api/projects/{project_id}/items"),
2414	"title=PRFOK+Track&price_cents=999&item_type=audio",
2415	)
2416	.await;
2417	let item: Value = resp.json();
2418	let item_id = item["id"].as_str().unwrap().to_string();
2419
2420	let session_id = "cs_prfok_001";
2421	let pi = "pi_prfok_001";
2422	sqlx::query(
2423	r"INSERT INTO transactions
2424	(buyer_id, seller_id, item_id, amount_cents, status,
2425	stripe_checkout_session_id, stripe_payment_intent_id, item_title, seller_username)
2426	VALUES ($1, $2, $3::uuid, 999, 'pending', $4, $5, 'PRFOK Track', 'prfok_seller')",
2427	)
2428	.bind(buyer_id)
2429	.bind(seller_id)
2430	.bind(&item_id)
2431	.bind(session_id)
2432	.bind(pi)
2433	.execute(&h.db)
2434	.await
2435	.unwrap();
2436	sqlx::query("INSERT INTO pending_refunds (payment_intent_id, amount, amount_refunded) VALUES ($1, 999, 999)")
2437	.bind(pi).execute(&h.db).await.unwrap();
2438
2439	let mut meta = HashMap::new();
2440	meta.insert("buyer_id".to_string(), buyer_id.to_string());
2441	meta.insert("seller_id".to_string(), seller_id.to_string());
2442	meta.insert("item_id".to_string(), item_id.clone());
2443	let session = serde_json::json!({
2444	"id": session_id, "object": "checkout_session", "mode": "payment",
2445	"metadata": meta, "payment_intent": pi,
2446	});
2447	let resp = post_event_json(&mut h, "checkout.session.completed", session).await;
2448	assert_eq!(
2449	resp.status.as_u16(),
2450	200,
2451	"checkout webhook failed: {}",
2452	resp.text
2453	);
2454
2455	let matched_at: Option<chrono::DateTime<chrono::Utc>> =
2456	sqlx::query_scalar("SELECT matched_at FROM pending_refunds WHERE payment_intent_id = $1")
2457	.bind(pi)
2458	.fetch_one(&h.db)
2459	.await
2460	.unwrap();
2461	assert!(
2462	matched_at.is_some(),
2463	"a successfully processed pending refund stays claimed"
2464	);
2465
2466	let status: String =
2467	sqlx::query_scalar("SELECT status FROM transactions WHERE stripe_payment_intent_id = $1")
2468	.bind(pi)
2469	.fetch_one(&h.db)
2470	.await
2471	.unwrap();
2472	assert_eq!(
2473	status, "refunded",
2474	"the out-of-order refund must apply on completion"
2475	);
2476	}
2477
2478	// PAY-S1 crash window: a refund claimed but never completed (process killed
2479	// between claim and completion) must be surfaced by the stale-refund sweep, not
2480	// silently lost. The old sweep filtered `matched_at IS NULL`, so a row that was
2481	// claimed (matched_at set) but whose refund never finished was invisible forever.
2482	// The fix adds `completed_at` and the sweep (`get_stale_refunds`) filters on that.
2483	// This pins the exact selection predicate the sweep relies on (db::pending_refunds
2484	// is pub(crate), so the invariant is asserted at the SQL layer it queries).
2485	#[tokio::test]
2486	async fn pending_refund_claimed_but_uncompleted_is_swept() {
2487	let h = TestHarness::with_stripe().await;
2488
2489	// A claimed-but-incomplete row: refund arrived >24h ago (created_at old),
2490	// claim taken (matched_at set), but completed_at NULL, the process died
2491	// before recording completion.
2492	let pi_stuck = "pi_prf_crash_001";
2493	sqlx::query(
2494	"INSERT INTO pending_refunds (payment_intent_id, amount, amount_refunded, created_at, matched_at, completed_at)
2495	VALUES ($1, 999, 999, NOW() - INTERVAL '25 hours', NOW() - INTERVAL '25 hours', NULL)",
2496	).bind(pi_stuck).execute(&h.db).await.unwrap();
2497
2498	// A fully-completed row of the same age must NOT be escalated.
2499	let pi_done = "pi_prf_done_001";
2500	sqlx::query(
2501	"INSERT INTO pending_refunds (payment_intent_id, amount, amount_refunded, created_at, matched_at, completed_at)
2502	VALUES ($1, 999, 999, NOW() - INTERVAL '25 hours', NOW() - INTERVAL '25 hours', NOW() - INTERVAL '25 hours')",
2503	).bind(pi_done).execute(&h.db).await.unwrap();
2504
2505	// The sweep predicate from `get_stale_refunds`: not completed, not escalated,
2506	// older than the cutoff. The stuck row surfaces; the completed one does not.
2507	// Under the old `matched_at IS NULL` filter the stuck row would have been
2508	// invisible (matched_at was set), silently dropping the owed refund.
2509	let swept: Vec<String> = sqlx::query_scalar(
2510	"SELECT payment_intent_id FROM pending_refunds
2511	WHERE completed_at IS NULL AND escalated_at IS NULL AND created_at < NOW() - INTERVAL '24 hours'
2512	AND payment_intent_id = ANY($1)",
2513	)
2514	.bind(vec![pi_stuck.to_string(), pi_done.to_string()])
2515	.fetch_all(&h.db)
2516	.await
2517	.unwrap();
2518
2519	assert!(
2520	swept.contains(&pi_stuck.to_string()),
2521	"a claimed-but-uncompleted refund must surface in the stale sweep (PAY-S1 crash window)"
2522	);
2523	assert!(
2524	!swept.contains(&pi_done.to_string()),
2525	"a completed refund must not be escalated"
2526	);
2527	}
2528