milky0217
3cce81321f
- auth.rs 新增 JWT 生成/验证回环、篡改拒绝、refresh token 随机性、临时 token 有效期测试 - payment.rs 新增套餐定义、URL 解码、支付宝表单解析（UTF-8/GBK）、RSA2 签名验签、UA 检测测试 - 修复 urlencoding 两个真实 bug：UTF-8 多字节序列乱码、不完整转义吞字符 - 移除 db.rs 悬空 doc 注释，修复 auth.rs 冗余引用（clippy 0 警告） - cargo test 从 2 个增至 36 个全过 - 同步 IMPROVEMENTS.md（索引/重复查询/测试覆盖/公告系统状态）
1302 lines
43 KiB
Rust
1302 lines
43 KiB
Rust
use sqlx::{PgPool, postgres::PgPoolOptions, FromRow};
|
||
use std::env;
|
||
use std::error::Error;
|
||
use chrono::Utc;
|
||
|
||
// 从 models 模块引入 WeatherData 结构体
|
||
use crate::models::{User, WeatherData, WeatherDataBrief, WeatherListResponse};
|
||
use crate::error::AppError;
|
||
|
||
// 用于插入weather_data的数据
|
||
pub async fn insert_weather_data(pool: &PgPool, weather_data: &WeatherData, user_id: i32) -> Result<i32, AppError> {
|
||
// 单次查询获取用户会员状态和数据条数（原 N+1：get_user_by_id + count_user_weather_data → 合并为 1 次）
|
||
#[derive(FromRow)]
|
||
struct UserQuota {
|
||
is_member: bool,
|
||
#[sqlx(rename = "membership_expires_at")]
|
||
membership_expires_at: Option<chrono::DateTime<Utc>>,
|
||
data_count: i64,
|
||
}
|
||
|
||
let quota = sqlx::query_as::<_, UserQuota>(
|
||
r#"
|
||
SELECT
|
||
u.is_member,
|
||
u.membership_expires_at,
|
||
(SELECT COUNT(*) FROM weather_data w WHERE w.user_id = $1)::BIGINT AS data_count
|
||
FROM users u
|
||
WHERE u.id = $1
|
||
"#,
|
||
)
|
||
.bind(user_id)
|
||
.fetch_optional(pool)
|
||
.await
|
||
.map_err(|e| AppError::Database(format!("查询用户配额失败: {}", e)))?
|
||
.ok_or_else(|| AppError::NotFound("用户不存在".to_string()))?;
|
||
|
||
let is_active_member = quota.is_member && quota.membership_expires_at.is_none_or(|expires| expires > Utc::now());
|
||
|
||
// 维护模式：非会员使用更高的临时限额，防止资源滥用
|
||
let is_maintenance = std::env::var("PAYMENT_MAINTENANCE_MODE").ok() == Some("true".to_string());
|
||
|
||
if !is_active_member {
|
||
let limit: i64 = if is_maintenance {
|
||
env::var("MAINTENANCE_MODE_DATA_LIMIT")
|
||
.unwrap_or_else(|_| "500".to_string())
|
||
.parse()
|
||
.unwrap_or(500)
|
||
} else {
|
||
env::var("FREE_USER_DATA_LIMIT")
|
||
.unwrap_or_else(|_| "20".to_string())
|
||
.parse()
|
||
.unwrap_or(20)
|
||
};
|
||
|
||
if quota.data_count >= limit {
|
||
return Err(AppError::Forbidden("数据条数已达上限，请升级为付费用户".to_string()));
|
||
}
|
||
}
|
||
|
||
// 准备插入数据的 SQL 语句
|
||
let insert_query = r#"
|
||
INSERT INTO weather_data (
|
||
user_id, title, date, hour, min, longitude, latitude, daysincejanfirst,
|
||
theta, solardeclination, sunaltitude, overallcloudiness, lowcloudiness,
|
||
cloudindex, solarradiationlevel, hasmeasuredwindspeed, measuredwindspeed,
|
||
convertedwindspeed, pointheight, areatype, pointwindspeed,
|
||
atmosphericstability, suitabilitydegree, winddirection, averagewinddirection,
|
||
winddirectionstandarddeviation, windspeed, averagewindspeed, windspeedsuitability,
|
||
winddirectionsuitability, overallsuitability, inspectiontype, assignmentnumber,
|
||
calculatedwindspeed, hasspotcheckwindspeed, spotcheckcount, version, is_favorite
|
||
) VALUES (
|
||
$1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19,
|
||
$20, $21, $22, $23, $24, $25, $26, $27, $28, $29, $30, $31, $32, $33, $34, $35, $36, $37, $38
|
||
) RETURNING id
|
||
"#;
|
||
|
||
// 3. 执行插入操作
|
||
let inserted_id = match sqlx::query_as::<_, (i32,)>(insert_query)
|
||
.bind(user_id)
|
||
.bind(&weather_data.title)
|
||
.bind(weather_data.date)
|
||
.bind(weather_data.hours)
|
||
.bind(weather_data.min)
|
||
.bind(&weather_data.longitude)
|
||
.bind(&weather_data.latitude)
|
||
.bind(weather_data.day_since_jan_first)
|
||
.bind(weather_data.theta)
|
||
.bind(weather_data.solar_declination)
|
||
.bind(weather_data.sun_altitude)
|
||
.bind(&weather_data.overall_cloudiness)
|
||
.bind(&weather_data.low_cloudiness)
|
||
.bind(&weather_data.cloud_index)
|
||
.bind(weather_data.solar_radiation_level)
|
||
.bind(weather_data.has_measured_wind_speed)
|
||
.bind(weather_data.measured_wind_speed) // <-- 修改: 直接绑定 Option<f64>
|
||
.bind(weather_data.converted_wind_speed) // <-- 修改: 直接绑定 Option<i32>，并移除 `as i32`
|
||
.bind(weather_data.point_height) // <-- 修改: 直接绑定 Option<f64>
|
||
.bind(&weather_data.area_type)
|
||
.bind(weather_data.point_wind_speed) // <-- 修改: 直接绑定 Option<f64>
|
||
.bind(&weather_data.atmospheric_stability)
|
||
.bind(&weather_data.suitability_degree)
|
||
.bind(
|
||
serde_json::to_value(&weather_data.wind_direction)
|
||
.map_err(|e| AppError::Internal(format!("JSON 序列化失败: {}", e)))?,
|
||
)
|
||
.bind(weather_data.average_wind_direction)
|
||
.bind(weather_data.wind_direction_standard_deviation)
|
||
.bind(
|
||
serde_json::to_value(&weather_data.wind_speed)
|
||
.map_err(|e| AppError::Internal(format!("JSON 序列化失败: {}", e)))?,
|
||
)
|
||
.bind(weather_data.average_wind_speed)
|
||
.bind(&weather_data.wind_speed_suitability)
|
||
.bind(&weather_data.wind_direction_suitability)
|
||
.bind(&weather_data.overall_suitability)
|
||
.bind(&weather_data.inspection_type)
|
||
.bind(&weather_data.assignment_number)
|
||
.bind(weather_data.calculated_wind_speed) // 这个本来就是 Option，无需修改
|
||
.bind(weather_data.has_spot_check_wind_speed)
|
||
.bind(weather_data.spot_check_count)
|
||
.bind(&weather_data.version)
|
||
.bind(weather_data.is_favorite)
|
||
.fetch_one(pool)
|
||
.await
|
||
{
|
||
Ok((id,)) => id,
|
||
Err(e) => {
|
||
return Err(AppError::Database(format!("插入数据失败: {}", e)));
|
||
}
|
||
};
|
||
|
||
// 4. 成功，返回插入的 ID
|
||
Ok(inserted_id)
|
||
}
|
||
|
||
pub async fn create_pool() -> Result<PgPool, Box<dyn Error>> {
|
||
// 获取数据库URL并提供友好错误信息
|
||
let database_url =
|
||
env::var("DATABASE_URL").map_err(|_| "环境变量DATABASE_URL未设置，请在.env文件中配置")?;
|
||
|
||
// 配置连接池参数
|
||
let pool = PgPoolOptions::new()
|
||
.max_connections(20) // 根据应用需求调整最大连接数
|
||
.acquire_timeout(std::time::Duration::from_secs(30)) // 获取连接的超时时间
|
||
.connect(&database_url)
|
||
.await
|
||
.map_err(|e| format!("数据库连接失败: {}", e))?;
|
||
|
||
Ok(pool)
|
||
}
|
||
|
||
// 获取天气数据详情
|
||
pub async fn get_weather_details(pool: &PgPool, weather_id: i32) -> Result<WeatherData, AppError> {
|
||
let query = r#"
|
||
SELECT
|
||
wd.id, wd.title, wd.date, wd.hour, wd.min,
|
||
wd.longitude, wd.latitude,
|
||
wd.daysincejanfirst, wd.theta, wd.solardeclination,
|
||
wd.sunaltitude, wd.overallcloudiness, wd.lowcloudiness,
|
||
wd.cloudindex, wd.solarradiationlevel, wd.hasmeasuredwindspeed,
|
||
wd.measuredwindspeed, wd.convertedwindspeed, wd.pointheight,
|
||
wd.areatype, wd.pointwindspeed, wd.atmosphericstability,
|
||
wd.suitabilitydegree, wd.winddirection, wd.averagewinddirection,
|
||
wd.winddirectionstandarddeviation, wd.windspeed, wd.averagewindspeed,
|
||
wd.windspeedsuitability, wd.winddirectionsuitability, wd.overallsuitability,
|
||
wd.inspectiontype, wd.assignmentnumber, wd.calculatedwindspeed, wd.hasspotcheckwindspeed,
|
||
wd.spotcheckcount, wd.is_favorite, u.openid, wd.version
|
||
FROM weather_data wd
|
||
JOIN users u ON wd.user_id = u.id
|
||
WHERE wd.id = $1
|
||
"#;
|
||
|
||
let row = match sqlx::query_as::<_, WeatherData>(query)
|
||
.bind(weather_id)
|
||
.fetch_optional(pool)
|
||
.await
|
||
{
|
||
Ok(Some(row)) => row,
|
||
Ok(None) => return Err(AppError::NotFound(format!("未找到ID为 {} 的天气数据", weather_id))),
|
||
Err(e) => return Err(AppError::Database(format!("查询天气数据失败: {}", e))),
|
||
};
|
||
|
||
Ok(row)
|
||
}
|
||
|
||
// get_weather_list函数
|
||
pub async fn get_weather_list(
|
||
pool: &PgPool,
|
||
user_id: i32,
|
||
page: i32,
|
||
limit: i32,
|
||
) -> Result<WeatherListResponse, AppError> {
|
||
let offset = (page - 1) * limit;
|
||
|
||
// 1. 查询符合条件的总条数
|
||
let total_query = r#"
|
||
SELECT COUNT(*) as total
|
||
FROM weather_data
|
||
WHERE user_id = $1
|
||
"#;
|
||
let total = match sqlx::query_as::<_, (i64,)>(total_query)
|
||
.bind(user_id)
|
||
.fetch_one(pool)
|
||
.await
|
||
{
|
||
Ok((count,)) => count,
|
||
Err(e) => return Err(AppError::Database(format!("查询总条数失败: {}", e))),
|
||
};
|
||
|
||
// 2. 查询当前页数据列表
|
||
let list_query = r#"
|
||
SELECT
|
||
wd.id, wd.title, wd.date, wd.hour, wd.min, wd.longitude, wd.latitude, wd.is_favorite
|
||
FROM weather_data wd
|
||
WHERE wd.user_id = $1
|
||
ORDER BY wd.date DESC, wd.hour DESC, wd.min DESC
|
||
LIMIT $2 OFFSET $3
|
||
"#;
|
||
let list = match sqlx::query_as::<_, WeatherDataBrief>(list_query)
|
||
.bind(user_id)
|
||
.bind(limit)
|
||
.bind(offset)
|
||
.fetch_all(pool)
|
||
.await
|
||
{
|
||
Ok(data) => data,
|
||
Err(e) => return Err(AppError::Database(format!("查询天气数据列表失败: {}", e))),
|
||
};
|
||
|
||
// 3. 包装结果并返回
|
||
Ok(WeatherListResponse { list, total })
|
||
}
|
||
|
||
pub async fn delete_weather_data(
|
||
pool: &PgPool,
|
||
weather_id: i32,
|
||
user_id: i32,
|
||
) -> Result<(), AppError> {
|
||
let query = r#"
|
||
DELETE FROM weather_data
|
||
WHERE id = $1 AND user_id = $2
|
||
"#;
|
||
|
||
let result = sqlx::query(query)
|
||
.bind(weather_id)
|
||
.bind(user_id)
|
||
.execute(pool)
|
||
.await
|
||
.map_err(|e| AppError::Database(format!("删除天气数据失败: {}", e)))?;
|
||
|
||
if result.rows_affected() == 0 {
|
||
return Err(AppError::NotFound(format!("未找到ID为 {} 的天气数据或无权限删除", weather_id)));
|
||
}
|
||
|
||
Ok(())
|
||
}
|
||
|
||
// 根据用户ID获取用户信息
|
||
pub async fn get_user_by_id(pool: &PgPool, user_id: i32) -> Result<User, AppError> {
|
||
let query = r#"
|
||
SELECT
|
||
id, name, openid, phone, type, "desc", is_member, is_admin, membership_expires_at,
|
||
avatar_url, nickname
|
||
FROM users
|
||
WHERE id = $1
|
||
"#;
|
||
|
||
let row = match sqlx::query_as::<_, User>(query)
|
||
.bind(user_id)
|
||
.fetch_optional(pool)
|
||
.await
|
||
{
|
||
Ok(Some(row)) => row,
|
||
Ok(None) => return Err(AppError::NotFound(format!("未找到ID为 {} 的用户", user_id))),
|
||
Err(e) => return Err(AppError::Database(format!("查询用户信息失败: {}", e))),
|
||
};
|
||
|
||
Ok(row)
|
||
}
|
||
|
||
// 统计用户的天气数据条数
|
||
pub async fn count_user_weather_data(pool: &PgPool, user_id: i32) -> Result<i64, AppError> {
|
||
let query = r#"
|
||
SELECT COUNT(*) FROM weather_data WHERE user_id = $1
|
||
"#;
|
||
|
||
let count = match sqlx::query_as::<_, (i64,)>(query)
|
||
.bind(user_id)
|
||
.fetch_one(pool)
|
||
.await
|
||
{
|
||
Ok((count,)) => count,
|
||
Err(e) => return Err(AppError::Database(format!("查询天气数据条数失败: {}", e))),
|
||
};
|
||
|
||
Ok(count)
|
||
}
|
||
|
||
// 更新用户付费状态
|
||
pub async fn update_user_payment_status(
|
||
pool: &PgPool,
|
||
user_id: i32,
|
||
is_member: bool,
|
||
membership_expires_at: Option<chrono::DateTime<chrono::Utc>>,
|
||
) -> Result<(), AppError> {
|
||
let query = r#"
|
||
UPDATE users
|
||
SET is_member = $1, membership_expires_at = $2
|
||
WHERE id = $3
|
||
"#;
|
||
|
||
match sqlx::query(query)
|
||
.bind(is_member)
|
||
.bind(membership_expires_at)
|
||
.bind(user_id)
|
||
.execute(pool)
|
||
.await
|
||
{
|
||
Ok(_) => Ok(()),
|
||
Err(e) => Err(AppError::Database(format!("更新用户付费状态失败: {}", e))),
|
||
}
|
||
}
|
||
|
||
// 更新用户个人信息
|
||
pub async fn update_user_profile(
|
||
pool: &PgPool,
|
||
user_id: i32,
|
||
nickname: &Option<String>,
|
||
avatar_url: &Option<String>,
|
||
) -> Result<(), AppError> {
|
||
let query = r#"
|
||
UPDATE users
|
||
SET nickname = COALESCE($1, nickname),
|
||
avatar_url = COALESCE($2, avatar_url)
|
||
WHERE id = $3
|
||
"#;
|
||
|
||
match sqlx::query(query)
|
||
.bind(nickname)
|
||
.bind(avatar_url)
|
||
.bind(user_id)
|
||
.execute(pool)
|
||
.await
|
||
{
|
||
Ok(_) => Ok(()),
|
||
Err(e) => Err(AppError::Database(format!("更新用户个人信息失败: {}", e))),
|
||
}
|
||
}
|
||
|
||
// ===== 支付系统 DB 函数 =====
|
||
|
||
/// 创建待支付订单
|
||
pub async fn get_user_orders(
|
||
pool: &PgPool,
|
||
user_id: i32,
|
||
) -> Result<Vec<crate::models::PaymentOrder>, AppError> {
|
||
let rows = sqlx::query_as::<_, crate::models::PaymentOrder>(
|
||
r#"
|
||
SELECT id, user_id, order_no, package_type, amount, status, paid_at, expires_at, created_at
|
||
FROM payment_orders
|
||
WHERE user_id = $1
|
||
ORDER BY created_at DESC
|
||
"#,
|
||
)
|
||
.bind(user_id)
|
||
.fetch_all(pool)
|
||
.await
|
||
.map_err(|e| AppError::Database(format!("查询订单失败: {}", e)))?;
|
||
|
||
Ok(rows)
|
||
}
|
||
|
||
/// 检查会员到期前 7 天的用户，发送即将到期通知
|
||
pub async fn check_member_expiry_soon(pool: &PgPool) -> Result<u64, AppError> {
|
||
let affected = sqlx::query(
|
||
r#"
|
||
INSERT INTO notifications (scope, user_id, type, title, content)
|
||
SELECT 'user', u.id, 'member_expiry_soon',
|
||
'会员即将到期',
|
||
CONCAT('您的会员将于 ', TO_CHAR(u.membership_expires_at, 'YYYY-MM-DD'), ' 到期，请及时续费')
|
||
FROM users u
|
||
WHERE u.is_member = true
|
||
AND u.membership_expires_at IS NOT NULL
|
||
AND u.membership_expires_at > NOW()
|
||
AND u.membership_expires_at <= NOW() + INTERVAL '7 days'
|
||
AND NOT EXISTS (
|
||
SELECT 1 FROM notifications n
|
||
WHERE n.user_id = u.id
|
||
AND n.type = 'member_expiry_soon'
|
||
AND n.created_at > NOW() - INTERVAL '1 day'
|
||
)
|
||
"#,
|
||
)
|
||
.execute(pool)
|
||
.await
|
||
.map_err(|e| AppError::Database(format!("检查会员到期失败: {}", e)))?
|
||
.rows_affected();
|
||
|
||
if affected > 0 {
|
||
tracing::info!("已发送 {} 条会员到期提醒", affected);
|
||
}
|
||
Ok(affected)
|
||
}
|
||
|
||
/// 清理超过 24 小时仍未支付的待处理订单
|
||
pub async fn cleanup_expired_pending_orders(pool: &PgPool, user_id: Option<i32>) -> Result<u64, AppError> {
|
||
let affected = if let Some(uid) = user_id {
|
||
sqlx::query(
|
||
r#"
|
||
UPDATE payment_orders
|
||
SET status = 'cancelled'
|
||
WHERE status = 'pending'
|
||
AND created_at < NOW() - INTERVAL '24 hours'
|
||
AND user_id = $1
|
||
"#,
|
||
)
|
||
.bind(uid)
|
||
.execute(pool)
|
||
.await
|
||
.map_err(|e| AppError::Database(format!("清理过期订单失败: {}", e)))?
|
||
.rows_affected()
|
||
} else {
|
||
sqlx::query(
|
||
r#"
|
||
UPDATE payment_orders
|
||
SET status = 'cancelled'
|
||
WHERE status = 'pending'
|
||
AND created_at < NOW() - INTERVAL '24 hours'
|
||
"#,
|
||
)
|
||
.execute(pool)
|
||
.await
|
||
.map_err(|e| AppError::Database(format!("清理过期订单失败: {}", e)))?
|
||
.rows_affected()
|
||
};
|
||
|
||
if affected > 0 {
|
||
tracing::info!("已清理 {} 个过期待支付订单", affected);
|
||
}
|
||
Ok(affected)
|
||
}
|
||
|
||
pub async fn create_payment_order(
|
||
pool: &PgPool,
|
||
user_id: i32,
|
||
order_no: &str,
|
||
package_type: &str,
|
||
amount: i32,
|
||
expires_at: Option<chrono::DateTime<chrono::Utc>>,
|
||
) -> Result<(), AppError> {
|
||
// 创建前先清理该用户的过期 pending 订单
|
||
let _ = cleanup_expired_pending_orders(pool, Some(user_id)).await;
|
||
|
||
let query = r#"
|
||
INSERT INTO payment_orders (user_id, order_no, package_type, amount, expires_at)
|
||
VALUES ($1, $2, $3, $4, $5)
|
||
"#;
|
||
|
||
sqlx::query(query)
|
||
.bind(user_id)
|
||
.bind(order_no)
|
||
.bind(package_type)
|
||
.bind(amount)
|
||
.bind(expires_at)
|
||
.execute(pool)
|
||
.await
|
||
.map_err(|e| AppError::Database(format!("创建订单失败: {}", e)))?;
|
||
|
||
Ok(())
|
||
}
|
||
|
||
/// 清理所有已过期的 web 登录码
|
||
pub async fn cleanup_expired_login_codes(pool: &PgPool) -> Result<u64, AppError> {
|
||
let affected = sqlx::query("DELETE FROM web_login_codes WHERE expires_at < NOW()")
|
||
.execute(pool)
|
||
.await
|
||
.map_err(|e| AppError::Database(format!("清理过期登录码失败: {}", e)))?
|
||
.rows_affected();
|
||
|
||
if affected > 0 {
|
||
tracing::info!("已清理 {} 个过期的 web 登录码", affected);
|
||
}
|
||
Ok(affected)
|
||
}
|
||
|
||
/// 确认订单支付（模拟）：更新订单状态 + 激活用户付费
|
||
///
|
||
/// 返回该订单的 expires_at（永久会员为 None）
|
||
/// 确认订单支付（模拟）：更新订单状态 + 激活用户付费（累加有效期）
|
||
///
|
||
/// 返回计算后的到期时间（永久会员返回 None）
|
||
/// 插入支付成功通知
|
||
async fn insert_payment_notification(pool: &PgPool, user_id: i32, order_no: &str) -> Result<(), AppError> {
|
||
// 查出套餐信息
|
||
let pkg_info: Option<(String, String)> = sqlx::query_as(
|
||
r#"SELECT package_type, status FROM payment_orders WHERE order_no = $1"#,
|
||
)
|
||
.bind(order_no)
|
||
.fetch_optional(pool)
|
||
.await
|
||
.map_err(|e| AppError::Database(format!("查询订单失败: {}", e)))?;
|
||
|
||
let (pkg_type, status) = match pkg_info {
|
||
Some(info) => info,
|
||
None => return Ok(()),
|
||
};
|
||
|
||
if status != "paid" {
|
||
return Ok(());
|
||
}
|
||
|
||
let pkg_label = match pkg_type.as_str() {
|
||
"monthly" => "包月会员",
|
||
"quarterly" => "季卡会员",
|
||
"half_year" => "半年会员",
|
||
"yearly" => "包年会员",
|
||
"permanent" => "永久会员",
|
||
_ => "会员",
|
||
};
|
||
|
||
let title = format!("{}开通成功", pkg_label);
|
||
let content = format!("恭喜！您已成功开通{}", pkg_label);
|
||
|
||
sqlx::query(
|
||
r#"INSERT INTO notifications (scope, user_id, type, title, content)
|
||
VALUES ('user', $1, 'payment_success', $2, $3)"#,
|
||
)
|
||
.bind(user_id)
|
||
.bind(&title)
|
||
.bind(&content)
|
||
.execute(pool)
|
||
.await
|
||
.ok();
|
||
|
||
Ok(())
|
||
}
|
||
|
||
pub async fn confirm_payment_order(
|
||
pool: &PgPool,
|
||
order_no: &str,
|
||
user_id: i32,
|
||
) -> Result<Option<chrono::DateTime<chrono::Utc>>, AppError> {
|
||
// 检查订单有效
|
||
let row = sqlx::query_as::<_, (i32, String)>(
|
||
r#"SELECT user_id, status FROM payment_orders WHERE order_no = $1"#,
|
||
)
|
||
.bind(order_no)
|
||
.fetch_optional(pool)
|
||
.await
|
||
.map_err(|e| AppError::Database(format!("查询订单失败: {}", e)))?;
|
||
|
||
let (order_user_id, status) = match row {
|
||
Some(r) => r,
|
||
None => return Err(AppError::NotFound("订单不存在".to_string())),
|
||
};
|
||
|
||
if order_user_id != user_id {
|
||
return Err(AppError::Forbidden("无权操作此订单".to_string()));
|
||
}
|
||
|
||
if status != "pending" {
|
||
let hint = if status == "paid" {
|
||
"该订单已支付成功，无需重复操作"
|
||
} else if status == "cancelled" {
|
||
"该订单已被取消，如已扣款请联系客服处理"
|
||
} else {
|
||
"该订单状态异常，请联系客服"
|
||
};
|
||
return Err(AppError::BadRequest(hint.to_string()));
|
||
}
|
||
|
||
// 一次性完成：更新订单状态 + 累加计算新的到期时间（纯 SQL）
|
||
let new_expires: Option<chrono::DateTime<chrono::Utc>> = sqlx::query_scalar(
|
||
r#"
|
||
WITH updated_order AS (
|
||
UPDATE payment_orders SET status = 'paid', paid_at = NOW() WHERE order_no = $1 AND status = 'pending' RETURNING package_type
|
||
)
|
||
UPDATE users SET
|
||
is_member = true,
|
||
membership_expires_at =
|
||
CASE
|
||
WHEN uo.package_type = 'permanent' THEN NULL
|
||
ELSE GREATEST(COALESCE(users.membership_expires_at, NOW()), NOW()) +
|
||
CASE
|
||
WHEN uo.package_type = 'monthly' THEN INTERVAL '30 days'
|
||
WHEN uo.package_type = 'quarterly' THEN INTERVAL '90 days'
|
||
WHEN uo.package_type = 'half_year' THEN INTERVAL '182 days'
|
||
WHEN uo.package_type = 'yearly' THEN INTERVAL '365 days'
|
||
ELSE INTERVAL '0 days'
|
||
END
|
||
END
|
||
FROM updated_order uo
|
||
WHERE users.id = $2
|
||
RETURNING users.membership_expires_at
|
||
"#,
|
||
)
|
||
.bind(order_no)
|
||
.bind(user_id)
|
||
.fetch_one(pool)
|
||
.await
|
||
.map_err(|e| AppError::Database(format!("支付确认失败: {}", e)))?;
|
||
|
||
// 发送支付成功通知
|
||
let _ = insert_payment_notification(pool, user_id, order_no).await;
|
||
|
||
// 验证会员到期时间是否正常
|
||
if let Err(e) = verify_membership_after_payment(pool, order_no).await {
|
||
tracing::warn!("会员到期时间验证异常 (订单 {}): {}", order_no, e);
|
||
|
||
}
|
||
|
||
Ok(new_expires)
|
||
}
|
||
|
||
/// 确认订单支付（支付宝异步回调用，通过 order_no 查找，不校验 user_id）
|
||
pub async fn confirm_payment_order_by_orderno(
|
||
pool: &PgPool,
|
||
order_no: &str,
|
||
) -> Result<(), AppError> {
|
||
// 检查订单有效
|
||
let status: Option<String> = sqlx::query_scalar(
|
||
r#"SELECT status FROM payment_orders WHERE order_no = $1"#,
|
||
)
|
||
.bind(order_no)
|
||
.fetch_optional(pool)
|
||
.await
|
||
.map_err(|e| AppError::Database(format!("查询订单失败: {}", e)))?;
|
||
|
||
let status = match status {
|
||
Some(s) => s,
|
||
None => return Err(AppError::NotFound("订单不存在".to_string())),
|
||
};
|
||
|
||
// 支付宝回调已通过 RSA2 验证。如果订单被用户误取消，重新激活。
|
||
if status == "cancelled" {
|
||
tracing::warn!("订单 {} 已被取消，但支付宝确认已收款，重新激活并处理支付", order_no);
|
||
} else if status != "pending" {
|
||
return Ok(());
|
||
}
|
||
|
||
// 查出订单关联的用户（用于审计日志）
|
||
let user_id: Option<i32> = sqlx::query_scalar(
|
||
r#"SELECT user_id FROM payment_orders WHERE order_no = $1"#,
|
||
)
|
||
.bind(order_no)
|
||
.fetch_optional(pool)
|
||
.await
|
||
.ok()
|
||
.flatten();
|
||
|
||
// 一次性完成：更新订单状态 + 累加计算新的到期时间
|
||
// 加上 AND status = 'pending' 防止竞态覆盖已取消/退款的订单
|
||
let result = sqlx::query(
|
||
r#"
|
||
WITH updated_order AS (
|
||
UPDATE payment_orders SET status = 'paid', paid_at = NOW() WHERE order_no = $1 AND status IN ('pending', 'cancelled') RETURNING package_type, user_id
|
||
)
|
||
UPDATE users SET
|
||
is_member = true,
|
||
membership_expires_at =
|
||
CASE
|
||
WHEN uo.package_type = 'permanent' THEN '2099-12-31'::TIMESTAMPTZ
|
||
ELSE GREATEST(COALESCE(users.membership_expires_at, NOW()), NOW()) +
|
||
CASE
|
||
WHEN uo.package_type = 'monthly' THEN INTERVAL '30 days'
|
||
WHEN uo.package_type = 'quarterly' THEN INTERVAL '90 days'
|
||
WHEN uo.package_type = 'half_year' THEN INTERVAL '182 days'
|
||
WHEN uo.package_type = 'yearly' THEN INTERVAL '365 days'
|
||
ELSE INTERVAL '0 days'
|
||
END
|
||
END
|
||
FROM updated_order uo
|
||
WHERE users.id = uo.user_id
|
||
"#,
|
||
)
|
||
.bind(order_no)
|
||
.execute(pool)
|
||
.await
|
||
.map_err(|e| AppError::Database(format!("支付确认失败: {}", e)))?;
|
||
|
||
// 0 行 update 说明订单已被取消/退款/已确认（竞态保护生效）
|
||
if result.rows_affected() == 0 {
|
||
tracing::warn!("支付确认: 订单 {} 状态已变更，跳过处理（可能是竞态或重复回调）", order_no);
|
||
return Ok(());
|
||
}
|
||
|
||
// 审计日志
|
||
if let Some(uid) = user_id {
|
||
if let Err(e) = insert_payment_audit_log(pool, order_no, uid, "paid", None, None).await {
|
||
tracing::warn!("插入支付审计日志失败: {}", e);
|
||
}
|
||
// 发送支付成功通知
|
||
if let Err(e) = insert_payment_notification(pool, uid, order_no).await {
|
||
tracing::warn!("插入支付通知失败: {}", e);
|
||
}
|
||
|
||
// 验证会员到期时间是否正常
|
||
if let Err(e) = verify_membership_after_payment(pool, order_no).await {
|
||
tracing::warn!("会员到期时间验证异常 (订单 {}): {}", order_no, e);
|
||
}
|
||
}
|
||
|
||
Ok(())
|
||
}
|
||
|
||
/// 退款处理：标记订单为 refunded + 撤销用户会员（如果无其他有效订单）
|
||
pub async fn refund_payment_order(
|
||
pool: &PgPool,
|
||
order_no: &str,
|
||
) -> Result<(), AppError> {
|
||
// 先查出订单关联的用户
|
||
let order_info: Option<(i32, String, Option<chrono::DateTime<chrono::Utc>>)> =
|
||
sqlx::query_as(
|
||
r#"SELECT user_id, status, paid_at FROM payment_orders WHERE order_no = $1"#,
|
||
)
|
||
.bind(order_no)
|
||
.fetch_optional(pool)
|
||
.await
|
||
.map_err(|e| AppError::Database(format!("查询订单失败: {}", e)))?;
|
||
|
||
let (user_id, status, _paid_at) = match order_info {
|
||
Some(info) => info,
|
||
None => return Err(AppError::NotFound("订单不存在".to_string())),
|
||
};
|
||
|
||
if status != "paid" {
|
||
tracing::warn!("订单 {} 状态为 {}，跳过退款处理", order_no, status);
|
||
return Ok(());
|
||
}
|
||
|
||
// 标记订单为 refunded
|
||
sqlx::query("UPDATE payment_orders SET status = 'refunded' WHERE order_no = $1")
|
||
.bind(order_no)
|
||
.execute(pool)
|
||
.await
|
||
.map_err(|e| AppError::Database(format!("退款标记失败: {}", e)))?;
|
||
|
||
// 检查该用户是否有其他有效的已支付订单
|
||
let other_active: (i64,) = sqlx::query_as(
|
||
r#"
|
||
SELECT COUNT(*) FROM payment_orders
|
||
WHERE user_id = $1
|
||
AND status = 'paid'
|
||
AND order_no != $2
|
||
AND (expires_at IS NULL OR expires_at > NOW())
|
||
"#,
|
||
)
|
||
.bind(user_id)
|
||
.bind(order_no)
|
||
.fetch_one(pool)
|
||
.await
|
||
.map_err(|e| AppError::Database(format!("查询其他有效订单失败: {}", e)))?;
|
||
|
||
// 如果没有其他有效订单，撤销会员状态
|
||
if other_active.0 == 0 {
|
||
sqlx::query(
|
||
"UPDATE users SET is_member = false, membership_expires_at = NULL WHERE id = $1",
|
||
)
|
||
.bind(user_id)
|
||
.execute(pool)
|
||
.await
|
||
.map_err(|e| AppError::Database(format!("撤销会员失败: {}", e)))?;
|
||
tracing::info!("用户 {} 的会员因订单 {} 退款已被撤销", user_id, order_no);
|
||
} else {
|
||
tracing::info!(
|
||
"用户 {} 有其他有效订单({}笔)，跳过会员撤销",
|
||
user_id,
|
||
other_active.0
|
||
);
|
||
}
|
||
|
||
// 审计日志
|
||
let detail = if other_active.0 == 0 { "退款，会员已撤销" } else { "退款，有其他有效订单，保留会员" };
|
||
if let Err(e) = insert_payment_audit_log(pool, order_no, user_id, "refunded", None, Some(detail)).await {
|
||
tracing::warn!("插入支付审计日志失败: {}", e);
|
||
}
|
||
|
||
Ok(())
|
||
}
|
||
|
||
/// 获取用户配额信息
|
||
///
|
||
/// 返回 (已用条数, 是否付费活跃, 到期时间)
|
||
pub async fn get_user_quota(
|
||
pool: &PgPool,
|
||
user_id: i32,
|
||
) -> Result<(i64, bool, Option<chrono::DateTime<chrono::Utc>>, bool), AppError> {
|
||
let user = get_user_by_id(pool, user_id).await?;
|
||
let is_active_member = user.is_member
|
||
&& user.membership_expires_at.is_none_or(|expires| expires > chrono::Utc::now());
|
||
let used = count_user_weather_data(pool, user_id).await?;
|
||
Ok((used, is_active_member, user.membership_expires_at, user.is_member))
|
||
}
|
||
|
||
// ===== 收藏功能 DB 函数 =====
|
||
|
||
/// 获取收藏列表（分页）
|
||
pub async fn get_favorites_list(
|
||
pool: &PgPool,
|
||
user_id: i32,
|
||
page: i32,
|
||
limit: i32,
|
||
) -> Result<WeatherListResponse, AppError> {
|
||
let offset = (page - 1) * limit;
|
||
|
||
let total_query = r#"
|
||
SELECT COUNT(*) as total
|
||
FROM weather_data
|
||
WHERE user_id = $1 AND is_favorite = true
|
||
"#;
|
||
let total = sqlx::query_as::<_, (i64,)>(total_query)
|
||
.bind(user_id)
|
||
.fetch_one(pool)
|
||
.await
|
||
.map_err(|e| AppError::Database(format!("查询收藏总数失败: {}", e)))?
|
||
.0;
|
||
|
||
let list_query = r#"
|
||
SELECT
|
||
id, title, date, hour, min, longitude, latitude, is_favorite
|
||
FROM weather_data
|
||
WHERE user_id = $1 AND is_favorite = true
|
||
ORDER BY date DESC, hour DESC, min DESC
|
||
LIMIT $2 OFFSET $3
|
||
"#;
|
||
let list = sqlx::query_as::<_, WeatherDataBrief>(list_query)
|
||
.bind(user_id)
|
||
.bind(limit)
|
||
.bind(offset)
|
||
.fetch_all(pool)
|
||
.await
|
||
.map_err(|e| AppError::Database(format!("查询收藏列表失败: {}", e)))?;
|
||
|
||
Ok(WeatherListResponse { list, total })
|
||
}
|
||
|
||
/// 设置天气数据的收藏状态
|
||
pub async fn set_weather_favorite(
|
||
pool: &PgPool,
|
||
weather_id: i32,
|
||
user_id: i32,
|
||
is_favorite: bool,
|
||
) -> Result<(), AppError> {
|
||
let query = r#"
|
||
UPDATE weather_data
|
||
SET is_favorite = $1
|
||
WHERE id = $2 AND user_id = $3
|
||
"#;
|
||
|
||
let result = sqlx::query(query)
|
||
.bind(is_favorite)
|
||
.bind(weather_id)
|
||
.bind(user_id)
|
||
.execute(pool)
|
||
.await
|
||
.map_err(|e| AppError::Database(format!("更新收藏状态失败: {}", e)))?;
|
||
|
||
if result.rows_affected() == 0 {
|
||
return Err(AppError::NotFound(format!("未找到ID为 {} 的天气数据或无权限修改", weather_id)));
|
||
}
|
||
|
||
Ok(())
|
||
}
|
||
|
||
// ===== Refresh Token DB 函数 =====
|
||
|
||
#[derive(Debug, FromRow)]
|
||
pub struct RefreshToken {
|
||
#[allow(dead_code)] // sqlx deserialization only; not read directly
|
||
pub id: i32,
|
||
pub user_id: i32,
|
||
#[allow(dead_code)] // sqlx deserialization only
|
||
pub token: String,
|
||
#[allow(dead_code)] // sqlx deserialization only
|
||
pub expires_at: chrono::DateTime<chrono::Utc>,
|
||
#[allow(dead_code)] // sqlx deserialization only
|
||
pub created_at: chrono::DateTime<chrono::Utc>,
|
||
}
|
||
|
||
pub async fn create_refresh_token(
|
||
pool: &PgPool,
|
||
user_id: i32,
|
||
token: &str,
|
||
expires_at: chrono::DateTime<chrono::Utc>,
|
||
) -> Result<(), AppError> {
|
||
let query = r#"
|
||
INSERT INTO refresh_tokens (user_id, token, expires_at)
|
||
VALUES ($1, $2, $3)
|
||
"#;
|
||
|
||
sqlx::query(query)
|
||
.bind(user_id)
|
||
.bind(token)
|
||
.bind(expires_at)
|
||
.execute(pool)
|
||
.await
|
||
.map_err(|e| AppError::Database(format!("创建 refresh_token 失败: {}", e)))?;
|
||
|
||
Ok(())
|
||
}
|
||
|
||
pub async fn verify_refresh_token(
|
||
pool: &PgPool,
|
||
token: &str,
|
||
) -> Result<RefreshToken, AppError> {
|
||
let query = r#"
|
||
SELECT id, user_id, token, expires_at, created_at
|
||
FROM refresh_tokens
|
||
WHERE token = $1 AND expires_at > NOW()
|
||
"#;
|
||
|
||
sqlx::query_as::<_, RefreshToken>(query)
|
||
.bind(token)
|
||
.fetch_optional(pool)
|
||
.await
|
||
.map_err(|e| AppError::Database(format!("验证 refresh_token 失败: {}", e)))?
|
||
.ok_or_else(|| AppError::Unauthorized("Refresh token 无效或已过期".to_string()))
|
||
}
|
||
|
||
#[allow(dead_code)]
|
||
pub async fn delete_refresh_token(pool: &PgPool, user_id: i32) -> Result<(), AppError> {
|
||
let query = r#"DELETE FROM refresh_tokens WHERE user_id = $1"#;
|
||
|
||
sqlx::query(query)
|
||
.bind(user_id)
|
||
.execute(pool)
|
||
.await
|
||
.map_err(|e| AppError::Database(format!("删除 refresh_token 失败: {}", e)))?;
|
||
|
||
Ok(())
|
||
}
|
||
|
||
pub async fn delete_specific_refresh_token(pool: &PgPool, token: &str) -> Result<(), AppError> {
|
||
let query = r#"DELETE FROM refresh_tokens WHERE token = $1"#;
|
||
|
||
sqlx::query(query)
|
||
.bind(token)
|
||
.execute(pool)
|
||
.await
|
||
.map_err(|e| AppError::Database(format!("删除 refresh_token 失败: {}", e)))?;
|
||
|
||
Ok(())
|
||
}
|
||
|
||
pub async fn cleanup_expired_refresh_tokens(pool: &PgPool) -> Result<u64, AppError> {
|
||
let query = r#"DELETE FROM refresh_tokens WHERE expires_at <= NOW()"#;
|
||
|
||
let result = sqlx::query(query)
|
||
.execute(pool)
|
||
.await
|
||
.map_err(|e| AppError::Database(format!("清理过期 refresh_token 失败: {}", e)))?;
|
||
|
||
Ok(result.rows_affected())
|
||
}
|
||
|
||
// ===== 支付审计日志 =====
|
||
|
||
/// 写入支付审计日志
|
||
pub async fn insert_payment_audit_log(
|
||
pool: &PgPool,
|
||
order_no: &str,
|
||
user_id: i32,
|
||
action: &str,
|
||
operator_id: Option<i32>,
|
||
detail: Option<&str>,
|
||
) -> Result<(), AppError> {
|
||
// 自动建表（幂等）
|
||
sqlx::query(
|
||
r#"
|
||
CREATE TABLE IF NOT EXISTS payment_audit_log (
|
||
id SERIAL PRIMARY KEY,
|
||
order_no VARCHAR(64) NOT NULL,
|
||
user_id INTEGER NOT NULL,
|
||
action VARCHAR(32) NOT NULL,
|
||
operator_id INTEGER,
|
||
detail TEXT,
|
||
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
|
||
)
|
||
"#,
|
||
)
|
||
.execute(pool)
|
||
.await
|
||
.map_err(|e| AppError::Database(format!("创建支付审计表失败: {}", e)))?;
|
||
|
||
sqlx::query(
|
||
r#"
|
||
INSERT INTO payment_audit_log (order_no, user_id, action, operator_id, detail)
|
||
VALUES ($1, $2, $3, $4, $5)
|
||
"#,
|
||
)
|
||
.bind(order_no)
|
||
.bind(user_id)
|
||
.bind(action)
|
||
.bind(operator_id)
|
||
.bind(detail)
|
||
.execute(pool)
|
||
.await
|
||
.map_err(|e| AppError::Database(format!("写入支付审计日志失败: {}", e)))?;
|
||
|
||
Ok(())
|
||
}
|
||
|
||
/// 管理员强制确认待支付订单（带审计日志）
|
||
pub async fn admin_force_confirm_order(
|
||
pool: &PgPool,
|
||
order_no: &str,
|
||
admin_user_id: i32,
|
||
) -> Result<Option<chrono::DateTime<chrono::Utc>>, AppError> {
|
||
// 查出订单关联的用户
|
||
let order = sqlx::query_as::<_, (i32, String)>(
|
||
r#"SELECT user_id, status FROM payment_orders WHERE order_no = $1"#,
|
||
)
|
||
.bind(order_no)
|
||
.fetch_optional(pool)
|
||
.await
|
||
.map_err(|e| AppError::Database(format!("查询订单失败: {}", e)))?;
|
||
|
||
let (user_id, status) = match order {
|
||
Some(o) => o,
|
||
None => return Err(AppError::NotFound("订单不存在".to_string())),
|
||
};
|
||
|
||
// 非管理员不能操作
|
||
// (调用方已校验)
|
||
|
||
let new_expires = if status == "pending" {
|
||
// 直接执行确认 SQL（跳过 user_id 校验，由管理员操作）
|
||
let result = sqlx::query_scalar::<_, Option<chrono::DateTime<chrono::Utc>>>(
|
||
r#"
|
||
WITH updated_order AS (
|
||
UPDATE payment_orders SET status = 'paid', paid_at = NOW() WHERE order_no = $1 AND status = 'pending' RETURNING package_type
|
||
)
|
||
UPDATE users SET
|
||
is_member = true,
|
||
membership_expires_at =
|
||
CASE
|
||
WHEN uo.package_type = 'permanent' THEN '2099-12-31'::TIMESTAMPTZ
|
||
ELSE GREATEST(COALESCE(users.membership_expires_at, NOW()), NOW()) +
|
||
CASE
|
||
WHEN uo.package_type = 'monthly' THEN INTERVAL '30 days'
|
||
WHEN uo.package_type = 'quarterly' THEN INTERVAL '90 days'
|
||
WHEN uo.package_type = 'half_year' THEN INTERVAL '182 days'
|
||
WHEN uo.package_type = 'yearly' THEN INTERVAL '365 days'
|
||
ELSE INTERVAL '0 days'
|
||
END
|
||
END
|
||
FROM updated_order uo
|
||
WHERE users.id = $2
|
||
RETURNING users.membership_expires_at
|
||
"#,
|
||
)
|
||
.bind(order_no)
|
||
.bind(user_id)
|
||
.fetch_one(pool)
|
||
.await
|
||
.map_err(|e| AppError::Database(format!("管理员确认订单失败: {}", e)))?;
|
||
|
||
// 写审计日志
|
||
if let Err(e) = insert_payment_audit_log(
|
||
pool,
|
||
order_no,
|
||
user_id,
|
||
"admin_confirm",
|
||
Some(admin_user_id),
|
||
Some("管理员强制确认支付"),
|
||
)
|
||
.await
|
||
{
|
||
tracing::warn!("插入支付审计日志失败: {}", e);
|
||
}
|
||
|
||
// 验证会员到期时间是否正常
|
||
if let Err(e) = verify_membership_after_payment(pool, order_no).await {
|
||
tracing::warn!("会员到期时间验证异常 (订单 {}): {}", order_no, e);
|
||
}
|
||
|
||
result
|
||
} else {
|
||
// 订单已处理，只写日志
|
||
if let Err(e) = insert_payment_audit_log(
|
||
pool,
|
||
order_no,
|
||
user_id,
|
||
"admin_confirm",
|
||
Some(admin_user_id),
|
||
Some(&format!("订单状态为 {}，跳过确认", status)),
|
||
)
|
||
.await
|
||
{
|
||
tracing::warn!("插入支付审计日志失败: {}", e);
|
||
}
|
||
|
||
// 返回当前到期时间
|
||
sqlx::query_scalar::<_, Option<chrono::DateTime<chrono::Utc>>>(
|
||
r#"SELECT membership_expires_at FROM users WHERE id = $1"#,
|
||
)
|
||
.bind(user_id)
|
||
.fetch_one(pool)
|
||
.await
|
||
.map_err(|e| AppError::Database(format!("查询用户到期时间失败: {}", e)))?
|
||
};
|
||
|
||
Ok(new_expires)
|
||
}
|
||
|
||
/// 用户主动取消待支付订单
|
||
pub async fn cancel_payment_order(
|
||
pool: &PgPool,
|
||
order_no: &str,
|
||
user_id: i32,
|
||
) -> Result<(), AppError> {
|
||
// 检查订单有效且属于该用户
|
||
let row = sqlx::query_as::<_, (i32, String)>(
|
||
r#"SELECT user_id, status FROM payment_orders WHERE order_no = $1"#,
|
||
)
|
||
.bind(order_no)
|
||
.fetch_optional(pool)
|
||
.await
|
||
.map_err(|e| AppError::Database(format!("查询订单失败: {}", e)))?;
|
||
|
||
let (order_user_id, status) = match row {
|
||
Some(r) => r,
|
||
None => return Err(AppError::NotFound("订单不存在".to_string())),
|
||
};
|
||
|
||
if order_user_id != user_id {
|
||
return Err(AppError::Forbidden("无权操作此订单".to_string()));
|
||
}
|
||
|
||
if status != "pending" {
|
||
return Err(AppError::BadRequest(format!("订单状态为 {}，无法取消", status)));
|
||
}
|
||
|
||
// 防止用户误取消：支付宝支付确认通常在 5-30 秒内到达
|
||
// 2 分钟内的订单不允许取消，避免用户付款后误触取消按钮
|
||
let order_age: f64 = sqlx::query_scalar(
|
||
r#"SELECT EXTRACT(EPOCH FROM (NOW() - created_at)) FROM payment_orders WHERE order_no = $1"#,
|
||
)
|
||
.bind(order_no)
|
||
.fetch_one(pool)
|
||
.await
|
||
.map_err(|e| AppError::Database(format!("查询订单创建时间失败: {}", e)))?;
|
||
|
||
if order_age < 120.0 {
|
||
return Err(AppError::BadRequest(
|
||
"订单刚刚创建，支付可能仍在处理中，请 2 分钟后再试".to_string(),
|
||
));
|
||
}
|
||
|
||
sqlx::query("UPDATE payment_orders SET status = 'cancelled' WHERE order_no = $1")
|
||
.bind(order_no)
|
||
.execute(pool)
|
||
.await
|
||
.map_err(|e| AppError::Database(format!("取消订单失败: {}", e)))?;
|
||
|
||
if let Err(e) = insert_payment_audit_log(pool, order_no, user_id, "cancelled", Some(user_id),
|
||
Some("用户主动取消待支付订单")).await
|
||
{
|
||
tracing::warn!("插入支付审计日志失败: {}", e);
|
||
}
|
||
|
||
Ok(())
|
||
}
|
||
|
||
/// 管理员手动退款（标记订单为 refunded + 重新计算会员）
|
||
pub async fn admin_refund_order(
|
||
pool: &PgPool,
|
||
order_no: &str,
|
||
admin_user_id: i32,
|
||
) -> Result<(), AppError> {
|
||
// 查出订单信息
|
||
let order = sqlx::query_as::<_, (i32, String)>(
|
||
r#"SELECT user_id, status FROM payment_orders WHERE order_no = $1"#,
|
||
)
|
||
.bind(order_no)
|
||
.fetch_optional(pool)
|
||
.await
|
||
.map_err(|e| AppError::Database(format!("查询订单失败: {}", e)))?;
|
||
|
||
let (user_id, status) = match order {
|
||
Some(o) => o,
|
||
None => return Err(AppError::NotFound("订单不存在".to_string())),
|
||
};
|
||
|
||
if status != "paid" && status != "pending" {
|
||
return Err(AppError::BadRequest(format!("订单状态为 {}，无法退款", status)));
|
||
}
|
||
|
||
// 标记为 refunded
|
||
sqlx::query("UPDATE payment_orders SET status = 'refunded' WHERE order_no = $1")
|
||
.bind(order_no)
|
||
.execute(pool)
|
||
.await
|
||
.map_err(|e| AppError::Database(format!("退款标记失败: {}", e)))?;
|
||
|
||
// 如果订单是已支付状态，重新计算会员
|
||
if status == "paid" {
|
||
// 检查该用户是否有其他有效的已支付订单
|
||
let other_active: (i64,) = sqlx::query_as(
|
||
r#"
|
||
SELECT COUNT(*) FROM payment_orders
|
||
WHERE user_id = $1
|
||
AND status = 'paid'
|
||
AND order_no != $2
|
||
AND (expires_at IS NULL OR expires_at > NOW())
|
||
"#,
|
||
)
|
||
.bind(user_id)
|
||
.bind(order_no)
|
||
.fetch_one(pool)
|
||
.await
|
||
.map_err(|e| AppError::Database(format!("查询其他有效订单失败: {}", e)))?;
|
||
|
||
if other_active.0 == 0 {
|
||
sqlx::query(
|
||
"UPDATE users SET is_member = false, membership_expires_at = NULL WHERE id = $1",
|
||
)
|
||
.bind(user_id)
|
||
.execute(pool)
|
||
.await
|
||
.map_err(|e| AppError::Database(format!("撤销会员失败: {}", e)))?;
|
||
tracing::info!("管理员退款: 用户 {} 的会员因订单 {} 退款已被撤销", user_id, order_no);
|
||
} else {
|
||
tracing::info!(
|
||
"管理员退款: 用户 {} 有其他有效订单({}笔)，保留会员",
|
||
user_id,
|
||
other_active.0
|
||
);
|
||
}
|
||
}
|
||
|
||
let detail = if status == "paid" { "管理员手动退款（已支付订单）" } else { "管理员取消订单（待支付订单）" };
|
||
if let Err(e) = insert_payment_audit_log(pool, order_no, user_id, "refunded",
|
||
Some(admin_user_id), Some(detail)).await
|
||
{
|
||
tracing::warn!("插入支付审计日志失败: {}", e);
|
||
}
|
||
|
||
Ok(())
|
||
}
|
||
|
||
|
||
/// 支付确认后验证会员到期时间是否正常
|
||
/// 用于捕获 ELSE INTERVAL '0 days' 的静默错误
|
||
async fn verify_membership_after_payment(
|
||
pool: &PgPool,
|
||
order_no: &str,
|
||
) -> Result<(), AppError> {
|
||
#[derive(Debug, sqlx::FromRow)]
|
||
struct PaymentResult {
|
||
package_type: String,
|
||
paid_at: chrono::DateTime<chrono::Utc>,
|
||
membership_expires_at: Option<chrono::DateTime<chrono::Utc>>,
|
||
}
|
||
|
||
let row = sqlx::query_as::<_, PaymentResult>(
|
||
r#"
|
||
SELECT po.package_type, po.paid_at, u.membership_expires_at
|
||
FROM payment_orders po
|
||
JOIN users u ON u.id = po.user_id
|
||
WHERE po.order_no = $1
|
||
"#,
|
||
)
|
||
.bind(order_no)
|
||
.fetch_optional(pool)
|
||
.await
|
||
.map_err(|e| AppError::Database(format!("验证查询失败: {}", e)))?;
|
||
|
||
let r = match row {
|
||
Some(r) => r,
|
||
None => return Err(AppError::NotFound("验证查询：订单不存在".to_string())),
|
||
};
|
||
|
||
if r.package_type == "permanent" {
|
||
return Ok(()); // 永久会员不累计
|
||
}
|
||
|
||
let expires = match r.membership_expires_at {
|
||
Some(e) => e,
|
||
None => {
|
||
let msg = format!("支付验证失败：{} 非永久套餐但会员到期为 NULL", r.package_type);
|
||
tracing::error!("{}", msg);
|
||
return Err(AppError::Internal(msg));
|
||
}
|
||
};
|
||
|
||
let actual_days = (expires - r.paid_at).num_days();
|
||
|
||
// 预期至少 25 天（最短月卡 30 天，允许 5 天误差）
|
||
if actual_days < 25 {
|
||
let msg = format!(
|
||
"支付验证失败：套餐 {} 仅增加 {} 天会员时长（预期 >= 25 天），可能触发了 ELSE 分支",
|
||
r.package_type, actual_days
|
||
);
|
||
tracing::error!("{}", msg);
|
||
return Err(AppError::Internal(msg));
|
||
}
|
||
|
||
tracing::info!("支付验证通过：{} 增加 {} 天", r.package_type, actual_days);
|
||
Ok(())
|
||
}
|