Tweaks and fixes to background tasks (#4447)

* adjustments

* chore: query cache, clippy, fmt

apps/labrinth/src/queue/billing.rs

@@ -343,6 +343,8 @@ pub async fn index_subscriptions(
             }
 
             for mut c in charges {
+                processed_charges += 1;
+
                 let payment_intent_id = c
                     .payment_platform_id
                     .as_deref()
@@ -465,8 +467,6 @@ pub async fn index_subscriptions(
                 c.tax_drift_loss = Some(drift);
                 c.tax_platform_id = Some(tax_platform_id);
                 c.upsert(&mut txn).await?;
-
-                processed_charges += 1;
             }
 
             txn.commit().await?;
@@ -477,14 +477,17 @@ pub async fn index_subscriptions(
         }
     }
 
-    let tax_charges_index_handle = tokio::spawn(anrok_api_operations(
+    anrok_api_operations(
         pool.clone(),
         redis.clone(),
         stripe_client.clone(),
         anrok_client.clone(),
-    ));
+    )
+    .await;
 
     let res = async {
+        info!("Gathering charges to unprovision");
+
         let mut transaction = pool.begin().await?;
         let mut clear_cache_users = Vec::new();
 
@@ -539,6 +542,8 @@ pub async fn index_subscriptions(
         .await?;
 
         for charge in all_charges {
+            info!("Indexing charge '{}'", to_base62(charge.id.0 as u64));
+
             let Some(subscription) = all_subscriptions
                 .iter_mut()
                 .find(|x| Some(x.id) == charge.subscription_id)
@@ -664,12 +669,6 @@ pub async fn index_subscriptions(
         warn!("Error indexing subscriptions: {:?}", e);
     }
 
-    if let Err(error) = tax_charges_index_handle.await
-        && error.is_panic()
-    {
-        std::panic::resume_unwind(error.into_panic());
-    }
-
     info!("Done indexing subscriptions");
 }
 

apps/labrinth/src/queue/email.rs

@@ -236,7 +236,7 @@ impl EmailQueue {
                         let update_next_attempt =
                             status == NotificationDeliveryStatus::Pending;
 
-                        let mut delivery = deliveries.remove(idx);
+                        let mut delivery = deliveries.swap_remove(idx);
                         delivery.status = status;
                         delivery.next_attempt += if update_next_attempt {
                             chrono::Duration::seconds(EMAIL_RETRY_DELAY_SECONDS)

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

@@ -559,7 +559,7 @@ async fn collect_template_variables(
 
             map.insert(
                 PAYOUTAVAILABLE_AMOUNT,
-                format!("USD${:.2}", (amount * 100.0) as i64),
+                format!("USD${:.2}", *amount as f64 / 100.0),
             );
 
             Ok(map)

apps/labrinth/src/queue/payouts.rs

@@ -24,7 +24,7 @@ use sqlx::PgPool;
 use sqlx::postgres::PgQueryResult;
 use std::collections::HashMap;
 use tokio::sync::RwLock;
-use tracing::error;
+use tracing::{error, info};
 
 pub struct PayoutsQueue {
     credential: RwLock<Option<PayPalCredentials>>,
@@ -1091,6 +1091,8 @@ pub async fn index_payouts_notifications(
     pool: &PgPool,
     redis: &RedisPool,
 ) -> Result<(), ApiError> {
+    info!("Updating payout notifications");
+
     let mut transaction = pool.begin().await?;
 
     payouts_values_notifications::synchronize_future_payout_values(