chore: fixes

apps/labrinth/.env.docker-compose

@@ -10,6 +10,7 @@ CDN_URL=file:///tmp/modrinth
 LABRINTH_ADMIN_KEY=feedbeef
 LABRINTH_MEDAL_KEY=
 LABRINTH_EXTERNAL_NOTIFICATION_KEY=beeffeed
+LABRINTH_SUBSCRIPTIONS_KEY=
 RATE_LIMIT_IGNORE_KEY=feedbeef
 
 DATABASE_URL=postgresql://labrinth:labrinth@labrinth-postgres/labrinth

apps/labrinth/.env.local

@@ -10,6 +10,7 @@ CDN_URL=file:///tmp/modrinth
 LABRINTH_ADMIN_KEY=feedbeef
 LABRINTH_MEDAL_KEY=
 LABRINTH_EXTERNAL_NOTIFICATION_KEY=beeffeed
+LABRINTH_SUBSCRIPTIONS_KEY=
 RATE_LIMIT_IGNORE_KEY=feedbeef
 
 DATABASE_URL=postgresql://labrinth:labrinth@localhost/labrinth

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

@@ -1,4 +1,3 @@
-use crate::database::PgTransaction;
 use crate::database::models::{
     DBProductPriceId, DBUserId, DBUserSubscriptionId, DatabaseError,
 };
@@ -131,7 +130,7 @@ impl DBUserSubscription {
 
     pub async fn upsert(
         &self,
-        transaction: &mut PgTransaction<'_>,
+        exec: impl crate::database::Executor<'_, Database = sqlx::Postgres>,
     ) -> Result<(), DatabaseError> {
         sqlx::query!(
             "
@@ -156,7 +155,7 @@ impl DBUserSubscription {
             self.status.as_str(),
             serde_json::to_value(&self.metadata)?,
         )
-        .execute(&mut *transaction)
+        .execute(exec)
         .await?;
 
         Ok(())

apps/labrinth/src/env.rs

@@ -127,6 +127,7 @@ vars! {
     LABRINTH_ADMIN_KEY: String = "";
     LABRINTH_MEDAL_KEY: String = "";
     LABRINTH_EXTERNAL_NOTIFICATION_KEY: String = "";
+    LABRINTH_SUBSCRIPTIONS_KEY: String = "";
     RATE_LIMIT_IGNORE_KEY: String = "";
     DATABASE_URL: String = "postgresql://labrinth:labrinth@localhost/labrinth";
     REDIS_URL: String = "redis://localhost";

apps/labrinth/src/routes/internal/billing.rs

@@ -1,4 +1,5 @@
 use self::payments::*;
+use self::update_subscriptions::*;
 use crate::auth::get_user_from_headers;
 use crate::database::models::charge_item::DBCharge;
 use crate::database::models::ids::DBUserSubscriptionId;
@@ -56,6 +57,7 @@ pub fn config(cfg: &mut actix_web::web::ServiceConfig) {
             .service(charges)
             .service(credit)
             .service(active_servers)
+            .service(update_many)
             .service(initiate_payment)
             .service(stripe_webhook)
             .service(refund_charge)
@@ -63,7 +65,7 @@ pub fn config(cfg: &mut actix_web::web::ServiceConfig) {
     );
 }
 
-/// List products.  
+/// List products.
 #[utoipa::path(
 	context_path = "/billing",
 	tag = "billing",
@@ -113,7 +115,7 @@ struct UserSubscriptionWithNextChargeTaxAmount {
     pub next_charge_tax_amount: Option<i64>,
 }
 
-/// List subscriptions.  
+/// List subscriptions.
 #[utoipa::path(
 	context_path = "/billing",
 	tag = "billing",
@@ -187,7 +189,7 @@ pub struct ChargeRefund {
     pub unprovision: Option<bool>,
 }
 
-/// Refund a charge.  
+/// Refund a charge.
 #[utoipa::path(
 	context_path = "/billing",
 	tag = "billing",
@@ -457,7 +459,7 @@ pub async fn refund_charge(
     Ok(HttpResponse::NoContent().finish())
 }
 
-/// Reprocess tax for a charge.  
+/// Reprocess tax for a charge.
 #[utoipa::path(
 	context_path = "/billing",
 	tag = "billing",
@@ -646,7 +648,7 @@ pub struct SubscriptionEditQuery {
     pub dry: Option<bool>,
 }
 
-/// Update a subscription.  
+/// Update a subscription.
 #[utoipa::path(
 	context_path = "/billing",
 	tag = "billing",
@@ -1146,7 +1148,7 @@ pub async fn edit_subscription(
     }
 }
 
-/// Get the current customer.  
+/// Get the current customer.
 #[utoipa::path(
 	context_path = "/billing",
 	tag = "billing",
@@ -1190,7 +1192,7 @@ pub struct ChargesQuery {
     pub user_id: Option<ariadne::ids::UserId>,
 }
 
-/// List payments.  
+/// List payments.
 #[utoipa::path(
 	context_path = "/billing",
 	tag = "billing",
@@ -1256,7 +1258,7 @@ pub async fn charges(
     ))
 }
 
-/// Start a payment method flow.  
+/// Start a payment method flow.
 #[utoipa::path(
 	context_path = "/billing",
 	tag = "billing",
@@ -1315,7 +1317,7 @@ pub struct EditPaymentMethod {
     pub primary: bool,
 }
 
-/// Update a payment method.  
+/// Update a payment method.
 #[utoipa::path(
 	context_path = "/billing",
 	tag = "billing",
@@ -1385,7 +1387,7 @@ pub async fn edit_payment_method(
     }
 }
 
-/// Remove a payment method.  
+/// Remove a payment method.
 #[utoipa::path(
 	context_path = "/billing",
 	tag = "billing",
@@ -1474,7 +1476,7 @@ pub async fn remove_payment_method(
     }
 }
 
-/// List payment methods.  
+/// List payment methods.
 #[utoipa::path(
 	context_path = "/billing",
 	tag = "billing",
@@ -1537,7 +1539,7 @@ struct ActiveServerResponse {
     pub region: Option<String>,
 }
 
-/// List active servers.  
+/// List active servers.
 #[utoipa::path(
 	context_path = "/billing",
 	tag = "billing",
@@ -1651,7 +1653,7 @@ pub struct PaymentRequest {
     pub metadata: Option<PaymentRequestMetadata>,
 }
 
-/// Initiate a payment.  
+/// Initiate a payment.
 #[utoipa::path(
 	context_path = "/billing",
 	tag = "billing",
@@ -1721,7 +1723,7 @@ pub async fn initiate_payment(
     }
 }
 
-/// Receive a Stripe webhook.  
+/// Receive a Stripe webhook.
 #[utoipa::path(
 	context_path = "/billing",
 	tag = "billing",
@@ -1967,7 +1969,7 @@ pub async fn stripe_webhook(
 
                                     if charge_status != ChargeStatus::Failed {
                                         subscription
-                                            .upsert(transaction)
+                                            .upsert(&mut *transaction)
                                             .await?;
                                     }
 
@@ -2009,7 +2011,7 @@ pub async fn stripe_webhook(
                         };
 
                         if charge_status != ChargeStatus::Failed {
-                            charge.upsert(transaction).await?;
+                            charge.upsert(&mut *transaction).await?;
                         }
 
                         (charge, price, product, subscription, new_region)
@@ -2644,7 +2646,7 @@ pub enum CreditTarget {
     },
 }
 
-/// Credit subscriptions.  
+/// Credit subscriptions.
 #[utoipa::path(
 	context_path = "/billing",
 	tag = "billing",
@@ -2773,3 +2775,4 @@ pub async fn credit(
 }
 
 pub mod payments;
+pub mod update_subscriptions;

apps/labrinth/src/routes/internal/billing/update_subscriptions.rs

@@ -0,0 +1,115 @@
+use std::collections::HashMap;
+
+use actix_web::{post, web};
+use ariadne::ids::UserId as Id;
+use serde::{Deserialize, Serialize};
+use utoipa::ToSchema;
+use uuid::Uuid;
+
+use crate::database::PgPool;
+use crate::database::models::user_subscription_item::DBUserSubscription;
+use crate::models::billing::SubscriptionMetadata;
+use crate::routes::ApiError;
+use crate::util::error::Context;
+use crate::util::guards::subscriptions_key_guard;
+
+#[derive(Debug, Serialize, Deserialize, ToSchema)]
+pub struct UpdateManySubscriptions {
+    pub subscriptions: Vec<SubscriptionUpdate>,
+}
+
+#[derive(Debug, Serialize, Deserialize, ToSchema)]
+pub struct SubscriptionUpdate {
+    pub target: SubscriptionTarget,
+    pub update_region: Option<String>,
+    pub ignore_if_missing: bool,
+}
+
+#[derive(
+    Debug, Clone, Copy, Hash, PartialEq, Eq, Serialize, Deserialize, ToSchema,
+)]
+pub enum SubscriptionTarget {
+    Pyro { user_id: Id, server_id: Uuid },
+}
+
+/// Update multiple managed subscriptions.
+#[utoipa::path(
+	context_path = "/billing",
+	tag = "billing",
+	request_body = UpdateManySubscriptions,
+	responses((status = OK))
+)]
+#[post(
+    "/subscriptions/manager/update_many",
+    guard = "subscriptions_key_guard"
+)]
+pub async fn update_many(
+    pool: web::Data<PgPool>,
+    body: web::Json<UpdateManySubscriptions>,
+) -> Result<(), ApiError> {
+    let UpdateManySubscriptions { subscriptions } = body.into_inner();
+
+    let mut txn = pool
+        .begin()
+        .await
+        .wrap_internal_err("failed to begin transaction")?;
+
+    // Only supports hosting subscriptions now, so gather the server IDs and
+    // fetch the subscriptions from them
+    let server_ids = subscriptions
+        .iter()
+        .map(|update| match update.target {
+            SubscriptionTarget::Pyro { server_id, .. } => server_id.to_string(),
+        })
+        .collect::<Vec<_>>();
+
+    let found_subscriptions =
+        DBUserSubscription::get_many_by_server_ids(&server_ids, &mut txn)
+            .await
+            .wrap_internal_err("failed to fetch subscriptions to update")?;
+
+    let mut subscriptions_by_target =
+        HashMap::<SubscriptionTarget, DBUserSubscription>::new();
+
+    // Creates a map of subscription "key" -> it's DB representation.
+    for subscription in found_subscriptions {
+        let Some(SubscriptionMetadata::Pyro { id, .. }) =
+            subscription.metadata.as_ref()
+        else {
+            continue;
+        };
+        let Ok(server_id) = Uuid::parse_str(id) else {
+            continue;
+        };
+        let target = SubscriptionTarget::Pyro {
+            user_id: subscription.user_id.into(),
+            server_id,
+        };
+
+        subscriptions_by_target.insert(target, subscription);
+    }
+
+    for update in subscriptions {
+        let Some(subscription) =
+            subscriptions_by_target.get_mut(&update.target)
+        else {
+            continue;
+        };
+
+        // Update the subscription region
+        if let Some(SubscriptionMetadata::Pyro { region, .. }) =
+            subscription.metadata.as_mut()
+            && let Some(new_region) = update.update_region.clone()
+        {
+            *region = Some(new_region);
+        }
+
+        subscription.upsert(&mut txn).await?;
+    }
+
+    txn.commit()
+        .await
+        .wrap_internal_err("failed to commit transaction")?;
+
+    Ok(())
+}

apps/labrinth/src/routes/internal/mod.rs

@@ -157,6 +157,7 @@ pub fn config(cfg: &mut web::ServiceConfig) {
 		billing::remove_payment_method,
 		billing::payment_methods,
 		billing::active_servers,
+		billing::update_subscriptions::update_many,
 		billing::initiate_payment,
 		billing::stripe_webhook,
 		billing::credit,

apps/labrinth/src/util/guards.rs

@@ -6,6 +6,7 @@ use crate::env::ENV;
 pub const ADMIN_KEY_HEADER: &str = "Modrinth-Admin";
 pub const MEDAL_KEY_HEADER: &str = "X-Medal-Access-Key";
 pub const EXTERNAL_NOTIFICATION_KEY_HEADER: &str = "External-Notification-Key";
+pub const SUBSCRIPTIONS_KEY_HEADER: &str = "Modrinth-Subscriptions-Key";
 
 pub fn admin_key_guard(ctx: &GuardContext) -> bool {
     ctx.head()
@@ -30,6 +31,20 @@ pub fn external_notification_key_guard(ctx: &GuardContext) -> bool {
         })
 }
 
+pub fn subscriptions_key_guard(ctx: &GuardContext) -> bool {
+    // Ensure the subs key is set and at least 32 characters
+    if ENV.LABRINTH_SUBSCRIPTIONS_KEY.chars().count() < 32 {
+        return false;
+    }
+
+    ctx.head()
+        .headers()
+        .get(SUBSCRIPTIONS_KEY_HEADER)
+        .is_some_and(|it| {
+            it.as_bytes() == ENV.LABRINTH_SUBSCRIPTIONS_KEY.as_bytes()
+        })
+}
+
 pub fn internal_network_guard(ctx: &GuardContext) -> bool {
     ctx.head()
 		.peer_addr