max / makenotwork

2.7 KB · 89 lines History Blame Raw

1	//! Idempotency key storage for safe POST retries.
2	//!
3	//! These functions are the storage layer for [`crate::metrics::idempotency_middleware`],
4	//! the global POST/PUT replay guard mounted on the router: it reads
5	//! [`get_cached_response`] before running a handler and writes [`store_response`]
6	//! (fire-and-forget, via the background pool) on a successful 2xx/3xx response,
7	//! scoped to `(key, user_id, method, path)`. The scheduler calls
8	//! [`cleanup_expired`] on its daily tick. This is distinct from, and additive
9	//! to, the structural checkout dedup (the `(buyer_id, item_id) WHERE
10	//! status='pending'` partial unique index): the index makes a double checkout
11	//! unrepresentable; this cache replays the response of any retried POST/PUT.
12	//! Exercised end-to-end by `tests/workflows/idempotency.rs`.
13
14	use crate::db::UserId;
15	use crate::error::Result;
16	use sqlx::PgPool;
17
18	/// A cached idempotency response.
19	#[derive(sqlx::FromRow)]
20	pub struct CachedResponse {
21	pub status_code: i16,
22	pub response_body: String,
23	}
24
25	/// Look up a cached response for an idempotency key.
26	/// Scoped to (key, user_id, method, path) to prevent cross-endpoint collisions.
27	#[tracing::instrument(skip_all)]
28	pub async fn get_cached_response(
29	pool: &PgPool,
30	key: &str,
31	user_id: UserId,
32	method: &str,
33	path: &str,
34	) -> Result<Option<CachedResponse>> {
35	let row = sqlx::query_as!(
36	CachedResponse,
37	"SELECT status_code, response_body FROM idempotency_keys WHERE key = $1 AND user_id = $2 AND method = $3 AND path = $4",
38	key,
39	user_id as UserId,
40	method,
41	path,
42	)
43	.fetch_optional(pool)
44	.await?;
45
46	Ok(row)
47	}
48
49	/// Store a response for an idempotency key. Uses ON CONFLICT to handle
50	/// race conditions (first writer wins).
51	#[tracing::instrument(skip_all)]
52	pub async fn store_response(
53	pool: &PgPool,
54	key: &str,
55	user_id: UserId,
56	method: &str,
57	path: &str,
58	status_code: u16,
59	response_body: &str,
60	) -> Result<()> {
61	sqlx::query!(
62	r#"INSERT INTO idempotency_keys (key, user_id, method, path, status_code, response_body)
63	VALUES ($1, $2, $3, $4, $5, $6)
64	ON CONFLICT (key, user_id, method, path) DO NOTHING"#,
65	key,
66	user_id as UserId,
67	method,
68	path,
69	status_code as i16,
70	response_body,
71	)
72	.execute(pool)
73	.await?;
74
75	Ok(())
76	}
77
78	/// Delete expired idempotency keys (older than 24 hours).
79	#[tracing::instrument(skip_all)]
80	pub async fn cleanup_expired(pool: &PgPool) -> Result<u64> {
81	let result = sqlx::query!(
82	"DELETE FROM idempotency_keys WHERE created_at < NOW() - INTERVAL '24 hours'",
83	)
84	.execute(pool)
85	.await?;
86
87	Ok(result.rows_affected())
88	}
89