max / makenotwork

5.8 KB · 168 lines History Blame Raw

1	//! Pending refunds queue for out-of-order webhook delivery.
2	//!
3	//! When a `charge.refunded` webhook arrives before its matching
4	//! `checkout.session.completed`, the refund data is stored here.
5	//! The scheduler and checkout handler both check for pending matches.
6
7	use sqlx::PgPool;
8
9	use super::validated_types::Cents;
10
11	use crate::error::Result;
12
13	/// Insert a pending refund for later matching. Deduplicates on
14	/// `payment_intent_id`; if a pending (unmatched) refund already exists
15	/// for this payment intent, the insert is silently skipped.
16	pub async fn insert_pending_refund(
17	pool: &PgPool,
18	payment_intent_id: &str,
19	amount: i64,
20	amount_refunded: i64,
21	) -> Result<()> {
22	sqlx::query!(
23	r#"
24	INSERT INTO pending_refunds (payment_intent_id, amount, amount_refunded)
25	VALUES ($1, $2, $3)
26	ON CONFLICT (payment_intent_id) WHERE matched_at IS NULL DO NOTHING
27	"#,
28	payment_intent_id,
29	amount,
30	amount_refunded,
31	)
32	.execute(pool)
33	.await?;
34
35	Ok(())
36	}
37
38	/// Row from the pending_refunds table.
39	#[derive(Debug, sqlx::FromRow)]
40	pub struct PendingRefund {
41	pub id: uuid::Uuid,
42	pub payment_intent_id: String,
43	pub amount: Cents,
44	pub amount_refunded: Cents,
45	}
46
47	/// Claim a pending refund matching a payment intent ID.
48	///
49	/// Atomically marks it as matched (so it is only claimed once) but NOT completed,
50	/// completion is recorded separately by [`mark_refund_completed`] only after the
51	/// fallible refund work succeeds. A claim that is never completed (process killed
52	/// mid-refund) leaves `completed_at IS NULL`, so the stale-refund sweep surfaces it
53	/// for human escalation (PAY-S1). Returns `None` if no unmatched pending refund exists.
54	pub async fn claim_pending_refund(
55	pool: &PgPool,
56	payment_intent_id: &str,
57	) -> Result<Option<PendingRefund>> {
58	let row = sqlx::query_as!(
59	PendingRefund,
60	r#"
61	UPDATE pending_refunds
62	SET matched_at = NOW()
63	WHERE id = (
64	SELECT id FROM pending_refunds
65	WHERE payment_intent_id = $1 AND matched_at IS NULL
66	LIMIT 1
67	FOR UPDATE SKIP LOCKED
68	)
69	RETURNING id, payment_intent_id, amount AS "amount: Cents", amount_refunded AS "amount_refunded: Cents"
70	"#,
71	payment_intent_id,
72	)
73	.fetch_optional(pool)
74	.await?;
75
76	Ok(row)
77	}
78
79	/// Record that a claimed pending refund's processing finished successfully.
80	///
81	/// Sets `completed_at`; only after this is the row considered fully handled. A
82	/// claimed row without a `completed_at` (the process died between claim and this
83	/// call) is surfaced by [`get_stale_refunds`] for manual reconciliation instead of
84	/// being auto-retried, re-issuing a refund that may already have reached Stripe
85	/// could double-refund (PAY-S1).
86	pub async fn mark_refund_completed(pool: &PgPool, id: uuid::Uuid) -> Result<()> {
87	sqlx::query!(
88	"UPDATE pending_refunds SET completed_at = NOW() WHERE id = $1",
89	id
90	)
91	.execute(pool)
92	.await?;
93	Ok(())
94	}
95
96	/// Release a claimed pending refund back to the queue (`matched_at` → NULL) after a
97	/// graceful processing failure (a transient error where the handler committed
98	/// nothing, it is atomic). Releasing re-opens the row so a later webhook delivery
99	/// can re-claim and retry. A non-graceful failure (process killed) cannot reach
100	/// here; that row stays matched-but-incomplete and is escalated by the sweep
101	/// instead (PAY-S1). Idempotent.
102	pub async fn unclaim_pending_refund(pool: &PgPool, id: uuid::Uuid) -> Result<()> {
103	sqlx::query!(
104	"UPDATE pending_refunds SET matched_at = NULL WHERE id = $1",
105	id
106	)
107	.execute(pool)
108	.await?;
109	Ok(())
110	}
111
112	/// Row for stale pending refunds that need escalation.
113	#[derive(Debug, sqlx::FromRow)]
114	pub struct StaleRefund {
115	pub id: uuid::Uuid,
116	pub payment_intent_id: String,
117	pub amount: Cents,
118	pub amount_refunded: Cents,
119	pub created_at: chrono::DateTime<chrono::Utc>,
120	}
121
122	/// Per-tick cap on the stale-refund escalation sweep. It runs every scheduler
123	/// tick under the tick-wide advisory lock; escalation is idempotent (sets
124	/// `escalated_at`), so a bound here drains a backlog across ticks instead of
125	/// letting one unbounded query stall the tick.
126	pub const STALE_REFUND_BATCH: i64 = 100;
127
128	/// Get up to [`STALE_REFUND_BATCH`] pending refunds older than `age` that still
129	/// need attention and have not been escalated, oldest first. "Need attention" means
130	/// the refund work never completed: either the row was never matched to a payment
131	/// (`matched_at IS NULL`), or it was claimed but the process died before recording
132	/// completion (`matched_at IS NOT NULL AND completed_at IS NULL`), the crash-window
133	/// case (PAY-S1). Both surface here for human reconciliation.
134	pub async fn get_stale_refunds(pool: &PgPool, age: chrono::Duration) -> Result<Vec<StaleRefund>> {
135	let cutoff = chrono::Utc::now() - age;
136	// runtime-checked: binds a chrono `DateTime<Utc>` cutoff (`$1`); a bind
137	// parameter's type can't be overridden in the macro when sqlx's `time` and
138	// `chrono` features are unified. See db::pending_uploads::get_stale_pending_uploads.
139	let rows = sqlx::query_as::<_, StaleRefund>(
140	r"
141	SELECT id, payment_intent_id, amount, amount_refunded, created_at
142	FROM pending_refunds
143	WHERE completed_at IS NULL
144	AND escalated_at IS NULL
145	AND created_at < $1
146	ORDER BY created_at
147	LIMIT $2
148	",
149	)
150	.bind(cutoff)
151	.bind(STALE_REFUND_BATCH)
152	.fetch_all(pool)
153	.await?;
154
155	Ok(rows)
156	}
157
158	/// Mark a pending refund as escalated (alert sent, won't be re-alerted).
159	pub async fn mark_escalated(pool: &PgPool, id: uuid::Uuid) -> Result<()> {
160	sqlx::query!(
161	"UPDATE pending_refunds SET escalated_at = NOW() WHERE id = $1",
162	id
163	)
164	.execute(pool)
165	.await?;
166	Ok(())
167	}
168