Skip to main content

max / makenotwork

payments: idempotent re-runnable webhook finalize (ultra-fuzz Run 4 M-Pay1, A+) Webhook secondary effects (license-key mint, revenue splits, mailing-list, bundle grants) ran after the DB commit but before the event was marked processed. The dedup is check-then-act, so a genuine duplicate is dropped early and the complete_transaction Ok(None) branch is reached only on crash-recovery redelivery (tx already completed, event unmarked). That branch only escalated, so a first attempt that crashed mid-finalize left the buyer with a completed purchase and no key/splits, never retried. - Consolidate the purchase/cart/guest effect blocks into one re-runnable finalize_purchase_transaction / finalize_guest_transaction so the paths cannot drift and a redelivery re-runs them. - Wire both branches of all three handlers: on Ok(None)/empty, re-fetch the session's completed rows and re-run the finalizer; escalate only if none exist (genuinely orphaned). - Make every effect idempotent: migration 151 adds a partial unique index on license_keys(transaction_id) and a unique index on revenue_splits(transaction_id, recipient_id); maybe_generate_license_key pre-checks for an existing key; create_transaction_splits uses ON CONFLICT DO NOTHING. Double-mint / double-split are now structurally impossible. - Integration test: a wiped-finalize redelivery backfills key + split and a second redelivery is a no-op; plus a direct unique-index rejection test. Also (gate-enabling, pre-existing): db/tips.rs string queries -> query_as! (A+ cold spot); clear two pre-existing clippy lints (lib.rs test-module ordering, synckit/rotation.rs type_complexity); refresh two stale slug tests to assert the deliberate Run 2 auto-suffix seal instead of the old reject behavior (they had been red since that seal landed). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-06-23 22:32 UTC
Commit: c929c2b2cc25ff666b36ad8149ebeaa1eabf2c61
Parent: 83a4e68
22 files changed, +1257 insertions, -167 deletions
A server/.sqlx/query-1b173822c0dfe66ea3bb9ae1aea044cb353aaeab38da1e015629fe1b571d19bd.json +22
@@ -0,0 +1,22 @@
1 + {
2 + "db_name": "PostgreSQL",
3 + "query": "SELECT COALESCE(SUM(amount_cents), 0)::BIGINT AS \"total!\" FROM tips WHERE recipient_id = $1 AND status = 'completed'",
4 + "describe": {
5 + "columns": [
6 + {
7 + "ordinal": 0,
8 + "name": "total!",
9 + "type_info": "Int8"
10 + }
11 + ],
12 + "parameters": {
13 + "Left": [
14 + "Uuid"
15 + ]
16 + },
17 + "nullable": [
18 + null
19 + ]
20 + },
21 + "hash": "1b173822c0dfe66ea3bb9ae1aea044cb353aaeab38da1e015629fe1b571d19bd"
22 + }
A server/.sqlx/query-365d41311e8f51f763af4850927b74bfdab786ec86ebe575dc560c4cc22600ae.json +70
@@ -0,0 +1,70 @@
1 + {
2 + "db_name": "PostgreSQL",
3 + "query": "\n SELECT id AS \"id: LicenseKeyId\", item_id AS \"item_id: ItemId\",\n owner_id AS \"owner_id: UserId\", transaction_id AS \"transaction_id: TransactionId\",\n key_code AS \"key_code: KeyCode\", max_activations, activation_count,\n revoked_at AS \"revoked_at: chrono::DateTime<chrono::Utc>\",\n created_at AS \"created_at: chrono::DateTime<chrono::Utc>\"\n FROM license_keys WHERE transaction_id = $1\n ",
4 + "describe": {
5 + "columns": [
6 + {
7 + "ordinal": 0,
8 + "name": "id: LicenseKeyId",
9 + "type_info": "Uuid"
10 + },
11 + {
12 + "ordinal": 1,
13 + "name": "item_id: ItemId",
14 + "type_info": "Uuid"
15 + },
16 + {
17 + "ordinal": 2,
18 + "name": "owner_id: UserId",
19 + "type_info": "Uuid"
20 + },
21 + {
22 + "ordinal": 3,
23 + "name": "transaction_id: TransactionId",
24 + "type_info": "Uuid"
25 + },
26 + {
27 + "ordinal": 4,
28 + "name": "key_code: KeyCode",
29 + "type_info": "Varchar"
30 + },
31 + {
32 + "ordinal": 5,
33 + "name": "max_activations",
34 + "type_info": "Int4"
35 + },
36 + {
37 + "ordinal": 6,
38 + "name": "activation_count",
39 + "type_info": "Int4"
40 + },
41 + {
42 + "ordinal": 7,
43 + "name": "revoked_at: chrono::DateTime<chrono::Utc>",
44 + "type_info": "Timestamptz"
45 + },
46 + {
47 + "ordinal": 8,
48 + "name": "created_at: chrono::DateTime<chrono::Utc>",
49 + "type_info": "Timestamptz"
50 + }
51 + ],
52 + "parameters": {
53 + "Left": [
54 + "Uuid"
55 + ]
56 + },
57 + "nullable": [
58 + false,
59 + false,
60 + false,
61 + true,
62 + false,
63 + true,
64 + false,
65 + true,
66 + false
67 + ]
68 + },
69 + "hash": "365d41311e8f51f763af4850927b74bfdab786ec86ebe575dc560c4cc22600ae"
70 + }
A server/.sqlx/query-36e369ea83eb4c8337f4a4c3ccde3c5ceabb68b38d458060ec45352f67a1c86e.json +90
@@ -0,0 +1,90 @@
1 + {
2 + "db_name": "PostgreSQL",
3 + "query": "\n SELECT\n id AS \"id: TipId\", tipper_id AS \"tipper_id: UserId\", recipient_id AS \"recipient_id: UserId\",\n project_id AS \"project_id: ProjectId\", amount_cents AS \"amount_cents: Cents\", message,\n status AS \"status: super::TransactionStatus\", stripe_payment_intent_id, stripe_checkout_session_id,\n stripe_transfer_group,\n created_at AS \"created_at: chrono::DateTime<chrono::Utc>\",\n completed_at AS \"completed_at: chrono::DateTime<chrono::Utc>\"\n FROM tips\n WHERE tipper_id = $1 AND status = 'completed'\n ORDER BY created_at DESC\n LIMIT $2 OFFSET $3\n ",
4 + "describe": {
5 + "columns": [
6 + {
7 + "ordinal": 0,
8 + "name": "id: TipId",
9 + "type_info": "Uuid"
10 + },
11 + {
12 + "ordinal": 1,
13 + "name": "tipper_id: UserId",
14 + "type_info": "Uuid"
15 + },
16 + {
17 + "ordinal": 2,
18 + "name": "recipient_id: UserId",
19 + "type_info": "Uuid"
20 + },
21 + {
22 + "ordinal": 3,
23 + "name": "project_id: ProjectId",
24 + "type_info": "Uuid"
25 + },
26 + {
27 + "ordinal": 4,
28 + "name": "amount_cents: Cents",
29 + "type_info": "Int4"
30 + },
31 + {
32 + "ordinal": 5,
33 + "name": "message",
34 + "type_info": "Varchar"
35 + },
36 + {
37 + "ordinal": 6,
38 + "name": "status: super::TransactionStatus",
39 + "type_info": "Varchar"
40 + },
41 + {
42 + "ordinal": 7,
43 + "name": "stripe_payment_intent_id",
44 + "type_info": "Varchar"
45 + },
46 + {
47 + "ordinal": 8,
48 + "name": "stripe_checkout_session_id",
49 + "type_info": "Varchar"
50 + },
51 + {
52 + "ordinal": 9,
53 + "name": "stripe_transfer_group",
54 + "type_info": "Varchar"
55 + },
56 + {
57 + "ordinal": 10,
58 + "name": "created_at: chrono::DateTime<chrono::Utc>",
59 + "type_info": "Timestamptz"
60 + },
61 + {
62 + "ordinal": 11,
63 + "name": "completed_at: chrono::DateTime<chrono::Utc>",
64 + "type_info": "Timestamptz"
65 + }
66 + ],
67 + "parameters": {
68 + "Left": [
69 + "Uuid",
70 + "Int8",
71 + "Int8"
72 + ]
73 + },
74 + "nullable": [
75 + false,
76 + false,
77 + false,
78 + true,
79 + false,
80 + true,
81 + false,
82 + true,
83 + true,
84 + true,
85 + false,
86 + true
87 + ]
88 + },
89 + "hash": "36e369ea83eb4c8337f4a4c3ccde3c5ceabb68b38d458060ec45352f67a1c86e"
90 + }
A server/.sqlx/query-6b1d0dbffd82a8902fa1e2cad2ae7c26135dca1ab462ee72d09bda12b4b3bb7d.json +89
@@ -0,0 +1,89 @@
1 + {
2 + "db_name": "PostgreSQL",
3 + "query": "\n UPDATE tips\n SET status = 'completed',\n stripe_payment_intent_id = $2,\n completed_at = NOW()\n WHERE stripe_checkout_session_id = $1\n AND status = 'pending'\n RETURNING\n id AS \"id: TipId\", tipper_id AS \"tipper_id: UserId\", recipient_id AS \"recipient_id: UserId\",\n project_id AS \"project_id: ProjectId\", amount_cents AS \"amount_cents: Cents\", message,\n status AS \"status: super::TransactionStatus\", stripe_payment_intent_id, stripe_checkout_session_id,\n stripe_transfer_group,\n created_at AS \"created_at: chrono::DateTime<chrono::Utc>\",\n completed_at AS \"completed_at: chrono::DateTime<chrono::Utc>\"\n ",
4 + "describe": {
5 + "columns": [
6 + {
7 + "ordinal": 0,
8 + "name": "id: TipId",
9 + "type_info": "Uuid"
10 + },
11 + {
12 + "ordinal": 1,
13 + "name": "tipper_id: UserId",
14 + "type_info": "Uuid"
15 + },
16 + {
17 + "ordinal": 2,
18 + "name": "recipient_id: UserId",
19 + "type_info": "Uuid"
20 + },
21 + {
22 + "ordinal": 3,
23 + "name": "project_id: ProjectId",
24 + "type_info": "Uuid"
25 + },
26 + {
27 + "ordinal": 4,
28 + "name": "amount_cents: Cents",
29 + "type_info": "Int4"
30 + },
31 + {
32 + "ordinal": 5,
33 + "name": "message",
34 + "type_info": "Varchar"
35 + },
36 + {
37 + "ordinal": 6,
38 + "name": "status: super::TransactionStatus",
39 + "type_info": "Varchar"
40 + },
41 + {
42 + "ordinal": 7,
43 + "name": "stripe_payment_intent_id",
44 + "type_info": "Varchar"
45 + },
46 + {
47 + "ordinal": 8,
48 + "name": "stripe_checkout_session_id",
49 + "type_info": "Varchar"
50 + },
51 + {
52 + "ordinal": 9,
53 + "name": "stripe_transfer_group",
54 + "type_info": "Varchar"
55 + },
56 + {
57 + "ordinal": 10,
58 + "name": "created_at: chrono::DateTime<chrono::Utc>",
59 + "type_info": "Timestamptz"
60 + },
61 + {
62 + "ordinal": 11,
63 + "name": "completed_at: chrono::DateTime<chrono::Utc>",
64 + "type_info": "Timestamptz"
65 + }
66 + ],
67 + "parameters": {
68 + "Left": [
69 + "Text",
70 + "Varchar"
71 + ]
72 + },
73 + "nullable": [
74 + false,
75 + false,
76 + false,
77 + true,
78 + false,
79 + true,
80 + false,
81 + true,
82 + true,
83 + true,
84 + false,
85 + true
86 + ]
87 + },
88 + "hash": "6b1d0dbffd82a8902fa1e2cad2ae7c26135dca1ab462ee72d09bda12b4b3bb7d"
89 + }
A server/.sqlx/query-8f93450ef8d93e5aa9955348623030d3779b9e1903a16a94b8107646ef0f3ef3.json +22
@@ -0,0 +1,22 @@
1 + {
2 + "db_name": "PostgreSQL",
3 + "query": "SELECT COUNT(*) AS \"count!\" FROM tips WHERE recipient_id = $1 AND status = 'completed'",
4 + "describe": {
5 + "columns": [
6 + {
7 + "ordinal": 0,
8 + "name": "count!",
9 + "type_info": "Int8"
10 + }
11 + ],
12 + "parameters": {
13 + "Left": [
14 + "Uuid"
15 + ]
16 + },
17 + "nullable": [
18 + null
19 + ]
20 + },
21 + "hash": "8f93450ef8d93e5aa9955348623030d3779b9e1903a16a94b8107646ef0f3ef3"
22 + }
A server/.sqlx/query-c442eef4152cb7aa4fcf4298f5ce170211c902f831a787eef055242d85dcbfe6.json +148
@@ -0,0 +1,148 @@
1 + {
2 + "db_name": "PostgreSQL",
3 + "query": "\n SELECT\n id AS \"id: TransactionId\", buyer_id AS \"buyer_id: UserId\", seller_id AS \"seller_id: UserId\",\n item_id AS \"item_id: ItemId\", amount_cents AS \"amount_cents: Cents\", platform_fee_cents AS \"platform_fee_cents: Cents\",\n currency, status AS \"status: super::TransactionStatus\", stripe_payment_intent_id, stripe_checkout_session_id,\n created_at AS \"created_at: chrono::DateTime<chrono::Utc>\", completed_at AS \"completed_at: chrono::DateTime<chrono::Utc>\",\n item_title, seller_username, share_contact, project_id AS \"project_id: ProjectId\",\n parent_transaction_id AS \"parent_transaction_id: TransactionId\", promo_code_id AS \"promo_code_id: PromoCodeId\",\n guest_email, claim_token AS \"claim_token: ClaimToken\", claimed_by AS \"claimed_by: UserId\",\n download_token AS \"download_token: DownloadToken\"\n FROM transactions\n WHERE stripe_checkout_session_id = $1 AND status = 'completed'\n ",
4 + "describe": {
5 + "columns": [
6 + {
7 + "ordinal": 0,
8 + "name": "id: TransactionId",
9 + "type_info": "Uuid"
10 + },
11 + {
12 + "ordinal": 1,
13 + "name": "buyer_id: UserId",
14 + "type_info": "Uuid"
15 + },
16 + {
17 + "ordinal": 2,
18 + "name": "seller_id: UserId",
19 + "type_info": "Uuid"
20 + },
21 + {
22 + "ordinal": 3,
23 + "name": "item_id: ItemId",
24 + "type_info": "Uuid"
25 + },
26 + {
27 + "ordinal": 4,
28 + "name": "amount_cents: Cents",
29 + "type_info": "Int4"
30 + },
31 + {
32 + "ordinal": 5,
33 + "name": "platform_fee_cents: Cents",
34 + "type_info": "Int4"
35 + },
36 + {
37 + "ordinal": 6,
38 + "name": "currency",
39 + "type_info": "Varchar"
40 + },
41 + {
42 + "ordinal": 7,
43 + "name": "status: super::TransactionStatus",
44 + "type_info": "Varchar"
45 + },
46 + {
47 + "ordinal": 8,
48 + "name": "stripe_payment_intent_id",
49 + "type_info": "Varchar"
50 + },
51 + {
52 + "ordinal": 9,
53 + "name": "stripe_checkout_session_id",
54 + "type_info": "Varchar"
55 + },
56 + {
57 + "ordinal": 10,
58 + "name": "created_at: chrono::DateTime<chrono::Utc>",
59 + "type_info": "Timestamptz"
60 + },
61 + {
62 + "ordinal": 11,
63 + "name": "completed_at: chrono::DateTime<chrono::Utc>",
64 + "type_info": "Timestamptz"
65 + },
66 + {
67 + "ordinal": 12,
68 + "name": "item_title",
69 + "type_info": "Varchar"
70 + },
71 + {
72 + "ordinal": 13,
73 + "name": "seller_username",
74 + "type_info": "Varchar"
75 + },
76 + {
77 + "ordinal": 14,
78 + "name": "share_contact",
79 + "type_info": "Bool"
80 + },
81 + {
82 + "ordinal": 15,
83 + "name": "project_id: ProjectId",
84 + "type_info": "Uuid"
85 + },
86 + {
87 + "ordinal": 16,
88 + "name": "parent_transaction_id: TransactionId",
89 + "type_info": "Uuid"
90 + },
91 + {
92 + "ordinal": 17,
93 + "name": "promo_code_id: PromoCodeId",
94 + "type_info": "Uuid"
95 + },
96 + {
97 + "ordinal": 18,
98 + "name": "guest_email",
99 + "type_info": "Varchar"
100 + },
101 + {
102 + "ordinal": 19,
103 + "name": "claim_token: ClaimToken",
104 + "type_info": "Uuid"
105 + },
106 + {
107 + "ordinal": 20,
108 + "name": "claimed_by: UserId",
109 + "type_info": "Uuid"
110 + },
111 + {
112 + "ordinal": 21,
113 + "name": "download_token: DownloadToken",
114 + "type_info": "Uuid"
115 + }
116 + ],
117 + "parameters": {
118 + "Left": [
119 + "Text"
120 + ]
121 + },
122 + "nullable": [
123 + false,
124 + true,
125 + true,
126 + true,
127 + false,
128 + false,
129 + false,
130 + false,
131 + true,
132 + true,
133 + false,
134 + true,
135 + true,
136 + true,
137 + false,
138 + true,
139 + true,
140 + true,
141 + true,
142 + true,
143 + true,
144 + true
145 + ]
146 + },
147 + "hash": "c442eef4152cb7aa4fcf4298f5ce170211c902f831a787eef055242d85dcbfe6"
148 + }
A server/.sqlx/query-c71cc7d733de0e4bebfd201d882a1a98213c4178481d0ef90baeb75314b66adf.json +84
@@ -0,0 +1,84 @@
1 + {
2 + "db_name": "PostgreSQL",
3 + "query": "\n SELECT t.id AS \"id: TipId\", t.tipper_id AS \"tipper_id: UserId\", t.recipient_id AS \"recipient_id: UserId\",\n t.project_id AS \"project_id: ProjectId\", t.amount_cents AS \"amount_cents: Cents\",\n t.message, t.status AS \"status: super::TransactionStatus\",\n t.created_at AS \"created_at: chrono::DateTime<chrono::Utc>\",\n t.completed_at AS \"completed_at: chrono::DateTime<chrono::Utc>\",\n u.username AS tipper_username, u.display_name AS tipper_display_name\n FROM tips t\n JOIN users u ON u.id = t.tipper_id\n WHERE t.recipient_id = $1 AND t.status = 'completed'\n ORDER BY t.created_at DESC\n LIMIT $2 OFFSET $3\n ",
4 + "describe": {
5 + "columns": [
6 + {
7 + "ordinal": 0,
8 + "name": "id: TipId",
9 + "type_info": "Uuid"
10 + },
11 + {
12 + "ordinal": 1,
13 + "name": "tipper_id: UserId",
14 + "type_info": "Uuid"
15 + },
16 + {
17 + "ordinal": 2,
18 + "name": "recipient_id: UserId",
19 + "type_info": "Uuid"
20 + },
21 + {
22 + "ordinal": 3,
23 + "name": "project_id: ProjectId",
24 + "type_info": "Uuid"
25 + },
26 + {
27 + "ordinal": 4,
28 + "name": "amount_cents: Cents",
29 + "type_info": "Int4"
30 + },
31 + {
32 + "ordinal": 5,
33 + "name": "message",
34 + "type_info": "Varchar"
35 + },
36 + {
37 + "ordinal": 6,
38 + "name": "status: super::TransactionStatus",
39 + "type_info": "Varchar"
40 + },
41 + {
42 + "ordinal": 7,
43 + "name": "created_at: chrono::DateTime<chrono::Utc>",
44 + "type_info": "Timestamptz"
45 + },
46 + {
47 + "ordinal": 8,
48 + "name": "completed_at: chrono::DateTime<chrono::Utc>",
49 + "type_info": "Timestamptz"
50 + },
51 + {
52 + "ordinal": 9,
53 + "name": "tipper_username",
54 + "type_info": "Varchar"
55 + },
56 + {
57 + "ordinal": 10,
58 + "name": "tipper_display_name",
59 + "type_info": "Varchar"
60 + }
61 + ],
62 + "parameters": {
63 + "Left": [
64 + "Uuid",
65 + "Int8",
66 + "Int8"
67 + ]
68 + },
69 + "nullable": [
70 + false,
71 + false,
72 + false,
73 + true,
74 + false,
75 + true,
76 + false,
77 + false,
78 + true,
79 + false,
80 + true
81 + ]
82 + },
83 + "hash": "c71cc7d733de0e4bebfd201d882a1a98213c4178481d0ef90baeb75314b66adf"
84 + }
A server/.sqlx/query-e8aff7effadd1600fcf9496564f52028340cc9187cae2778984fe90e9e3eda77.json +93
@@ -0,0 +1,93 @@
1 + {
2 + "db_name": "PostgreSQL",
3 + "query": "\n INSERT INTO tips (tipper_id, recipient_id, project_id, amount_cents, message, stripe_checkout_session_id)\n VALUES ($1, $2, $3, $4, $5, $6)\n RETURNING\n id AS \"id: TipId\", tipper_id AS \"tipper_id: UserId\", recipient_id AS \"recipient_id: UserId\",\n project_id AS \"project_id: ProjectId\", amount_cents AS \"amount_cents: Cents\", message,\n status AS \"status: super::TransactionStatus\", stripe_payment_intent_id, stripe_checkout_session_id,\n stripe_transfer_group,\n created_at AS \"created_at: chrono::DateTime<chrono::Utc>\",\n completed_at AS \"completed_at: chrono::DateTime<chrono::Utc>\"\n ",
4 + "describe": {
5 + "columns": [
6 + {
7 + "ordinal": 0,
8 + "name": "id: TipId",
9 + "type_info": "Uuid"
10 + },
11 + {
12 + "ordinal": 1,
13 + "name": "tipper_id: UserId",
14 + "type_info": "Uuid"
15 + },
16 + {
17 + "ordinal": 2,
18 + "name": "recipient_id: UserId",
19 + "type_info": "Uuid"
20 + },
21 + {
22 + "ordinal": 3,
23 + "name": "project_id: ProjectId",
24 + "type_info": "Uuid"
25 + },
26 + {
27 + "ordinal": 4,
28 + "name": "amount_cents: Cents",
29 + "type_info": "Int4"
30 + },
31 + {
32 + "ordinal": 5,
33 + "name": "message",
34 + "type_info": "Varchar"
35 + },
36 + {
37 + "ordinal": 6,
38 + "name": "status: super::TransactionStatus",
39 + "type_info": "Varchar"
40 + },
41 + {
42 + "ordinal": 7,
43 + "name": "stripe_payment_intent_id",
44 + "type_info": "Varchar"
45 + },
46 + {
47 + "ordinal": 8,
48 + "name": "stripe_checkout_session_id",
49 + "type_info": "Varchar"
50 + },
51 + {
52 + "ordinal": 9,
53 + "name": "stripe_transfer_group",
54 + "type_info": "Varchar"
55 + },
56 + {
57 + "ordinal": 10,
58 + "name": "created_at: chrono::DateTime<chrono::Utc>",
59 + "type_info": "Timestamptz"
60 + },
61 + {
62 + "ordinal": 11,
63 + "name": "completed_at: chrono::DateTime<chrono::Utc>",
64 + "type_info": "Timestamptz"
65 + }
66 + ],
67 + "parameters": {
68 + "Left": [
69 + "Uuid",
70 + "Uuid",
71 + "Uuid",
72 + "Int4",
73 + "Varchar",
74 + "Varchar"
75 + ]
76 + },
77 + "nullable": [
78 + false,
79 + false,
80 + false,
81 + true,
82 + false,
83 + true,
84 + false,
85 + true,
86 + true,
87 + true,
88 + false,
89 + true
90 + ]
91 + },
92 + "hash": "e8aff7effadd1600fcf9496564f52028340cc9187cae2778984fe90e9e3eda77"
93 + }
A server/.sqlx/query-ea0c7de9b7e1a7eaa6be491ba0acac1fe935f83bcaaa925f6e133c2b5f033ea9.json +14
@@ -0,0 +1,14 @@
1 + {
2 + "db_name": "PostgreSQL",
3 + "query": "\n UPDATE tips\n SET status = 'refunded'\n WHERE stripe_payment_intent_id = $1 AND status = 'completed'\n ",
4 + "describe": {
5 + "columns": [],
6 + "parameters": {
7 + "Left": [
8 + "Text"
9 + ]
10 + },
11 + "nullable": []
12 + },
13 + "hash": "ea0c7de9b7e1a7eaa6be491ba0acac1fe935f83bcaaa925f6e133c2b5f033ea9"
14 + }
A server/migrations/151_payment_finalize_idempotency.sql +24
@@ -0,0 +1,24 @@
1 + -- Payment finalize idempotency (ultra-fuzz Payments A+, finding M-Pay1).
2 + --
3 + -- Webhook secondary effects (license-key mint, revenue splits, etc.) run after
4 + -- the transaction is committed. On crash-recovery redelivery the finalizer
5 + -- re-runs, so the side-effecting writes must be structurally safe to repeat.
6 + -- These two indexes are the backstop that makes a double-mint / double-split
7 + -- impossible even if application-level pre-checks are bypassed.
8 + --
9 + -- NOTE: both indexes are created NON-CONCURRENTLY (sqlx runs each migration
10 + -- inside its own transaction, and CREATE INDEX CONCURRENTLY cannot run there).
11 + -- Acceptable at current table sizes; revisit if license_keys / revenue_splits
12 + -- grow large enough that the brief write lock matters.
13 +
14 + -- One auto-minted license key per purchase. The partial predicate excludes the
15 + -- manually-created keys (transaction_id IS NULL) that creators issue by hand,
16 + -- so it constrains only purchase-minted keys.
17 + CREATE UNIQUE INDEX IF NOT EXISTS license_keys_transaction_id_key
18 + ON license_keys (transaction_id)
19 + WHERE transaction_id IS NOT NULL;
20 +
21 + -- One revenue split per recipient per transaction. Lets create_transaction_splits
22 + -- use ON CONFLICT DO NOTHING so a finalize re-run is a no-op.
23 + CREATE UNIQUE INDEX IF NOT EXISTS revenue_splits_tx_recipient_key
24 + ON revenue_splits (transaction_id, recipient_id);
M server/src/db/license_keys.rs +28
@@ -93,6 +93,34 @@ pub async fn get_license_key_by_code(pool: &PgPool, key_code: &KeyCode) -> Resul
93 93 Ok(key)
94 94 }
95 95
96 + /// Look up the auto-minted license key for a purchase transaction, if any.
97 + ///
98 + /// Used by the finalize pre-check so a crash-recovery redelivery does not mint
99 + /// a second key (the `license_keys_transaction_id_key` partial unique index is
100 + /// the structural backstop). At most one such key exists per transaction.
101 + #[tracing::instrument(skip_all)]
102 + pub async fn get_license_key_by_transaction_id(
103 + pool: &PgPool,
104 + transaction_id: TransactionId,
105 + ) -> Result<Option<DbLicenseKey>> {
106 + let key = sqlx::query_as!(
107 + DbLicenseKey,
108 + r#"
109 + SELECT id AS "id: LicenseKeyId", item_id AS "item_id: ItemId",
110 + owner_id AS "owner_id: UserId", transaction_id AS "transaction_id: TransactionId",
111 + key_code AS "key_code: KeyCode", max_activations, activation_count,
112 + revoked_at AS "revoked_at: chrono::DateTime<chrono::Utc>",
113 + created_at AS "created_at: chrono::DateTime<chrono::Utc>"
114 + FROM license_keys WHERE transaction_id = $1
115 + "#,
116 + transaction_id as TransactionId,
117 + )
118 + .fetch_optional(pool)
119 + .await?;
120 +
121 + Ok(key)
122 + }
123 +
96 124 /// Get a license key by ID.
97 125 #[tracing::instrument(skip_all)]
98 126 pub async fn get_license_key_by_id(pool: &PgPool, id: LicenseKeyId) -> Result<Option<DbLicenseKey>> {
M server/src/db/project_members.rs +3
@@ -248,10 +248,13 @@ pub async fn create_transaction_splits(
248 248 let recipient_ids: Vec<UserId> = splits.iter().map(|(id, _, _)| *id).collect();
249 249 let amounts: Vec<i32> = splits.iter().map(|(_, a, _)| *a as i32).collect();
250 250 let percents: Vec<i16> = splits.iter().map(|(_, _, p)| *p).collect();
251 + // ON CONFLICT DO NOTHING (against revenue_splits_tx_recipient_key): a
252 + // crash-recovery finalize re-run records no duplicate splits.
251 253 sqlx::query(
252 254 r#"
253 255 INSERT INTO revenue_splits (transaction_id, recipient_id, amount_cents, split_percent, status)
254 256 SELECT $1, UNNEST($2::uuid[]), UNNEST($3::int[]), UNNEST($4::smallint[]), 'pending'
257 + ON CONFLICT (transaction_id, recipient_id) DO NOTHING
255 258 "#,
256 259 )
257 260 .bind(transaction_id)
M server/src/db/synckit/rotation.rs +5 -2
@@ -103,6 +103,9 @@ pub async fn get_key_rotation(
103 103 Ok(rotation)
104 104 }
105 105
106 + /// One sync-log entry awaiting re-encryption: (seq, table_name, row_id, data).
107 + pub type RotationEntry = (i64, String, String, Option<JsonValue>);
108 +
106 109 /// Pull sync log entries that need re-encryption (key_id != new_key_id).
107 110 /// Returns entries ordered by seq, paginated by after_seq.
108 111 #[tracing::instrument(skip_all)]
@@ -113,8 +116,8 @@ pub async fn get_rotation_entries(
113 116 new_key_id: i32,
114 117 after_seq: i64,
115 118 limit: i64,
116 - ) -> Result<Vec<(i64, String, String, Option<JsonValue>)>> {
117 - let entries: Vec<(i64, String, String, Option<JsonValue>)> = sqlx::query_as(
119 + ) -> Result<Vec<RotationEntry>> {
120 + let entries: Vec<RotationEntry> = sqlx::query_as(
118 121 r#"
119 122 SELECT seq, table_name, row_id, data FROM sync_log
120 123 WHERE app_id = $1 AND user_id = $2
M server/src/db/tips.rs +62 -35
@@ -4,6 +4,7 @@ use sqlx::PgPool;
4 4
5 5 use super::id_types::*;
6 6 use super::models::*;
7 + use super::validated_types::Cents;
7 8 use crate::error::Result;
8 9
9 10 /// Create a pending tip record before redirecting to Stripe Checkout.
@@ -17,19 +18,26 @@ pub async fn create_tip(
17 18 message: Option<&str>,
18 19 stripe_checkout_session_id: &str,
19 20 ) -> Result<DbTip> {
20 - let tip = sqlx::query_as::<_, DbTip>(
21 + let tip = sqlx::query_as!(
22 + DbTip,
21 23 r#"
22 24 INSERT INTO tips (tipper_id, recipient_id, project_id, amount_cents, message, stripe_checkout_session_id)
23 25 VALUES ($1, $2, $3, $4, $5, $6)
24 - RETURNING *
26 + RETURNING
27 + id AS "id: TipId", tipper_id AS "tipper_id: UserId", recipient_id AS "recipient_id: UserId",
28 + project_id AS "project_id: ProjectId", amount_cents AS "amount_cents: Cents", message,
29 + status AS "status: super::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
30 + stripe_transfer_group,
31 + created_at AS "created_at: chrono::DateTime<chrono::Utc>",
32 + completed_at AS "completed_at: chrono::DateTime<chrono::Utc>"
25 33 "#,
34 + tipper_id as UserId,
35 + recipient_id as UserId,
36 + project_id as Option<ProjectId>,
37 + amount_cents,
38 + message,
39 + stripe_checkout_session_id,
26 40 )
27 - .bind(tipper_id)
28 - .bind(recipient_id)
29 - .bind(project_id)
30 - .bind(amount_cents)
31 - .bind(message)
32 - .bind(stripe_checkout_session_id)
33 41 .fetch_one(pool)
34 42 .await?;
35 43
@@ -39,12 +47,13 @@ pub async fn create_tip(
39 47 /// Mark a tip as completed after Stripe confirms payment.
40 48 /// Returns `Some(tip)` if updated, `None` if already completed (idempotent).
41 49 #[tracing::instrument(skip(executor))]
42 - pub async fn complete_tip(
43 - executor: impl sqlx::Executor<'_, Database = sqlx::Postgres>,
50 + pub async fn complete_tip<'e>(
51 + executor: impl sqlx::PgExecutor<'e>,
44 52 stripe_checkout_session_id: &str,
45 53 stripe_payment_intent_id: &str,
46 54 ) -> Result<Option<DbTip>> {
47 - let tip = sqlx::query_as::<_, DbTip>(
55 + let tip = sqlx::query_as!(
56 + DbTip,
48 57 r#"
49 58 UPDATE tips
50 59 SET status = 'completed',
@@ -52,11 +61,17 @@ pub async fn complete_tip(
52 61 completed_at = NOW()
53 62 WHERE stripe_checkout_session_id = $1
54 63 AND status = 'pending'
55 - RETURNING *
64 + RETURNING
65 + id AS "id: TipId", tipper_id AS "tipper_id: UserId", recipient_id AS "recipient_id: UserId",
66 + project_id AS "project_id: ProjectId", amount_cents AS "amount_cents: Cents", message,
67 + status AS "status: super::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
68 + stripe_transfer_group,
69 + created_at AS "created_at: chrono::DateTime<chrono::Utc>",
70 + completed_at AS "completed_at: chrono::DateTime<chrono::Utc>"
56 71 "#,
72 + stripe_checkout_session_id,
73 + stripe_payment_intent_id,
57 74 )
58 - .bind(stripe_checkout_session_id)
59 - .bind(stripe_payment_intent_id)
60 75 .fetch_optional(executor)
61 76 .await?;
62 77
@@ -71,10 +86,14 @@ pub async fn get_tips_received(
71 86 limit: i64,
72 87 offset: i64,
73 88 ) -> Result<Vec<DbTipWithUser>> {
74 - let tips = sqlx::query_as::<_, DbTipWithUser>(
89 + let tips = sqlx::query_as!(
90 + DbTipWithUser,
75 91 r#"
76 - SELECT t.id, t.tipper_id, t.recipient_id, t.project_id, t.amount_cents,
77 - t.message, t.status, t.created_at, t.completed_at,
92 + SELECT t.id AS "id: TipId", t.tipper_id AS "tipper_id: UserId", t.recipient_id AS "recipient_id: UserId",
93 + t.project_id AS "project_id: ProjectId", t.amount_cents AS "amount_cents: Cents",
94 + t.message, t.status AS "status: super::TransactionStatus",
95 + t.created_at AS "created_at: chrono::DateTime<chrono::Utc>",
96 + t.completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
78 97 u.username AS tipper_username, u.display_name AS tipper_display_name
79 98 FROM tips t
80 99 JOIN users u ON u.id = t.tipper_id
@@ -82,10 +101,10 @@ pub async fn get_tips_received(
82 101 ORDER BY t.created_at DESC
83 102 LIMIT $2 OFFSET $3
84 103 "#,
104 + recipient_id as UserId,
105 + limit,
106 + offset,
85 107 )
86 - .bind(recipient_id)
87 - .bind(limit)
88 - .bind(offset)
89 108 .fetch_all(pool)
90 109 .await?;
91 110
@@ -95,41 +114,41 @@ pub async fn get_tips_received(
95 114 /// Total tip revenue received by a creator (completed tips only).
96 115 #[tracing::instrument(skip(pool))]
97 116 pub async fn total_tips_received(pool: &PgPool, recipient_id: UserId) -> Result<i64> {
98 - let row: (Option<i64>,) = sqlx::query_as(
99 - "SELECT SUM(amount_cents)::BIGINT FROM tips WHERE recipient_id = $1 AND status = 'completed'",
117 + let total = sqlx::query_scalar!(
118 + r#"SELECT COALESCE(SUM(amount_cents), 0)::BIGINT AS "total!" FROM tips WHERE recipient_id = $1 AND status = 'completed'"#,
119 + recipient_id as UserId,
100 120 )
101 - .bind(recipient_id)
102 121 .fetch_one(pool)
103 122 .await?;
104 123
105 - Ok(row.0.unwrap_or(0))
124 + Ok(total)
106 125 }
107 126
108 127 /// Count of completed tips received.
109 128 #[tracing::instrument(skip(pool))]
110 129 pub async fn count_tips_received(pool: &PgPool, recipient_id: UserId) -> Result<i64> {
111 - let row: (i64,) = sqlx::query_as(
112 - "SELECT COUNT(*) FROM tips WHERE recipient_id = $1 AND status = 'completed'",
130 + let count = sqlx::query_scalar!(
131 + r#"SELECT COUNT(*) AS "count!" FROM tips WHERE recipient_id = $1 AND status = 'completed'"#,
132 + recipient_id as UserId,
113 133 )
114 - .bind(recipient_id)
115 134 .fetch_one(pool)
116 135 .await?;
117 136
118 - Ok(row.0)
137 + Ok(count)
119 138 }
120 139
121 140 /// Mark a tip as refunded by payment intent ID.
122 141 /// Returns true if a tip was refunded, false if not found (idempotent).
123 142 #[tracing::instrument(skip(pool))]
124 143 pub async fn refund_tip_by_payment_intent(pool: &PgPool, payment_intent_id: &str) -> Result<bool> {
125 - let result = sqlx::query(
144 + let result = sqlx::query!(
126 145 r#"
127 146 UPDATE tips
128 147 SET status = 'refunded'
129 148 WHERE stripe_payment_intent_id = $1 AND status = 'completed'
130 149 "#,
150 + payment_intent_id,
131 151 )
132 - .bind(payment_intent_id)
133 152 .execute(pool)
134 153 .await?;
135 154
@@ -145,17 +164,25 @@ pub async fn get_tips_sent(
145 164 limit: i64,
146 165 offset: i64,
147 166 ) -> Result<Vec<DbTip>> {
148 - let tips = sqlx::query_as::<_, DbTip>(
167 + let tips = sqlx::query_as!(
168 + DbTip,
149 169 r#"
150 - SELECT * FROM tips
170 + SELECT
171 + id AS "id: TipId", tipper_id AS "tipper_id: UserId", recipient_id AS "recipient_id: UserId",
172 + project_id AS "project_id: ProjectId", amount_cents AS "amount_cents: Cents", message,
173 + status AS "status: super::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
174 + stripe_transfer_group,
175 + created_at AS "created_at: chrono::DateTime<chrono::Utc>",
176 + completed_at AS "completed_at: chrono::DateTime<chrono::Utc>"
177 + FROM tips
151 178 WHERE tipper_id = $1 AND status = 'completed'
152 179 ORDER BY created_at DESC
153 180 LIMIT $2 OFFSET $3
154 181 "#,
182 + tipper_id as UserId,
183 + limit,
184 + offset,
155 185 )
156 - .bind(tipper_id)
157 - .bind(limit)
158 - .bind(offset)
159 186 .fetch_all(pool)
160 187 .await?;
161 188
M server/src/db/transactions.rs +35
@@ -292,6 +292,41 @@ pub async fn complete_cart_transactions<'e>(
292 292 Ok(txs)
293 293 }
294 294
295 + /// Fetch all completed transactions for a checkout session.
296 + ///
297 + /// Used on the crash-recovery branch of the purchase/cart webhook handlers: when
298 + /// `complete_transaction` / `complete_cart_transactions` return nothing (the
299 + /// rows were already flipped to completed by a first attempt that crashed before
300 + /// running finalize), this re-reads those completed rows so finalize can re-run
301 + /// idempotently. Covers single and cart purchases since both key on the session.
302 + #[tracing::instrument(skip_all)]
303 + pub async fn get_completed_transactions_for_session<'e>(
304 + executor: impl sqlx::PgExecutor<'e>,
305 + stripe_checkout_session_id: &str,
306 + ) -> Result<Vec<DbTransaction>> {
307 + let txs = sqlx::query_as!(
308 + DbTransaction,
309 + r#"
310 + SELECT
311 + id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
312 + item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
313 + currency, status AS "status: super::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
314 + created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
315 + item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
316 + parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
317 + guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
318 + download_token AS "download_token: DownloadToken"
319 + FROM transactions
320 + WHERE stripe_checkout_session_id = $1 AND status = 'completed'
321 + "#,
322 + stripe_checkout_session_id,
323 + )
324 + .fetch_all(executor)
325 + .await?;
326 +
327 + Ok(txs)
328 + }
329 +
295 330 /// List transactions where the user is the buyer, newest first.
296 331 ///
297 332 /// Pass `limit: None` for all rows (exports), or `Some(n)` for dashboard display.
M server/src/lib.rs +29 -29
@@ -304,35 +304,6 @@ async fn request_timeout_middleware(
304 304 }
305 305 }
306 306
307 - #[cfg(test)]
308 - mod timeout_exempt_tests {
309 - use super::timeout_exempt;
310 -
311 - #[test]
312 - fn exempts_only_anchored_long_running_routes() {
313 - // Genuinely long / streaming routes stay exempt.
314 - for p in [
315 - "/git/foo/bar.git/info/refs",
316 - "/api/export/content",
317 - "/api/internal/creator/export/sales",
318 - "/api/sync/subscribe",
319 - "/api/v1/sync/subscribe",
320 - "/api/sync/ota/apps/x/releases",
321 - "/api/v1/sync/ota/slug/macos/arm64/1.0.0",
322 - ] {
323 - assert!(timeout_exempt(p), "{p} should be exempt");
324 - }
325 - // The substring-match hazard: these contain a token but must NOT be exempt.
326 - for p in [
327 - "/dashboard/export", // a normal page, not a long export
328 - "/u/somecreator/export-notes", // creator-controlled slug
329 - "/items/sync/ota-recap", // happens to contain the token mid-path
330 - ] {
331 - assert!(!timeout_exempt(p), "{p} must not be exempt");
332 - }
333 - }
334 - }
335 -
336 307 /// Middleware that sets security headers on all responses.
337 308 /// Embed routes (`/embed/`) get permissive frame headers for iframe embedding.
338 309 async fn security_headers_middleware(
@@ -429,3 +400,32 @@ async fn security_headers_middleware(
429 400 );
430 401 response
431 402 }
403 +
404 + #[cfg(test)]
405 + mod timeout_exempt_tests {
406 + use super::timeout_exempt;
407 +
408 + #[test]
409 + fn exempts_only_anchored_long_running_routes() {
410 + // Genuinely long / streaming routes stay exempt.
411 + for p in [
412 + "/git/foo/bar.git/info/refs",
413 + "/api/export/content",
414 + "/api/internal/creator/export/sales",
415 + "/api/sync/subscribe",
416 + "/api/v1/sync/subscribe",
417 + "/api/sync/ota/apps/x/releases",
418 + "/api/v1/sync/ota/slug/macos/arm64/1.0.0",
419 + ] {
420 + assert!(timeout_exempt(p), "{p} should be exempt");
421 + }
422 + // The substring-match hazard: these contain a token but must NOT be exempt.
423 + for p in [
424 + "/dashboard/export", // a normal page, not a long export
425 + "/u/somecreator/export-notes", // creator-controlled slug
426 + "/items/sync/ota-recap", // happens to contain the token mid-path
427 + ] {
428 + assert!(!timeout_exempt(p), "{p} must not be exempt");
429 + }
430 + }
431 + }
M server/src/routes/stripe/webhook/checkout.rs +70 -90
@@ -9,9 +9,8 @@ use crate::{
9 9 };
10 10
11 11 use super::checkout_helpers::{
12 - check_pending_refund, maybe_generate_license_key, record_tip_splits,
13 - record_transaction_splits, send_guest_sale_notification, send_purchase_emails,
14 - send_tip_email, subscribe_buyer_to_mailing_list,
12 + check_pending_refund, finalize_guest_transaction, finalize_purchase_transaction,
13 + record_tip_splits, send_tip_email,
15 14 };
16 15
17 16 /// A `checkout.session.completed` produced no transaction to complete. Tell a
@@ -131,29 +130,9 @@ pub(super) async fn handle_purchase_checkout_completed(
131 130 db_tx.commit().await.context("commit purchase webhook transaction")?;
132 131
133 132 // --- Secondary effects below (outside transaction) ---
134 -
135 - // Grant access to bundle child items (if this is a bundle)
136 - if let Some(iid) = item_id
137 - && let Ok(Some(purchased_item)) = db::items::get_item_by_id(&state.db, iid).await
138 - && purchased_item.item_type == db::ItemType::Bundle
139 - {
140 - crate::routes::stripe::checkout::grant_bundle_items(state, iid, buyer_id, seller_id, Some(tx.id)).await;
141 - }
142 -
143 - if tx.share_contact {
144 - db::transactions::clear_contact_revocation(&state.db, buyer_id, seller_id)
145 - .await
146 - .context("clear contact revocation after purchase")?;
147 - }
148 -
149 - // Record revenue splits if the item's project has members
150 - if let Some(iid) = item_id {
151 - record_transaction_splits(state, tx.id, iid, tx.amount_cents).await;
152 - maybe_generate_license_key(state, iid, buyer_id, tx.id).await;
153 - subscribe_buyer_to_mailing_list(state, iid, buyer_id);
154 - }
155 -
156 - send_purchase_emails(state, &tx, buyer_id, seller_id);
133 + // Consolidated in one re-runnable finalizer so the purchase and cart
134 + // paths can't drift and a crash-recovery redelivery re-runs safely.
135 + finalize_purchase_transaction(state, &tx, buyer_id, seller_id).await;
157 136
158 137 if let Err(e) = db::subscriptions::log_subscription_event(
159 138 &state.db, None, event_id, "checkout.session.completed.purchase",
@@ -167,7 +146,27 @@ pub(super) async fn handle_purchase_checkout_completed(
167 146 check_pending_refund(state, &payment_intent_id).await;
168 147 }
169 148 Ok(None) => {
170 - escalate_if_orphaned_session(state, &session_id, &payment_intent_id, "transaction").await?;
149 + // No row flipped to completed. Either this is a benign duplicate of
150 + // an already-finalized purchase, OR the first attempt crashed AFTER
151 + // committing the completed status but BEFORE running finalize — in
152 + // which case the buyer holds a completed purchase with no license
153 + // key / no splits. Re-fetch completed rows for the session and
154 + // re-run the idempotent finalizer; only escalate if none exist
155 + // (genuinely orphaned: payment took, rows never created).
156 + let completed = db::transactions::get_completed_transactions_for_session(&state.db, &session_id)
157 + .await
158 + .context("re-fetch completed transactions for crash recovery")?;
159 + if completed.is_empty() {
160 + escalate_if_orphaned_session(state, &session_id, &payment_intent_id, "transaction").await?;
161 + } else {
162 + for tx in &completed {
163 + tracing::info!(
164 + session_id = %session_id, transaction_id = %tx.id,
165 + "crash-recovery: re-running finalize for already-completed session"
166 + );
167 + finalize_purchase_transaction(state, tx, buyer_id, seller_id).await;
168 + }
169 + }
171 170 }
172 171 Err(e) => {
173 172 tracing::error!(session_id = %session_id, error = ?e, "failed to complete transaction");
@@ -204,7 +203,24 @@ pub(super) async fn handle_cart_checkout_completed(
204 203 .context("complete cart transactions")?;
205 204
206 205 if completed_txs.is_empty() {
207 - escalate_if_orphaned_session(state, &session_id, &payment_intent_id, "cart transactions").await?;
206 + // Same crash-recovery shape as the single-item handler: re-fetch
207 + // completed rows for the session and re-run the idempotent finalizer
208 + // before falling through to orphan escalation.
209 + db_tx.commit().await.ok();
210 + let completed = db::transactions::get_completed_transactions_for_session(&state.db, &session_id)
211 + .await
212 + .context("re-fetch completed cart transactions for crash recovery")?;
213 + if completed.is_empty() {
214 + escalate_if_orphaned_session(state, &session_id, &payment_intent_id, "cart transactions").await?;
215 + } else {
216 + for tx in &completed {
217 + tracing::info!(
218 + session_id = %session_id, transaction_id = %tx.id,
219 + "crash-recovery: re-running finalize for already-completed cart session"
220 + );
221 + finalize_purchase_transaction(state, tx, buyer_id, seller_id).await;
222 + }
223 + }
208 224 return Ok(());
209 225 }
210 226
@@ -257,40 +273,12 @@ pub(super) async fn handle_cart_checkout_completed(
257 273 .context("remove cart items after successful payment")?;
258 274
259 275 // --- Secondary effects (outside transaction) ---
260 -
261 - for tx in &completed_txs {
262 - if let Some(item_id) = tx.item_id {
263 - // Bundle grants
264 - if let Ok(Some(purchased_item)) = db::items::get_item_by_id(&state.db, item_id).await
265 - && purchased_item.item_type == db::ItemType::Bundle
266 - {
267 - crate::routes::stripe::checkout::grant_bundle_items(
268 - state, item_id, buyer_id, seller_id, Some(tx.id),
269 - )
270 - .await;
271 - }
272 -
273 - // Revenue splits
274 - record_transaction_splits(state, tx.id, item_id, tx.amount_cents).await;
275 -
276 - // License keys
277 - maybe_generate_license_key(state, item_id, buyer_id, tx.id).await;
278 -
279 - // Mailing list
280 - subscribe_buyer_to_mailing_list(state, item_id, buyer_id);
281 - }
282 - }
283 -
284 - // Contact sharing (once per seller)
285 - if completed_txs.iter().any(|t| t.share_contact) {
286 - db::transactions::clear_contact_revocation(&state.db, buyer_id, seller_id)
287 - .await
288 - .context("clear contact revocation after cart purchase")?;
289 - }
290 -
291 - // Send purchase emails for each item (reuse existing per-item emails)
276 + // One re-runnable finalizer per transaction; shared with the single-item
277 + // path so the effect blocks can't drift. Idempotent on a crash-recovery
278 + // redelivery. (clear_contact_revocation runs per-tx that opted in, which is
279 + // a no-op once already cleared.)
292 280 for tx in &completed_txs {
293 - send_purchase_emails(state, tx, buyer_id, seller_id);
281 + finalize_purchase_transaction(state, tx, buyer_id, seller_id).await;
294 282 }
295 283
296 284 // Log event
@@ -722,38 +710,30 @@ pub(super) async fn handle_guest_checkout_completed(
722 710 db_tx.commit().await.context("commit guest checkout webhook transaction")?;
723 711
724 712 // --- Secondary effects below (outside transaction) ---
725 -
726 - // A guest purchase is never auto-attached, so there's no buyer yet:
727 - // the license key (if any) is minted when the buyer claims, in
728 - // `claim_purchase`. Revenue splits are recorded now regardless.
729 - record_transaction_splits(state, tx.id, meta.item_id, tx.amount_cents).await;
730 -
731 - // Always send the guest purchase confirmation with the claim link.
732 - if let (Some(download_token), Some(claim_token)) = (tx.download_token, tx.claim_token)
733 - {
734 - let email_client = state.email.clone();
735 - let host_url = state.config.host_url.clone();
736 - let item_title = tx.item_title.clone().unwrap_or_else(|| "your item".to_string());
737 - let price = helpers::format_price(tx.amount_cents);
738 - let guest_email_addr = guest_email.clone();
739 - let download_url = format!("{}/download/{}", host_url, download_token);
740 - let claim_url = format!("{}/claim?token={}", host_url, claim_token);
741 -
742 - state.bg.spawn("guest purchase confirmation", async move {
743 - if let Err(e) = email_client.send_guest_purchase_confirmation(
744 - &guest_email_addr, &item_title, &price, &download_url, &claim_url,
745 - ).await {
746 - tracing::error!(error = ?e, "failed to send guest purchase confirmation email");
747 - }
748 - });
749 - }
750 -
751 - // Send sale notification to seller
752 - send_guest_sale_notification(state, &tx, &guest_email, meta.seller_id);
713 + // One re-runnable finalizer (splits + confirmation + sale email).
714 + finalize_guest_transaction(state, &tx, &guest_email, meta.item_id, meta.seller_id);
753 715 }
754 716 None => {
755 717 db_tx.commit().await.ok();
756 - escalate_if_orphaned_session(state, &session_id, &payment_intent_id, "guest transaction").await?;
718 + // Crash-recovery: re-fetch completed guest rows for the session and
719 + // re-run the idempotent finalizer; only escalate if none exist.
720 + let completed = db::transactions::get_completed_transactions_for_session(&state.db, &session_id)
721 + .await
722 + .context("re-fetch completed guest transactions for crash recovery")?;
723 + if completed.is_empty() {
724 + escalate_if_orphaned_session(state, &session_id, &payment_intent_id, "guest transaction").await?;
725 + } else {
726 + for tx in &completed {
727 + let guest_email = tx.guest_email.clone().unwrap_or_else(|| guest_email.clone());
728 + tracing::info!(
729 + session_id = %session_id, transaction_id = %tx.id,
730 + "crash-recovery: re-running finalize for already-completed guest session"
731 + );
732 + if let Some(item_id) = tx.item_id {
733 + finalize_guest_transaction(state, tx, &guest_email, item_id, meta.seller_id);
734 + }
735 + }
736 + }
757 737 }
758 738 }
759 739
M server/src/routes/stripe/webhook/checkout_helpers.rs +107
@@ -19,6 +19,22 @@ pub(crate) async fn maybe_generate_license_key(
19 19 _ => return,
20 20 };
21 21
22 + // Idempotency pre-check: a crash-recovery redelivery re-runs finalize, but a
23 + // purchase mints at most one auto key. If one already exists for this
24 + // transaction, skip the mint. The `license_keys_transaction_id_key` partial
25 + // unique index is the structural backstop if this check is ever bypassed.
26 + match db::license_keys::get_license_key_by_transaction_id(&state.db, transaction_id).await {
27 + Ok(Some(_)) => {
28 + tracing::debug!(transaction_id = %transaction_id, item_id = %item_id, "license key already minted for transaction; skipping");
29 + return;
30 + }
31 + Ok(None) => {}
32 + Err(e) => {
33 + tracing::error!(transaction_id = %transaction_id, error = ?e, "failed to check for existing license key; skipping mint to avoid duplicate");
34 + return;
35 + }
36 + }
37 +
22 38 let key_code = helpers::generate_key_code();
23 39 match db::license_keys::create_license_key(
24 40 &state.db, item_id, buyer_id, Some(transaction_id),
@@ -41,6 +57,50 @@ pub(crate) async fn maybe_generate_license_key(
41 57 }
42 58 }
43 59
60 + /// Run every secondary effect of a completed (logged-in) purchase, in order:
61 + /// bundle grants, contact-revocation clear, revenue splits, license-key mint,
62 + /// mailing-list subscribe, and the purchase/sale emails.
63 + ///
64 + /// This is the single place the purchase and cart handlers funnel their effect
65 + /// blocks through, so the two can't drift, and it is safe to re-run: a
66 + /// crash-recovery redelivery (transaction already completed, event not yet
67 + /// marked processed) re-invokes it. Every DB effect here is now idempotent
68 + /// (ON CONFLICT writes or a pre-check guarded by a unique index). The fire-and-
69 + /// forget emails may re-send on that rare redelivery; that is acceptable and
70 + /// consistent with the existing webhook architecture (handlers are idempotent
71 + /// on data, best-effort on notifications).
72 + pub(super) async fn finalize_purchase_transaction(
73 + state: &AppState,
74 + tx: &db::DbTransaction,
75 + buyer_id: db::UserId,
76 + seller_id: db::UserId,
77 + ) {
78 + // Grant access to bundle child items (if this purchase is a bundle).
79 + if let Some(item_id) = tx.item_id
80 + && let Ok(Some(purchased_item)) = db::items::get_item_by_id(&state.db, item_id).await
81 + && purchased_item.item_type == db::ItemType::Bundle
82 + {
83 + crate::routes::stripe::checkout::grant_bundle_items(state, item_id, buyer_id, seller_id, Some(tx.id)).await;
84 + }
85 +
86 + // Contact-revocation clear (if the buyer opted to share contact).
87 + if tx.share_contact
88 + && let Err(e) = db::transactions::clear_contact_revocation(&state.db, buyer_id, seller_id).await
89 + {
90 + tracing::error!(transaction_id = %tx.id, error = ?e, "failed to clear contact revocation after purchase");
91 + }
92 +
93 + // Revenue splits, license key, mailing list (each keyed to the item).
94 + if let Some(item_id) = tx.item_id {
95 + record_transaction_splits(state, tx.id, item_id, tx.amount_cents).await;
96 + maybe_generate_license_key(state, item_id, buyer_id, tx.id).await;
97 + subscribe_buyer_to_mailing_list(state, item_id, buyer_id);
98 + }
99 +
100 + // Purchase confirmation + sale notification (fire-and-forget).
101 + send_purchase_emails(state, tx, buyer_id, seller_id);
102 + }
103 +
44 104 /// Send purchase confirmation to buyer and sale notification to seller (fire-and-forget).
45 105 pub(super) fn send_purchase_emails(
46 106 state: &AppState,
@@ -287,6 +347,53 @@ fn compute_splits(
287 347 splits
288 348 }
289 349
350 + /// Run every secondary effect of a completed guest purchase, in order: revenue
351 + /// splits, the guest purchase confirmation (with claim + download links), and
352 + /// the seller sale notification.
353 + ///
354 + /// Mirrors [`finalize_purchase_transaction`] but for the guest path, which has
355 + /// no buyer account yet (the license key, if any, is minted at claim time in
356 + /// `claim_purchase`, not here). Re-runnable on a crash-recovery redelivery:
357 + /// splits go through an ON CONFLICT write and the emails are fire-and-forget
358 + /// (a re-send on that rare redelivery is acceptable).
359 + pub(super) fn finalize_guest_transaction(
360 + state: &AppState,
361 + tx: &db::DbTransaction,
362 + guest_email: &str,
363 + item_id: db::ItemId,
364 + seller_id: db::UserId,
365 + ) {
366 + // Revenue splits (idempotent).
367 + let state_for_splits = state.clone();
368 + let tx_id = tx.id;
369 + let amount_cents = tx.amount_cents;
370 + state.bg.spawn("guest revenue splits", async move {
371 + record_transaction_splits(&state_for_splits, tx_id, item_id, amount_cents).await;
372 + });
373 +
374 + // Guest purchase confirmation with the claim link (fire-and-forget).
375 + if let (Some(download_token), Some(claim_token)) = (tx.download_token, tx.claim_token) {
376 + let email_client = state.email.clone();
377 + let host_url = state.config.host_url.clone();
378 + let item_title = tx.item_title.clone().unwrap_or_else(|| "your item".to_string());
379 + let price = helpers::format_price(tx.amount_cents);
380 + let guest_email_addr = guest_email.to_string();
381 + let download_url = format!("{}/download/{}", host_url, download_token);
382 + let claim_url = format!("{}/claim?token={}", host_url, claim_token);
383 +
384 + state.bg.spawn("guest purchase confirmation", async move {
385 + if let Err(e) = email_client.send_guest_purchase_confirmation(
386 + &guest_email_addr, &item_title, &price, &download_url, &claim_url,
387 + ).await {
388 + tracing::error!(error = ?e, "failed to send guest purchase confirmation email");
389 + }
390 + });
391 + }
392 +
393 + // Sale notification to the seller (fire-and-forget).
394 + send_guest_sale_notification(state, tx, guest_email, seller_id);
395 + }
396 +
290 397 /// Send sale notification to the seller for a guest purchase.
291 398 pub(super) fn send_guest_sale_notification(
292 399 state: &AppState,
M server/tests/workflows/adversarial_input.rs +20 -7
@@ -258,23 +258,36 @@ async fn sql_injection_in_title_harmless() {
258 258 // Duplicate creation
259 259 // =============================================================================
260 260
261 - /// Vulnerability tested: Duplicate project slug creates confusion or overwrites.
262 - /// Second project with same slug should be rejected (unique constraint).
261 + /// Vulnerability tested: a duplicate project slug must never overwrite the
262 + /// existing project or surface a raw 500. The create path auto-suffixes the
263 + /// collision (`inputtest-proj` -> `inputtest-proj-2`) via `insert_with_unique_slug`
264 + /// and retries — the deliberate Run 2 UX seal. The second project gets a fresh
265 + /// id and a distinct slug, so no confusion or overwrite is possible.
263 266 #[tokio::test]
264 - async fn duplicate_project_slug_rejected() {
267 + async fn duplicate_project_slug_auto_suffixed() {
265 268 let mut h = TestHarness::new().await;
266 - let (_project_id, _item_id) = setup_creator_with_item(&mut h).await;
269 + let (project_id, _item_id) = setup_creator_with_item(&mut h).await;
267 270
268 - // Try creating another project with the same slug
271 + // Creating another project with the same slug succeeds with a suffixed slug.
269 272 let resp = h
270 273 .client
271 274 .post_form("/api/projects", "slug=inputtest-proj&title=Duplicate+Shop")
272 275 .await;
273 276 assert!(
274 - !resp.status.is_success(),
275 - "Duplicate slug must not succeed: {} {}",
277 + resp.status.is_success(),
278 + "Duplicate slug should auto-suffix, not fail: {} {}",
276 279 resp.status, resp.text
277 280 );
281 + let second: serde_json::Value = resp.json();
282 + assert_eq!(
283 + second["slug"], "inputtest-proj-2",
284 + "collision must auto-suffix to inputtest-proj-2: {}", second
285 + );
286 + assert_ne!(
287 + second["id"].as_str().unwrap(),
288 + project_id,
289 + "the duplicate must be a new project, never an overwrite"
290 + );
278 291 }
279 292
280 293 /// Vulnerability tested: Duplicate username on signup.
M server/tests/workflows/mod.rs +1
@@ -84,6 +84,7 @@ mod imports;
84 84 mod media_library;
85 85 mod synckit_sse;
86 86 mod mock_payment_flows;
87 + mod payment_crash_recovery;
87 88 mod revenue_splits;
88 89 mod synckit_selective;
89 90 mod guest_checkout;
A server/tests/workflows/payment_crash_recovery.rs +227
M server/tests/workflows/project_management.rs +14 -4