Tweaks and fixes to background tasks (#4447)

* adjustments

* chore: query cache, clippy, fmt

Author: fetchfern

apps/labrinth/.sqlx/query-1dacc8ebab576d595a1a5b6a44f1c00ed9709a76cc3a65f48e710d4e73129114.json

@@ -260,13 +260,15 @@ impl DBCharge {
         let charge_type = ChargeType::Subscription.as_str();
         let res = select_charges_with_predicate!(
             r#"
+            INNER JOIN users_subscriptions us ON us.id = charges.subscription_id
             WHERE
-                charge_type = $1 AND
+                charges.charge_type = $1 AND
                 (
-                    (status = 'cancelled' AND due < NOW()) OR
-                    (status = 'expiring' AND due < NOW()) OR
-                    (status = 'failed' AND last_attempt < NOW() - INTERVAL '2 days')
+                    (charges.status = 'cancelled' AND charges.due < NOW()) OR
+                    (charges.status = 'expiring' AND charges.due < NOW()) OR
+                    (charges.status = 'failed' AND charges.last_attempt < NOW() - INTERVAL '2 days')
                 )
+                AND us.status = 'provisioned'
             "#,
             charge_type
         )
@@ -321,6 +323,7 @@ impl DBCharge {
 			  AND COALESCE(tax_last_updated, '-infinity' :: TIMESTAMPTZ) < NOW() - INTERVAL '1 day'
 			  AND u.email IS NOT NULL
 			  AND due - INTERVAL '7 days' > NOW()
+              AND due - INTERVAL '14 days' < NOW() -- Due between 7 and 14 days from now
 			ORDER BY COALESCE(tax_last_updated, '-infinity' :: TIMESTAMPTZ) ASC
 			FOR NO KEY UPDATE SKIP LOCKED
 			LIMIT $1

apps/labrinth/.sqlx/query-7910f8c982de4c9c8f3b40cecda741d6d43c3084b7f347e5da31c353cc9bae53.json

@@ -64,7 +64,7 @@ impl NotificationBuilder {
                   ids.notification_id,
                   ids.user_id,
                   ids.date_available,
-                  COALESCE(SUM(pv.amount), 0.0) sum
+                  FLOOR(COALESCE(SUM(pv.amount), 0.0) * 100) :: BIGINT sum -- Convert to cents
                   FROM UNNEST($1::bigint[], $2::bigint[], $3::timestamptz[]) AS ids(notification_id, user_id, date_available)
                 LEFT JOIN payouts_values pv ON pv.user_id = ids.user_id AND pv.date_available = ids.date_available
                 GROUP BY ids.user_id, ids.notification_id, ids.date_available
@@ -81,6 +81,7 @@ impl NotificationBuilder {
                 'amount', to_jsonb(sum)
               ) body
             FROM period_payouts
+            WHERE sum > 0
             ",
             &notification_ids[..],
             &users_raw_ids[..],

apps/labrinth/.sqlx/query-6ca75db4bd4260c888a8f30fa4da6ac919a7bedf6f07d9d66a9793bf0f7171dd.json renamed to apps/labrinth/.sqlx/query-9f0c73fabe99d9891faaebdd3518b362437dcdcef9cd9a68b950fba61218bb4d.json

@@ -7,7 +7,8 @@ const TEMPLATES_NAMESPACE: &str = "notifications_templates";
 const TEMPLATES_HTML_DATA_NAMESPACE: &str = "notifications_templates_html_data";
 const HTML_DATA_CACHE_EXPIRY: i64 = 60 * 15; // 15 minutes
 
-#[derive(Clone, Serialize, Deserialize)]
+#[derive(Debug, Clone, Serialize, Deserialize)]
+
 pub struct NotificationTemplate {
     pub id: i64,
     pub channel: NotificationChannel,

apps/labrinth/.sqlx/query-f3f819d5761dcd562d13f98826fd2827b25fbdad4898100cf242b21ce8e9713a.json renamed to apps/labrinth/.sqlx/query-cd18ae8abe81a159a134923957f4cd6d0ba3b1bcbc89df349cf7b5b1897603b8.json

@@ -136,7 +136,7 @@ pub enum LegacyNotificationBody {
         new_owner_organization_id: Option<OrganizationId>,
     },
     PayoutAvailable {
-        amount: f64,
+        amount: u64,
         date_available: DateTime<Utc>,
     },
     Unknown,

apps/labrinth/src/database/models/charge_item.rs

@@ -27,7 +27,7 @@ pub struct Notification {
     pub actions: Vec<NotificationAction>,
 }
 
-#[derive(Serialize, Deserialize, Clone, Copy, Eq, PartialEq)]
+#[derive(Debug, Serialize, Deserialize, Clone, Copy, Eq, PartialEq)]
 #[serde(rename_all = "snake_case")]
 pub enum NotificationType {
     // If adding a notification type, add a variant in `NotificationBody` of the same name!
@@ -114,6 +114,17 @@ impl NotificationType {
             "email_changed" => NotificationType::EmailChanged,
             "payment_failed" => NotificationType::PaymentFailed,
             "tax_notification" => NotificationType::TaxNotification,
+            "payout_available" => NotificationType::PayoutAvailable,
+            "moderation_message_received" => {
+                NotificationType::ModerationMessageReceived
+            }
+            "report_status_updated" => NotificationType::ReportStatusUpdated,
+            "report_submitted" => NotificationType::ReportSubmitted,
+            "project_status_approved" => {
+                NotificationType::ProjectStatusApproved
+            }
+            "project_status_neutral" => NotificationType::ProjectStatusNeutral,
+            "project_transferred" => NotificationType::ProjectTransferred,
             "unknown" => NotificationType::Unknown,
             _ => NotificationType::Unknown,
         }
@@ -223,7 +234,7 @@ pub enum NotificationBody {
     },
     PayoutAvailable {
         date_available: DateTime<Utc>,
-        amount: f64,
+        amount: u64,
     },
     Unknown,
 }

apps/labrinth/src/database/models/notification_item.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");
 }