12451 f80b3b634c fix: 五角度审查后的安全/正确性/契约修复(含 order_timeout 反条件、连接接管、权限兜底)
## 严重问题
- order_timeout: 超时条件方向写反(`start_time < TIMESTAMPADD(HOUR,10,NOW())` 等价于
  `start_time < NOW()+10h`,条件恒真)→ 活跃订单会被当超时单提前结掉。改为
  `TIMESTAMPADD(HOUR, ?, start_time) < NOW()`,并补"刚建单不命中/超期命中"双向断言
- C1 device-server 连接池可被未验签的报文 dev_id 顶替(覆盖式 insert + 断连无条件 remove):
  改为 conn_id 归属模型 —— 唯一连接标识、remove_if/touch/is_authenticated 均带 conn 归属校验、
  接管时用 oneshot 信号主动驱逐旧连接、每条上行前用 is_owner 复核持有者身份
- C2 POST /api/devices/{imei}/command 零归属校验 + act 自由字符串 + 把 IMEI 塞进协议 id:
  补 scope_guard 归属校验 + act 白名单 + id 从 params 显式取(拒绝 15 位 IMEI)
- 非总管理员权限集恒为空(user_roles 从不写入且无分配入口)→ 全站 403:
  user_roles 为空时按 users.role 回退内置角色权限 + 建/改用户幂等同步映射(仅角色变化时同步)

## 订单生命周期并发正确性
- finalize_order 先取电量累加器后抢结束权 → off/bat_out 成对到达时 energy 被写 0:
  改为先原子抢占(WHERE compartment_id=? AND status=0 ORDER BY id DESC LIMIT 1),
  抢到者才 take_and_reset 取电量;抢不到直接返回且不碰累加器
- 建单"先查后插"无锁无唯一约束 → 同仓体两笔活跃单:
  条件插入(INSERT ... SELECT ... WHERE NOT EXISTS)+ 按仓体进程内锁
  + DB 层 active_flag 生成列 + UNIQUE 唯一索引(幂等迁移,有重复则记 ERROR 跳过,绝不阻断启动)
- handle_bat_out: updated==0 时不再复位占用;通道登记删除改为 battery_id 一致才删;
  新增 should_finalize_bat_out 挡住"迟到 bat_out 结束新单"
- 电压/电流加量程夹取(0–100000 mV/mA 之外按 0 计并 WARN),防位翻转导致积分暴增

## 协议/契约
- spec 绝不可能是 "0000"(生产 42/68 台拓扑为空):新增 build_spec 三级兜底
  (真实拓扑 → cabinets.cell×6 → 6×6),兜底生效记 WARN 报出拓扑数据
- /api/cabinets/realtime-status 改读真实字段(原来读的设备链路从不写的旧字段,恒返回 idle/空),
  用 derive_channel_view 推导旧字段以保持响应字段名不变;修掉"把通道号当板号"的幽灵板问题
- 小程序通道详情/PC 详情改读通道 Hash(弃用无写入方的 compartment:status:{id}),
  取不到返回 null 而非假的 0;补 voltage/current/temp/batt_status/pow_status
- 反馈列表补权限码 operation_log:view + status 筛选真正落到 SQL
- 小程序柜温换算(29800 曾被显示成 29800°C);签名注释 7→8 位;msg_id 序列化省略 null;
  parse_compartment_id 与 status_post 补板号 1–15/通道 1–8 量程校验(不再写 board:0 垃圾 key)

## 文档
- docs/柜控板固件对接文档.md:确立数值口径统一规则(hex 字节流=原始硬件单位·平台换算;
  JSON 数值字段=平台单位·设备换算)、修正 A.5.3 的 id 说明、作废变更记录原位标注(不删历史)、
  spec 表述改为"按拓扑现算+兜底、恒 4 位不会是 0000"、修复文首 7 项表格被引用块切断丢表头、
  修正 §10 误引 channels[].channel、写明无电池兜底阈值与 #16 风险

## 运维加固(服务器侧,非代码)
- 关闭 MySQL/Redis 公网暴露:nginx stream 由 0.0.0.0 改为只监听 127.0.0.1 + ZeroTier
- worker_shutdown_timeout 30s(原先每次 reload 泄漏永不退出的 worker)
- 测试证书改指向 app 证书(原测试证书 9 天后过期且无续期机制);补测试环境 uploads 目录;.env 0600

## 验收
- cargo check 两服务 0 error / 0 warning;cargo test api 88 + device 22 全过
- 路由 79 → 79 逐条一致;web tsc 干净
- 端到端联调(测试环境)13/13,订单结算与重构前逐字段一致;唯一索引已由迁移真实建出
- 生产已部署并验证:服务 active、health 200、ERROR 0、消费链路正常、迁移已建索引
2026-09-20 00:36:37 +08:00

768 lines
33 KiB
Rust
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

//! 数据库迁移 — 自动建表
use sqlx::MySqlPool;
/// 全量建表语句(幂等:全部 `CREATE TABLE IF NOT EXISTS`)。
/// 抽成常量是为了让 [`run`] 保持可读(此前 280 行的函数里有 233 行是这段 SQL)。
const SCHEMA_SQL: &str = r#"
CREATE TABLE IF NOT EXISTS organizations (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
name VARCHAR(100) NOT NULL COMMENT '组织名称',
parent_id BIGINT NULL COMMENT '父组织ID(NULL=顶级组织)',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
CREATE TABLE IF NOT EXISTS projects (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
organization_id BIGINT NOT NULL,
name VARCHAR(100) NOT NULL COMMENT '项目名称',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
FOREIGN KEY (organization_id) REFERENCES organizations(id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
CREATE TABLE IF NOT EXISTS cabinets (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
project_id BIGINT,
abstract_id VARCHAR(20) NULL UNIQUE COMMENT '平台抽象ID,如10-00000001',
imei VARCHAR(15) NOT NULL UNIQUE COMMENT '4G模块IMEI',
iccid VARCHAR(20) COMMENT 'SIM卡ICCID',
auth_str VARCHAR(8) COMMENT '签名安全码',
auth_str_issued BOOLEAN DEFAULT FALSE COMMENT '安全码是否已下发',
cell TINYINT DEFAULT 6 COMMENT '仓控板数量',
work_mode TINYINT DEFAULT 0 COMMENT '充电工作模式: 0=平台指令模式(默认), 1=自动充电工作模式',
name VARCHAR(100) COMMENT '柜子名称',
status TINYINT DEFAULT 0 COMMENT '0离线 1在线',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
FOREIGN KEY (project_id) REFERENCES projects(id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
CREATE TABLE IF NOT EXISTS sim_change_logs (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
imei VARCHAR(15) NOT NULL COMMENT '设备IMEI',
old_iccid VARCHAR(20) COMMENT '旧ICCID',
new_iccid VARCHAR(20) NOT NULL COMMENT '新ICCID',
changed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT '变更时间',
INDEX idx_imei (imei)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
CREATE TABLE IF NOT EXISTS batteries (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
battery_id VARCHAR(50) NOT NULL UNIQUE COMMENT '电池唯一标识',
soc INT DEFAULT 0 COMMENT '电量(‰)',
brand VARCHAR(50) COMMENT '品牌(由 brand_code 映射出的名称,代码表未定前为空)',
model VARCHAR(50) COMMENT '型号(由 model_code 映射出的名称,代码表未定前为空)',
sn VARCHAR(100) COMMENT '序列号',
-- 以下为协议 §5.5「电池属性」原值(硬件代码,平台侧负责映射,见交付文档 §10 定稿第 7 条)
brand_code VARCHAR(20) COMMENT '厂家代码(原值透传)',
model_code VARCHAR(20) COMMENT '电池型号代码(原值透传)',
material_code VARCHAR(20) COMMENT '电芯材料代码(原值透传)',
cell_count INT COMMENT '电芯节数',
rated_voltage BIGINT COMMENT '额定电压(mV)',
rated_capacity BIGINT COMMENT '额定容量(mAh)',
status TINYINT DEFAULT 0 COMMENT '0空闲 1使用中 2故障 3报废',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
CREATE TABLE IF NOT EXISTS charge_orders (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
compartment_id BIGINT NOT NULL COMMENT '仓体ID',
cabinet_id BIGINT NOT NULL COMMENT '柜子ID',
battery_id VARCHAR(50) NOT NULL COMMENT '电池ID',
start_time TIMESTAMP NULL COMMENT '充电开始时间',
end_time TIMESTAMP NULL COMMENT '充电结束时间',
start_soc INT COMMENT '起始电量(‰)',
end_soc INT COMMENT '结束电量(‰)',
energy INT DEFAULT 0 COMMENT '充电量(Wh)',
reason VARCHAR(20) COMMENT '结束原因: normal/user_stop/forced/timeout',
status TINYINT DEFAULT 0 COMMENT '0充电中 1已完成 2异常',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (compartment_id) REFERENCES compartments(id),
FOREIGN KEY (cabinet_id) REFERENCES cabinets(id),
INDEX idx_battery (battery_id),
INDEX idx_compartment (compartment_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
CREATE TABLE IF NOT EXISTS cabin_boards (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
cabinet_id BIGINT NOT NULL,
board_id TINYINT NOT NULL COMMENT '仓控板ID 1-6',
status TINYINT DEFAULT 0,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
FOREIGN KEY (cabinet_id) REFERENCES cabinets(id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
CREATE TABLE IF NOT EXISTS compartments (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
cabin_board_id BIGINT NOT NULL,
channel_id TINYINT NOT NULL COMMENT '通道ID 1-6',
status TINYINT DEFAULT 0 COMMENT '0空闲 1充电中 2故障 3禁用',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
FOREIGN KEY (cabin_board_id) REFERENCES cabin_boards(id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
CREATE TABLE IF NOT EXISTS users (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
phone VARCHAR(11) NOT NULL UNIQUE COMMENT '手机号',
name VARCHAR(50) COMMENT '姓名',
password VARCHAR(255) NOT NULL,
role TINYINT DEFAULT 0 COMMENT '0操作员 1企业管理员 2总管理员',
organization_id BIGINT,
status TINYINT DEFAULT 1 COMMENT '0禁用 1启用',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
FOREIGN KEY (organization_id) REFERENCES organizations(id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
CREATE TABLE IF NOT EXISTS charge_records (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
compartment_id BIGINT NOT NULL,
cabinet_id BIGINT NOT NULL,
start_time TIMESTAMP NULL COMMENT '充电开始时间',
end_time TIMESTAMP NULL COMMENT '充电结束时间',
energy BIGINT NOT NULL DEFAULT 0 COMMENT '充电量(Wh)',
max_temperature BIGINT NOT NULL DEFAULT 0 COMMENT '最高温度(K×100)',
status TINYINT DEFAULT 0 COMMENT '0正常 1异常 2手动停止',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (compartment_id) REFERENCES compartments(id),
FOREIGN KEY (cabinet_id) REFERENCES cabinets(id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
CREATE TABLE IF NOT EXISTS device_logs (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
cabinet_id BIGINT NOT NULL,
log_type TINYINT NOT NULL COMMENT '1状态上报 2故障 3告警 4停电',
content TEXT COMMENT '日志内容JSON',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (cabinet_id) REFERENCES cabinets(id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
CREATE TABLE IF NOT EXISTS energy_stats (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
cabinet_id BIGINT NOT NULL,
stat_date DATE NOT NULL,
total_energy BIGINT NOT NULL DEFAULT 0 COMMENT '总能耗(Wh)',
charge_count INT DEFAULT 0 COMMENT '充电次数',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (cabinet_id) REFERENCES cabinets(id),
UNIQUE KEY uk_cabinet_date (cabinet_id, stat_date)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
CREATE TABLE IF NOT EXISTS operation_logs (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
user_id BIGINT NOT NULL,
action VARCHAR(100) NOT NULL COMMENT '操作动作',
target_type VARCHAR(50) COMMENT '操作对象类型',
target_id BIGINT COMMENT '操作对象ID',
detail TEXT COMMENT '操作详情',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (user_id) REFERENCES users(id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
CREATE TABLE IF NOT EXISTS roles (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
name VARCHAR(50) NOT NULL UNIQUE COMMENT '角色名称',
description VARCHAR(200) COMMENT '角色描述',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
CREATE TABLE IF NOT EXISTS permissions (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
code VARCHAR(50) NOT NULL UNIQUE COMMENT '权限码,如device:view',
name VARCHAR(100) NOT NULL COMMENT '权限名称',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
CREATE TABLE IF NOT EXISTS role_permissions (
role_id BIGINT NOT NULL,
permission_id BIGINT NOT NULL,
PRIMARY KEY (role_id, permission_id),
FOREIGN KEY (role_id) REFERENCES roles(id) ON DELETE CASCADE,
FOREIGN KEY (permission_id) REFERENCES permissions(id) ON DELETE CASCADE
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
CREATE TABLE IF NOT EXISTS user_roles (
user_id BIGINT NOT NULL,
role_id BIGINT NOT NULL,
PRIMARY KEY (user_id, role_id),
FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE,
FOREIGN KEY (role_id) REFERENCES roles(id) ON DELETE CASCADE
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
CREATE TABLE IF NOT EXISTS user_scopes (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
user_id BIGINT NOT NULL,
scope_type TINYINT NOT NULL COMMENT '1组织 2项目 3设备',
scope_id BIGINT NOT NULL COMMENT '对应组织/项目/设备ID',
permission_type TINYINT NOT NULL COMMENT '1查看 2操作',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
CREATE TABLE IF NOT EXISTS download_tasks (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
user_id BIGINT NOT NULL,
task_type VARCHAR(50) NOT NULL COMMENT 'charge_records/energy_stats/device_logs',
status TINYINT DEFAULT 0 COMMENT '0排队 1处理中 2完成 3失败',
file_path VARCHAR(500) COMMENT '文件路径',
file_name VARCHAR(200) COMMENT '文件名',
progress INT DEFAULT 0 COMMENT '进度百分比',
error_message TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
completed_at TIMESTAMP NULL COMMENT '完成时间',
FOREIGN KEY (user_id) REFERENCES users(id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
CREATE TABLE IF NOT EXISTS abstract_bindings (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
abstract_id VARCHAR(20) NOT NULL UNIQUE COMMENT '平台抽象ID,如10-00000001',
cabinet_id BIGINT NOT NULL UNIQUE COMMENT '柜子ID',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (cabinet_id) REFERENCES cabinets(id) ON DELETE CASCADE
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
CREATE TABLE IF NOT EXISTS user_feedback (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
user_id BIGINT NOT NULL COMMENT '提交用户ID',
content TEXT NOT NULL COMMENT '反馈内容',
contact VARCHAR(200) COMMENT '联系方式',
status TINYINT DEFAULT 0 COMMENT '0待处理 1已处理',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (user_id) REFERENCES users(id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
"#;
/// 执行数据库迁移,创建所有必要的数据表
pub async fn run(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?;
}
// 兼容已有数据库:将 abstract_id 改为可空
let _ = sqlx::query(r#"ALTER TABLE cabinets MODIFY abstract_id VARCHAR(20) NULL"#)
.execute(pool).await;
// 更新 charge_records.max_temperature 注释(0.1°C → K×100)
// 只在注释确实过时(老库)时才 ALTER:避免每次启动都对整表做一次 MODIFY
ensure_column_comment(
pool,
"charge_records",
"max_temperature",
"K×100",
r#"ALTER TABLE charge_records MODIFY max_temperature BIGINT NOT NULL DEFAULT 0 COMMENT '最高温度(K×100)'"#,
)
.await?;
// 更新 charge_orders.reason 注释:新增 timeout
// (订单超过 10 小时由 order_timeout 定时任务自动结束,落该枚举值)
ensure_column_comment(
pool,
"charge_orders",
"reason",
"timeout",
r#"ALTER TABLE charge_orders MODIFY COLUMN reason VARCHAR(20) NULL COMMENT '结束原因: normal/user_stop/forced/timeout'"#,
)
.await?;
// 创建关键索引(IF NOT EXISTS 避免重复)
create_indexes(pool).await?;
// 兼容已有数据库:新增字段(幂等,列已存在则跳过)
add_columns_if_not_exists(pool).await?;
// P0-2 的 DB 层兜底:**同一仓体最多一笔活跃订单**(`status = 0`)的唯一约束。
//
// ⚠️ 这一步的失败**绝不能中断启动**:`run()` 返回 Err 会让 main panic 退出(生产事故),
// 而业务侧已经有「按仓体的进程内锁 + 条件插入」两层防护,DB 约束只是第三层兜底。
// 因此这里只记 ERROR 后继续(与上方 `ALTER TABLE cabinets ...` 的容错口径一致)。
if let Err(e) = ensure_active_order_unique(pool).await {
tracing::error!(
"[migrate] 活跃订单唯一约束迁移失败(服务继续启动;业务侧仍有进程内锁+条件插入兜底)err={}",
e
);
}
// 初始化默认权限码(INSERT IGNORE 避免重复插入)
seed_permissions(pool).await?;
// 初始化默认角色
seed_roles(pool).await?;
tracing::info!("数据库迁移完成");
Ok(())
}
/// 创建关键查询索引(幂等,已存在则跳过)
async fn create_indexes(pool: &MySqlPool) -> Result<(), sqlx::Error> {
/// 辅助:检查索引是否存在,不存在则创建
async fn ensure_index(
pool: &MySqlPool,
table: &str,
index: &str,
columns: &str,
) -> Result<(), sqlx::Error> {
let exists: (i64,) = sqlx::query_as(
"SELECT COUNT(*) FROM information_schema.statistics \
WHERE table_schema = DATABASE() AND table_name = ? AND index_name = ?",
)
.bind(table)
.bind(index)
.fetch_one(pool)
.await?;
if exists.0 == 0 {
let ddl = format!("CREATE INDEX {} ON {}({})", index, table, columns);
sqlx::query(&ddl).execute(pool).await?;
}
Ok(())
}
ensure_index(pool, "cabinets", "idx_cabinets_project", "project_id").await?;
ensure_index(pool, "cabin_boards", "idx_cabin_boards_cabinet", "cabinet_id").await?;
ensure_index(pool, "compartments", "idx_compartments_board", "cabin_board_id").await?;
ensure_index(pool, "charge_records", "idx_charge_records_cabinet", "cabinet_id").await?;
ensure_index(pool, "device_logs", "idx_device_logs_cabinet", "cabinet_id").await?;
// 订单超时兜底扫描(status = 0 AND start_time < ...)走该联合索引,避免全表扫描
ensure_index(pool, "charge_orders", "idx_charge_orders_status_start", "status, start_time").await?;
// device_logs 保留策略按 created_at 分批删除,必须有该列索引,否则每次清理全表扫描
ensure_index(pool, "device_logs", "idx_device_logs_created", "created_at").await?;
tracing::info!("数据库索引创建完成");
Ok(())
}
/// DB 层兜底:**同一仓体最多只能有一笔活跃订单**(`status = 0`)的唯一约束。
///
/// # 为什么需要(P0-2)
/// `workers/report_worker.rs` 的 `ensure_charge_order` 原本是"先查活跃单、再插入":
/// 判定与插入之间没有事务、没有锁、也没有唯一键,并发两次建单
/// (H5 双击/重试、两人同时点、设备上报 `on` 与平台下发 `on` 撞在一起)
/// 会各插一笔 `status = 0`。后果:结单只收最新一笔、老单**永久卡在 `status = 0`**,
/// 还会被 `order_timeout` 扫到后**把正在充电的新单当超时单结束**,
/// 一次充电会话产生两条 `charge_records`("今日充电次数"虚增)。
///
/// 代码侧已加「按仓体的进程内锁 + 条件插入」,本迁移是**第三层(DB 层)兜底**:
/// 即使将来有人绕过封装直接 `INSERT`,也写不出第二笔活跃订单。
///
/// # 方案:生成列 + 唯一索引(MySQL 不支持条件唯一索引)
/// ```sql
/// ALTER TABLE charge_orders ADD COLUMN active_flag BIGINT
/// GENERATED ALWAYS AS (CASE WHEN status = 0 THEN compartment_id ELSE NULL END) STORED;
/// CREATE UNIQUE INDEX uk_charge_orders_active ON charge_orders(active_flag);
/// ```
/// `status <> 0`(已完成/异常)的行 `active_flag = NULL`,而 **MySQL 唯一索引不约束 NULL**
/// (多行 NULL 可以共存)—— 正是需要的语义:历史订单可以有任意多笔,
/// **活跃订单同一仓体只能有一笔**。
///
/// # 幂等与安全(三步各自独立可重入)
/// 1. 索引已存在 ⇒ 直接返回(OLD 库/第二次启动都不会重复 DDL,避免每次启动重建整表);
/// 2. 加约束前**先预检重复活跃单**:存在重复就**只记 ERROR、放弃本次 DDL**
/// (强行加约束会报 1062;脏数据必须先人工清理,不能让迁移失败);
/// 3. 生成列已存在但索引缺失(上次"加列成功、建索引失败")⇒ 只补建索引,不重复 `ALTER`。
///
/// 失败一律上抛给 [`run`],由 `run` 记 ERROR 后**继续启动**(绝不让服务起不来)。
async fn ensure_active_order_unique(pool: &MySqlPool) -> Result<(), sqlx::Error> {
/// 目标表名(仅用于 information_schema 查询与日志)
const TABLE: &str = "charge_orders";
/// 生成列名(活跃订单哨兵)
const COLUMN: &str = "active_flag";
/// 唯一索引名(幂等判定依据)
const INDEX: &str = "uk_charge_orders_active";
// 1) 索引已存在 ⇒ 幂等跳过
let index_exists: (i64,) = sqlx::query_as(
"SELECT COUNT(*) FROM information_schema.statistics \
WHERE table_schema = DATABASE() AND table_name = ? AND index_name = ?",
)
.bind(TABLE)
.bind(INDEX)
.fetch_one(pool)
.await?;
if index_exists.0 > 0 {
return Ok(());
}
// 2) 预检:已存在重复活跃单时**不能**加唯一约束,只能记 ERROR 交人工处理
let duplicates: (i64,) = sqlx::query_as(
"SELECT COUNT(*) FROM ( \
SELECT compartment_id FROM charge_orders WHERE status = 0 \
GROUP BY compartment_id HAVING COUNT(*) > 1 \
) AS dup",
)
.fetch_one(pool)
.await?;
if duplicates.0 > 0 {
tracing::error!(
"[migrate] 检测到 {} 个仓体存在多笔活跃订单(charge_orders.status=0),\
已跳过 {} 唯一约束;请先人工清理重复活跃单(保留最新一笔、其余置为结束态)再重启服务",
duplicates.0,
INDEX
);
return Ok(());
}
// 3) 生成列(幂等:列已存在则只补建索引)
let column_exists: (i64,) = sqlx::query_as(
"SELECT COUNT(*) FROM information_schema.COLUMNS \
WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ? AND COLUMN_NAME = ?",
)
.bind(TABLE)
.bind(COLUMN)
.fetch_one(pool)
.await?;
if column_exists.0 == 0 {
sqlx::query(
"ALTER TABLE charge_orders ADD COLUMN active_flag BIGINT \
GENERATED ALWAYS AS (CASE WHEN status = 0 THEN compartment_id ELSE NULL END) STORED \
COMMENT '活跃订单唯一性哨兵:status=0 时=compartment_id,否则 NULL(NULL 不参与唯一性)'",
)
.execute(pool)
.await?;
tracing::info!("[migrate] 新增生成列 {}.{}", TABLE, COLUMN);
}
sqlx::query("CREATE UNIQUE INDEX uk_charge_orders_active ON charge_orders(active_flag)")
.execute(pool)
.await?;
tracing::info!(
"[migrate] 已创建活跃订单唯一约束 {}(同一仓体最多一笔 status=0)",
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(
"SELECT COUNT(*) FROM information_schema.COLUMNS \
WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ? AND COLUMN_NAME = ?",
)
.bind(table)
.bind(column)
.fetch_one(pool)
.await?;
if exists.0 == 0 {
let alter = format!("ALTER TABLE {} ADD COLUMN {} {}", table, column, ddl);
sqlx::query(&alter).execute(pool).await?;
tracing::info!("[migrate] 新增列 {}.{}", table, column);
}
Ok(())
}
/// 列注释幂等更新:仅当现有注释不包含 `marker` 时才 ALTER
///
/// 为什么:老库的列注释不会随 CREATE TABLE IF NOT EXISTS 更新;
/// 而 MODIFY 会重建整表,因此必须先比对注释,避免每次启动都白跑一次 DDL。
async fn ensure_column_comment(
pool: &MySqlPool,
table: &str,
column: &str,
marker: &str,
ddl: &str,
) -> Result<(), sqlx::Error> {
// ⚠️ `information_schema.COLUMNS.COLUMN_COMMENT` 在 MySQL 8 上会被 sqlx 报告为
// **VARBINARY**,直接按 `Option<String>` 解码会 panic(实测阿里云 RDS MySQL 8.0.36):
// `mismatched types; Rust type String (as SQL type VARCHAR) is not compatible with SQL type VARBINARY`
// 必须显式 `CAST(... AS CHAR)`。
let comment: Option<String> = sqlx::query_scalar(
"SELECT CAST(COLUMN_COMMENT AS CHAR) FROM information_schema.COLUMNS \
WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ? AND COLUMN_NAME = ?",
)
.bind(table)
.bind(column)
.fetch_optional(pool)
.await?;
if let Some(current) = comment {
if !current.contains(marker) {
sqlx::query(ddl).execute(pool).await?;
tracing::info!("[migrate] 更新列注释 {}.{}", table, column);
}
}
Ok(())
}
/// 逐列补齐历史版本遗漏的列(幂等,兼容线上已存在的旧表)
///
/// 线上 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, "users", "avatar", "VARCHAR(500) NULL COMMENT '头像URL'").await?;
ensure_column(pool, "users", "openid", "VARCHAR(64) NULL UNIQUE COMMENT '微信openid'")
.await?;
// organizations:父组织(组织树层级)+ 禁用标记(甲方禁用客户组织,其下柜子全部失效)
ensure_column(pool, "organizations", "parent_id", "BIGINT NULL COMMENT '父组织ID(NULL=顶级组织)'")
.await?;
ensure_column(
pool,
"organizations",
"disabled",
"TINYINT NOT NULL DEFAULT 0 COMMENT '0正常 1禁用'",
)
.await?;
// projects:地址 + 经纬度
ensure_column(pool, "projects", "address", "VARCHAR(200) NULL COMMENT '详细地址'").await?;
ensure_column(pool, "projects", "latitude", "DECIMAL(10,7) NULL COMMENT '纬度'").await?;
ensure_column(pool, "projects", "longitude", "DECIMAL(10,7) NULL COMMENT '经度'").await?;
// cabinets:地址 + 经纬度
ensure_column(pool, "cabinets", "address", "VARCHAR(200) NULL COMMENT '详细地址'").await?;
ensure_column(pool, "cabinets", "latitude", "DECIMAL(10,7) NULL COMMENT '纬度'").await?;
ensure_column(pool, "cabinets", "longitude", "DECIMAL(10,7) NULL COMMENT '经度'").await?;
// cabinets:设备认证 / 拓扑 / 工作模式(线上曾缺失,必须补齐)
ensure_column(
pool,
"cabinets",
"auth_str_issued",
"BOOLEAN DEFAULT FALSE COMMENT '安全码是否已下发'",
)
.await?;
ensure_column(pool, "cabinets", "cell", "TINYINT DEFAULT 6 COMMENT '仓控板数量'").await?;
ensure_column(
pool,
"cabinets",
"work_mode",
"TINYINT DEFAULT 0 COMMENT '充电工作模式: 0=平台指令模式(默认), 1=自动充电工作模式'",
)
.await?;
// batteries:协议 §5.5「电池属性」的原值字段(老库缺失,必须补齐,否则台账只能丢到 device_logs)
ensure_column(pool, "batteries", "brand_code", "VARCHAR(20) NULL COMMENT '厂家代码(原值透传)'").await?;
ensure_column(pool, "batteries", "model_code", "VARCHAR(20) NULL COMMENT '电池型号代码(原值透传)'").await?;
ensure_column(pool, "batteries", "material_code", "VARCHAR(20) NULL COMMENT '电芯材料代码(原值透传)'")
.await?;
ensure_column(pool, "batteries", "cell_count", "INT NULL COMMENT '电芯节数'").await?;
ensure_column(pool, "batteries", "rated_voltage", "BIGINT NULL COMMENT '额定电压(mV)'").await?;
ensure_column(pool, "batteries", "rated_capacity", "BIGINT NULL COMMENT '额定容量(mAh)'").await?;
// user_feedback:反馈图片(JSON 数组字符串)
ensure_column(
pool,
"user_feedback",
"images",
"TEXT NULL COMMENT '反馈图片URL列表(JSON数组)'",
)
.await?;
tracing::info!("数据库列级迁移完成");
Ok(())
}
/// 插入默认权限码
async fn seed_permissions(pool: &MySqlPool) -> Result<(), sqlx::Error> {
let perms: Vec<(&str, &str)> = vec![
("device:view", "查看设备"),
("device:operate", "操作设备"),
("device:create", "添加设备"),
("device:edit", "编辑设备"),
("device:delete", "删除设备"),
// 安全码重置(仅总管理员,不授予企业管理员/操作员)
("device:auth_str:reset", "重置安全码"),
("charge:start", "开始充电"),
("charge:stop", "停止充电"),
("charge:open_door", "开门"),
("charge_record:view", "查看充电记录"),
("charge_record:export", "导出充电记录"),
("device_log:view", "查看设备日志"),
("energy:view", "查看能耗"),
("energy:export", "导出能耗"),
("user:view", "查看用户"),
("user:create", "创建用户"),
("user:edit", "编辑用户"),
("user:delete", "删除用户"),
("role:view", "查看角色"),
("role:create", "创建角色"),
("role:edit", "编辑角色"),
("role:delete", "删除角色"),
("operation_log:view", "查看操作日志"),
// 组织管理
("org:view", "查看组织"),
("org:create", "创建组织"),
("org:edit", "编辑组织"),
("org:delete", "删除组织"),
// 项目管理
("project:view", "查看项目"),
("project:create", "创建项目"),
("project:edit", "编辑项目"),
("project:delete", "删除项目"),
];
for (code, name) in perms {
sqlx::query(
"INSERT IGNORE INTO permissions (code, name) VALUES (?, ?)"
)
.bind(code)
.bind(name)
.execute(pool)
.await?;
}
tracing::info!("默认权限码初始化完成");
Ok(())
}
/// 创建默认角色并分配权限
async fn seed_roles(pool: &MySqlPool) -> Result<(), sqlx::Error> {
insert_default_roles(pool).await?;
// 为总管理员分配所有权限
let super_admin_id: Option<(i64,)> = sqlx::query_as(
"SELECT id FROM roles WHERE name = 'super_admin'"
)
.fetch_optional(pool)
.await?;
if let Some((role_id,)) = super_admin_id {
grant_all_permissions(pool, role_id).await?;
}
// 为企业管理员分配非系统管理权限
let org_admin_id: Option<(i64,)> = sqlx::query_as(
"SELECT id FROM roles WHERE name = 'org_admin'"
)
.fetch_optional(pool)
.await?;
let org_perm_codes = [
"device:view", "device:operate", "device:create", "device:edit",
"charge:start", "charge:stop", "charge:open_door",
"charge_record:view", "charge_record:export",
"device_log:view", "energy:view", "energy:export",
"user:view", "user:create", "user:edit",
"operation_log:view",
// 组织管理
"org:view", "org:create", "org:edit", "org:delete",
// 项目管理
"project:view", "project:create", "project:edit", "project:delete",
];
if let Some((role_id,)) = org_admin_id {
grant_permissions_by_codes(pool, role_id, &org_perm_codes).await?;
}
// 为操作员分配查看和操作权限
let normal_id: Option<(i64,)> = sqlx::query_as(
"SELECT id FROM roles WHERE name = 'normal_user'"
)
.fetch_optional(pool)
.await?;
let normal_perm_codes = [
"device:view", "device:operate",
"charge:start", "charge:stop", "charge:open_door",
"charge_record:view", "device_log:view", "energy:view",
];
if let Some((role_id,)) = normal_id {
grant_permissions_by_codes(pool, role_id, &normal_perm_codes).await?;
}
tracing::info!("默认角色及权限分配完成");
Ok(())
}
/// 创建三个默认角色(INSERT IGNORE,已存在则跳过)
async fn insert_default_roles(pool: &MySqlPool) -> Result<(), sqlx::Error> {
// 创建总管理员角色
sqlx::query(
"INSERT IGNORE INTO roles (name, description) VALUES (?, ?)"
)
.bind("super_admin")
.bind("总管理员 — 全部权限")
.execute(pool)
.await?;
// 创建企业管理员角色
sqlx::query(
"INSERT IGNORE INTO roles (name, description) VALUES (?, ?)"
)
.bind("org_admin")
.bind("企业管理员 — 管理本组织下项目/设备/人员")
.execute(pool)
.await?;
// 创建操作员角色
sqlx::query(
"INSERT IGNORE INTO roles (name, description) VALUES (?, ?)"
)
.bind("normal_user")
.bind("操作员 — 查看/操作指定范围设备")
.execute(pool)
.await?;
Ok(())
}
/// 为指定角色分配全部权限(授权总管理员)
async fn grant_all_permissions(pool: &MySqlPool, role_id: i64) -> Result<(), sqlx::Error> {
// 查询所有权限ID
let perm_ids: Vec<(i64,)> = sqlx::query_as(
"SELECT id FROM permissions"
)
.fetch_all(pool)
.await?;
for (perm_id,) in perm_ids {
sqlx::query(
"INSERT IGNORE INTO role_permissions (role_id, permission_id) VALUES (?, ?)"
)
.bind(role_id)
.bind(perm_id)
.execute(pool)
.await?;
}
Ok(())
}
/// 按权限码逐个为指定角色授权(权限码不存在则跳过)
async fn grant_permissions_by_codes(
pool: &MySqlPool,
role_id: i64,
codes: &[&str],
) -> Result<(), sqlx::Error> {
for code in codes {
let perm: Option<(i64,)> = sqlx::query_as(
"SELECT id FROM permissions WHERE code = ?"
)
.bind(code)
.fetch_optional(pool)
.await?;
if let Some((perm_id,)) = perm {
sqlx::query(
"INSERT IGNORE INTO role_permissions (role_id, permission_id) VALUES (?, ?)"
)
.bind(role_id)
.bind(perm_id)
.execute(pool)
.await?;
}
}
Ok(())
}