Compare commits

..

No commits in common. "9e935fdda9b2d3c2c524e3b16fd93acff04e2991" and "401f84bda69cffd661bbcaace2a28067c00bdd20" have entirely different histories.

8 changed files with 145 additions and 998 deletions

View File

@ -266,105 +266,16 @@ const CABINETS_STATUS_COMMENT_ALIGN: (&str, &str) = (
r#"ALTER TABLE cabinets MODIFY COLUMN status TINYINT DEFAULT 0 COMMENT '启用/禁用配置态(不表示在线):1启用 2禁用(拦截指令链路) 0=建表默认值产生的取值,语义等同启用'"#,
);
/// 执行数据库迁移。
///
/// **本函数只做编排**:五个阶段各自封装在下方私有函数里,**顺序即依赖顺序,不得调整**
/// (历史上改错过一次顺序:`SCHEMA_SQL` 前向引用导致新库初始化报 MySQL 1824 无限重启)。
///
/// 每个阶段的"失败是否中断启动"口径**各不相同且是刻意设计的**,见各阶段函数的文档说明:
/// 结构性前提(建表/索引/种子数据)失败必须中断;纯元信息/纯优化/业务侧已有兜底的失败
/// 一律只记 `ERROR` 后继续 —— 因为 `run()` 返回 Err 会让 `main.rs` 的
/// `.expect("数据库迁移失败")` panic 成 systemd 崩溃循环。
/// 执行数据库迁移,创建所有必要的数据表
pub async fn run(pool: &MySqlPool) -> Result<(), sqlx::Error> {
apply_schema(pool).await?;
align_column_metadata(pool).await?;
create_indexes_and_columns(pool).await?;
align_legacy_structures(pool).await?;
seed_defaults(pool).await?;
tracing::info!("数据库迁移完成");
Ok(())
}
/// 阶段 A:执行全量建表语句(`SCHEMA_SQL`)。
///
/// ⛔ **失败必须中断启动**:表都不在,后续任何查询都无从谈起。
///
/// ⚠️ `SCHEMA_SQL` 里建表语句的**书写顺序即依赖顺序**(被引用表必须先建),
/// 否则新库初始化会报 MySQL **1824**(外键找不到被引用表),
/// 叠加 `Restart=always` = **无限重启**。改 `SCHEMA_SQL` 时必须同步
/// `LEGACY_COLUMN_TARGETS`(有单测守门)。
async fn apply_schema(pool: &MySqlPool) -> Result<(), sqlx::Error> {
// 逐条执行全量建表语句,避免多语句执行问题
for statement in SCHEMA_SQL.split(';').map(|s| s.trim()).filter(|s| !s.is_empty()) {
sqlx::query(statement).execute(pool).await?;
}
Ok(())
}
/// 阶段 B:对齐列的**可空性与注释**(老库与 `SCHEMA_SQL` 的元信息口径)。
///
/// 顺序即原实现顺序,且**每一步的容错口径单独保留**。
///
/// # 口径:纯元信息失败只记 ERROR,绝不阻断启动
///
/// 注释只是元信息,不配享有"迁移失败 → main panic → systemd 崩溃循环"的杀伤力
/// (`run()` 返回 Err 会让 `main.rs` 的 `.expect("数据库迁移失败")` panic)。
/// 例外是 `charge_records.max_temperature` / `charge_orders.reason` /
/// `compartments.status` / `charge_orders.status` 四条 —— 它们从最初就是**上抛**口径,保持不变。
///
/// # 各列为什么要改(逐条复核结论,原文保留)
///
/// ## `charge_records.max_temperature`:`0.1°C` → `K×100`
/// 只在该列注释确实过时(老库)时才 ALTER,避免每次启动都对整表做一次 MODIFY。
///
/// ## `charge_orders.reason`:新增 `timeout` 枚举值
/// 订单超过 10 小时由 `order_timeout` 定时任务自动结束,落该枚举值。
///
/// ## `compartments.status`:**旧注释是错的**
/// 老库写的是 `0空闲 1充电中 2故障 3禁用`,但全仓没有任何地方会写 2/3 ——
/// 平台只写 0/1(`workers/report_worker.rs::mark_compartment_status` 的调用点只有 0 与 1);
/// 而"禁用"是设备/组织级状态,仓体级并不存在。
/// 本列语义从此**只表达占用**:"充满 / 故障"由 Redis 通道实时状态派生
/// (`routes/cabinet_tree.rs::derive_channel_flags`),不落库。
///
/// ## `cabinets.status`:**本列是"启用/禁用"配置态,不是"在线"**
/// 全仓复核结论(grep 实测):
/// - **唯一写入方** `routes/cabinets.rs::set_cabinet_status`:
/// `UPDATE cabinets SET status = ? WHERE id = ?`,语义是 `2` = 禁用 / 其它值 = 启用;
/// - **没有任何地方写 0**(0 只可能由建表默认值 `DEFAULT 0` 或该接口传入 0 时产生);
/// - 消费方只有"禁用"一档:`commands.rs::CABINET_DISABLED_SQL`
/// (`SELECT COUNT(*) > 0 FROM cabinets WHERE id = ? AND status = 2`)与
/// `routes/h5.rs::ensure_cabinet_enabled` 据此拦截开合闸 / 开门 / 充电。
///
/// 旧注释 `0离线 1在线 2禁用` 里的"**在线**"是**语义错配**:
/// 生产实测 68 台柜子 `status` 全 = 1,而当时**一台设备都没连** ——
/// 按它统计在线(`energy_stats` 的 `online_count`、H5 首页 `online_cabinets`)
/// 会恒定显示"全部在线"。在线已全部改为 **Redis 判据**
/// (`routes/cabinet_tree.rs::device_online_flag` / `count_online_devices`,
/// 与设备列表 `is_online` 同源),本列不再参与任何在线判定。
/// 只改注释,**不动类型与默认值**(仍是 `TINYINT DEFAULT 0`)。
///
/// ## `charge_orders.status`:**旧注释把从未写入的 `2` 写成了既成事实**
/// 实测(全仓 grep):**生产代码只有 0 与 1 两个写入值** ——
/// 建单 `report_worker.rs::try_create_order` 固定写 `0`;所有结束路径
/// (`off` / `bat_out` / 换电池 / 10 小时超时 / 无电池对账)都走
/// `finalize_order*`,`order_status` 恒为 `ORDER_STATUS_DONE = 1`。
/// 唯一的 `2` 出现在 `test/reset_fixture.py`(一次性夹具脚本,非交付物)。
///
/// 但 `2` **有消费方**:PC 订单页 `web/src/pages/charge-orders/index.tsx` 的筛选下拉
/// 提供 `<Option value={2}>异常</Option>`、`STATUS_COLOR[2] = 'red'`,
/// 且 `routes/charge_orders.rs::status_text` 把 `2` 映射为"异常" ——
/// 即"异常"曾被设计成一个**状态**,而现实现把它降级成 `reason` 里的**结束原因**。
/// 这是**产品语义缺口,不是笔误**,故**不删除 `2`**,只在注释里注明"保留未使用",
/// 并作为待产品决策项上报(见 docs/tech/API.md 的 `GET /api/charge-orders` 一节)。
///
/// # marker 机制
/// 各条 `marker` 取**新注释里的独有片段**:老注释不含它 ⇒ 只 ALTER 一次,
/// 之后每次启动不再重复 DDL(幂等,见 `ensure_column_comment`)。
async fn align_column_metadata(pool: &MySqlPool) -> Result<(), sqlx::Error> {
// ① cabinets.abstract_id 可空化(老库为 NOT NULL 时才 ALTER)
// 容错口径与修改前一致(原先为 `let _ =`):失败只记 ERROR,绝不中断启动
// 兼容已有数据库:将 abstract_id 改为可空
// 只在列当前确实为 NOT NULL(老库)时才 ALTER:避免每次重启都对 cabinets 做一次 DDL
// 容错口径与修改前一致(原先为 `let _ =`):失败只记 ERROR,绝不中断启动
if let Err(e) = ensure_column_nullable(
pool,
"cabinets",
@ -379,7 +290,8 @@ async fn align_column_metadata(pool: &MySqlPool) -> Result<(), sqlx::Error> {
);
}
// ② charge_records.max_temperature 注释(0.1°C → K×100)—— 上抛口径
// 更新 charge_records.max_temperature 注释(0.1°C → K×100)
// 只在注释确实过时(老库)时才 ALTER:避免每次启动都对整表做一次 MODIFY
ensure_column_comment(
pool,
"charge_records",
@ -389,7 +301,8 @@ async fn align_column_metadata(pool: &MySqlPool) -> Result<(), sqlx::Error> {
)
.await?;
// ③ charge_orders.reason 注释:新增 timeout —— 上抛口径
// 更新 charge_orders.reason 注释:新增 timeout
// (订单超过 10 小时由 order_timeout 定时任务自动结束,落该枚举值)
ensure_column_comment(
pool,
"charge_orders",
@ -399,7 +312,13 @@ async fn align_column_metadata(pool: &MySqlPool) -> Result<(), sqlx::Error> {
)
.await?;
// ④ compartments.status 注释:语义收敛为"只表达占用" —— 上抛口径
// 更新 compartments.status 注释(**旧的注释是错的**:老库写的是
// `0空闲 1充电中 2故障 3禁用`,但全仓没有任何地方会写 2/3 —— 平台只写 0/1,
// `routes/workers/report_worker.rs::mark_compartment_status` 的调用点只有 0 与 1;
// 而"禁用"是设备/组织级状态,仓体级并不存在)。
// 本列语义从此**只表达占用**:"充满 / 故障"由 Redis 通道实时状态派生
// (`routes/cabinet_tree.rs::derive_channel_flags`),不落库。
// marker 取注释里的独有片段:老注释不含它 → 只 ALTER 一次,之后启动不再重复 DDL。
ensure_column_comment(
pool,
"compartments",
@ -409,7 +328,33 @@ async fn align_column_metadata(pool: &MySqlPool) -> Result<(), sqlx::Error> {
)
.await?;
// ⑤ cabinets.status 注释:"启用/禁用配置态,不是在线" —— **容错:只记 ERROR**
// 更新 cabinets.status 注释(**本列是"启用/禁用"配置态,不是"在线"**)。
//
// 全仓复核结论(grep 实测):
// - **唯一写入方** `routes/cabinets.rs::set_cabinet_status`:
// `UPDATE cabinets SET status = ? WHERE id = ?`,语义是 `2` = 禁用 / 其它值 = 启用;
// - **没有任何地方写 0**(0 只可能由建表默认值 `DEFAULT 0` 或该接口传入 0 时产生);
// - 消费方只有"禁用"一档:`commands.rs::CABINET_DISABLED_SQL`
// (`SELECT COUNT(*) > 0 FROM cabinets WHERE id = ? AND status = 2`)与
// `routes/h5.rs::ensure_cabinet_enabled` 据此拦截开合闸 / 开门 / 充电。
//
// 旧注释 `0离线 1在线 2禁用` 里的"**在线**"是**语义错配**:
// 生产实测 68 台柜子 `status` 全 = 1,而当时**一台设备都没连** ——
// 按它统计在线(`energy_stats` 的 `online_count`、H5 首页 `online_cabinets`)
// 会恒定显示"全部在线"。在线已全部改为 **Redis 判据**
// (`routes/cabinet_tree.rs::device_online_flag` / `count_online_devices`,
// 与设备列表 `is_online` 同源),本列不再参与任何在线判定。
//
// 只改注释,**不动类型与默认值**(仍是 `TINYINT DEFAULT 0`);
// `0` 是建表默认值产生的取值(`create_cabinet` 的 INSERT 未显式给 status),
// 语义等同"启用"(除 `2` 外一律按启用处理),代码不显式写它。
//
// marker 取新注释里的独有片段(旧注释 `0离线 1在线 2禁用(禁用后指令链路拦截)` 不含它)
// → 只 ALTER 一次,之后启动不再重复 DDL。
//
// ⚠️ 失败只记 ERROR、**绝不阻断启动**:注释只是元信息,
// 不配享有"迁移失败 → main panic → systemd 崩溃循环"的杀伤力
// (与上面 `cabinets.abstract_id` 可空化、下面 `align_legacy_columns` 同口径)。
if let Err(e) = ensure_column_comment(
pool,
"cabinets",
@ -425,7 +370,20 @@ async fn align_column_metadata(pool: &MySqlPool) -> Result<(), sqlx::Error> {
);
}
// ⑥ charge_orders.status 注释:标注 `2` 保留未使用 —— 上抛口径
// 更新 charge_orders.status 注释(**旧注释把从未写入的 `2` 写成了既成事实**)。
//
// 实测(全仓 grep):**生产代码只有 0 与 1 两个写入值** ——
// 建单 `report_worker.rs::try_create_order` 固定写 `0`;所有结束路径
// (`off` / `bat_out` / 换电池 / 10 小时超时 / 无电池对账)都走
// `finalize_order*`,`order_status` 恒为 `ORDER_STATUS_DONE = 1`。
// 唯一的 `2` 出现在 `test/reset_fixture.py`(一次性夹具脚本,非交付物)。
//
// 但 `2` **有消费方**:PC 订单页 `web/src/pages/charge-orders/index.tsx` 的筛选下拉
// 提供 `<Option value={2}>异常</Option>`、`STATUS_COLOR[2] = 'red'`,
// 且 `routes/charge_orders.rs::status_text` 把 `2` 映射为"异常" ——
// 即"异常"曾被设计成一个**状态**,而现实现把它降级成 `reason` 里的**结束原因**。
// 这是**产品语义缺口,不是笔误**,故**不删除 `2`**,只在注释里注明"保留未使用",
// 并作为待产品决策项上报(见 docs/tech/API.md 的 `GET /api/charge-orders` 一节)。
ensure_column_comment(
pool,
"charge_orders",
@ -435,15 +393,6 @@ async fn align_column_metadata(pool: &MySqlPool) -> Result<(), sqlx::Error> {
)
.await?;
Ok(())
}
/// 阶段 C:关键索引 + 老库缺失列的补齐。
///
/// 容错口径**两条不同**,逐一保留:
/// - [`create_indexes`] **上抛** ⇒ 经 `run` 中断启动(索引是可用性前提);
/// - [`ensure_energy_report_indexes`] 只记 `ERROR` ⇒ 纯性能优化,报表查询退化为全表扫描也不能阻断启动。
async fn create_indexes_and_columns(pool: &MySqlPool) -> Result<(), sqlx::Error> {
// 创建关键索引(IF NOT EXISTS 避免重复)
create_indexes(pool).await?;
@ -459,13 +408,6 @@ async fn create_indexes_and_columns(pool: &MySqlPool) -> Result<(), sqlx::Error>
// 兼容已有数据库:新增字段(幂等,列已存在则跳过)
add_columns_if_not_exists(pool).await?;
Ok(())
}
/// 阶段 D:历史库的**结构漂移对齐**与 DB 层约束兜底。
///
/// ⚠️ 三步的**顺序不可换**(见下),容错口径:第 1 步上抛,后两步只记 `ERROR`。
async fn align_legacy_structures(pool: &MySqlPool) -> Result<(), sqlx::Error> {
// 历史库「列在、但类型/可空性/默认值错了」的漂移对齐 + organizations.parent_id 索引/外键。
//
// ⚠️ 必须放在 `add_columns_if_not_exists` **之后**:它要建索引的
@ -483,7 +425,7 @@ async fn align_legacy_structures(pool: &MySqlPool) -> Result<(), sqlx::Error> {
//
// ⚠️ 这一步的失败**绝不能中断启动**:`run()` 返回 Err 会让 main panic 退出(生产事故),
// 而业务侧已经有「按仓体的进程内锁 + 条件插入」两层防护,DB 约束只是第三层兜底。
// 因此这里只记 ERROR 后继续(与上方各 `ALTER TABLE ...` 的容错口径一致)。
// 因此这里只记 ERROR 后继续(与上方 `ALTER TABLE cabinets ...` 的容错口径一致)。
if let Err(e) = ensure_active_order_unique(pool).await {
tracing::error!(
"[migrate] 活跃订单唯一约束迁移失败(服务继续启动;业务侧仍有进程内锁+条件插入兜底)err={}",
@ -491,18 +433,12 @@ async fn align_legacy_structures(pool: &MySqlPool) -> Result<(), sqlx::Error> {
);
}
Ok(())
}
/// 阶段 E:种子数据(默认权限码 + 默认角色)。
///
/// ⛔ **失败必须中断启动**:没有权限码,全部 RBAC 校验都会拒绝,平台等于不可用。
/// 两条语句都用 `INSERT IGNORE` / 幂等写法,重复执行无副作用。
async fn seed_defaults(pool: &MySqlPool) -> Result<(), sqlx::Error> {
// 初始化默认权限码(INSERT IGNORE 避免重复插入)
seed_permissions(pool).await?;
// 初始化默认角色
seed_roles(pool).await?;
tracing::info!("数据库迁移完成");
Ok(())
}

View File

@ -230,9 +230,22 @@ pub async fn auth_middleware(
let (role_level, organization_id) = role_and_org;
// 加载有效数据范围(`user_scopes`):整块搬进 `load_effective_scope`,
// 提权豁免与失败兜底口径见该函数的文档(**逐字未改**)。
let scope = load_effective_scope(pool, &claims, role_level, organization_id).await;
// 加载有效数据范围(`user_scopes`)。
// - 总管理员(role_level >= 2)**豁免**:恒为"不限制",连库都不查;
// - 只有配过 scope 的客户侧用户才会多出这条查询(见 `load_user_scopes` 的首查短路)。
let scope = if role_level >= SUPER_ADMIN_ROLE_LEVEL {
UserScope::default()
} else {
match load_user_scopes(pool, claims.sub, organization_id).await {
Ok(scope) => scope,
Err(e) => {
// 失败安全:范围查询失败**不能让请求 500**(否则一次脏数据/超时就全网不可用)。
// 保守退回"不限制",与"未配置"同义 —— 组织隔离仍然生效,不会跨组织泄露。
tracing::error!("查询用户数据范围失败,按「未配置=不限制」继续: {}", e);
UserScope::default()
}
}
};
// 构建当前用户信息并注入请求扩展
let current_user = CurrentUser {
@ -252,36 +265,6 @@ pub async fn auth_middleware(
next.run(req).await
}
/// 加载用户的**有效数据范围**(`user_scopes`)。
///
/// - 总管理员(`role_level >= 2`)**豁免**:恒为"不限制",连库都不查;
/// - 只有配过 scope 的客户侧用户才会多出这条查询(见 [`load_user_scopes`] 的首查短路);
/// - **失败安全**:范围查询失败**不能让请求 500**(否则一次脏数据/超时就全网不可用),
/// 保守退回"不限制",与"未配置"同义 —— 组织隔离仍然生效,不会跨组织泄露。
///
/// 本函数由 [`auth_middleware`] **原样搬出**:语句、顺序、失败口径逐字未改。
async fn load_effective_scope(
pool: &MySqlPool,
claims: &Claims,
role_level: i32,
organization_id: Option<i64>,
) -> UserScope {
let scope = if role_level >= SUPER_ADMIN_ROLE_LEVEL {
UserScope::default()
} else {
match load_user_scopes(pool, claims.sub, organization_id).await {
Ok(scope) => scope,
Err(e) => {
// 失败安全:范围查询失败**不能让请求 500**(否则一次脏数据/超时就全网不可用)。
// 保守退回"不限制",与"未配置"同义 —— 组织隔离仍然生效,不会跨组织泄露。
tracing::error!("查询用户数据范围失败,按「未配置=不限制」继续: {}", e);
UserScope::default()
}
}
};
scope
}
/// 从请求中提取当前用户(Axum Extractor 实现)
impl FromRequestParts<AppState> for CurrentUser {
type Rejection = Response;

View File

@ -170,46 +170,6 @@ pub async fn get_challenge(
})))
}
/// 登录前置闸门:失败计数锁定检查 + 滑块验证码校验。
///
/// 本函数由 `login` **原样搬出**,可执行语句、SQL、控制流、错误口径与执行顺序逐条不变:
/// - 锁定检查放在验证码之前:锁定期直接拒绝,不消耗验证码、不做图像/DB 开销;
/// 锁定期与密码错误返回**完全一致**的文案(`LOGIN_FAIL_MESSAGE`),
/// 不泄露账号是否存在 / 是否被锁。
/// - 校验滑块验证码(防暴力破解 + 人机验证):缺失验证参数或验证码错误返回 `BadRequest`。
///
/// 返回 `Ok(())` 表示前置检查全部通过;所有错误口径与搬出前一致,
/// 由调用方 `login` 以 `?` 原样上抛(`Unauthorized` / `BadRequest` 不变)。
async fn login_guard(
state: &AppState,
input: &LoginInput,
phone_key: &str,
ip_key: &str,
ip: &str,
) -> Result<(), AppError> {
// 锁定检查放在验证码之前:锁定期直接拒绝,不消耗验证码、不做图像/DB 开销
let phone_fails = read_counter(&state, &phone_key).await;
let ip_fails = read_counter(&state, &ip_key).await;
if phone_fails >= LOGIN_FAIL_MAX || ip_fails >= LOGIN_FAIL_MAX {
tracing::warn!(
phone = %input.phone, ip = %ip, phone_fails, ip_fails,
"登录已锁定,拒绝本次尝试(返回统一文案)"
);
return Err(AppError::Unauthorized(LOGIN_FAIL_MESSAGE.into()));
}
// 校验滑块验证码(防暴力破解 + 人机验证)
if input.challenge_id.is_empty() {
return Err(AppError::BadRequest("缺少登录验证参数".into()));
}
let captcha_x = input.captcha_x.unwrap_or(i32::MAX);
let valid = captcha::verify_offset(&state.redis_url, &input.challenge_id, captcha_x).await?;
if !valid {
return Err(AppError::BadRequest("验证码错误,请刷新页面重试".into()));
}
Ok(())
}
/// POST /api/auth/login — 用户登录
///
/// 防暴力破解:按「手机号 + 客户端 IP」双维度计数(Redis,TTL 15 分钟),
@ -230,8 +190,26 @@ pub async fn login(
let phone_key = format!("auth:fail:phone:{}", input.phone);
let ip_key = format!("auth:fail:ip:{}", ip);
// 锁定检查 + 滑块验证码校验:原样搬入 login_guard,此处仅按原顺序调用
login_guard(&state, &input, &phone_key, &ip_key, &ip).await?;
// 锁定检查放在验证码之前:锁定期直接拒绝,不消耗验证码、不做图像/DB 开销
let phone_fails = read_counter(&state, &phone_key).await;
let ip_fails = read_counter(&state, &ip_key).await;
if phone_fails >= LOGIN_FAIL_MAX || ip_fails >= LOGIN_FAIL_MAX {
tracing::warn!(
phone = %input.phone, ip = %ip, phone_fails, ip_fails,
"登录已锁定,拒绝本次尝试(返回统一文案)"
);
return Err(AppError::Unauthorized(LOGIN_FAIL_MESSAGE.into()));
}
// 校验滑块验证码(防暴力破解 + 人机验证)
if input.challenge_id.is_empty() {
return Err(AppError::BadRequest("缺少登录验证参数".into()));
}
let captcha_x = input.captcha_x.unwrap_or(i32::MAX);
let valid = captcha::verify_offset(&state.redis_url, &input.challenge_id, captcha_x).await?;
if !valid {
return Err(AppError::BadRequest("验证码错误,请刷新页面重试".into()));
}
// 查询用户(含角色和组织信息)
let user_row: Option<(i64, Option<String>, String, String, i32, i32, Option<i64>)> = sqlx::query_as(

View File

@ -598,7 +598,16 @@ pub async fn get_cabinet_detail(
.await?
.ok_or_else(|| AppError::NotFound(format!("柜子 {} 不存在", id)))?;
ensure_cabinet_access(&user, db, &cabinet).await?;
// 验证用户是否有权访问该柜子所属的项目
if let Some(pid) = cabinet.project_id {
check_project_access(&user, db, pid).await?;
}
// 数据范围兜底:**未划拨到项目的柜子**(`project_id` 为空)走不到上面的项目校验,
// 配了范围的用户一律不可见(`allows_project(None)` 在有限制时为 false)。
if !user.effective_scope().allows_project(cabinet.project_id) {
return Err(AppError::Forbidden("无权访问该设备".into()));
}
// 仓控板 + 仓体(含实时字段占位),并收集通道 Redis key 所需的映射
let (mut cabin_boards, comp_ids, comp_to_board) = load_boards_with_compartments(db, id).await?;
@ -656,28 +665,6 @@ pub async fn get_cabinet_detail(
})))
}
/// 详情接口的访问门禁:项目归属校验 + 数据范围兜底
///
/// 由 `get_cabinet_detail` 原样搬出(**语句、顺序、错误口径逐字未改**):
/// 先按柜子所属项目校验组织归属与项目范围,再按用户数据范围兜底。
async fn ensure_cabinet_access(
user: &CurrentUser,
db: &MySqlPool,
cabinet: &CabinetBasicRow,
) -> Result<(), AppError> {
// 验证用户是否有权访问该柜子所属的项目
if let Some(pid) = cabinet.project_id {
check_project_access(&user, db, pid).await?;
}
// 数据范围兜底:**未划拨到项目的柜子**(`project_id` 为空)走不到上面的项目校验,
// 配了范围的用户一律不可见(`allows_project(None)` 在有限制时为 false)。
if !user.effective_scope().allows_project(cabinet.project_id) {
return Err(AppError::Forbidden("无权访问该设备".into()));
}
Ok(())
}
// ---------------------------------------------------------------------------
// 设备详情 — Redis 实时状态(柜级 + 通道级)
// ---------------------------------------------------------------------------

View File

@ -134,7 +134,30 @@ pub async fn list_charge_orders(
let mut count_query = "SELECT COUNT(*) FROM charge_orders WHERE 1=1".to_string();
// 按固定顺序追加 `?` 占位符,绑定顺序必须与下方 build_count_query / build_list_query 一致
append_order_filters(&mut query, &mut count_query, &params);
if params.cabinet_id.is_some() {
query.push_str(" AND cabinet_id = ?");
count_query.push_str(" AND cabinet_id = ?");
}
if params.compartment_id.is_some() {
query.push_str(" AND compartment_id = ?");
count_query.push_str(" AND compartment_id = ?");
}
if params.battery_id.is_some() {
query.push_str(" AND battery_id LIKE ?");
count_query.push_str(" AND battery_id LIKE ?");
}
if params.status.is_some() {
query.push_str(" AND status = ?");
count_query.push_str(" AND status = ?");
}
if params.start_time.is_some() {
query.push_str(" AND start_time >= ?");
count_query.push_str(" AND start_time >= ?");
}
if params.end_time.is_some() {
query.push_str(" AND end_time <= ?");
count_query.push_str(" AND end_time <= ?");
}
// 组织隔离 + 数据范围:必须追加在 LIMIT/OFFSET 之前,绑定值也必须排在分页参数之前
let (list_scope_cond, scope_binds) = order_scope_conditions(&user, "co.");
@ -179,43 +202,6 @@ pub async fn list_charge_orders(
})))
}
/// 按固定顺序向 `query` / `count_query` 追加筛选条件的 `?` 占位符。
///
/// 原注释(逐字保留):按固定顺序追加 `?` 占位符,绑定顺序必须与下方 build_count_query / build_list_query 一致
///
/// 本函数只把 `list_charge_orders` 中原有的六个筛选片段**原样搬出**,
/// SQL 文本、条件顺序、追加顺序全部逐字不变:
/// cabinet_id → compartment_id → battery_id → status → start_time → end_time
///
/// 本函数是**纯字符串拼接**,不碰 IO,故为同步函数(无 `async`/`await`),
/// 也不会失败(返回 `()`,不引入任何错误处理口径变化)。
fn append_order_filters(query: &mut String, count_query: &mut String, params: &ChargeOrderQuery) {
if params.cabinet_id.is_some() {
query.push_str(" AND cabinet_id = ?");
count_query.push_str(" AND cabinet_id = ?");
}
if params.compartment_id.is_some() {
query.push_str(" AND compartment_id = ?");
count_query.push_str(" AND compartment_id = ?");
}
if params.battery_id.is_some() {
query.push_str(" AND battery_id LIKE ?");
count_query.push_str(" AND battery_id LIKE ?");
}
if params.status.is_some() {
query.push_str(" AND status = ?");
count_query.push_str(" AND status = ?");
}
if params.start_time.is_some() {
query.push_str(" AND start_time >= ?");
count_query.push_str(" AND start_time >= ?");
}
if params.end_time.is_some() {
query.push_str(" AND end_time <= ?");
count_query.push_str(" AND end_time <= ?");
}
}
/// 执行 COUNT 查询并返回总数。
///
/// 参数绑定顺序必须与 `list_charge_orders` 中向 `query` 追加 `?` 占位符的顺序一致:

View File

@ -1316,28 +1316,6 @@ pub async fn get_compartment_detail(
let flags = derive_channel_flags(realtime.batt_status.as_deref(), realtime.pow_status.as_deref());
let display_status = if flags.has_data { flags.status_code() } else { status };
build_compartment_detail_response(
comp_id,
channel_id,
display_status,
cabin_board_id,
&realtime,
records,
)
}
/// 组装仓体详情的响应体(**接口契约,字段名/顺序不得改动**)。
///
/// 由 [`get_compartment_detail`] **原样搬出**:`json!` 的字段顺序、`null` 语义
/// (无数据返回 `null` 而非 0)、三个空数组占位(无写入方但需维持前端类型)逐字未改。
fn build_compartment_detail_response(
comp_id: i64,
channel_id: i8,
display_status: i8,
cabin_board_id: i64,
realtime: &CompartmentRealtime,
records: Vec<ChargeRecordBriefRow>,
) -> Result<Json<Value>, AppError> {
Ok(Json(json!({
"code": 0,
"data": {

View File

@ -270,678 +270,3 @@ mod tests {
assert!(!is_allowed_origin("http://localhost:"));
}
}
// ---------------------------------------------------------------------------
// SQL 快照守门断言(长函数重构时补)
//
// 这些 handler **原本没有任何测试覆盖**。重构只做「搬动代码」,但后续任何人改动
// SQL 文本、字段顺序、`?` 占位符或 `json!` 字段名时,这里会立即失败。
//
// 判据来源:`include_str!` 把源文件本身作为被测对象 —— 因为这些 SQL 多为
// 函数内联字面量,单测无法引用局部变量。已自证:故意改一个字符即失败。
// ---------------------------------------------------------------------------
#[cfg(test)]
mod sql_snapshot_guards {
/// 断言 `haystack` 含 `needle`,失败时打印出错的片段便于定位。
fn assert_contains(what: &str, haystack: &str, needle: &str) {
assert!(
haystack.contains(needle),
"{what} 的 SQL 片段已改动(长函数重构后不应发生):\n{needle}"
);
}
/// `routes/auth.rs` 的 SQL 快照(5 条)。
#[test]
fn test_sql_snapshot_routes_auth_rs() {
let src = include_str!(r#"auth.rs"#);
assert_contains(r#"routes/auth.rs"#, src, r#"SELECT id, name, phone, password, status, role, organization_id FROM users WHERE phone = ?"#);
assert_contains(r#"routes/auth.rs"#, src, r#"SELECT id, name, phone, role, organization_id, avatar FROM users WHERE id = ?"#);
assert_contains(r#"routes/auth.rs"#, src, r#"SELECT id, name, phone, status, role, organization_id FROM users WHERE phone = ?"#);
assert_contains(r#"routes/auth.rs"#, src, r#"UPDATE users SET openid = ? WHERE id = ?"#);
assert_contains(r#"routes/auth.rs"#, src, r#"UPDATE users SET avatar = ? WHERE id = ?"#);
}
/// `routes/cabinets.rs` 的 SQL 快照(27 条)。
#[test]
fn test_sql_snapshot_routes_cabinets_rs() {
let src = include_str!(r#"cabinets.rs"#);
assert_contains(r#"routes/cabinets.rs"#, src, r#"SELECT id, project_id, abstract_id, imei, iccid, name, address,
CAST(latitude AS CHAR) AS latitude, CAST(longitude AS CHAR) AS longitude,
status, work_mode
FROM cabinets WHERE project_id = ? ORDER BY id"#);
assert_contains(r#"routes/cabinets.rs"#, src, r#"INSERT INTO cabinets (project_id, imei, auth_str, status)
VALUES (?, ?, ?, 1)"#);
assert_contains(r#"routes/cabinets.rs"#, src, r#"INSERT INTO cabin_boards (cabinet_id, board_id) VALUES (?, 1)"#);
assert_contains(r#"routes/cabinets.rs"#, src, r#"INSERT INTO compartments (cabin_board_id, channel_id) VALUES (?, ?)"#);
assert_contains(r#"routes/cabinets.rs"#, src, r#"UPDATE cabinets SET
project_id = COALESCE(?, project_id),
name = COALESCE(?, name),
abstract_id = COALESCE(?, abstract_id)
WHERE id = ?"#);
assert_contains(r#"routes/cabinets.rs"#, src, r#"DELETE compartments FROM compartments
INNER JOIN cabin_boards ON compartments.cabin_board_id = cabin_boards.id
WHERE cabin_boards.cabinet_id = ?"#);
assert_contains(r#"routes/cabinets.rs"#, src, r#"DELETE FROM cabin_boards WHERE cabinet_id = ?"#);
assert_contains(r#"routes/cabinets.rs"#, src, r#"DELETE FROM cabinets WHERE id = ?"#);
assert_contains(r#"routes/cabinets.rs"#, src, r#"UPDATE cabinets SET auth_str = ?, auth_str_issued = 0 WHERE id = ?"#);
assert_contains(r#"routes/cabinets.rs"#, src, r#"SELECT id, project_id, abstract_id, imei, iccid, name, address, CAST(latitude AS CHAR) AS latitude, CAST(longitude AS CHAR) AS longitude, status, work_mode
FROM cabinets WHERE id = ?"#);
assert_contains(r#"routes/cabinets.rs"#, src, r#"SELECT id, cabinet_id, board_id, status
FROM cabin_boards WHERE cabinet_id = ? ORDER BY board_id"#);
assert_contains(r#"routes/cabinets.rs"#, src, r#"SELECT id, cabin_board_id, channel_id, status
FROM compartments WHERE cabin_board_id = ? ORDER BY channel_id"#);
assert_contains(r#"routes/cabinets.rs"#, src, r#"SELECT id FROM projects WHERE id = ? AND organization_id = ?{}"#);
assert_contains(r#"routes/cabinets.rs"#, src, r#"UPDATE cabinets SET status = ? WHERE id = ?"#);
assert_contains(r#"routes/cabinets.rs"#, src, r#"UPDATE cabinets SET work_mode = ? WHERE id = ?"#);
assert_contains(r#"routes/cabinets.rs"#, src, r#"SELECT COUNT(*) FROM cabinets WHERE id = ? AND abstract_id IS NULL"#);
assert_contains(r#"routes/cabinets.rs"#, src, r#"INSERT INTO abstract_bindings (cabinet_id) VALUES (?)"#);
assert_contains(r#"routes/cabinets.rs"#, src, r#"UPDATE cabinets SET abstract_id = ? WHERE id = ?"#);
assert_contains(r#"routes/cabinets.rs"#, src, r#"SELECT COUNT(*) FROM cabinets WHERE id = ?"#);
assert_contains(r#"routes/cabinets.rs"#, src, r#"SELECT id, name FROM cabinets WHERE imei = ? AND id != ?"#);
assert_contains(r#"routes/cabinets.rs"#, src, r#"UPDATE cabinets SET imei = ? WHERE id = ?"#);
assert_contains(r#"routes/cabinets.rs"#, src, r#"SELECT imei FROM cabinets WHERE id = ?"#);
assert_contains(r#"routes/cabinets.rs"#, src, r#"SELECT name FROM projects WHERE id = ?"#);
assert_contains(r#"routes/cabinets.rs"#, src, r#"SELECT {} FROM cabinets"#);
assert_contains(r#"routes/cabinets.rs"#, src, r#"SELECT id, board_id FROM cabin_boards WHERE cabinet_id = ? ORDER BY board_id"#);
assert_contains(r#"routes/cabinets.rs"#, src, r#"SELECT cabinet_id FROM cabin_boards WHERE id = ?"#);
assert_contains(r#"routes/cabinets.rs"#, src, r#"SELECT id, channel_id FROM compartments WHERE cabin_board_id = ? ORDER BY channel_id"#);
}
/// `routes/charge_orders.rs` 的 SQL 快照(14 条)。
#[test]
fn test_sql_snapshot_routes_charge_orders_rs() {
let src = include_str!(r#"charge_orders.rs"#);
assert_contains(r#"routes/charge_orders.rs"#, src, r#" AND {} IN (SELECT c.id FROM cabinets c JOIN projects p ON c.project_id = p.id WHERE p.organization_id = ?)"#);
assert_contains(r#"routes/charge_orders.rs"#, src, r#"SELECT co.*, cab.name as cabinet_name FROM charge_orders co LEFT JOIN cabinets cab ON co.cabinet_id = cab.id WHERE 1=1"#);
assert_contains(r#"routes/charge_orders.rs"#, src, r#"SELECT COUNT(*) FROM charge_orders WHERE 1=1"#);
assert_contains(r#"routes/charge_orders.rs"#, src, r#" ORDER BY id DESC LIMIT ? OFFSET ?"#);
assert_contains(r#"routes/charge_orders.rs"#, src, r#" AND cabinet_id = ?"#);
assert_contains(r#"routes/charge_orders.rs"#, src, r#" AND compartment_id = ?"#);
assert_contains(r#"routes/charge_orders.rs"#, src, r#" AND battery_id LIKE ?"#);
assert_contains(r#"routes/charge_orders.rs"#, src, r#" AND status = ?"#);
assert_contains(r#"routes/charge_orders.rs"#, src, r#" AND start_time >= ?"#);
assert_contains(r#"routes/charge_orders.rs"#, src, r#" AND end_time <= ?"#);
assert_contains(r#"routes/charge_orders.rs"#, src, r#"co.cabinet_id IN (SELECT c.id FROM cabinets c JOIN projects p ON c.project_id = p.id WHERE p.organization_id = ?)"#);
assert_contains(r#"routes/charge_orders.rs"#, src, r#"co.cabinet_id IN (SELECT c.id FROM cabinets c WHERE c.project_id IN (?,?))"#);
assert_contains(r#"routes/charge_orders.rs"#, src, r#"cabinet_id IN (SELECT c.id FROM cabinets c JOIN projects p ON c.project_id = p.id WHERE p.organization_id = ?)"#);
assert_contains(r#"routes/charge_orders.rs"#, src, r#"cabinet_id IN (SELECT c.id FROM cabinets c WHERE c.project_id IN (?,?))"#);
}
/// `routes/h5.rs` 的 SQL 快照(52 条)。
#[test]
fn test_sql_snapshot_routes_h5_rs() {
let src = include_str!(r#"h5.rs"#);
assert_contains(r#"routes/h5.rs"#, src, r#"SELECT cab.imei FROM cabinets cab
INNER JOIN projects p ON cab.project_id = p.id
WHERE cab.imei IS NOT NULL AND p.organization_id = ?{}"#);
assert_contains(r#"routes/h5.rs"#, src, r#"SELECT COUNT(*) FROM compartments comp
JOIN cabin_boards cb ON comp.cabin_board_id = cb.id
JOIN cabinets c ON cb.cabinet_id = c.id
WHERE c.status = 1"#);
assert_contains(r#"routes/h5.rs"#, src, r#"SELECT COUNT(*) FROM compartments comp
JOIN cabin_boards cb ON comp.cabin_board_id = cb.id
JOIN cabinets c ON cb.cabinet_id = c.id
JOIN projects p ON c.project_id = p.id
WHERE c.status = 1 AND p.organization_id = ?{}"#);
assert_contains(r#"routes/h5.rs"#, src, r#"SELECT COUNT(*) FROM compartments comp
INNER JOIN cabin_boards cb ON comp.cabin_board_id = cb.id
INNER JOIN cabinets cab ON cb.cabinet_id = cab.id
INNER JOIN projects p ON cab.project_id = p.id
WHERE comp.status = ? AND p.organization_id = ?{}"#);
assert_contains(r#"routes/h5.rs"#, src, r#"SELECT c.id FROM cabinets c
INNER JOIN projects p ON c.project_id = p.id
WHERE p.organization_id = ?{}"#);
assert_contains(r#"routes/h5.rs"#, src, r#"SELECT COUNT(*) FROM charge_records cr
INNER JOIN cabinets cab ON cr.cabinet_id = cab.id
INNER JOIN projects p ON cab.project_id = p.id
WHERE DATE(cr.start_time) = CURDATE() AND p.organization_id = ?{}"#);
assert_contains(r#"routes/h5.rs"#, src, r#"SELECT CAST(COALESCE(SUM(cr.energy), 0) AS SIGNED) FROM charge_records cr
INNER JOIN cabinets cab ON cr.cabinet_id = cab.id
INNER JOIN projects p ON cab.project_id = p.id
WHERE DATE(cr.start_time) = CURDATE() AND p.organization_id = ?{}"#);
assert_contains(r#"routes/h5.rs"#, src, r#"SELECT CAST(COALESCE(SUM(
TIMESTAMPDIFF(SECOND, start_time, COALESCE(end_time, NOW()))
), 0) AS SIGNED) FROM charge_records
WHERE DATE(start_time) = CURDATE()"#);
assert_contains(r#"routes/h5.rs"#, src, r#"SELECT CAST(COALESCE(SUM(
TIMESTAMPDIFF(SECOND, cr.start_time, COALESCE(cr.end_time, NOW()))
), 0) AS SIGNED) FROM charge_records cr
INNER JOIN cabinets cab ON cr.cabinet_id = cab.id
INNER JOIN projects p ON cab.project_id = p.id
WHERE DATE(cr.start_time) = CURDATE() AND p.organization_id = ?{}"#);
assert_contains(r#"routes/h5.rs"#, src, r#"SELECT IFNULL(p.name, '') AS project, IFNULL(cab.name, cab.abstract_id) AS cabinet,
comp.channel_id
FROM compartments comp
JOIN cabin_boards cb ON comp.cabin_board_id = cb.id
JOIN cabinets cab ON cb.cabinet_id = cab.id
LEFT JOIN projects p ON cab.project_id = p.id
WHERE comp.status = 2
ORDER BY comp.updated_at DESC
LIMIT 10"#);
assert_contains(r#"routes/h5.rs"#, src, r#"SELECT IFNULL(p.name, '') AS project, IFNULL(cab.name, cab.abstract_id) AS cabinet,
comp.channel_id
FROM compartments comp
JOIN cabin_boards cb ON comp.cabin_board_id = cb.id
JOIN cabinets cab ON cb.cabinet_id = cab.id
INNER JOIN projects p ON cab.project_id = p.id
WHERE comp.status = 2 AND p.organization_id = ?{}
ORDER BY comp.updated_at DESC
LIMIT 10"#);
assert_contains(r#"routes/h5.rs"#, src, r#"SELECT COUNT(*) FROM compartments comp
JOIN cabin_boards cb ON comp.cabin_board_id = cb.id
JOIN cabinets cab ON cb.cabinet_id = cab.id
WHERE cab.project_id = ? AND comp.status = 1"#);
assert_contains(r#"routes/h5.rs"#, src, r#"SELECT COUNT(*) FROM compartments comp
JOIN cabin_boards cb ON comp.cabin_board_id = cb.id
JOIN cabinets cab ON cb.cabinet_id = cab.id
WHERE cab.project_id = ? AND comp.status = 0"#);
assert_contains(r#"routes/h5.rs"#, src, r#"SELECT COUNT(*) FROM compartments comp
JOIN cabin_boards cb ON comp.cabin_board_id = cb.id
JOIN cabinets cab ON cb.cabinet_id = cab.id
WHERE cab.project_id = ? AND comp.status = 2"#);
assert_contains(r#"routes/h5.rs"#, src, r#"SELECT c.id, c.abstract_id, c.name, c.status, c.imei, c.iccid, c.project_id, c.address, CAST(c.latitude AS CHAR) AS latitude, CAST(c.longitude AS CHAR) AS longitude,"#);
assert_contains(r#"routes/h5.rs"#, src, r#"(SELECT COUNT(*) FROM compartments comp JOIN cabin_boards cb ON comp.cabin_board_id = cb.id WHERE cb.cabinet_id = c.id) as channel_count,"#);
assert_contains(r#"routes/h5.rs"#, src, r#"(SELECT COUNT(*) FROM compartments comp JOIN cabin_boards cb ON comp.cabin_board_id = cb.id WHERE cb.cabinet_id = c.id AND comp.status = 0) as idle_count,"#);
assert_contains(r#"routes/h5.rs"#, src, r#"(SELECT COUNT(*) FROM compartments comp JOIN cabin_boards cb ON comp.cabin_board_id = cb.id WHERE cb.cabinet_id = c.id AND comp.status = 1) as charging_count"#);
assert_contains(r#"routes/h5.rs"#, src, r#"SELECT c.id, c.abstract_id, c.name, c.status, c.imei, c.iccid, c.project_id, c.address, CAST(c.latitude AS CHAR) AS latitude, CAST(c.longitude AS CHAR) AS longitude,"#);
assert_contains(r#"routes/h5.rs"#, src, r#"(SELECT COUNT(*) FROM compartments comp JOIN cabin_boards cb ON comp.cabin_board_id = cb.id WHERE cb.cabinet_id = c.id) as channel_count,"#);
assert_contains(r#"routes/h5.rs"#, src, r#"(SELECT COUNT(*) FROM compartments comp JOIN cabin_boards cb ON comp.cabin_board_id = cb.id WHERE cb.cabinet_id = c.id AND comp.status = 0) as idle_count,"#);
assert_contains(r#"routes/h5.rs"#, src, r#"(SELECT COUNT(*) FROM compartments comp JOIN cabin_boards cb ON comp.cabin_board_id = cb.id WHERE cb.cabinet_id = c.id AND comp.status = 1) as charging_count"#);
assert_contains(r#"routes/h5.rs"#, src, r#"SELECT c.id, c.abstract_id, c.name, c.status, c.imei, c.iccid, c.project_id, c.address, CAST(c.latitude AS CHAR) AS latitude, CAST(c.longitude AS CHAR) AS longitude,"#);
assert_contains(r#"routes/h5.rs"#, src, r#"(SELECT COUNT(*) FROM compartments comp JOIN cabin_boards cb ON comp.cabin_board_id = cb.id WHERE cb.cabinet_id = c.id) as channel_count,"#);
assert_contains(r#"routes/h5.rs"#, src, r#"(SELECT COUNT(*) FROM compartments comp JOIN cabin_boards cb ON comp.cabin_board_id = cb.id WHERE cb.cabinet_id = c.id AND comp.status = 0) as idle_count,"#);
assert_contains(r#"routes/h5.rs"#, src, r#"(SELECT COUNT(*) FROM compartments comp JOIN cabin_boards cb ON comp.cabin_board_id = cb.id WHERE cb.cabinet_id = c.id AND comp.status = 1) as charging_count"#);
assert_contains(r#"routes/h5.rs"#, src, r#"SELECT c.id, c.abstract_id, c.name, c.status, c.imei, c.iccid, c.project_id, c.address, CAST(c.latitude AS CHAR) AS latitude, CAST(c.longitude AS CHAR) AS longitude,"#);
assert_contains(r#"routes/h5.rs"#, src, r#"(SELECT COUNT(*) FROM compartments comp JOIN cabin_boards cb ON comp.cabin_board_id = cb.id WHERE cb.cabinet_id = c.id) as channel_count,"#);
assert_contains(r#"routes/h5.rs"#, src, r#"(SELECT COUNT(*) FROM compartments comp JOIN cabin_boards cb ON comp.cabin_board_id = cb.id WHERE cb.cabinet_id = c.id AND comp.status = 0) as idle_count,"#);
assert_contains(r#"routes/h5.rs"#, src, r#"(SELECT COUNT(*) FROM compartments comp JOIN cabin_boards cb ON comp.cabin_board_id = cb.id WHERE cb.cabinet_id = c.id AND comp.status = 1) as charging_count"#);
assert_contains(r#"routes/h5.rs"#, src, r#"SELECT c.id, c.abstract_id, c.name, c.status, c.imei, c.iccid, c.project_id, c.address, CAST(c.latitude AS CHAR) AS latitude, CAST(c.longitude AS CHAR) AS longitude,"#);
assert_contains(r#"routes/h5.rs"#, src, r#"(SELECT COUNT(*) FROM compartments comp JOIN cabin_boards cb ON comp.cabin_board_id = cb.id WHERE cb.cabinet_id = c.id) as channel_count,"#);
assert_contains(r#"routes/h5.rs"#, src, r#"(SELECT COUNT(*) FROM compartments comp JOIN cabin_boards cb ON comp.cabin_board_id = cb.id WHERE cb.cabinet_id = c.id AND comp.status = 0) as idle_count,"#);
assert_contains(r#"routes/h5.rs"#, src, r#"(SELECT COUNT(*) FROM compartments comp JOIN cabin_boards cb ON comp.cabin_board_id = cb.id WHERE cb.cabinet_id = c.id AND comp.status = 1) as charging_count"#);
assert_contains(r#"routes/h5.rs"#, src, r#"SELECT p.id, IFNULL(o.disabled, 0) FROM projects p
LEFT JOIN organizations o ON p.organization_id = o.id
WHERE p.id IN ({})"#);
assert_contains(r#"routes/h5.rs"#, src, r#"SELECT cb.board_id, c.imei
FROM cabin_boards cb
INNER JOIN cabinets c ON c.id = cb.cabinet_id
WHERE cb.id = ?"#);
assert_contains(r#"routes/h5.rs"#, src, r#"SELECT id, start_time, end_time, energy
FROM charge_records
WHERE compartment_id = ?
ORDER BY start_time DESC LIMIT 10"#);
assert_contains(r#"routes/h5.rs"#, src, r#"SELECT cab.id, cab.imei, cb.board_id, comp.channel_id, cab.work_mode
FROM cabinets cab
JOIN cabin_boards cb ON cb.cabinet_id = cab.id
JOIN compartments comp ON comp.cabin_board_id = cb.id
WHERE comp.id = ?"#);
assert_contains(r#"routes/h5.rs"#, src, r#"SELECT cb.cabinet_id FROM compartments comp
JOIN cabin_boards cb ON cb.id = comp.cabin_board_id
WHERE comp.id = ?"#);
assert_contains(r#"routes/h5.rs"#, src, r#"SELECT cab.id AS cabinet_id, cab.imei, cb.board_id, comp.channel_id
FROM cabinets cab
JOIN cabin_boards cb ON cb.cabinet_id = cab.id
JOIN compartments comp ON comp.cabin_board_id = cb.id
WHERE comp.id = ?"#);
assert_contains(r#"routes/h5.rs"#, src, r#"SELECT COUNT(*) > 0 FROM cabinets c
INNER JOIN projects p ON c.project_id = p.id
INNER JOIN organizations o ON p.organization_id = o.id
WHERE c.id = ? AND o.disabled = 1"#);
assert_contains(r#"routes/h5.rs"#, src, r#"SELECT cab.id FROM cabinets cab
INNER JOIN projects p ON cab.project_id = p.id
WHERE cab.id = ? AND p.organization_id = ?{}"#);
assert_contains(r#"routes/h5.rs"#, src, r#"SELECT cab.imei FROM cabinets cab WHERE cab.imei IS NOT NULL"#);
assert_contains(r#"routes/h5.rs"#, src, r#"SELECT COUNT(*) FROM compartments WHERE status = ?"#);
assert_contains(r#"routes/h5.rs"#, src, r#"SELECT id FROM cabinets"#);
assert_contains(r#"routes/h5.rs"#, src, r#"SELECT COUNT(*) FROM charge_records WHERE DATE(start_time) = CURDATE()"#);
assert_contains(r#"routes/h5.rs"#, src, r#"SELECT CAST(COALESCE(SUM(energy), 0) AS SIGNED) FROM charge_records WHERE DATE(start_time) = CURDATE()"#);
assert_contains(r#"routes/h5.rs"#, src, r#"SELECT id, name FROM projects WHERE organization_id = ? ORDER BY id"#);
assert_contains(r#"routes/h5.rs"#, src, r#"SELECT id, name FROM projects ORDER BY id"#);
assert_contains(r#"routes/h5.rs"#, src, r#"SELECT COUNT(*) FROM cabinets WHERE project_id = ?"#);
assert_contains(r#"routes/h5.rs"#, src, r#"SELECT id, name FROM organizations ORDER BY id"#);
assert_contains(r#"routes/h5.rs"#, src, r#"SELECT id, name FROM organizations WHERE id = ?"#);
assert_contains(r#"routes/h5.rs"#, src, r#"SELECT id, abstract_id, imei, name, status FROM cabinets WHERE abstract_id = ?"#);
assert_contains(r#"routes/h5.rs"#, src, r#"SELECT id, board_id, status FROM cabin_boards WHERE cabinet_id = ? ORDER BY board_id"#);
assert_contains(r#"routes/h5.rs"#, src, r#"SELECT id, channel_id, status FROM compartments WHERE cabin_board_id = ? ORDER BY channel_id"#);
assert_contains(r#"routes/h5.rs"#, src, r#"SELECT id, channel_id, status, cabin_board_id FROM compartments WHERE id = ?"#);
assert_contains(r#"routes/h5.rs"#, src, r#"SELECT cabinet_id FROM cabin_boards WHERE id = ?"#);
assert_contains(r#"routes/h5.rs"#, src, r#"SELECT imei FROM cabinets WHERE id = ?"#);
assert_contains(r#"routes/h5.rs"#, src, r#"SELECT COUNT(*) > 0 FROM cabinets WHERE id = ? AND status = 2"#);
assert_contains(r#"routes/h5.rs"#, src, r#"SELECT cabinet_id, board_id FROM cabin_boards WHERE id = ?"#);
assert_contains(r#"routes/h5.rs"#, src, r#"INSERT INTO user_feedback (user_id, content, contact, images) VALUES (?, ?, ?, ?)"#);
assert_contains(r#"routes/h5.rs"#, src, r#", Uuid::new_v4().simple(), ext);
let filepath = dir.join(&filename);
let mut f = std::fs::File::create(&filepath)
.map_err(|e| AppError::Internal(format!("#);
assert_contains(r#"routes/h5.rs"#, src, r#".into()));
}
// 归属校验:客户侧只能处置本组织成员提交的反馈(工厂豁免)。
// 历史缺陷:原先只校验权限码就直接 UPDATE ⇒ 企业管理员能改别的组织的反馈状态。
verify_feedback_ownership(pool, &user, id).await?;
let affected = sqlx::query("#);
assert_contains(r#"routes/h5.rs"#, src, r#"WHERE id = ?"#);
assert_contains(r#"routes/h5.rs"#, src, r#"SELECT cab.imei FROM cabinets cab"#);
assert_contains(r#"routes/h5.rs"#, src, r#" AND p.id IN (?,?)"#);
assert_contains(r#"routes/h5.rs"#, src, r#"AND p.id IN (?,?)"#);
}
/// `middleware/auth.rs` 的 SQL 快照(18 条)。
#[test]
fn test_sql_snapshot_middleware_auth_rs() {
let src = include_str!(r#"../middleware/auth.rs"#);
assert_contains(r#"middleware/auth.rs"#, src, r#"
SELECT DISTINCT p.code
FROM permissions p
INNER JOIN role_permissions rp ON rp.permission_id = p.id
INNER JOIN user_roles ur ON ur.role_id = rp.role_id
WHERE ur.user_id = ?
"#);
assert_contains(r#"middleware/auth.rs"#, src, r#"
SELECT DISTINCT p.code
FROM permissions p
INNER JOIN role_permissions rp ON rp.permission_id = p.id
INNER JOIN roles r ON r.id = rp.role_id
WHERE r.name = ?
"#);
assert_contains(r#"middleware/auth.rs"#, src, r#"
SELECT DISTINCT t.pid FROM (
SELECT p.id AS pid
FROM user_scopes us
JOIN projects p ON p.organization_id = us.scope_id
WHERE us.user_id = ? AND us.scope_type = 1 AND us.scope_id = ?
UNION
SELECT p.id AS pid
FROM user_scopes us
JOIN projects p ON p.id = us.scope_id
WHERE us.user_id = ? AND us.scope_type = 2 AND p.organization_id = ?
) t
"#);
assert_contains(r#"middleware/auth.rs"#, src, r#" AND {} IN (SELECT c.id FROM cabinets c WHERE c.project_id IN ({}))"#);
assert_contains(r#"middleware/auth.rs"#, src, r#"SELECT role FROM users WHERE id = ?"#);
assert_contains(r#"middleware/auth.rs"#, src, r#"SELECT code FROM permissions"#);
assert_contains(r#"middleware/auth.rs"#, src, r#"SELECT role, organization_id FROM users WHERE id = ?"#);
assert_contains(r#"middleware/auth.rs"#, src, r#"SELECT COUNT(*) FROM user_scopes WHERE user_id = ?"#);
assert_contains(r#"middleware/auth.rs"#, src, r#" AND {}.cabinet_id IN (SELECT c.id FROM cabinets c JOIN projects p ON c.project_id = p.id WHERE p.organization_id = ?)"#);
assert_contains(r#"middleware/auth.rs"#, src, r#" AND {} = ?"#);
assert_contains(r#"middleware/auth.rs"#, src, r#" AND {} IN (SELECT pj.id FROM projects pj WHERE pj.organization_id = ?)"#);
assert_contains(r#"middleware/auth.rs"#, src, r#" AND {}.user_id IN (SELECT id FROM users WHERE organization_id = ?)"#);
assert_contains(r#"middleware/auth.rs"#, src, r#" AND c.project_id IN (?,?)"#);
assert_contains(r#"middleware/auth.rs"#, src, r#"cr.cabinet_id IN (SELECT c.id FROM cabinets c WHERE c.project_id IN (?,?))"#);
assert_contains(r#"middleware/auth.rs"#, src, r#" AND p.organization_id = ?"#);
assert_contains(r#"middleware/auth.rs"#, src, r#" AND c.project_id IN (SELECT pj.id FROM projects pj WHERE pj.organization_id = ?)"#);
assert_contains(r#"middleware/auth.rs"#, src, r#" AND c.project_id IN (?)"#);
assert_contains(r#"middleware/auth.rs"#, src, r#"WHERE 1=1 AND c.project_id IN (?)"#);
}
/// `workers/report_worker.rs` 的 SQL 快照(27 条)。
#[test]
fn test_sql_snapshot_workers_report_worker_rs() {
let src = include_str!(r#"../workers/report_worker.rs"#);
assert_contains(r#"workers/report_worker.rs"#, src, r#"SELECT auth_str, auth_str_issued FROM cabinets WHERE imei = ?"#);
assert_contains(r#"workers/report_worker.rs"#, src, r#"UPDATE cabinets SET auth_str_issued = 1, auth_str = ? WHERE imei = ?"#);
assert_contains(r#"workers/report_worker.rs"#, src, r#"SELECT auth_str FROM cabinets WHERE imei = ?"#);
assert_contains(r#"workers/report_worker.rs"#, src, r#"SELECT COUNT(DISTINCT cb.id), COUNT(comp.id)
FROM cabin_boards cb
JOIN cabinets cab ON cb.cabinet_id = cab.id
LEFT JOIN compartments comp ON comp.cabin_board_id = cb.id
WHERE cab.imei = ?"#);
assert_contains(r#"workers/report_worker.rs"#, src, r#"SELECT cell FROM cabinets WHERE imei = ?"#);
assert_contains(r#"workers/report_worker.rs"#, src, r#", byte(1)?),
soc: i64::from(byte(2)?) * 10,
voltage: i64::from(be16(3)?) * 10,
current: i64::from(be16(5)?) * 10,
temp: i64::from(be16(7)?) * 100,
})
}
/// 校验并规范化 2 位 hex 状态字节(协议 §3.8);缺失或非法 → `None`(调用方记 WARN)
fn normalize_hex_byte(raw: Option<&str>) -> Option<String> {
let raw = raw?;
if raw.len() != 2 {
return None;
}
u8::from_str_radix(raw, 16).ok().map(|b| format!("#);
assert_contains(r#"workers/report_worker.rs"#, src, r#").and_then(|v| v.as_str())?;"#);
assert_contains(r#"workers/report_worker.rs"#, src, r#"let parts: Vec<&str> = id.split('-').collect();"#);
assert_contains(r#"workers/report_worker.rs"#, src, r#"if parts.len() != COMPARTMENT_ID_SEGMENTS {"#);
assert_contains(r#"workers/report_worker.rs"#, src, r#"let board_id: i8 = parts.first()?.trim().parse().ok()?;"#);
assert_contains(r#"workers/report_worker.rs"#, src, r#"这条核心规则)
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum OrderDecision {
/// 无活跃订单 → 新建
Create,
/// 已有活跃订单且电池未变(含任一方 id 为空串)→ 不新建
Skip,
/// 已有活跃订单但换电池了 → 老订单强制结束后再新建
Replace,
}
/// 纯函数:根据「当前活跃订单的电池 id」与「本次要充电的电池 id」决定建单动作。
///
/// 规则依据交付文档 §5.5 / §5.7:
/// - **一个通道只可能有一块电池** —— 所以 id 变了必然是换了新电池,老订单必须先结束;
/// - **任一方为空串**(BMS 读不到序列号)→ **视为未变**,不新建(否则每次重复上报都会多一笔订单);
/// - 双方 id 相同 → 同一次充电的重复上报(设备重发 `on`)→ 不新建。
///
/// 抽成纯函数的原因:这段判定是**订单生命周期里最容易出错、又最需要被证明正确**的一环,
/// 而它原本内嵌在"#);
assert_contains(r#"workers/report_worker.rs"#, src, r#"pub fn decide_order_action(active_battery: Option<&str>, new_battery: &str) -> OrderDecision {"#);
assert_contains(r#"workers/report_worker.rs"#, src, r#"let Some(old) = active_battery else {"#);
assert_contains(r#"workers/report_worker.rs"#, src, r#"return OrderDecision::Create;"#);
assert_contains(r#"workers/report_worker.rs"#, src, r#"// 任一方未知(空串)⇒ 无法证明换了电池 ⇒ 保守视为未变"#);
assert_contains(r#"workers/report_worker.rs"#, src, r#"let _guard = acquire_compartment_lock(compartment_id).await;"#);
assert_contains(r#"workers/report_worker.rs"#, src, r#"if !reconcile_active_order("#);
assert_contains(r#"workers/report_worker.rs"#, src, r#"let start_soc = match start_soc {"#);
assert_contains(r#"workers/report_worker.rs"#, src, r#"None => resolve_start_soc(pool, redis, cabinet_id, board_id, channel_id).await,"#);
assert_contains(r#"workers/report_worker.rs"#, src, r#"/// 换电池时结束老订单并复位老电池台账([`ensure_charge_order`] 判定为 [`OrderDecision::Replace`] 的分支)。"#);
assert_contains(r#"workers/report_worker.rs"#, src, r#"/// 一个通道只可能有一块电池:id 变了必然是换了新电池,老订单必须先结束。"#);
assert_contains(r#"workers/report_worker.rs"#, src, r#"/// 走统一结算:写回本次充电电量(平台侧积分)并生成充电记录。"#);
assert_contains(r#"workers/report_worker.rs"#, src, r#"/// ⚠️ 调用方 [`ensure_charge_order`] **已持有该仓体的进程内锁**,所以这里必须走"#);
assert_contains(r#"workers/report_worker.rs"#, src, r#":
/// 两个任务都先取(谁先取谁拿到电量,另一个拿到 0),再各自 UPDATE ——
/// **拿到电量的那个若没抢到 UPDATE(InnoDB 行锁让另一个先提交),落库的就是 0**,
/// 真实电量被静默丢弃(旧代码 `affected == 0` 时直接 `return`)。
/// 而"#);
assert_contains(r#"workers/report_worker.rs"#, src, r#"恰恰是**最常见的结束路径**,于是电量与统计直接错。
///
/// # 新顺序
/// 1. 按 `compartment_id` **原子抢占结束权**:
/// `UPDATE ... WHERE compartment_id = ? AND status = 0 ORDER BY id DESC LIMIT 1`
/// (语义与旧实现"#);
assert_contains(r#"workers/report_worker.rs"#, src, r#",
compartment_id,
reason
);
return Ok(0);
}
// 2) 抢到结束权之后**才**取电量(谁结束谁结算)
let order_id = match find_just_finished_order(pool, compartment_id, order_status).await? {
Some(id) => id,
None => {
// 理论上不可达:上一条 UPDATE 至少改动了 1 行。留作防御 ——
// 宁可不写电量,也绝不把电量写到别的订单行上。
tracing::error!(
"#);
assert_contains(r#"workers/report_worker.rs"#, src, r#"的 `UPDATE`。
/// 依据:订单 id 自增,且同一仓体同一时刻只允许一笔活跃订单(`active_flag` 唯一键 +
/// [`acquire_compartment_lock`]),因此"#);
assert_contains(r#"workers/report_worker.rs"#, src, r#"let energy_wh = energy::mwh_to_wh(mwh);"#);
assert_contains(r#"workers/report_worker.rs"#, src, r#"// 只改**已经收尾**的那一行(`status <> 0`):任何情况下都不碰活跃单"#);
assert_contains(r#"workers/report_worker.rs"#, src, r#"write_back_order_energy(pool, order_id, energy_wh, take_energy).await?;"#);
assert_contains(r#"workers/report_worker.rs"#, src, r#"// 生成充电记录(供后台「充电记录」页 / 导出 / 小程序今日统计使用)"#);
assert_contains(r#"workers/report_worker.rs"#, src, r#"作为兜底(依据见该常量)。
///
/// 复位按**单行 + 同样的守卫条件**执行(`UPDATE ... WHERE id=? AND status=1 AND NOT EXISTS(...)`),
/// 避免"#);
assert_contains(r#"workers/report_worker.rs"#, src, r#"(见该函数说明)。
/// 两处都改才算真正调整阈值。
pub const ORDER_TIMEOUT_HOURS: i64 = 10;
/// **超时选单 SQL —— 方向即语义**(`start_time` 必须早于 `NOW() - ORDER_TIMEOUT_HOURS`)。
///
/// ⛔ **方向极易写反,且写反了不报错**:
/// 若写成 `start_time < TIMESTAMPADD(HOUR, ?, NOW())` 并**绑正数**,得到的是**未来时刻**,
/// "#);
assert_contains(r#"workers/report_worker.rs"#, src, r#"对**几乎所有订单恒真** ⇒ 刚建几分钟的活跃订单会被当成超时单提前结掉、
/// 电量被提前结算(`workers/order_timeout.rs` 的注释记录了这个坑)。
///
/// ✅ 两种**等价**的正确形态(本项目同时存在,改一处必须确认另一处):
/// - **绑负数**(本常量,`finalize_timeout_order` 用):`start_time < TIMESTAMPADD(HOUR, -10, NOW())`
/// - **把 start_time 往前推**(`order_timeout.rs::scan_once` 用):
/// `TIMESTAMPADD(HOUR, +10, start_time) < NOW()`
///
/// 守门单测(**抽成常量就是为了让方向能被断言**):
/// [`tests::test_expired_order_condition_points_backwards`]、
/// [`tests::test_is_expired_two_way_assertions_catch_reversed_direction`]。
const FINALIZE_EXPIRED_ORDER_SQL: &str = "#);
assert_contains(r#"workers/report_worker.rs"#, src, r#"**:先定位该仓体 `start_time` 早于 `NOW() - 10h` 的活跃订单"#);
assert_contains(r#"workers/report_worker.rs"#, src, r#"/// (按 `start_time` 升序取最早一笔)。找不到就返回 0 ——"#);
assert_contains(r#"workers/report_worker.rs"#, src, r#"/// 这样**刚建的新单绝不会被误结**:即使调用方的扫描条件写错(该条件的方向极易搞反:"#);
assert_contains(r#"workers/report_worker.rs"#, src, r#"/// 写成 `start_time < TIMESTAMPADD(HOUR, 10, NOW())` 会命中**所有**活跃订单),"#);
assert_contains(r#"workers/report_worker.rs"#, src, r#",
compartment_id,
order_id
);
return Ok(0);
};
finalize_order_by_id(
pool,
redis,
&imei,
i64::from(board_id),
i64::from(channel_id),
compartment_id,
order_id,
None,
ORDER_STATUS_DONE,
"#);
assert_contains(r#"workers/report_worker.rs"#, src, r#"抢占收尾(设备上报
/// `off`/`bat_out` 的场景,报文里只有仓体),而本函数只认 `order_id` ——
/// 用于**扫描结果可能过期**的路径(超时兜底、无电池对账兜底):若那笔单已被别的路径结束,
/// `WHERE id = ? AND status = 0` 会返回 0 行,**绝不会误结该仓体后来新建的订单**。
///
/// 电量归属:通道电量累加器属于该仓体**最新的活跃订单**(新单建单时会 `energy::reset`)。
/// 因此只有"#);
assert_contains(r#"workers/report_worker.rs"#, src, r#"async fn finalize_active_order_for_bat_out("#);
assert_contains(r#"workers/report_worker.rs"#, src, r#"redis: &ConnectionManager,"#);
assert_contains(r#"workers/report_worker.rs"#, src, r#") -> Result<u64, Box<dyn std::error::Error + Send + Sync>> {"#);
assert_contains(r#"workers/report_worker.rs"#, src, r#"let active = find_active_order(mysql, compartment_id).await?;"#);
assert_contains(r#"workers/report_worker.rs"#, src, r#"))
.and_then(|v| v.as_str())
.and_then(parse_channel_hex)
.map(|c| c.soc);
// 结束订单(统一结算:写回平台侧积分电量 + 生成充电记录)
let updated = finalize_order(
mysql,
redis,
dev_id,
i64::from(board_id),
i64::from(channel_id),
compartment_id,
end_soc,
1,
"#);
assert_contains(r#"workers/report_worker.rs"#, src, r#"), OrderDecision::Create);
// 电池 id 为空也照常新建(BMS 读不到序列号不影响充电)
assert_eq!(decide_order_action(None, "#);
assert_contains(r#"workers/report_worker.rs"#, src, r#"), OrderDecision::Create);
}
/// 同一块电池重复上报 `on` → 不新建(设备重发同一次充电)
#[test]
fn test_decide_order_action_skip_same_battery() {
assert_eq!(
decide_order_action(Some("#);
assert_contains(r#"workers/report_worker.rs"#, src, r#", OrderDecision::Create),
(Some("#);
}
/// `db/migrate.rs` 的 SQL 快照(28 条)。
#[test]
fn test_sql_snapshot_db_migrate_rs() {
let src = include_str!(r#"../db/migrate.rs"#);
assert_contains(r#"db/migrate.rs"#, src, r#"CREATE TABLE IF NOT EXISTS organizations ("#);
assert_contains(r#"db/migrate.rs"#, src, r#"id BIGINT AUTO_INCREMENT PRIMARY KEY,"#);
assert_contains(r#"db/migrate.rs"#, src, r#"name VARCHAR(100) NOT NULL COMMENT '组织名称',"#);
assert_contains(r#"db/migrate.rs"#, src, r#"parent_id BIGINT NULL COMMENT '父组织ID(NULL=顶级组织)',"#);
assert_contains(r#"db/migrate.rs"#, src, r#"ALTER TABLE cabinets MODIFY COLUMN status TINYINT DEFAULT 0 COMMENT '启用/禁用配置态(不表示在线):1启用 2禁用(拦截指令链路) 0=建表默认值产生的取值,语义等同启用'"#);
assert_contains(r#"db/migrate.rs"#, src, r#"ALTER TABLE cabinets MODIFY abstract_id VARCHAR(20) NULL"#);
assert_contains(r#"db/migrate.rs"#, src, r#"ALTER TABLE charge_records MODIFY max_temperature BIGINT NOT NULL DEFAULT 0 COMMENT '最高温度(K×100)'"#);
assert_contains(r#"db/migrate.rs"#, src, r#"ALTER TABLE charge_orders MODIFY COLUMN reason VARCHAR(20) NULL COMMENT '结束原因: normal/user_stop/forced/timeout'"#);
assert_contains(r#"db/migrate.rs"#, src, r#"ALTER TABLE compartments MODIFY COLUMN status TINYINT DEFAULT 0 COMMENT '0空闲 1充电中(仅表示占用);充满/故障由通道实时状态派生,不写入本列'"#);
assert_contains(r#"db/migrate.rs"#, src, r#"ALTER TABLE charge_orders MODIFY COLUMN status TINYINT DEFAULT 0 COMMENT '0充电中 1已完成 2异常(保留未使用:平台当前无写入方,结束原因见 reason)'"#);
assert_contains(r#"db/migrate.rs"#, src, r#"/// 代码侧已加「按仓体的进程内锁 + 条件插入」,本迁移是**第三层(DB 层)兜底**:"#);
assert_contains(r#"db/migrate.rs"#, src, r#"/// 即使将来有人绕过封装直接 `INSERT`,也写不出第二笔活跃订单。"#);
assert_contains(r#"db/migrate.rs"#, src, r#"/// # 方案:生成列 + 唯一索引(MySQL 不支持条件唯一索引)"#);
assert_contains(r#"db/migrate.rs"#, src, r#"/// ALTER TABLE charge_orders ADD COLUMN active_flag BIGINT"#);
assert_contains(r#"db/migrate.rs"#, src, r#")⇒ 只补建索引,不重复 `ALTER`。
///
/// 失败一律上抛给 [`run`],由 `run` 记 ERROR 后**继续启动**(绝不让服务起不来)。
async fn ensure_active_order_unique(pool: &MySqlPool) -> Result<(), sqlx::Error> {
/// 目标表名(仅用于 information_schema 查询与日志)
const TABLE: &str = "#);
assert_contains(r#"db/migrate.rs"#, src, r#",
INDEX
);
Ok(())
}
/// 列级幂等迁移:列不存在时才 ALTER TABLE 补列
///
/// 为什么需要:`CREATE TABLE IF NOT EXISTS` 只保证「表存在」,
/// 对**已存在**的表永远不会补后加的列,导致线上 schema 漂移。
/// 安全性:表名/列名/DDL 全部来自本文件内的白名单常量,不含任何外部输入拼接。
async fn ensure_column(
pool: &MySqlPool,
table: &str,
column: &str,
ddl: &str,
) -> Result<(), sqlx::Error> {
let exists: (i64,) = sqlx::query_as(
"#);
assert_contains(r#"db/migrate.rs"#, src, r#",
)
.bind(table)
.bind(column)
.fetch_one(pool)
.await?;
if exists.0 == 0 {
let alter = format!("#);
assert_contains(r#"db/migrate.rs"#, src, r#", table, column, ddl);
sqlx::query(&alter).execute(pool).await?;
tracing::info!("#);
assert_contains(r#"db/migrate.rs"#, src, r#"/// 列可空性幂等迁移:仅当该列**当前为 NOT NULL**(老库)时才 ALTER 为可空"#);
assert_contains(r#"db/migrate.rs"#, src, r#"/// 为什么:`cabinets.abstract_id` 在老库上是 `NOT NULL`,需要改成可空;"#);
assert_contains(r#"db/migrate.rs"#, src, r#"/// 但无条件执行 MODIFY 会让**每次服务启动**都对 `cabinets` 做一次 DDL"#);
assert_contains(r#"db/migrate.rs"#, src, r#"/// (MODIFY 会走「重建整表」路径,大表上会长时间持有元数据锁、阻塞并发读写)。"#);
assert_contains(r#"db/migrate.rs"#, src, r#"/// 列注释幂等更新:仅当现有注释不包含 `marker` 时才 ALTER"#);
assert_contains(r#"db/migrate.rs"#, src, r#"/// 为什么:老库的列注释不会随 CREATE TABLE IF NOT EXISTS 更新;"#);
assert_contains(r#"db/migrate.rs"#, src, r#"/// 而 MODIFY 会重建整表,因此必须先比对注释,避免每次启动都白跑一次 DDL。"#);
assert_contains(r#"db/migrate.rs"#, src, r#"/// 返回 `Ok(true)` 表示本次真的执行了 DDL(供 [`align_legacy_columns`] 统计),"#);
assert_contains(r#"db/migrate.rs"#, src, r#",
table,
column,
current,
marker,
ddl
);
return Ok(true);
}
}
Ok(false)
}
/// 逐列补齐历史版本遗漏的列(幂等,兼容线上已存在的旧表)
///
/// 线上 pms 库曾因缺少 `cabinets.auth_str_issued` / `cabinets.cell`,
/// 导致设备认证链路直接报 `1054 Unknown column`,只能人工 ALTER 补救;
/// 本函数就是把这个人工步骤固化进启动迁移。
async fn add_columns_if_not_exists(pool: &MySqlPool) -> Result<(), sqlx::Error> {
// users:头像 + 微信 openid
ensure_column(pool, "#);
assert_contains(r#"db/migrate.rs"#, src, r#"/// 表名(本文件内的白名单常量,不含任何外部输入)"#);
assert_contains(r#"db/migrate.rs"#, src, r#"/// `information_schema.COLUMNS.DATA_TYPE` 的目标值(大小写不敏感比较)"#);
assert_contains(r#"db/migrate.rs"#, src, r#"data_type: &'static str,"#);
assert_contains(r#"db/migrate.rs"#, src, r#"/// 目标可空性(`true` = 允许 NULL,对应 `IS_NULLABLE = 'YES'`)"#);
assert_contains(r#"db/migrate.rs"#, src, r#", table, column, target),"#);
assert_contains(r#"db/migrate.rs"#, src, r#"/// **列定义对齐清单**:`(表名, 列名, 目标定义)` —— 目标定义 = `SCHEMA_SQL` 里该列的 DDL 原文"#);
assert_contains(r#"db/migrate.rs"#, src, r#"/// 两件事叠加,让「列在、但类型/默认值/单位错了」的漂移**永远不会被自动修复**:"#);
assert_contains(r#"db/migrate.rs"#, src, r#"/// 1. `CREATE TABLE IF NOT EXISTS` 只保证「表存在」,对**已存在**的表不改任何列定义;"#);
assert_contains(r#"db/migrate.rs"#, src, r#"/// Rust type `Option<NaiveDateTime>` (as SQL type `DATETIME`)"#);
assert_contains(r#"db/migrate.rs"#, src, r#"/// is not compatible with SQL type `TIMESTAMP`"#);
assert_contains(r#"db/migrate.rs"#, src, r#"/// 已实测的两处后果:测试环境充电记录导出必败(`pms_dev.charge_records.start_time` 曾是"#);
assert_contains(r#"db/migrate.rs"#, src, r#"/// `TIMESTAMP`)、生产 `/api/charge-orders` 一有订单就 500(`pms.charge_orders.start_time`"#);
assert_contains(r#"db/migrate.rs"#, src, r#"),
// `updated_at` 家族:多一条 ON UPDATE(SCHEMA_SQL 原文如此,缺了它 `updated_at` 不会自动刷新)
("#);
assert_contains(r#"db/migrate.rs"#, src, r#",
),
];
/// **纯注释口径对齐清单**:`(表名, 列名, 注释里必须出现的片段, 需要时执行的 ALTER 原文)`
///
/// # 为什么与 [`LEGACY_COLUMN_TARGETS`] 分开列
/// 1. 这 3 列**只有注释措辞不同**(类型/可空性/默认值两库本来就一致),走
/// [`ensure_column_comment`] 这条既有范式即可,不必伪装成"#);
assert_contains(r#"db/migrate.rs"#, src, r#"/// 单列对齐:先查 `information_schema.COLUMNS`,**只有与目标不一致才 ALTER**(幂等)"#);
assert_contains(r#"db/migrate.rs"#, src, r#"/// 返回 `Ok(true)` 表示本次真的执行了 DDL;`Ok(false)` 表示本来就已经一致(no-op)。"#);
assert_contains(r#"db/migrate.rs"#, src, r#"/// 比对五个维度 —— 类型 / 可空性 / 默认值 / `ON UPDATE` / 注释,"#);
assert_contains(r#"db/migrate.rs"#, src, r#"/// 与 [`LEGACY_COLUMN_TARGETS`] 里的目标定义原文逐项对应:"#);
assert_contains(r#"db/migrate.rs"#, src, r#");
Ok(())
}
/// 创建三个默认角色(INSERT IGNORE,已存在则跳过)
async fn insert_default_roles(pool: &MySqlPool) -> Result<(), sqlx::Error> {
// 创建总管理员角色
sqlx::query(
"#);
assert_contains(r#"db/migrate.rs"#, src, r#"]);
}
/// 解析器自检:目标定义原文 → 判定字段(类型/可空性/默认值/ON UPDATE/注释)
///
/// 判定字段与 DDL 全部由目标定义原文推出。解析一旦错了,就会出现
/// 「判定说不一致 → ALTER 跑了 → 下一轮判定仍说不一致 → 每次启动都 DDL」的死循环。
#[test]
fn test_column_spec_from_target_parses_all_dimensions() {
// NOT NULL + 默认值 + ON UPDATE
let spec = ColumnSpec::from_target(
"#);
assert_contains(r#"db/migrate.rs"#, src, r#",
expected.0,
expected.1,
actual
);
}
}
/// 注释口径清单自检:`marker` 必须真的出现在 `ddl` 写进去的注释里
///
/// 否则就会出现「marker 永远匹配不上 → 每次启动都 ALTER」的死循环。
/// 另外 `ddl` 必须显式写清可空性(这两张表的相关列本来就允许 NULL,
/// `MODIFY` 漏写 `NULL` 会把列改成 NOT NULL —— 那就是改结构而不是改注释了)。
#[test]
fn test_legacy_column_comments_marker_matches_ddl() {
assert_eq!(
LEGACY_COLUMN_COMMENTS.len(),
3,
"#);
assert_contains(r#"db/migrate.rs"#, src, r#",下一个人就会照着它再写一次 `WHERE status = 1` 当在线统计。
#[test]
fn test_cabinets_status_comment_is_config_state_not_online() {
let line = cabinets_status_ddl_line();
assert!(line.contains("#);
assert_contains(r#"db/migrate.rs"#, src, r#", line);
}
/// 启动迁移的注释对齐**幂等且只改注释**:
/// ① `marker` 必须出现在 DDL 的 `COMMENT` 里(否则每次启动都重复 `MODIFY`);
/// ② DDL 保留 `TINYINT DEFAULT 0`(不动类型/默认值),也不改可空性;
/// ③ 与 `SCHEMA_SQL` 建表语句的注释口径**一致**(否则新库与老库再次分叉)。
#[test]
fn test_cabinets_status_comment_alignment_is_idempotent_and_comment_only() {
let (marker, ddl) = CABINETS_STATUS_COMMENT_ALIGN;
assert!(
ddl.contains(marker),
"#);
assert_contains(r#"db/migrate.rs"#, src, r#", ddl);
// 与建表语句口径一致:新库(SCHEMA_SQL)与老库(启动 ALTER)不得分叉
let line = cabinets_status_ddl_line();
assert!(
line.contains(marker),
"#);
assert_contains(r#"db/migrate.rs"#, src, r#"/// 建表语句里**不允许再出现 `TIMESTAMP` 列类型**"#);
assert_contains(r#"db/migrate.rs"#, src, r#"/// `TIMESTAMP` 只与 `DateTime<Utc>` 兼容,而本项目时间列在 Rust 侧全是"#);
assert_contains(r#"db/migrate.rs"#, src, r#"/// `NaiveDateTime` / `Option<NaiveDateTime>` ⇒ 任何一处 TIMESTAMP 都会让"#);
assert_contains(r#"db/migrate.rs"#, src, r#"/// 对应接口 500(实测:测试环境导出失败、生产订单列表 500)。"#);
}
}

View File

@ -382,19 +382,6 @@ async fn handle_auth_str(
.collect(),
};
// 写入 Redis 缓存、标记已下发、回包(权威值在 DB,随后一并写回 DB)
deliver_auth_str(mysql, redis, msg, dev_id, auth_str).await?;
Ok(())
}
/// 下发安全码的收尾三步:写 Redis 缓存 → 标记 DB/Redis 已下发 → 回包。
///
/// 错误处理口径与内联时逐条一致(**不得改动**):
/// - Redis 缓存 `SET` 失败**只记 ERROR、不中断**(权威值在 DB,随后由 `mark_auth_str_issued` 一并写回);
/// - `mark_auth_str_issued` 内部同样是"两处失败都只记 ERROR"(见其文档);
/// - 只有 `send_response_to_device` 沿原样用 `?` 上抛。
async fn deliver_auth_str(mysql: &MySqlPool, redis: &ConnectionManager, msg: &serde_json::Value, dev_id: &str, auth_str: String) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
// 写入 Redis 缓存(权威值在 DB,随后一并写回 DB)
let key = format!("{REDIS_KEY_AUTH_STR}{dev_id}");
let mut conn = redis.clone();
@ -2092,8 +2079,22 @@ async fn settle_finished_order(
};
let energy_wh = energy::mwh_to_wh(mwh);
// 只改**已经收尾**的那一行(`status <> 0`):任何情况下都不碰活跃单
write_back_order_energy(pool, order_id, energy_wh, take_energy).await?;
if take_energy {
// 只改**已经收尾**的那一行(`status <> 0`):任何情况下都不碰活跃单
let updated = sqlx::query("UPDATE charge_orders SET energy = ? WHERE id = ? AND status <> 0")
.bind(energy_wh)
.bind(order_id)
.execute(pool)
.await?
.rows_affected();
if updated == 0 {
tracing::error!(
"[order] 电量写回未命中订单行(订单未处于结束态?)order_id={} energy={}Wh",
order_id,
energy_wh
);
}
}
// 生成充电记录(供后台「充电记录」页 / 导出 / 小程序今日统计使用)
// 字段从订单行直接 SELECT 过来,避免再查一次仓体与柜子
@ -2142,33 +2143,6 @@ async fn settle_finished_order(
Ok(())
}
/// 把累计电量写回**已经收尾**的订单行(从 `settle_finished_order` 原样搬出)。
///
/// - `take_energy = false`:本单不是通道累加器的归属者(历史的重复活跃单)⇒ 整段跳过,不写电量;
/// - `take_energy = true`:只改 `status <> 0` 的行,**任何情况下都不碰活跃单**。
///
/// 错误处理口径与内联时逐条一致(**不得改动**):SQL 执行失败仍用 `?` 上抛;
/// `rows_affected() == 0` 仍**只记 ERROR、不中断**(订单已正常收尾,不能因此回滚)。
async fn write_back_order_energy(pool: &MySqlPool, order_id: i64, energy_wh: i64, take_energy: bool) -> Result<(), sqlx::Error> {
if take_energy {
// 只改**已经收尾**的那一行(`status <> 0`):任何情况下都不碰活跃单
let updated = sqlx::query("UPDATE charge_orders SET energy = ? WHERE id = ? AND status <> 0")
.bind(energy_wh)
.bind(order_id)
.execute(pool)
.await?
.rows_affected();
if updated == 0 {
tracing::error!(
"[order] 电量写回未命中订单行(订单未处于结束态?)order_id={} energy={}Wh",
order_id,
energy_wh
);
}
}
Ok(())
}
/// 复位 / 置位仓体占用状态(0 空闲 / 1 充电中)。失败只记 ERROR,不影响设备流程。
async fn mark_compartment_status(pool: &MySqlPool, compartment_id: i64, status: i8) {
let result: Result<sqlx::mysql::MySqlQueryResult, sqlx::Error> =