Skip to main content

max / makenotwork

server: convert pending_uploads/pending_s3_deletions/pending_refunds to compile-checked SQL (ultra-fuzz Run #1 --deep Phase 4) Convert the runtime sqlx::query() strings in three more money-path modules to query!/query_as! macros validated against the schema at build time, refreshing the committed .sqlx cache. Following mt-db's playbook, queries that bind a chrono DateTime parameter stay runtime-checked (the macro infers the TIMESTAMPTZ bind as time::OffsetDateTime under the unified time+chrono features, and a bind parameter's type can't be overridden) — get_stale_pending_uploads, get_stale_pending, get_stale_refunds. db::pending_s3_deletions::is_s3_key_live also stays runtime: its SQL is built dynamically from the S3_KEY_REFS registry. Cents/UserId columns use output-type overrides. Gate: SQLX_OFFLINE=true build + storage/webhook/purchase/scanning workflows (88) green offline; clippy clean. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-06-22 22:57 UTC
Commit: eb3b19a115b42e53be32b5fb8d86c8d8657ef708
Parent: 897378b
13 files changed, +215 insertions, -32 deletions
A server/.sqlx/query-37fb85cee293011c18dac118334b6bb00bacce263c26aa9a49a47331a98d2971.json +16
@@ -0,0 +1,16 @@
1 + {
2 + "db_name": "PostgreSQL",
3 + "query": "DELETE FROM pending_uploads WHERE s3_key = $1 AND user_id = $2 AND bucket = $3",
4 + "describe": {
5 + "columns": [],
6 + "parameters": {
7 + "Left": [
8 + "Text",
9 + "Uuid",
10 + "Text"
11 + ]
12 + },
13 + "nullable": []
14 + },
15 + "hash": "37fb85cee293011c18dac118334b6bb00bacce263c26aa9a49a47331a98d2971"
16 + }
A server/.sqlx/query-442ca0ce542463afe513e17133bfd22c3a2d1e45ddb90064a64bf30b64925196.json +14
@@ -0,0 +1,14 @@
1 + {
2 + "db_name": "PostgreSQL",
3 + "query": "UPDATE pending_refunds SET escalated_at = NOW() WHERE id = $1",
4 + "describe": {
5 + "columns": [],
6 + "parameters": {
7 + "Left": [
8 + "Uuid"
9 + ]
10 + },
11 + "nullable": []
12 + },
13 + "hash": "442ca0ce542463afe513e17133bfd22c3a2d1e45ddb90064a64bf30b64925196"
14 + }
A server/.sqlx/query-4678a710329dbabd3196939141bb6ade132baa5485bad5159936fd8d4c77671f.json +16
@@ -0,0 +1,16 @@
1 + {
2 + "db_name": "PostgreSQL",
3 + "query": "INSERT INTO pending_s3_deletions (s3_key, bucket, source) SELECT * FROM unnest($1::text[], $2::text[], $3::text[]) ON CONFLICT DO NOTHING",
4 + "describe": {
5 + "columns": [],
6 + "parameters": {
7 + "Left": [
8 + "TextArray",
9 + "TextArray",
10 + "TextArray"
11 + ]
12 + },
13 + "nullable": []
14 + },
15 + "hash": "4678a710329dbabd3196939141bb6ade132baa5485bad5159936fd8d4c77671f"
16 + }
A server/.sqlx/query-6aadbc1bdb6d65f46a458c7fb8eeb05642ebcee9c03fe7635ade9406ae62b64f.json +14
@@ -0,0 +1,14 @@
1 + {
2 + "db_name": "PostgreSQL",
3 + "query": "DELETE FROM pending_s3_deletions WHERE id = ANY($1)",
4 + "describe": {
5 + "columns": [],
6 + "parameters": {
7 + "Left": [
8 + "UuidArray"
9 + ]
10 + },
11 + "nullable": []
12 + },
13 + "hash": "6aadbc1bdb6d65f46a458c7fb8eeb05642ebcee9c03fe7635ade9406ae62b64f"
14 + }
A server/.sqlx/query-7483a786cf09b76ae683083960962d59e2f27c7c498028276ab820dde5d14b7e.json +15
@@ -0,0 +1,15 @@
1 + {
2 + "db_name": "PostgreSQL",
3 + "query": "DELETE FROM pending_uploads pu\n USING UNNEST($1::text[], $2::text[]) AS t(s3_key, bucket)\n WHERE pu.s3_key = t.s3_key AND pu.bucket = t.bucket",
4 + "describe": {
5 + "columns": [],
6 + "parameters": {
7 + "Left": [
8 + "TextArray",
9 + "TextArray"
10 + ]
11 + },
12 + "nullable": []
13 + },
14 + "hash": "7483a786cf09b76ae683083960962d59e2f27c7c498028276ab820dde5d14b7e"
15 + }
A server/.sqlx/query-90738ee4514950905808dde549a3572b5bd5c3f508c3c3139f32b695aead0f1c.json +16
@@ -0,0 +1,16 @@
1 + {
2 + "db_name": "PostgreSQL",
3 + "query": "INSERT INTO pending_uploads (user_id, s3_key, bucket) VALUES ($1, $2, $3)\n ON CONFLICT (s3_key, bucket) DO UPDATE SET created_at = NOW()\n WHERE pending_uploads.user_id = EXCLUDED.user_id",
4 + "describe": {
5 + "columns": [],
6 + "parameters": {
7 + "Left": [
8 + "Uuid",
9 + "Text",
10 + "Text"
11 + ]
12 + },
13 + "nullable": []
14 + },
15 + "hash": "90738ee4514950905808dde549a3572b5bd5c3f508c3c3139f32b695aead0f1c"
16 + }
A server/.sqlx/query-94e5ea852c5bf35a798c65965823071169cbb0de4e3fa024bbf8604b0ba75063.json +14
@@ -0,0 +1,14 @@
1 + {
2 + "db_name": "PostgreSQL",
3 + "query": "\n WITH moved AS (\n DELETE FROM pending_s3_deletions\n WHERE id = ANY($1)\n RETURNING id, s3_key, bucket, source, created_at, attempts, last_attempted_at\n )\n INSERT INTO pending_s3_deletions_dead_letter\n (id, s3_key, bucket, source, created_at, attempts, last_attempted_at)\n SELECT id, s3_key, bucket, source, created_at, attempts, last_attempted_at\n FROM moved\n ON CONFLICT (id) DO NOTHING\n ",
4 + "describe": {
5 + "columns": [],
6 + "parameters": {
7 + "Left": [
8 + "UuidArray"
9 + ]
10 + },
11 + "nullable": []
12 + },
13 + "hash": "94e5ea852c5bf35a798c65965823071169cbb0de4e3fa024bbf8604b0ba75063"
14 + }
A server/.sqlx/query-9d70d52ffa4da48ff434a1ee4e50bfe0774734b73c79c9ed8a0b26727c5eddae.json +16
@@ -0,0 +1,16 @@
1 + {
2 + "db_name": "PostgreSQL",
3 + "query": "\n INSERT INTO pending_refunds (payment_intent_id, amount, amount_refunded)\n VALUES ($1, $2, $3)\n ON CONFLICT (payment_intent_id) WHERE matched_at IS NULL DO NOTHING\n ",
4 + "describe": {
5 + "columns": [],
6 + "parameters": {
7 + "Left": [
8 + "Text",
9 + "Int8",
10 + "Int8"
11 + ]
12 + },
13 + "nullable": []
14 + },
15 + "hash": "9d70d52ffa4da48ff434a1ee4e50bfe0774734b73c79c9ed8a0b26727c5eddae"
16 + }
A server/.sqlx/query-ae75b716dca52b00027933de0a2b912d9fd3331452b72ae34834734716dfa804.json +40
@@ -0,0 +1,40 @@
1 + {
2 + "db_name": "PostgreSQL",
3 + "query": "\n UPDATE pending_refunds\n SET matched_at = NOW()\n WHERE id = (\n SELECT id FROM pending_refunds\n WHERE payment_intent_id = $1 AND matched_at IS NULL\n LIMIT 1\n FOR UPDATE SKIP LOCKED\n )\n RETURNING id, payment_intent_id, amount AS \"amount: Cents\", amount_refunded AS \"amount_refunded: Cents\"\n ",
4 + "describe": {
5 + "columns": [
6 + {
7 + "ordinal": 0,
8 + "name": "id",
9 + "type_info": "Uuid"
10 + },
11 + {
12 + "ordinal": 1,
13 + "name": "payment_intent_id",
14 + "type_info": "Text"
15 + },
16 + {
17 + "ordinal": 2,
18 + "name": "amount: Cents",
19 + "type_info": "Int8"
20 + },
21 + {
22 + "ordinal": 3,
23 + "name": "amount_refunded: Cents",
24 + "type_info": "Int8"
25 + }
26 + ],
27 + "parameters": {
28 + "Left": [
29 + "Text"
30 + ]
31 + },
32 + "nullable": [
33 + false,
34 + false,
35 + false,
36 + false
37 + ]
38 + },
39 + "hash": "ae75b716dca52b00027933de0a2b912d9fd3331452b72ae34834734716dfa804"
40 + }
A server/.sqlx/query-f6e778c8973a2db6103d3f2f1c30f6446599a5d56b5f7897453738c404968b01.json +14
@@ -0,0 +1,14 @@
1 + {
2 + "db_name": "PostgreSQL",
3 + "query": "UPDATE pending_refunds SET matched_at = NULL WHERE id = $1",
4 + "describe": {
5 + "columns": [],
6 + "parameters": {
7 + "Left": [
8 + "Uuid"
9 + ]
10 + },
11 + "nullable": []
12 + },
13 + "hash": "f6e778c8973a2db6103d3f2f1c30f6446599a5d56b5f7897453738c404968b01"
14 + }
M server/src/db/pending_refunds.rs +13 -11
@@ -19,16 +19,16 @@ pub async fn insert_pending_refund(
19 19 amount: i64,
20 20 amount_refunded: i64,
21 21 ) -> Result<()> {
22 - sqlx::query(
22 + sqlx::query!(
23 23 r#"
24 24 INSERT INTO pending_refunds (payment_intent_id, amount, amount_refunded)
25 25 VALUES ($1, $2, $3)
26 26 ON CONFLICT (payment_intent_id) WHERE matched_at IS NULL DO NOTHING
27 27 "#,
28 + payment_intent_id,
29 + amount,
30 + amount_refunded,
28 31 )
29 - .bind(payment_intent_id)
30 - .bind(amount)
31 - .bind(amount_refunded)
32 32 .execute(pool)
33 33 .await?;
34 34
@@ -52,7 +52,8 @@ pub async fn claim_pending_refund(
52 52 pool: &PgPool,
53 53 payment_intent_id: &str,
54 54 ) -> Result<Option<PendingRefund>> {
55 - let row = sqlx::query_as::<_, PendingRefund>(
55 + let row = sqlx::query_as!(
56 + PendingRefund,
56 57 r#"
57 58 UPDATE pending_refunds
58 59 SET matched_at = NOW()
@@ -62,10 +63,10 @@ pub async fn claim_pending_refund(
62 63 LIMIT 1
63 64 FOR UPDATE SKIP LOCKED
64 65 )
65 - RETURNING id, payment_intent_id, amount, amount_refunded
66 + RETURNING id, payment_intent_id, amount AS "amount: Cents", amount_refunded AS "amount_refunded: Cents"
66 67 "#,
68 + payment_intent_id,
67 69 )
68 - .bind(payment_intent_id)
69 70 .fetch_optional(pool)
70 71 .await?;
71 72
@@ -84,8 +85,7 @@ pub async fn claim_pending_refund(
84 85 /// retry — safe because the refund handler is atomic, so a failed attempt
85 86 /// committed nothing. Idempotent.
86 87 pub async fn unclaim_pending_refund(pool: &PgPool, id: uuid::Uuid) -> Result<()> {
87 - sqlx::query("UPDATE pending_refunds SET matched_at = NULL WHERE id = $1")
88 - .bind(id)
88 + sqlx::query!("UPDATE pending_refunds SET matched_at = NULL WHERE id = $1", id)
89 89 .execute(pool)
90 90 .await?;
91 91 Ok(())
@@ -114,6 +114,9 @@ pub async fn get_stale_refunds(
114 114 age: chrono::Duration,
115 115 ) -> Result<Vec<StaleRefund>> {
116 116 let cutoff = chrono::Utc::now() - age;
117 + // runtime-checked: binds a chrono `DateTime<Utc>` cutoff (`$1`); a bind
118 + // parameter's type can't be overridden in the macro when sqlx's `time` and
119 + // `chrono` features are unified. See db::pending_uploads::get_stale_pending_uploads.
117 120 let rows = sqlx::query_as::<_, StaleRefund>(
118 121 r#"
119 122 SELECT id, payment_intent_id, amount, amount_refunded, created_at
@@ -135,8 +138,7 @@ pub async fn get_stale_refunds(
135 138
136 139 /// Mark a pending refund as escalated (alert sent, won't be re-alerted).
137 140 pub async fn mark_escalated(pool: &PgPool, id: uuid::Uuid) -> Result<()> {
138 - sqlx::query("UPDATE pending_refunds SET escalated_at = NOW() WHERE id = $1")
139 - .bind(id)
141 + sqlx::query!("UPDATE pending_refunds SET escalated_at = NOW() WHERE id = $1", id)
140 142 .execute(pool)
141 143 .await?;
142 144 Ok(())
M server/src/db/pending_s3_deletions.rs +7 -8
@@ -29,12 +29,12 @@ pub async fn enqueue_deletions<'e>(
29 29 // Takes any `PgExecutor` so callers can enqueue inside a transaction —
30 30 // letting a row delete, its storage refund, and the S3-delete enqueue commit
31 31 // atomically (see `versions::delete_version`).
32 - sqlx::query(
32 + sqlx::query!(
33 33 "INSERT INTO pending_s3_deletions (s3_key, bucket, source) SELECT * FROM unnest($1::text[], $2::text[], $3::text[]) ON CONFLICT DO NOTHING",
34 + &s3_keys as &[&str],
35 + &buckets as &[&str],
36 + &vec![source; keys.len()] as &[&str],
34 37 )
35 - .bind(&s3_keys)
36 - .bind(&buckets)
37 - .bind(vec![source; keys.len()])
38 38 .execute(executor)
39 39 .await?;
40 40 Ok(())
@@ -46,8 +46,7 @@ pub async fn remove_completed(pool: &PgPool, ids: &[Uuid]) -> Result<()> {
46 46 if ids.is_empty() {
47 47 return Ok(());
48 48 }
49 - sqlx::query("DELETE FROM pending_s3_deletions WHERE id = ANY($1)")
50 - .bind(ids)
49 + sqlx::query!("DELETE FROM pending_s3_deletions WHERE id = ANY($1)", ids)
51 50 .execute(pool)
52 51 .await?;
53 52 Ok(())
@@ -67,7 +66,7 @@ pub async fn move_to_dead_letter(pool: &PgPool, ids: &[Uuid]) -> Result<u64> {
67 66 if ids.is_empty() {
68 67 return Ok(0);
69 68 }
70 - let result = sqlx::query(
69 + let result = sqlx::query!(
71 70 r#"
72 71 WITH moved AS (
73 72 DELETE FROM pending_s3_deletions
@@ -80,8 +79,8 @@ pub async fn move_to_dead_letter(pool: &PgPool, ids: &[Uuid]) -> Result<u64> {
80 79 FROM moved
81 80 ON CONFLICT (id) DO NOTHING
82 81 "#,
82 + ids,
83 83 )
84 - .bind(ids)
85 84 .execute(pool)
86 85 .await?;
87 86 Ok(result.rows_affected())
M server/src/db/pending_uploads.rs +20 -13
@@ -20,14 +20,14 @@ pub async fn record_pending_upload(
20 20 // same key, do NOT refresh — let the reaper age the original row out on its
21 21 // own schedule. Otherwise a re-presign loop by another principal could keep
22 22 // an orphan object alive indefinitely.
23 - sqlx::query(
23 + sqlx::query!(
24 24 "INSERT INTO pending_uploads (user_id, s3_key, bucket) VALUES ($1, $2, $3)
25 25 ON CONFLICT (s3_key, bucket) DO UPDATE SET created_at = NOW()
26 26 WHERE pending_uploads.user_id = EXCLUDED.user_id",
27 + user_id as UserId,
28 + s3_key,
29 + bucket,
27 30 )
28 - .bind(user_id)
29 - .bind(s3_key)
30 - .bind(bucket)
31 31 .execute(pool)
32 32 .await?;
33 33 Ok(())
@@ -49,12 +49,14 @@ pub async fn remove_pending_upload<'e>(
49 49 s3_key: &str,
50 50 bucket: &str,
51 51 ) -> Result<()> {
52 - sqlx::query("DELETE FROM pending_uploads WHERE s3_key = $1 AND user_id = $2 AND bucket = $3")
53 - .bind(s3_key)
54 - .bind(user_id)
55 - .bind(bucket)
56 - .execute(executor)
57 - .await?;
52 + sqlx::query!(
53 + "DELETE FROM pending_uploads WHERE s3_key = $1 AND user_id = $2 AND bucket = $3",
54 + s3_key,
55 + user_id as UserId,
56 + bucket,
57 + )
58 + .execute(executor)
59 + .await?;
58 60 Ok(())
59 61 }
60 62
@@ -71,6 +73,11 @@ pub async fn get_stale_pending_uploads(
71 73 max_age: chrono::Duration,
72 74 ) -> Result<Vec<(String, String)>> {
73 75 let cutoff = chrono::Utc::now() - max_age;
76 + // runtime-checked: binds a chrono `DateTime<Utc>` (`$1`). With sqlx's `time`
77 + // and `chrono` features unified (the session store pulls `time`), the macro
78 + // infers the TIMESTAMPTZ bind as `time::OffsetDateTime`, which a chrono value
79 + // won't satisfy, and a bind parameter's type can't be overridden in the macro
80 + // (only output columns can). Mirrors mt-db's one chrono-binding write path.
74 81 let rows: Vec<(String, String)> = sqlx::query_as(
75 82 "SELECT s3_key, bucket FROM pending_uploads WHERE created_at < $1 \
76 83 ORDER BY created_at LIMIT $2",
@@ -94,13 +101,13 @@ pub async fn delete_pending_uploads(pool: &PgPool, keys: &[(String, String)]) ->
94 101 let buckets: Vec<&str> = keys.iter().map(|(_, b)| b.as_str()).collect();
95 102 // Match each (key, bucket) pair positionally via UNNEST so a key in two
96 103 // buckets only clears the rows whose bucket was actually reaped.
97 - sqlx::query(
104 + sqlx::query!(
98 105 "DELETE FROM pending_uploads pu
99 106 USING UNNEST($1::text[], $2::text[]) AS t(s3_key, bucket)
100 107 WHERE pu.s3_key = t.s3_key AND pu.bucket = t.bucket",
108 + &s3_keys as &[&str],
109 + &buckets as &[&str],
101 110 )
102 - .bind(&s3_keys)
103 - .bind(&buckets)
104 111 .execute(pool)
105 112 .await?;
106 113 Ok(())