fix: 添加后台定时重试 + 管理员强制确认 + 审计日志 + mock守卫
This commit is contained in:
14
migrations/009_add_payment_audit_log.sql
Normal file
14
migrations/009_add_payment_audit_log.sql
Normal file
@@ -0,0 +1,14 @@
|
||||
-- 支付操作审计日志表
|
||||
CREATE TABLE IF NOT EXISTS payment_audit_log (
|
||||
id SERIAL PRIMARY KEY,
|
||||
order_no VARCHAR(64) NOT NULL REFERENCES payment_orders(order_no),
|
||||
user_id INTEGER NOT NULL REFERENCES users(id),
|
||||
action VARCHAR(32) NOT NULL, -- paid, refunded, cancelled, admin_confirm, admin_revoke, expired
|
||||
operator_id INTEGER REFERENCES users(id), -- NULL 表示系统操作
|
||||
detail TEXT,
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_payment_audit_log_order_no ON payment_audit_log(order_no);
|
||||
CREATE INDEX IF NOT EXISTS idx_payment_audit_log_user_id ON payment_audit_log(user_id);
|
||||
CREATE INDEX IF NOT EXISTS idx_payment_audit_log_created_at ON payment_audit_log(created_at);
|
||||
197
src/db.rs
197
src/db.rs
@@ -338,6 +338,47 @@ pub async fn get_user_orders(
|
||||
Ok(rows)
|
||||
}
|
||||
|
||||
/// 扫描超过 10 分钟的待支付订单,尝试自动确认(用于回调重试)
|
||||
/// 返回成功确认的订单数量
|
||||
pub async fn check_and_retry_pending_orders(pool: &PgPool) -> Result<u64, AppError> {
|
||||
// 查找超过 10 分钟仍为 pending 的订单
|
||||
let stuck_orders: Vec<(String,)> = sqlx::query_as(
|
||||
r#"
|
||||
SELECT order_no FROM payment_orders
|
||||
WHERE status = 'pending'
|
||||
AND created_at < NOW() - INTERVAL '10 minutes'
|
||||
AND created_at > NOW() - INTERVAL '24 hours'
|
||||
ORDER BY created_at ASC
|
||||
"#,
|
||||
)
|
||||
.fetch_all(pool)
|
||||
.await
|
||||
.map_err(|e| AppError::Database(format!("查询待处理订单失败: {}", e)))?;
|
||||
|
||||
if stuck_orders.is_empty() {
|
||||
return Ok(0);
|
||||
}
|
||||
|
||||
tracing::info!(
|
||||
"发现 {} 个待处理订单,尝试自动确认...",
|
||||
stuck_orders.len()
|
||||
);
|
||||
|
||||
let mut confirmed: u64 = 0;
|
||||
for (order_no,) in &stuck_orders {
|
||||
match confirm_payment_order_by_orderno(pool, order_no).await {
|
||||
Ok(_) => {
|
||||
confirmed += 1;
|
||||
tracing::info!("定时任务自动确认订单: {}", order_no);
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::warn!("定时任务无法确认订单 {}: {}", order_no, e);
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(confirmed)
|
||||
}
|
||||
|
||||
/// 清理超过 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 {
|
||||
@@ -507,6 +548,16 @@ pub async fn confirm_payment_order_by_orderno(
|
||||
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();
|
||||
|
||||
// 一次性完成:更新订单状态 + 累加计算新的到期时间
|
||||
sqlx::query(
|
||||
r#"
|
||||
@@ -534,6 +585,11 @@ pub async fn confirm_payment_order_by_orderno(
|
||||
.await
|
||||
.map_err(|e| AppError::Database(format!("支付确认失败: {}", e)))?;
|
||||
|
||||
// 审计日志
|
||||
if let Some(uid) = user_id {
|
||||
let _ = insert_payment_audit_log(pool, order_no, uid, "paid", None, None).await;
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -603,6 +659,10 @@ pub async fn refund_payment_order(
|
||||
);
|
||||
}
|
||||
|
||||
// 审计日志
|
||||
let detail = if other_active.0 == 0 { "退款,会员已撤销" } else { "退款,有其他有效订单,保留会员" };
|
||||
let _ = insert_payment_audit_log(pool, order_no, user_id, "refunded", None, Some(detail)).await;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -775,3 +835,140 @@ pub async fn cleanup_expired_refresh_tokens(pool: &PgPool) -> Result<u64, AppErr
|
||||
|
||||
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 RETURNING package_type
|
||||
)
|
||||
UPDATE users SET
|
||||
is_paid = true,
|
||||
paid_expires_at =
|
||||
CASE
|
||||
WHEN uo.package_type = 'permanent' THEN NULL
|
||||
ELSE GREATEST(COALESCE(users.paid_expires_at, NOW()), NOW()) +
|
||||
CASE
|
||||
WHEN uo.package_type = 'monthly' THEN INTERVAL '30 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.paid_expires_at
|
||||
"#,
|
||||
)
|
||||
.bind(order_no)
|
||||
.bind(user_id)
|
||||
.fetch_one(pool)
|
||||
.await
|
||||
.map_err(|e| AppError::Database(format!("管理员确认订单失败: {}", e)))?;
|
||||
|
||||
// 写审计日志
|
||||
let _ = insert_payment_audit_log(
|
||||
pool,
|
||||
order_no,
|
||||
user_id,
|
||||
"admin_confirm",
|
||||
Some(admin_user_id),
|
||||
Some("管理员强制确认支付"),
|
||||
)
|
||||
.await;
|
||||
|
||||
result
|
||||
} else {
|
||||
// 订单已处理,只写日志
|
||||
let _ = insert_payment_audit_log(
|
||||
pool,
|
||||
order_no,
|
||||
user_id,
|
||||
"admin_confirm",
|
||||
Some(admin_user_id),
|
||||
Some(&format!("订单状态为 {},跳过确认", status)),
|
||||
)
|
||||
.await;
|
||||
|
||||
// 返回当前到期时间
|
||||
sqlx::query_scalar::<_, Option<chrono::DateTime<chrono::Utc>>>(
|
||||
r#"SELECT paid_expires_at FROM users WHERE id = $1"#,
|
||||
)
|
||||
.bind(user_id)
|
||||
.fetch_one(pool)
|
||||
.await
|
||||
.map_err(|e| AppError::Database(format!("查询用户到期时间失败: {}", e)))?
|
||||
};
|
||||
|
||||
Ok(new_expires)
|
||||
}
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
use actix_web::{web, get, put, HttpResponse};
|
||||
use actix_web::{web, get, post, put, HttpResponse};
|
||||
use sqlx::postgres::PgPool;
|
||||
use chrono::{DateTime, Utc};
|
||||
|
||||
@@ -58,8 +58,42 @@ pub async fn admin_update_user_payment(
|
||||
|
||||
// 更新用户付费状态
|
||||
db::update_user_payment_status(pool.get_ref(), target_user_id, body.is_paid, paid_expires_at).await?;
|
||||
// 审计日志
|
||||
let _ = db::insert_payment_audit_log(
|
||||
pool.get_ref(), "ADMIN_MANUAL", target_user_id, "admin_revoke",
|
||||
Some(claims.user_id),
|
||||
Some(&format!("管理员手动更新付费状态: is_paid={}, expires_at={:?}", body.is_paid, body.paid_expires_at)),
|
||||
).await;
|
||||
Ok(HttpResponse::Ok().json(serde_json::json!({
|
||||
"success": true,
|
||||
"message": "用户付费状态已更新"
|
||||
})))
|
||||
}
|
||||
|
||||
/// POST /api/admin/orders/{order_no}/force-confirm
|
||||
/// 管理员强制确认指定订单(用于回调失败后的手动恢复)
|
||||
#[post("/api/admin/orders/{order_no}/force-confirm")]
|
||||
pub async fn admin_force_confirm_order(
|
||||
path: web::Path<String>,
|
||||
pool: web::Data<PgPool>,
|
||||
claims: web::ReqData<Claims>,
|
||||
) -> Result<HttpResponse, AppError> {
|
||||
let order_no = path.into_inner();
|
||||
tracing::info!("管理员强制确认订单: {}", order_no);
|
||||
|
||||
// 验证当前用户是否为管理员
|
||||
let current_user = db::get_user_by_id(pool.get_ref(), claims.user_id).await?;
|
||||
if !current_user.is_admin {
|
||||
return Err(AppError::Forbidden("无权限执行此操作".to_string()));
|
||||
}
|
||||
|
||||
let new_expires = db::admin_force_confirm_order(pool.get_ref(), &order_no, claims.user_id).await?;
|
||||
|
||||
Ok(HttpResponse::Ok().json(serde_json::json!({
|
||||
"success": true,
|
||||
"message": "订单已确认",
|
||||
"data": {
|
||||
"new_expires_at": new_expires,
|
||||
}
|
||||
})))
|
||||
}
|
||||
|
||||
@@ -590,13 +590,16 @@ pub async fn web_login_confirm(
|
||||
// 查询付费状态
|
||||
let (is_paid_active, paid_expires_at): (bool, Option<String>) =
|
||||
match sqlx::query_as::<_, (bool, Option<chrono::DateTime<chrono::Utc>>)>(
|
||||
"SELECT is_paid_active($1)",
|
||||
"SELECT is_paid, paid_expires_at FROM users WHERE id = $1",
|
||||
)
|
||||
.bind(user_id)
|
||||
.fetch_optional(pool.get_ref())
|
||||
.await
|
||||
{
|
||||
Ok(Some((active, expires))) => (active, expires.map(|e| e.to_rfc3339())),
|
||||
Ok(Some((is_paid, expires))) => {
|
||||
let active = is_paid && expires.map_or(true, |e| e > Utc::now());
|
||||
(active, expires.map(|e| e.to_rfc3339()))
|
||||
}
|
||||
_ => (false, None),
|
||||
};
|
||||
|
||||
@@ -642,7 +645,9 @@ pub async fn web_login_auto_confirm(
|
||||
) -> impl Responder {
|
||||
let code = req.code.clone();
|
||||
|
||||
let openid = if code.starts_with("mock_") || code == "test_mock" {
|
||||
let openid = if (code.starts_with("mock_") || code == "test_mock")
|
||||
&& std::env::var("MOCK_LOGIN_ENABLED").ok() == Some("true".to_string())
|
||||
{
|
||||
format!("mock_openid_{}", Utc::now().timestamp_millis())
|
||||
} else {
|
||||
let url = format!(
|
||||
|
||||
@@ -18,6 +18,7 @@ pub static TEMPLATES_DIR: Dir = include_dir!("$CARGO_MANIFEST_DIR/templates");
|
||||
// re-export handlers for convenient use in main.rs
|
||||
pub use admin::admin_get_user;
|
||||
pub use admin::admin_update_user_payment;
|
||||
pub use admin::admin_force_confirm_order;
|
||||
pub use auth::login;
|
||||
pub use auth::mock_login;
|
||||
pub use auth::refresh_token;
|
||||
|
||||
@@ -1173,12 +1173,17 @@ pub async fn payment_login_status(
|
||||
let user_id = user_id.unwrap();
|
||||
|
||||
let (is_paid_active, paid_expires_at): (bool, Option<String>) =
|
||||
match sqlx::query_as::<_, (bool, Option<chrono::DateTime<chrono::Utc>>)>("SELECT is_paid_active($1)")
|
||||
match sqlx::query_as::<_, (bool, Option<chrono::DateTime<chrono::Utc>>)>(
|
||||
"SELECT is_paid, paid_expires_at FROM users WHERE id = $1"
|
||||
)
|
||||
.bind(user_id)
|
||||
.fetch_optional(pool.get_ref())
|
||||
.await
|
||||
{
|
||||
Ok(Some((active, expires))) => (active, expires.map(|e| e.to_rfc3339())),
|
||||
Ok(Some((is_paid, expires))) => {
|
||||
let active = is_paid && expires.map_or(true, |e| e > Utc::now());
|
||||
(active, expires.map(|e| e.to_rfc3339()))
|
||||
}
|
||||
_ => (false, None),
|
||||
};
|
||||
|
||||
|
||||
25
src/main.rs
25
src/main.rs
@@ -18,7 +18,7 @@ use auth::jwt_middleware;
|
||||
use config::AppConfig;
|
||||
use db::create_pool;
|
||||
use handlers::{
|
||||
admin_get_user, admin_update_user_payment, add_favorite, alipay_notify, alipay_pay_page, alipay_refund_notify,
|
||||
admin_force_confirm_order, admin_get_user, admin_update_user_payment, add_favorite, alipay_notify, alipay_pay_page, alipay_refund_notify,
|
||||
create_order, delete_weather, generate_code, generate_temp_token_handler, get_current_user_profile,
|
||||
get_favorites, get_user_quota, get_weather_brief, get_weather_details,
|
||||
get_user_orders, health_check, login, mock_login, mock_confirm, sync_order, payment_index, payment_login_status, payment_page, payment_success,
|
||||
@@ -107,6 +107,7 @@ fn create_server_config(
|
||||
.service(save_user_profile)
|
||||
.service(admin_get_user)
|
||||
.service(admin_update_user_payment)
|
||||
.service(admin_force_confirm_order)
|
||||
.service(create_order)
|
||||
.service(mock_confirm)
|
||||
.service(sync_order)
|
||||
@@ -231,17 +232,35 @@ async fn main() -> std::io::Result<()> {
|
||||
|
||||
let http_client = Client::new();
|
||||
|
||||
// 启动时清理所有超过 24 小时的待支付订单
|
||||
// 启动时清理:过期订单 + 过期登录码
|
||||
match db::cleanup_expired_pending_orders(&pool, None).await {
|
||||
Ok(n) => info!("已清理 {} 个过期待支付订单(启动时)", n),
|
||||
Err(e) => warn!("启动时清理过期待支付订单失败: {}", e),
|
||||
}
|
||||
// 启动时清理所有过期的 web 登录码
|
||||
match db::cleanup_expired_login_codes(&pool).await {
|
||||
Ok(n) => info!("已清理 {} 个过期的 web 登录码(启动时)", n),
|
||||
Err(e) => warn!("启动时清理 web 登录码失败: {}", e),
|
||||
}
|
||||
|
||||
// 后台定时扫描待支付订单(每 5 分钟)
|
||||
let pool_clone = pool.clone();
|
||||
actix_web::rt::spawn(async move {
|
||||
let mut interval = actix_web::rt::time::interval(std::time::Duration::from_secs(300));
|
||||
interval.tick().await; // 跳过立即执行
|
||||
loop {
|
||||
interval.tick().await;
|
||||
tracing::info!("[定时任务] 开始扫描待支付订单...");
|
||||
match db::check_and_retry_pending_orders(&pool_clone).await {
|
||||
Ok(confirmed) => {
|
||||
if confirmed > 0 {
|
||||
tracing::info!("[定时任务] 自动确认了 {} 个待支付订单", confirmed);
|
||||
}
|
||||
}
|
||||
Err(e) => tracing::warn!("[定时任务] 扫描待支付订单失败: {}", e),
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
info!("Attempting to start server...");
|
||||
|
||||
let is_production = std::env::var("APP_ENV").unwrap_or_else(|_| "development".into()) == "production";
|
||||
|
||||
Reference in New Issue
Block a user