Skip to main content

max / makenotwork

server: convert remaining money-path DB modules to compile-checked SQL (ultra-fuzz Run #1 --deep Phase 4) Finish D4: convert media_files, versions, sessions, auth, license_keys, promo_codes, subscriptions, transactions from runtime sqlx::query strings to compile-checked query!/query_as! macros (sqlx-offline), completing the money-path migration. Queries that bind a chrono DateTime<Utc> or build SQL dynamically stay runtime with a documented reason. Fixes surfaced and resolved by the offline gate: - ORDER BY on a macro-aliased column failed (sqlx sends the "col!: Type" suffix as the literal column name); reorder on the underlying expression instead. - get_user_purchases.license_key_code: a column selected from a derived subquery is inferred NOT NULL by sqlx, so a bare "col: Type" override panicked with UnexpectedNull at runtime when the LEFT JOIN produced no key. Forced nullable with "?: KeyCode".
Author: Max Johnson <me@maxj.phd> · 2026-06-23 00:01 UTC
Signed with PGP, not checked
Commit: 52a315ed7feade137c150f9259b129d26bed6bd6
Parent: b7eadd1
145 files changed, +7784 insertions, -493 deletions
@@ -4,7 +4,7 @@
4 4 use sqlx::PgPool;
5 5
6 6 use super::models::*;
7 - use super::UserId;
7 + use super::{LoginTokenId, UserId};
8 8 use crate::error::Result;
9 9
10 10 /// Result of an atomic failed-login increment.
@@ -37,7 +37,7 @@
37 37 // window (counter already >= threshold) — so the lockout notification was
38 38 // silently skipped on every re-lock. Now `just_locked` is exactly "the lock
39 39 // was (re)set on this call".
40 - let row: (i32, bool) = sqlx::query_as(
40 + let row = sqlx::query!(
41 41 r#"
42 42 WITH prev AS (
43 43 SELECT failed_login_attempts AS old_attempts, locked_until AS old_lock
@@ -64,28 +64,28 @@
64 64 RETURNING
65 65 u.failed_login_attempts,
66 66 (prev.old_attempts + 1 >= $2
67 - AND (prev.old_lock IS NULL OR prev.old_lock <= NOW())) AS just_locked
67 + AND (prev.old_lock IS NULL OR prev.old_lock <= NOW())) AS "just_locked!"
68 68 "#,
69 + user_id as UserId,
70 + max_attempts,
71 + lockout_minutes.to_string(),
69 72 )
70 - .bind(user_id)
71 - .bind(max_attempts)
72 - .bind(lockout_minutes.to_string())
73 73 .fetch_one(pool)
74 74 .await?;
75 75
76 76 Ok(FailedLoginResult {
77 - attempts: row.0,
78 - just_locked: row.1,
77 + attempts: row.failed_login_attempts,
78 + just_locked: row.just_locked,
79 79 })
80 80 }
81 81
82 82 /// Reset failed login attempts (on successful login)
83 83 #[tracing::instrument(skip_all)]
84 84 pub async fn reset_failed_login(pool: &PgPool, user_id: UserId) -> Result<()> {
85 - sqlx::query(
85 + sqlx::query!(
86 86 "UPDATE users SET failed_login_attempts = 0, locked_until = NULL WHERE id = $1",
87 + user_id as UserId,
87 88 )
88 - .bind(user_id)
89 89 .execute(pool)
90 90 .await?;
91 91
@@ -100,6 +100,7 @@
100 100 token_hash: &str,
101 101 expires_at: DateTime<Utc>,
102 102 ) -> Result<DbLoginToken> {
103 + // runtime-checked: binds a chrono DateTime<Utc> param; a bind param's type can't be overridden in the macro when sqlx time+chrono features are unified.
103 104 let token = sqlx::query_as::<_, DbLoginToken>(
104 105 r#"
105 106 INSERT INTO login_tokens (user_id, token_hash, expires_at)
@@ -127,17 +128,24 @@
127 128 pool: &PgPool,
128 129 token_hash: &str,
129 130 ) -> Result<Option<DbLoginToken>> {
130 - let token = sqlx::query_as::<_, DbLoginToken>(
131 + let token = sqlx::query_as!(
132 + DbLoginToken,
131 133 r#"
132 134 UPDATE login_tokens
133 135 SET used_at = NOW()
134 136 WHERE token_hash = $1
135 137 AND used_at IS NULL
136 138 AND expires_at > NOW()
137 - RETURNING *
139 + RETURNING
140 + id AS "id: LoginTokenId",
141 + user_id AS "user_id: UserId",
142 + token_hash,
143 + expires_at AS "expires_at: chrono::DateTime<chrono::Utc>",
144 + used_at AS "used_at: chrono::DateTime<chrono::Utc>",
145 + created_at AS "created_at: chrono::DateTime<chrono::Utc>"
138 146 "#,
147 + token_hash,
139 148 )
140 - .bind(token_hash)
141 149 .fetch_optional(pool)
142 150 .await?;
143 151
@@ -152,6 +160,7 @@
152 160 token_hash: &str,
153 161 expires_at: DateTime<Utc>,
154 162 ) -> Result<()> {
163 + // runtime-checked: binds a chrono DateTime<Utc> param; a bind param's type can't be overridden in the macro when sqlx time+chrono features are unified.
155 164 sqlx::query(
156 165 r#"
157 166 INSERT INTO password_reset_tokens (user_id, token_hash, expires_at)
@@ -176,15 +185,15 @@
176 185 pool: &PgPool,
177 186 token_hash: &str,
178 187 ) -> Result<Option<UserId>> {
179 - let user_id = sqlx::query_scalar::<_, UserId>(
188 + let user_id = sqlx::query_scalar!(
180 189 r#"
181 - SELECT user_id FROM password_reset_tokens
190 + SELECT user_id AS "user_id: UserId" FROM password_reset_tokens
182 191 WHERE token_hash = $1
183 192 AND used_at IS NULL
184 193 AND expires_at > NOW()
185 194 "#,
195 + token_hash,
186 196 )
187 - .bind(token_hash)
188 197 .fetch_optional(pool)
189 198 .await?;
190 199
@@ -201,17 +210,17 @@
201 210 pool: &PgPool,
202 211 token_hash: &str,
203 212 ) -> Result<Option<UserId>> {
204 - let user_id = sqlx::query_scalar::<_, UserId>(
213 + let user_id = sqlx::query_scalar!(
205 214 r#"
206 215 UPDATE password_reset_tokens
207 216 SET used_at = NOW()
208 217 WHERE token_hash = $1
209 218 AND used_at IS NULL
210 219 AND expires_at > NOW()
211 - RETURNING user_id
220 + RETURNING user_id AS "user_id: UserId"
212 221 "#,
222 + token_hash,
213 223 )
214 - .bind(token_hash)
215 224 .fetch_optional(pool)
216 225 .await?;
217 226
@@ -222,7 +231,7 @@
222 231 /// doesn't accumulate dead rows). Keeps recently-used rows briefly for audit.
223 232 #[tracing::instrument(skip_all)]
224 233 pub async fn prune_password_reset_tokens(pool: &PgPool) -> Result<u64> {
225 - let result = sqlx::query(
234 + let result = sqlx::query!(
226 235 r#"
227 236 DELETE FROM password_reset_tokens
228 237 WHERE expires_at < NOW() - interval '7 days'
@@ -241,14 +250,14 @@
241 250 /// reset kills all outstanding links" behavior.
242 251 #[tracing::instrument(skip_all)]
243 252 pub async fn invalidate_password_reset_tokens(pool: &PgPool, user_id: UserId) -> Result<()> {
244 - sqlx::query(
253 + sqlx::query!(
245 254 r#"
246 255 UPDATE password_reset_tokens
247 256 SET used_at = NOW()
248 257 WHERE user_id = $1 AND used_at IS NULL
249 258 "#,
259 + user_id as UserId,
250 260 )
251 - .bind(user_id)
252 261 .execute(pool)
253 262 .await?;
254 263
@@ -22,34 +22,50 @@
22 22 key_code: &KeyCode,
23 23 max_activations: Option<i32>,
24 24 ) -> Result<DbLicenseKey> {
25 - const SQL: &str = r#"
25 + let first = sqlx::query_as!(
26 + DbLicenseKey,
27 + r#"
26 28 INSERT INTO license_keys (item_id, owner_id, transaction_id, key_code, max_activations)
27 29 VALUES ($1, $2, $3, $4, $5)
28 - RETURNING *
29 - "#;
30 -
31 - let first = sqlx::query_as::<_, DbLicenseKey>(SQL)
32 - .bind(item_id)
33 - .bind(owner_id)
34 - .bind(transaction_id)
35 - .bind(key_code)
36 - .bind(max_activations)
37 - .fetch_one(pool)
38 - .await;
30 + RETURNING id AS "id: LicenseKeyId", item_id AS "item_id: ItemId",
31 + owner_id AS "owner_id: UserId", transaction_id AS "transaction_id: TransactionId",
32 + key_code AS "key_code: KeyCode", max_activations, activation_count,
33 + revoked_at AS "revoked_at: chrono::DateTime<chrono::Utc>",
34 + created_at AS "created_at: chrono::DateTime<chrono::Utc>"
35 + "#,
36 + item_id as ItemId,
37 + owner_id as UserId,
38 + transaction_id as Option<TransactionId>,
39 + key_code as &KeyCode,
40 + max_activations,
41 + )
42 + .fetch_one(pool)
43 + .await;
39 44
40 45 match first {
41 46 Ok(key) => Ok(key),
42 47 Err(sqlx::Error::Database(e)) if e.code().as_deref() == Some("23505") => {
43 48 let retry_code = crate::helpers::generate_key_code();
44 49 tracing::warn!(item_id = %item_id, "license key 23505 collision; retrying once");
45 - let key = sqlx::query_as::<_, DbLicenseKey>(SQL)
46 - .bind(item_id)
47 - .bind(owner_id)
48 - .bind(transaction_id)
49 - .bind(&retry_code)
50 - .bind(max_activations)
51 - .fetch_one(pool)
52 - .await?;
50 + let key = sqlx::query_as!(
51 + DbLicenseKey,
52 + r#"
53 + INSERT INTO license_keys (item_id, owner_id, transaction_id, key_code, max_activations)
54 + VALUES ($1, $2, $3, $4, $5)
55 + RETURNING id AS "id: LicenseKeyId", item_id AS "item_id: ItemId",
56 + owner_id AS "owner_id: UserId", transaction_id AS "transaction_id: TransactionId",
57 + key_code AS "key_code: KeyCode", max_activations, activation_count,
58 + revoked_at AS "revoked_at: chrono::DateTime<chrono::Utc>",
59 + created_at AS "created_at: chrono::DateTime<chrono::Utc>"
60 + "#,
61 + item_id as ItemId,
62 + owner_id as UserId,
63 + transaction_id as Option<TransactionId>,
64 + &retry_code as &KeyCode,
65 + max_activations,
66 + )
67 + .fetch_one(pool)
68 + .await?;
53 69 Ok(key)
54 70 }
55 71 Err(e) => Err(e.into()),
@@ -59,10 +75,18 @@
59 75 /// Look up a license key by its code.
60 76 #[tracing::instrument(skip_all)]
61 77 pub async fn get_license_key_by_code(pool: &PgPool, key_code: &KeyCode) -> Result<Option<DbLicenseKey>> {
62 - let key = sqlx::query_as::<_, DbLicenseKey>(
63 - "SELECT * FROM license_keys WHERE key_code = $1",
78 + let key = sqlx::query_as!(
79 + DbLicenseKey,
80 + r#"
81 + SELECT id AS "id: LicenseKeyId", item_id AS "item_id: ItemId",
82 + owner_id AS "owner_id: UserId", transaction_id AS "transaction_id: TransactionId",
83 + key_code AS "key_code: KeyCode", max_activations, activation_count,
84 + revoked_at AS "revoked_at: chrono::DateTime<chrono::Utc>",
85 + created_at AS "created_at: chrono::DateTime<chrono::Utc>"
86 + FROM license_keys WHERE key_code = $1
87 + "#,
88 + key_code as &KeyCode,
64 89 )
65 - .bind(key_code)
66 90 .fetch_optional(pool)
67 91 .await?;
68 92
@@ -72,10 +96,18 @@
72 96 /// Get a license key by ID.
73 97 #[tracing::instrument(skip_all)]
74 98 pub async fn get_license_key_by_id(pool: &PgPool, id: LicenseKeyId) -> Result<Option<DbLicenseKey>> {
75 - let key = sqlx::query_as::<_, DbLicenseKey>(
76 - "SELECT * FROM license_keys WHERE id = $1",
99 + let key = sqlx::query_as!(
100 + DbLicenseKey,
101 + r#"
102 + SELECT id AS "id: LicenseKeyId", item_id AS "item_id: ItemId",
103 + owner_id AS "owner_id: UserId", transaction_id AS "transaction_id: TransactionId",
104 + key_code AS "key_code: KeyCode", max_activations, activation_count,
105 + revoked_at AS "revoked_at: chrono::DateTime<chrono::Utc>",
106 + created_at AS "created_at: chrono::DateTime<chrono::Utc>"
107 + FROM license_keys WHERE id = $1
108 + "#,
109 + id as LicenseKeyId,
77 110 )
78 - .bind(id)
79 111 .fetch_optional(pool)
80 112 .await?;
81 113
@@ -85,10 +117,10 @@
85 117 /// Count license keys for an item.
86 118 #[tracing::instrument(skip_all)]
87 119 pub async fn count_keys_by_item(pool: &PgPool, item_id: ItemId) -> Result<i64> {
88 - let count: i64 = sqlx::query_scalar(
89 - "SELECT COUNT(*) FROM license_keys WHERE item_id = $1",
120 + let count = sqlx::query_scalar!(
121 + r#"SELECT COUNT(*) AS "count!" FROM license_keys WHERE item_id = $1"#,
122 + item_id as ItemId,
90 123 )
91 - .bind(item_id)
92 124 .fetch_one(pool)
93 125 .await?;
94 126
@@ -102,10 +134,18 @@
102 134 /// future work could add cursor-based pagination if needed.
103 135 #[tracing::instrument(skip_all)]
104 136 pub async fn get_license_keys_by_item(pool: &PgPool, item_id: ItemId) -> Result<Vec<DbLicenseKey>> {
105 - let keys = sqlx::query_as::<_, DbLicenseKey>(
106 - "SELECT * FROM license_keys WHERE item_id = $1 ORDER BY created_at DESC LIMIT 500",
137 + let keys = sqlx::query_as!(
138 + DbLicenseKey,
139 + r#"
140 + SELECT id AS "id: LicenseKeyId", item_id AS "item_id: ItemId",
141 + owner_id AS "owner_id: UserId", transaction_id AS "transaction_id: TransactionId",
142 + key_code AS "key_code: KeyCode", max_activations, activation_count,
143 + revoked_at AS "revoked_at: chrono::DateTime<chrono::Utc>",
144 + created_at AS "created_at: chrono::DateTime<chrono::Utc>"
145 + FROM license_keys WHERE item_id = $1 ORDER BY created_at DESC LIMIT 500
146 + "#,
147 + item_id as ItemId,
107 148 )
108 - .bind(item_id)
109 149 .fetch_all(pool)
110 150 .await?;
111 151
@@ -118,10 +158,18 @@
118 158 pool: &PgPool,
119 159 item_ids: &[ItemId],
120 160 ) -> Result<std::collections::HashMap<ItemId, Vec<DbLicenseKey>>> {
121 - let keys = sqlx::query_as::<_, DbLicenseKey>(
122 - "SELECT * FROM license_keys WHERE item_id = ANY($1) ORDER BY item_id, created_at DESC",
161 + let keys = sqlx::query_as!(
162 + DbLicenseKey,
163 + r#"
164 + SELECT id AS "id: LicenseKeyId", item_id AS "item_id: ItemId",
165 + owner_id AS "owner_id: UserId", transaction_id AS "transaction_id: TransactionId",
166 + key_code AS "key_code: KeyCode", max_activations, activation_count,
167 + revoked_at AS "revoked_at: chrono::DateTime<chrono::Utc>",
168 + created_at AS "created_at: chrono::DateTime<chrono::Utc>"
169 + FROM license_keys WHERE item_id = ANY($1) ORDER BY item_id, created_at DESC
170 + "#,
171 + item_ids as &[ItemId],
123 172 )
124 - .bind(item_ids)
125 173 .fetch_all(pool)
126 174 .await?;
127 175
@@ -139,11 +187,18 @@
139 187 license_key_id: LicenseKeyId,
140 188 machine_id: &str,
141 189 ) -> Result<Option<DbLicenseActivation>> {
142 - let activation = sqlx::query_as::<_, DbLicenseActivation>(
143 - "SELECT * FROM license_activations WHERE license_key_id = $1 AND machine_id = $2",
190 + let activation = sqlx::query_as!(
191 + DbLicenseActivation,
192 + r#"
193 + SELECT id AS "id: LicenseActivationId", license_key_id AS "license_key_id: LicenseKeyId",
194 + machine_id, label,
195 + activated_at AS "activated_at: chrono::DateTime<chrono::Utc>",
196 + last_validated_at AS "last_validated_at: chrono::DateTime<chrono::Utc>", is_active
197 + FROM license_activations WHERE license_key_id = $1 AND machine_id = $2
198 + "#,
199 + license_key_id as LicenseKeyId,
200 + machine_id,
144 201 )
145 - .bind(license_key_id)
146 - .bind(machine_id)
147 202 .fetch_optional(pool)
148 203 .await?;
149 204
@@ -153,10 +208,10 @@
153 208 /// Update the last_validated_at timestamp for an existing activation.
154 209 #[tracing::instrument(skip_all)]
155 210 pub async fn touch_activation(pool: &PgPool, activation_id: LicenseActivationId) -> Result<()> {
156 - sqlx::query(
211 + sqlx::query!(
157 212 "UPDATE license_activations SET last_validated_at = NOW() WHERE id = $1",
213 + activation_id as LicenseActivationId,
158 214 )
159 - .bind(activation_id)
160 215 .execute(pool)
161 216 .await?;
162 217
@@ -171,10 +226,10 @@
171 226 /// under concurrent activations (ultra-fuzz Run #1 Payments MINOR).
172 227 #[tracing::instrument(skip_all)]
173 228 pub async fn get_activation_count(pool: &PgPool, license_key_id: LicenseKeyId) -> Result<i32> {
174 - let count: i32 = sqlx::query_scalar(
229 + let count = sqlx::query_scalar!(
175 230 "SELECT activation_count FROM license_keys WHERE id = $1",
231 + license_key_id as LicenseKeyId,
176 232 )
177 - .bind(license_key_id)
178 233 .fetch_one(pool)
179 234 .await?;
180 235 Ok(count)
@@ -206,21 +261,30 @@
206 261 // revocation INSIDE the lock (MINOR TOCTOU, Run #23): a charge.refunded that
207 262 // revokes the key between the caller's pre-check and here must block the
208 263 // activation. No eligible (non-revoked) row => no activation.
209 - let locked = sqlx::query("SELECT 1 FROM license_keys WHERE id = $1 AND revoked_at IS NULL FOR UPDATE")
210 - .bind(license_key_id)
211 - .fetch_optional(&mut *tx)
212 - .await?;
264 + let locked = sqlx::query_scalar!(
265 + r#"SELECT 1 AS "one!" FROM license_keys WHERE id = $1 AND revoked_at IS NULL FOR UPDATE"#,
266 + license_key_id as LicenseKeyId,
267 + )
268 + .fetch_optional(&mut *tx)
269 + .await?;
213 270 if locked.is_none() {
214 271 tx.rollback().await?;
215 272 return Ok(None);
216 273 }
217 274
218 275 // Check if this machine already has an activation (re-activation is always OK)
219 - let existing: Option<DbLicenseActivation> = sqlx::query_as(
220 - "SELECT * FROM license_activations WHERE license_key_id = $1 AND machine_id = $2",
276 + let existing: Option<DbLicenseActivation> = sqlx::query_as!(
277 + DbLicenseActivation,
278 + r#"
279 + SELECT id AS "id: LicenseActivationId", license_key_id AS "license_key_id: LicenseKeyId",
280 + machine_id, label,
281 + activated_at AS "activated_at: chrono::DateTime<chrono::Utc>",
282 + last_validated_at AS "last_validated_at: chrono::DateTime<chrono::Utc>", is_active
283 + FROM license_activations WHERE license_key_id = $1 AND machine_id = $2
284 + "#,
285 + license_key_id as LicenseKeyId,
286 + machine_id,
221 287 )
222 - .bind(license_key_id)
223 - .bind(machine_id)
224 288 .fetch_optional(&mut *tx)
225 289 .await?;
226 290
@@ -228,10 +292,10 @@
228 292 if existing.is_none()
229 293 && let Some(max) = max_activations
230 294 {
231 - let count: i64 = sqlx::query_scalar(
232 - "SELECT COUNT(*) FROM license_activations WHERE license_key_id = $1 AND is_active = true",
295 + let count = sqlx::query_scalar!(
296 + r#"SELECT COUNT(*) AS "count!" FROM license_activations WHERE license_key_id = $1 AND is_active = true"#,
297 + license_key_id as LicenseKeyId,
233 298 )
234 - .bind(license_key_id)
235 299 .fetch_one(&mut *tx)
236 300 .await?;
237 301
@@ -242,24 +306,28 @@
242 306 }
243 307
244 308 // Upsert: if same machine_id re-activates, reactivate it
245 - let activation = sqlx::query_as::<_, DbLicenseActivation>(
309 + let activation = sqlx::query_as!(
310 + DbLicenseActivation,
246 311 r#"
247 312 INSERT INTO license_activations (license_key_id, machine_id, label)
248 313 VALUES ($1, $2, $3)
249 314 ON CONFLICT (license_key_id, machine_id)
250 315 DO UPDATE SET is_active = true, last_validated_at = NOW(),
251 316 label = COALESCE(EXCLUDED.label, license_activations.label)
252 - RETURNING *
317 + RETURNING id AS "id: LicenseActivationId", license_key_id AS "license_key_id: LicenseKeyId",
318 + machine_id, label,
319 + activated_at AS "activated_at: chrono::DateTime<chrono::Utc>",
320 + last_validated_at AS "last_validated_at: chrono::DateTime<chrono::Utc>", is_active
253 321 "#,
322 + license_key_id as LicenseKeyId,
323 + machine_id,
324 + label,
254 325 )
255 - .bind(license_key_id)
256 - .bind(machine_id)
257 - .bind(label)
258 326 .fetch_one(&mut *tx)
259 327 .await?;
260 328
261 329 // Recount active activations to keep denormalized count accurate
262 - sqlx::query(
330 + sqlx::query!(
263 331 r#"
264 332 UPDATE license_keys
265 333 SET activation_count = (
@@ -268,8 +336,8 @@
268 336 )
269 337 WHERE id = $1
270 338 "#,
339 + license_key_id as LicenseKeyId,
271 340 )
272 - .bind(license_key_id)
273 341 .execute(&mut *tx)
274 342 .await?;
275 343
@@ -290,21 +358,21 @@
290 358 ) -> Result<bool> {
291 359 let mut tx = pool.begin().await?;
292 360
293 - let result = sqlx::query(
361 + let result = sqlx::query!(
294 362 r#"
295 363 UPDATE license_activations
296 364 SET is_active = false
297 365 WHERE license_key_id = $1 AND machine_id = $2 AND is_active = true
298 366 "#,
367 + license_key_id as LicenseKeyId,
368 + machine_id,
299 369 )
300 - .bind(license_key_id)
301 - .bind(machine_id)
302 370 .execute(&mut *tx)
303 371 .await?;
304 372
305 373 if result.rows_affected() > 0 {
306 374 // Recount active activations
307 - sqlx::query(
375 + sqlx::query!(
308 376 r#"
309 377 UPDATE license_keys
310 378 SET activation_count = (
@@ -313,8 +381,8 @@
313 381 )
314 382 WHERE id = $1
315 383 "#,
384 + license_key_id as LicenseKeyId,
316 385 )
317 - .bind(license_key_id)
318 386 .execute(&mut *tx)
319 387 .await?;
320 388
@@ -335,21 +403,21 @@
335 403 pub async fn revoke_license_key(pool: &PgPool, key_id: LicenseKeyId) -> Result<()> {
336 404 let mut tx = pool.begin().await?;
337 405
338 - sqlx::query(
406 + sqlx::query!(
339 407 r#"
340 408 UPDATE license_keys
341 409 SET revoked_at = NOW()
342 410 WHERE id = $1
343 411 "#,
412 + key_id as LicenseKeyId,
344 413 )
345 - .bind(key_id)
346 414 .execute(&mut *tx)
347 415 .await?;
348 416
349 - sqlx::query(
417 + sqlx::query!(
350 418 "UPDATE license_activations SET is_active = false WHERE license_key_id = $1",
419 + key_id as LicenseKeyId,
351 420 )
352 - .bind(key_id)
353 421 .execute(&mut *tx)
354 422 .await?;
355 423
@@ -369,10 +437,10 @@
369 437 transaction_id: TransactionId,
370 438 ) -> Result<u64> {
371 439 // Get all key IDs for this transaction
372 - let key_ids: Vec<LicenseKeyId> = sqlx::query_scalar(
373 - "SELECT id FROM license_keys WHERE transaction_id = $1 AND revoked_at IS NULL",
440 + let key_ids: Vec<LicenseKeyId> = sqlx::query_scalar!(
441 + r#"SELECT id AS "id: LicenseKeyId" FROM license_keys WHERE transaction_id = $1 AND revoked_at IS NULL"#,
442 + transaction_id as TransactionId,
374 443 )
375 - .bind(transaction_id)
376 444 .fetch_all(&mut *conn)
377 445 .await?;
378 446
@@ -381,23 +449,23 @@
381 449 }
382 450
383 451 // Revoke the keys
384 - let result = sqlx::query(
452 + let result = sqlx::query!(
385 453 r#"
386 454 UPDATE license_keys
387 455 SET revoked_at = NOW()
388 456 WHERE transaction_id = $1 AND revoked_at IS NULL
389 457 "#,
458 + transaction_id as TransactionId,
390 459 )
391 - .bind(transaction_id)
392 460 .execute(&mut *conn)
393 461 .await?;
394 462
395 463 // Deactivate all activations for those keys in a single query
396 464 if !key_ids.is_empty() {
397 - sqlx::query(
465 + sqlx::query!(
398 466 "UPDATE license_activations SET is_active = false WHERE license_key_id = ANY($1)",
467 + &key_ids as &[LicenseKeyId],
399 468 )
400 - .bind(&key_ids)
401 469 .execute(&mut *conn)
402 470 .await?;
403 471 }
@@ -20,21 +20,24 @@
20 20 media_type: &str,
21 21 scan_status: &str,
22 22 ) -> Result<DbMediaFile> {
23 - let row = sqlx::query_as::<_, DbMediaFile>(
23 + let row = sqlx::query_as!(
24 + DbMediaFile,
24 25 r#"
25 26 INSERT INTO media_files (user_id, folder, filename, s3_key, content_type, file_size_bytes, media_type, scan_status)
26 27 VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
27 - RETURNING *
28 + RETURNING id AS "id: MediaFileId", user_id AS "user_id: UserId", folder, filename,
29 + s3_key, content_type, file_size_bytes, media_type, scan_status,
30 + created_at AS "created_at: chrono::DateTime<chrono::Utc>"
28 31 "#,
32 + user_id as UserId,
33 + folder,
34 + filename,
35 + s3_key,
36 + content_type,
37 + file_size_bytes,
38 + media_type,
39 + scan_status,
29 40 )
30 - .bind(user_id)
31 - .bind(folder)
32 - .bind(filename)
33 - .bind(s3_key)
34 - .bind(content_type)
35 - .bind(file_size_bytes)
36 - .bind(media_type)
37 - .bind(scan_status)
38 41 .fetch_one(executor)
39 42 .await?;
40 43
@@ -54,20 +57,32 @@
54 57 folder: Option<&str>,
55 58 ) -> Result<Vec<DbMediaFile>> {
56 59 let rows = if let Some(f) = folder {
57 - sqlx::query_as::<_, DbMediaFile>(
58 - "SELECT * FROM media_files WHERE user_id = $1 AND folder = $2 AND scan_status = 'clean' ORDER BY created_at DESC LIMIT $3",
60 + sqlx::query_as!(
61 + DbMediaFile,
62 + r#"SELECT id AS "id: MediaFileId", user_id AS "user_id: UserId", folder, filename,
63 + s3_key, content_type, file_size_bytes, media_type, scan_status,
64 + created_at AS "created_at: chrono::DateTime<chrono::Utc>"
65 + FROM media_files
66 + WHERE user_id = $1 AND folder = $2 AND scan_status = 'clean'
67 + ORDER BY created_at DESC LIMIT $3"#,
68 + user_id as UserId,
69 + f,
70 + MEDIA_LIST_HARD_CAP,
59 71 )
60 - .bind(user_id)
61 - .bind(f)
62 - .bind(MEDIA_LIST_HARD_CAP)
63 72 .fetch_all(pool)
64 73 .await?
65 74 } else {
66 - sqlx::query_as::<_, DbMediaFile>(
67 - "SELECT * FROM media_files WHERE user_id = $1 AND scan_status = 'clean' ORDER BY created_at DESC LIMIT $2",
75 + sqlx::query_as!(
76 + DbMediaFile,
77 + r#"SELECT id AS "id: MediaFileId", user_id AS "user_id: UserId", folder, filename,
78 + s3_key, content_type, file_size_bytes, media_type, scan_status,
79 + created_at AS "created_at: chrono::DateTime<chrono::Utc>"
80 + FROM media_files
81 + WHERE user_id = $1 AND scan_status = 'clean'
82 + ORDER BY created_at DESC LIMIT $2"#,
83 + user_id as UserId,
84 + MEDIA_LIST_HARD_CAP,
68 85 )
69 - .bind(user_id)
70 - .bind(MEDIA_LIST_HARD_CAP)
71 86 .fetch_all(pool)
72 87 .await?
73 88 };
@@ -85,10 +100,10 @@
85 100 /// List distinct folder names for a user.
86 101 #[tracing::instrument(skip_all)]
87 102 pub async fn list_folders(pool: &PgPool, user_id: UserId) -> Result<Vec<String>> {
88 - let folders: Vec<String> = sqlx::query_scalar(
103 + let folders: Vec<String> = sqlx::query_scalar!(
89 104 "SELECT DISTINCT folder FROM media_files WHERE user_id = $1 ORDER BY folder",
105 + user_id as UserId,
90 106 )
91 - .bind(user_id)
92 107 .fetch_all(pool)
93 108 .await?;
94 109
@@ -98,10 +113,14 @@
98 113 /// Get a single media file by ID.
99 114 #[tracing::instrument(skip_all)]
100 115 pub async fn get_by_id(pool: &PgPool, id: MediaFileId) -> Result<Option<DbMediaFile>> {
101 - let row = sqlx::query_as::<_, DbMediaFile>(
102 - "SELECT * FROM media_files WHERE id = $1",
116 + let row = sqlx::query_as!(
117 + DbMediaFile,
118 + r#"SELECT id AS "id: MediaFileId", user_id AS "user_id: UserId", folder, filename,
119 + s3_key, content_type, file_size_bytes, media_type, scan_status,
120 + created_at AS "created_at: chrono::DateTime<chrono::Utc>"
121 + FROM media_files WHERE id = $1"#,
122 + id as MediaFileId,
103 123 )
104 - .bind(id)
105 124 .fetch_optional(pool)
106 125 .await?;
107 126
@@ -114,10 +133,14 @@
114 133 executor: impl sqlx::PgExecutor<'e>,
115 134 id: MediaFileId,
116 135 ) -> Result<Option<DbMediaFile>> {
117 - let row = sqlx::query_as::<_, DbMediaFile>(
118 - "DELETE FROM media_files WHERE id = $1 RETURNING *",
136 + let row = sqlx::query_as!(
137 + DbMediaFile,
138 + r#"DELETE FROM media_files WHERE id = $1
139 + RETURNING id AS "id: MediaFileId", user_id AS "user_id: UserId", folder, filename,
140 + s3_key, content_type, file_size_bytes, media_type, scan_status,
141 + created_at AS "created_at: chrono::DateTime<chrono::Utc>""#,
142 + id as MediaFileId,
119 143 )
120 - .bind(id)
121 144 .fetch_optional(executor)
122 145 .await?;
123 146
@@ -128,10 +151,10 @@
128 151 #[allow(dead_code)]
129 152 #[tracing::instrument(skip_all)]
130 153 pub async fn count_by_user(pool: &PgPool, user_id: UserId) -> Result<i64> {
131 - let count: i64 = sqlx::query_scalar(
132 - "SELECT COUNT(*) FROM media_files WHERE user_id = $1",
154 + let count: i64 = sqlx::query_scalar!(
155 + r#"SELECT COUNT(*) AS "count!" FROM media_files WHERE user_id = $1"#,
156 + user_id as UserId,
133 157 )
134 - .bind(user_id)
135 158 .fetch_one(pool)
136 159 .await?;
137 160
@@ -29,6 +29,7 @@
29 29 project_id: Option<ProjectId>,
30 30 tier_id: Option<SubscriptionTierId>,
31 31 ) -> Result<DbPromoCode> {
32 + // runtime-checked: binds a chrono DateTime<Utc> param; a bind param's type can't be overridden in the macro when sqlx time+chrono features are unified.
32 33 let promo_code = sqlx::query_as::<_, DbPromoCode>(
33 34 r#"
34 35 INSERT INTO promo_codes (creator_id, code, code_purpose, discount_type, discount_value,
@@ -59,10 +60,19 @@
59 60 /// Fetch a promo code by primary key.
60 61 #[tracing::instrument(skip_all)]
61 62 pub async fn get_promo_code_by_id(pool: &PgPool, id: PromoCodeId) -> Result<Option<DbPromoCode>> {
62 - let code = sqlx::query_as::<_, DbPromoCode>(
63 - "SELECT * FROM promo_codes WHERE id = $1",
63 + let code = sqlx::query_as!(
64 + DbPromoCode,
65 + r#"SELECT id AS "id: PromoCodeId", creator_id AS "creator_id: UserId", code,
66 + code_purpose AS "code_purpose: super::CodePurpose",
67 + discount_type AS "discount_type: DiscountType", discount_value, min_price_cents,
68 + trial_days, item_id AS "item_id: ItemId", project_id AS "project_id: ProjectId",
69 + tier_id AS "tier_id: SubscriptionTierId", max_uses, use_count,
70 + expires_at AS "expires_at: chrono::DateTime<chrono::Utc>",
71 + starts_at AS "starts_at: chrono::DateTime<chrono::Utc>",
72 + created_at AS "created_at: chrono::DateTime<chrono::Utc>", is_platform_wide
73 + FROM promo_codes WHERE id = $1"#,
74 + id as PromoCodeId,
64 75 )
65 - .bind(id)
66 76 .fetch_optional(pool)
67 77 .await?;
68 78
@@ -77,11 +87,19 @@
77 87 creator_id: UserId,
78 88 code: &str,
79 89 ) -> Result<Option<DbPromoCode>> {
80 - let promo_code = sqlx::query_as::<_, DbPromoCode>(
81 - "SELECT * FROM promo_codes WHERE creator_id = $1 AND upper(code) = upper($2)",
90 + let promo_code = sqlx::query_as!(
91 + DbPromoCode,
92 + r#"SELECT id AS "id: PromoCodeId", creator_id AS "creator_id: UserId", code,
93 + code_purpose AS "code_purpose: super::CodePurpose",
94 + discount_type AS "discount_type: DiscountType", discount_value, min_price_cents,
95 + trial_days, item_id AS "item_id: ItemId", project_id AS "project_id: ProjectId",
96 + tier_id AS "tier_id: SubscriptionTierId", max_uses, use_count,
97 + expires_at AS "expires_at: chrono::DateTime<chrono::Utc>",
98 + starts_at AS "starts_at: chrono::DateTime<chrono::Utc>",
99 + created_at AS "created_at: chrono::DateTime<chrono::Utc>", is_platform_wide
100 + FROM promo_codes WHERE creator_id = $1 AND upper(code) = upper($2)"#,
101 + creator_id as UserId, code,
82 102 )
83 - .bind(creator_id)
84 - .bind(code)
85 103 .fetch_optional(pool)
86 104 .await?;
87 105
@@ -96,10 +114,19 @@
96 114 pool: &PgPool,
97 115 code: &str,
98 116 ) -> Result<Option<DbPromoCode>> {
99 - let promo_code = sqlx::query_as::<_, DbPromoCode>(
100 - "SELECT * FROM promo_codes WHERE upper(code) = upper($1) AND code_purpose = 'free_access'",
117 + let promo_code = sqlx::query_as!(
118 + DbPromoCode,
119 + r#"SELECT id AS "id: PromoCodeId", creator_id AS "creator_id: UserId", code,
120 + code_purpose AS "code_purpose: super::CodePurpose",
121 + discount_type AS "discount_type: DiscountType", discount_value, min_price_cents,
122 + trial_days, item_id AS "item_id: ItemId", project_id AS "project_id: ProjectId",
123 + tier_id AS "tier_id: SubscriptionTierId", max_uses, use_count,
124 + expires_at AS "expires_at: chrono::DateTime<chrono::Utc>",
125 + starts_at AS "starts_at: chrono::DateTime<chrono::Utc>",
126 + created_at AS "created_at: chrono::DateTime<chrono::Utc>", is_platform_wide
127 + FROM promo_codes WHERE upper(code) = upper($1) AND code_purpose = 'free_access'"#,
128 + code,
101 129 )
102 - .bind(code)
103 130 .fetch_optional(pool)
104 131 .await?;
105 132
@@ -118,6 +145,7 @@
118 145 /// List all promo codes for a creator, newest first. Capped at 500.
119 146 #[tracing::instrument(skip_all)]
120 147 pub async fn get_promo_codes_by_creator(pool: &PgPool, creator_id: UserId) -> Result<Vec<DbPromoCodeWithNames>> {
148 + // runtime-checked: dynamically-built SQL string (PROMO_CODE_WITH_NAMES_SELECT fragment).
121 149 let query = format!("{PROMO_CODE_WITH_NAMES_SELECT} WHERE pc.creator_id = $1 ORDER BY pc.created_at DESC LIMIT 500");
122 150 let codes = sqlx::query_as::<_, DbPromoCodeWithNames>(&query)
123 151 .bind(creator_id)
@@ -130,6 +158,7 @@
130 158 /// List all promo codes scoped to a project, newest first. Capped at 500.
131 159 #[tracing::instrument(skip_all)]
132 160 pub async fn get_promo_codes_by_project(pool: &PgPool, project_id: ProjectId) -> Result<Vec<DbPromoCodeWithNames>> {
161 + // runtime-checked: dynamically-built SQL string (PROMO_CODE_WITH_NAMES_SELECT fragment).
133 162 let query = format!("{PROMO_CODE_WITH_NAMES_SELECT} WHERE pc.project_id = $1 ORDER BY pc.created_at DESC LIMIT 500");
134 163 let codes = sqlx::query_as::<_, DbPromoCodeWithNames>(&query)
135 164 .bind(project_id)
@@ -142,6 +171,7 @@
142 171 /// List all promo codes scoped to an item, newest first. Capped at 500.
143 172 #[tracing::instrument(skip_all)]
144 173 pub async fn get_promo_codes_by_item(pool: &PgPool, item_id: ItemId) -> Result<Vec<DbPromoCodeWithNames>> {
174 + // runtime-checked: dynamically-built SQL string (PROMO_CODE_WITH_NAMES_SELECT fragment).
145 175 let query = format!("{PROMO_CODE_WITH_NAMES_SELECT} WHERE pc.item_id = $1 ORDER BY pc.created_at DESC LIMIT 500");
146 176 let codes = sqlx::query_as::<_, DbPromoCodeWithNames>(&query)
147 177 .bind(item_id)
@@ -157,6 +187,7 @@
157 187 pool: &PgPool,
158 188 item_ids: &[ItemId],
159 189 ) -> Result<std::collections::HashMap<ItemId, Vec<DbPromoCodeWithNames>>> {
190 + // runtime-checked: dynamically-built SQL string (PROMO_CODE_WITH_NAMES_SELECT fragment).
160 191 let query = format!("{PROMO_CODE_WITH_NAMES_SELECT} WHERE pc.item_id = ANY($1) ORDER BY pc.item_id, pc.created_at DESC");
161 192 let codes = sqlx::query_as::<_, DbPromoCodeWithNames>(&query)
162 193 .bind(item_ids)
@@ -185,14 +216,14 @@
185 216 executor: impl sqlx::PgExecutor<'e>,
186 217 id: PromoCodeId,
187 218 ) -> Result<bool> {
188 - let result = sqlx::query(
219 + let result = sqlx::query!(
189 220 "UPDATE promo_codes SET use_count = use_count + 1 \
190 221 WHERE id = $1 \
191 222 AND (max_uses IS NULL OR use_count < max_uses) \
192 223 AND (expires_at IS NULL OR expires_at > NOW()) \
193 224 AND (starts_at IS NULL OR starts_at <= NOW())",
225 + id as PromoCodeId,
194 226 )
195 - .bind(id)
196 227 .execute(executor)
197 228 .await?;
198 229
@@ -218,10 +249,10 @@
218 249 /// zero) but the structural fix above prevents it from happening at all.
219 250 #[tracing::instrument(skip_all)]
220 251 pub async fn release_use_count(pool: &PgPool, id: PromoCodeId) -> Result<()> {
221 - sqlx::query(
252 + sqlx::query!(
222 253 "UPDATE promo_codes SET use_count = GREATEST(0, use_count - 1) WHERE id = $1",
254 + id as PromoCodeId,
223 255 )
224 - .bind(id)
225 256 .execute(pool)
226 257 .await?;
227 258
@@ -245,19 +276,19 @@
245 276 ) -> Result<()> {
246 277 let mut tx = pool.begin().await?;
247 278
248 - sqlx::query(
279 + sqlx::query!(
249 280 "UPDATE transactions SET promo_code_id = NULL \
250 281 WHERE buyer_id = $1 AND promo_code_id = $2 AND status = 'pending'",
282 + buyer_id as UserId,
283 + id as PromoCodeId,
251 284 )
252 - .bind(buyer_id)
253 - .bind(id)
254 285 .execute(&mut *tx)
255 286 .await?;
256 287
257 - sqlx::query(
288 + sqlx::query!(
258 289 "UPDATE promo_codes SET use_count = GREATEST(0, use_count - 1) WHERE id = $1",
290 + id as PromoCodeId,
259 291 )
260 - .bind(id)
261 292 .execute(&mut *tx)
262 293 .await?;
263 294
@@ -300,6 +331,7 @@
300 331 .ok_or_else(|| crate::error::AppError::NotFound);
301 332 }
302 333
334 + // runtime-checked: dynamically-built SQL string (SET clause assembled from provided fields).
303 335 let sql = format!("UPDATE promo_codes SET {} WHERE id = $1 RETURNING *", sets.join(", "));
304 336 let mut query = sqlx::query_as::<_, DbPromoCode>(&sql).bind(id);
305 337
@@ -320,10 +352,10 @@
320 352 /// Delete all expired promo codes for a creator. Returns number of rows deleted.
321 353 #[tracing::instrument(skip_all)]
322 354 pub async fn delete_expired_by_creator(pool: &PgPool, creator_id: UserId) -> Result<u64> {
323 - let result = sqlx::query(
355 + let result = sqlx::query!(
324 356 "DELETE FROM promo_codes WHERE creator_id = $1 AND expires_at IS NOT NULL AND expires_at < NOW()",
357 + creator_id as UserId,
325 358 )
326 - .bind(creator_id)
327 359 .execute(pool)
328 360 .await?;
329 361
@@ -333,8 +365,7 @@
333 365 /// Delete a promo code permanently.
334 366 #[tracing::instrument(skip_all)]
335 367 pub async fn delete_promo_code(pool: &PgPool, id: PromoCodeId) -> Result<()> {
336 - sqlx::query("DELETE FROM promo_codes WHERE id = $1")
337 - .bind(id)
368 + sqlx::query!("DELETE FROM promo_codes WHERE id = $1", id as PromoCodeId)
338 369 .execute(pool)
339 370 .await?;
340 371
@@ -367,12 +398,13 @@
367 398 pool: &PgPool,
368 399 id: PromoCodeId,
369 400 ) -> Result<Vec<PromoRedemption>> {
370 - let rows = sqlx::query_as::<_, PromoRedemption>(
401 + let rows = sqlx::query_as!(
402 + PromoRedemption,
371 403 r#"
372 404 SELECT
373 - COALESCE(t.completed_at, t.created_at) AS redeemed_at,
405 + COALESCE(t.completed_at, t.created_at) AS "redeemed_at!: chrono::DateTime<chrono::Utc>",
374 406 u.display_name AS display_name,
375 - u.username AS username,
407 + u.username AS "username?",
376 408 t.guest_email AS guest_email,
377 409 t.item_title AS item_title,
378 410 t.amount_cents AS amount_cents
@@ -380,11 +412,11 @@
380 412 LEFT JOIN users u ON u.id = t.buyer_id
381 413 WHERE t.promo_code_id = $1
382 414 AND t.status = 'completed'
383 - ORDER BY redeemed_at DESC
415 + ORDER BY COALESCE(t.completed_at, t.created_at) DESC
384 416 LIMIT 500
385 417 "#,
418 + id as PromoCodeId,
386 419 )
387 - .bind(id)
388 420 .fetch_all(pool)
389 421 .await?;
390 422
@@ -409,6 +441,7 @@
409 441 max_uses: Option<i32>,
410 442 expires_at: Option<chrono::DateTime<chrono::Utc>>,
411 443 ) -> Result<DbPromoCode> {
444 + // runtime-checked: binds a chrono DateTime<Utc> param; a bind param's type can't be overridden in the macro when sqlx time+chrono features are unified.
412 445 let promo_code = sqlx::query_as::<_, DbPromoCode>(
413 446 r#"
414 447 INSERT INTO promo_codes (creator_id, code, code_purpose, discount_type, discount_value,
@@ -468,12 +501,12 @@
468 501 stripe_sub_id: &str,
469 502 period_end: i64,
470 503 ) -> Result<Option<FanPlusCreditClaim>> {
471 - let result = sqlx::query(
504 + let result = sqlx::query!(
472 505 "INSERT INTO fan_plus_credit_issuance (stripe_sub_id, period_end) \
473 506 VALUES ($1, $2) ON CONFLICT DO NOTHING",
507 + stripe_sub_id,
508 + period_end,
474 509 )
475 - .bind(stripe_sub_id)
476 - .bind(period_end)
477 510 .execute(pool)
478 511 .await?;
479 512
@@ -520,11 +553,19 @@
520 553 user_id: UserId,
521 554 code: &str,
522 555 ) -> Result<Option<DbPromoCode>> {
523 - let promo_code = sqlx::query_as::<_, DbPromoCode>(
524 - "SELECT * FROM promo_codes WHERE creator_id = $1 AND upper(code) = upper($2) AND is_platform_wide = true",
556 + let promo_code = sqlx::query_as!(
557 + DbPromoCode,
558 + r#"SELECT id AS "id: PromoCodeId", creator_id AS "creator_id: UserId", code,
559 + code_purpose AS "code_purpose: super::CodePurpose",
560 + discount_type AS "discount_type: DiscountType", discount_value, min_price_cents,
561 + trial_days, item_id AS "item_id: ItemId", project_id AS "project_id: ProjectId",
562 + tier_id AS "tier_id: SubscriptionTierId", max_uses, use_count,
563 + expires_at AS "expires_at: chrono::DateTime<chrono::Utc>",
564 + starts_at AS "starts_at: chrono::DateTime<chrono::Utc>",
565 + created_at AS "created_at: chrono::DateTime<chrono::Utc>", is_platform_wide
566 + FROM promo_codes WHERE creator_id = $1 AND upper(code) = upper($2) AND is_platform_wide = true"#,
567 + user_id as UserId, code,
525 568 )
526 - .bind(user_id)
527 - .bind(code)
528 569 .fetch_optional(pool)
529 570 .await?;
530 571
@@ -543,11 +584,20 @@
543 584 pool: &PgPool,
544 585 code: &str,
545 586 ) -> Result<Option<DbPromoCode>> {
546 - let promo_code = sqlx::query_as::<_, DbPromoCode>(
547 - "SELECT * FROM promo_codes \
548 - WHERE upper(code) = upper($1) AND code_purpose = 'free_trial' AND is_platform_wide = true",
587 + let promo_code = sqlx::query_as!(
588 + DbPromoCode,
589 + r#"SELECT id AS "id: PromoCodeId", creator_id AS "creator_id: UserId", code,
590 + code_purpose AS "code_purpose: super::CodePurpose",
591 + discount_type AS "discount_type: DiscountType", discount_value, min_price_cents,
592 + trial_days, item_id AS "item_id: ItemId", project_id AS "project_id: ProjectId",
593 + tier_id AS "tier_id: SubscriptionTierId", max_uses, use_count,
594 + expires_at AS "expires_at: chrono::DateTime<chrono::Utc>",
595 + starts_at AS "starts_at: chrono::DateTime<chrono::Utc>",
596 + created_at AS "created_at: chrono::DateTime<chrono::Utc>", is_platform_wide
597 + FROM promo_codes
598 + WHERE upper(code) = upper($1) AND code_purpose = 'free_trial' AND is_platform_wide = true"#,
599 + code,
549 600 )
550 - .bind(code)
551 601 .fetch_optional(pool)
552 602 .await?;
553 603
@@ -566,12 +616,12 @@
566 616 code_id: PromoCodeId,
567 617 user_id: UserId,
568 618 ) -> Result<bool> {
569 - let result = sqlx::query(
619 + let result = sqlx::query!(
570 620 "INSERT INTO promo_code_redemptions (promo_code_id, user_id) \
571 621 VALUES ($1, $2) ON CONFLICT DO NOTHING",
622 + code_id as PromoCodeId,
623 + user_id as UserId,
572 624 )
573 - .bind(code_id)
574 - .bind(user_id)
575 625 .execute(pool)
576 626 .await?;
577 627
@@ -583,11 +633,13 @@
583 633 /// redemption row was inserted.
584 634 #[tracing::instrument(skip_all)]
585 635 pub async fn remove_redemption(pool: &PgPool, code_id: PromoCodeId, user_id: UserId) -> Result<()> {
586 - sqlx::query("DELETE FROM promo_code_redemptions WHERE promo_code_id = $1 AND user_id = $2")
587 - .bind(code_id)
588 - .bind(user_id)
589 - .execute(pool)
590 - .await?;
636 + sqlx::query!(
637 + "DELETE FROM promo_code_redemptions WHERE promo_code_id = $1 AND user_id = $2",
638 + code_id as PromoCodeId,
639 + user_id as UserId,
640 + )
641 + .execute(pool)
642 + .await?;
591 643 Ok(())
592 644 }
593 645
@@ -595,10 +647,19 @@
595 647 /// Powers the admin comp-codes dashboard. Capped at 500.
596 648 #[tracing::instrument(skip_all)]
597 649 pub async fn get_platform_trial_codes(pool: &PgPool) -> Result<Vec<DbPromoCode>> {
598 - let codes = sqlx::query_as::<_, DbPromoCode>(
599 - "SELECT * FROM promo_codes \
600 - WHERE code_purpose = 'free_trial' AND is_platform_wide = true \
601 - ORDER BY created_at DESC LIMIT 500",
650 + let codes = sqlx::query_as!(
651 + DbPromoCode,
652 + r#"SELECT id AS "id: PromoCodeId", creator_id AS "creator_id: UserId", code,
653 + code_purpose AS "code_purpose: super::CodePurpose",
654 + discount_type AS "discount_type: DiscountType", discount_value, min_price_cents,
655 + trial_days, item_id AS "item_id: ItemId", project_id AS "project_id: ProjectId",
656 + tier_id AS "tier_id: SubscriptionTierId", max_uses, use_count,
657 + expires_at AS "expires_at: chrono::DateTime<chrono::Utc>",
658 + starts_at AS "starts_at: chrono::DateTime<chrono::Utc>",
659 + created_at AS "created_at: chrono::DateTime<chrono::Utc>", is_platform_wide
660 + FROM promo_codes
661 + WHERE code_purpose = 'free_trial' AND is_platform_wide = true
662 + ORDER BY created_at DESC LIMIT 500"#,
602 663 )
603 664 .fetch_all(pool)
604 665 .await?;
@@ -13,12 +13,12 @@
13 13 user_agent: Option<&str>,
14 14 ip_address: Option<&str>,
15 15 ) -> Result<UserSessionId> {
16 - let row = sqlx::query_scalar::<_, UserSessionId>(
17 - "INSERT INTO user_sessions (user_id, user_agent, ip_address) VALUES ($1, $2, $3) RETURNING id",
16 + let row = sqlx::query_scalar!(
17 + r#"INSERT INTO user_sessions (user_id, user_agent, ip_address) VALUES ($1, $2, $3) RETURNING id AS "id: UserSessionId""#,
18 + user_id as UserId,
19 + user_agent,
20 + ip_address,
18 21 )
19 - .bind(user_id)
20 - .bind(user_agent)
21 - .bind(ip_address)
22 22 .fetch_one(pool)
23 23 .await?;
24 24
@@ -38,7 +38,7 @@
38 38 user_id: UserId,
39 39 max: i64,
40 40 ) -> Result<u64> {
41 - let result = sqlx::query(
41 + let result = sqlx::query!(
42 42 r#"
43 43 DELETE FROM user_sessions
44 44 WHERE user_id = $1 AND kind = 'active'
@@ -49,9 +49,9 @@
49 49 LIMIT $2
50 50 )
51 51 "#,
52 + user_id as UserId,
53 + max,
52 54 )
53 - .bind(user_id)
54 - .bind(max)
55 55 .execute(pool)
56 56 .await?;
57 57
@@ -69,13 +69,13 @@
69 69 user_agent: Option<&str>,
70 70 ip_address: Option<&str>,
71 71 ) -> Result<UserSessionId> {
72 - let row = sqlx::query_scalar::<_, UserSessionId>(
73 - "INSERT INTO user_sessions (user_id, user_agent, ip_address, kind)
74 - VALUES ($1, $2, $3, 'pending_2fa') RETURNING id",
72 + let row = sqlx::query_scalar!(
73 + r#"INSERT INTO user_sessions (user_id, user_agent, ip_address, kind)
74 + VALUES ($1, $2, $3, 'pending_2fa') RETURNING id AS "id: UserSessionId""#,
75 + user_id as UserId,
76 + user_agent,
77 + ip_address,
75 78 )
76 - .bind(user_id)
77 - .bind(user_agent)
78 - .bind(ip_address)
79 79 .fetch_one(pool)
80 80 .await?;
81 81
@@ -90,11 +90,11 @@
90 90 id: UserSessionId,
91 91 user_id: UserId,
92 92 ) -> Result<bool> {
93 - let exists: bool = sqlx::query_scalar(
94 - "SELECT EXISTS(SELECT 1 FROM user_sessions WHERE id = $1 AND user_id = $2 AND kind = 'pending_2fa')",
93 + let exists = sqlx::query_scalar!(
94 + r#"SELECT EXISTS(SELECT 1 FROM user_sessions WHERE id = $1 AND user_id = $2 AND kind = 'pending_2fa') AS "exists!""#,
95 + id as UserSessionId,
96 + user_id as UserId,
95 97 )
96 - .bind(id)
97 - .bind(user_id)
98 98 .fetch_one(pool)
99 99 .await?;
100 100 Ok(exists)
@@ -105,10 +105,12 @@
105 105 /// cleared (expiry, account lockout, navigation away).
106 106 #[tracing::instrument(skip_all)]
107 107 pub async fn delete_pending_2fa_session(pool: &PgPool, id: UserSessionId) -> Result<()> {
108 - sqlx::query("DELETE FROM user_sessions WHERE id = $1 AND kind = 'pending_2fa'")
109 - .bind(id)
110 - .execute(pool)
111 - .await?;
108 + sqlx::query!(
109 + "DELETE FROM user_sessions WHERE id = $1 AND kind = 'pending_2fa'",
110 + id as UserSessionId,
111 + )
112 + .execute(pool)
113 + .await?;
112 114 Ok(())
113 115 }
114 116
@@ -138,20 +140,21 @@
138 140 pub async fn touch_session(pool: &PgPool, session_id: UserSessionId) -> Result<TouchResult> {
139 141 // Single query: update last_active_at, join users for live status, and check
140 142 // fan_plus + creator_tier via subqueries (avoids 2 extra round-trips in auth extractor).
141 - let row = sqlx::query_as::<_, (bool, bool, bool, Option<String>)>(
143 + let row = sqlx::query!(
142 144 r#"
143 145 UPDATE user_sessions us
144 146 SET last_active_at = NOW()
145 147 FROM users u
146 148 WHERE us.id = $1 AND u.id = us.user_id
147 149 RETURNING
148 - u.suspended_at IS NOT NULL,
149 - u.can_create_projects,
150 - EXISTS(SELECT 1 FROM fan_plus_subscriptions fps WHERE fps.user_id = u.id AND fps.status = 'active'),
151 - (SELECT cs.tier FROM creator_subscriptions cs WHERE cs.user_id = u.id AND cs.status = 'active')
150 + u.suspended_at IS NOT NULL AS "suspended!",
151 + u.can_create_projects AS "can_create_projects!",
152 + EXISTS(SELECT 1 FROM fan_plus_subscriptions fps WHERE fps.user_id = u.id AND fps.status = 'active') AS "is_fan_plus!",
153 + (SELECT cs.tier FROM creator_subscriptions cs WHERE cs.user_id = u.id AND cs.status = 'active') AS "creator_tier"
152 154 "#,
155 + session_id as UserSessionId,
153 156 )
154 - .bind(session_id)
157 + .map(|r| (r.suspended, r.can_create_projects, r.is_fan_plus, r.creator_tier))
155 158 .fetch_optional(pool)
156 159 .await?;
157 160
@@ -170,14 +173,18 @@
170 173 /// List all active sessions for a user, newest first.
171 174 #[tracing::instrument(skip_all)]
172 175 pub async fn get_user_sessions(pool: &PgPool, user_id: UserId) -> Result<Vec<DbUserSession>> {
173 - let sessions = sqlx::query_as::<_, DbUserSession>(
174 - "SELECT id, user_id, created_at, last_active_at, user_agent, ip_address
176 + let sessions = sqlx::query_as!(
177 + DbUserSession,
178 + r#"SELECT id AS "id: UserSessionId", user_id AS "user_id: UserId",
179 + created_at AS "created_at: chrono::DateTime<chrono::Utc>",
180 + last_active_at AS "last_active_at: chrono::DateTime<chrono::Utc>",
181 + user_agent, ip_address
175 182 FROM user_sessions
176 183 WHERE user_id = $1
177 184 ORDER BY last_active_at DESC
178 - LIMIT 100",
185 + LIMIT 100"#,
186 + user_id as UserId,
179 187 )
180 - .bind(user_id)
181 188 .fetch_all(pool)
182 189 .await?;
183 190
@@ -187,10 +194,10 @@
187 194 /// Count active sessions for a user.
188 195 #[tracing::instrument(skip_all)]
189 196 pub async fn count_user_sessions(pool: &PgPool, user_id: UserId) -> Result<i64> {
190 - let count = sqlx::query_scalar::<_, i64>(
191 - "SELECT COUNT(*) FROM user_sessions WHERE user_id = $1",
197 + let count = sqlx::query_scalar!(
198 + r#"SELECT COUNT(*) AS "count!" FROM user_sessions WHERE user_id = $1"#,
199 + user_id as UserId,
192 200 )
193 - .bind(user_id)
194 201 .fetch_one(pool)
195 202 .await?;
196 203
@@ -204,11 +211,11 @@
204 211 session_id: UserSessionId,
205 212 user_id: UserId,
206 213 ) -> Result<bool> {
207 - let rows = sqlx::query(
214 + let rows = sqlx::query!(
208 215 "DELETE FROM user_sessions WHERE id = $1 AND user_id = $2",
216 + session_id as UserSessionId,
217 + user_id as UserId,
209 218 )
210 - .bind(session_id)
211 - .bind(user_id)
212 219 .execute(pool)
213 220 .await?;
214 221
@@ -229,11 +236,13 @@
229 236 session_id: UserSessionId,
230 237 user_id: UserId,
231 238 ) -> Result<bool> {
232 - let rows = sqlx::query("DELETE FROM user_sessions WHERE id = $1 AND user_id = $2")
233 - .bind(session_id)
234 - .bind(user_id)
235 - .execute(pool)
236 - .await?;
239 + let rows = sqlx::query!(
240 + "DELETE FROM user_sessions WHERE id = $1 AND user_id = $2",
241 + session_id as UserSessionId,
242 + user_id as UserId,
243 + )
244 + .execute(pool)
245 + .await?;
237 246
238 247 Ok(rows.rows_affected() > 0)
239 248 }
@@ -242,6 +251,7 @@
242 251 /// Returns the number of rows removed.
243 252 #[tracing::instrument(skip_all)]
244 253 pub async fn prune_expired_sessions(pool: &PgPool, stale_threshold: chrono::DateTime<chrono::Utc>) -> Result<u64> {
254 + // runtime-checked: binds a chrono DateTime<Utc> param; a bind param's type can't be overridden in the macro when sqlx time+chrono features are unified.
245 255 let result = sqlx::query(
246 256 "DELETE FROM user_sessions WHERE last_active_at < $1",
247 257 )
@@ -259,11 +269,11 @@
259 269 current_session_id: UserSessionId,
260 270 user_id: UserId,
261 271 ) -> Result<Vec<UserSessionId>> {
262 - let ids: Vec<UserSessionId> = sqlx::query_scalar(
263 - "DELETE FROM user_sessions WHERE user_id = $1 AND id != $2 RETURNING id",
272 + let ids = sqlx::query_scalar!(
273 + r#"DELETE FROM user_sessions WHERE user_id = $1 AND id != $2 RETURNING id AS "id: UserSessionId""#,
274 + user_id as UserId,
275 + current_session_id as UserSessionId,
264 276 )
265 - .bind(user_id)
266 - .bind(current_session_id)
267 277 .fetch_all(pool)
268 278 .await?;
269 279
@@ -283,17 +293,19 @@
283 293 ) -> Result<Vec<UserSessionId>> {
284 294 let mut tx = pool.begin().await?;
285 295
286 - let ids: Vec<UserSessionId> = sqlx::query_scalar(
287 - "DELETE FROM user_sessions WHERE user_id = $1 RETURNING id",
296 + let ids = sqlx::query_scalar!(
297 + r#"DELETE FROM user_sessions WHERE user_id = $1 RETURNING id AS "id: UserSessionId""#,
298 + user_id as UserId,
288 299 )
289 - .bind(user_id)
290 300 .fetch_all(&mut *tx)
291 301 .await?;
292 302
293 - sqlx::query("UPDATE users SET jwt_invalidated_at = NOW() WHERE id = $1")
294 - .bind(user_id)
295 - .execute(&mut *tx)
296 - .await?;
303 + sqlx::query!(
304 + "UPDATE users SET jwt_invalidated_at = NOW() WHERE id = $1",
305 + user_id as UserId,
306 + )
307 + .execute(&mut *tx)
308 + .await?;
297 309
298 310 tx.commit().await?;
299 311
@@ -3,7 +3,7 @@
3 3 use sqlx::PgPool;
4 4
5 5 use super::models::*;
6 - use super::{PriceCents, ProjectId, SubscriptionId, SubscriptionTierId, UserId};
6 + use super::{ItemId, PriceCents, ProjectId, SubscriptionId, SubscriptionTierId, UserId};
7 7 use crate::error::Result;
8 8
9 9 // ── Tier CRUD ──
@@ -17,17 +17,23 @@
17 17 description: Option<&str>,
18 18 price_cents: PriceCents,
19 19 ) -> Result<DbSubscriptionTier> {
20 - let tier = sqlx::query_as::<_, DbSubscriptionTier>(
20 + let tier = sqlx::query_as!(
21 + DbSubscriptionTier,
21 22 r#"
22 23 INSERT INTO subscription_tiers (project_id, name, description, price_cents)
23 24 VALUES ($1, $2, $3, $4)
24 - RETURNING *
25 + RETURNING id AS "id: SubscriptionTierId", project_id AS "project_id: ProjectId",
26 + name, description, price_cents, stripe_product_id, stripe_price_id,
27 + sort_order, is_active,
28 + created_at AS "created_at: chrono::DateTime<chrono::Utc>",
29 + updated_at AS "updated_at: chrono::DateTime<chrono::Utc>",
30 + item_id AS "item_id: ItemId"
25 31 "#,
32 + project_id as ProjectId,
33 + name,
34 + description,
35 + price_cents.as_i32(),
26 36 )
27 - .bind(project_id)
28 - .bind(name)
29 - .bind(description)
30 - .bind(price_cents.as_i32())
31 37 .fetch_one(pool)
32 38 .await?;
33 39
@@ -40,10 +46,19 @@
40 46 pool: &PgPool,
41 47 id: SubscriptionTierId,
42 48 ) -> Result<Option<DbSubscriptionTier>> {
43 - let tier = sqlx::query_as::<_, DbSubscriptionTier>(
44 - "SELECT * FROM subscription_tiers WHERE id = $1",
49 + let tier = sqlx::query_as!(
50 + DbSubscriptionTier,
51 + r#"
52 + SELECT id AS "id: SubscriptionTierId", project_id AS "project_id: ProjectId",
53 + name, description, price_cents, stripe_product_id, stripe_price_id,
54 + sort_order, is_active,
55 + created_at AS "created_at: chrono::DateTime<chrono::Utc>",
56 + updated_at AS "updated_at: chrono::DateTime<chrono::Utc>",
57 + item_id AS "item_id: ItemId"
58 + FROM subscription_tiers WHERE id = $1
59 + "#,
60 + id as SubscriptionTierId,
45 61 )
46 - .bind(id)
47 62 .fetch_optional(pool)
48 63 .await?;
49 64
@@ -56,10 +71,19 @@
56 71 pool: &PgPool,
57 72 project_id: ProjectId,
58 73 ) -> Result<Vec<DbSubscriptionTier>> {
59 - let tiers = sqlx::query_as::<_, DbSubscriptionTier>(
60 - "SELECT * FROM subscription_tiers WHERE project_id = $1 AND is_active = true ORDER BY sort_order, created_at",
74 + let tiers = sqlx::query_as!(
75 + DbSubscriptionTier,
76 + r#"
77 + SELECT id AS "id: SubscriptionTierId", project_id AS "project_id: ProjectId",
78 + name, description, price_cents, stripe_product_id, stripe_price_id,
79 + sort_order, is_active,
80 + created_at AS "created_at: chrono::DateTime<chrono::Utc>",
81 + updated_at AS "updated_at: chrono::DateTime<chrono::Utc>",
82 + item_id AS "item_id: ItemId"
83 + FROM subscription_tiers WHERE project_id = $1 AND is_active = true ORDER BY sort_order, created_at
84 + "#,
85 + project_id as ProjectId,
61 86 )
62 - .bind(project_id)
63 87 .fetch_all(pool)
64 88 .await?;
65 89
@@ -72,10 +96,19 @@
72 96 pool: &PgPool,
73 97 project_id: ProjectId,
74 98 ) -> Result<Vec<DbSubscriptionTier>> {
75 - let tiers = sqlx::query_as::<_, DbSubscriptionTier>(
76 - "SELECT * FROM subscription_tiers WHERE project_id = $1 ORDER BY sort_order, created_at",
99 + let tiers = sqlx::query_as!(
100 + DbSubscriptionTier,
101 + r#"
102 + SELECT id AS "id: SubscriptionTierId", project_id AS "project_id: ProjectId",
103 + name, description, price_cents, stripe_product_id, stripe_price_id,
104 + sort_order, is_active,
105 + created_at AS "created_at: chrono::DateTime<chrono::Utc>",
106 + updated_at AS "updated_at: chrono::DateTime<chrono::Utc>",
107 + item_id AS "item_id: ItemId"
108 + FROM subscription_tiers WHERE project_id = $1 ORDER BY sort_order, created_at
109 + "#,
110 + project_id as ProjectId,
77 111 )
78 - .bind(project_id)
79 112 .fetch_all(pool)
80 113 .await?;
81 114
@@ -91,18 +124,24 @@
91 124 description: Option<&str>,
92 125 is_active: bool,
93 126 ) -> Result<DbSubscriptionTier> {
94 - let tier = sqlx::query_as::<_, DbSubscriptionTier>(
127 + let tier = sqlx::query_as!(
128 + DbSubscriptionTier,
95 129 r#"
96 130 UPDATE subscription_tiers
97 131 SET name = $2, description = $3, is_active = $4
98 132 WHERE id = $1
99 - RETURNING *
133 + RETURNING id AS "id: SubscriptionTierId", project_id AS "project_id: ProjectId",
134 + name, description, price_cents, stripe_product_id, stripe_price_id,
135 + sort_order, is_active,
136 + created_at AS "created_at: chrono::DateTime<chrono::Utc>",
137 + updated_at AS "updated_at: chrono::DateTime<chrono::Utc>",
138 + item_id AS "item_id: ItemId"
100 139 "#,
140 + id as SubscriptionTierId,
141 + name,
142 + description,
143 + is_active,
101 144 )
102 - .bind(id)
103 - .bind(name)
104 - .bind(description)
105 - .bind(is_active)
106 145 .fetch_one(pool)
107 146 .await?;
108 147
@@ -117,16 +156,16 @@
117 156 product_id: &str,
118 157 price_id: &str,
119 158 ) -> Result<()> {
120 - sqlx::query(
159 + sqlx::query!(
121 160 r#"
122 161 UPDATE subscription_tiers
123 162 SET stripe_product_id = $2, stripe_price_id = $3
124 163 WHERE id = $1
125 164 "#,
165 + tier_id as SubscriptionTierId,
166 + product_id,
167 + price_id,
126 168 )
127 - .bind(tier_id)
128 - .bind(product_id)
129 - .bind(price_id)
130 169 .execute(pool)
131 170 .await?;
132 171
@@ -143,29 +182,35 @@
143 182 let mut tx = pool.begin().await?;
144 183
145 184 // Lock the tier row to serialize against concurrent subscription creation
146 - sqlx::query("SELECT id FROM subscription_tiers WHERE id = $1 FOR UPDATE")
147 - .bind(id)
148 - .fetch_optional(&mut *tx)
149 - .await?
150 - .ok_or(sqlx::Error::RowNotFound)?;
151 -
152 - let has_subscriptions: bool = sqlx::query_scalar(
153 - "SELECT EXISTS(SELECT 1 FROM subscriptions WHERE tier_id = $1)",
185 + sqlx::query!(
186 + "SELECT id FROM subscription_tiers WHERE id = $1 FOR UPDATE",
187 + id as SubscriptionTierId
188 + )
189 + .fetch_optional(&mut *tx)
190 + .await?
191 + .ok_or(sqlx::Error::RowNotFound)?;
192 +
193 + let has_subscriptions: bool = sqlx::query_scalar!(
194 + r#"SELECT EXISTS(SELECT 1 FROM subscriptions WHERE tier_id = $1) AS "exists!""#,
195 + id as SubscriptionTierId,
154 196 )
155 - .bind(id)
156 197 .fetch_one(&mut *tx)
157 198 .await?;
158 199
159 200 if has_subscriptions {
160 - sqlx::query("UPDATE subscription_tiers SET is_active = false WHERE id = $1")
161 - .bind(id)
162 - .execute(&mut *tx)
163 - .await?;
201 + sqlx::query!(
202 + "UPDATE subscription_tiers SET is_active = false WHERE id = $1",
203 + id as SubscriptionTierId
204 + )
205 + .execute(&mut *tx)
206 + .await?;
164 207 } else {
165 - sqlx::query("DELETE FROM subscription_tiers WHERE id = $1")
166 - .bind(id)
167 - .execute(&mut *tx)
168 - .await?;
208 + sqlx::query!(
209 + "DELETE FROM subscription_tiers WHERE id = $1",
210 + id as SubscriptionTierId
211 + )
212 + .execute(&mut *tx)
213 + .await?;
169 214 }
170 215
171 216 tx.commit().await?;
@@ -189,19 +234,30 @@
189 234 stripe_subscription_id: &str,
190 235 stripe_customer_id: &str,
191 236 ) -> Result<Option<DbSubscription>> {
192 - let sub = sqlx::query_as::<_, DbSubscription>(
237 + let sub = sqlx::query_as!(
238 + DbSubscription,
193 239 r#"
194 240 INSERT INTO subscriptions (subscriber_id, tier_id, project_id, stripe_subscription_id, stripe_customer_id)
195 241 VALUES ($1, $2, $3, $4, $5)
196 242 ON CONFLICT DO NOTHING
197 - RETURNING *
243 + RETURNING id AS "id: SubscriptionId", subscriber_id AS "subscriber_id: UserId",
244 + tier_id AS "tier_id: SubscriptionTierId", project_id AS "project_id: ProjectId",
245 + stripe_subscription_id, stripe_customer_id,
246 + status AS "status: super::SubscriptionStatus",
247 + current_period_start AS "current_period_start: chrono::DateTime<chrono::Utc>",
248 + current_period_end AS "current_period_end: chrono::DateTime<chrono::Utc>",
249 + canceled_at AS "canceled_at: chrono::DateTime<chrono::Utc>",
250 + created_at AS "created_at: chrono::DateTime<chrono::Utc>",
251 + updated_at AS "updated_at: chrono::DateTime<chrono::Utc>",
252 + item_id AS "item_id: ItemId",
253 + paused_at AS "paused_at: chrono::DateTime<chrono::Utc>"
198 254 "#,
255 + subscriber_id as UserId,
256 + tier_id as SubscriptionTierId,
257 + project_id as ProjectId,
258 + stripe_subscription_id,
259 + stripe_customer_id,
199 260 )
200 - .bind(subscriber_id)
201 - .bind(tier_id)
202 - .bind(project_id)
203 - .bind(stripe_subscription_id)
204 - .bind(stripe_customer_id)
205 261 .fetch_optional(executor)
206 262 .await?;
207 263
@@ -214,10 +270,24 @@
214 270 pool: &PgPool,
215 271 stripe_sub_id: &str,
216 272 ) -> Result<Option<DbSubscription>> {
217 - let sub = sqlx::query_as::<_, DbSubscription>(
218 - "SELECT * FROM subscriptions WHERE stripe_subscription_id = $1",
273 + let sub = sqlx::query_as!(
274 + DbSubscription,
275 + r#"
276 + SELECT id AS "id: SubscriptionId", subscriber_id AS "subscriber_id: UserId",
277 + tier_id AS "tier_id: SubscriptionTierId", project_id AS "project_id: ProjectId",
278 + stripe_subscription_id, stripe_customer_id,
279 + status AS "status: super::SubscriptionStatus",
280 + current_period_start AS "current_period_start: chrono::DateTime<chrono::Utc>",
281 + current_period_end AS "current_period_end: chrono::DateTime<chrono::Utc>",
282 + canceled_at AS "canceled_at: chrono::DateTime<chrono::Utc>",
283 + created_at AS "created_at: chrono::DateTime<chrono::Utc>",
284 + updated_at AS "updated_at: chrono::DateTime<chrono::Utc>",
285 + item_id AS "item_id: ItemId",
286 + paused_at AS "paused_at: chrono::DateTime<chrono::Utc>"
287 + FROM subscriptions WHERE stripe_subscription_id = $1
288 + "#,
289 + stripe_sub_id,
219 290 )
220 - .bind(stripe_sub_id)
221 291 .fetch_optional(pool)
222 292 .await?;
223 293
@@ -243,15 +313,26 @@
243 313 pool: &PgPool,
244 314 stripe_sub_id: &str,
245 315 ) -> Result<Option<DbSubscription>> {
246 - let sub = sqlx::query_as::<_, DbSubscription>(
316 + let sub = sqlx::query_as!(
317 + DbSubscription,
247 318 r#"
248 319 UPDATE subscriptions
249 320 SET status = 'canceled', canceled_at = COALESCE(canceled_at, NOW())
250 321 WHERE stripe_subscription_id = $1
251 - RETURNING *
322 + RETURNING id AS "id: SubscriptionId", subscriber_id AS "subscriber_id: UserId",
323 + tier_id AS "tier_id: SubscriptionTierId", project_id AS "project_id: ProjectId",
324 + stripe_subscription_id, stripe_customer_id,
325 + status AS "status: super::SubscriptionStatus",
326 + current_period_start AS "current_period_start: chrono::DateTime<chrono::Utc>",
327 + current_period_end AS "current_period_end: chrono::DateTime<chrono::Utc>",
328 + canceled_at AS "canceled_at: chrono::DateTime<chrono::Utc>",
329 + created_at AS "created_at: chrono::DateTime<chrono::Utc>",
330 + updated_at AS "updated_at: chrono::DateTime<chrono::Utc>",
331 + item_id AS "item_id: ItemId",
332 + paused_at AS "paused_at: chrono::DateTime<chrono::Utc>"
252 333 "#,
334 + stripe_sub_id,
253 335 )
254 - .bind(stripe_sub_id)
255 336 .fetch_optional(pool)
256 337 .await?;
257 338
@@ -266,15 +347,27 @@
266 347 pool: &PgPool,
267 348 creator_id: UserId,
268 349 ) -> Result<Vec<DbSubscription>> {
269 - let subs = sqlx::query_as::<_, DbSubscription>(
350 + let subs = sqlx::query_as!(
351 + DbSubscription,
270 352 r#"
271 - SELECT s.* FROM subscriptions s
353 + SELECT s.id AS "id: SubscriptionId", s.subscriber_id AS "subscriber_id: UserId",
354 + s.tier_id AS "tier_id: SubscriptionTierId", s.project_id AS "project_id: ProjectId",
355 + s.stripe_subscription_id, s.stripe_customer_id,
356 + s.status AS "status: super::SubscriptionStatus",
357 + s.current_period_start AS "current_period_start: chrono::DateTime<chrono::Utc>",
358 + s.current_period_end AS "current_period_end: chrono::DateTime<chrono::Utc>",
359 + s.canceled_at AS "canceled_at: chrono::DateTime<chrono::Utc>",
360 + s.created_at AS "created_at: chrono::DateTime<chrono::Utc>",
361 + s.updated_at AS "updated_at: chrono::DateTime<chrono::Utc>",
362 + s.item_id AS "item_id: ItemId",
363 + s.paused_at AS "paused_at: chrono::DateTime<chrono::Utc>"
364 + FROM subscriptions s
272 365 WHERE s.project_id IN (SELECT id FROM projects WHERE user_id = $1)
273 366 AND s.status = 'active'
274 367 AND s.paused_at IS NULL
275 368 "#,
369 + creator_id as UserId,
276 370 )
277 - .bind(creator_id)
278 371 .fetch_all(pool)
279 372 .await?;
280 373
@@ -283,19 +376,16 @@
283 376
284 377 /// Mark all active subscriptions to a creator's projects as paused.
285 378 #[tracing::instrument(skip_all)]
286 - pub async fn pause_subscriptions_for_creator(
287 - pool: &PgPool,
288 - creator_id: UserId,
289 - ) -> Result<u64> {
290 - let result = sqlx::query(
379 + pub async fn pause_subscriptions_for_creator(pool: &PgPool, creator_id: UserId) -> Result<u64> {
380 + let result = sqlx::query!(
291 381 r#"
292 382 UPDATE subscriptions SET paused_at = NOW()
293 383 WHERE project_id IN (SELECT id FROM projects WHERE user_id = $1)
294 384 AND status = 'active'
295 385 AND paused_at IS NULL
296 386 "#,
387 + creator_id as UserId,
297 388 )
298 - .bind(creator_id)
299 389 .execute(pool)
300 390 .await?;
301 391
@@ -308,15 +398,27 @@
308 398 pool: &PgPool,
309 399 creator_id: UserId,
310 400 ) -> Result<Vec<DbSubscription>> {
311 - let subs = sqlx::query_as::<_, DbSubscription>(
401 + let subs = sqlx::query_as!(
402 + DbSubscription,
312 403 r#"
313 - SELECT s.* FROM subscriptions s
404 + SELECT s.id AS "id: SubscriptionId", s.subscriber_id AS "subscriber_id: UserId",
405 + s.tier_id AS "tier_id: SubscriptionTierId", s.project_id AS "project_id: ProjectId",
406 + s.stripe_subscription_id, s.stripe_customer_id,
407 + s.status AS "status: super::SubscriptionStatus",
408 + s.current_period_start AS "current_period_start: chrono::DateTime<chrono::Utc>",
409 + s.current_period_end AS "current_period_end: chrono::DateTime<chrono::Utc>",
410 + s.canceled_at AS "canceled_at: chrono::DateTime<chrono::Utc>",
411 + s.created_at AS "created_at: chrono::DateTime<chrono::Utc>",
412 + s.updated_at AS "updated_at: chrono::DateTime<chrono::Utc>",
413 + s.item_id AS "item_id: ItemId",
414 + s.paused_at AS "paused_at: chrono::DateTime<chrono::Utc>"
415 + FROM subscriptions s
314 416 WHERE s.project_id IN (SELECT id FROM projects WHERE user_id = $1)
315 417 AND s.status = 'active'
316 418 AND s.paused_at IS NOT NULL
317 419 "#,
420 + creator_id as UserId,
318 421 )
319 - .bind(creator_id)
320 422 .fetch_all(pool)
321 423 .await?;
322 424
@@ -329,16 +431,27 @@
329 431 pool: &PgPool,
330 432 creator_id: UserId,
331 433 ) -> Result<Vec<DbSubscription>> {
332 - let subs = sqlx::query_as::<_, DbSubscription>(
434 + let subs = sqlx::query_as!(
435 + DbSubscription,
333 436 r#"
334 437 UPDATE subscriptions SET paused_at = NULL
335 438 WHERE project_id IN (SELECT id FROM projects WHERE user_id = $1)
336 439 AND status = 'active'
337 440 AND paused_at IS NOT NULL
338 - RETURNING *
441 + RETURNING id AS "id: SubscriptionId", subscriber_id AS "subscriber_id: UserId",
442 + tier_id AS "tier_id: SubscriptionTierId", project_id AS "project_id: ProjectId",
443 + stripe_subscription_id, stripe_customer_id,
444 + status AS "status: super::SubscriptionStatus",
445 + current_period_start AS "current_period_start: chrono::DateTime<chrono::Utc>",
446 + current_period_end AS "current_period_end: chrono::DateTime<chrono::Utc>",
447 + canceled_at AS "canceled_at: chrono::DateTime<chrono::Utc>",
448 + created_at AS "created_at: chrono::DateTime<chrono::Utc>",
449 + updated_at AS "updated_at: chrono::DateTime<chrono::Utc>",
450 + item_id AS "item_id: ItemId",
451 + paused_at AS "paused_at: chrono::DateTime<chrono::Utc>"
339 452 "#,
453 + creator_id as UserId,
340 454 )
341 - .bind(creator_id)
342 455 .fetch_all(pool)
343 456 .await?;
344 457
@@ -408,6 +521,9 @@
408 521 user_id: UserId,
409 522 scope: SubscriptionScope,
410 523 ) -> Result<Option<SubscriptionGate>> {
524 + // runtime-checked: dynamically-built SQL — the access predicate is
525 + // interpolated from the sealed `PREDICATE` const via format!, so the
526 + // statement text isn't a compile-time literal the macro can verify.
411 527 let exists: bool = match scope {
412 528 SubscriptionScope::Project(project_id) => {
413 529 sqlx::query_scalar(&format!(
@@ -441,6 +557,9 @@
441 557 /// batch path cannot drift from the single-item gate.
442 558 #[tracing::instrument(skip_all)]
443 559 pub async fn accessible_item_ids(pool: &PgPool, user_id: UserId) -> Result<Vec<ItemId>> {
560 + // runtime-checked: dynamically-built SQL — the access predicate is
561 + // interpolated from the sealed `PREDICATE` const via format!, so the
562 + // statement text isn't a compile-time literal the macro can verify.
444 563 let item_ids: Vec<ItemId> = sqlx::query_scalar(&format!(
445 564 "SELECT DISTINCT item_id FROM subscriptions \
446 565 WHERE subscriber_id = $1 AND item_id IS NOT NULL AND {}",
@@ -463,7 +582,10 @@
463 582 user_id: UserId,
464 583 ) -> Result<HashMap<ItemId, SubscriptionGate>> {
465 584 let ids = Self::accessible_item_ids(pool, user_id).await?;
466 - Ok(ids.into_iter().map(|id| (id, SubscriptionGate(()))).collect())
585 + Ok(ids
586 + .into_iter()
587 + .map(|id| (id, SubscriptionGate(())))
588 + .collect())
467 589 }
468 590
469 591 /// Test-only constructor. Real gates can only be minted by running the
@@ -483,12 +605,10 @@
483 605 /// taking the [`SubscriptionGate`] witness directly where a proof of access is
484 606 /// useful downstream.
485 607 #[tracing::instrument(skip_all)]
486 - pub async fn has_access(
487 - pool: &PgPool,
488 - user_id: UserId,
489 - scope: SubscriptionScope,
490 - ) -> Result<bool> {
491 - Ok(SubscriptionGate::check(pool, user_id, scope).await?.is_some())
608 + pub async fn has_access(pool: &PgPool, user_id: UserId, scope: SubscriptionScope) -> Result<bool> {
609 + Ok(SubscriptionGate::check(pool, user_id, scope)
610 + .await?
611 + .is_some())
492 612 }
493 613
494 614 /// Get user subscriptions joined with project and tier data (for library display).
@@ -497,18 +617,23 @@
497 617 pool: &PgPool,
498 618 user_id: UserId,
499 619 ) -> Result<Vec<DbUserSubscriptionRow>> {
500 - let rows = sqlx::query_as::<_, DbUserSubscriptionRow>(
501 - "SELECT s.id, s.project_id, p.title AS project_title, p.slug AS project_slug,
502 - t.name AS tier_name, t.price_cents, s.status,
503 - s.current_period_end, s.stripe_subscription_id
504 - FROM subscriptions s
505 - JOIN projects p ON p.id = s.project_id
506 - JOIN subscription_tiers t ON t.id = s.tier_id
507 - WHERE s.subscriber_id = $1
508 - ORDER BY s.created_at DESC
509 - LIMIT 1000",
620 + let rows = sqlx::query_as!(
621 + DbUserSubscriptionRow,
622 + r#"
623 + SELECT s.id AS "id: SubscriptionId", s.project_id AS "project_id!: ProjectId",
624 + p.title AS project_title, p.slug AS "project_slug: super::Slug",
625 + t.name AS tier_name, t.price_cents, s.status AS "status: super::SubscriptionStatus",
626 + s.current_period_end AS "current_period_end: chrono::DateTime<chrono::Utc>",
627 + s.stripe_subscription_id
628 + FROM subscriptions s
629 + JOIN projects p ON p.id = s.project_id
630 + JOIN subscription_tiers t ON t.id = s.tier_id
631 + WHERE s.subscriber_id = $1
632 + ORDER BY s.created_at DESC
633 + LIMIT 1000
634 + "#,
635 + user_id as UserId,
510 636 )
511 - .bind(user_id)
512 637 .fetch_all(pool)
513 638 .await?;
514 639
@@ -522,14 +647,11 @@
522 647 /// its grace window is still a subscriber). Do not "align" it with
523 648 /// [`GRANTS_ACCESS_PREDICATE`]; the divergence here is intentional.
524 649 #[tracing::instrument(skip_all)]
525 - pub async fn get_project_subscriber_count(
526 - pool: &PgPool,
527 - project_id: ProjectId,
528 - ) -> Result<i64> {
529 - let count: i64 = sqlx::query_scalar(
530 - "SELECT COUNT(*) FROM subscriptions WHERE project_id = $1 AND status = 'active' AND paused_at IS NULL",
650 + pub async fn get_project_subscriber_count(pool: &PgPool, project_id: ProjectId) -> Result<i64> {
651 + let count: i64 = sqlx::query_scalar!(
Lines truncated
@@ -46,25 +46,34 @@
46 46 executor: impl sqlx::PgExecutor<'e>,
47 47 params: &CreateTransactionParams<'_>,
48 48 ) -> Result<DbTransaction> {
49 - let tx = sqlx::query_as::<_, DbTransaction>(
49 + let tx = sqlx::query_as!(
50 + DbTransaction,
50 51 r#"
51 52 INSERT INTO transactions (buyer_id, seller_id, item_id, amount_cents, platform_fee_cents, stripe_checkout_session_id, item_title, seller_username, share_contact, project_id, promo_code_id, guest_email)
52 53 VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)
53 - RETURNING *
54 + RETURNING
55 + id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
56 + item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
57 + currency, status AS "status: super::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
58 + created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
59 + item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
60 + parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
61 + guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
62 + download_token AS "download_token: DownloadToken"
54 63 "#,
64 + params.buyer_id as Option<UserId>,
65 + params.seller_id as UserId,
66 + params.item_id as Option<ItemId>,
67 + params.amount_cents as Cents,
68 + params.platform_fee_cents as Cents,
69 + params.stripe_checkout_session_id,
70 + params.item_title,
71 + params.seller_username,
72 + params.share_contact,
73 + params.project_id as Option<ProjectId>,
74 + params.promo_code_id as Option<PromoCodeId>,
75 + params.guest_email,
55 76 )
56 - .bind(params.buyer_id)
57 - .bind(params.seller_id)
58 - .bind(params.item_id)
59 - .bind(params.amount_cents)
60 - .bind(params.platform_fee_cents)
61 - .bind(params.stripe_checkout_session_id)
62 - .bind(params.item_title)
63 - .bind(params.seller_username)
64 - .bind(params.share_contact)
65 - .bind(params.project_id)
66 - .bind(params.promo_code_id)
67 - .bind(params.guest_email)
68 77 .fetch_one(executor)
69 78 .await?;
70 79
@@ -87,7 +96,8 @@
87 96 ) -> Result<Option<DbTransaction>> {
88 97 let claim_token = ClaimToken::new();
89 98
90 - let tx = sqlx::query_as::<_, DbTransaction>(
99 + let tx = sqlx::query_as!(
100 + DbTransaction,
91 101 r#"
92 102 UPDATE transactions
93 103 SET status = 'completed',
@@ -98,13 +108,21 @@
98 108 buyer_id = NULL
99 109 WHERE stripe_checkout_session_id = $1
100 110 AND status = 'pending'
101 - RETURNING *
111 + RETURNING
112 + id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
113 + item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
114 + currency, status AS "status: super::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
115 + created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
116 + item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
117 + parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
118 + guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
119 + download_token AS "download_token: DownloadToken"
102 120 "#,
121 + stripe_checkout_session_id,
122 + stripe_payment_intent_id,
123 + guest_email,
124 + claim_token as ClaimToken,
103 125 )
104 - .bind(stripe_checkout_session_id)
105 - .bind(stripe_payment_intent_id)
106 - .bind(guest_email)
107 - .bind(claim_token)
108 126 .fetch_optional(executor)
109 127 .await?;
110 128
@@ -119,7 +137,7 @@
119 137 email: &str,
120 138 user_id: UserId,
121 139 ) -> Result<u64> {
122 - let result = sqlx::query(
140 + let result = sqlx::query!(
123 141 r#"
124 142 UPDATE transactions
125 143 SET buyer_id = $1, claimed_by = $1, claim_token = NULL
@@ -127,9 +145,9 @@
127 145 AND buyer_id IS NULL
128 146 AND status = 'completed'
129 147 "#,
148 + user_id as UserId,
149 + email,
130 150 )
131 - .bind(user_id)
132 - .bind(email)
133 151 .execute(pool)
134 152 .await?;
135 153
@@ -143,18 +161,27 @@
143 161 claim_token: ClaimToken,
144 162 user_id: UserId,
145 163 ) -> Result<Option<DbTransaction>> {
146 - let tx = sqlx::query_as::<_, DbTransaction>(
164 + let tx = sqlx::query_as!(
165 + DbTransaction,
147 166 r#"
148 167 UPDATE transactions
149 168 SET buyer_id = $2, claimed_by = $2, claim_token = NULL
150 169 WHERE claim_token = $1
151 170 AND buyer_id IS NULL
152 171 AND status = 'completed'
153 - RETURNING *
172 + RETURNING
173 + id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
174 + item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
175 + currency, status AS "status: super::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
176 + created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
177 + item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
178 + parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
179 + guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
180 + download_token AS "download_token: DownloadToken"
154 181 "#,
182 + claim_token as ClaimToken,
183 + user_id as UserId,
155 184 )
156 - .bind(claim_token)
157 - .bind(user_id)
158 185 .fetch_optional(pool)
159 186 .await?;
160 187
@@ -167,10 +194,22 @@
167 194 pool: &PgPool,
168 195 download_token: DownloadToken,
169 196 ) -> Result<Option<DbTransaction>> {
170 - let tx = sqlx::query_as::<_, DbTransaction>(
171 - "SELECT * FROM transactions WHERE download_token = $1 AND status = 'completed'",
197 + let tx = sqlx::query_as!(
198 + DbTransaction,
199 + r#"
200 + SELECT
201 + id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
202 + item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
203 + currency, status AS "status: super::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
204 + created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
205 + item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
206 + parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
207 + guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
208 + download_token AS "download_token: DownloadToken"
209 + FROM transactions WHERE download_token = $1 AND status = 'completed'
210 + "#,
211 + download_token as DownloadToken,
172 212 )
173 - .bind(download_token)
174 213 .fetch_optional(pool)
175 214 .await?;
176 215
@@ -189,7 +228,8 @@
189 228 ) -> Result<Option<DbTransaction>> {
190 229 // Only update if status is 'pending' for idempotency
191 230 // Returns None if transaction was already completed (duplicate webhook)
192 - let tx = sqlx::query_as::<_, DbTransaction>(
231 + let tx = sqlx::query_as!(
232 + DbTransaction,
193 233 r#"
194 234 UPDATE transactions
195 235 SET status = 'completed',
@@ -197,11 +237,19 @@
197 237 completed_at = NOW()
198 238 WHERE stripe_checkout_session_id = $1
199 239 AND status = 'pending'
200 - RETURNING *
240 + RETURNING
241 + id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
242 + item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
243 + currency, status AS "status: super::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
244 + created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
245 + item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
246 + parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
247 + guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
248 + download_token AS "download_token: DownloadToken"
201 249 "#,
250 + stripe_checkout_session_id,
251 + stripe_payment_intent_id,
202 252 )
203 - .bind(stripe_checkout_session_id)
204 - .bind(stripe_payment_intent_id)
205 253 .fetch_optional(executor)
206 254 .await?;
207 255
@@ -216,7 +264,8 @@
216 264 stripe_checkout_session_id: &str,
217 265 stripe_payment_intent_id: &str,
218 266 ) -> Result<Vec<DbTransaction>> {
219 - let txs = sqlx::query_as::<_, DbTransaction>(
267 + let txs = sqlx::query_as!(
268 + DbTransaction,
220 269 r#"
221 270 UPDATE transactions
222 271 SET status = 'completed',
@@ -224,11 +273,19 @@
224 273 completed_at = NOW()
225 274 WHERE stripe_checkout_session_id = $1
226 275 AND status = 'pending'
227 - RETURNING *
276 + RETURNING
277 + id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
278 + item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
279 + currency, status AS "status: super::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
280 + created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
281 + item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
282 + parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
283 + guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
284 + download_token AS "download_token: DownloadToken"
228 285 "#,
286 + stripe_checkout_session_id,
287 + stripe_payment_intent_id,
229 288 )
230 - .bind(stripe_checkout_session_id)
231 - .bind(stripe_payment_intent_id)
232 289 .fetch_all(executor)
233 290 .await?;
234 291
@@ -244,11 +301,23 @@
244 301 buyer_id: UserId,
245 302 limit: Option<i64>,
246 303 ) -> Result<Vec<DbTransaction>> {
247 - let txs = sqlx::query_as::<_, DbTransaction>(
248 - "SELECT * FROM transactions WHERE buyer_id = $1 ORDER BY created_at DESC LIMIT $2",
304 + let txs = sqlx::query_as!(
305 + DbTransaction,
306 + r#"
307 + SELECT
308 + id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
309 + item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
310 + currency, status AS "status: super::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
311 + created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
312 + item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
313 + parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
314 + guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
315 + download_token AS "download_token: DownloadToken"
316 + FROM transactions WHERE buyer_id = $1 ORDER BY created_at DESC LIMIT $2
317 + "#,
318 + buyer_id as UserId,
319 + limit,
249 320 )
250 - .bind(buyer_id)
251 - .bind(limit)
252 321 .fetch_all(pool)
253 322 .await?;
254 323
@@ -264,11 +333,23 @@
264 333 seller_id: UserId,
265 334 limit: Option<i64>,
266 335 ) -> Result<Vec<DbTransaction>> {
267 - let txs = sqlx::query_as::<_, DbTransaction>(
268 - "SELECT * FROM transactions WHERE seller_id = $1 ORDER BY created_at DESC LIMIT $2",
336 + let txs = sqlx::query_as!(
337 + DbTransaction,
338 + r#"
339 + SELECT
340 + id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
341 + item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
342 + currency, status AS "status: super::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
343 + created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
344 + item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
345 + parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
346 + guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
347 + download_token AS "download_token: DownloadToken"
348 + FROM transactions WHERE seller_id = $1 ORDER BY created_at DESC LIMIT $2
349 + "#,
350 + seller_id as UserId,
351 + limit,
269 352 )
270 - .bind(seller_id)
271 - .bind(limit)
272 353 .fetch_all(pool)
273 354 .await?;
274 355
@@ -278,11 +359,11 @@
278 359 /// Check whether a user has a completed purchase for a given item.
279 360 #[tracing::instrument(skip_all)]
280 361 pub async fn has_purchased_item(pool: &PgPool, user_id: UserId, item_id: ItemId) -> Result<bool> {
281 - let count: i64 = sqlx::query_scalar(
282 - "SELECT COUNT(*) FROM transactions WHERE buyer_id = $1 AND item_id = $2 AND status = 'completed'",
362 + let count: i64 = sqlx::query_scalar!(
363 + r#"SELECT COUNT(*) AS "count!" FROM transactions WHERE buyer_id = $1 AND item_id = $2 AND status = 'completed'"#,
364 + user_id as UserId,
365 + item_id as ItemId,
283 366 )
284 - .bind(user_id)
285 - .bind(item_id)
286 367 .fetch_one(pool)
287 368 .await?;
288 369
@@ -300,24 +381,24 @@
300 381 if item_ids.is_empty() {
301 382 return Ok(std::collections::HashSet::new());
302 383 }
303 - let rows: Vec<(ItemId,)> = sqlx::query_as(
304 - "SELECT DISTINCT item_id FROM transactions
305 - WHERE buyer_id = $1 AND status = 'completed' AND item_id = ANY($2)",
384 + let rows = sqlx::query_scalar!(
385 + r#"SELECT DISTINCT item_id AS "item_id!: ItemId" FROM transactions
386 + WHERE buyer_id = $1 AND status = 'completed' AND item_id = ANY($2)"#,
387 + user_id as UserId,
388 + item_ids as &[ItemId],
306 389 )
307 - .bind(user_id)
308 - .bind(item_ids)
309 390 .fetch_all(pool)
310 391 .await?;
311 - Ok(rows.into_iter().map(|(id,)| id).collect())
392 + Ok(rows.into_iter().collect())
312 393 }
313 394
314 395 /// Get all item IDs that a user has purchased (for batch access checks)
315 396 #[tracing::instrument(skip_all)]
316 397 pub async fn get_user_purchased_item_ids(pool: &PgPool, user_id: UserId) -> Result<Vec<ItemId>> {
317 - let item_ids: Vec<ItemId> = sqlx::query_scalar(
318 - "SELECT DISTINCT item_id FROM transactions WHERE buyer_id = $1 AND status = 'completed'",
398 + let item_ids: Vec<ItemId> = sqlx::query_scalar!(
399 + r#"SELECT DISTINCT item_id AS "item_id!: ItemId" FROM transactions WHERE buyer_id = $1 AND status = 'completed' AND item_id IS NOT NULL"#,
400 + user_id as UserId,
319 401 )
320 - .bind(user_id)
321 402 .fetch_all(pool)
322 403 .await?;
323 404
@@ -336,21 +417,21 @@
336 417 params: &ClaimParams<'_>,
337 418 ) -> Result<bool> {
338 419 let claim_id = format!("free-claim-{}-{}", params.buyer_id, params.item_id);
339 - let result = sqlx::query(
420 + let result = sqlx::query!(
340 421 r#"
341 422 INSERT INTO transactions (buyer_id, seller_id, item_id, amount_cents, platform_fee_cents, stripe_checkout_session_id, status, completed_at, item_title, seller_username, share_contact, parent_transaction_id)
342 423 VALUES ($1, $2, $3, 0, 0, $4, 'completed', NOW(), $5, $6, $7, $8)
343 424 ON CONFLICT (buyer_id, item_id) WHERE status = 'completed' AND item_id IS NOT NULL DO NOTHING
344 425 "#,
426 + params.buyer_id as UserId,
427 + params.seller_id as UserId,
428 + params.item_id as ItemId,
429 + claim_id,
430 + params.item_title,
431 + params.seller_username,
432 + params.share_contact,
433 + params.parent_transaction_id as Option<TransactionId>,
345 434 )
346 - .bind(params.buyer_id)
347 - .bind(params.seller_id)
348 - .bind(params.item_id)
349 - .bind(&claim_id)
350 - .bind(params.item_title)
351 - .bind(params.seller_username)
352 - .bind(params.share_contact)
353 - .bind(params.parent_transaction_id)
354 435 .execute(executor)
355 436 .await?;
356 437
@@ -386,21 +467,21 @@
386 467
387 468 // Step 1: Attempt to claim the item first
388 469 let claim_id = format!("free-claim-{}-{}", params.buyer_id, params.item_id);
389 - let result = sqlx::query(
470 + let result = sqlx::query!(
390 471 r#"
391 472 INSERT INTO transactions (buyer_id, seller_id, item_id, amount_cents, platform_fee_cents, stripe_checkout_session_id, status, completed_at, item_title, seller_username, share_contact, promo_code_id)
392 473 VALUES ($1, $2, $3, 0, 0, $4, 'completed', NOW(), $5, $6, $7, $8)
393 474 ON CONFLICT (buyer_id, item_id) WHERE status = 'completed' AND item_id IS NOT NULL DO NOTHING
394 475 "#,
476 + params.buyer_id as UserId,
477 + params.seller_id as UserId,
478 + params.item_id as ItemId,
479 + claim_id,
480 + params.item_title,
481 + params.seller_username,
482 + params.share_contact,
483 + promo_code_id as PromoCodeId,
395 484 )
396 - .bind(params.buyer_id)
397 - .bind(params.seller_id)
398 - .bind(params.item_id)
399 - .bind(&claim_id)
400 - .bind(params.item_title)
401 - .bind(params.seller_username)
402 - .bind(params.share_contact)
403 - .bind(promo_code_id)
404 485 .execute(&mut *tx)
405 486 .await?;
406 487
@@ -413,10 +494,10 @@
413 494 }
414 495
415 496 // Step 3: Increment the promo code use count
416 - let code_result = sqlx::query(
497 + let code_result = sqlx::query!(
417 498 "UPDATE promo_codes SET use_count = use_count + 1 WHERE id = $1 AND (max_uses IS NULL OR use_count < max_uses)",
499 + promo_code_id as PromoCodeId,
418 500 )
419 - .bind(promo_code_id)
420 501 .execute(&mut *tx)
421 502 .await?;
422 503
@@ -433,16 +514,16 @@
433 514 // flip headroom, so an actual collision is vanishingly rare, but the
434 515 // alternative is surfacing a 500 to a buyer mid-claim — cheap to handle.
435 516 if let Some(lk) = license_key_params {
436 - let attempt = sqlx::query(
517 + let attempt = sqlx::query!(
437 518 r#"
438 519 INSERT INTO license_keys (item_id, owner_id, transaction_id, key_code, max_activations)
439 520 VALUES ($1, $2, NULL, $3, $4)
440 521 "#,
522 + params.item_id as ItemId,
523 + params.buyer_id as UserId,
524 + lk.key_code as &KeyCode,
525 + lk.max_activations,
441 526 )
442 - .bind(params.item_id)
443 - .bind(params.buyer_id)
444 - .bind(lk.key_code)
445 - .bind(lk.max_activations)
446 527 .execute(&mut *tx)
447 528 .await;
448 529
@@ -451,16 +532,16 @@
451 532 {
452 533 let retry_code = crate::helpers::generate_key_code();
453 534 tracing::warn!(item_id = %params.item_id, "license key 23505 collision; retrying once");
454 - sqlx::query(
535 + sqlx::query!(
455 536 r#"
456 537 INSERT INTO license_keys (item_id, owner_id, transaction_id, key_code, max_activations)
457 538 VALUES ($1, $2, NULL, $3, $4)
458 539 "#,
540 + params.item_id as ItemId,
541 + params.buyer_id as UserId,
542 + retry_code as KeyCode,
543 + lk.max_activations,
459 544 )
460 - .bind(params.item_id)
461 - .bind(params.buyer_id)
462 - .bind(&retry_code)
463 - .bind(lk.max_activations)
464 545 .execute(&mut *tx)
465 546 .await?;
466 547 } else {
@@ -477,11 +558,11 @@
477 558 /// Check whether a user has a completed purchase for a given project.
478 559 #[tracing::instrument(skip_all)]
479 560 pub async fn has_purchased_project(pool: &PgPool, user_id: UserId, project_id: ProjectId) -> Result<bool> {
480 - let count: i64 = sqlx::query_scalar(
481 - "SELECT COUNT(*) FROM transactions WHERE buyer_id = $1 AND project_id = $2 AND status = 'completed'",
561 + let count: i64 = sqlx::query_scalar!(
562 + r#"SELECT COUNT(*) AS "count!" FROM transactions WHERE buyer_id = $1 AND project_id = $2 AND status = 'completed'"#,
563 + user_id as UserId,
564 + project_id as ProjectId,
482 565 )
483 - .bind(user_id)
484 - .bind(project_id)
485 566 .fetch_one(pool)
486 567 .await?;
487 568
@@ -506,21 +587,30 @@
506 587 pool: &PgPool,
507 588 params: &CreateProjectTransactionParams<'_>,
508 589 ) -> Result<DbTransaction> {
509 - let tx = sqlx::query_as::<_, DbTransaction>(
590 + let tx = sqlx::query_as!(
591 + DbTransaction,
510 592 r#"
511 593 INSERT INTO transactions (buyer_id, seller_id, project_id, amount_cents, platform_fee_cents, stripe_checkout_session_id, item_title, seller_username, share_contact)
512 594 VALUES ($1, $2, $3, $4, 0, $5, $6, $7, $8)
513 - RETURNING *
595 + RETURNING
596 + id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
597 + item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
598 + currency, status AS "status: super::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
599 + created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
600 + item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
601 + parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
602 + guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
603 + download_token AS "download_token: DownloadToken"
514 604 "#,
605 + params.buyer_id as UserId,
606 + params.seller_id as UserId,
607 + params.project_id as ProjectId,
608 + params.amount_cents,
609 + params.stripe_checkout_session_id,
610 + params.project_title,
611 + params.seller_username,
612 + params.share_contact,
515 613 )
516 - .bind(params.buyer_id)
517 - .bind(params.seller_id)
518 - .bind(params.project_id)
519 - .bind(params.amount_cents)
520 - .bind(params.stripe_checkout_session_id)
521 - .bind(params.project_title)
522 - .bind(params.seller_username)
523 - .bind(params.share_contact)
524 614 .fetch_one(pool)
525 615 .await?;
526 616
@@ -536,9 +626,20 @@
536 626 /// without a separate lookup. Capped at 20 rows for the dashboard summary.
537 627 #[tracing::instrument(skip_all)]
538 628 pub async fn get_user_purchases(pool: &PgPool, user_id: UserId) -> Result<Vec<DbPurchaseRow>> {
539 - let purchases = sqlx::query_as::<_, DbPurchaseRow>(
629 + let purchases = sqlx::query_as!(
630 + DbPurchaseRow,
540 631 r#"
541 - SELECT * FROM (
632 + SELECT
Lines truncated
@@ -26,27 +26,33 @@
26 26 let mut tx = pool.begin().await?;
27 27
28 28 // Unset current on older version numbers (versions with the same number stay current)
29 - sqlx::query("UPDATE versions SET is_current = false WHERE item_id = $1 AND version_number != $2")
30 - .bind(item_id)
31 - .bind(version_number)
32 - .execute(&mut *tx)
33 - .await?;
29 + sqlx::query!(
30 + "UPDATE versions SET is_current = false WHERE item_id = $1 AND version_number != $2",
31 + item_id as ItemId,
32 + version_number,
33 + )
34 + .execute(&mut *tx)
35 + .await?;
34 36
35 37 // Create new version as current
36 - let version = sqlx::query_as::<_, DbVersion>(
38 + let version = sqlx::query_as!(
39 + DbVersion,
37 40 r#"
38 41 INSERT INTO versions (item_id, version_number, changelog, file_url, file_size_bytes, file_name, is_current, label)
39 42 VALUES ($1, $2, $3, $4, $5, $6, true, $7)
40 - RETURNING *
43 + RETURNING id AS "id: VersionId", item_id AS "item_id: ItemId", version_number, changelog,
44 + file_url, file_size_bytes, file_name, download_count, is_current,
45 + created_at AS "created_at: chrono::DateTime<chrono::Utc>", s3_key,
46 + scan_status AS "scan_status: super::FileScanStatus", label
41 47 "#,
48 + item_id as ItemId,
49 + version_number,
50 + changelog,
51 + file_url,
52 + file_size_bytes,
53 + file_name,
54 + label,
42 55 )
43 - .bind(item_id)
44 - .bind(version_number)
45 - .bind(changelog)
46 - .bind(file_url)
47 - .bind(file_size_bytes)
48 - .bind(file_name)
49 - .bind(label)
50 56 .fetch_one(&mut *tx)
51 57 .await?;
52 58
@@ -67,11 +73,18 @@
67 73 /// logged at WARN so we notice before a real user gets silently truncated.
68 74 #[tracing::instrument(skip_all)]
69 75 pub async fn get_versions_by_item(pool: &PgPool, item_id: ItemId) -> Result<Vec<DbVersion>> {
70 - let versions = sqlx::query_as::<_, DbVersion>(
71 - "SELECT * FROM versions WHERE item_id = $1 ORDER BY created_at DESC LIMIT $2",
76 + let versions = sqlx::query_as!(
77 + DbVersion,
78 + r#"
79 + SELECT id AS "id: VersionId", item_id AS "item_id: ItemId", version_number, changelog,
80 + file_url, file_size_bytes, file_name, download_count, is_current,
81 + created_at AS "created_at: chrono::DateTime<chrono::Utc>", s3_key,
82 + scan_status AS "scan_status: super::FileScanStatus", label
83 + FROM versions WHERE item_id = $1 ORDER BY created_at DESC LIMIT $2
84 + "#,
85 + item_id as ItemId,
86 + VERSIONS_LIST_HARD_CAP,
72 87 )
73 - .bind(item_id)
74 - .bind(VERSIONS_LIST_HARD_CAP)
75 88 .fetch_all(pool)
76 89 .await?;
77 90
@@ -91,10 +104,17 @@
91 104 pool: &PgPool,
92 105 item_ids: &[ItemId],
93 106 ) -> Result<std::collections::HashMap<ItemId, Vec<DbVersion>>> {
94 - let versions = sqlx::query_as::<_, DbVersion>(
95 - "SELECT * FROM versions WHERE item_id = ANY($1) ORDER BY item_id, created_at DESC",
107 + let versions = sqlx::query_as!(
108 + DbVersion,
109 + r#"
110 + SELECT id AS "id: VersionId", item_id AS "item_id: ItemId", version_number, changelog,
111 + file_url, file_size_bytes, file_name, download_count, is_current,
112 + created_at AS "created_at: chrono::DateTime<chrono::Utc>", s3_key,
113 + scan_status AS "scan_status: super::FileScanStatus", label
114 + FROM versions WHERE item_id = ANY($1) ORDER BY item_id, created_at DESC
115 + "#,
116 + item_ids as &[ItemId],
96 117 )
97 - .bind(item_ids)
98 118 .fetch_all(pool)
99 119 .await?;
100 120
@@ -108,10 +128,12 @@
108 128 /// Atomically increment the download counter for a version.
109 129 #[tracing::instrument(skip_all)]
110 130 pub async fn increment_download_count(pool: &PgPool, version_id: VersionId) -> Result<()> {
111 - sqlx::query("UPDATE versions SET download_count = download_count + 1 WHERE id = $1")
112 - .bind(version_id)
113 - .execute(pool)
114 - .await?;
131 + sqlx::query!(
132 + "UPDATE versions SET download_count = download_count + 1 WHERE id = $1",
133 + version_id as VersionId,
134 + )
135 + .execute(pool)
136 + .await?;
115 137
116 138 Ok(())
117 139 }
@@ -129,16 +151,16 @@
129 151 // a target would silently swallow conflicts on ANY future constraint
130 152 // we add (a unique index on downloaded_at, say). Naming the target
131 153 // means a new constraint surfaces as an error rather than a no-op.
132 - sqlx::query(
154 + sqlx::query!(
133 155 r#"
134 156 INSERT INTO user_downloads (user_id, item_id, version_id)
135 157 VALUES ($1, $2, $3)
136 158 ON CONFLICT (user_id, item_id, version_id) DO NOTHING
137 159 "#,
160 + user_id as UserId,
161 + item_id as ItemId,
162 + version_id as VersionId,
138 163 )
139 - .bind(user_id)
140 - .bind(item_id)
141 - .bind(version_id)
142 164 .execute(pool)
143 165 .await?;
144 166
@@ -152,30 +174,39 @@
152 174 user_id: UserId,
153 175 item_id: ItemId,
154 176 ) -> Result<Option<VersionId>> {
155 - let row: Option<(VersionId,)> = sqlx::query_as(
177 + let row = sqlx::query_scalar!(
156 178 r#"
157 - SELECT ud.version_id FROM user_downloads ud
179 + SELECT ud.version_id AS "version_id: VersionId" FROM user_downloads ud
158 180 JOIN versions v ON v.id = ud.version_id
159 181 WHERE ud.user_id = $1 AND ud.item_id = $2
160 182 ORDER BY v.created_at DESC
161 183 LIMIT 1
162 184 "#,
185 + user_id as UserId,
186 + item_id as ItemId,
163 187 )
164 - .bind(user_id)
165 - .bind(item_id)
166 188 .fetch_optional(pool)
167 189 .await?;
168 190
169 - Ok(row.map(|(id,)| id))
191 + Ok(row)
170 192 }
171 193
172 194 /// Fetch a version by primary key. Returns `None` if not found.
173 195 #[tracing::instrument(skip_all)]
174 196 pub async fn get_version_by_id(pool: &PgPool, version_id: VersionId) -> Result<Option<DbVersion>> {
175 - let version = sqlx::query_as::<_, DbVersion>("SELECT * FROM versions WHERE id = $1")
176 - .bind(version_id)
177 - .fetch_optional(pool)
178 - .await?;
197 + let version = sqlx::query_as!(
198 + DbVersion,
199 + r#"
200 + SELECT id AS "id: VersionId", item_id AS "item_id: ItemId", version_number, changelog,
201 + file_url, file_size_bytes, file_name, download_count, is_current,
202 + created_at AS "created_at: chrono::DateTime<chrono::Utc>", s3_key,
203 + scan_status AS "scan_status: super::FileScanStatus", label
204 + FROM versions WHERE id = $1
205 + "#,
206 + version_id as VersionId,
207 + )
208 + .fetch_optional(pool)
209 + .await?;
179 210
180 211 Ok(version)
181 212 }
@@ -189,10 +220,11 @@
189 220 pool: &PgPool,
190 221 user_id: super::UserId,
191 222 ) -> Result<Vec<VersionS3KeyRow>> {
192 - let rows = sqlx::query_as::<_, VersionS3KeyRow>(
223 + let rows = sqlx::query_as!(
224 + VersionS3KeyRow,
193 225 r#"
194 - SELECT v.s3_key, v.file_name, v.version_number, i.title AS item_title,
195 - p.id AS project_id, p.slug AS project_slug, v.file_size_bytes
226 + SELECT v.s3_key, v.file_name, v.version_number AS "version_number!", i.title AS "item_title!",
227 + p.id AS "project_id!: super::ProjectId", p.slug AS "project_slug!: super::Slug", v.file_size_bytes
196 228 FROM versions v
197 229 JOIN items i ON v.item_id = i.id
198 230 JOIN projects p ON i.project_id = p.id
@@ -200,9 +232,9 @@
200 232 ORDER BY p.slug, i.sort_order, v.created_at DESC
201 233 LIMIT $2
202 234 "#,
235 + user_id as UserId,
236 + VERSIONS_LIST_HARD_CAP,
203 237 )
204 - .bind(user_id)
205 - .bind(VERSIONS_LIST_HARD_CAP)
206 238 .fetch_all(pool)
207 239 .await?;
208 240
@@ -236,20 +268,24 @@
236 268 file_size_bytes: Option<i64>,
237 269 file_name: Option<&str>,
238 270 ) -> Result<Option<DbVersion>> {
239 - let version = sqlx::query_as::<_, DbVersion>(
271 + let version = sqlx::query_as!(
272 + DbVersion,
240 273 r#"
241 274 UPDATE versions
242 275 SET s3_key = $2, file_size_bytes = $3, file_name = $4
243 276 WHERE id = $1
244 277 AND s3_key IS NOT DISTINCT FROM $5
245 - RETURNING *
278 + RETURNING id AS "id: VersionId", item_id AS "item_id: ItemId", version_number, changelog,
279 + file_url, file_size_bytes, file_name, download_count, is_current,
280 + created_at AS "created_at: chrono::DateTime<chrono::Utc>", s3_key,
281 + scan_status AS "scan_status: super::FileScanStatus", label
246 282 "#,
283 + version_id as VersionId,
284 + s3_key,
285 + file_size_bytes,
286 + file_name,
287 + expected_old_s3_key,
247 288 )
248 - .bind(version_id)
249 - .bind(s3_key)
250 - .bind(file_size_bytes)
251 - .bind(file_name)
252 - .bind(expected_old_s3_key)
253 289 .fetch_optional(executor)
254 290 .await?;
255 291
@@ -270,16 +306,16 @@
270 306 // a version's lifetime; its size + key, by contrast, can change under a
271 307 // concurrent replace-confirm, so those come from the DELETE's RETURNING
272 308 // below — never a pre-tx read (Run #18 Storage B5).
273 - let owner_id: Option<super::UserId> = sqlx::query_scalar(
309 + let owner_id: Option<super::UserId> = sqlx::query_scalar!(
274 310 r#"
275 - SELECT p.user_id
311 + SELECT p.user_id AS "user_id!: super::UserId"
276 312 FROM versions v
277 313 JOIN items i ON v.item_id = i.id
278 314 JOIN projects p ON i.project_id = p.id
279 315 WHERE v.id = $1
280 316 "#,
317 + version_id as VersionId,
281 318 )
282 - .bind(version_id)
283 319 .fetch_optional(pool)
284 320 .await?;
285 321
@@ -291,12 +327,13 @@
291 327 // the OLD key while leaking the new one. RETURNING also gives us the
292 328 // rows-affected discipline for free: a concurrent double-delete finds no row
293 329 // and refunds nothing (Run #12 LOW + Run #18 Storage B5).
294 - let deleted: Option<(Option<String>, Option<i64>)> = sqlx::query_as(
330 + let deleted: Option<(Option<String>, Option<i64>)> = sqlx::query!(
295 331 "DELETE FROM versions WHERE id = $1 RETURNING s3_key, file_size_bytes",
332 + version_id as VersionId,
296 333 )
297 - .bind(version_id)
298 334 .fetch_optional(&mut *tx)
299 - .await?;
335 + .await?
336 + .map(|r| (r.s3_key, r.file_size_bytes));
300 337
301 338 if let Some((s3_key, file_size_bytes)) = deleted {
302 339 if let Some(user_id) = owner_id
@@ -333,10 +370,10 @@
333 370 // sides (>=0 and <=i64::MAX) before casting back to BIGINT — without
334 371 // GREATEST(0, ...), a corrupt-negative row could propagate a negative
335 372 // total that later under-flows storage accounting.
336 - let total: i64 = sqlx::query_scalar(
337 - "SELECT COALESCE(GREATEST(0, LEAST(SUM(file_size_bytes), 9223372036854775807))::BIGINT, 0) FROM versions WHERE item_id = $1 AND file_size_bytes IS NOT NULL",
373 + let total: i64 = sqlx::query_scalar!(
374 + r#"SELECT COALESCE(GREATEST(0, LEAST(SUM(file_size_bytes), 9223372036854775807))::BIGINT, 0) AS "total!" FROM versions WHERE item_id = $1 AND file_size_bytes IS NOT NULL"#,
375 + item_id as ItemId,
338 376 )
339 - .bind(item_id)
340 377 .fetch_one(pool)
341 378 .await?;
342 379
@@ -1,0 +1,16 @@
1 + {
2 + "db_name": "PostgreSQL",
3 + "query": "\n INSERT INTO user_downloads (user_id, item_id, version_id)\n VALUES ($1, $2, $3)\n ON CONFLICT (user_id, item_id, version_id) DO NOTHING\n ",
4 + "describe": {
5 + "columns": [],
6 + "parameters": {
7 + "Left": [
8 + "Uuid",
9 + "Uuid",
10 + "Uuid"
11 + ]
12 + },
13 + "nullable": []
14 + },
15 + "hash": "03e1ede9cf99290dabd6c809568142ff8e2e00ec13014d0b4b3e5eb3f5aa274b"
16 + }
@@ -1,0 +1,15 @@
1 + {
2 + "db_name": "PostgreSQL",
3 + "query": "UPDATE versions SET is_current = false WHERE item_id = $1 AND version_number != $2",
4 + "describe": {
5 + "columns": [],
6 + "parameters": {
7 + "Left": [
8 + "Uuid",
9 + "Text"
10 + ]
11 + },
12 + "nullable": []
13 + },
14 + "hash": "0550dd5416b339101f50fe8d1a02b5bd50c1f8ac5ec919c9d3de0add29d73f92"
15 + }
@@ -1,0 +1,100 @@
1 + {
2 + "db_name": "PostgreSQL",
3 + "query": "\n UPDATE subscriptions SET paused_at = NULL\n WHERE project_id IN (SELECT id FROM projects WHERE user_id = $1)\n AND status = 'active'\n AND paused_at IS NOT NULL\n RETURNING id AS \"id: SubscriptionId\", subscriber_id AS \"subscriber_id: UserId\",\n tier_id AS \"tier_id: SubscriptionTierId\", project_id AS \"project_id: ProjectId\",\n stripe_subscription_id, stripe_customer_id,\n status AS \"status: super::SubscriptionStatus\",\n current_period_start AS \"current_period_start: chrono::DateTime<chrono::Utc>\",\n current_period_end AS \"current_period_end: chrono::DateTime<chrono::Utc>\",\n canceled_at AS \"canceled_at: chrono::DateTime<chrono::Utc>\",\n created_at AS \"created_at: chrono::DateTime<chrono::Utc>\",\n updated_at AS \"updated_at: chrono::DateTime<chrono::Utc>\",\n item_id AS \"item_id: ItemId\",\n paused_at AS \"paused_at: chrono::DateTime<chrono::Utc>\"\n ",
4 + "describe": {
5 + "columns": [
6 + {
7 + "ordinal": 0,
8 + "name": "id: SubscriptionId",
9 + "type_info": "Uuid"
10 + },
11 + {
12 + "ordinal": 1,
13 + "name": "subscriber_id: UserId",
14 + "type_info": "Uuid"
15 + },
16 + {
17 + "ordinal": 2,
18 + "name": "tier_id: SubscriptionTierId",
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": "stripe_subscription_id",
29 + "type_info": "Text"
30 + },
31 + {
32 + "ordinal": 5,
33 + "name": "stripe_customer_id",
34 + "type_info": "Text"
35 + },
36 + {
37 + "ordinal": 6,
38 + "name": "status: super::SubscriptionStatus",
39 + "type_info": "Varchar"
40 + },
41 + {
42 + "ordinal": 7,
43 + "name": "current_period_start: chrono::DateTime<chrono::Utc>",
44 + "type_info": "Timestamptz"
45 + },
46 + {
47 + "ordinal": 8,
48 + "name": "current_period_end: chrono::DateTime<chrono::Utc>",
49 + "type_info": "Timestamptz"
50 + },
51 + {
52 + "ordinal": 9,
53 + "name": "canceled_at: chrono::DateTime<chrono::Utc>",
54 + "type_info": "Timestamptz"
55 + },
56 + {
57 + "ordinal": 10,
58 + "name": "created_at: chrono::DateTime<chrono::Utc>",
59 + "type_info": "Timestamptz"
60 + },
61 + {
62 + "ordinal": 11,
63 + "name": "updated_at: chrono::DateTime<chrono::Utc>",
64 + "type_info": "Timestamptz"
65 + },
66 + {
67 + "ordinal": 12,
68 + "name": "item_id: ItemId",
69 + "type_info": "Uuid"
70 + },
71 + {
72 + "ordinal": 13,
73 + "name": "paused_at: chrono::DateTime<chrono::Utc>",
74 + "type_info": "Timestamptz"
75 + }
76 + ],
77 + "parameters": {
78 + "Left": [
79 + "Uuid"
80 + ]
81 + },
82 + "nullable": [
83 + false,
84 + false,
85 + false,
86 + true,
87 + false,
88 + false,
89 + false,
90 + true,
91 + true,
92 + true,
93 + false,
94 + false,
95 + true,
96 + true
97 + ]
98 + },
99 + "hash": "05fc1e551a7fb89119de0ac0801fdd5219e7e478dd8b63d7665dba25243a4c26"
100 + }
@@ -1,0 +1,14 @@
1 + {
2 + "db_name": "PostgreSQL",
3 + "query": "DELETE FROM promo_codes WHERE creator_id = $1 AND expires_at IS NOT NULL AND expires_at < NOW()",
4 + "describe": {
5 + "columns": [],
6 + "parameters": {
7 + "Left": [
8 + "Uuid"
9 + ]
10 + },
11 + "nullable": []
12 + },
13 + "hash": "074a33db48f3349c0e8dcacb93cf70d6a66fb78866e7af46e988e952406461a6"
14 + }
@@ -1,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 item_id = $1 ORDER BY created_at DESC LIMIT 500\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": "08a3603078a3aee0577f94add6b32ceb5633667a5a7dbcb6476c658dbed32ba7"
70 + }
@@ -1,0 +1,14 @@
1 + {
2 + "db_name": "PostgreSQL",
3 + "query": "\n UPDATE license_keys\n SET revoked_at = NOW()\n WHERE transaction_id = $1 AND revoked_at IS NULL\n ",
4 + "describe": {
5 + "columns": [],
6 + "parameters": {
7 + "Left": [
8 + "Uuid"
9 + ]
10 + },
11 + "nullable": []
12 + },
13 + "hash": "0cc4a78544490a1bb1f21db1f53309f5f30fba02f1ddeafda6b761b62a9c8c6c"
14 + }
@@ -1,0 +1,159 @@
1 + {
2 + "db_name": "PostgreSQL",
3 + "query": "\n INSERT INTO transactions (buyer_id, seller_id, item_id, amount_cents, platform_fee_cents, stripe_checkout_session_id, item_title, seller_username, share_contact, project_id, promo_code_id, guest_email)\n VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)\n RETURNING\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 ",
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 + "Uuid",
120 + "Uuid",
121 + "Uuid",
122 + "Int4",
123 + "Int4",
124 + "Varchar",
125 + "Varchar",
126 + "Varchar",
127 + "Bool",
128 + "Uuid",
129 + "Uuid",
130 + "Varchar"
131 + ]
132 + },
133 + "nullable": [
134 + false,
135 + true,
136 + true,
137 + true,
138 + false,
139 + false,
140 + false,
141 + false,
142 + true,
143 + true,
144 + false,
145 + true,
146 + true,
147 + true,
148 + false,
149 + true,
150 + true,
151 + true,
152 + true,
153 + true,
154 + true,
155 + true
156 + ]
157 + },
158 + "hash": "0e3c392cde613d2447dbbba61bccccf276fb01ea2a550f50afbf540cd5f6ea2c"
159 + }
@@ -1,0 +1,149 @@
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 WHERE seller_id = $1 ORDER BY created_at DESC LIMIT $2\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 + "Uuid",
120 + "Int8"
121 + ]
122 + },
123 + "nullable": [
124 + false,
125 + true,
126 + true,
127 + true,
128 + false,
129 + false,
130 + false,
131 + false,
132 + true,
133 + true,
134 + false,
135 + true,
136 + true,
137 + true,
138 + false,
139 + true,
140 + true,
141 + true,
142 + true,
143 + true,
144 + true,
145 + true
146 + ]
147 + },
148 + "hash": "0edcbc78d86e90f5845d72797f65ed8d1071327610ef8a51dd775b3f709bec66"
149 + }