Skip to main content

max / makenotwork

21.2 KB · 563 lines History Blame Raw
1 //! Route-layer contract tests for `routes::stripe::webhook::checkout`, the
2 //! session half of the exactly-once protocol.
3 //!
4 //! Stripe redelivers. It redelivers the same event id after a timeout, it
5 //! delivers a second event for the same checkout session when an asynchronous
6 //! payment settles, and it delivers events out of the order they happened in.
7 //! The money contract is that none of that pays anyone twice: one completed
8 //! transaction per session, one sales-count increment, one license key, one
9 //! purchase receipt, one Fan+ subscription row.
10 //!
11 //! What these tests pin, at the HTTP boundary the real deliveries arrive on:
12 //! - a redelivered `checkout.session.completed` (same event id) is a no-op,
13 //! down to `completed_at` being the same instant it was;
14 //! - the same session arriving under a NEW event id, which defeats the
15 //! event-id dedup, is still completed only once, because the
16 //! status-guarded UPDATE underneath it refuses the second pass;
17 //! - a one-time checkout that reports `payment_status: "unpaid"` delivers
18 //! nothing until the settled event arrives (out-of-order settlement);
19 //! - a forged signature is refused with 400 and writes nothing at all;
20 //! - a redelivered Fan+ checkout creates one subscription.
21 //!
22 //! The invoice half is `stripe_webhook_billing_replay`.
23 //!
24 //! Delete this file and the suite keeps saying that one delivery of each event
25 //! works, which was never the risk.
26
27 use crate::harness::TestHarness;
28 use crate::harness::stripe::{TEST_WEBHOOK_SECRET, sign_webhook_payload};
29 use makenotwork::db::UserId;
30
31 /// The item price used throughout. Deliberately not 0, 1, or a round dollar:
32 /// 1499 tells cents from dollars ($14.99), tells a doubled credit (2998) from
33 /// a single one, and cannot agree with a sum that dropped a term.
34 const PRICE_CENTS: i64 = 1499;
35
36 // ── posting events ──────────────────────────────────────────────────────────
37
38 /// Sign an event envelope with the harness webhook secret and POST it.
39 async fn post_event(
40 h: &mut TestHarness,
41 event_id: &str,
42 event_type: &str,
43 object: serde_json::Value,
44 ) -> crate::harness::client::TestResponse {
45 let payload = serde_json::json!({
46 "id": event_id,
47 "type": event_type,
48 "data": {"object": object},
49 })
50 .to_string();
51 let signature = sign_webhook_payload(&payload, TEST_WEBHOOK_SECRET);
52
53 // 503 is the route's documented "redeliver me" answer, not a failure: the
54 // handler takes a `pg_try_advisory_xact_lock` on the event id and answers
55 // 503 rather than parking a connection when that lock is held. sqlx rolls a
56 // dropped transaction back lazily, so the lock from the delivery that just
57 // returned can still be held for a moment, and a test that posts the
58 // redelivery immediately would be asserting that timing rather than the
59 // contract. Stripe's own answer to a 503 is to send the event again, so
60 // that is what this does, bounded. The status the caller asserts is the
61 // one the route settles on.
62 for _ in 0..200 {
63 let resp = post_signed(h, &payload, &signature).await;
64 if resp.status != 503 {
65 return resp;
66 }
67 tokio::time::sleep(std::time::Duration::from_millis(10)).await;
68 }
69 panic!("{event_id}: the webhook route answered 503 for two seconds");
70 }
71
72 /// POST a payload with a caller-supplied signature header, so a test can hand
73 /// over one that does not match the body.
74 async fn post_signed(
75 h: &mut TestHarness,
76 payload: &str,
77 signature: &str,
78 ) -> crate::harness::client::TestResponse {
79 h.client
80 .request_with_headers(
81 "POST",
82 "/stripe/webhook",
83 Some(payload),
84 &[
85 ("stripe-signature", signature),
86 ("content-type", "application/json"),
87 ],
88 )
89 .await
90 }
91
92 // ── fixtures ────────────────────────────────────────────────────────────────
93
94 /// A seller with one license-key-enabled item, plus a buyer, plus a pending
95 /// transaction for `session_id` waiting on the webhook. Returns
96 /// `(buyer_id, seller_id, item_id)`.
97 async fn pending_purchase(
98 h: &mut TestHarness,
99 tag: &str,
100 session_id: &str,
101 ) -> (UserId, UserId, String) {
102 let setup = h
103 .create_creator_with_item(&format!("seller{tag}"), "audio", PRICE_CENTS)
104 .await;
105
106 // License keys are the mint-once oracle for the finalize step: the handler
107 // mints at most one per transaction, so a second run of finalize shows up
108 // here as a second row.
109 sqlx::query("UPDATE items SET enable_license_keys = true WHERE id = $1::uuid")
110 .bind(&setup.item_id)
111 .execute(&h.db)
112 .await
113 .expect("enable license keys");
114
115 h.client.post_form("/logout", "").await;
116 let buyer_id = h
117 .signup(
118 &format!("buyer{tag}"),
119 &format!("buyer{tag}@test.com"),
120 "password123",
121 )
122 .await;
123
124 sqlx::query(
125 r"INSERT INTO transactions
126 (buyer_id, seller_id, item_id, amount_cents, status,
127 stripe_checkout_session_id, item_title, seller_username)
128 VALUES ($1, $2, $3::uuid, $4, 'pending', $5, 'Test Item', 'seller')",
129 )
130 .bind(buyer_id)
131 .bind(setup.user_id)
132 .bind(&setup.item_id)
133 .bind(PRICE_CENTS)
134 .bind(session_id)
135 .execute(&h.db)
136 .await
137 .expect("insert pending transaction");
138
139 (buyer_id, setup.user_id, setup.item_id)
140 }
141
142 /// A `checkout.session.*` object for the one-time purchase path.
143 fn purchase_session(
144 session_id: &str,
145 buyer_id: UserId,
146 seller_id: UserId,
147 item_id: &str,
148 payment_status: &str,
149 ) -> serde_json::Value {
150 serde_json::json!({
151 "id": session_id,
152 "object": "checkout.session",
153 "payment_intent": "pi_exactly_once",
154 "payment_status": payment_status,
155 "currency": "usd",
156 "amount_subtotal": PRICE_CENTS,
157 "metadata": {
158 "buyer_id": buyer_id.to_string(),
159 "seller_id": seller_id.to_string(),
160 "item_id": item_id,
161 },
162 })
163 }
164
165 // ── observations ────────────────────────────────────────────────────────────
166
167 /// Everything one checkout session's completion touches, read together so the
168 /// state after a redelivery can be compared with the state after the first
169 /// delivery field by field. A single-field oracle would miss the case where the
170 /// transaction is guarded but the counters are not.
171 struct PurchaseState {
172 /// Transaction rows carrying this session id. More than one is a second charge.
173 rows: i64,
174 /// The transaction's status: `pending` until the completion write lands.
175 status: String,
176 /// When the completion write landed. A redelivery that re-ran it moves this,
177 /// which is the difference between a guarded UPDATE and an unguarded one.
178 completed_at: Option<chrono::DateTime<chrono::Utc>>,
179 /// The item's denormalised sales counter, incremented once per purchase.
180 sales_count: i32,
181 /// License keys minted for the item. Enabled on the fixture item precisely
182 /// so a second run of the finalize step shows up here as a second row
183 /// rather than disappearing into an idempotent upsert.
184 license_keys: i64,
185 }
186
187 async fn purchase_state(h: &TestHarness, session_id: &str, item_id: &str) -> PurchaseState {
188 // MIN over the session's rows rather than `fetch_one`: `rows` is asserted
189 // separately, so a duplicate row must be reported as a count rather than
190 // blowing up the read that would have shown it.
191 let (rows, status, completed_at): (i64, Option<String>, Option<chrono::DateTime<chrono::Utc>>) =
192 sqlx::query_as(
193 "SELECT COUNT(*), MIN(status), MIN(completed_at) FROM transactions \
194 WHERE stripe_checkout_session_id = $1",
195 )
196 .bind(session_id)
197 .fetch_one(&h.db)
198 .await
199 .expect("read the session's transaction rows");
200
201 let sales_count: i32 = sqlx::query_scalar("SELECT sales_count FROM items WHERE id = $1::uuid")
202 .bind(item_id)
203 .fetch_one(&h.db)
204 .await
205 .expect("read item sales count");
206
207 let license_keys: i64 =
208 sqlx::query_scalar("SELECT COUNT(*) FROM license_keys WHERE item_id = $1::uuid")
209 .bind(item_id)
210 .fetch_one(&h.db)
211 .await
212 .expect("count license keys");
213
214 PurchaseState {
215 rows,
216 status: status.unwrap_or_default(),
217 completed_at,
218 sales_count,
219 license_keys,
220 }
221 }
222
223 /// How many times the event-id dedup recorded this delivery. Two rows would
224 /// mean the dedup key stopped being the primary key it is.
225 async fn processed_event_rows(h: &TestHarness, event_id: &str) -> i64 {
226 sqlx::query_scalar("SELECT COUNT(*) FROM processed_webhook_events WHERE event_id = $1")
227 .bind(event_id)
228 .fetch_one(&h.db)
229 .await
230 .expect("count processed webhook events")
231 }
232
233 /// Wait for the fire-and-forget mail tasks to settle and return the purchase
234 /// receipts sent to `address`.
235 ///
236 /// The receipt is spawned on the background pool, so it lands after the webhook
237 /// response and cannot be read synchronously. This is a wait on a condition
238 /// rather than a fixed sleep: it returns as soon as the count has held at
239 /// `expected` for a stretch, and it fails the instant a second copy appears,
240 /// which is the failure the redelivery test is hunting.
241 async fn settled_receipts(h: &TestHarness, address: &str, expected: usize) -> usize {
242 let transport = h
243 .mock_email
244 .as_ref()
245 .expect("with_mocks configures a mock email transport");
246 let mut stable = 0;
247 for _ in 0..400 {
248 let n = transport
249 .sent_to(address)
250 .into_iter()
251 .filter(|e| e.subject == "Your purchase is confirmed")
252 .count();
253 assert!(
254 n <= expected,
255 "{address} received {n} purchase receipts, contract allows {expected}"
256 );
257 if n == expected {
258 stable += 1;
259 if stable == 20 {
260 return n;
261 }
262 } else {
263 stable = 0;
264 }
265 tokio::time::sleep(std::time::Duration::from_millis(10)).await;
266 }
267 panic!("purchase receipts to {address} never settled at {expected}");
268 }
269
270 // ── checkout: redelivery ────────────────────────────────────────────────────
271
272 #[tokio::test]
273 async fn redelivered_checkout_session_credits_the_purchase_exactly_once() {
274 let mut h = TestHarness::with_mocks().await;
275 let session_id = "cs_exactly_once_replay";
276 let (buyer_id, seller_id, item_id) = pending_purchase(&mut h, "replay", session_id).await;
277 let event_id = "evt_exactly_once_replay";
278 let session = purchase_session(session_id, buyer_id, seller_id, &item_id, "paid");
279
280 let first = post_event(
281 &mut h,
282 event_id,
283 "checkout.session.completed",
284 session.clone(),
285 )
286 .await;
287 assert_eq!(first.status, 200, "first delivery rejected: {}", first.text);
288
289 let after_first = purchase_state(&h, session_id, &item_id).await;
290 assert_eq!(
291 after_first.status, "completed",
292 "the pending transaction must be completed by the first delivery"
293 );
294 let n = after_first.sales_count;
295 assert_eq!(n, 1, "one purchase is one sale, got {n}");
296 let n = after_first.license_keys;
297 assert_eq!(n, 1, "one purchase mints one license key, got {n}");
298 assert_eq!(
299 processed_event_rows(&h, event_id).await,
300 1,
301 "the event must be marked processed once the work committed"
302 );
303
304 // The redelivery Stripe actually sends: byte-identical event, same id.
305 let second = post_event(&mut h, event_id, "checkout.session.completed", session).await;
306 assert_eq!(second.status, 200, "redelivery rejected: {}", second.text);
307
308 let after_second = purchase_state(&h, session_id, &item_id).await;
309 let n = after_second.rows;
310 assert_eq!(
311 n, 1,
312 "the session must hold exactly one transaction row, got {n}"
313 );
314 assert_eq!(
315 after_second.completed_at, after_first.completed_at,
316 "completed_at moved: the redelivery re-ran the completion write"
317 );
318 let n = after_second.sales_count;
319 assert_eq!(n, 1, "the redelivery moved sales_count to {n}");
320 let n = after_second.license_keys;
321 assert_eq!(n, 1, "the redelivery left {n} license keys");
322 assert_eq!(
323 processed_event_rows(&h, event_id).await,
324 1,
325 "the processed-event mark must stay a single row"
326 );
327
328 // One receipt, and it quotes the credited amount in dollars-and-cents. A
329 // handler that read 1499 as dollars, or credited twice, says something else.
330 let receipts = settled_receipts(&h, "buyerreplay@test.com", 1).await;
331 assert_eq!(receipts, 1, "exactly one purchase receipt for one purchase");
332 let receipt = h
333 .mock_email
334 .as_ref()
335 .unwrap()
336 .sent_to("buyerreplay@test.com")
337 .into_iter()
338 .find(|e| e.subject == "Your purchase is confirmed")
339 .expect("receipt captured");
340 assert!(
341 receipt.body.contains("$14.99"),
342 "receipt must quote the 1499-cent price as $14.99, got: {}",
343 receipt.body
344 );
345 }
346
347 #[tokio::test]
348 async fn a_new_event_id_for_the_same_session_does_not_complete_it_twice() {
349 let mut h = TestHarness::with_mocks().await;
350 let session_id = "cs_exactly_once_newid";
351 let (buyer_id, seller_id, item_id) = pending_purchase(&mut h, "newid", session_id).await;
352 let session = purchase_session(session_id, buyer_id, seller_id, &item_id, "paid");
353
354 let first = post_event(
355 &mut h,
356 "evt_exactly_once_newid_a",
357 "checkout.session.completed",
358 session.clone(),
359 )
360 .await;
361 assert_eq!(first.status, 200, "first delivery failed: {}", first.text);
362 let after_first = purchase_state(&h, session_id, &item_id).await;
363
364 // A different event id defeats the event-id dedup on purpose: this is what
365 // is left when that layer does not fire, and the status-guarded UPDATE
366 // under it is the thing being pinned.
367 let second = post_event(
368 &mut h,
369 "evt_exactly_once_newid_b",
370 "checkout.session.completed",
371 session,
372 )
373 .await;
374 assert_eq!(second.status, 200, "second event rejected: {}", second.text);
375
376 let after_second = purchase_state(&h, session_id, &item_id).await;
377 assert_eq!(
378 after_second.completed_at, after_first.completed_at,
379 "the second event re-ran the completion write on an already-completed session"
380 );
381 let n = after_second.sales_count;
382 assert_eq!(
383 n, 1,
384 "sales_count must stay 1 across two event ids, got {n}"
385 );
386 let n = after_second.license_keys;
387 assert_eq!(
388 n, 1,
389 "crash-recovery finalize must mint no second key, got {n}"
390 );
391 }
392
393 // ── checkout: out-of-order settlement ───────────────────────────────────────
394
395 #[tokio::test]
396 async fn an_unpaid_checkout_delivers_nothing_until_the_settled_event_arrives() {
397 let mut h = TestHarness::with_mocks().await;
398 let session_id = "cs_exactly_once_async";
399 let (buyer_id, seller_id, item_id) = pending_purchase(&mut h, "async", session_id).await;
400
401 // An asynchronous method (ACH, SEPA) reports `unpaid` on
402 // checkout.session.completed. Funds are not captured, so nothing may ship.
403 let unpaid = purchase_session(session_id, buyer_id, seller_id, &item_id, "unpaid");
404 let deferred = post_event(
405 &mut h,
406 "evt_exactly_once_async_a",
407 "checkout.session.completed",
408 unpaid,
409 )
410 .await;
411 assert_eq!(deferred.status, 200, "unsettled session: {}", deferred.text);
412
413 let waiting = purchase_state(&h, session_id, &item_id).await;
414 assert_eq!(
415 waiting.status, "pending",
416 "an unpaid session must leave the transaction pending"
417 );
418 let n = waiting.sales_count;
419 assert_eq!(n, 0, "an unpaid session must not count a sale, got {n}");
420 let n = waiting.license_keys;
421 assert_eq!(
422 n, 0,
423 "an unpaid session must not mint a license key, got {n}"
424 );
425
426 // Stripe settles it later, under its own event id.
427 let paid = purchase_session(session_id, buyer_id, seller_id, &item_id, "paid");
428 let settled = post_event(
429 &mut h,
430 "evt_exactly_once_async_b",
431 "checkout.session.async_payment_succeeded",
432 paid,
433 )
434 .await;
435 assert_eq!(
436 settled.status, 200,
437 "settled event rejected: {}",
438 settled.text
439 );
440
441 let delivered = purchase_state(&h, session_id, &item_id).await;
442 assert_eq!(
443 delivered.status, "completed",
444 "the settled event must complete the transaction"
445 );
446 let n = delivered.sales_count;
447 assert_eq!(n, 1, "settlement counts exactly one sale, got {n}");
448 let n = delivered.license_keys;
449 assert_eq!(n, 1, "settlement mints exactly one license key, got {n}");
450 }
451
452 // ── checkout: signature ─────────────────────────────────────────────────────
453
454 #[tokio::test]
455 async fn a_forged_signature_is_refused_and_writes_nothing() {
456 let mut h = TestHarness::with_mocks().await;
457 let session_id = "cs_exactly_once_forged";
458 let (buyer_id, seller_id, item_id) = pending_purchase(&mut h, "forged", session_id).await;
459
460 let event_id = "evt_exactly_once_forged";
461 let payload = serde_json::json!({
462 "id": event_id,
463 "type": "checkout.session.completed",
464 "data": {"object": purchase_session(session_id, buyer_id, seller_id, &item_id, "paid")},
465 })
466 .to_string();
467
468 // Correctly shaped header, signed with a secret we do not hold. The payload
469 // is a real, fully valid completion, so only the signature separates this
470 // from the accepted delivery above.
471 let forged = sign_webhook_payload(&payload, "whsec_not_our_secret");
472 let resp = post_signed(&mut h, &payload, &forged).await;
473 assert_eq!(
474 resp.status, 400,
475 "a forged signature must be refused with 400: {}",
476 resp.text
477 );
478
479 let state = purchase_state(&h, session_id, &item_id).await;
480 assert_eq!(
481 state.status, "pending",
482 "a refused event must leave the transaction pending"
483 );
484 let n = state.sales_count;
485 assert_eq!(n, 0, "a refused event must not count a sale, got {n}");
486 let n = state.license_keys;
487 assert_eq!(n, 0, "a refused event must not mint a license key, got {n}");
488 assert_eq!(
489 processed_event_rows(&h, event_id).await,
490 0,
491 "a refused event must not be marked processed"
492 );
493 let queued: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM webhook_events")
494 .fetch_one(&h.db)
495 .await
496 .expect("count retry queue");
497 assert_eq!(
498 queued, 0,
499 "a signature failure is not a handler failure and must not enter the retry queue"
500 );
501 }
502
503 // ── checkout: Fan+ subscription creation ────────────────────────────────────
504
505 #[tokio::test]
506 async fn redelivered_fan_plus_checkout_creates_one_subscription() {
507 let mut h = TestHarness::with_mocks().await;
508 let user_id = h
509 .signup("fanplusonce", "fanplusonce@test.com", "password123")
510 .await;
511
512 let session = serde_json::json!({
513 "id": "cs_exactly_once_fanplus",
514 "object": "checkout.session",
515 "subscription": "sub_exactly_once_fanplus",
516 "customer": "cus_exactly_once_fanplus",
517 "payment_status": "paid",
518 "currency": "usd",
519 "metadata": {"checkout_type": "fan_plus", "user_id": user_id.to_string()},
520 });
521
522 let first = post_event(
523 &mut h,
524 "evt_exactly_once_fanplus_a",
525 "checkout.session.completed",
526 session.clone(),
527 )
528 .await;
529 assert_eq!(first.status, 200, "Fan+ checkout failed: {}", first.text);
530
531 // Redelivery under the same id (dedup) and under a fresh id (the ON CONFLICT
532 // guard beneath it): both must leave one active subscription, not two.
533 let replay = post_event(
534 &mut h,
535 "evt_exactly_once_fanplus_a",
536 "checkout.session.completed",
537 session.clone(),
538 )
539 .await;
540 assert_eq!(replay.status, 200, "Fan+ replay failed: {}", replay.text);
541 let reissued = post_event(
542 &mut h,
543 "evt_exactly_once_fanplus_b",
544 "checkout.session.completed",
545 session,
546 )
547 .await;
548 assert_eq!(reissued.status, 200, "Fan+ second event: {}", reissued.text);
549
550 let (rows, status): (i64, Option<String>) = sqlx::query_as(
551 "SELECT COUNT(*), MIN(status::text) FROM fan_plus_subscriptions WHERE user_id = $1",
552 )
553 .bind(user_id)
554 .fetch_one(&h.db)
555 .await
556 .expect("read fan+ subscriptions");
557 assert_eq!(
558 rows, 1,
559 "three deliveries of one Fan+ checkout must leave one subscription, got {rows}"
560 );
561 assert_eq!(status.as_deref(), Some("active"), "must still be active");
562 }
563