237899745/ystp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add configurable verification and redemption codes

Browse Source
This commit is contained in:
237899745
2026-07-25 13:57:39 +08:00
parent d1f093685d
commit c6e96464d3
32 changed files with 2108 additions and 261 deletions
Download Patch File Download Diff File Expand all files Collapse all files

112
src/api/admin.rs
View File

@@ -4,7 +4,9 @@ use crate::error::{AppError, ErrorCode};
use crate::services::billing;
use crate::services::mail;
use crate::services::settings;
use crate::services::settings::{MailConfigStored, MailCustomSmtp, StripeConfigStored};
use crate::services::settings::{
AuthConfigStored, MailConfigStored, MailCustomSmtp, StripeConfigStored,
};
use crate::state::AppState;
use axum::extract::{ConnectInfo, Path, Query, State};
@@ -36,6 +38,8 @@ pub fn router() -> Router<AppState> {
.route("/admin/mail", get(get_mail_config))
.route("/admin/mail", put(update_mail_config))
.route("/admin/mail/test", post(test_mail))
.route("/admin/auth", get(get_auth_config))
.route("/admin/auth", put(update_auth_config))
.route("/admin/config", get(get_config))
.route("/admin/config", put(update_config))
}
@@ -180,7 +184,13 @@ async fn get_stats(
.map_err(|err| AppError::new(ErrorCode::Internal, "查询用量统计失败").with_source(err))?;
let active_subscriptions: i64 = sqlx::query_scalar(
"SELECT COUNT(*) FROM subscriptions WHERE status IN ('active', 'trialing', 'past_due')",
r#"
SELECT COUNT(*)
FROM subscriptions
WHERE status IN ('active', 'trialing', 'past_due')
AND current_period_start <= NOW()
AND current_period_end > NOW()
"#,
)
.fetch_one(&state.db)
.await
@@ -689,8 +699,9 @@ struct GrantCreditsResponse {
period_end: DateTime<Utc>,
used_units: i32,
bonus_units: i32,
total_units: i32,
remaining_units: i32,
redeemed_units: i64,
total_units: i64,
remaining_units: i64,
}
async fn grant_credits(
@@ -722,6 +733,9 @@ async fn grant_credits(
SELECT id, plan_id, current_period_start, current_period_end
FROM subscriptions
WHERE user_id = $1
AND status IN ('active', 'trialing')
AND current_period_start <= NOW()
AND current_period_end > NOW()
ORDER BY current_period_end DESC
LIMIT 1
"#,
@@ -781,6 +795,7 @@ async fn grant_credits(
struct UsageRow {
used_units: i32,
bonus_units: i32,
grant_used_units: i32,
}
let usage = sqlx::query_as::<_, UsageRow>(
@@ -789,7 +804,7 @@ async fn grant_credits(
SET bonus_units = bonus_units + $1,
updated_at = NOW()
WHERE user_id = $2 AND period_start = $3 AND period_end = $4
RETURNING used_units, bonus_units
RETURNING used_units, bonus_units, grant_used_units
"#,
)
.bind(req.units)
@@ -818,8 +833,27 @@ async fn grant_credits(
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "提交事务失败").with_source(err))?;
let total_units = plan_units + usage.bonus_units;
let remaining = (total_units - usage.used_units).max(0);
let redeemed_units: i64 = sqlx::query_scalar(
r#"
SELECT COALESCE(SUM(remaining_units), 0)::bigint
FROM unit_grants
WHERE user_id = $1
AND starts_at <= NOW()
AND expires_at > NOW()
AND remaining_units > 0
"#,
)
.bind(user_id)
.fetch_one(&state.db)
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "查询兑换额度失败").with_source(err))?;
let base_total = i64::from(plan_units.saturating_add(usage.bonus_units));
let base_used = i64::from(usage.used_units.saturating_sub(usage.grant_used_units));
let remaining = base_total
.saturating_sub(base_used)
.max(0)
.saturating_add(redeemed_units);
let total_units = i64::from(usage.used_units).saturating_add(remaining);
Ok(Json(Envelope {
success: true,
@@ -829,6 +863,7 @@ async fn grant_credits(
period_end,
used_units: usage.used_units,
bonus_units: usage.bonus_units,
redeemed_units,
total_units,
remaining_units: remaining,
},
@@ -1274,6 +1309,69 @@ async fn update_stripe_config(
}))
}
#[derive(Debug, Serialize)]
struct AuthConfigView {
email_verification_required: bool,
}
async fn get_auth_config(
State(state): State<AppState>,
jar: axum_extra::extract::cookie::CookieJar,
ConnectInfo(addr): ConnectInfo<SocketAddr>,
headers: HeaderMap,
) -> Result<Json<Envelope<AuthConfigView>>, AppError> {
let ip = context::client_ip(&headers, addr.ip());
let (_jar, _admin_id) = require_admin(&state, jar, &headers, ip).await?;
let config = settings::load_system_config::<AuthConfigStored>(&state, "auth")
.await?
.unwrap_or(AuthConfigStored {
email_verification_required: true,
});
Ok(Json(Envelope {
success: true,
data: AuthConfigView {
email_verification_required: config.email_verification_required,
},
}))
}
#[derive(Debug, Deserialize)]
struct AuthConfigRequest {
email_verification_required: bool,
}
async fn update_auth_config(
State(state): State<AppState>,
jar: axum_extra::extract::cookie::CookieJar,
ConnectInfo(addr): ConnectInfo<SocketAddr>,
headers: HeaderMap,
Json(req): Json<AuthConfigRequest>,
) -> Result<Json<Envelope<AuthConfigView>>, AppError> {
let ip = context::client_ip(&headers, addr.ip());
let (_jar, admin_id) = require_admin(&state, jar, &headers, ip).await?;
let config = AuthConfigStored {
email_verification_required: req.email_verification_required,
};
settings::upsert_system_config(
&state,
"auth",
serde_json::to_value(&config).map_err(|err| {
AppError::new(ErrorCode::Internal, "序列化认证配置失败").with_source(err)
})?,
Some("认证功能开关"),
Some(admin_id),
)
.await?;
Ok(Json(Envelope {
success: true,
data: AuthConfigView {
email_verification_required: config.email_verification_required,
},
}))
}
#[derive(Debug, Serialize)]
struct MailConfigView {
enabled: bool,

76
src/api/auth.rs
View File

@@ -2,6 +2,7 @@ use crate::api::envelope::Envelope;
use crate::auth;
use crate::error::{AppError, ErrorCode};
use crate::services::mail;
use crate::services::settings;
use crate::state::AppState;
use argon2::{Argon2, PasswordHash, PasswordHasher, PasswordVerifier};
@@ -79,11 +80,13 @@ async fn register(
validate_password(&req.password)?;
let password_hash = hash_password(&req.password)?;
let verification_required = settings::email_verification_required(&state).await?;
let verified_at = (!verification_required).then(Utc::now);
let user = sqlx::query_as::<_, UserRow>(
r#"
INSERT INTO users (email, username, password_hash)
VALUES ($1, $2, $3)
INSERT INTO users (email, username, password_hash, email_verified_at)
VALUES ($1, $2, $3, $4)
RETURNING
id,
email,
@@ -97,6 +100,7 @@ async fn register(
.bind(req.email.to_lowercase())
.bind(&req.username)
.bind(password_hash)
.bind(verified_at)
.fetch_one(&state.db)
.await
.map_err(map_unique_violation)?;
@@ -108,34 +112,38 @@ async fn register(
&user.role,
)?;
let verification_token = generate_token();
let token_hash = sha256_hex(&verification_token);
let expires_at_db = Utc::now() + Duration::hours(24);
if verification_required {
let verification_token = generate_token();
let token_hash = sha256_hex(&verification_token);
let expires_at_db = Utc::now() + Duration::hours(24);
sqlx::query(
r#"
INSERT INTO email_verifications (user_id, token_hash, expires_at)
VALUES ($1, $2, $3)
"#,
)
.bind(user.id)
.bind(token_hash)
.bind(expires_at_db)
.execute(&state.db)
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "创建邮箱验证记录失败").with_source(err))?;
let verification_url = format!(
"{}/verify-email?token={}",
state.config.public_base_url, verification_token
);
mail::send_verification_email(&state, &user.email, &user.username, &verification_url)
sqlx::query(
r#"
INSERT INTO email_verifications (user_id, token_hash, expires_at)
VALUES ($1, $2, $3)
"#,
)
.bind(user.id)
.bind(token_hash)
.bind(expires_at_db)
.execute(&state.db)
.await
.map_err(|err| {
AppError::new(ErrorCode::MailSendFailed, "验证邮件发送失败").with_source(err)
AppError::new(ErrorCode::Internal, "创建邮箱验证记录失败").with_source(err)
})?;
let verification_url = format!(
"{}/verify-email?token={}",
state.config.public_base_url, verification_token
);
mail::send_verification_email(&state, &user.email, &user.username, &verification_url)
.await
.map_err(|err| {
AppError::new(ErrorCode::MailSendFailed, "验证邮件发送失败").with_source(err)
})?;
}
let body = RegisterResponse {
user: UserView {
id: user.id,
@@ -145,7 +153,11 @@ async fn register(
email_verified: user.email_verified_at.is_some(),
},
token,
message: "注册成功，验证邮件已发送至您的邮箱".to_string(),
message: if verification_required {
"注册成功，验证邮件已发送至您的邮箱".to_string()
} else {
"注册成功".to_string()
},
};
Ok(Json(Envelope {
@@ -213,6 +225,7 @@ async fn login(
}
verify_password(&req.password, &user.password_hash)?;
let verification_required = settings::email_verification_required(&state).await?;
let (token, expires_at) = auth::issue_jwt(
&state.config.jwt_secret,
@@ -231,7 +244,7 @@ async fn login(
email: user.email,
username: user.username,
role: user.role,
email_verified: user.email_verified_at.is_some(),
email_verified: user.email_verified_at.is_some() || !verification_required,
},
},
}))
@@ -248,6 +261,15 @@ async fn send_verification(
) -> Result<Json<Envelope<MessageResponse>>, AppError> {
let claims = auth::require_jwt(&state.config.jwt_secret, &headers)?;
if !settings::email_verification_required(&state).await? {
return Ok(Json(Envelope {
success: true,
data: MessageResponse {
message: "邮箱验证功能当前已关闭，无需验证".to_string(),
},
}));
}
// Rate limit: 1 per minute per user
let key = format!(
"rate:send_verification:{}:{}",

51
src/api/billing.rs
View File

@@ -3,6 +3,7 @@ use crate::api::envelope::Envelope;
use crate::error::{AppError, ErrorCode};
use crate::services::billing;
use crate::services::idempotency;
use crate::services::quota;
use crate::services::settings;
use crate::state::AppState;
@@ -153,7 +154,9 @@ async fn get_subscription(
FROM subscriptions s
JOIN plans p ON p.id = s.plan_id
WHERE s.user_id = $1
AND s.status IN ('active', 'trialing', 'past_due', 'canceled', 'incomplete')
AND s.status IN ('active', 'trialing', 'past_due')
AND s.current_period_start <= NOW()
AND s.current_period_end > NOW()
ORDER BY s.current_period_end DESC
LIMIT 1
"#,
@@ -239,11 +242,12 @@ async fn get_subscription(
struct UsageResponse {
period_start: DateTime<Utc>,
period_end: DateTime<Utc>,
used_units: i32,
included_units: i32,
bonus_units: i32,
total_units: i32,
remaining_units: i32,
used_units: i64,
included_units: i64,
bonus_units: i64,
redeemed_units: i64,
total_units: i64,
remaining_units: i64,
}
async fn get_usage(
@@ -262,33 +266,7 @@ async fn get_usage(
let billing = billing::get_user_billing(&state, user_id).await?;
#[derive(Debug, FromRow)]
struct UsageRow {
used_units: i32,
bonus_units: i32,
}
let usage = sqlx::query_as::<_, UsageRow>(
r#"
SELECT used_units, bonus_units
FROM usage_periods
WHERE user_id = $1 AND period_start = $2 AND period_end = $3
"#,
)
.bind(user_id)
.bind(billing.period_start)
.bind(billing.period_end)
.fetch_optional(&state.db)
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "查询用量失败").with_source(err))?
.unwrap_or(UsageRow {
used_units: 0,
bonus_units: 0,
});
let included = billing.plan.included_units_per_period;
let total = included + usage.bonus_units;
let remaining = (total - usage.used_units).max(0);
let usage = quota::user_usage_balance(&state, &billing).await?;
Ok(Json(Envelope {
success: true,
@@ -296,10 +274,11 @@ async fn get_usage(
period_start: billing.period_start,
period_end: billing.period_end,
used_units: usage.used_units,
included_units: included,
included_units: usage.included_units,
bonus_units: usage.bonus_units,
total_units: total,
remaining_units: remaining,
redeemed_units: usage.redeemed_units,
total_units: usage.total_units,
remaining_units: usage.remaining_units,
},
}))
}

79
src/api/compress.rs
View File

@@ -1044,41 +1044,7 @@ async fn ensure_quota_available(
ctx: &BillingContext,
needed_units: i32,
) -> Result<(), AppError> {
if needed_units <= 0 {
return Ok(());
}
#[derive(Debug, FromRow)]
struct UsageRow {
used_units: i32,
bonus_units: i32,
}
let usage = sqlx::query_as::<_, UsageRow>(
r#"
SELECT used_units, bonus_units
FROM usage_periods
WHERE user_id = $1 AND period_start = $2 AND period_end = $3
"#,
)
.bind(ctx.user_id)
.bind(ctx.period_start)
.bind(ctx.period_end)
.fetch_optional(&state.db)
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "查询用量失败").with_source(err))?
.unwrap_or(UsageRow {
used_units: 0,
bonus_units: 0,
});
let total_units = ctx.plan.included_units_per_period + usage.bonus_units;
let remaining = total_units - usage.used_units;
if remaining < needed_units {
return Err(AppError::new(ErrorCode::QuotaExceeded, "当期配额已用完"));
}
Ok(())
quota::ensure_user_units(state, ctx, needed_units).await
}
#[allow(clippy::too_many_arguments)]
@@ -1252,48 +1218,7 @@ async fn charge_one_unit(
bytes_in: u64,
bytes_out: u64,
) -> Result<(), AppError> {
sqlx::query(
r#"
INSERT INTO usage_periods (user_id, subscription_id, period_start, period_end)
VALUES ($1, $2, $3, $4)
ON CONFLICT (user_id, period_start, period_end) DO NOTHING
"#,
)
.bind(billing.user_id)
.bind(billing.subscription_id)
.bind(billing.period_start)
.bind(billing.period_end)
.execute(&mut **tx)
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "初始化用量周期失败").with_source(err))?;
let updated: Option<i32> = sqlx::query_scalar(
r#"
UPDATE usage_periods
SET used_units = used_units + 1,
bytes_in = bytes_in + $1,
bytes_out = bytes_out + $2,
updated_at = NOW()
WHERE user_id = $3
AND period_start = $4
AND period_end = $5
AND used_units + 1 <= $6 + bonus_units
RETURNING used_units
"#,
)
.bind(bytes_in as i64)
.bind(bytes_out as i64)
.bind(billing.user_id)
.bind(billing.period_start)
.bind(billing.period_end)
.bind(billing.plan.included_units_per_period)
.fetch_optional(&mut **tx)
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "扣减配额失败").with_source(err))?;
if updated.is_none() {
return Err(AppError::new(ErrorCode::QuotaExceeded, "当期配额已用完"));
}
quota::consume_user_unit(tx, billing, bytes_in, bytes_out).await?;
sqlx::query(
r#"

10
src/api/context.rs
View File

@@ -130,10 +130,13 @@ async fn try_jwt(state: &AppState, headers: &HeaderMap) -> Result<Option<Princip
return Err(AppError::new(ErrorCode::Forbidden, "账号已被禁用"));
}
let verification_required =
crate::services::settings::email_verification_required(state).await?;
Ok(Some(Principal::User {
user_id: user.id,
role: user.role,
email_verified: user.email_verified_at.is_some(),
email_verified: user.email_verified_at.is_some() || !verification_required,
}))
}
@@ -205,11 +208,14 @@ async fn try_api_key(
.execute(&state.db)
.await;
let verification_required =
crate::services::settings::email_verification_required(state).await?;
Ok(Some(Principal::ApiKey {
user_id: row.user_id,
api_key_id: row.id,
role: row.user_role,
email_verified: row.email_verified_at.is_some(),
email_verified: row.email_verified_at.is_some() || !verification_required,
}))
}

2
src/api/mod.rs
View File

@@ -7,6 +7,7 @@ mod context;
mod downloads;
mod envelope;
mod health;
mod redemption;
mod response;
mod tasks;
mod user;
@@ -62,6 +63,7 @@ fn v1_router() -> Router<AppState> {
.merge(billing::router())
.merge(webhooks::router())
.merge(user::router())
.merge(redemption::router())
.merge(admin::router())
.merge(admin_storage::router())
.fallback(response::not_found)

883
src/api/redemption.rs Normal file
View File

@@ -0,0 +1,883 @@
use crate::api::envelope::Envelope;
use crate::api::{admin, context};
use crate::error::{AppError, ErrorCode};
use crate::state::AppState;
use axum::extract::{ConnectInfo, Path, Query, State};
use axum::http::HeaderMap;
use axum::routing::{get, post, put};
use axum::{Json, Router};
use chrono::{DateTime, Duration, Utc};
use rand::RngCore;
use serde::{Deserialize, Serialize};
use sqlx::{FromRow, Postgres, Transaction};
use std::net::SocketAddr;
use uuid::Uuid;
const CODE_PREFIX: &str = "IMG";
const CODE_ALPHABET: &[u8; 32] = b"ABCDEFGHJKLMNPQRSTUVWXYZ23456789";
pub fn router() -> Router<AppState> {
Router::new()
.route("/redemptions", get(list_user_redemptions))
.route("/redemptions/redeem", post(redeem_code))
.route(
"/admin/redemption-codes",
get(list_admin_codes).post(create_codes),
)
.route("/admin/redemption-codes/{code_id}", put(update_code_status))
}
#[derive(Debug, Deserialize)]
struct PagingQuery {
page: Option<u32>,
limit: Option<u32>,
}
#[derive(Debug, FromRow, Serialize)]
struct AdminCodeView {
id: Uuid,
code_hint: String,
benefit_kind: String,
plan_id: Option<Uuid>,
plan_name: Option<String>,
units: Option<i32>,
duration_days: i32,
redeem_before: Option<DateTime<Utc>>,
is_active: bool,
note: Option<String>,
created_at: DateTime<Utc>,
redeemed_at: Option<DateTime<Utc>>,
redeemed_by: Option<Uuid>,
redeemed_username: Option<String>,
}
#[derive(Debug, Serialize)]
struct AdminCodesResponse {
codes: Vec<AdminCodeView>,
page: u32,
limit: u32,
total: i64,
}
async fn list_admin_codes(
State(state): State<AppState>,
jar: axum_extra::extract::cookie::CookieJar,
ConnectInfo(addr): ConnectInfo<SocketAddr>,
headers: HeaderMap,
Query(query): Query<PagingQuery>,
) -> Result<Json<Envelope<AdminCodesResponse>>, AppError> {
let ip = context::client_ip(&headers, addr.ip());
let (_jar, _admin_id) = admin::require_admin(&state, jar, &headers, ip).await?;
let page = query.page.unwrap_or(1).max(1);
let limit = query.limit.unwrap_or(50).clamp(1, 200);
let offset = (page - 1) * limit;
let total: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM redemption_codes")
.fetch_one(&state.db)
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "查询兑换码失败").with_source(err))?;
let codes = sqlx::query_as::<_, AdminCodeView>(
r#"
SELECT
c.id,
c.code_hint,
c.benefit_kind,
c.plan_id,
p.name AS plan_name,
c.units,
c.duration_days,
c.redeem_before,
c.is_active,
c.note,
c.created_at,
r.redeemed_at,
r.user_id AS redeemed_by,
u.username AS redeemed_username
FROM redemption_codes c
LEFT JOIN plans p ON p.id = c.plan_id
LEFT JOIN redemption_records r ON r.code_id = c.id
LEFT JOIN users u ON u.id = r.user_id
ORDER BY c.created_at DESC
LIMIT $1 OFFSET $2
"#,
)
.bind(i64::from(limit))
.bind(i64::from(offset))
.fetch_all(&state.db)
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "查询兑换码失败").with_source(err))?;
Ok(Json(Envelope {
success: true,
data: AdminCodesResponse {
codes,
page,
limit,
total,
},
}))
}
#[derive(Debug, Deserialize)]
struct CreateCodesRequest {
benefit_kind: String,
plan_id: Option<Uuid>,
units: Option<i32>,
duration_days: i32,
redeem_before: Option<DateTime<Utc>>,
quantity: Option<u32>,
note: Option<String>,
}
#[derive(Debug, Serialize)]
struct GeneratedCodeView {
id: Uuid,
code: String,
code_hint: String,
}
#[derive(Debug, Serialize)]
struct CreateCodesResponse {
message: String,
codes: Vec<GeneratedCodeView>,
}
async fn create_codes(
State(state): State<AppState>,
jar: axum_extra::extract::cookie::CookieJar,
ConnectInfo(addr): ConnectInfo<SocketAddr>,
headers: HeaderMap,
Json(req): Json<CreateCodesRequest>,
) -> Result<Json<Envelope<CreateCodesResponse>>, AppError> {
let ip = context::client_ip(&headers, addr.ip());
let (_jar, admin_id) = admin::require_admin(&state, jar, &headers, ip).await?;
let quantity = req.quantity.unwrap_or(1);
if !(1..=200).contains(&quantity) {
return Err(AppError::new(
ErrorCode::InvalidRequest,
"quantity 需在 1-200 之间",
));
}
if !(1..=3650).contains(&req.duration_days) {
return Err(AppError::new(
ErrorCode::InvalidRequest,
"duration_days 需在 1-3650 之间",
));
}
if req
.redeem_before
.is_some_and(|deadline| deadline <= Utc::now())
{
return Err(AppError::new(
ErrorCode::InvalidRequest,
"兑换截止时间必须晚于当前时间",
));
}
let benefit_kind = req.benefit_kind.trim().to_ascii_lowercase();
let (plan_id, units) = match benefit_kind.as_str() {
"plan" => {
let plan_id = req
.plan_id
.ok_or_else(|| AppError::new(ErrorCode::InvalidRequest, "套餐卡必须选择套餐"))?;
let available: bool = sqlx::query_scalar(
"SELECT EXISTS(SELECT 1 FROM plans WHERE id = $1 AND is_active = true)",
)
.bind(plan_id)
.fetch_one(&state.db)
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "查询套餐失败").with_source(err))?;
if !available {
return Err(AppError::new(ErrorCode::NotFound, "套餐不存在或已停用"));
}
(Some(plan_id), None)
}
"units" => {
let units = req
.units
.ok_or_else(|| AppError::new(ErrorCode::InvalidRequest, "次数卡必须填写次数"))?;
if !(1..=10_000_000).contains(&units) {
return Err(AppError::new(
ErrorCode::InvalidRequest,
"units 需在 1-10000000 之间",
));
}
(None, Some(units))
}
_ => {
return Err(AppError::new(
ErrorCode::InvalidRequest,
"benefit_kind 仅支持 plan/units",
))
}
};
let note = req
.note
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty())
.map(str::to_string);
if note
.as_ref()
.is_some_and(|value| value.chars().count() > 500)
{
return Err(AppError::new(
ErrorCode::InvalidRequest,
"备注不能超过 500 字",
));
}
let mut tx = state
.db
.begin()
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "开启事务失败").with_source(err))?;
let mut generated = Vec::with_capacity(quantity as usize);
for _ in 0..quantity {
let code = generate_code();
let compact = normalize_code(&code)
.ok_or_else(|| AppError::new(ErrorCode::Internal, "生成兑换码格式失败"))?;
let code_hash = context::api_key_hash(&compact, &state.config.api_key_pepper)?;
let code_hint = code_hint(&compact);
let id: Uuid = sqlx::query_scalar(
r#"
INSERT INTO redemption_codes (
code_hash, code_hint, benefit_kind, plan_id, units,
duration_days, redeem_before, note, created_by
) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
RETURNING id
"#,
)
.bind(code_hash)
.bind(&code_hint)
.bind(&benefit_kind)
.bind(plan_id)
.bind(units)
.bind(req.duration_days)
.bind(req.redeem_before)
.bind(&note)
.bind(admin_id)
.fetch_one(&mut *tx)
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "生成兑换码失败").with_source(err))?;
generated.push(GeneratedCodeView {
id,
code,
code_hint,
});
}
sqlx::query(
r#"
INSERT INTO audit_logs (user_id, action, resource_type, details, ip_address)
VALUES ($1, 'redemption_codes_created', 'redemption_code', $2, $3::inet)
"#,
)
.bind(admin_id)
.bind(serde_json::json!({
"benefit_kind": benefit_kind,
"plan_id": plan_id,
"units": units,
"duration_days": req.duration_days,
"redeem_before": req.redeem_before,
"quantity": quantity,
"note": note,
}))
.bind(ip.to_string())
.execute(&mut *tx)
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "写入审计日志失败").with_source(err))?;
tx.commit()
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "提交事务失败").with_source(err))?;
Ok(Json(Envelope {
success: true,
data: CreateCodesResponse {
message: "兑换码已生成，完整码仅显示本次".to_string(),
codes: generated,
},
}))
}
#[derive(Debug, Deserialize)]
struct UpdateCodeStatusRequest {
is_active: bool,
}
#[derive(Debug, Serialize)]
struct MessageResponse {
message: String,
}
async fn update_code_status(
State(state): State<AppState>,
jar: axum_extra::extract::cookie::CookieJar,
ConnectInfo(addr): ConnectInfo<SocketAddr>,
headers: HeaderMap,
Path(code_id): Path<Uuid>,
Json(req): Json<UpdateCodeStatusRequest>,
) -> Result<Json<Envelope<MessageResponse>>, AppError> {
let ip = context::client_ip(&headers, addr.ip());
let (_jar, admin_id) = admin::require_admin(&state, jar, &headers, ip).await?;
let mut tx = state
.db
.begin()
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "开启事务失败").with_source(err))?;
let updated =
sqlx::query("UPDATE redemption_codes SET is_active = $2, updated_at = NOW() WHERE id = $1")
.bind(code_id)
.bind(req.is_active)
.execute(&mut *tx)
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "更新兑换码失败").with_source(err))?;
if updated.rows_affected() == 0 {
return Err(AppError::new(ErrorCode::NotFound, "兑换码不存在"));
}
sqlx::query(
r#"
INSERT INTO audit_logs (user_id, action, resource_type, resource_id, details, ip_address)
VALUES ($1, 'redemption_code_status', 'redemption_code', $2, $3, $4::inet)
"#,
)
.bind(admin_id)
.bind(code_id)
.bind(serde_json::json!({ "is_active": req.is_active }))
.bind(ip.to_string())
.execute(&mut *tx)
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "写入审计日志失败").with_source(err))?;
tx.commit()
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "提交事务失败").with_source(err))?;
Ok(Json(Envelope {
success: true,
data: MessageResponse {
message: if req.is_active {
"兑换码已启用".to_string()
} else {
"兑换码已停用".to_string()
},
},
}))
}
#[derive(Debug, Deserialize)]
struct RedeemCodeRequest {
code: String,
}
#[derive(Debug, Serialize)]
struct RedeemCodeResponse {
message: String,
benefit_kind: String,
plan_id: Option<Uuid>,
plan_name: Option<String>,
units: Option<i32>,
benefit_starts_at: DateTime<Utc>,
benefit_expires_at: DateTime<Utc>,
}
#[derive(Debug, FromRow)]
struct RedeemableCodeRow {
id: Uuid,
benefit_kind: String,
plan_id: Option<Uuid>,
plan_name: Option<String>,
plan_is_active: Option<bool>,
units: Option<i32>,
duration_days: i32,
redeem_before: Option<DateTime<Utc>>,
is_active: bool,
redemption_id: Option<Uuid>,
}
async fn redeem_code(
State(state): State<AppState>,
jar: axum_extra::extract::cookie::CookieJar,
ConnectInfo(addr): ConnectInfo<SocketAddr>,
headers: HeaderMap,
Json(req): Json<RedeemCodeRequest>,
) -> Result<Json<Envelope<RedeemCodeResponse>>, AppError> {
let ip = context::client_ip(&headers, addr.ip());
let (_jar, principal) = context::authenticate(&state, jar, &headers, ip).await?;
let user_id = match principal {
context::Principal::User { user_id, .. } => user_id,
_ => return Err(AppError::new(ErrorCode::Unauthorized, "请先登录")),
};
enforce_redeem_rate_limit(&state, user_id).await?;
if req.code.len() > 64 {
return Err(AppError::new(ErrorCode::InvalidRequest, "兑换码无效"));
}
let compact = normalize_code(&req.code)
.ok_or_else(|| AppError::new(ErrorCode::InvalidRequest, "兑换码无效"))?;
let code_hash = context::api_key_hash(&compact, &state.config.api_key_pepper)?;
let mut tx = state
.db
.begin()
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "开启事务失败").with_source(err))?;
let _: Uuid = sqlx::query_scalar("SELECT id FROM users WHERE id = $1 FOR UPDATE")
.bind(user_id)
.fetch_one(&mut *tx)
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "锁定兑换用户失败").with_source(err))?;
let code = sqlx::query_as::<_, RedeemableCodeRow>(
r#"
SELECT
c.id,
c.benefit_kind,
c.plan_id,
p.name AS plan_name,
p.is_active AS plan_is_active,
c.units,
c.duration_days,
c.redeem_before,
c.is_active,
r.id AS redemption_id
FROM redemption_codes c
LEFT JOIN plans p ON p.id = c.plan_id
LEFT JOIN redemption_records r ON r.code_id = c.id
WHERE c.code_hash = $1
FOR UPDATE OF c
"#,
)
.bind(code_hash)
.fetch_optional(&mut *tx)
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "查询兑换码失败").with_source(err))?
.ok_or_else(|| AppError::new(ErrorCode::InvalidRequest, "兑换码无效"))?;
let already_redeemed: bool =
sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM redemption_records WHERE code_id = $1)")
.bind(code.id)
.fetch_one(&mut *tx)
.await
.map_err(|err| {
AppError::new(ErrorCode::Internal, "检查兑换码状态失败").with_source(err)
})?;
let now = Utc::now();
if !code.is_active
|| code.redemption_id.is_some()
|| already_redeemed
|| code.redeem_before.is_some_and(|deadline| deadline <= now)
{
return Err(AppError::new(
ErrorCode::InvalidRequest,
"兑换码无效、已使用或已过期",
));
}
let duration = Duration::days(i64::from(code.duration_days));
let (benefit_starts_at, benefit_expires_at) = match code.benefit_kind.as_str() {
"plan" => {
if code.plan_is_active != Some(true) {
return Err(AppError::new(ErrorCode::Forbidden, "兑换码对应套餐已停用"));
}
let plan_id = code
.plan_id
.ok_or_else(|| AppError::new(ErrorCode::Internal, "套餐兑换码数据不完整"))?;
apply_plan_benefit(&mut tx, user_id, plan_id, duration, now).await?
}
"units" => (now, now + duration),
_ => return Err(AppError::new(ErrorCode::Internal, "兑换码类型错误")),
};
let record_id: Uuid = sqlx::query_scalar(
r#"
INSERT INTO redemption_records (
code_id, user_id, benefit_kind, plan_id, units,
benefit_starts_at, benefit_expires_at, redeemed_ip
) VALUES ($1, $2, $3, $4, $5, $6, $7, $8::inet)
RETURNING id
"#,
)
.bind(code.id)
.bind(user_id)
.bind(&code.benefit_kind)
.bind(code.plan_id)
.bind(code.units)
.bind(benefit_starts_at)
.bind(benefit_expires_at)
.bind(ip.to_string())
.fetch_one(&mut *tx)
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "记录兑换结果失败").with_source(err))?;
if code.benefit_kind == "units" {
let units = code
.units
.ok_or_else(|| AppError::new(ErrorCode::Internal, "次数兑换码数据不完整"))?;
sqlx::query(
r#"
INSERT INTO unit_grants (
user_id, redemption_record_id, total_units, remaining_units,
starts_at, expires_at
) VALUES ($1, $2, $3, $3, $4, $5)
"#,
)
.bind(user_id)
.bind(record_id)
.bind(units)
.bind(benefit_starts_at)
.bind(benefit_expires_at)
.execute(&mut *tx)
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "发放次数额度失败").with_source(err))?;
}
sqlx::query(
r#"
INSERT INTO audit_logs (user_id, action, resource_type, resource_id, details, ip_address)
VALUES ($1, 'redemption_code_redeemed', 'redemption_code', $2, $3, $4::inet)
"#,
)
.bind(user_id)
.bind(code.id)
.bind(serde_json::json!({
"benefit_kind": code.benefit_kind,
"plan_id": code.plan_id,
"units": code.units,
"benefit_expires_at": benefit_expires_at,
}))
.bind(ip.to_string())
.execute(&mut *tx)
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "写入审计日志失败").with_source(err))?;
tx.commit()
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "提交事务失败").with_source(err))?;
let message = if code.benefit_kind == "plan" {
"套餐兑换成功"
} else {
"次数额度兑换成功"
};
Ok(Json(Envelope {
success: true,
data: RedeemCodeResponse {
message: message.to_string(),
benefit_kind: code.benefit_kind,
plan_id: code.plan_id,
plan_name: code.plan_name,
units: code.units,
benefit_starts_at,
benefit_expires_at,
},
}))
}
#[derive(Debug, FromRow)]
struct ActiveSubscriptionRow {
id: Uuid,
plan_id: Uuid,
provider: String,
current_period_start: DateTime<Utc>,
current_period_end: DateTime<Utc>,
}
async fn apply_plan_benefit(
tx: &mut Transaction<'_, Postgres>,
user_id: Uuid,
plan_id: Uuid,
duration: Duration,
now: DateTime<Utc>,
) -> Result<(DateTime<Utc>, DateTime<Utc>), AppError> {
let active = sqlx::query_as::<_, ActiveSubscriptionRow>(
r#"
SELECT id, plan_id, provider, current_period_start, current_period_end
FROM subscriptions
WHERE user_id = $1
AND status IN ('active', 'trialing', 'past_due')
AND current_period_start <= $2
AND current_period_end > $2
ORDER BY current_period_end DESC
FOR UPDATE
LIMIT 1
"#,
)
.bind(user_id)
.bind(now)
.fetch_optional(&mut **tx)
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "查询当前套餐失败").with_source(err))?;
if let Some(active) = active {
if active.provider == "stripe" {
return Err(AppError::new(
ErrorCode::Forbidden,
"当前存在有效的 Stripe 订阅，请在订阅到期后兑换套餐卡",
));
}
let new_end = active.current_period_end + duration;
if active.plan_id == plan_id {
sqlx::query(
r#"
UPDATE usage_periods
SET period_end = $2, updated_at = NOW()
WHERE subscription_id = $1
AND period_start = $3
AND period_end = $4
"#,
)
.bind(active.id)
.bind(new_end)
.bind(active.current_period_start)
.bind(active.current_period_end)
.execute(&mut **tx)
.await
.map_err(|err| {
AppError::new(ErrorCode::Internal, "延长用量周期失败").with_source(err)
})?;
sqlx::query(
"UPDATE subscriptions SET current_period_end = $2, updated_at = NOW() WHERE id = $1",
)
.bind(active.id)
.bind(new_end)
.execute(&mut **tx)
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "延长套餐失败").with_source(err))?;
return Ok((now, new_end));
}
sqlx::query(
r#"
UPDATE subscriptions
SET status = 'canceled', canceled_at = NOW(), updated_at = NOW()
WHERE id = $1
"#,
)
.bind(active.id)
.execute(&mut **tx)
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "切换套餐失败").with_source(err))?;
create_redemption_subscription(tx, user_id, plan_id, now, new_end).await?;
return Ok((now, new_end));
}
let end = now + duration;
create_redemption_subscription(tx, user_id, plan_id, now, end).await?;
Ok((now, end))
}
async fn create_redemption_subscription(
tx: &mut Transaction<'_, Postgres>,
user_id: Uuid,
plan_id: Uuid,
start: DateTime<Utc>,
end: DateTime<Utc>,
) -> Result<Uuid, AppError> {
let subscription_id: Uuid = sqlx::query_scalar(
r#"
INSERT INTO subscriptions (
user_id, plan_id, status, current_period_start, current_period_end,
cancel_at_period_end, provider, created_at, updated_at
) VALUES ($1, $2, 'active', $3, $4, false, 'redemption', NOW(), NOW())
RETURNING id
"#,
)
.bind(user_id)
.bind(plan_id)
.bind(start)
.bind(end)
.fetch_one(&mut **tx)
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "开通兑换套餐失败").with_source(err))?;
sqlx::query(
r#"
INSERT INTO usage_periods (user_id, subscription_id, period_start, period_end)
VALUES ($1, $2, $3, $4)
ON CONFLICT (user_id, period_start, period_end) DO NOTHING
"#,
)
.bind(user_id)
.bind(subscription_id)
.bind(start)
.bind(end)
.execute(&mut **tx)
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "初始化兑换套餐用量失败").with_source(err))?;
Ok(subscription_id)
}
#[derive(Debug, FromRow, Serialize)]
struct UserRedemptionView {
id: Uuid,
code_hint: String,
benefit_kind: String,
plan_name: Option<String>,
units: Option<i32>,
remaining_units: Option<i32>,
benefit_starts_at: DateTime<Utc>,
benefit_expires_at: DateTime<Utc>,
redeemed_at: DateTime<Utc>,
}
#[derive(Debug, Serialize)]
struct UserRedemptionsResponse {
redemptions: Vec<UserRedemptionView>,
}
async fn list_user_redemptions(
State(state): State<AppState>,
jar: axum_extra::extract::cookie::CookieJar,
ConnectInfo(addr): ConnectInfo<SocketAddr>,
headers: HeaderMap,
) -> Result<Json<Envelope<UserRedemptionsResponse>>, AppError> {
let ip = context::client_ip(&headers, addr.ip());
let (_jar, principal) = context::authenticate(&state, jar, &headers, ip).await?;
let user_id = match principal {
context::Principal::User { user_id, .. } => user_id,
_ => return Err(AppError::new(ErrorCode::Unauthorized, "请先登录")),
};
let redemptions = sqlx::query_as::<_, UserRedemptionView>(
r#"
SELECT
r.id,
c.code_hint,
r.benefit_kind,
p.name AS plan_name,
r.units,
CASE WHEN g.expires_at > NOW() THEN g.remaining_units ELSE 0 END AS remaining_units,
r.benefit_starts_at,
r.benefit_expires_at,
r.redeemed_at
FROM redemption_records r
JOIN redemption_codes c ON c.id = r.code_id
LEFT JOIN plans p ON p.id = r.plan_id
LEFT JOIN unit_grants g ON g.redemption_record_id = r.id
WHERE r.user_id = $1
ORDER BY r.redeemed_at DESC
LIMIT 100
"#,
)
.bind(user_id)
.fetch_all(&state.db)
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "查询兑换记录失败").with_source(err))?;
Ok(Json(Envelope {
success: true,
data: UserRedemptionsResponse { redemptions },
}))
}
fn generate_code() -> String {
let mut random = [0u8; 16];
rand::rngs::OsRng.fill_bytes(&mut random);
let payload = random
.iter()
.map(|byte| CODE_ALPHABET[usize::from(*byte & 31)] as char)
.collect::<String>();
format!(
"{CODE_PREFIX}-{}-{}-{}-{}",
&payload[0..4],
&payload[4..8],
&payload[8..12],
&payload[12..16]
)
}
fn normalize_code(input: &str) -> Option<String> {
let compact = input
.chars()
.filter(|ch| *ch != '-' && !ch.is_ascii_whitespace())
.collect::<String>()
.to_ascii_uppercase();
if compact.len() != CODE_PREFIX.len() + 16 || !compact.starts_with(CODE_PREFIX) {
return None;
}
if !compact[CODE_PREFIX.len()..]
.bytes()
.all(|byte| CODE_ALPHABET.contains(&byte))
{
return None;
}
Some(compact)
}
fn code_hint(compact: &str) -> String {
format!("{}-...-{}", CODE_PREFIX, &compact[15..19])
}
async fn enforce_redeem_rate_limit(state: &AppState, user_id: Uuid) -> Result<(), AppError> {
let key = format!(
"rate:redemption:{}:{}",
user_id,
Utc::now().format("%Y%m%d%H%M")
);
let mut redis = state.redis.clone();
let count: i64 = redis::cmd("INCR")
.arg(&key)
.query_async(&mut redis)
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "Redis 限流失败").with_source(err))?;
if count == 1 {
let _: () = redis::cmd("EXPIRE")
.arg(&key)
.arg(60)
.query_async(&mut redis)
.await
.unwrap_or(());
}
if count > 20 {
return Err(AppError::new(
ErrorCode::RateLimited,
"兑换尝试过于频繁，请稍后再试",
));
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn generated_code_round_trips_through_normalization() {
let code = generate_code();
let compact = normalize_code(&code).unwrap();
assert_eq!(compact.len(), 19);
assert!(code.starts_with("IMG-"));
}
#[test]
fn normalization_accepts_lowercase_and_spaces() {
assert_eq!(
normalize_code(" img-abcd-efgh-jkmn-pqrs "),
Some("IMGABCDEFGHJKMNPQRS".to_string())
);
}
#[test]
fn normalization_rejects_ambiguous_characters() {
assert!(normalize_code("IMG-ABCI-EFGH-JKMN-PQRS").is_none());
assert!(normalize_code("IMG-ABCO-EFGH-JKMN-PQRS").is_none());
}
#[test]
fn code_hint_only_exposes_the_last_group() {
assert_eq!(code_hint("IMGABCDEFGHJKMNPQRS"), "IMG-...-PQRS");
}
}

37
src/api/tasks.rs
View File

@@ -6,6 +6,7 @@ use crate::services::billing::{BillingContext, Plan};
use crate::services::compress;
use crate::services::compress::{CompressionLevel, ImageFmt};
use crate::services::idempotency;
use crate::services::quota;
use crate::services::storage;
use crate::state::AppState;
@@ -647,41 +648,7 @@ async fn ensure_quota_available(
ctx: &BillingContext,
needed_units: i32,
) -> Result<(), AppError> {
if needed_units <= 0 {
return Ok(());
}
#[derive(Debug, FromRow)]
struct UsageRow {
used_units: i32,
bonus_units: i32,
}
let usage = sqlx::query_as::<_, UsageRow>(
r#"
SELECT used_units, bonus_units
FROM usage_periods
WHERE user_id = $1 AND period_start = $2 AND period_end = $3
"#,
)
.bind(ctx.user_id)
.bind(ctx.period_start)
.bind(ctx.period_end)
.fetch_optional(&state.db)
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "查询用量失败").with_source(err))?
.unwrap_or(UsageRow {
used_units: 0,
bonus_units: 0,
});
let total_units = ctx.plan.included_units_per_period + usage.bonus_units;
let remaining = total_units - usage.used_units;
if remaining < needed_units {
return Err(AppError::new(ErrorCode::QuotaExceeded, "当期配额已用完"));
}
Ok(())
quota::ensure_user_units(state, ctx, needed_units).await
}
async fn anonymous_remaining_units(

18
src/api/user.rs
View File

@@ -3,6 +3,7 @@ use crate::api::envelope::Envelope;
use crate::error::{AppError, ErrorCode};
use crate::services::billing;
use crate::services::mail;
use crate::services::settings;
use crate::state::AppState;
use argon2::{Argon2, PasswordHash, PasswordHasher, PasswordVerifier};
@@ -98,6 +99,8 @@ async fn get_profile(
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "查询用户失败").with_source(err))?;
let verification_required = settings::email_verification_required(&state).await?;
Ok(Json(Envelope {
success: true,
data: UserView {
@@ -105,7 +108,7 @@ async fn get_profile(
email: user.email,
username: user.username,
role: user.role,
email_verified: user.email_verified_at.is_some(),
email_verified: user.email_verified_at.is_some() || !verification_required,
},
}))
}
@@ -165,6 +168,7 @@ async fn update_profile(
let mut next_email = user.email.clone();
let mut next_username = user.username.clone();
let mut email_changed = false;
let verification_required = settings::email_verification_required(&state).await?;
if let Some(email) = req.email.as_ref() {
let email = email.trim().to_lowercase();
@@ -192,7 +196,7 @@ async fn update_profile(
email: user.email,
username: user.username,
role: user.role,
email_verified: user.email_verified_at.is_some(),
email_verified: user.email_verified_at.is_some() || !verification_required,
},
message: "暂无更新".to_string(),
},
@@ -205,8 +209,10 @@ async fn update_profile(
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "开启事务失败").with_source(err))?;
let email_verified_at = if email_changed {
let email_verified_at = if email_changed && verification_required {
None
} else if email_changed {
Some(Utc::now())
} else {
user.email_verified_at
};
@@ -231,7 +237,7 @@ async fn update_profile(
.map_err(map_unique_violation)?;
let mut verification_link: Option<String> = None;
if email_changed {
if email_changed && verification_required {
let token = generate_token();
let token_hash = sha256_hex(&token);
let expires_at = Utc::now() + Duration::hours(24);
@@ -269,7 +275,7 @@ async fn update_profile(
})?;
}
let message = if email_changed {
let message = if email_changed && verification_required {
"资料已更新，请验证新邮箱".to_string()
} else {
"资料已更新".to_string()
@@ -283,7 +289,7 @@ async fn update_profile(
email: updated.email,
username: updated.username,
role: updated.role,
email_verified: updated.email_verified_at.is_some(),
email_verified: updated.email_verified_at.is_some() || !verification_required,
},
message,
},

2
src/services/billing.rs
View File

@@ -48,6 +48,8 @@ pub async fn get_user_billing(state: &AppState, user_id: Uuid) -> Result<Billing
FROM subscriptions
WHERE user_id = $1
AND status IN ('active', 'trialing', 'past_due')
AND current_period_start <= NOW()
AND current_period_end > NOW()
ORDER BY current_period_end DESC
LIMIT 1
"#,

46
src/services/bootstrap.rs
View File

@@ -12,44 +12,65 @@ static MIGRATOR: sqlx::migrate::Migrator = sqlx::migrate!("./migrations");
#[derive(Debug, FromRow)]
struct AdminRow {
id: Uuid,
email: String,
username: String,
role: String,
}
pub async fn ensure_admin_user(state: &AppState) -> Result<(), AppError> {
let Some(admin_email) = env_string("ADMIN_EMAIL") else {
return Ok(());
};
let Some(admin_password) = env_string("ADMIN_PASSWORD") else {
return Ok(());
};
let configured_email = env_string("ADMIN_EMAIL");
let configured_username = env_string("ADMIN_USERNAME");
if configured_email.is_none() && configured_username.is_none() {
return Ok(());
}
let admin_username = configured_username.unwrap_or_else(|| {
configured_email
.as_deref()
.and_then(|email| email.split('@').next())
.unwrap_or("admin")
.to_string()
});
let admin_username = admin_username.trim().to_string();
let admin_email = configured_email
.unwrap_or_else(|| format!("{}@local.invalid", admin_username.to_ascii_lowercase()));
let admin_email = admin_email.trim().to_lowercase();
let admin_password = admin_password.trim().to_string();
if admin_email.is_empty() || admin_password.is_empty() {
return Ok(());
}
let admin_username = env_string("ADMIN_USERNAME")
.unwrap_or_else(|| admin_email.split('@').next().unwrap_or("admin").to_string());
let admin_username = admin_username.trim().to_string();
validate_email(&admin_email)?;
validate_username(&admin_username)?;
validate_password(&admin_password)?;
let existing = sqlx::query_as::<_, AdminRow>(
let mut matching = sqlx::query_as::<_, AdminRow>(
r#"
SELECT id, username, role::text AS role
SELECT id, email, username, role::text AS role
FROM users
WHERE email = $1
WHERE email = $1 OR username = $2
ORDER BY (email = $1) DESC
LIMIT 2
"#,
)
.bind(&admin_email)
.fetch_optional(&state.db)
.bind(&admin_username)
.fetch_all(&state.db)
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "查询管理员账号失败").with_source(err))?;
if matching.len() > 1 {
return Err(AppError::new(
ErrorCode::InvalidRequest,
"管理员邮箱和用户名分别属于不同账号",
));
}
let existing = matching.pop();
let password_hash = hash_password(&admin_password)?;
if let Some(row) = existing {
@@ -113,6 +134,9 @@ pub async fn ensure_admin_user(state: &AppState) -> Result<(), AppError> {
if row.role != "admin" {
info!(admin_email = %admin_email, "管理员权限已启用");
}
if row.email != admin_email {
info!(admin_username = %admin_username, "按用户名匹配到已有管理员，保留原邮箱");
}
} else {
sqlx::query(
r#"

280
src/services/quota.rs
View File

@@ -1,8 +1,247 @@
use crate::error::{AppError, ErrorCode};
use crate::services::billing::BillingContext;
use crate::state::AppState;
use chrono::{Duration, Utc};
use chrono::{DateTime, Duration, Utc};
use sqlx::{FromRow, Postgres, Transaction};
use std::net::IpAddr;
use uuid::Uuid;
#[derive(Debug, Clone, Copy)]
pub struct UserUsageBalance {
pub used_units: i64,
pub included_units: i64,
pub bonus_units: i64,
pub redeemed_units: i64,
pub total_units: i64,
pub remaining_units: i64,
}
#[derive(Debug, FromRow)]
struct UsagePeriodRow {
used_units: i32,
bonus_units: i32,
grant_used_units: i32,
}
#[derive(Debug, FromRow)]
struct AvailableGrantRow {
id: Uuid,
expires_at: DateTime<Utc>,
}
pub async fn user_usage_balance(
state: &AppState,
billing: &BillingContext,
) -> Result<UserUsageBalance, AppError> {
let usage = sqlx::query_as::<_, UsagePeriodRow>(
r#"
SELECT used_units, bonus_units, grant_used_units
FROM usage_periods
WHERE user_id = $1 AND period_start = $2 AND period_end = $3
"#,
)
.bind(billing.user_id)
.bind(billing.period_start)
.bind(billing.period_end)
.fetch_optional(&state.db)
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "查询用量失败").with_source(err))?
.unwrap_or(UsagePeriodRow {
used_units: 0,
bonus_units: 0,
grant_used_units: 0,
});
let redeemed_units: i64 = sqlx::query_scalar(
r#"
SELECT COALESCE(SUM(remaining_units), 0)::bigint
FROM unit_grants
WHERE user_id = $1
AND starts_at <= NOW()
AND expires_at > NOW()
AND remaining_units > 0
"#,
)
.bind(billing.user_id)
.fetch_one(&state.db)
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "查询兑换额度失败").with_source(err))?;
Ok(calculate_user_balance(
billing.plan.included_units_per_period,
usage.used_units,
usage.bonus_units,
usage.grant_used_units,
redeemed_units,
))
}
pub async fn ensure_user_units(
state: &AppState,
billing: &BillingContext,
needed_units: i32,
) -> Result<(), AppError> {
if needed_units <= 0 {
return Ok(());
}
let balance = user_usage_balance(state, billing).await?;
if balance.remaining_units < i64::from(needed_units) {
return Err(AppError::new(ErrorCode::QuotaExceeded, "可用配额已用完"));
}
Ok(())
}
pub async fn consume_user_unit(
tx: &mut Transaction<'_, Postgres>,
billing: &BillingContext,
bytes_in: u64,
bytes_out: u64,
) -> Result<(), AppError> {
sqlx::query(
r#"
INSERT INTO usage_periods (user_id, subscription_id, period_start, period_end)
VALUES ($1, $2, $3, $4)
ON CONFLICT (user_id, period_start, period_end) DO NOTHING
"#,
)
.bind(billing.user_id)
.bind(billing.subscription_id)
.bind(billing.period_start)
.bind(billing.period_end)
.execute(&mut **tx)
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "初始化用量周期失败").with_source(err))?;
let usage = sqlx::query_as::<_, UsagePeriodRow>(
r#"
SELECT used_units, bonus_units, grant_used_units
FROM usage_periods
WHERE user_id = $1 AND period_start = $2 AND period_end = $3
FOR UPDATE
"#,
)
.bind(billing.user_id)
.bind(billing.period_start)
.bind(billing.period_end)
.fetch_one(&mut **tx)
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "锁定用量周期失败").with_source(err))?;
let plan_used = usage.used_units.saturating_sub(usage.grant_used_units);
let plan_capacity = billing
.plan
.included_units_per_period
.saturating_add(usage.bonus_units);
let plan_available = plan_used < plan_capacity;
let grant = sqlx::query_as::<_, AvailableGrantRow>(
r#"
SELECT id, expires_at
FROM unit_grants
WHERE user_id = $1
AND starts_at <= NOW()
AND expires_at > NOW()
AND remaining_units > 0
ORDER BY expires_at ASC, created_at ASC
FOR UPDATE
LIMIT 1
"#,
)
.bind(billing.user_id)
.fetch_optional(&mut **tx)
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "锁定兑换额度失败").with_source(err))?;
let use_grant = should_consume_grant(
plan_available,
grant.as_ref().map(|grant| grant.expires_at),
billing.period_end,
);
if !plan_available && !use_grant {
return Err(AppError::new(ErrorCode::QuotaExceeded, "可用配额已用完"));
}
if use_grant {
let grant_id = grant
.ok_or_else(|| AppError::new(ErrorCode::QuotaExceeded, "可用配额已用完"))?
.id;
let updated = sqlx::query(
r#"
UPDATE unit_grants
SET remaining_units = remaining_units - 1,
updated_at = NOW()
WHERE id = $1
AND remaining_units > 0
AND expires_at > NOW()
"#,
)
.bind(grant_id)
.execute(&mut **tx)
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "扣减兑换额度失败").with_source(err))?;
if updated.rows_affected() != 1 {
return Err(AppError::new(ErrorCode::QuotaExceeded, "可用配额已用完"));
}
}
sqlx::query(
r#"
UPDATE usage_periods
SET used_units = used_units + 1,
grant_used_units = grant_used_units + $1,
bytes_in = bytes_in + $2,
bytes_out = bytes_out + $3,
updated_at = NOW()
WHERE user_id = $4 AND period_start = $5 AND period_end = $6
"#,
)
.bind(use_grant as i32)
.bind(bytes_in as i64)
.bind(bytes_out as i64)
.bind(billing.user_id)
.bind(billing.period_start)
.bind(billing.period_end)
.execute(&mut **tx)
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "记录用量失败").with_source(err))?;
Ok(())
}
fn calculate_user_balance(
included_units: i32,
used_units: i32,
bonus_units: i32,
grant_used_units: i32,
redeemed_units: i64,
) -> UserUsageBalance {
let base_capacity = i64::from(included_units.saturating_add(bonus_units));
let base_used = i64::from(used_units.saturating_sub(grant_used_units));
let base_remaining = base_capacity.saturating_sub(base_used).max(0);
let remaining_units = base_remaining.saturating_add(redeemed_units.max(0));
let used_units = i64::from(used_units.max(0));
UserUsageBalance {
used_units,
included_units: i64::from(included_units),
bonus_units: i64::from(bonus_units),
redeemed_units: redeemed_units.max(0),
total_units: used_units.saturating_add(remaining_units),
remaining_units,
}
}
fn should_consume_grant(
plan_available: bool,
grant_expires_at: Option<DateTime<Utc>>,
plan_expires_at: DateTime<Utc>,
) -> bool {
grant_expires_at.is_some_and(|expires_at| !plan_available || expires_at <= plan_expires_at)
}
pub async fn consume_anonymous_units(
state: &AppState,
@@ -71,3 +310,42 @@ fn utc8_date() -> String {
let now = Utc::now() + Duration::hours(8);
now.format("%Y-%m-%d").to_string()
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn balance_keeps_redeemed_units_separate_from_plan_usage() {
let balance = calculate_user_balance(10, 13, 0, 3, 7);
assert_eq!(balance.remaining_units, 7);
assert_eq!(balance.total_units, 20);
}
#[test]
fn balance_uses_plan_capacity_before_redeemed_units() {
let balance = calculate_user_balance(10, 4, 2, 0, 5);
assert_eq!(balance.remaining_units, 13);
assert_eq!(balance.total_units, 17);
}
#[test]
fn earlier_expiring_entitlement_is_consumed_first() {
let now = Utc::now();
assert!(should_consume_grant(
true,
Some(now + Duration::days(2)),
now + Duration::days(20),
));
assert!(!should_consume_grant(
true,
Some(now + Duration::days(30)),
now + Duration::days(20),
));
assert!(should_consume_grant(
false,
Some(now + Duration::days(30)),
now + Duration::days(20),
));
}
}

28
src/services/settings.rs
View File

@@ -28,6 +28,16 @@ pub struct MailConfigStored {
pub log_links_when_disabled: Option<bool>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AuthConfigStored {
#[serde(default = "default_true")]
pub email_verification_required: bool,
}
fn default_true() -> bool {
true
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct StripeConfigStored {
pub secret_key_encrypted: Option<String>,
@@ -117,6 +127,13 @@ pub async fn load_mail_settings(state: &AppState) -> Result<Option<MailSettings>
}))
}
pub async fn email_verification_required(state: &AppState) -> Result<bool, AppError> {
Ok(load_system_config::<AuthConfigStored>(state, "auth")
.await?
.map(|config| config.email_verification_required)
.unwrap_or(true))
}
pub async fn load_stripe_secrets(state: &AppState) -> Result<Option<StripeSecrets>, AppError> {
let Some(cfg) = load_system_config::<StripeConfigStored>(state, "stripe").await? else {
return Ok(None);
@@ -225,3 +242,14 @@ pub async fn get_stripe_webhook_secret(state: &AppState) -> Result<String, AppEr
.filter(|v| !v.trim().is_empty())
.ok_or_else(|| AppError::new(ErrorCode::InvalidRequest, "未配置 Stripe Webhook Secret"))
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn auth_config_defaults_to_requiring_verification() {
let config: AuthConfigStored = serde_json::from_value(serde_json::json!({})).unwrap();
assert!(config.email_verification_required);
}
}

44
src/worker/mod.rs
View File

@@ -791,49 +791,7 @@ async fn charge_one_unit(
bytes_in: u64,
bytes_out: u64,
) -> Result<(), AppError> {
// Ensure usage period row exists.
sqlx::query(
r#"
INSERT INTO usage_periods (user_id, subscription_id, period_start, period_end)
VALUES ($1, $2, $3, $4)
ON CONFLICT (user_id, period_start, period_end) DO NOTHING
"#,
)
.bind(billing.user_id)
.bind(billing.subscription_id)
.bind(billing.period_start)
.bind(billing.period_end)
.execute(&mut **tx)
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "初始化用量周期失败").with_source(err))?;
let updated: Option<i32> = sqlx::query_scalar(
r#"
UPDATE usage_periods
SET used_units = used_units + 1,
bytes_in = bytes_in + $1,
bytes_out = bytes_out + $2,
updated_at = NOW()
WHERE user_id = $3
AND period_start = $4
AND period_end = $5
AND used_units + 1 <= $6 + bonus_units
RETURNING used_units
"#,
)
.bind(bytes_in as i64)
.bind(bytes_out as i64)
.bind(billing.user_id)
.bind(billing.period_start)
.bind(billing.period_end)
.bind(billing.plan.included_units_per_period)
.fetch_optional(&mut **tx)
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "扣减配额失败").with_source(err))?;
if updated.is_none() {
return Err(AppError::new(ErrorCode::QuotaExceeded, "当期配额已用完"));
}
quota::consume_user_unit(tx, billing, bytes_in, bytes_out).await?;
sqlx::query(
r#"