diff --git a/apps/labrinth/migrations/20250804221014_users-redeemals.sql b/apps/labrinth/migrations/20250804221014_users-redeemals.sql index 55b965cbf..94c0511a6 100644 --- a/apps/labrinth/migrations/20250804221014_users-redeemals.sql +++ b/apps/labrinth/migrations/20250804221014_users-redeemals.sql @@ -1,9 +1,11 @@ -- Add migration script here CREATE TABLE IF NOT EXISTS users_redeemals ( - id SERIAL PRIMARY KEY, - user_id BIGINT NOT NULL REFERENCES users(id), - offer VARCHAR NOT NULL, - redeemed TIMESTAMP WITH TIME ZONE NOT NULL, - status VARCHAR NOT NULL + id SERIAL PRIMARY KEY, + user_id BIGINT NOT NULL REFERENCES users(id), + offer VARCHAR NOT NULL, + redeemed TIMESTAMP WITH TIME ZONE NOT NULL, + status VARCHAR NOT NULL, + last_attempt TIMESTAMP WITH TIME ZONE, + n_attempts INTEGER NOT NULL ); \ No newline at end of file diff --git a/apps/labrinth/src/database/models/users_redeemals.rs b/apps/labrinth/src/database/models/users_redeemals.rs index 9faa775e6..2fb62695f 100644 --- a/apps/labrinth/src/database/models/users_redeemals.rs +++ b/apps/labrinth/src/database/models/users_redeemals.rs @@ -41,25 +41,25 @@ impl fmt::Display for Offer { pub enum Status { #[default] Pending, - Redeemed, - Expired, + Processing, + Processed, } impl Status { pub fn as_str(&self) -> &'static str { match self { Status::Pending => "pending", - Status::Redeemed => "redeemed", - Status::Expired => "expired", + Status::Processing => "processing", + Status::Processed => "processed", } } pub fn from_str_or_default(s: &str) -> Self { match s { "pending" => Status::Pending, - "redeemed" => Status::Redeemed, - "expired" => Status::Expired, - _ => Status::Pending, + "processing" => Status::Processing, + "processed" => Status::Processed, + _ => Default::default(), } } } @@ -76,10 +76,62 @@ pub struct UserRedeemal { pub user_id: DBUserId, pub offer: Offer, pub redeemed: DateTime, + pub last_attempt: Option>, + pub n_attempts: i32, pub status: Status, } impl UserRedeemal { + pub async fn get_pending<'a, E>( + exec: E, + limit: i64, + ) -> sqlx::Result> + where + E: sqlx::PgExecutor<'a>, + { + let redeemals = query!( + r#"SELECT * FROM users_redeemals WHERE status = $1 LIMIT $2"#, + Status::Pending.as_str(), + limit + ) + .fetch_all(exec) + .await? + .into_iter() + .map(|row| UserRedeemal { + id: row.id, + user_id: DBUserId(row.user_id), + offer: Offer::from_str_or_default(&row.offer), + redeemed: row.redeemed, + last_attempt: row.last_attempt, + n_attempts: row.n_attempts, + status: Status::from_str_or_default(&row.status), + }) + .collect(); + + Ok(redeemals) + } + + pub async fn update_stuck_5_minutes<'a, E>(exec: E) -> sqlx::Result<()> + where + E: sqlx::PgExecutor<'a>, + { + query!( + r#" + UPDATE users_redeemals + SET status = $1 + WHERE + status = $2 + AND NOW() - last_attempt > INTERVAL '5 minutes' + "#, + Status::Pending.as_str(), + Status::Processing.as_str(), + ) + .execute(exec) + .await?; + + Ok(()) + } + pub async fn exists_by_user_and_offer<'a, E>( exec: E, user_id: DBUserId, @@ -114,13 +166,15 @@ impl UserRedeemal { let query = query_scalar!( r#" INSERT INTO users_redeemals - (user_id, offer, redeemed, status) - VALUES ($1, $2, $3, $4) + (user_id, offer, redeemed, status, last_attempt, n_attempts) + VALUES ($1, $2, $3, $4, $5, $6) RETURNING id"#, self.user_id.0, self.offer.as_str(), self.redeemed, self.status.as_str(), + self.last_attempt, + self.n_attempts, ); let id = query.fetch_one(exec).await?; @@ -130,6 +184,36 @@ impl UserRedeemal { Ok(()) } + /// Updates `status`, `last_attempt`, and `n_attempts` only if `status` is currently pending. + /// Returns `true` if the status was updated, `false` otherwise. + pub async fn update_status_if_pending<'a, E>( + &self, + exec: E, + ) -> sqlx::Result + where + E: sqlx::PgExecutor<'a>, + { + let query = query!( + r#" + UPDATE users_redeemals + SET + status = $3, + last_attempt = $4, + n_attempts = $5 + WHERE id = $1 AND status = $2 + "#, + self.id, + Status::Pending.as_str(), + self.status.as_str(), + self.last_attempt, + self.n_attempts, + ); + + let query_result = query.execute(exec).await?; + + Ok(query_result.rows_affected() > 0) + } + pub async fn update<'a, E>(&self, exec: E) -> sqlx::Result<()> where E: sqlx::PgExecutor<'a>, @@ -140,13 +224,17 @@ impl UserRedeemal { SET offer = $2, status = $3, - redeemed = $4 + redeemed = $4, + last_attempt = $5, + n_attempts = $6 WHERE id = $1 "#, self.id, self.offer.as_str(), self.status.as_str(), self.redeemed, + self.last_attempt, + self.n_attempts, ); query.execute(exec).await?; diff --git a/apps/labrinth/src/queue/servers.rs b/apps/labrinth/src/queue/servers.rs new file mode 100644 index 000000000..e69de29bb diff --git a/apps/labrinth/src/routes/internal/billing.rs b/apps/labrinth/src/routes/internal/billing.rs index a9f2958af..8c2158f0a 100644 --- a/apps/labrinth/src/routes/internal/billing.rs +++ b/apps/labrinth/src/routes/internal/billing.rs @@ -1,5 +1,8 @@ use crate::auth::{get_user_from_headers, send_email}; use crate::database::models::charge_item::DBCharge; +use crate::database::models::user_item::DBUser; +use crate::database::models::user_subscription_item::DBUserSubscription; +use crate::database::models::users_redeemals::{self, UserRedeemal}; use crate::database::models::{ generate_charge_id, generate_user_subscription_id, product_item, user_subscription_item, @@ -14,6 +17,7 @@ use crate::models::pats::Scopes; use crate::models::users::Badges; use crate::queue::session::AuthQueue; use crate::routes::ApiError; +use crate::util::archon::{ArchonClient, CreateServerRequest, Specs}; use actix_web::{HttpRequest, HttpResponse, delete, get, patch, post, web}; use ariadne::ids::base62_impl::{parse_base62, to_base62}; use chrono::{Duration, Utc}; @@ -1717,16 +1721,11 @@ pub async fn stripe_webhook( // Provision subscription match metadata.product_item.metadata { - ProductMetadata::Medal { - cpu: _, - ram: _, - swap: _, - storage: _, - region: _, - } => { - todo!( - "Promote Medal subscription to Pyro subscription" - ) + // A payment shouldn't be processed for Medal subscriptions. + ProductMetadata::Medal { .. } => { + warn!( + "A payment processed for a free subscription" + ); } ProductMetadata::Midas => { @@ -2286,6 +2285,18 @@ pub async fn index_subscriptions(pool: PgPool, redis: RedisPool) { .await?; transaction.commit().await?; + // If an offer redeemal has been processing for over 5 minutes, it should be set pending. + UserRedeemal::update_stuck_5_minutes(&pool).await?; + + // If an offer redeemal is pending, try processing it. + // Try processing it. + let pending_redeemals = UserRedeemal::get_pending(&pool, 100).await?; + for redeemal in pending_redeemals { + if let Err(error) = try_process_user_redeemal(&pool, &redis, redeemal).await { + warn!(%error, "Failed to process a redeemal.") + } + } + Ok::<(), ApiError>(()) }; @@ -2296,6 +2307,160 @@ pub async fn index_subscriptions(pool: PgPool, redis: RedisPool) { info!("Done indexing subscriptions"); } +/// Attempts to process a user redeemal. +/// +/// Returns `Ok` if the entry has been succesfully processed, or will not be processed. +pub async fn try_process_user_redeemal( + pool: &PgPool, + redis: &RedisPool, + mut user_redeemal: UserRedeemal, +) -> Result<(), ApiError> { + // Immediately update redeemal row + 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?; + + if !updated { + return Ok(()); + } + + let user_id = user_redeemal.user_id; + + // Find the Medal product's price & metadata + + let mut medal_products = + product_item::QueryProductWithPrices::list_by_product_type( + pool, "medal", + ) + .await?; + + let Some(product_item::QueryProductWithPrices { + id: _product_id, + metadata, + mut prices, + unitary: _, + }) = medal_products.pop() + else { + return Err(ApiError::Conflict( + "Missing Medal subscription product".to_owned(), + )); + }; + + let ProductMetadata::Medal { + cpu, + ram, + swap, + storage, + region, + } = metadata + else { + return Err(ApiError::Conflict( + "Missing or incorrect metadata for Medal subscription".to_owned(), + )); + }; + + let Some(medal_price) = prices.pop() else { + return Err(ApiError::Conflict( + "Missing price for Medal subscription".to_owned(), + )); + }; + + let (price_duration, price_amount) = match medal_price.prices { + Price::OneTime { price: _ } => { + return Err(ApiError::Conflict( + "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( + "Missing price interval for Medal subscription".to_owned(), + )); + }; + + (price_duration, price_amount) + } + }; + + let price_id = medal_price.id; + + // Get the user's username + + let user = DBUser::get_id(user_id, pool, redis) + .await? + .ok_or(ApiError::NotFound)?; + + // 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 server_id = archon_client + .create_server(&CreateServerRequest { + user_id: to_base62(user_id.0 as u64), + name: format!("{}'s Medal server", user.username), + specs: Specs { + memory_mb: ram, + cpu, + swap_mb: swap, + storage_mb: storage, + }, + source: Default::default(), + region, + }) + .await?; + + let mut txn = pool.begin().await?; + + // Build a subscription using this price ID. + let subscription = DBUserSubscription { + id: generate_user_subscription_id(&mut txn).await?, + user_id, + price_id, + interval: PriceDuration::FiveDays, + created: Utc::now(), + status: SubscriptionStatus::Provisioned, + metadata: Some(SubscriptionMetadata::Medal { + id: server_id.to_string(), + }), + }; + + subscription.upsert(&mut txn).await?; + + // Insert an expiring charge, `index_subscriptions` will unprovision the + // subscription when expired. + DBCharge { + id: generate_charge_id(&mut txn).await?, + user_id, + price_id, + amount: price_amount.into(), + currency_code: medal_price.currency_code, + status: ChargeStatus::Expiring, + due: Utc::now() + price_duration.duration(), + last_attempt: None, + type_: ChargeType::Subscription, + subscription_id: Some(subscription.id), + subscription_interval: Some(subscription.interval), + payment_platform: PaymentPlatform::None, + payment_platform_id: None, + parent_charge_id: None, + net: None, + } + .upsert(&mut txn) + .await?; + + // Update `users_redeemal`, mark subscription as redeemed. + user_redeemal.status = users_redeemals::Status::Processed; + user_redeemal.update(&mut *txn).await?; + + txn.commit().await?; + + Ok(()) +} + pub async fn index_billing( stripe_client: stripe::Client, pool: PgPool, diff --git a/apps/labrinth/src/routes/internal/medal.rs b/apps/labrinth/src/routes/internal/medal.rs index 1c862340a..4bc29e399 100644 --- a/apps/labrinth/src/routes/internal/medal.rs +++ b/apps/labrinth/src/routes/internal/medal.rs @@ -1,24 +1,16 @@ use actix_web::{HttpResponse, post, web}; use ariadne::ids::UserId; -use ariadne::ids::base62_impl::to_base62; use chrono::Utc; use serde::{Deserialize, Serialize}; use sqlx::PgPool; +use tracing::warn; -use crate::database::models::charge_item::DBCharge; -use crate::database::models::product_item; -use crate::database::models::user_subscription_item::DBUserSubscription; use crate::database::models::users_redeemals::{ Offer, RedeemalLookupFields, Status, UserRedeemal, }; -use crate::database::models::{ - generate_charge_id, generate_user_subscription_id, -}; -use crate::models::v3::billing::{ - ChargeStatus, ChargeType, PaymentPlatform, Price, PriceDuration, - ProductMetadata, SubscriptionMetadata, SubscriptionStatus, -}; +use crate::database::redis::RedisPool; use crate::routes::ApiError; +use crate::routes::internal::billing::try_process_user_redeemal; use crate::util::guards::medal_key_guard; pub fn config(cfg: &mut web::ServiceConfig) { @@ -61,9 +53,11 @@ pub async fn verify( #[post("redeem", guard = "medal_key_guard")] pub async fn redeem( pool: web::Data, + redis: web::Data, web::Query(MedalQuery { username }): web::Query, ) -> Result { // Check the offer hasn't been redeemed yet, then insert into the table. + // In a transaction to avoid double inserts. let mut txn = pool.begin().await?; @@ -95,144 +89,21 @@ pub async fn redeem( offer: Offer::Medal, redeemed: Utc::now(), status: Status::Pending, + last_attempt: None, + n_attempts: 0, }; redeemal.insert(&mut *txn).await?; txn.commit().await?; - let client = reqwest::Client::new(); + // Immediately try to process the redeemal + if let Err(error) = try_process_user_redeemal(&pool, &redis, redeemal).await + { + warn!(%error, "Medal redeemal processing failed"); - // Find the Medal product price - let mut medal_products = - product_item::QueryProductWithPrices::list_by_product_type( - &**pool, "medal", - ) - .await?; - - let Some(product_item::QueryProductWithPrices { - id: _product_id, - metadata, - mut prices, - unitary: _, - }) = medal_products.pop() - else { - return Ok(HttpResponse::NotImplemented() - .body("Missing Medal subscription product")); - }; - - let ProductMetadata::Medal { - cpu, - ram, - swap, - storage, - region, - } = metadata - else { - return Ok(HttpResponse::NotImplemented() - .body("Missing or incorrect metadata for Medal subscription")); - }; - - let Some(medal_price) = prices.pop() else { - return Ok(HttpResponse::NotImplemented() - .body("Missing price for Medal subscription")); - }; - - let (price_duration, price_amount) = match medal_price.prices { - Price::OneTime { price: _ } => { - return Ok(HttpResponse::NotImplemented() - .body("Unexpected metadata for Medal subscription price")); - } - - Price::Recurring { intervals } => { - let Some((price_duration, price_amount)) = - intervals.into_iter().next() - else { - return Ok(HttpResponse::NotImplemented() - .body("Missing price interval for Medal subscription")); - }; - - (price_duration, price_amount) - } - }; - - let price_id = medal_price.id; - - #[derive(Deserialize)] - struct PyroServerResponse { - uuid: uuid::Uuid, + Ok(HttpResponse::Accepted().finish()) + } else { + Ok(HttpResponse::Created().finish()) } - - // TODO: archon-client module - let pyro_response = client - .post(format!( - "{}/modrinth/v0/servers/create", - dotenvy::var("ARCHON_URL")?, - )) - .header("X-Master-Key", dotenvy::var("PYRO_API_KEY")?) - .json(&serde_json::json!({ - "user_id": to_base62(user_id.0 as u64), - "name": format!("{}'s Medal server", username), - "specs": { - "memory_mb": ram, - "cpu": cpu, - "swap_mb": swap, - "storage_mb": storage, - }, - "region": region, - "source": {}, // Don't install anything by default (field is ignored on Archon anyways) - "payment_interval": 1, // Doesn't matter, not used on Archon anymore anyways - })) - .send() - .await? - .error_for_status()? - .json::() - .await?; - - let mut txn = pool.begin().await?; - - // Build a subscription using this price ID. - let subscription = DBUserSubscription { - id: generate_user_subscription_id(&mut txn).await?, - user_id, - price_id, - interval: PriceDuration::FiveDays, - created: Utc::now(), - status: SubscriptionStatus::Provisioned, - metadata: Some(SubscriptionMetadata::Medal { - id: pyro_response.uuid.to_string(), - }), - }; - - subscription.upsert(&mut txn).await?; - - // Insert an expiring charge, `index_subscriptions` will unprovision the - // subscription when expired. - DBCharge { - id: generate_charge_id(&mut txn).await?, - user_id, - price_id, - amount: price_amount.into(), - currency_code: medal_price.currency_code, - status: ChargeStatus::Expiring, - due: Utc::now() + price_duration.duration(), - last_attempt: None, - type_: ChargeType::Subscription, - subscription_id: Some(subscription.id), - subscription_interval: Some(subscription.interval), - payment_platform: PaymentPlatform::None, - payment_platform_id: None, - parent_charge_id: None, - net: None, - } - .upsert(&mut txn) - .await?; - - // Update `users_redeemal`, mark subscription as redeemed. - redeemal.status = Status::Redeemed; - redeemal.update(&mut *txn).await?; - - txn.commit().await?; - - Ok(HttpResponse::Ok().finish()) } diff --git a/apps/labrinth/src/util/archon.rs b/apps/labrinth/src/util/archon.rs new file mode 100644 index 000000000..ac10817d8 --- /dev/null +++ b/apps/labrinth/src/util/archon.rs @@ -0,0 +1,74 @@ +use reqwest::header::HeaderName; +use serde::{Deserialize, Serialize}; +use uuid::Uuid; + +use crate::routes::ApiError; + +const X_MASTER_KEY: HeaderName = HeaderName::from_static("x-master-key"); + +#[derive(Debug, Default, Serialize, Deserialize)] +pub struct Empty {} + +#[derive(Debug, Serialize, Deserialize)] +pub struct Specs { + pub memory_mb: u32, + pub cpu: u32, + pub swap_mb: u32, + pub storage_mb: u32, +} + +#[derive(Debug, Serialize, Deserialize)] +pub struct CreateServerRequest { + pub user_id: String, + pub name: String, + pub specs: Specs, + // Must be included because archon doesn't accept null values, only + // an empty struct, as a source. + pub source: Empty, + pub region: String, +} + +#[derive(Clone)] +pub struct ArchonClient { + client: reqwest::Client, + base_url: String, + pyro_api_key: String, +} + +impl ArchonClient { + /// Builds an Archon client from environment variables. Returns `None` if the + /// required environment variables are not set. + pub fn from_env() -> Result { + let client = reqwest::Client::new(); + + let base_url = + dotenvy::var("ARCHON_URL")?.trim_end_matches('/').to_owned(); + + Ok(Self { + client, + base_url, + pyro_api_key: dotenvy::var("PYRO_API_KEY")?, + }) + } + + pub async fn create_server( + &self, + request: &CreateServerRequest, + ) -> Result { + #[derive(Deserialize)] + struct CreateServerResponse { + uuid: Uuid, + } + + let response = self + .client + .post(format!("{}/modrinth/v0/servers/create", self.base_url)) + .header(X_MASTER_KEY, &self.pyro_api_key) + .json(request) + .send() + .await? + .error_for_status()?; + + Ok(response.json::().await?.uuid) + } +} diff --git a/apps/labrinth/src/util/mod.rs b/apps/labrinth/src/util/mod.rs index 94eefffd8..fa9b16ff7 100644 --- a/apps/labrinth/src/util/mod.rs +++ b/apps/labrinth/src/util/mod.rs @@ -1,4 +1,5 @@ pub mod actix; +pub mod archon; pub mod bitflag; pub mod captcha; pub mod cors;