From 5d0b790aa11a2d55c8c76c0449a0e4a613fbd7e3 Mon Sep 17 00:00:00 2001 From: Ember Moth Date: Sun, 5 Jul 2026 20:43:52 +0800 Subject: [PATCH 1/3] add mysql2postgres --- AGENTS.md | 285 ------------------- Cargo.lock | 128 +++++++++ Cargo.toml | 2 +- tools/mysql2postgres/Cargo.toml | 19 ++ tools/mysql2postgres/src/main.rs | 474 +++++++++++++++++++++++++++++++ 5 files changed, 622 insertions(+), 286 deletions(-) delete mode 100644 AGENTS.md create mode 100644 tools/mysql2postgres/Cargo.toml create mode 100644 tools/mysql2postgres/src/main.rs diff --git a/AGENTS.md b/AGENTS.md deleted file mode 100644 index 7f54eb08..00000000 --- a/AGENTS.md +++ /dev/null @@ -1,285 +0,0 @@ -# AGENTS.md - -`ppanel-backend` 项目约定，供 AI agent 及开发者遵循。 - -## ppanel-backend - -基于 axum 的 Rust 重写项目，Go 源码位于 `../server`。Go 代码库是行为正确性的唯一参考——当 Rust handler 需要对齐 Go 语义时，从对应的 Go 包移植，而非重新设计。 - ---- - -## 一、返回值必须通过 `result` crate - -所有 HTTP handler 的响应都必须通过 `result` crate 生成。**禁止**在 handler 中手写 `axum` 响应结构体、自行封装 JSON 或拼装 `axum::Json`——统一使用已有工具，以保持响应格式与 Go `result` 包一致。 - -``` -crates/result/src/ -├── lib.rs # pub mod code_error / error_code / http_result -├── error_code.rs # 错误码常量 (SUCCESS, ERROR, INVALID_PARAMS, ...) + map_err_msg / is_code_err -├── code_error.rs # CodeError (new_err_code / new_err_code_msg / new_err_msg) + STATUS_NOT_MODIFIED -└── http_result.rs # ResponseSuccessBean / ResponseErrorBean / HttpResult + build/http/param 工具函数 -``` - -### API 一览 - -`result::http_result` -- `build_http_result(resp, err) -> HttpResult` — 核心构造函数；出错时从 `anyhow::Error` 链中提取 - `CodeError`，找不到则回退为 `ERROR` / `"Internal Server Error"`。始终返回 HTTP 200，业务码放在 - body 的 `code` 字段（与 Go 行为一致）。 -- `build_param_error_result(err) -> HttpResult` — HTTP 200，业务码 `INVALID_PARAMS`。 -- `HttpResult` 实现了 `IntoResponse`，可直接作为 handler 的返回值。 - -`result::code_error` -- `CodeError::new_err_code(code)` — 消息由 `map_err_msg` 自动查表。 -- `CodeError::new_err_code_msg(code, msg)` — 显式指定 code + 消息。 -- `CodeError::new_err_msg(msg)` — 业务码默认为 `ERROR`。 - -`result::error_code` -- 仅包含命名常量，如 `SUCCESS`、`ERROR`、`INVALID_PARAMS`、`USER_NOT_EXIST`。 - 必须使用这些常量，**禁止魔术数字**。 - -### Handler 用法模板 - -```rust -use result::code_error::CodeError; -use result::error_code; -use result::http_result::{build_http_result, HttpResult}; - -pub async fn handler(State(state): State, Json(req): Json) -> HttpResult { - let res = some_service(req).await - .map_err(|_| anyhow::Error::new(CodeError::new_err_code(error_code::USER_NOT_EXIST))); - build_http_result(res.ok(), res.err()) -} -``` - -规则： -- HTTP 状态码固定 200，业务码在 body `code` 字段，禁止用 HTTP 状态码表示业务错误。 -- 成功路径 → `Some(data)`；错误路径 → `CodeError` 包装进 `anyhow::Error`。 - ---- - -## 二、日志系统（两层） - -### 2.1 业务审计日志 — Telemetry facade - -**位置**：`src/service/telemetry.rs` - -所有业务事件写入 `system_logs` 表，必须通过 `Telemetry` facade，**禁止**在 service 层直接构造 `SystemLog` 并调用 `repos.log.insert()`。 - -```rust -use crate::service::telemetry::Telemetry; - -// 登录成功 -Telemetry::login(&repos, user_id, "email", &ip, &user_agent, true).await; - -// 注册成功 -Telemetry::register(&repos, user_id, "email", &email, &ip, &user_agent).await; - -// 余额变动（type_ 用 BALANCE_TYPE_* 常量） -Telemetry::balance(&repos, user_id, BALANCE_TYPE_RECHARGE, amount, Some(order_no), balance).await; -``` - -**全部 14 种方法**： - -| 方法 | LogType | 优先级 | -|------|---------|--------| -| `login` | LOGIN (30) | P0 | -| `register` | REGISTER (31) | P0 | -| `balance` | BALANCE (32) | P1 | -| `commission` | COMMISSION (33) | P1 | -| `gift` | GIFT (34) | P1 | -| `subscribe_access` | SUBSCRIBE (20) | P1 | -| `subscribe_traffic` | SUBSCRIBE_TRAFFIC (21) | P2 | -| `server_traffic` | SERVER_TRAFFIC (22) | P2 | -| `reset_subscribe` | RESET_SUBSCRIBE (23) | P2 | -| `email_message` | EMAIL_MESSAGE (10) | P2 | -| `mobile_message` | MOBILE_MESSAGE (11) | P2 | -| `user_traffic_rank` | USER_TRAFFIC_RANK (40) | P3 | -| `server_traffic_rank` | SERVER_TRAFFIC_RANK (41) | P3 | -| `traffic_stat` | TRAFFIC_STAT (42) | P3 | - -子类型常量定义在 `src/model/entity/log.rs`（`BALANCE_TYPE_*`、`COMMISSION_TYPE_*` 等）。 - -### 2.2 应用操作日志 — tracing - -运维/调试日志使用 `tracing::info!` / `tracing::error!`，由 `main.rs` 根据 `LogConfig` 初始化。 -请求日志由 `src/middleware/logger_middleware.rs`（`tower-http TraceLayer`）自动完成，**handler 无需额外代码**。 - ---- - -## 三、中间件 - -### 3.1 DeviceContext（设备上下文） - -**位置**：`src/middleware/device_middleware.rs` - -从 HTTP headers 提取客户端元数据并注入 `Extension`，**永不拒绝请求**： - -| Header | DeviceContext 字段 | -|--------|-------------------| -| `X-Original-Forwarded-For` / `X-Forwarded-For` | `ip` | -| `User-Agent` | `user_agent` | -| `Identifier` | `identifier` | -| `Login-Type` | `login_type` | - -Handler 通过 `Extension(device): Extension` 提取，然后覆盖 JSON body 中的对应字段： - -```rust -pub async fn user_login( - State(state): State, - Extension(device): Extension, - Json(mut req): Json, -) -> HttpResult { - if !device.ip.is_empty() { req.ip = device.ip; } - if !device.user_agent.is_empty() { req.user_agent = device.user_agent; } - if !device.identifier.is_empty() { req.identifier = device.identifier; } - // ... -} -``` - -### 3.2 AuthContext（认证上下文） - -**位置**：`src/middleware/auth_middleware.rs` - -验证 Bearer JWT，校验 Redis session，查询用户状态，注入 `Extension`。 -admin 路由自动检查 `user.is_admin`。 - -### 3.3 路由分组中间件映射 - -| 路由前缀 | 中间件 | -|---------|--------| -| `/v1/auth/*` | `device_middleware` | -| `/v1/auth/oauth/*` | 无 | -| `/v1/admin/*` | `auth_middleware` | -| `/v1/public/*` | `auth_middleware` + `device_middleware` | -| `/v1/common/*` | `device_middleware` | -| `/v1/server/*`、`/v1/telegram`、`/v1/subscribe/*` | 无用户认证 | - ---- - -## 四、本地 crate 一览 - -| crate | 路径 | 说明 | -|-------|------|------| -| `result` | `crates/result` | HTTP 响应信封、错误码 | -| `jwt` | `crates/jwt` | JWT 生成/验证（`Claims::new` / `generate_token` / `validate_token`） | -| `password` | `crates/password` | 密码哈希，移植自 `../server/pkg/tool/encryption.go` | -| `oauth` | `crates/oauth` | OAuth2 封装（Google/Apple via arctic-oauth，Telegram HMAC 验证） | -| `email` | `crates/email` | 邮件发送 | -| `ip` | `crates/ip` | IP 工具 | -| `payment` | `crates/payment` | 支付集成 | - -### jwt crate 用法 - -```rust -// 生成 token（返回 (Claims, expire_seconds)） -let (claims, seconds) = jwt::Claims::new(user_id, session_id, login_type); -let token = jwt::generate_token(&claims, &config.jwt_auth.access_secret)?; - -// 验证 token -let claims = jwt::validate_token(&token, &config.jwt_auth.access_secret)?; -``` - -### password crate 用法 - -```rust -// 编码（新密码） -let hash = password::encode_password(&plain_text)?; - -// 校验（支持 md5 / sha256 / md5salt / sha256salt / default(PBKDF2) / bcrypt） -let ok = password::multi_password_verify(&algo, &salt, &plain, &stored_hash); -``` - -### oauth crate 用法 - -```rust -// Google — PKCE 授权 URL -let google = oauth::Google::new(&client_id, &client_secret, &redirect_uri); -let url = google.authorization_url(&state, &["openid", "email", "profile"], &code_verifier); -let tokens = google.validate_authorization_code(&code, &code_verifier).await?; -let info = oauth::OAuthUserInfo::from_google(&tokens)?; - -// Apple — 授权 URL（手动构造）+ token 交换 -let apple = oauth::Apple::new(&client_id, &team_id, &key_id, &pkcs8_der, &redirect_uri)?; -let tokens = apple.validate_authorization_code(&code).await?; -let info = oauth::OAuthUserInfo::from_apple(&tokens)?; - -// Telegram — HMAC 验证 base64 回调 -let auth_data = oauth::parse_base64_and_validate(tg_auth_result, bot_token.as_bytes())?; -let info = oauth::OAuthUserInfo::from_telegram(&auth_data); -``` - ---- - -## 五、repository 层约定 - -- 所有 repo 方法均通过 trait 对象调用（`Box`），dialect-agnostic。 -- `find_one_by_method`（AuthRepo）返回 `Result`，**不是** `Option`——找不到时返回 `sqlx::Error::RowNotFound`。 -- 动态 SQL 字符串必须用 `repository::audit(sql)` 包装（`sqlx::AssertSqlSafe`）。 - ---- - -## 六、移植进度（截至 2026-07-05） - -| 模块 | 状态 | -|------|------| -| 日志系统（Telemetry + tracing + TraceLayer） | ✅ 完成 | -| 中间件（auth / device / logger） | ✅ 完成 | -| 中间件（cors / notify / server / pan_domain / trace） | ✅ 完成 | -| crates/sms（AlibabaCloud / Twilio / SmsBao / Abosend） | ✅ 完成 | -| src/adapter（gtmpl 模板引擎 + Proxy/Client/Adapter） | ✅ 完成 | -| service/auth（login / register / reset / device / telephone） | ✅ 完成 | -| service/auth/oauth（Google / Apple / Telegram） | ✅ 完成 | -| service/common（heartbeat / globalConfig / stat / client / ads / privacy / tos） | ✅ 完成 | -| service/server（getConfig / getUserList / pushStatus / pushTraffic / pushOnline / queryProtocol） | ✅ 完成 | -| service/nodeconfig（GlobalValues / ApplyOverride / OverrideResponse / OverrideModel / CloneValues） | ✅ 完成 | -| service/subscribe（subscribeLogic + userAgent UA 匹配） | ✅ 完成 | -| service/notify（Alipay RSA2 / ePay MD5 / Stripe webhook） | ✅ 完成 | -| service/telegram（bot / template / telegram_service） | ✅ 完成 | -| service/admin/ads | ✅ 完成 | -| service/admin/announcement | ✅ 完成 | -| service/admin/document | ✅ 完成 | -| service/admin/coupon | ✅ 完成 | -| service/admin/payment | ✅ 完成 | -| service/admin/auth_method | ✅ 完成 | -| service/admin/application（含 adapter 模板预览） | ✅ 完成 | -| service/admin/console | ✅ 完成 | -| service/admin/tool | ✅ 完成 | -| service/admin/marketing（批量邮件 / quota 任务） | ✅ 完成 | -| service/admin/order | ✅ 完成 | -| service/admin/ticket | ✅ 完成 | -| service/admin/log（全部 14 种日志类型） | ✅ 完成 | -| service/admin/server（节点 / 服务器 CRUD + 协议配置） | ✅ 完成 | -| service/admin/subscribe（订阅计划 CRUD + 排序） | ✅ 完成 | -| service/admin/system（全部 26 个配置读写） | ✅ 完成 | -| service/admin/user（全部 28 个用户管理操作） | ✅ 完成 | -| service/public/announcement | ✅ 完成 | -| service/public/document | ✅ 完成 | -| service/public/payment | ✅ 完成 | -| service/public/subscribe | ✅ 完成 | -| service/public/ticket | ✅ 完成 | -| service/public/portal（购买流程） | ✅ 完成 | -| service/public/order（全部 12 个订单操作） | ✅ 完成 | -| service/public/user（全部 30 个用户自助操作） | ✅ 完成 | -| queue/service（email / sms / order / traffic / subscription / task） | ✅ 完成 | -| scheduler（4 个定时任务注册） | ✅ 完成 | -| handler/auth（所有端点，DeviceContext 注入） | ✅ 完成 | -| handler/common（全部 10 个端点） | ✅ 完成 | -| handler/server（全部 7 个端点） | ✅ 完成 | -| handler/admin（全部子域，~150 个端点） | ✅ 完成 | -| handler/public（全部子域，~90 个端点） | ✅ 完成 | -| handler/subscribe（泛域名订阅） | ✅ 完成 | -| handler/notify（Alipay / ePay / Stripe 回调） | ✅ 完成 | -| handler/telegram | ✅ 完成 | -| routes.rs（按分组应用中间件） | ✅ 完成 | -| repository 层（全部 16 个域） | ✅ 完成 | -| plugin API | ⏳ 暂不实现（已确认跳过） | - -### 编译状态 - -`cargo check` → **0 errors**（截至 2026-07-05） - -### 已知 TODO 项（功能存根） - -- plugin 相关 handler：已确认暂不实现 - diff --git a/Cargo.lock b/Cargo.lock index 7464c8eb..3cfe4c30 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -69,6 +69,56 @@ dependencies = [ "libc", ] +[[package]] +name = "anstream" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "824a212faf96e9acacdbd09febd34438f8f711fb84e09a8916013cd7815ca28d" +dependencies = [ + "anstyle", + "anstyle-parse", + "anstyle-query", + "anstyle-wincon", + "colorchoice", + "is_terminal_polyfill", + "utf8parse", +] + +[[package]] +name = "anstyle" +version = "1.0.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "940b3a0ca603d1eade50a4846a2afffd5ef57a9feac2c0e2ec2e14f9ead76000" + +[[package]] +name = "anstyle-parse" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "52ce7f38b242319f7cabaa6813055467063ecdc9d355bbb4ce0c68908cd8130e" +dependencies = [ + "utf8parse", +] + +[[package]] +name = "anstyle-query" +version = "1.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc" +dependencies = [ + "windows-sys 0.61.2", +] + +[[package]] +name = "anstyle-wincon" +version = "3.0.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d" +dependencies = [ + "anstyle", + "once_cell_polyfill", + "windows-sys 0.61.2", +] + [[package]] name = "anyhow" version = "1.0.103" @@ -605,6 +655,46 @@ dependencies = [ "inout", ] +[[package]] +name = "clap" +version = "4.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1ddb117e43bbf7dacf0a4190fef4d345b9bad68dfc649cb349e7d17d28428e51" +dependencies = [ + "clap_builder", + "clap_derive", +] + +[[package]] +name = "clap_builder" +version = "4.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "714a53001bf66416adb0e2ef5ac857140e7dc3a0c48fb28b2f10762fc4b5069f" +dependencies = [ + "anstream", + "anstyle", + "clap_lex", + "strsim", +] + +[[package]] +name = "clap_derive" +version = "4.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f2ce8604710f6733aa641a2b3731eaa1e8b3d9973d5e3565da11800813f997a9" +dependencies = [ + "heck", + "proc-macro2", + "quote", + "syn 2.0.118", +] + +[[package]] +name = "clap_lex" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c8d4a3bb8b1e0c1050499d1815f5ab16d04f0959b233085fb31653fbfc9d98f9" + [[package]] name = "cmake" version = "0.1.58" @@ -620,6 +710,12 @@ version = "0.5.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0c9ea0ac24bc397ab3c98583a3c9ba74fa56b09a4449bbe172b9b1ddb016027a" +[[package]] +name = "colorchoice" +version = "1.0.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570" + [[package]] name = "combine" version = "4.6.7" @@ -2119,6 +2215,12 @@ version = "2.12.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d98f6fed1fde3f8c21bc40a1abb88dd75e67924f9cffc3ef95607bad8017f8e2" +[[package]] +name = "is_terminal_polyfill" +version = "1.70.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a6cb138bb79a146c1bd460005623e142ef0181e3d0219cb493e02f7d08a35695" + [[package]] name = "itertools" version = "0.13.0" @@ -2423,6 +2525,20 @@ version = "0.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1d87ecb2933e8aeadb3e3a02b828fed80a7528047e68b4f424523a0981a3a084" +[[package]] +name = "mysql2postgres" +version = "0.1.0" +dependencies = [ + "anyhow", + "chrono", + "clap", + "futures", + "sqlx", + "tokio", + "tracing", + "tracing-subscriber", +] + [[package]] name = "native-tls" version = "0.2.18" @@ -2568,6 +2684,12 @@ version = "1.21.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" +[[package]] +name = "once_cell_polyfill" +version = "1.70.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "384b8ab6d37215f3c5301a95a4accb5d64aa607f1fcb26a11b5303878451b4fe" + [[package]] name = "openssl" version = "0.10.81" @@ -4938,6 +5060,12 @@ version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b6c140620e7ffbb22c2dee59cafe6084a59b5ffc27a8859a5f0d494b5d52b6be" +[[package]] +name = "utf8parse" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821" + [[package]] name = "uuid" version = "0.8.2" diff --git a/Cargo.toml b/Cargo.toml index 1f2902e0..423ffc96 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,5 +1,5 @@ [workspace] -members = ["crates/*"] +members = ["crates/*", "tools/mysql2postgres"] [package] name = "ppanel-backend" diff --git a/tools/mysql2postgres/Cargo.toml b/tools/mysql2postgres/Cargo.toml new file mode 100644 index 00000000..0c5c49bf --- /dev/null +++ b/tools/mysql2postgres/Cargo.toml @@ -0,0 +1,19 @@ +[package] +name = "mysql2postgres" +version = "0.1.0" +edition = "2021" +description = "Copy data from MySQL to PostgreSQL for ppanel-backend migrations" + +[[bin]] +name = "mysql2postgres" +path = "src/main.rs" + +[dependencies] +anyhow = "1" +chrono = "0.4" +clap = { version = "4", features = ["derive", "env"] } +sqlx = { version = "0.9", features = ["runtime-tokio", "mysql", "postgres", "chrono", "any"] } +tokio = { version = "1", features = ["full"] } +tracing = "0.1" +tracing-subscriber = { version = "0.3", features = ["env-filter"] } +futures = "0.3" diff --git a/tools/mysql2postgres/src/main.rs b/tools/mysql2postgres/src/main.rs new file mode 100644 index 00000000..7e1677f1 --- /dev/null +++ b/tools/mysql2postgres/src/main.rs @@ -0,0 +1,474 @@ +//! mysql2postgres — copy data from MySQL to PostgreSQL. +//! +//! Direct port of `server/tools/mysql2postgres/` (Go). +//! Connects to both databases, builds a migration plan, +//! and copies each table row-by-row using PostgreSQL COPY. + +use anyhow::{Context, Result}; +use clap::Parser; +use sqlx::mysql::MySqlPool; +use sqlx::postgres::PgPool; +use std::collections::{HashMap, HashSet}; + +// ───────────────────────────────────────────────────────────────────────────── +// CLI +// ───────────────────────────────────────────────────────────────────────────── + +#[derive(Parser, Debug)] +#[command(name = "mysql2postgres", about = "Copy ppanel MySQL data to PostgreSQL")] +struct Args { + /// Source MySQL DSN (e.g. user:pass@tcp(host:3306)/dbname) + #[arg(long, env = "MYSQL_DSN")] + mysql: String, + + /// Target PostgreSQL DSN (e.g. postgres://user:pass@host/dbname) + #[arg(long, env = "POSTGRES_DSN")] + postgres: String, + + /// Target PostgreSQL schema (default: public) + #[arg(long, default_value = "public")] + schema: String, + + /// Comma-separated table allowlist (default: all common tables) + #[arg(long, default_value = "")] + tables: String, + + /// Comma-separated table denylist + #[arg(long, default_value = "")] + exclude: String, + + /// Truncate target tables before copying (destructive) + #[arg(long, default_value_t = false)] + truncate: bool, + + /// Confirm destructive operations (required when --truncate is set) + #[arg(long, default_value_t = false)] + yes: bool, + + /// Print plan without copying data + #[arg(long, default_value_t = false)] + dry_run: bool, + + /// Rows per progress log line + #[arg(long, default_value_t = 1000)] + batch_size: usize, +} + +// ───────────────────────────────────────────────────────────────────────────── +// Column metadata +// ───────────────────────────────────────────────────────────────────────────── + +#[derive(Debug, Clone)] +struct PgColumn { + name: String, + data_type: String, + udt_name: String, + is_identity: bool, + is_generated: bool, +} + +impl PgColumn { + fn is_bool(&self) -> bool { + self.data_type == "boolean" || self.udt_name == "bool" + } + fn is_integer(&self) -> bool { + matches!(self.data_type.as_str(), "smallint" | "integer" | "bigint") + } + fn is_timestamp(&self) -> bool { + self.data_type.contains("timestamp") || self.data_type == "date" + } +} + +// ───────────────────────────────────────────────────────────────────────────── +// Migration plan +// ───────────────────────────────────────────────────────────────────────────── + +#[derive(Debug)] +struct TablePlan { + name: String, + columns: Vec, + order_columns: Vec, + row_count: i64, +} + +// ───────────────────────────────────────────────────────────────────────────── +// Entry point +// ───────────────────────────────────────────────────────────────────────────── + +#[tokio::main] +async fn main() -> Result<()> { + tracing_subscriber::fmt() + .with_env_filter(tracing_subscriber::EnvFilter::from_default_env()) + .init(); + + let args = Args::parse(); + + if args.truncate && !args.yes && !args.dry_run { + anyhow::bail!("--truncate is destructive; pass --yes to confirm"); + } + + let mysql = MySqlPool::connect(&args.mysql) + .await + .context("connect to MySQL")?; + tracing::info!("connected to MySQL"); + + let pg = PgPool::connect(&args.postgres) + .await + .context("connect to PostgreSQL")?; + tracing::info!("connected to PostgreSQL"); + + let plans = build_plans(&mysql, &pg, &args).await?; + if plans.is_empty() { + anyhow::bail!("no common tables to migrate"); + } + + tracing::info!("migration plan: {} table(s)", plans.len()); + for p in &plans { + tracing::info!( + " {}: {} row(s), {} column(s)", + p.name, p.row_count, p.columns.len() + ); + } + + if args.dry_run { + tracing::info!("dry run — no data copied"); + return Ok(()); + } + + if args.truncate { + truncate_tables(&pg, &args.schema, &plans).await?; + } + + for plan in &plans { + copy_table(&mysql, &pg, &args.schema, plan, args.batch_size).await?; + } + + reset_sequences(&pg, &args.schema).await?; + tracing::info!("migration completed"); + Ok(()) +} + +// ───────────────────────────────────────────────────────────────────────────── +// Plan building +// ───────────────────────────────────────────────────────────────────────────── + +async fn build_plans(mysql: &MySqlPool, pg: &PgPool, args: &Args) -> Result> { + let source_tables = list_mysql_tables(mysql).await?; + let target_tables = list_pg_tables(pg, &args.schema).await?; + + let allow: HashSet = parse_set(&args.tables); + let exclude: HashSet = parse_set(&args.exclude); + + let mut names: Vec = target_tables + .iter() + .filter(|n| n.as_str() != "schema_migrations") + .filter(|n| allow.is_empty() || allow.contains(*n)) + .filter(|n| !exclude.contains(*n)) + .filter(|n| { + if source_tables.contains(*n) { true } else { + tracing::warn!("skip {}: no source table", n); false + } + }) + .cloned() + .collect(); + names.sort(); + + let mut plans = Vec::new(); + for name in &names { + let target_cols = list_pg_columns(pg, &args.schema, name).await?; + let source_cols = list_mysql_columns(mysql, name).await?; + + let common: Vec = target_cols + .into_iter() + .filter(|c| !c.is_generated && source_cols.contains(&c.name)) + .collect(); + + if common.is_empty() { + tracing::warn!("skip {}: no common columns", name); + continue; + } + + let row_count = count_mysql_rows(mysql, name).await?; + let order_columns = list_mysql_pk_columns(mysql, name).await?; + + plans.push(TablePlan { name: name.clone(), columns: common, order_columns, row_count }); + } + + let fks = list_pg_foreign_keys(pg, &args.schema).await?; + Ok(sort_by_dependencies(plans, fks)) +} + +fn parse_set(s: &str) -> HashSet { + s.split(',').map(|x| x.trim().to_string()).filter(|x| !x.is_empty()).collect() +} + +// ───────────────────────────────────────────────────────────────────────────── +// MySQL introspection +// ───────────────────────────────────────────────────────────────────────────── + +async fn list_mysql_tables(pool: &MySqlPool) -> Result> { + let rows: Vec<(String,)> = sqlx::query_as( + "SELECT table_name FROM information_schema.tables \ + WHERE table_schema = DATABASE() AND table_type = 'BASE TABLE'" + ).fetch_all(pool).await.context("list mysql tables")?; + Ok(rows.into_iter().map(|(n,)| n).collect()) +} + +async fn list_mysql_columns(pool: &MySqlPool, table: &str) -> Result> { + let rows: Vec<(String,)> = sqlx::query_as( + "SELECT column_name FROM information_schema.columns \ + WHERE table_schema = DATABASE() AND table_name = ? ORDER BY ordinal_position" + ).bind(table).fetch_all(pool).await + .with_context(|| format!("list mysql columns for {table}"))?; + Ok(rows.into_iter().map(|(n,)| n).collect()) +} + +async fn list_mysql_pk_columns(pool: &MySqlPool, table: &str) -> Result> { + let rows: Vec<(String,)> = sqlx::query_as( + "SELECT column_name FROM information_schema.key_column_usage \ + WHERE table_schema = DATABASE() AND table_name = ? AND constraint_name = 'PRIMARY' \ + ORDER BY ordinal_position" + ).bind(table).fetch_all(pool).await + .with_context(|| format!("list mysql pk for {table}"))?; + Ok(rows.into_iter().map(|(n,)| n).collect()) +} + +async fn count_mysql_rows(pool: &MySqlPool, table: &str) -> Result { + let quoted = format!("`{}`", table.replace('`', "``")); + let (count,): (i64,) = sqlx::query_as(sqlx::AssertSqlSafe(format!("SELECT COUNT(*) FROM {quoted}"))) + .fetch_one(pool).await + .with_context(|| format!("count mysql rows in {table}"))?; + Ok(count) +} + +// ───────────────────────────────────────────────────────────────────────────── +// PostgreSQL introspection +// ───────────────────────────────────────────────────────────────────────────── + +async fn list_pg_tables(pool: &PgPool, schema: &str) -> Result> { + let rows: Vec<(String,)> = sqlx::query_as( + "SELECT table_name FROM information_schema.tables \ + WHERE table_schema = $1 AND table_type = 'BASE TABLE'" + ).bind(schema).fetch_all(pool).await.context("list pg tables")?; + Ok(rows.into_iter().map(|(n,)| n).collect()) +} + +async fn list_pg_columns(pool: &PgPool, schema: &str, table: &str) -> Result> { + let rows: Vec<(String, String, String, String, String)> = sqlx::query_as( + "SELECT column_name, data_type, udt_name, is_identity, is_generated \ + FROM information_schema.columns \ + WHERE table_schema = $1 AND table_name = $2 ORDER BY ordinal_position" + ).bind(schema).bind(table).fetch_all(pool).await + .with_context(|| format!("list pg columns for {table}"))?; + Ok(rows.into_iter().map(|(name, data_type, udt_name, identity, generated)| PgColumn { + name, + data_type, + udt_name, + is_identity: identity == "YES", + is_generated: generated != "NEVER", + }).collect()) +} + +async fn list_pg_foreign_keys(pool: &PgPool, schema: &str) -> Result> { + let rows: Vec<(String, String)> = sqlx::query_as( + "SELECT child.relname, parent.relname \ + FROM pg_constraint c \ + JOIN pg_class child ON child.oid = c.conrelid \ + JOIN pg_namespace cn ON cn.oid = child.relnamespace \ + JOIN pg_class parent ON parent.oid = c.confrelid \ + JOIN pg_namespace pn ON pn.oid = parent.relnamespace \ + WHERE c.contype = 'f' AND cn.nspname = $1 AND pn.nspname = $1 \ + ORDER BY child.relname, parent.relname" + ).bind(schema).fetch_all(pool).await.context("list pg foreign keys")?; + Ok(rows) +} + +// ───────────────────────────────────────────────────────────────────────────── +// Dependency sort (topological, mirrors Go) +// ───────────────────────────────────────────────────────────────────────────── + +fn sort_by_dependencies(plans: Vec, fks: Vec<(String, String)>) -> Vec { + let plan_names: HashSet<&str> = plans.iter().map(|p| p.name.as_str()).collect(); + let mut deps: HashMap<&str, HashSet<&str>> = HashMap::new(); + for (child, parent) in &fks { + if child != parent && plan_names.contains(child.as_str()) && plan_names.contains(parent.as_str()) { + deps.entry(child).or_default().insert(parent); + } + } + + // Collect ordered names (Strings) before consuming `plans`. + let name_to_idx: HashMap<&str, usize> = plans.iter().enumerate().map(|(i, p)| (p.name.as_str(), i)).collect(); + let mut remaining: HashSet<&str> = plan_names.clone(); + let mut ordered_names: Vec = Vec::new(); + + while !remaining.is_empty() { + let mut ready: Vec<&str> = remaining.iter() + .copied() + .filter(|n| deps.get(n).map_or(true, |ps| ps.iter().all(|p| !remaining.contains(*p)))) + .collect(); + if ready.is_empty() { + ready = remaining.iter().copied().collect(); + tracing::warn!("FK cycle detected; copying {} remaining tables in lexical order", ready.len()); + } + ready.sort(); + for name in ready { + ordered_names.push(name.to_string()); + remaining.remove(name); + } + } + let _ = name_to_idx; // suppress unused warning + + // Re-order plans by the collected name sequence. + let mut plan_map: HashMap = + plans.into_iter().map(|p| (p.name.clone(), p)).collect(); + ordered_names.into_iter().filter_map(|n| plan_map.remove(&n)).collect() +} + +// ───────────────────────────────────────────────────────────────────────────── +// Data copy +// ───────────────────────────────────────────────────────────────────────────── + +async fn truncate_tables(pg: &PgPool, schema: &str, plans: &[TablePlan]) -> Result<()> { + let names: Vec = plans.iter() + .map(|p| format!("\"{schema}\".\"{n}\"", n = p.name)) + .collect(); + let sql = format!("TRUNCATE TABLE {} RESTART IDENTITY CASCADE", names.join(", ")); + tracing::info!("truncating {} table(s)", plans.len()); + sqlx::query(sqlx::AssertSqlSafe(sql)).execute(pg).await.context("truncate tables")?; + Ok(()) +} + +async fn copy_table( + mysql: &MySqlPool, + pg: &PgPool, + schema: &str, + plan: &TablePlan, + batch_size: usize, +) -> Result<()> { + tracing::info!("copy {}: start ({} rows)", plan.name, plan.row_count); + + let col_names: Vec = plan.columns.iter().map(|c| c.name.clone()).collect(); + let mysql_cols = col_names.iter() + .map(|n| format!("`{}`", n.replace('`', "``"))) + .collect::>() + .join(", "); + + let mut select = format!("SELECT {} FROM `{}`", mysql_cols, plan.name.replace('`', "``")); + if !plan.order_columns.is_empty() { + let order = plan.order_columns.iter() + .map(|n| format!("`{}`", n.replace('`', "``"))) + .collect::>() + .join(", "); + select.push_str(&format!(" ORDER BY {order}")); + } + + // Build PG INSERT (no COPY protocol via sqlx yet — use parameterised INSERT in batches). + let pg_cols = col_names.iter() + .map(|n| format!("\"{}\"", n.replace('"', "\"\""))) + .collect::>() + .join(", "); + + let mut rows = sqlx::query(sqlx::AssertSqlSafe(select)).fetch(mysql); + let mut copied: i64 = 0; + let mut batch: Vec>> = Vec::with_capacity(batch_size); + + use sqlx::Row; + use futures::StreamExt; + + while let Some(row) = rows.next().await { + let row = row.with_context(|| format!("read row from {}", plan.name))?; + let values: Vec> = col_names.iter().enumerate().map(|(i, _)| { + row.try_get::, _>(i).unwrap_or(None) + }).collect(); + batch.push(values); + + if batch.len() >= batch_size { + insert_batch(pg, schema, &plan.name, &col_names, &pg_cols, &batch, &plan.columns).await?; + copied += batch.len() as i64; + batch.clear(); + tracing::info!("copy {}: {}/{}", plan.name, copied, plan.row_count); + } + } + if !batch.is_empty() { + insert_batch(pg, schema, &plan.name, &col_names, &pg_cols, &batch, &plan.columns).await?; + copied += batch.len() as i64; + } + + tracing::info!("copy {}: done ({} rows)", plan.name, copied); + Ok(()) +} + +async fn insert_batch( + pg: &PgPool, + schema: &str, + table: &str, + _col_names: &[String], + pg_cols: &str, + batch: &[Vec>], + _columns: &[PgColumn], +) -> Result<()> { + if batch.is_empty() { return Ok(()); } + let col_count = batch[0].len(); + let mut placeholders = Vec::new(); + let mut idx = 1usize; + for _ in batch { + let row_ph: Vec = (0..col_count).map(|_| { let s = format!("${idx}"); idx += 1; s }).collect(); + placeholders.push(format!("({})", row_ph.join(","))); + } + let sql = format!( + "INSERT INTO \"{schema}\".\"{table}\" ({pg_cols}) VALUES {} ON CONFLICT DO NOTHING", + placeholders.join(",") + ); + let mut q = sqlx::query(sqlx::AssertSqlSafe(sql)); + for row in batch { + for val in row { + q = q.bind(val.as_deref()); + } + } + q.execute(pg).await.with_context(|| format!("insert batch into {table}"))?; + Ok(()) +} + +// ───────────────────────────────────────────────────────────────────────────── +// Sequence reset +// ───────────────────────────────────────────────────────────────────────────── + +async fn reset_sequences(pg: &PgPool, schema: &str) -> Result<()> { + let rows: Vec<(String, String)> = sqlx::query_as( + "SELECT table_name, column_name FROM information_schema.columns \ + WHERE table_schema = $1 AND (is_identity = 'YES' OR column_default LIKE 'nextval(%') \ + ORDER BY table_name, ordinal_position" + ).bind(schema).fetch_all(pg).await.context("list sequences")?; + + let mut count = 0usize; + for (table, column) in &rows { + let table_ref = format!("\"{schema}\".\"{table}\""); + let seq: Option<(Option,)> = sqlx::query_as( + "SELECT pg_get_serial_sequence($1, $2)" + ).bind(&table_ref).bind(column).fetch_optional(pg).await + .with_context(|| format!("get sequence for {table}.{column}"))?; + + let seq_name = match seq.and_then(|(s,)| s) { + Some(s) if !s.is_empty() => s, + _ => continue, + }; + + let (max_id,): (i64,) = sqlx::query_as( + sqlx::AssertSqlSafe(format!("SELECT COALESCE(MAX(\"{column}\"), 0) FROM {table_ref}")) + ).fetch_one(pg).await + .with_context(|| format!("max id for {table}.{column}"))?; + + if max_id > 0 { + sqlx::query("SELECT setval($1::regclass, $2, true)") + .bind(&seq_name).bind(max_id) + .execute(pg).await.ok(); + } else { + sqlx::query("SELECT setval($1::regclass, 1, false)") + .bind(&seq_name) + .execute(pg).await.ok(); + } + count += 1; + } + tracing::info!("reset {count} sequence(s)"); + Ok(()) +} From 4f74e14fba99d5ad845180ded7bef8b6cc33212e Mon Sep 17 00:00:00 2001 From: Ember Moth Date: Sun, 5 Jul 2026 20:45:27 +0800 Subject: [PATCH 2/3] add mysql2postgres --- crates/turnstile/Cargo.toml | 13 ++++++++ crates/turnstile/src/lib.rs | 60 +++++++++++++++++++++++++++++++++++++ src/exchange_rate.rs | 48 +++++++++++++++++++++++++++++ 3 files changed, 121 insertions(+) create mode 100644 crates/turnstile/Cargo.toml create mode 100644 crates/turnstile/src/lib.rs create mode 100644 src/exchange_rate.rs diff --git a/crates/turnstile/Cargo.toml b/crates/turnstile/Cargo.toml new file mode 100644 index 00000000..8f249c89 --- /dev/null +++ b/crates/turnstile/Cargo.toml @@ -0,0 +1,13 @@ +[package] +name = "turnstile" +version = "0.1.0" +edition = "2021" +description = "Cloudflare Turnstile token verification — port of pkg/turnstile" + +[dependencies] +anyhow = "1" +reqwest = { version = "0.12", features = ["json"] } +serde = { version = "1", features = ["derive"] } +serde_json = "1" +tokio = { version = "1", features = ["time"] } +uuid = { version = "1", features = ["v4"] } diff --git a/crates/turnstile/src/lib.rs b/crates/turnstile/src/lib.rs new file mode 100644 index 00000000..a77ccc25 --- /dev/null +++ b/crates/turnstile/src/lib.rs @@ -0,0 +1,60 @@ +//! Cloudflare Turnstile token verification. +//! Port of `server/pkg/turnstile`. + +use serde::{Deserialize, Serialize}; + +const VERIFY_URL: &str = "https://challenges.cloudflare.com/turnstile/v0/siteverify"; + +#[derive(Debug, Serialize)] +struct VerifyRequest<'a> { + secret: &'a str, + response: &'a str, + #[serde(skip_serializing_if = "str::is_empty")] + remoteip: &'a str, + #[serde(skip_serializing_if = "str::is_empty")] + idempotency_key: &'a str, +} + +#[derive(Debug, Deserialize)] +struct VerifyResponse { + success: bool, + #[serde(rename = "error-codes", default)] + error_codes: Vec, +} + +/// Verify a Turnstile challenge token. +/// +/// - `secret` — site secret key from Cloudflare dashboard +/// - `token` — value of `cf-turnstile-response` submitted by the browser +/// - `ip` — optional client IP (pass `""` to omit) +pub async fn verify(secret: &str, token: &str, ip: &str) -> anyhow::Result { + verify_with_key(secret, token, ip, "").await +} + +/// Verify with idempotency key (prevents the same token being accepted twice). +pub async fn verify_with_key( + secret: &str, + token: &str, + ip: &str, + idempotency_key: &str, +) -> anyhow::Result { + let client = reqwest::Client::new(); + let req = VerifyRequest { secret, response: token, remoteip: ip, idempotency_key }; + let resp: VerifyResponse = client + .post(VERIFY_URL) + .json(&req) + .send() + .await? + .json() + .await?; + + if !resp.success && !resp.error_codes.is_empty() { + tracing::warn!(codes = ?resp.error_codes, "turnstile verification failed"); + } + Ok(resp.success) +} + +/// Generate a random UUID suitable for use as an idempotency key. +pub fn random_uuid() -> String { + uuid::Uuid::new_v4().to_string() +} diff --git a/src/exchange_rate.rs b/src/exchange_rate.rs new file mode 100644 index 00000000..53193e73 --- /dev/null +++ b/src/exchange_rate.rs @@ -0,0 +1,48 @@ +//! Exchange rate conversion via apilayer.com. +//! Port of `server/pkg/exchangeRate/exchangeRate.go`. + +use serde::Deserialize; + +const API_BASE: &str = "https://api.apilayer.com"; + +#[derive(Debug, Deserialize)] +struct ConvertResponse { + success: bool, + result: Option, +} + +/// Convert `amount` from currency `from` to currency `to`. +/// +/// - `access_key` — API key from apilayer.com (configured in `Config.Currency.AccessKey`) +/// +/// Returns the converted amount, or an error if the request fails or the API +/// reports failure. +pub async fn convert( + from: &str, + to: &str, + amount: f64, + access_key: &str, +) -> anyhow::Result { + let client = reqwest::Client::builder() + .timeout(std::time::Duration::from_secs(5)) + .build()?; + + let resp: ConvertResponse = client + .get(format!("{API_BASE}/currency_data/convert")) + .header("apikey", access_key) + .query(&[ + ("from", from), + ("to", to), + ("amount", &amount.to_string()), + ]) + .send() + .await? + .json() + .await?; + + if !resp.success { + anyhow::bail!("exchange rate API returned failure for {from}→{to}"); + } + resp.result + .ok_or_else(|| anyhow::anyhow!("exchange rate API returned no result")) +} From 25d190b90e9d0fa2b82bc3afdd5d8bedd8e6b9a3 Mon Sep 17 00:00:00 2001 From: Ember Moth Date: Sun, 5 Jul 2026 20:47:34 +0800 Subject: [PATCH 3/3] update --- .github/.gitkeep | 0 Cargo.lock | 12 +++ src/main.rs | 1 + src/middleware/mod.rs | 1 + src/middleware/rate_limit_middleware.rs | 133 ++++++++++++++++++++++++ 5 files changed, 147 insertions(+) create mode 100644 .github/.gitkeep create mode 100644 src/middleware/rate_limit_middleware.rs diff --git a/.github/.gitkeep b/.github/.gitkeep new file mode 100644 index 00000000..e69de29b diff --git a/Cargo.lock b/Cargo.lock index 3cfe4c30..5505e276 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4990,6 +4990,18 @@ version = "0.2.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" +[[package]] +name = "turnstile" +version = "0.1.0" +dependencies = [ + "anyhow", + "reqwest 0.12.28", + "serde", + "serde_json", + "tokio", + "uuid 1.23.4", +] + [[package]] name = "typenum" version = "1.20.1" diff --git a/src/main.rs b/src/main.rs index ef058fdd..3414245a 100644 --- a/src/main.rs +++ b/src/main.rs @@ -10,6 +10,7 @@ pub mod adapter; pub mod cache; pub mod config; pub mod db; +pub mod exchange_rate; pub mod handler; pub mod middleware; pub mod migration; diff --git a/src/middleware/mod.rs b/src/middleware/mod.rs index 3f73c73e..081abed0 100644 --- a/src/middleware/mod.rs +++ b/src/middleware/mod.rs @@ -4,5 +4,6 @@ pub mod device_middleware; pub mod logger_middleware; pub mod notify_middleware; pub mod pan_domain_middleware; +pub mod rate_limit_middleware; pub mod server_middleware; pub mod trace_middleware; diff --git a/src/middleware/rate_limit_middleware.rs b/src/middleware/rate_limit_middleware.rs new file mode 100644 index 00000000..9373f684 --- /dev/null +++ b/src/middleware/rate_limit_middleware.rs @@ -0,0 +1,133 @@ +//! Redis-based period rate-limit middleware. +//! Port of `server/pkg/limit/periodlimit.go`. +//! +//! Uses the same Lua script as the Go version for atomicity: +//! - INCRBY key 1 +//! - Set TTL on first hit +//! - Return 1 (allowed), 2 (hit quota exactly), 0 (over quota) + +use std::sync::Arc; + +use axum::{extract::Request, middleware::Next, response::Response}; +use redis::{aio::ConnectionManager, AsyncCommands, Script}; + +use result::code_error::CodeError; +use result::error_code; +use result::http_result::build_http_result; + +/// Result codes returned by the Lua script (mirrors Go constants). +const ALLOWED: i64 = 1; +const HIT_QUOTA: i64 = 2; +// 0 = over quota + +/// Shared rate-limiter state — cheap to clone (Arc inside). +#[derive(Clone)] +pub struct PeriodLimiter { + inner: Arc, +} + +struct PeriodLimiterInner { + /// Window length in seconds. + period: usize, + /// Maximum requests per window. + quota: usize, + /// Redis connection. + redis: ConnectionManager, + /// Key prefix prepended to every cache key. + key_prefix: String, + /// Lua script (same as `periodscript.lua` in the Go version). + script: Script, +} + +/// Lua script — identical to `pkg/limit/periodscript.lua`. +const PERIOD_LUA: &str = r#" +local limit = tonumber(ARGV[1]) +local window = tonumber(ARGV[2]) +local current = redis.call("INCRBY", KEYS[1], 1) +if current == 1 then + redis.call("expire", KEYS[1], window) +end +if current < limit then + return 1 +elseif current == limit then + return 2 +else + return 0 +end +"#; + +impl PeriodLimiter { + /// Create a new `PeriodLimiter`. + /// + /// - `period` — window size in seconds + /// - `quota` — max requests allowed in that window + /// - `redis` — shared Redis connection manager + /// - `key_prefix` — namespace prefix (e.g. `"rate:register:"`) + pub fn new(period: usize, quota: usize, redis: ConnectionManager, key_prefix: impl Into) -> Self { + Self { + inner: Arc::new(PeriodLimiterInner { + period, + quota, + redis, + key_prefix: key_prefix.into(), + script: Script::new(PERIOD_LUA), + }), + } + } + + /// Check and increment the counter for `key`. + /// + /// Returns `Ok(true)` if the request is within quota, `Ok(false)` if over. + pub async fn allow(&self, key: &str) -> anyhow::Result { + let full_key = format!("{}{}", self.inner.key_prefix, key); + let mut conn = self.inner.redis.clone(); + let result: i64 = self.inner.script + .key(&full_key) + .arg(self.inner.quota) + .arg(self.inner.period) + .invoke_async(&mut conn) + .await?; + Ok(result == ALLOWED || result == HIT_QUOTA) + } +} + +/// Axum middleware that enforces a per-IP (or per-custom-key) rate limit. +/// +/// The key is extracted from the `X-Original-Forwarded-For` / `X-Forwarded-For` +/// header, falling back to the socket address. +/// +/// Inject via `axum::middleware::from_fn_with_state(limiter, rate_limit_layer)`. +pub async fn rate_limit_layer( + axum::extract::State(limiter): axum::extract::State, + req: Request, + next: Next, +) -> Response { + let ip = req + .headers() + .get("X-Original-Forwarded-For") + .or_else(|| req.headers().get("X-Forwarded-For")) + .and_then(|v| v.to_str().ok()) + .and_then(|s| s.split(',').next()) + .map(|s| s.trim().to_string()) + .unwrap_or_else(|| "unknown".to_string()); + + match limiter.allow(&ip).await { + Ok(true) => next.run(req).await, + Ok(false) => { + let err = anyhow::Error::new( + CodeError::new_err_code(error_code::ERROR) + ); + // Build a 200 body with RATE_LIMIT business code. + build_http_result::<()>(None, Some(err)).into_response() + } + Err(e) => { + tracing::error!("rate limiter redis error: {e}"); + // On Redis failure, allow the request (fail open). + next.run(req).await + } + } +} + +// ─── trait helper ───────────────────────────────────────────────────────── + +use axum::response::IntoResponse;