Skip to main content

max / makenotwork

audit Run 15 Phase 1: payments & data-integrity fixes - Stripe refund creation now uses a deterministic idempotency key (refund-{txid}), mirroring the platform-credit transfer, so a retry can't double-debit the connected balance. - Stripe client built via ClientBuilder with a 30s per-attempt HTTP timeout (was Client::new with no timeout); boot exits code 2 on a build error rather than running with payments silently broken. - Guest download resolves version-delivered files (current/latest Clean version), not just inline audio/video keys; gates on the resolved file's scan status. Fixes paid-but-undownloadable digital-file items. - Import pipeline subscribes in batched UNNEST inserts instead of one query per email (100k-row import: 100k INSERTs -> ~100); removed the now-unused single-email subscribe helper. - Reorder (item_sections, custom_links, collection_items) rewritten as single atomic UNNEST ... WITH ORDINALITY updates instead of per-row UPDATE loops. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-07-01 23:53 UTC
Commit: 00144af7b5c5b4be9b467e7036f5942fa2f4af28
Parent: fca9902
10 files changed, +158 insertions, -69 deletions
M server/src/db/collections.rs +15 -10
@@ -321,17 +321,22 @@ pub async fn reorder_collection_items(
321 321 collection_id: CollectionId,
322 322 item_ids: &[ItemId],
323 323 ) -> Result<()> {
324 + let ids: Vec<uuid::Uuid> = item_ids.iter().map(|id| *id.as_uuid()).collect();
324 325 let mut tx = pool.begin().await?;
325 - for (index, item_id) in item_ids.iter().enumerate() {
326 - sqlx::query(
327 - "UPDATE collection_items SET position = $1 WHERE collection_id = $2 AND item_id = $3",
328 - )
329 - .bind(index as i32)
330 - .bind(collection_id)
331 - .bind(item_id)
332 - .execute(&mut *tx)
333 - .await?;
334 - }
326 + // Single `UNNEST ... WITH ORDINALITY` update instead of one query per item —
327 + // atomic within the same transaction that bumps the collection's timestamp.
328 + sqlx::query(
329 + r#"
330 + UPDATE collection_items AS ci
331 + SET position = ord.pos::int - 1
332 + FROM UNNEST($1::uuid[]) WITH ORDINALITY AS ord(id, pos)
333 + WHERE ci.collection_id = $2 AND ci.item_id = ord.id
334 + "#,
335 + )
336 + .bind(&ids)
337 + .bind(collection_id)
338 + .execute(&mut *tx)
339 + .await?;
335 340
336 341 sqlx::query("UPDATE collections SET updated_at = NOW() WHERE id = $1")
337 342 .bind(collection_id)
M server/src/db/custom_links.rs +17 -11
@@ -92,19 +92,25 @@ pub async fn delete_custom_link(pool: &PgPool, id: CustomLinkId, user_id: UserId
92 92 }
93 93
94 94 /// Reorder a user's custom links by assigning sort_order from the given ID sequence.
95 - /// Wrapped in a transaction so a crash mid-reorder doesn't leave inconsistent state.
95 + ///
96 + /// A single `UNNEST ... WITH ORDINALITY` update — inherently atomic (no partial
97 + /// ordering on failure) and one round-trip instead of one query per link, so the
98 + /// prior explicit transaction is no longer needed.
96 99 #[tracing::instrument(skip_all)]
97 100 pub async fn reorder_custom_links(pool: &PgPool, user_id: UserId, link_ids: &[CustomLinkId]) -> Result<()> {
98 - let mut tx = pool.begin().await?;
99 - for (index, link_id) in link_ids.iter().enumerate() {
100 - sqlx::query("UPDATE custom_links SET sort_order = $1 WHERE id = $2 AND user_id = $3")
101 - .bind(index as i32)
102 - .bind(link_id)
103 - .bind(user_id)
104 - .execute(&mut *tx)
105 - .await?;
106 - }
107 - tx.commit().await?;
101 + let ids: Vec<uuid::Uuid> = link_ids.iter().map(|id| *id.as_uuid()).collect();
102 + sqlx::query(
103 + r#"
104 + UPDATE custom_links AS c
105 + SET sort_order = ord.pos::int - 1
106 + FROM UNNEST($1::uuid[]) WITH ORDINALITY AS ord(id, pos)
107 + WHERE c.id = ord.id AND c.user_id = $2
108 + "#,
109 + )
110 + .bind(&ids)
111 + .bind(user_id)
112 + .execute(pool)
113 + .await?;
108 114
109 115 Ok(())
110 116 }
M server/src/db/item_sections.rs +16 -10
@@ -98,18 +98,24 @@ pub async fn delete(pool: &PgPool, section_id: ItemSectionId) -> Result<()> {
98 98 }
99 99
100 100 /// Reorder sections by setting sort_order from an ordered list of IDs.
101 + ///
102 + /// A single `UNNEST ... WITH ORDINALITY` update: atomic (no partial ordering on
103 + /// failure) and one round-trip instead of one query per section.
101 104 #[tracing::instrument(skip_all)]
102 105 pub async fn reorder(pool: &PgPool, item_id: ItemId, section_ids: &[ItemSectionId]) -> Result<()> {
103 - for (i, id) in section_ids.iter().enumerate() {
104 - sqlx::query(
105 - "UPDATE item_sections SET sort_order = $1, updated_at = now() WHERE id = $2 AND item_id = $3",
106 - )
107 - .bind(i as i32)
108 - .bind(id)
109 - .bind(item_id)
110 - .execute(pool)
111 - .await?;
112 - }
106 + let ids: Vec<uuid::Uuid> = section_ids.iter().map(|id| *id.as_uuid()).collect();
107 + sqlx::query(
108 + r#"
109 + UPDATE item_sections AS s
110 + SET sort_order = ord.pos::int - 1, updated_at = now()
111 + FROM UNNEST($1::uuid[]) WITH ORDINALITY AS ord(id, pos)
112 + WHERE s.id = ord.id AND s.item_id = $2
113 + "#,
114 + )
115 + .bind(&ids)
116 + .bind(item_id)
117 + .execute(pool)
118 + .await?;
113 119
114 120 Ok(())
115 121 }
M server/src/db/mailing_lists.rs +18 -10
@@ -169,28 +169,36 @@ pub async fn create_default_lists(
169 169 Ok(())
170 170 }
171 171
172 - /// Subscribe an email address (without a user account) to a mailing list.
173 - /// Used for importing subscriber lists from external platforms.
174 - /// Returns `true` if a new row was created, `false` if the email already exists.
175 - #[tracing::instrument(skip_all)]
176 - pub async fn subscribe_by_email(
172 + /// Batch-subscribe many emails to a list in a single statement.
173 + ///
174 + /// Used by the import pipeline, where subscribing one email per query is an
175 + /// N+1 storm (a 100k-subscriber import would issue 100k sequential INSERTs on
176 + /// the shared pool). Emails are lowercased and deduplicated within the batch;
177 + /// `ON CONFLICT DO NOTHING` skips ones already subscribed. Returns the number of
178 + /// rows actually inserted (new subscribers).
179 + #[tracing::instrument(skip_all, fields(count = emails.len()))]
180 + pub async fn subscribe_many_by_email(
177 181 pool: &PgPool,
178 182 list_id: MailingListId,
179 - email: &str,
180 - ) -> Result<bool> {
183 + emails: &[String],
184 + ) -> Result<u64> {
185 + if emails.is_empty() {
186 + return Ok(0);
187 + }
188 + let lowered: Vec<String> = emails.iter().map(|e| e.to_lowercase()).collect();
181 189 let result = sqlx::query(
182 190 r#"
183 191 INSERT INTO mailing_list_subscribers (list_id, email)
184 - VALUES ($1, $2)
192 + SELECT $1, sub_email FROM UNNEST($2::text[]) AS sub_email
185 193 ON CONFLICT DO NOTHING
186 194 "#,
187 195 )
188 196 .bind(list_id)
189 - .bind(email.to_lowercase())
197 + .bind(&lowered)
190 198 .execute(pool)
191 199 .await?;
192 200
193 - Ok(result.rows_affected() > 0)
201 + Ok(result.rows_affected())
194 202 }
195 203
196 204 /// Convenience: find the content list for a project and subscribe a user.
M server/src/import/pipeline.rs +26 -12
@@ -1,7 +1,10 @@
1 1 //! Generic import pipeline: takes an `ImportPayload` and creates MNW entities.
2 2 //!
3 - //! Processing happens in chunks of 50 rows with progress updates after each chunk.
4 - //! Individual row failures are logged but don't abort the import.
3 + //! Items are processed in chunks with progress updates after each chunk;
4 + //! subscribers are written in batched `UNNEST` inserts (see `SUBSCRIBER_BATCH_SIZE`)
5 + //! rather than one query per email. Individual item failures are logged but don't
6 + //! abort the import; a hard error (e.g. mailing-list creation) propagates and the
7 + //! caller (`routes/api/imports.rs`) marks the job `failed` via `fail_import_job`.
5 8
6 9 use sqlx::PgPool;
7 10
@@ -12,6 +15,13 @@ use crate::error::Result;
12 15 /// Chunk size for progress updates.
13 16 const CHUNK_SIZE: usize = 50;
14 17
18 + /// Batch size for the subscriber insert. Subscribers are written with a single
19 + /// `UNNEST` INSERT per batch (not one query per email), so this is bounded by
20 + /// how many parameters we want in one statement rather than progress-update
21 + /// cadence — 1,000 keeps the array small while collapsing a 100k-row import
22 + /// from 100k INSERTs to 100.
23 + const SUBSCRIBER_BATCH_SIZE: usize = 1_000;
24 +
15 25 /// Run the full import pipeline for a job.
16 26 ///
17 27 /// Creates MNW entities (tiers, items, tags, mailing list subscribers) from the
@@ -67,18 +77,22 @@ pub async fn run_import(
67 77 )
68 78 .await?;
69 79
70 - for chunk in payload.subscribers.chunks(CHUNK_SIZE) {
71 - for sub in chunk {
72 - match db::mailing_lists::subscribe_by_email(pool, list.id, &sub.email).await {
73 - Ok(true) => created += 1,
74 - Ok(false) => skipped += 1,
75 - Err(e) => {
76 - errors.push(format!("Subscriber '{}': {}", sub.email, e));
77 - skipped += 1;
78 - }
80 + for chunk in payload.subscribers.chunks(SUBSCRIBER_BATCH_SIZE) {
81 + let emails: Vec<String> = chunk.iter().map(|s| s.email.clone()).collect();
82 + match db::mailing_lists::subscribe_many_by_email(pool, list.id, &emails).await {
83 + Ok(inserted) => {
84 + let inserted = inserted as i32;
85 + created += inserted;
86 + skipped += chunk.len() as i32 - inserted;
87 + }
88 + Err(e) => {
89 + // A batch failure is a hard DB error (not a per-row skip); surface
90 + // it and count the whole chunk as skipped rather than N+1-retrying.
91 + errors.push(format!("Subscriber batch ({} rows): {}", chunk.len(), e));
92 + skipped += chunk.len() as i32;
79 93 }
80 - processed += 1;
81 94 }
95 + processed += chunk.len() as i32;
82 96 update_progress(pool, job_id, processed, created, skipped).await;
83 97 }
84 98
M server/src/main.rs +12 -2
@@ -202,8 +202,18 @@ async fn main() {
202 202
203 203 // Initialize Stripe client if configured
204 204 let stripe: Option<std::sync::Arc<dyn makenotwork::payments::PaymentProvider>> = if let Some(ref stripe_config) = config.stripe {
205 - tracing::info!("Stripe payments initialized");
206 - Some(std::sync::Arc::new(StripeClient::new(stripe_config)))
205 + match StripeClient::new(stripe_config) {
206 + Ok(client) => {
207 + tracing::info!("Stripe payments initialized");
208 + Some(std::sync::Arc::new(client) as std::sync::Arc<dyn makenotwork::payments::PaymentProvider>)
209 + }
210 + Err(e) => {
211 + // Invalid client config is a boot invariant violation — exit
212 + // without restart rather than run with payments silently broken.
213 + tracing::error!(error = %e, "Failed to build Stripe client — exiting without restart");
214 + std::process::exit(2);
215 + }
216 + }
207 217 } else {
208 218 tracing::info!("Stripe not configured. Payments will be unavailable.");
209 219 None
M server/src/payments/connect.rs +8
@@ -311,6 +311,13 @@ impl StripeClient {
311 311 transaction_id: crate::db::TransactionId,
312 312 ) -> Result<()> {
313 313 let acct = Self::parse_account_id(connected_account_id)?;
314 + // Deterministic idempotency key (`refund-{transaction_id}`), mirroring the
315 + // platform-credit transfer below: a retry after a crash or transient
316 + // failure returns the same refund rather than double-debiting the
317 + // creator's connected balance. A transaction is refunded in full exactly
318 + // once, so keying on its id is the correct dedup scope.
319 + let key = IdempotencyKey::new(format!("refund-{transaction_id}"))
320 + .map_err(|e| AppError::Internal(anyhow::anyhow!("invalid idempotency key: {e}")))?;
314 321 let metadata = std::collections::HashMap::from([(
315 322 "mnw_transaction_id".to_string(),
316 323 transaction_id.to_string(),
@@ -321,6 +328,7 @@ impl StripeClient {
321 328 .metadata(metadata)
322 329 .customize()
323 330 .account_id(acct)
331 + .request_strategy(RequestStrategy::Idempotent(key))
324 332 .send(&self.client)
325 333 .await
326 334 .map_err(|e| {
M server/src/payments/mod.rs +22 -6
@@ -28,9 +28,18 @@ pub use synckit_app_pricing::{quote_price_cents, SyncBillingInterval, ANNUAL_MUL
28 28 pub use synckit_billing::SynckitSubResult;
29 29 pub use webhooks::*;
30 30
31 - use stripe::Client;
31 + use std::time::Duration;
32 +
33 + use stripe::{Client, ClientBuilder};
32 34 use crate::config::StripeConfig;
33 35
36 + /// Per-attempt HTTP timeout for outbound Stripe calls. The async client has no
37 + /// timeout by default, so a hung connection would otherwise stall the caller
38 + /// indefinitely — on a webhook handler that holds the response open and invites
39 + /// Stripe's retry storm. 30s matches async-stripe's own blocking-client default;
40 + /// the request strategy still retries a timed-out attempt where permitted.
41 + const STRIPE_HTTP_TIMEOUT: Duration = Duration::from_secs(30);
42 +
34 43 /// Stripe client wrapper for payment operations
35 44 #[derive(Clone)]
36 45 pub struct StripeClient {
@@ -39,13 +48,20 @@ pub struct StripeClient {
39 48 }
40 49
41 50 impl StripeClient {
42 - /// Create a new Stripe client from configuration
43 - pub fn new(config: &StripeConfig) -> Self {
44 - let client = Client::new(&config.secret_key);
45 - StripeClient {
51 + /// Create a new Stripe client from configuration.
52 + ///
53 + /// Fallible because the builder validates the client config; a build error is
54 + /// an internal invariant violation (the secret key comes from validated
55 + /// config), so it is classified `Internal` and surfaces at boot.
56 + pub fn new(config: &StripeConfig) -> Result<Self> {
57 + let client = ClientBuilder::new(&config.secret_key)
58 + .timeout(STRIPE_HTTP_TIMEOUT)
59 + .build()
60 + .map_err(|e| AppError::Internal(anyhow::anyhow!("failed to build Stripe client: {e}")))?;
61 + Ok(StripeClient {
46 62 client,
47 63 config: config.clone(),
48 - }
64 + })
49 65 }
50 66
51 67 /// Parse a connected account ID string into an `AccountId`.
M server/src/routes/api/guest_checkout.rs +22 -7
@@ -271,24 +271,39 @@ pub(super) async fn guest_download(
271 271 .await?
272 272 .ok_or(AppError::NotFound)?;
273 273
274 + // Resolve the downloadable file and the scan status that gates it. Items
275 + // deliver either via an inline audio/video key or via a version file
276 + // (software / digital-file items). A guest download token is item-scoped, so
277 + // a version-delivered item resolves to its current file — the same file an
278 + // authenticated buyer gets from the item page. Previously this path checked
279 + // only the inline keys, so a guest who bought a version-delivered item got a
280 + // token that 404'd despite having paid (Run 15 fulfillment gap).
281 + let (s3_key, scan_status) = match item.audio_s3_key.clone().or(item.video_s3_key.clone()) {
282 + Some(key) => (key, item.scan_status),
283 + None => db::versions::get_versions_by_item(&state.db, item_id)
284 + .await?
285 + .into_iter()
286 + .filter_map(|v| v.s3_key.map(|k| (k, v.scan_status, v.is_current, v.created_at)))
287 + // Prefer the current version, then the most recent, matching what the
288 + // item page hands an authenticated buyer.
289 + .max_by(|a, b| a.2.cmp(&b.2).then(a.3.cmp(&b.3)))
290 + .map(|(key, status, _, _)| (key, status))
291 + .ok_or(AppError::NotFound)?,
292 + };
293 +
274 294 // Gate on scan status, exactly as the authenticated download path does
275 295 // (downloads.rs). A guest is never the creator, so there is no preview
276 296 // exemption: only Clean content is downloadable — Pending/HeldForReview/
277 297 // Quarantined are withheld until the scan clears (Sec NOTE, Run 7; the guest
278 298 // path previously skipped this quarantine check entirely).
279 - if item.scan_status != db::FileScanStatus::Clean {
299 + if scan_status != db::FileScanStatus::Clean {
280 300 return Err(AppError::NotFound);
281 301 }
282 302
283 - // Get the S3 key for the content
284 - let s3_key = item.audio_s3_key.as_deref()
285 - .or(item.video_s3_key.as_deref())
286 - .ok_or_else(|| AppError::NotFound)?;
287 -
288 303 let s3 = state.s3.as_ref()
289 304 .ok_or_else(|| AppError::ServiceUnavailable("File storage is not configured".to_string()))?;
290 305
291 - let download_url = s3.presign_download(&crate::storage::S3Key::from_stored(s3_key), Some(3600)).await?;
306 + let download_url = s3.presign_download(&crate::storage::S3Key::from_stored(&s3_key), Some(3600)).await?;
292 307
293 308 Ok(Redirect::temporary(&download_url).into_response())
294 309 }
M server/tests/harness/mod.rs +2 -1
@@ -168,7 +168,8 @@ impl TestHarness {
168 168 webhook_secret: vec![stripe::TEST_WEBHOOK_SECRET.to_string()],
169 169 webhook_secret_v2: Some(stripe::TEST_WEBHOOK_SECRET_V2.to_string()),
170 170 };
171 - let stripe_client: Arc<dyn PaymentProvider> = Arc::new(StripeClient::new(&stripe_config));
171 + let stripe_client: Arc<dyn PaymentProvider> =
172 + Arc::new(StripeClient::new(&stripe_config).expect("test Stripe client builds"));
172 173 Self::build(BuildOptions {
173 174 stripe_client: Some(stripe_client),
174 175 ..Default::default()