refactor: labrinth ApiError and error reporting (#6981)

* refactor: labrinth `ApiError` and error reporting

* fix clippy

* fix ci

apps/labrinth/src/queue/analytics/mod.rs

@@ -4,7 +4,7 @@ use crate::models::analytics::{
 };
 use crate::routes::ApiError;
 use crate::routes::analytics::MINECRAFT_SERVER_PLAYS;
-use crate::util::error::Context;
+use crate::util::error::Context as _;
 use dashmap::{DashMap, DashSet};
 use std::collections::HashMap;
 use tracing::trace;
@@ -100,25 +100,39 @@ impl AnalyticsQueue {
         if !affiliate_code_clicks_queue.is_empty() {
             let mut insert_clicks = client
                 .insert::<AffiliateCodeClick>("affiliate_code_clicks")
-                .await?;
+                .await
+                .wrap_internal_err("writing analytics data to ClickHouse")?;
 
             for (_, click_vec) in affiliate_code_clicks_queue {
                 for click in click_vec {
-                    insert_clicks.write(&click).await?;
+                    insert_clicks.write(&click).await.wrap_internal_err(
+                        "writing analytics data to ClickHouse",
+                    )?;
                 }
             }
 
-            insert_clicks.end().await?;
+            insert_clicks
+                .end()
+                .await
+                .wrap_internal_err("writing analytics data to ClickHouse")?;
         }
 
         if !playtime_queue.is_empty() {
-            let mut playtimes = client.insert::<Playtime>("playtime").await?;
+            let mut playtimes = client
+                .insert::<Playtime>("playtime")
+                .await
+                .wrap_internal_err("writing analytics data to ClickHouse")?;
 
             for playtime in playtime_queue {
-                playtimes.write(&playtime).await?;
+                playtimes.write(&playtime).await.wrap_internal_err(
+                    "writing analytics data to ClickHouse",
+                )?;
             }
 
-            playtimes.end().await?;
+            playtimes
+                .end()
+                .await
+                .wrap_internal_err("writing analytics data to ClickHouse")?;
         }
 
         if !minecraft_server_plays_queue.is_empty() {
@@ -176,13 +190,19 @@ impl AnalyticsQueue {
 
             let mut plays = client
                 .insert::<MinecraftServerPlay>(MINECRAFT_SERVER_PLAYS)
-                .await?;
+                .await
+                .wrap_internal_err("writing analytics data to ClickHouse")?;
 
             for (_, play) in raw_plays {
-                plays.write(&play).await?;
+                plays.write(&play).await.wrap_internal_err(
+                    "writing analytics data to ClickHouse",
+                )?;
             }
 
-            plays.end().await?;
+            plays
+                .end()
+                .await
+                .wrap_internal_err("writing analytics data to ClickHouse")?;
         }
 
         if !views_queue.is_empty() {
@@ -242,7 +262,10 @@ impl AnalyticsQueue {
                     .wrap_internal_err("writing view count to redis")?;
             }
 
-            let mut views = client.insert::<PageView>("views").await?;
+            let mut views = client
+                .insert::<PageView>("views")
+                .await
+                .wrap_internal_err("writing analytics data to ClickHouse")?;
 
             for (all_views, monetized) in raw_views {
                 for (idx, mut view) in all_views.into_iter().enumerate() {
@@ -250,11 +273,16 @@ impl AnalyticsQueue {
                         view.monetized = false;
                     }
 
-                    views.write(&view).await?;
+                    views.write(&view).await.wrap_internal_err(
+                        "writing analytics data to ClickHouse",
+                    )?;
                 }
             }
 
-            views.end().await?;
+            views
+                .end()
+                .await
+                .wrap_internal_err("writing analytics data to ClickHouse")?;
         }
 
         if !downloads_queue.is_empty() {
@@ -308,8 +336,14 @@ impl AnalyticsQueue {
                     .wrap_internal_err("writing download count to redis")?;
             }
 
-            let mut transaction = pool.begin().await?;
-            let mut downloads = client.insert::<Download>("downloads").await?;
+            let mut transaction = pool
+                .begin()
+                .await
+                .wrap_internal_err("starting database transaction")?;
+            let mut downloads = client
+                .insert::<Download>("downloads")
+                .await
+                .wrap_internal_err("writing analytics data to ClickHouse")?;
 
             let mut version_downloads: HashMap<i64, i32> = HashMap::new();
             let mut project_downloads: HashMap<i64, i32> = HashMap::new();
@@ -329,7 +363,9 @@ impl AnalyticsQueue {
 
                 trace!("writing download {download:?}");
 
-                downloads.write(&download).await?;
+                downloads.write(&download).await.wrap_internal_err(
+                    "writing analytics data to ClickHouse",
+                )?;
             }
 
             sqlx::query!(
@@ -343,7 +379,8 @@ impl AnalyticsQueue {
                 &version_downloads.values().copied().collect::<Vec<_>>(),
             )
             .execute(&mut transaction)
-            .await?;
+            .await
+            .wrap_internal_err("incrementing version download counts")?;
 
             sqlx::query!(
                 "
@@ -356,10 +393,17 @@ impl AnalyticsQueue {
                 &project_downloads.values().copied().collect::<Vec<_>>(),
             )
             .execute(&mut transaction)
-            .await?;
+            .await
+            .wrap_internal_err("incrementing project download counts")?;
 
-            transaction.commit().await?;
-            downloads.end().await?;
+            transaction
+                .commit()
+                .await
+                .wrap_internal_err("committing database transaction")?;
+            downloads
+                .end()
+                .await
+                .wrap_internal_err("writing analytics data to ClickHouse")?;
         }
 
         Ok(())

apps/labrinth/src/queue/billing.rs

@@ -24,6 +24,7 @@ use crate::routes::internal::billing::payments::*;
 use crate::util::anrok;
 use crate::util::archon::ArchonClient;
 use crate::util::archon::{CreateServerRequest, Specs};
+use crate::util::error::ApiContext as _;
 use crate::util::error::Context;
 use ariadne::ids::base62_impl::to_base62;
 use chrono::Utc;
@@ -48,9 +49,14 @@ async fn update_tax_amounts(
     let mut processed_charges = 0;
 
     loop {
-        let mut txn = pg.begin().await?;
+        let mut txn = pg
+            .begin()
+            .await
+            .wrap_internal_err("starting database transaction")?;
 
-        let charges = DBCharge::get_updateable_lock(&mut txn, 5).await?;
+        let charges = DBCharge::get_updateable_lock(&mut txn, 5)
+            .await
+            .wrap_internal_err("fetching charges from database")?;
 
         if charges.is_empty() {
             info!("No more charges to process");
@@ -82,67 +88,70 @@ async fn update_tax_amounts(
                         charge.price_id,
                         &pg,
                     )
-                    .await?
+                    .await
+                    .wrap_api_err("fetching price")?
                     .ok_or_else(|| {
                         DatabaseError::Database(sqlx::Error::RowNotFound)
-                    })?;
+                    })
+                    .wrap_internal_err("querying database for `update_tax_amounts`")?;
 
-                    let product =
-                        DBProduct::get_price(charge.price_id, &pg)
-                            .await?
-                            .ok_or_else(|| {
-                            DatabaseError::Database(
-                                sqlx::Error::RowNotFound,
-                            )
-                        })?;
+                    let product = DBProduct::get_price(charge.price_id, &pg)
+                        .await
+                        .wrap_internal_err(
+                            "fetching product price from database",
+                        )?
+                        .ok_or_else(|| {
+                            DatabaseError::Database(sqlx::Error::RowNotFound)
+                        })
+                        .wrap_internal_err(
+                            "finding product price in database",
+                        )?;
 
                     let stripe_address = 'a: {
-                        let stripe_customer_id =
+                        let stripe_customer =
                             DBUser::get_id(charge.user_id, &pg, &redis)
-                                .await?
-                                .ok_or_else(|| {
-                                    ApiError::from(DatabaseError::Database(
-                                        sqlx::Error::RowNotFound,
-                                    ))
-                                })
-                                .and_then(|user| {
-                                    user.stripe_customer_id.ok_or_else(
-                                        || {
-                                            ApiError::InvalidInput(
-                                            "User has no Stripe customer ID"
-                                                .to_owned(),
-                                        )
-                                        },
-                                    )
-                                })?
-                                .parse()
-                                .map_err(|_| {
-                                    ApiError::InvalidInput(
-                                        "User Stripe customer ID was invalid".to_owned(),
-                                    )
-                                })?;
+                                .await
+                                .wrap_internal_err(
+                                    "fetching Stripe customer from database",
+                                )?
+                                .wrap_internal_err(
+                                    "finding Stripe customer in database",
+                                )?;
+                        let stripe_customer_id = stripe_customer
+                            .stripe_customer_id
+                            .wrap_request_err(
+                                "finding Stripe customer ID on user",
+                            )?
+                            .parse()
+                            .wrap_request_err(
+                                "parsing user Stripe customer ID",
+                            )?;
 
                         let customer = stripe::Customer::retrieve(
                             &stripe_client,
                             &stripe_customer_id,
                             &["invoice_settings.default_payment_method"],
                         )
-                        .await?;
+                        .await
+                        .wrap_failed_dependency_err(
+                            "communicating with payment provider",
+                        )?;
 
                         // A customer should have a default payment method if they have an active subscription.
 
                         let payment_method = customer
                             .invoice_settings
                             .and_then(|x| {
-                                x.default_payment_method.and_then(|x| x.into_object())
+                                x.default_payment_method
+                                    .and_then(|x| x.into_object())
                             })
-                            .ok_or_else(|| {
-                                ApiError::InvalidInput(
-                                    "Customer has no default payment method!".to_string(),
-                                )
+                            .wrap_request_err_with(|| {
+                                "customer has no default payment method!"
+                                    .to_string()
                             })?;
 
-                        let stripe_address = payment_method.billing_details.address;
+                        let stripe_address =
+                            payment_method.billing_details.address;
 
                         // Attempt the default payment method's address first, then the customer's address.
                         match stripe_address {
@@ -152,18 +161,14 @@ async fn update_tax_amounts(
                             }
                         };
 
-                        customer.address.ok_or_else(|| {
-                            ApiError::InvalidInput(
-                                "Couldn't get an address for the Stripe customer"
-                                    .to_owned(),
-                            )
+                        customer.address.wrap_request_err_with(|| {
+                            "couldn't get an address for the Stripe customer"
+                                .to_owned()
                         })?
                     };
 
                     let customer_address =
-                        anrok::Address::from_stripe_address(
-                            &stripe_address,
-                        );
+                        anrok::Address::from_stripe_address(&stripe_address);
 
                     let tax_amount = anrok_client
                         .create_ephemeral_txn(&anrok::TransactionFields {
@@ -179,17 +184,16 @@ async fn update_tax_amounts(
                             customer_id: None,
                             customer_name: None,
                         })
-                        .await?
+                        .await
+                        .wrap_internal_err("inserting database records for `update_tax_amounts`")?
                         .tax_amount_to_collect;
 
-                    Result::<ProcessedCharge, ApiError>::Ok(
-                        ProcessedCharge {
-                            new_tax_amount: tax_amount,
-                            product_name: product
-                                .name
-                                .unwrap_or_else(|| "Modrinth".to_owned()),
-                        },
-                    )
+                    Result::<ProcessedCharge, ApiError>::Ok(ProcessedCharge {
+                        new_tax_amount: tax_amount,
+                        product_name: product
+                            .name
+                            .unwrap_or_else(|| "Modrinth".to_owned()),
+                    })
                 };
 
                 op_fut.then(move |res| async move { (charge_clone, res) })
@@ -212,11 +216,9 @@ async fn update_tax_amounts(
                         // for this.
 
                         let subscription_id =
-                            charge.subscription_id.ok_or_else(|| {
-                                ApiError::InvalidInput(
-                                    "Charge has no subscription ID".to_owned(),
-                                )
-                            })?;
+                            charge.subscription_id.wrap_request_err_with(
+                                || "charge has no subscription ID".to_owned(),
+                            )?;
 
                         NotificationBuilder {
                             body: NotificationBody::TaxNotification {
@@ -234,7 +236,8 @@ async fn update_tax_amounts(
                             },
                         }
                         .insert(charge.user_id, &mut txn, redis)
-                        .await?;
+                        .await
+                        .wrap_internal_err("inserting database records for `update_tax_amounts`")?;
 
                         charge.tax_amount = new_tax_amount;
                     }
@@ -248,10 +251,15 @@ async fn update_tax_amounts(
             };
 
             charge.tax_last_updated = Some(Utc::now());
-            charge.upsert(&mut txn).await?;
+            charge
+                .upsert(&mut txn)
+                .await
+                .wrap_internal_err("updating subscription id in database")?;
         }
 
-        txn.commit().await?;
+        txn.commit()
+            .await
+            .wrap_internal_err("committing database transaction")?;
 
         if processed_charges >= limit {
             break Ok(());
@@ -287,11 +295,9 @@ async fn update_anrok_transactions(
                     .payment_platform_id
                     .as_ref()
                     .and_then(|x| x.parse().ok())
-                    .ok_or_else(|| {
-                        ApiError::InvalidInput(
-                            "Refund charge has no or an invalid refund ID"
-                                .to_owned(),
-                        )
+                    .wrap_request_err_with(|| {
+                        "refund charge has no or an invalid refund ID"
+                            .to_owned()
                     })?;
 
                 let refund = stripe::Refund::retrieve(
@@ -299,15 +305,16 @@ async fn update_anrok_transactions(
                     &refund_id,
                     &["payment_intent.payment_method"],
                 )
-                .await?;
+                .await
+                .wrap_failed_dependency_err(
+                    "communicating with payment provider",
+                )?;
 
                 let pi = refund
                     .payment_intent
                     .and_then(|x| x.into_object())
-                    .ok_or_else(|| {
-                        ApiError::InvalidInput(
-                            "Refund charge has no payment intent".to_owned(),
-                        )
+                    .wrap_request_err_with(|| {
+                        "refund charge has no payment intent".to_owned()
                     })?;
 
                 (pi, anrok::transaction_id_stripe_pyr(&refund_id))
@@ -316,10 +323,8 @@ async fn update_anrok_transactions(
                     .payment_platform_id
                     .as_ref()
                     .and_then(|x| x.parse().ok())
-                    .ok_or_else(|| {
-                        ApiError::InvalidInput(
-                            "Charge has no payment platform ID".to_owned(),
-                        )
+                    .wrap_request_err_with(|| {
+                        "charge has no payment platform ID".to_owned()
                     })?;
 
                 // Attempt retrieving the address via the payment intent's payment method
@@ -329,7 +334,10 @@ async fn update_anrok_transactions(
                     &stripe_id,
                     &["payment_method"],
                 )
-                .await?;
+                .await
+                .wrap_failed_dependency_err(
+                    "communicating with payment provider",
+                )?;
 
                 let anrok_id = anrok::transaction_id_stripe_pi(&stripe_id);
 
@@ -341,27 +349,17 @@ async fn update_anrok_transactions(
                 .and_then(|x| x.into_object())
                 .and_then(|x| x.billing_details.address);
 
-            let stripe_customer_id =
-                DBUser::get_id(c.user_id, &mut *txn, redis)
-                    .await?
-                    .ok_or_else(|| {
-                        ApiError::from(DatabaseError::Database(
-                            sqlx::Error::RowNotFound,
-                        ))
-                    })
-                    .and_then(|user| {
-                        user.stripe_customer_id.ok_or_else(|| {
-                            ApiError::InvalidInput(
-                                "User has no Stripe customer ID".to_owned(),
-                            )
-                        })
-                    })?;
+            let stripe_customer = DBUser::get_id(c.user_id, &mut *txn, redis)
+                .await
+                .wrap_internal_err("fetching Stripe customer from database")?
+                .wrap_internal_err("finding Stripe customer in database")?;
+            let stripe_customer_id = stripe_customer
+                .stripe_customer_id
+                .wrap_request_err("finding Stripe customer ID on user")?;
 
-            let customer_id = stripe_customer_id.parse().map_err(|e| {
-                ApiError::InvalidInput(format!(
-                    "Charge's Stripe customer ID was invalid ({e})"
-                ))
-            })?;
+            let customer_id = stripe_customer_id
+                .parse()
+                .wrap_request_err("parsing request value")?;
 
             match pi_stripe_address {
                 Some(address) => {
@@ -377,7 +375,10 @@ async fn update_anrok_transactions(
 
             let customer =
                 stripe::Customer::retrieve(stripe_client, &customer_id, &[])
-                    .await?;
+                    .await
+                    .wrap_failed_dependency_err(
+                        "communicating with payment provider",
+                    )?;
 
             let Some(address) = customer.address else {
                 // We won't really be able to do anything about this.
@@ -388,7 +389,9 @@ async fn update_anrok_transactions(
                 );
 
                 c.tax_platform_id = Some("unresolved".to_owned());
-                c.upsert(txn).await?;
+                c.upsert(txn)
+                    .await
+                    .wrap_internal_err("updating customer in database")?;
 
                 return Ok(());
             };
@@ -397,8 +400,12 @@ async fn update_anrok_transactions(
         };
 
         let tax_id = DBProductsTaxIdentifier::get_price(c.price_id, &mut *txn)
-            .await?
-            .ok_or_else(|| DatabaseError::Database(sqlx::Error::RowNotFound))?;
+            .await
+            .wrap_api_err("fetching price")?
+            .ok_or_else(|| DatabaseError::Database(sqlx::Error::RowNotFound))
+            .wrap_internal_err(
+                "fetching products tax identifier from database",
+            )?;
 
         // Note: if the tax amount that was charged to the customer is *different* than
         // what it *should* be NOW, we will take on a loss here.
@@ -427,18 +434,18 @@ async fn update_anrok_transactions(
 
         match result {
             Ok(response) => {
-                let version = response.version.ok_or_else(|| {
-                    ApiError::InvalidInput(
-                        "Anrok response is missing tax transaction version"
-                            .to_owned(),
-                    )
+                let version = response.version.wrap_request_err_with(|| {
+                    "anrok response is missing tax transaction version"
+                        .to_owned()
                 })?;
 
                 c.tax_drift_loss = Some(response.tax_amount_to_collect);
                 c.tax_platform_id = Some(tax_platform_id);
                 c.tax_transaction_version = Some(version);
                 c.tax_platform_accounting_time = Some(c.due);
-                c.upsert(txn).await?;
+                c.upsert(txn)
+                    .await
+                    .wrap_internal_err("updating version in database")?;
 
                 Ok(())
             }
@@ -449,11 +456,15 @@ async fn update_anrok_transactions(
                     .is_conflict_and(|x| x == "customerAddressCouldNotResolve")
                 {
                     c.tax_platform_id = Some("unresolved".to_owned());
-                    c.upsert(txn).await?;
+                    c.upsert(txn)
+                        .await
+                        .wrap_internal_err("updating version in database")?;
 
                     Ok(())
                 } else {
-                    Err(error.into())
+                    Err(ApiError::Internal(eyre::eyre!(
+                        "calculating tax with Anrok: {error}"
+                    )))
                 }
             }
         }
@@ -464,11 +475,15 @@ async fn update_anrok_transactions(
     let mut offset = 0;
 
     loop {
-        let mut txn = pg.begin().await?;
+        let mut txn = pg
+            .begin()
+            .await
+            .wrap_internal_err("starting database transaction")?;
 
         let mut charges =
             DBCharge::get_missing_tax_identifier_lock(&mut txn, offset, 1)
-                .await?;
+                .await
+                .wrap_internal_err("fetching charges from database")?;
 
         let Some(c) = charges.pop() else {
             info!("No more charges to process");
@@ -492,7 +507,9 @@ async fn update_anrok_transactions(
             offset += 1;
         }
 
-        txn.commit().await?;
+        txn.commit()
+            .await
+            .wrap_internal_err("committing database transaction")?;
 
         if processed_charges >= limit {
             break Ok(());
@@ -512,7 +529,10 @@ pub async fn try_process_user_redeemal(
     user_redeemal.last_attempt = Some(Utc::now());
     user_redeemal.n_attempts += 1;
     user_redeemal.status = users_redeemals::Status::Processing;
-    let updated = user_redeemal.update_status_if_pending(pool).await?;
+    let updated = user_redeemal
+        .update_status_if_pending(pool)
+        .await
+        .wrap_internal_err("updating updated in database")?;
 
     if !updated {
         return Ok(());
@@ -526,7 +546,8 @@ pub async fn try_process_user_redeemal(
         product_item::QueryProductWithPrices::list_by_product_type(
             pool, "medal",
         )
-        .await?;
+        .await
+        .wrap_internal_err("fetching query product with prices from Redis")?;
 
     let Some(product_item::QueryProductWithPrices {
         id: _product_id,
@@ -536,9 +557,9 @@ pub async fn try_process_user_redeemal(
         name: _,
     }) = medal_products.pop()
     else {
-        return Err(ApiError::Conflict(
+        return Err(ApiError::Conflict(eyre::eyre!(
             "Missing Medal subscription product".to_owned(),
-        ));
+        )));
     };
 
     let ProductMetadata::Medal {
@@ -549,31 +570,31 @@ pub async fn try_process_user_redeemal(
         region,
     } = metadata
     else {
-        return Err(ApiError::Conflict(
+        return Err(ApiError::Conflict(eyre::eyre!(
             "Missing or incorrect metadata for Medal subscription".to_owned(),
-        ));
+        )));
     };
 
     let Some(medal_price) = prices.pop() else {
-        return Err(ApiError::Conflict(
+        return Err(ApiError::Conflict(eyre::eyre!(
             "Missing price for Medal subscription".to_owned(),
-        ));
+        )));
     };
 
     let (price_duration, price_amount) = match medal_price.prices {
         Price::OneTime { price: _ } => {
-            return Err(ApiError::Conflict(
+            return Err(ApiError::Conflict(eyre::eyre!(
                 "Unexpected metadata for Medal subscription price".to_owned(),
-            ));
+            )));
         }
 
         Price::Recurring { intervals } => {
             let Some((price_duration, price_amount)) =
                 intervals.into_iter().next()
             else {
-                return Err(ApiError::Conflict(
+                return Err(ApiError::Conflict(eyre::eyre!(
                     "Missing price interval for Medal subscription".to_owned(),
-                ));
+                )));
             };
 
             (price_duration, price_amount)
@@ -585,13 +606,15 @@ pub async fn try_process_user_redeemal(
     // Get the user's username
 
     let user = DBUser::get_id(user_id, pool, redis)
-        .await?
-        .ok_or(ApiError::NotFound)?;
+        .await
+        .wrap_internal_err("fetching user from database")?
+        .wrap_not_found_err("resource not found")?;
 
     // Send the provision request to Archon. On failure, the redeemal will be "stuck" processing,
     // and moved back to pending by `index_subscriptions`.
 
-    let archon_client = ArchonClient::from_env()?;
+    let archon_client = ArchonClient::from_env()
+        .wrap_api_err("executing `ArchonClient::from_env`")?;
     let server_id = archon_client
         .create_server(&CreateServerRequest {
             user_id: to_base62(user_id.0 as u64),
@@ -606,13 +629,21 @@ pub async fn try_process_user_redeemal(
             region,
             tags: vec!["medal".to_owned()],
         })
-        .await?;
+        .await
+        .wrap_internal_err(
+            "inserting database records for `try_process_user_redeemal`",
+        )?;
 
-    let mut txn = pool.begin().await?;
+    let mut txn = pool
+        .begin()
+        .await
+        .wrap_internal_err("starting database transaction")?;
 
     // Build a subscription using this price ID.
     let subscription = DBUserSubscription {
-        id: generate_user_subscription_id(&mut txn).await?,
+        id: generate_user_subscription_id(&mut txn)
+            .await
+            .wrap_internal_err("generating user subscription ID")?,
         user_id,
         price_id,
         interval: PriceDuration::FiveDays,
@@ -623,12 +654,17 @@ pub async fn try_process_user_redeemal(
         }),
     };
 
-    subscription.upsert(&mut txn).await?;
+    subscription
+        .upsert(&mut txn)
+        .await
+        .wrap_internal_err("generating user subscription ID")?;
 
     // Insert an expiring charge, `index_subscriptions` will unprovision the
     // subscription when expired.
     DBCharge {
-        id: generate_charge_id(&mut txn).await?,
+        id: generate_charge_id(&mut txn)
+            .await
+            .wrap_internal_err("generating redeemal charge ID")?,
         user_id,
         price_id,
         amount: price_amount.into(),
@@ -651,26 +687,42 @@ pub async fn try_process_user_redeemal(
         tax_platform_accounting_time: None,
     }
     .upsert(&mut txn)
-    .await?;
+    .await
+    .wrap_internal_err("upserting redeemal charge")?;
 
     // Update `users_redeemal`, mark subscription as redeemed.
     user_redeemal.status = users_redeemals::Status::Processed;
-    user_redeemal.update(&mut txn).await?;
+    user_redeemal.update(&mut txn).await.wrap_internal_err(
+        "updating database records for `try_process_user_redeemal`",
+    )?;
 
-    txn.commit().await?;
+    txn.commit()
+        .await
+        .wrap_internal_err("committing database transaction")?;
 
     Ok(())
 }
 
 pub async fn cancel_failing_charges(pool: &PgPool) -> Result<(), ApiError> {
-    let charges_to_cancel = DBCharge::get_cancellable(pool).await?;
+    let charges_to_cancel = DBCharge::get_cancellable(pool)
+        .await
+        .wrap_internal_err("fetching charge from database")?;
 
     for mut charge in charges_to_cancel {
         charge.status = ChargeStatus::Cancelled;
 
-        let mut transaction = pool.begin().await?;
-        charge.upsert(&mut transaction).await?;
-        transaction.commit().await?;
+        let mut transaction = pool
+            .begin()
+            .await
+            .wrap_internal_err("starting database transaction")?;
+        charge
+            .upsert(&mut transaction)
+            .await
+            .wrap_internal_err("updating transaction in database")?;
+        transaction
+            .commit()
+            .await
+            .wrap_internal_err("committing database transaction")?;
     }
 
     Ok(())
@@ -682,7 +734,9 @@ pub async fn process_chargeable_charges(
     stripe_client: &stripe::Client,
     anrok_client: &anrok::Client,
 ) -> Result<(), ApiError> {
-    let charges_to_do = DBCharge::get_chargeable(pool).await?;
+    let charges_to_do = DBCharge::get_chargeable(pool)
+        .await
+        .wrap_internal_err("fetching charge from database")?;
 
     let prices = product_item::DBProductPrice::get_many(
         &charges_to_do
@@ -693,7 +747,8 @@ pub async fn process_chargeable_charges(
             .collect::<Vec<_>>(),
         pool,
     )
-    .await?;
+    .await
+    .wrap_internal_err("fetching product prices from database")?;
 
     let users = crate::database::models::DBUser::get_many_ids(
         &charges_to_do
@@ -705,7 +760,8 @@ pub async fn process_chargeable_charges(
         pool,
         redis,
     )
-    .await?;
+    .await
+    .wrap_internal_err("fetching users from database")?;
 
     for mut charge in charges_to_do {
         let Some(product_price) =
@@ -784,9 +840,18 @@ pub async fn process_chargeable_charges(
             charge.status = ChargeStatus::Failed;
         }
 
-        let mut transaction = pool.begin().await?;
-        charge.upsert(&mut transaction).await?;
-        transaction.commit().await?;
+        let mut transaction = pool
+            .begin()
+            .await
+            .wrap_internal_err("starting database transaction")?;
+        charge
+            .upsert(&mut transaction)
+            .await
+            .wrap_internal_err("updating transaction in database")?;
+        transaction
+            .commit()
+            .await
+            .wrap_internal_err("committing database transaction")?;
     }
 
     Ok(())
@@ -798,7 +863,10 @@ async fn unprovision_subscriptions(
 ) -> Result<(), ApiError> {
     info!("Gathering charges to unprovision");
 
-    let mut transaction = pool.begin().await?;
+    let mut transaction = pool
+        .begin()
+        .await
+        .wrap_internal_err("starting database transaction")?;
     let mut clear_cache_users = Vec::new();
 
     // If an active subscription has:
@@ -806,7 +874,9 @@ async fn unprovision_subscriptions(
     // - An expiring charge due now
     // - A failed charge more than two days ago
     // It should be unprovisioned
-    let all_charges = DBCharge::get_unprovision(pool).await?;
+    let all_charges = DBCharge::get_unprovision(pool)
+        .await
+        .wrap_internal_err("fetching charges from database")?;
 
     let mut all_subscriptions =
         user_subscription_item::DBUserSubscription::get_many(
@@ -818,7 +888,8 @@ async fn unprovision_subscriptions(
                 .collect::<Vec<_>>(),
             pool,
         )
-        .await?;
+        .await
+        .wrap_internal_err("fetching user subscriptions from database")?;
     let subscription_prices = product_item::DBProductPrice::get_many(
         &all_subscriptions
             .iter()
@@ -828,7 +899,8 @@ async fn unprovision_subscriptions(
             .collect::<Vec<_>>(),
         pool,
     )
-    .await?;
+    .await
+    .wrap_internal_err("fetching product prices from database")?;
     let subscription_products = product_item::DBProduct::get_many(
         &subscription_prices
             .iter()
@@ -838,7 +910,8 @@ async fn unprovision_subscriptions(
             .collect::<Vec<_>>(),
         pool,
     )
-    .await?;
+    .await
+    .wrap_internal_err("fetching products from database")?;
     let users = DBUser::get_many_ids(
         &all_subscriptions
             .iter()
@@ -849,7 +922,8 @@ async fn unprovision_subscriptions(
         pool,
         redis,
     )
-    .await?;
+    .await
+    .wrap_internal_err("fetching users from database")?;
 
     for charge in all_charges {
         debug!("Unprovisioning charge '{}'", to_base62(charge.id.0 as u64));
@@ -898,7 +972,10 @@ async fn unprovision_subscriptions(
                     user.id as DBUserId,
                 )
                 .execute(&mut transaction)
-                .await?;
+                .await
+                .wrap_internal_err(
+                    "querying database for `unprovision_subscriptions`",
+                )?;
 
                 true
             }
@@ -939,7 +1016,10 @@ async fn unprovision_subscriptions(
 
         if unprovisioned {
             subscription.status = SubscriptionStatus::Unprovisioned;
-            subscription.upsert(&mut transaction).await?;
+            subscription
+                .upsert(&mut transaction)
+                .await
+                .wrap_internal_err("updating err in database")?;
 
             DBUsersSubscriptionsAffiliations::deactivate(
                 subscription.id,
@@ -961,8 +1041,12 @@ async fn unprovision_subscriptions(
             .collect::<Vec<_>>(),
         redis,
     )
-    .await?;
-    transaction.commit().await?;
+    .await
+    .wrap_internal_err("clearing cached data from Redis")?;
+    transaction
+        .commit()
+        .await
+        .wrap_internal_err("committing database transaction")?;
 
     Ok(())
 }
@@ -972,11 +1056,17 @@ async fn process_redeemals(
     redis: &RedisPool,
 ) -> Result<(), ApiError> {
     // If an offer redeemal has been processing for over 5 minutes, it should be set pending.
-    UserRedeemal::update_stuck_5_minutes(pool).await?;
+    UserRedeemal::update_stuck_5_minutes(pool)
+        .await
+        .wrap_internal_err(
+            "updating database records for `process_redeemals`",
+        )?;
 
     // If an offer redeemal is pending, try processing it.
     // Try processing it.
-    let pending_redeemals = UserRedeemal::get_pending(pool, 100).await?;
+    let pending_redeemals = UserRedeemal::get_pending(pool, 100)
+        .await
+        .wrap_internal_err("fetching user redeemal from Redis")?;
     for redeemal in pending_redeemals {
         if let Err(error) =
             try_process_user_redeemal(pool, redis, redeemal).await

apps/labrinth/src/queue/email.rs

@@ -10,6 +10,8 @@ use crate::models::v3::notifications::{
     NotificationChannel, NotificationDeliveryStatus,
 };
 use crate::routes::ApiError;
+use crate::util::error::ApiContext as _;
+use crate::util::error::Context as _;
 use chrono::Utc;
 use futures::stream::{FuturesUnordered, StreamExt};
 use lettre::message::Mailbox;
@@ -150,7 +152,13 @@ impl EmailQueue {
     /// Returns `Ok(false)` if no emails were processed, `Ok(true)` if some were processed.
     #[instrument(name = "EmailQueue::index", skip_all)]
     pub async fn index(&self, limit: i64) -> Result<bool, ApiError> {
-        let transport = self.mailer.lock().await.to_transport().await?;
+        let transport = self
+            .mailer
+            .lock()
+            .await
+            .to_transport()
+            .await
+            .wrap_internal_err("creating email transport")?;
 
         let begin = std::time::Instant::now();
 
@@ -159,7 +167,8 @@ impl EmailQueue {
             limit,
             &self.pg,
         )
-        .await?;
+        .await
+        .wrap_internal_err("creating email transport")?;
 
         if deliveries.is_empty() {
             return Ok(false);
@@ -171,7 +180,9 @@ impl EmailQueue {
         // ballooning the error rate.
         for d in deliveries.iter_mut().filter(|d| d.attempt_count >= 3) {
             d.status = NotificationDeliveryStatus::PermanentlyFailed;
-            d.update(&self.pg).await?;
+            d.update(&self.pg).await.wrap_internal_err(
+                "marking exhausted email delivery as failed",
+            )?;
         }
 
         // We hold a FOR UPDATE lock on the rows here, so no other workers are accessing them
@@ -183,7 +194,9 @@ impl EmailQueue {
             .map(|d| d.notification_id)
             .collect::<Vec<_>>();
         let notifications =
-            DBNotification::get_many(&notification_ids, &self.pg).await?;
+            DBNotification::get_many(&notification_ids, &self.pg)
+                .await
+                .wrap_internal_err("fetching notifications from database")?;
 
         // For all notifications we collected, fill out the template
         // and send it via SMTP in parallel.
@@ -201,11 +214,16 @@ impl EmailQueue {
             let seq = Arc::clone(&sequential_processing);
 
             futures.push(async move {
-                let mut txn = this.pg.begin().await?;
+                let mut txn = this
+                    .pg
+                    .begin()
+                    .await
+                    .wrap_internal_err("starting database transaction")?;
 
                 let maybe_user =
                     DBUser::get_id(notification.user_id, &mut txn, &this.redis)
-                        .await?;
+                        .await
+                        .wrap_internal_err("fetching user from database")?;
 
                 let Some(mailbox) = maybe_user
                     .and_then(|user| user.email)
@@ -268,7 +286,9 @@ impl EmailQueue {
                         };
 
                         delivery.attempt_count += 1;
-                        delivery.update(&self.pg).await?;
+                        delivery.update(&self.pg).await.wrap_internal_err(
+                            "updating processed email delivery",
+                        )?;
                     }
                 }
 
@@ -283,7 +303,10 @@ impl EmailQueue {
             delivery.next_attempt = Utc::now()
                 + chrono::Duration::seconds(EMAIL_RETRY_DELAY_SECONDS);
 
-            delivery.update(&self.pg).await?;
+            delivery
+                .update(&self.pg)
+                .await
+                .wrap_internal_err("scheduling email delivery retry")?;
         }
 
         info!(
@@ -302,7 +325,13 @@ impl EmailQueue {
         user_id: DBUserId,
         address: Mailbox,
     ) -> Result<NotificationDeliveryStatus, ApiError> {
-        let transport = self.mailer.lock().await.to_transport().await?;
+        let transport = self
+            .mailer
+            .lock()
+            .await
+            .to_transport()
+            .await
+            .wrap_internal_err("creating email transport")?;
         self.send_one_with_transport(
             txn,
             transport,
@@ -329,7 +358,8 @@ impl EmailQueue {
             &mut *txn,
             &self.redis,
         )
-        .await?
+        .await
+        .wrap_internal_err("creating email transport")?
         .into_iter()
         .find(|t| t.notification_type == notification.notification_type()) else {
             return Ok(NotificationDeliveryStatus::SkippedDefault);
@@ -345,7 +375,8 @@ impl EmailQueue {
             self.identity.clone(),
             address,
         )
-        .await?;
+        .await
+        .wrap_api_err("executing `templates::build_email`")?;
 
         let send_result = transport.send(message).await;
 

apps/labrinth/src/queue/email/templates.rs

@@ -11,6 +11,7 @@ use crate::database::models::{
 use crate::env::ENV;
 use crate::models::v3::notifications::NotificationBody;
 use crate::routes::ApiError;
+use crate::util::error::ApiContext as _;
 use crate::util::error::Context;
 use crate::util::http::HTTP_CLIENT;
 use ariadne::ids::base62_impl::to_base62;
@@ -135,7 +136,11 @@ pub async fn build_email(
 ) -> Result<Message, ApiError> {
     let get_html_body = async {
         let result: Result<Result<String, reqwest::Error>, ApiError> =
-            match template.get_cached_html_data(redis).await? {
+            match template
+                .get_cached_html_data(redis)
+                .await
+                .wrap_internal_err("fetching email template HTML from Redis")?
+            {
                 Some(html_body) => Ok(Ok(html_body)),
                 None => {
                     let result = client
@@ -149,7 +154,10 @@ pub async fn build_email(
                     if let Ok(ref body) = result {
                         template
                             .set_cached_html_data(body.clone(), redis)
-                            .await?;
+                            .await
+                            .wrap_internal_err(
+                                "updating database records for `build_email`",
+                            )?;
                     }
 
                     Ok(result)
@@ -167,8 +175,10 @@ pub async fn build_email(
     } = from;
 
     let db_user = DBUser::get_id(user_id, &mut *exec, redis)
-        .await?
-        .ok_or(DatabaseError::Database(sqlx::Error::RowNotFound))?;
+        .await
+        .wrap_internal_err("fetching user from database")?
+        .ok_or(DatabaseError::Database(sqlx::Error::RowNotFound))
+        .wrap_internal_err("fetching user from database")?;
 
     let map = [
         (USER_NAME, db_user.username),
@@ -180,17 +190,24 @@ pub async fn build_email(
     let (html_body_result, either) = futures::try_join!(
         get_html_body,
         collect_template_variables(exec, redis, user_id, body, map)
-    )?;
+    )
+    .wrap_api_err("executing `collect_template_variables`")?;
 
     let mut message_builder = Message::builder().from(Mailbox::new(
         Some(from_name),
-        from_address.parse().map_err(MailError::from)?,
+        from_address
+            .parse()
+            .map_err(MailError::from)
+            .wrap_internal_err("reading HTTP response body")?,
     ));
 
     if let Some((name, address)) = reply_name.zip(reply_address) {
         message_builder = message_builder.reply_to(Mailbox::new(
             Some(name),
-            address.parse().map_err(MailError::from)?,
+            address
+                .parse()
+                .map_err(MailError::from)
+                .wrap_internal_err("reading HTTP response body")?,
         ));
     }
 
@@ -244,21 +261,26 @@ pub async fn build_email(
             html: Some(html),
         } => message_builder
             .multipart(MultiPart::alternative_plain_html(plaintext, html))
-            .map_err(MailError::from)?,
+            .map_err(MailError::from)
+            .wrap_internal_err(
+                "executing `MultiPart::alternative_plain_html`",
+            )?,
 
         Body {
             plaintext: Some(plaintext),
             html: None,
         } => message_builder
             .singlepart(SinglePart::plain(plaintext))
-            .map_err(MailError::from)?,
+            .map_err(MailError::from)
+            .wrap_internal_err("executing `SinglePart::plain`")?,
 
         Body {
             plaintext: None,
             html: Some(html),
         } => message_builder
             .singlepart(SinglePart::html(html))
-            .map_err(MailError::from)?,
+            .map_err(MailError::from)
+            .wrap_internal_err("executing `SinglePart::html`")?,
 
         Body {
             plaintext: None,
@@ -329,7 +351,10 @@ async fn resolve_report_title(
         return Ok(title);
     }
 
-    let Some(report) = DBReport::get(report_id, &mut *exec).await? else {
+    let Some(report) = DBReport::get(report_id, &mut *exec)
+        .await
+        .wrap_internal_err("fetching report from database")?
+    else {
         return Ok(title);
     };
     let Some(shared_instance_id) = report.shared_instance_id else {
@@ -398,8 +423,10 @@ async fn collect_template_variables(
                 exec,
                 redis,
             )
-            .await?
-            .ok_or_else(|| DatabaseError::Database(sqlx::Error::RowNotFound))?
+            .await
+            .wrap_api_err("fetching email project")?
+            .ok_or_else(|| DatabaseError::Database(sqlx::Error::RowNotFound))
+            .wrap_internal_err("fetching project from database")?
             .inner;
 
             map.insert(PROJECT_ID, to_base62(project_id.0));
@@ -423,7 +450,7 @@ async fn collect_template_variables(
                 report_id.0 as i64
             )
             .fetch_one(&mut *exec)
-            .await?;
+            .await.wrap_internal_err("querying database for `collect_template_variables`")?;
 
             map.insert(REPORT_ID, to_base62(report_id.0));
             map.insert(
@@ -433,7 +460,8 @@ async fn collect_template_variables(
                     DBReportId(report_id.0 as i64),
                     result.title,
                 )
-                .await?,
+                .await
+                .wrap_api_err("executing `resolve_report_title`")?,
             );
             map.insert(REPORT_DATE, date_human_readable(result.created));
             Ok(EmailTemplate::Static(map))
@@ -453,7 +481,7 @@ async fn collect_template_variables(
                 report_id.0 as i64
             )
             .fetch_one(&mut *exec)
-            .await?;
+            .await.wrap_internal_err("querying database for `collect_template_variables`")?;
 
             map.insert(
                 REPORT_TITLE,
@@ -462,7 +490,8 @@ async fn collect_template_variables(
                     DBReportId(report_id.0 as i64),
                     result.title,
                 )
-                .await?,
+                .await
+                .wrap_api_err("executing `resolve_report_title`")?,
             );
             map.insert(NEWREPORT_ID, to_base62(report_id.0));
             Ok(EmailTemplate::Static(map))
@@ -476,7 +505,10 @@ async fn collect_template_variables(
                 project_id.0 as i64
             )
             .fetch_one(&mut *exec)
-            .await?;
+            .await
+            .wrap_internal_err(
+                "querying database for `collect_template_variables`",
+            )?;
 
             map.insert(PROJECT_ID, to_base62(project_id.0));
             map.insert(PROJECT_NAME, result.name);
@@ -494,8 +526,10 @@ async fn collect_template_variables(
                 exec,
                 redis,
             )
-            .await?
-            .ok_or_else(|| DatabaseError::Database(sqlx::Error::RowNotFound))?
+            .await
+            .wrap_api_err("fetching email project")?
+            .ok_or_else(|| DatabaseError::Database(sqlx::Error::RowNotFound))
+            .wrap_internal_err("fetching project from database")?
             .inner;
 
             map.insert(PROJECT_ID, to_base62(project_id.0));
@@ -516,8 +550,10 @@ async fn collect_template_variables(
                 &mut *exec,
                 redis,
             )
-            .await?
-            .ok_or_else(|| DatabaseError::Database(sqlx::Error::RowNotFound))?
+            .await
+            .wrap_api_err("fetching email project")?
+            .ok_or_else(|| DatabaseError::Database(sqlx::Error::RowNotFound))
+            .wrap_internal_err("fetching project from database")?
             .inner;
 
             map.insert(PROJECT_ID, to_base62(project_id.0));
@@ -530,10 +566,14 @@ async fn collect_template_variables(
                     &mut *exec,
                     redis,
                 )
-                .await?
+                .await
+                .wrap_internal_err("fetching user from database")?
                 .ok_or_else(|| {
                     DatabaseError::Database(sqlx::Error::RowNotFound)
-                })?;
+                })
+                .wrap_internal_err(
+                    "querying database for `collect_template_variables`",
+                )?;
 
                 map.insert(NEWOWNER_TYPE, "user".to_string());
                 map.insert(NEWOWNER_TYPE_CAPITALIZED, "User".to_string());
@@ -546,10 +586,14 @@ async fn collect_template_variables(
                     &mut *exec,
                     redis,
                 )
-                .await?
+                .await
+                .wrap_internal_err("fetching organization from database")?
                 .ok_or_else(|| {
                     DatabaseError::Database(sqlx::Error::RowNotFound)
-                })?;
+                })
+                .wrap_internal_err(
+                    "querying database for `collect_template_variables`",
+                )?;
 
                 map.insert(NEWOWNER_TYPE, "organization".to_string());
                 map.insert(
@@ -584,7 +628,10 @@ async fn collect_template_variables(
                 user_id.0 as i64
             )
             .fetch_one(&mut *exec)
-            .await?;
+            .await
+            .wrap_internal_err(
+                "querying database for `collect_template_variables`",
+            )?;
 
             map.insert(TEAMINVITE_INVITER_NAME, result.inviter_name);
             map.insert(TEAMINVITE_PROJECT_NAME, result.project_name);
@@ -616,7 +663,10 @@ async fn collect_template_variables(
                 user_id.0 as i64
             )
             .fetch_one(&mut *exec)
-            .await?;
+            .await
+            .wrap_internal_err(
+                "querying database for `collect_template_variables`",
+            )?;
 
             map.insert(ORGINVITE_INVITER_NAME, result.inviter_name);
             map.insert(ORGINVITE_ORG_NAME, result.organization_name);
@@ -644,7 +694,10 @@ async fn collect_template_variables(
                 user_id.0 as i64,
             )
             .fetch_one(&mut *exec)
-            .await?;
+            .await
+            .wrap_internal_err(
+                "querying database for `collect_template_variables`",
+            )?;
 
             map.insert(STATUSCHANGE_PROJECT_NAME, result.project_name);
             map.insert(STATUSCHANGE_OLD_STATUS, old_status.as_str().to_owned());
@@ -829,7 +882,9 @@ async fn collect_template_variables(
             key,
         } => Ok(EmailTemplate::Dynamic {
             variables: map,
-            body: dynamic_email_body(redis, title, body_md, key).await?,
+            body: dynamic_email_body(redis, title, body_md, key)
+                .await
+                .wrap_api_err("executing `dynamic_email_body`")?,
             title: title.to_string(),
         }),
 
@@ -844,8 +899,10 @@ async fn collect_template_variables(
                 &mut *exec,
                 redis,
             )
-            .await?
-            .ok_or_else(|| DatabaseError::Database(sqlx::Error::RowNotFound))?;
+            .await
+            .wrap_internal_err("fetching user from database")?
+            .ok_or_else(|| DatabaseError::Database(sqlx::Error::RowNotFound))
+            .wrap_internal_err("fetching user from database")?;
 
             map.insert(SERVERINVITE_INVITER_NAME, inviter.username);
             map.insert(SERVERINVITE_SERVER_NAME, server_name.clone());
@@ -883,9 +940,11 @@ async fn dynamic_email_body(
                 }))
                 .send()
                 .await
-                .and_then(|res| res.error_for_status())?
+                .and_then(|res| res.error_for_status())
+                .wrap_internal_err("deserializing HTTP response")?
                 .bytes()
-                .await?
+                .await
+                .wrap_internal_err("deserializing HTTP response")?
                 .as_ref(),
         )
         .wrap_internal_err("email body is not valid UTF-8")

apps/labrinth/src/queue/payouts/flow/mod.rs

@@ -1,6 +1,7 @@
 //! Centralized place where payout rails are defined - their fees, minimum and
 //! maximum withdraw amounts, and execution logic.
 
+use crate::util::error::ApiContext as _;
 use eyre::eyre;
 use modrinth_util::decimal::Decimal2dp;
 use rust_decimal::Decimal;
@@ -59,7 +60,9 @@ impl PayoutsQueue {
                     self,
                     withdrawal.amount,
                     method_details,
-                    &get_method.await?,
+                    &get_method
+                        .await
+                        .wrap_api_err("executing `tremendous::create`")?,
                 )
                 .await
             }

apps/labrinth/src/queue/payouts/flow/mural.rs

@@ -19,7 +19,7 @@ use crate::{
         mural::MuralPayoutRequest,
     },
     routes::ApiError,
-    util::error::Context,
+    util::error::{ApiContext as _, Context},
 };
 
 pub const PLATFORM_FEE: PayoutMethodFee = PayoutMethodFee {
@@ -67,7 +67,7 @@ pub(super) async fn create(
     let mural = queue.muralpay.load();
     let mural = mural
         .as_ref()
-        .wrap_internal_err("Mural client not available")?;
+        .wrap_internal_err("required Mural client is not available")?;
 
     let method_fee_usd;
     let forex_usd_to_currency;
@@ -158,14 +158,15 @@ pub(super) async fn execute(
         recipient_info,
     }: MuralFlow,
 ) -> Result<(), ApiError> {
-    let user_email = get_verified_email(user)?;
+    let user_email = get_verified_email(user)
+        .wrap_api_err("fetching verified user email")?;
     let sent_to_method_usd = net_usd + method_fee_usd;
     let total_fee_usd = method_fee_usd + platform_fee_usd;
 
     let mural = queue.muralpay.load();
     let mural = mural
         .as_ref()
-        .wrap_internal_err("Mural client not available")?;
+        .wrap_internal_err("required Mural client is not available")?;
 
     let payment_statement_doc = queue
         .create_mural_payment_statement_doc(
@@ -175,7 +176,8 @@ pub(super) async fn execute(
             &recipient_info,
             gotenberg,
         )
-        .await?;
+        .await
+        .wrap_api_err("creating Mural payment statement document")?;
 
     let user_id = UserId::from(user.id);
     let method_id = match &payout_details {
@@ -235,13 +237,18 @@ pub(super) async fn execute(
             Some(format!("User {user_id}")),
             &[payout],
         )
-        .await
-        .map_err(|err| match err {
-            muralpay::MuralError::Api(err) => ApiError::Mural(Box::new(err)),
-            err => ApiError::Internal(
+        .await;
+    let payout_request = match payout_request {
+        Ok(payout_request) => payout_request,
+        Err(muralpay::MuralError::Api(err)) => {
+            return Err(ApiError::Request(eyre::eyre!(Box::new(err))));
+        }
+        Err(err) => {
+            return Err(ApiError::Internal(
                 eyre!(err).wrap_err("failed to create payout request"),
-            ),
-        })?;
+            ));
+        }
+    };
 
     // Once the Mural payout request has been created successfully,
     // then we *must* commit *a* payout row into the DB, to link the Mural

apps/labrinth/src/queue/payouts/flow/paypal.rs

@@ -102,21 +102,20 @@ pub(super) async fn execute(
         if let Some(venmo) = &user.venmo_handle {
             ("Venmo", "user_handle", venmo.clone(), venmo)
         } else {
-            return Err(ApiError::InvalidInput(
-                "Venmo address has not been set for account!".to_string(),
-            ));
+            return Err(ApiError::Request(eyre::eyre!(
+                "Venmo address has not been set for account!",
+            )));
         }
     } else if let Some(paypal_id) = &user.paypal_id {
         if let Some(paypal_country) = &user.paypal_country {
             if paypal_country == "US" && method_id != "paypal_us" {
-                return Err(ApiError::InvalidInput(
-                    "Please use the US PayPal transfer option!".to_string(),
-                ));
+                return Err(ApiError::Request(eyre::eyre!(
+                    "Please use the US PayPal transfer option!",
+                )));
             } else if paypal_country != "US" && method_id == "paypal_us" {
-                return Err(ApiError::InvalidInput(
-                    "Please use the International PayPal transfer option!"
-                        .to_string(),
-                ));
+                return Err(ApiError::Request(eyre::eyre!(
+                    "Please use the International PayPal transfer option!",
+                )));
             }
 
             (
@@ -126,14 +125,14 @@ pub(super) async fn execute(
                 user.paypal_email.as_ref().unwrap_or(paypal_id),
             )
         } else {
-            return Err(ApiError::InvalidInput(
-                "Please re-link your PayPal account!".to_string(),
-            ));
+            return Err(ApiError::Request(eyre::eyre!(
+                "Please re-link your PayPal account!",
+            )));
         }
     } else {
-        return Err(ApiError::InvalidInput(
-            "You have not linked a PayPal account!".to_string(),
-        ));
+        return Err(ApiError::Request(eyre::eyre!(
+            "You have not linked a PayPal account!",
+        )));
     };
 
     let payout_req = json!({

apps/labrinth/src/queue/payouts/flow/tremendous.rs

@@ -1,3 +1,4 @@
+use crate::util::error::ApiContext as _;
 use chrono::Utc;
 use eyre::eyre;
 use modrinth_util::decimal::Decimal2dp;
@@ -64,7 +65,8 @@ pub(super) async fn create(
         "paypal" | "venmo" => {
             let currency = details.currency.unwrap_or(TremendousCurrency::Usd);
             let currency_code = currency.to_string();
-            let usd_to_currency = usd_to_currency_for(&currency_code)?;
+            let usd_to_currency = usd_to_currency_for(&currency_code)
+                .wrap_api_err("executing `usd_to_currency_for`")?;
 
             let fee = PayoutMethodFee {
                 // If a user withdraws $10:
@@ -127,7 +129,8 @@ pub(super) async fn create(
                 } else {
                     TremendousCurrency::Usd.to_string()
                 };
-            let usd_to_currency = usd_to_currency_for(&currency_code)?;
+            let usd_to_currency = usd_to_currency_for(&currency_code)
+                .wrap_api_err("executing `usd_to_currency_for`")?;
             let currency_to_usd = dec!(1) / usd_to_currency;
 
             // no fees
@@ -190,7 +193,8 @@ pub(super) async fn execute(
         pub order: Order,
     }
 
-    let user_email = get_verified_email(user)?;
+    let user_email = get_verified_email(user)
+        .wrap_api_err("fetching verified user email")?;
 
     let order_req = json!({
         "payment": {

apps/labrinth/src/queue/payouts/mod.rs

@@ -8,6 +8,7 @@ use crate::models::payouts::{
 };
 use crate::models::projects::MonetizationStatus;
 use crate::routes::ApiError;
+use crate::util::error::ApiContext as _;
 use crate::util::error::Context;
 use crate::util::webhook::{
     PayoutSourceAlertType, send_slack_payout_source_alert_webhook,
@@ -207,19 +208,17 @@ impl PayoutsQueue {
             .form(&form)
             .send()
             .await
-            .map_err(|_| {
-                ApiError::Payments(
-                    "Error while authenticating with PayPal".to_string(),
-                )
-            })?
+            .map_err(|err| eyre::eyre!(err))
+            .wrap_failed_dependency_err(
+                "error while authenticating with PayPal".to_string(),
+            )?
             .json()
             .await
-            .map_err(|_| {
-                ApiError::Payments(
-                    "Error while authenticating with PayPal (deser error)"
-                        .to_string(),
-                )
-            })?;
+            .map_err(|err| eyre::eyre!(err))
+            .wrap_failed_dependency_err(
+                "error while authenticating with PayPal (deser error)"
+                    .to_string(),
+            )?;
 
         let new_creds = PayPalCredentials {
             access_token: credential.access_token,
@@ -244,21 +243,23 @@ impl PayoutsQueue {
         let credentials = if let Some(credentials) = read.as_ref() {
             if credentials.expires < Utc::now() {
                 drop(read);
-                self.refresh_token().await.map_err(|_| {
-                    ApiError::Payments(
-                        "Error while authenticating with PayPal".to_string(),
-                    )
-                })?
+                self.refresh_token()
+                    .await
+                    .map_err(|err| eyre::eyre!(err))
+                    .wrap_failed_dependency_err(
+                        "error while authenticating with PayPal".to_string(),
+                    )?
             } else {
                 credentials.clone()
             }
         } else {
             drop(read);
-            self.refresh_token().await.map_err(|_| {
-                ApiError::Payments(
-                    "Error while authenticating with PayPal".to_string(),
-                )
-            })?
+            self.refresh_token()
+                .await
+                .map_err(|err| eyre::eyre!(err))
+                .wrap_failed_dependency_err(
+                    "error while authenticating with PayPal".to_string(),
+                )?
         };
 
         let client = reqwest::Client::new();
@@ -287,17 +288,23 @@ impl PayoutsQueue {
                 .body(body);
         }
 
-        let resp = request.send().await.map_err(|_| {
-            ApiError::Payments("could not communicate with PayPal".to_string())
-        })?;
+        let resp = request
+            .send()
+            .await
+            .map_err(|err| eyre::eyre!(err))
+            .wrap_failed_dependency_err(
+                "could not communicate with PayPal".to_string(),
+            )?;
 
         let status = resp.status();
 
-        let value = resp.json::<Value>().await.map_err(|_| {
-            ApiError::Payments(
+        let value = resp
+            .json::<Value>()
+            .await
+            .map_err(|err| eyre::eyre!(err))
+            .wrap_failed_dependency_err(
                 "could not retrieve PayPal response body".to_string(),
-            )
-        })?;
+            )?;
 
         if !status.is_success() {
             #[derive(Deserialize)]
@@ -318,27 +325,28 @@ impl PayoutsQueue {
                 if error.name == "INSUFFICIENT_FUNDS" {
                     error.message = "We're currently transferring funds to our PayPal account. Please try again in a couple days.".to_string();
                 }
-                return Err(ApiError::Payments(format!(
+                return Err(ApiError::FailedDependency(eyre::eyre!(format!(
                     "error name: {}, message: {}",
                     error.name, error.message
-                )));
+                ))));
             }
 
             if let Ok(error) =
                 serde_json::from_value::<PayPalIdentityError>(value)
             {
-                return Err(ApiError::Payments(format!(
+                return Err(ApiError::FailedDependency(eyre::eyre!(format!(
                     "error name: {}, message: {}",
                     error.error, error.error_description
-                )));
+                ))));
             }
 
-            return Err(ApiError::Payments(
-                "could not retrieve PayPal error body".to_string(),
-            ));
+            return Err(ApiError::FailedDependency(eyre::eyre!(
+                "could not retrieve PayPal error body",
+            )));
         }
 
-        Ok(serde_json::from_value(value)?)
+        serde_json::from_value(value)
+            .wrap_request_err("deserializing JSON data")
     }
 
     pub async fn make_tremendous_request<T: Serialize, X: DeserializeOwned>(
@@ -359,19 +367,23 @@ impl PayoutsQueue {
             request = request.json(&body);
         }
 
-        let resp = request.send().await.map_err(|_| {
-            ApiError::Payments(
+        let resp = request
+            .send()
+            .await
+            .map_err(|err| eyre::eyre!(err))
+            .wrap_failed_dependency_err(
                 "could not communicate with Tremendous".to_string(),
-            )
-        })?;
+            )?;
 
         let status = resp.status();
 
-        let value = resp.json::<Value>().await.map_err(|_| {
-            ApiError::Payments(
+        let value = resp
+            .json::<Value>()
+            .await
+            .map_err(|err| eyre::eyre!(err))
+            .wrap_failed_dependency_err(
                 "could not retrieve Tremendous response body".to_string(),
-            )
-        })?;
+            )?;
 
         if !status.is_success()
             && let Some(obj) = value.as_object()
@@ -385,25 +397,25 @@ impl PayoutsQueue {
 
                 let err =
                     serde_json::from_value::<TremendousError>(array.clone())
-                        .map_err(|_| {
-                            ApiError::Payments(
-                                "could not retrieve Tremendous error json body"
-                                    .to_string(),
-                            )
-                        })?;
+                        .map_err(|err| eyre::eyre!(err))
+                        .wrap_failed_dependency_err(
+                            "could not retrieve Tremendous error json body"
+                                .to_string(),
+                        )?;
 
-                return Err(ApiError::Payments(format!(
+                return Err(ApiError::FailedDependency(eyre::eyre!(format!(
                     "Tremendous error: {} ({:?})",
                     err.message, err.payload
-                )));
+                ))));
             }
 
-            return Err(ApiError::Payments(
-                "could not retrieve Tremendous error body".to_string(),
-            ));
+            return Err(ApiError::FailedDependency(eyre::eyre!(
+                "could not retrieve Tremendous error body",
+            )));
         }
 
-        Ok(serde_json::from_value(value)?)
+        serde_json::from_value(value)
+            .wrap_request_err("deserializing JSON data")
     }
 
     pub async fn get_payout_methods(
@@ -469,13 +481,17 @@ impl PayoutsQueue {
         let options = if let Some(options) = read.as_ref() {
             if options.expires < Utc::now() {
                 drop(read);
-                refresh_payout_methods(self).await?
+                refresh_payout_methods(self)
+                    .await
+                    .wrap_api_err("executing `refresh_payout_methods`")?
             } else {
                 options.clone()
             }
         } else {
             drop(read);
-            refresh_payout_methods(self).await?
+            refresh_payout_methods(self)
+                .await
+                .wrap_api_err("executing `refresh_payout_methods`")?
         };
 
         Ok(options.options)
@@ -876,12 +892,18 @@ pub async fn make_aditude_request(
             "interval": interval
         }))
         .send()
-        .await?
-        .error_for_status()?;
+        .await
+        .wrap_internal_err("deserializing HTTP response")?
+        .error_for_status()
+        .wrap_internal_err("deserializing HTTP response")?;
 
-    let text = request.text().await?;
+    let text = request
+        .text()
+        .await
+        .wrap_internal_err("reading HTTP response body")?;
 
-    let json: Vec<AditudePoints> = serde_json::from_str(&text)?;
+    let json: Vec<AditudePoints> = serde_json::from_str(&text)
+        .wrap_request_err("deserializing JSON data")?;
 
     Ok(json)
 }
@@ -900,7 +922,8 @@ pub async fn process_payout(
         crate::models::payouts::PayoutStatus::InTransit.as_str(),
     )
     .execute(pool)
-    .await?;
+    .await
+    .wrap_internal_err("writing analytics data to ClickHouse")?;
 
     let start: DateTime<Utc> = DateTime::from_naive_utc_and_offset(
         (Utc::now() - Duration::days(1))
@@ -915,7 +938,8 @@ pub async fn process_payout(
         start,
     )
     .fetch_one(pool)
-    .await?;
+    .await
+    .wrap_internal_err("querying database for `process_payout`")?;
 
     if results.exists.unwrap_or(false) {
         return Ok(());
@@ -966,9 +990,12 @@ pub async fn process_payout(
             .bind(end.timestamp())
             .fetch_one::<u64>(),
     )
-        .await?;
+        .await.wrap_internal_err("querying database for `process_payout`")?;
 
-    let mut transaction = pool.begin().await?;
+    let mut transaction = pool
+        .begin()
+        .await
+        .wrap_internal_err("starting database transaction")?;
 
     struct PayoutMultipliers {
         sum: u64,
@@ -1029,7 +1056,7 @@ pub async fn process_payout(
             .insert(r.user_id, r.payouts_split);
         async move { Ok(acc) }
     })
-    .await?;
+    .await.wrap_internal_err("inserting project org members into database")?;
 
     let project_team_members = sqlx::query!(
         "
@@ -1055,7 +1082,7 @@ pub async fn process_payout(
             async move { Ok(acc) }
         },
     )
-    .await?;
+    .await.wrap_internal_err("inserting project team members into database")?;
 
     for project_id in project_ids {
         let team_members: HashMap<i64, Decimal> = project_team_members
@@ -1098,7 +1125,8 @@ pub async fn process_payout(
         "Yesterday",
         "1d",
     )
-    .await?;
+    .await
+    .wrap_api_err("executing `make_aditude_request`")?;
 
     let aditude_amount: Decimal = aditude_res
         .iter()
@@ -1193,9 +1221,12 @@ pub async fn process_payout(
         &insert_availables[..]
     )
     .execute(&mut transaction)
-    .await?;
+    .await.wrap_internal_err("inserting database records for `process_payout`")?;
 
-    transaction.commit().await?;
+    transaction
+        .commit()
+        .await
+        .wrap_internal_err("committing database transaction")?;
 
     Ok(())
 }
@@ -1230,14 +1261,18 @@ pub async fn index_payouts_notifications(
 ) -> Result<(), ApiError> {
     info!("Updating payout notifications");
 
-    let mut transaction = pool.begin().await?;
+    let mut transaction = pool
+        .begin()
+        .await
+        .wrap_internal_err("starting database transaction")?;
 
     payouts_values_notifications::synchronize_future_payout_values(
         &mut transaction,
         200,
     )
-    .await?;
-    let items = payouts_values_notifications::PayoutsValuesNotification::unnotified_users_with_available_payouts_with_limit(&mut transaction, 200).await?;
+    .await
+    .wrap_internal_err("executing `payouts_values_notifications::synchronize_future_payout_values`")?;
+    let items = payouts_values_notifications::PayoutsValuesNotification::unnotified_users_with_available_payouts_with_limit(&mut transaction, 200).await.wrap_internal_err("executing `PayoutsValuesNotification::unnotified_users_with_available_payouts_with_limit`")?;
 
     let payout_ref_ids = items.iter().map(|x| x.id).collect::<Vec<_>>();
     let dates_available =
@@ -1250,14 +1285,23 @@ pub async fn index_payouts_notifications(
         &mut transaction,
         redis,
     )
-    .await?;
+    .await
+    .wrap_internal_err(
+        "inserting database records for `index_payouts_notifications`",
+    )?;
     payouts_values_notifications::PayoutsValuesNotification::set_notified_many(
         &payout_ref_ids,
         &mut transaction,
     )
-    .await?;
+    .await
+    .wrap_internal_err(
+        "updating database records for `index_payouts_notifications`",
+    )?;
 
-    transaction.commit().await?;
+    transaction
+        .commit()
+        .await
+        .wrap_internal_err("committing database transaction")?;
 
     Ok(())
 }