Bommels05/Modrinth
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Use a queue

Browse Source
This commit is contained in:
fetch
2025-08-06 20:19:45 -04:00
parent d6ee0c42c8
commit e8639510aa
7 changed files with 369 additions and 168 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
apps/labrinth/migrations/20250804221014_users-redeemals.sql
View File

@@ -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
);

108
apps/labrinth/src/database/models/users_redeemals.rs
View File

@@ -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<Utc>,
pub last_attempt: Option<DateTime<Utc>>,
pub n_attempts: i32,
pub status: Status,
}
impl UserRedeemal {
pub async fn get_pending<'a, E>(
exec: E,
limit: i64,
) -> sqlx::Result<Vec<UserRedeemal>>
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<bool>
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?;

0
apps/labrinth/src/queue/servers.rs Normal file
View File

185
apps/labrinth/src/routes/internal/billing.rs
View File

@@ -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,

157
apps/labrinth/src/routes/internal/medal.rs
View File

@@ -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<PgPool>,
redis: web::Data<RedisPool>,
web::Query(MedalQuery { username }): web::Query<MedalQuery>,
) -> Result<HttpResponse, ApiError> {
// 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::<PyroServerResponse>()
.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())
}

74
apps/labrinth/src/util/archon.rs Normal file
View File

@@ -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<Self, ApiError> {
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<Uuid, reqwest::Error> {
#[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::<CreateServerResponse>().await?.uuid)
}
}

1
apps/labrinth/src/util/mod.rs
View File

@@ -1,4 +1,5 @@
pub mod actix;
pub mod archon;
pub mod bitflag;
pub mod captcha;
pub mod cors;