12451 c3a3ea7c3a fix(api)+fix(device): 状态取值收敛、超时方向守门补齐、在线判据口径订正
1) cabinets.status 写入侧收敛(此前 body.status 是 i8,0/99/-1 都能落库)
   - create_cabinet 的 INSERT **显式写 status = 1**,不再依赖列默认值 0:
     冻结口径是「2 = 禁用,其余 = 启用」,但列默认会产生第二种"启用"表示,
     任何按 status = 1 做的统计都会悄悄漏掉新建柜子(见「踩过的坑」第 5 条)。
   - PUT /api/cabinets/:id/status 增加白名单 {1,2},抽成纯函数 is_valid_cabinet_status
     并配守门单测(i8 全量遍历断言,杜绝"只挡住我知道的那几个")。
     白名单校验放在**鉴权之后**:越权恒 403,不会因参数非法而先返回 400
     (否则 400/403 的差异会变成"该柜子是否存在"的探测信号)。
   - 已实测自证:把白名单放宽成 [0,1,2] → 用例 FAILED / exit 101;还原 → 通过。
   - 生产库实测 68 台**全部已是 status=1**,故无需数据迁移。

2) order_timeout.rs 的超时方向守门(此前只在文档里留了「遗留」标记)
   - 选单 SQL 抽成常量 EXPIRED_ORDER_SCAN_SQL;
   - 补 3 个单测:SQL 文本方向断言(含**反证**写反形态)+ 纯函数孪生体双向断言
     (刚建 5 分钟不命中 / 超 10 小时 1 分钟命中 / 差 1 分钟不命中 + 反证)+ 常量合理性。
   - Why:本文件与 report_worker.rs 是**等价但写法不同**的超时判据
     (那边绑负数、这边把 start_time 往前推),修好一边而另一边没有守门 = 漏网;
     方向写反**零报错、零日志、单测全绿**,只会静默提前结单、提前结算电量。
   - 已实测自证:反写成 start_time < TIMESTAMPADD(HOUR, ?, NOW())
     → test_expired_order_scan_sql_points_backwards FAILED / exit 101;还原 → 3 passed。

3) device-server 的在线判据注释订正(原文与代码事实不符)
   - 原文称在线键 device:{imei} "在验签前就会被写" —— **逐行核对后确认是错的**:
     online=1 只由 api-server 的 report_worker 在 handle_login 的
     sign == expected_sign 分支内(mark_device_online)以及 handle_status_post
     (write_device_state,该 act 已被登录门控挡住)写入;handle_auth_str 不写 online。
   - 真正不能拿它判下发的原因是另一条:它的 TTL 语义是"最近 300 秒收到过任意报文",
     断连后仍存活最多 5 分钟 ⇒ 会出现"管理端显示已下发成功、设备实际收不到"。
   - 另注:server.rs::update_device_route 在 auth_str/login **到达时**写 node_id 并
     EXPIRE 300 —— 只刷 TTL、不写 online;与协议 §2.3 一致,故保留,但同样说明
     该键不能当"可下发"判据。

验收:api cargo check --all-targets 零 error/零 warning,单测 182 passed;
device 30 passed;两产物交叉编译通过,已按 device→api 顺序发布到测试与生产。
2026-09-20 04:31:42 +08:00

1919 lines
82 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.

//! 组织树 API
//!
//! 两个独立接口:
//! 1. /api/organization-tree — 返回多级组织->项目树
//! 2. /api/cabinets/filter — 根据组织/项目筛选设备列表
use axum::extract::{State, Query};
use axum::Json;
use serde::Deserialize;
use serde_json::{json, Value};
use std::collections::{BTreeMap, HashMap};
use super::organizations::{CabinetRow, OrganizationRow, ProjectRow, TreeNode};
use crate::error::AppError;
use crate::middleware::auth::{self, CurrentUser};
use crate::commands::AppState;
/// SQL查询专用结构体(只包含数据库字段)
#[derive(Debug, Clone, sqlx::FromRow)]
struct CabinetSqlRow {
id: i64,
project_id: Option<i64>,
abstract_id: Option<String>,
imei: String,
iccid: Option<String>,
name: Option<String>,
address: Option<String>,
project_address: Option<String>,
status: i8,
board_count: Option<i64>,
channel_count: Option<i64>,
idle_count: Option<i64>,
charging_count: Option<i64>,
full_count: Option<i64>,
fault_count: Option<i64>,
}
impl From<CabinetSqlRow> for CabinetRow {
fn from(row: CabinetSqlRow) -> Self {
CabinetRow {
id: row.id,
project_id: row.project_id,
abstract_id: row.abstract_id,
imei: row.imei,
iccid: row.iccid,
name: row.name,
address: row.address,
project_address: row.project_address,
status: row.status,
board_count: row.board_count,
channel_count: row.channel_count,
idle_count: row.idle_count,
charging_count: row.charging_count,
full_count: row.full_count,
fault_count: row.fault_count,
rssi: None,
pow_fail_dc: None,
pow_fail_ac: None,
// 柜级位域由 attach_realtime_status 从 Redis 补取(取不到保持 null)
status1: None,
status2: None,
is_online: None,
org_disabled: None,
}
}
}
// ---------------------------------------------------------------------------
// SQL 常量
//
// 说明:以下 SQL 均由原函数体**原样搬移**,未改动任何 SELECT 字段列表、
// 子查询、JOIN 条件与 WHERE 条件(含 CAST(... AS CHAR) 与 DBX 占位符 `?`)。
// ---------------------------------------------------------------------------
/// 全部组织(工厂/总管理员可见全量)
const ORGS_ALL_SQL: &str = "SELECT id, name, parent_id, disabled FROM organizations WHERE 1=1{org_cond} ORDER BY id";
/// 全部项目(工厂/总管理员可见全量;经纬度转 CHAR,规避 sqlx 的 DECIMAL 解码问题)
///
/// `{org_cond}` = 组织隔离片段(`p.organization_id = ?` / `1=0` / 空),
/// `{scope_cond}` = 数据范围片段(`p.id IN (...)` / 空),**两者都在 `WHERE 1=1` 之后追加**。
const PROJECTS_ALL_SQL: &str = "SELECT p.id, p.organization_id, p.name, p.address, \
CAST(p.latitude AS CHAR) AS latitude, CAST(p.longitude AS CHAR) AS longitude \
FROM projects p WHERE 1=1{org_cond}{scope_cond} ORDER BY p.id";
/// 各项目下的设备数量
const PROJECT_CABINET_COUNTS_SQL: &str =
"SELECT project_id, COUNT(*) as cnt FROM cabinets GROUP BY project_id";
/// 设备列表基础 SQL:联表查询获取通道状态统计(子查询 idle_count/charging_count/... 原样保留)
///
/// ⚠️ **口径说明(本次产品决定)**:`compartments.status` 只表达**占用**语义
/// (`0 空闲 / 1 充电中`,这正是平台实际写入的两个值),"充满/故障"不落库。
/// 因此:
/// - `idle_count` / `charging_count` 两条子查询(`comp.status = 0 / 1`)**保持原样**,语义正确;
/// - `full_count` / `fault_count` 两条子查询(`comp.status = 2 / 3`)在 DB 里**恒为 0**
/// (全仓没有任何写入点会写 2/3),本次采用「**SQL 保留、结果由派生值覆盖**」方案:
/// 这两列仍被 SELECT,但响应值一律由 [`apply_derived_counts`] 从通道实时状态派生后覆盖,
/// 详见 [`ChannelCounts`] 的注释。
const CABINET_FILTER_BASE_SQL: &str = "
SELECT
c.id, c.project_id, c.abstract_id, c.imei, c.iccid, c.name, c.address,
(SELECT address FROM projects WHERE id = c.project_id) as project_address,
c.status,
(SELECT COUNT(*) FROM cabin_boards cb WHERE cb.cabinet_id = c.id) as board_count,
(SELECT COUNT(*) FROM cabin_boards cb2
JOIN compartments comp ON comp.cabin_board_id = cb2.id
WHERE cb2.cabinet_id = c.id) as channel_count,
(SELECT COUNT(*) FROM cabin_boards cb3
JOIN compartments comp2 ON comp2.cabin_board_id = cb3.id
WHERE cb3.cabinet_id = c.id AND comp2.status = 0) as idle_count,
(SELECT COUNT(*) FROM cabin_boards cb4
JOIN compartments comp3 ON comp3.cabin_board_id = cb4.id
WHERE cb4.cabinet_id = c.id AND comp3.status = 1) as charging_count,
(SELECT COUNT(*) FROM cabin_boards cb5
JOIN compartments comp4 ON comp4.cabin_board_id = cb5.id
WHERE cb5.cabinet_id = c.id AND comp4.status = 2) as full_count,
(SELECT COUNT(*) FROM cabin_boards cb6
JOIN compartments comp5 ON comp5.cabin_board_id = cb6.id
WHERE cb6.cabinet_id = c.id AND comp5.status = 3) as fault_count
FROM cabinets c
";
/// 设备归属校验(非总管理员):设备必须属于当前用户组织 **且落在其数据范围内**
///
/// 数据范围未配置时该附加条件为空串 ⇒ 与今天的行为完全一致(向后兼容)。
const DEVICE_ACCESS_SQL: &str = "SELECT EXISTS(SELECT 1 FROM cabinets c \
JOIN projects p ON c.project_id = p.id \
WHERE c.imei = ? AND p.organization_id = ?{})";
// ---------------------------------------------------------------------------
// 组织树
// ---------------------------------------------------------------------------
/// 组织树接口 — 返回多级组织->项目树(含设备数量)
///
/// **组织隔离(与 `GET /api/my-organizations` 同一口径,见 AGENTS.md「层级与可见性」)**:
/// 工厂(`role_level >= 2`)→ 全量组织与全量项目;
/// 客户侧(组织管理员 `1` / 操作员 `0`)→ **只有自己所属组织及其项目**;
/// `users.organization_id` 为 `NULL` → 空集(**绝不退化成全量**)。
///
/// 历史缺陷:本接口原先直接 `SELECT ... FROM organizations` / `FROM projects` 全量查询,
/// 只校验了 `device:view` ⇒ 任何客户账号都能拿到全平台组织名 + 项目树(含地址/经纬度)。
///
/// 数据范围(`user_scopes`)在**项目**这一层再收窄(未配置 = 不限制,与今天一致)。
pub async fn organization_tree(
user: CurrentUser,
State(state): State<AppState>,
) -> Result<Json<Value>, AppError> {
auth::check_permission(&user, "device:view")?;
let db = &state.mysql;
// 可见组织(含隔离条件)与可见项目(组织隔离 ∧ 数据范围)
let (orgs_sql, orgs_binds) = visible_orgs_sql(&user);
let mut orgs_query = sqlx::query_as::<_, OrganizationRow>(&orgs_sql);
for bind in &orgs_binds {
orgs_query = orgs_query.bind(bind);
}
let orgs: Vec<OrganizationRow> = orgs_query.fetch_all(db).await?;
let (projects_sql, projects_binds) = visible_projects_sql(&user);
let mut projects_query = sqlx::query_as::<_, ProjectRow>(&projects_sql);
for bind in &projects_binds {
projects_query = projects_query.bind(bind);
}
let projects: Vec<ProjectRow> = projects_query.fetch_all(db).await?;
// 获取每个项目的设备数量(只统计柜子数,项目集合已被上面收窄)
let project_count_map = fetch_project_counts(db).await?;
// 递归构建树
let tree = build_tree(&orgs, &projects, &project_count_map, None);
Ok(Json(json!({ "code": 0, "data": tree, "message": "ok" })))
}
/// 可见组织的查询 SQL 与绑定值(纯函数,便于单测断言组织隔离)
///
/// 组织表自带组织 ID 列 ⇒ 直接按 `id` 过滤(客户侧 `id = users.organization_id`)。
fn visible_orgs_sql(user: &CurrentUser) -> (String, Vec<String>) {
let (org_cond, org_binds) = auth::org_condition_on_column(user, "id");
(ORGS_ALL_SQL.replace("{org_cond}", &org_cond), org_binds)
}
/// 可见项目的查询 SQL 与绑定值(纯函数,便于单测断言两道收窄)
///
/// 组织隔离(本组织)**之后**再叠加数据范围(项目白名单),占位符顺序 = 绑定值顺序。
fn visible_projects_sql(user: &CurrentUser) -> (String, Vec<String>) {
let (org_cond, mut binds) = auth::org_condition_on_column(user, "p.organization_id");
let (scope_cond, scope_binds) = user.effective_scope().project_condition("p.id");
binds.extend(scope_binds);
(
PROJECTS_ALL_SQL
.replace("{org_cond}", &org_cond)
.replace("{scope_cond}", &scope_cond),
binds,
)
}
/// 查询「项目ID -> 设备数量」映射
async fn fetch_project_counts(db: &sqlx::MySqlPool) -> Result<HashMap<i64, i64>, AppError> {
let project_counts: Vec<(i64, i64)> = sqlx::query_as(PROJECT_CABINET_COUNTS_SQL)
.fetch_all(db)
.await?;
Ok(project_counts.into_iter().collect())
}
/// 递归构建「组织 -> 项目」树节点,项目名称携带设备数量
fn build_tree(
orgs: &[OrganizationRow],
projects: &[ProjectRow],
project_count_map: &HashMap<i64, i64>,
parent_id: Option<i64>,
) -> Vec<TreeNode> {
orgs.iter()
.filter(|o| o.parent_id == parent_id)
.map(|org| {
// 递归获取子组织
let children = build_tree(orgs, projects, project_count_map, Some(org.id));
// 获取该组织下的项目
let org_projects: Vec<TreeNode> = projects
.iter()
.filter(|p| p.organization_id == Some(org.id))
.map(|p| {
let count = project_count_map.get(&p.id).copied().unwrap_or(0);
TreeNode {
id: p.id,
name: format!("{} ({})", p.name, count),
children: None,
abstract_id: None,
imei: None,
status: None,
disabled: None,
address: p.address.clone(),
latitude: p.latitude.clone(),
longitude: p.longitude.clone(),
}
})
.collect();
// 合并子组织和项目
let mut all_children = children;
all_children.extend(org_projects);
// 计算本组织下的总设备数量
let total_count: i64 = all_children
.iter()
.map(|c| {
// 从项目名称中提取数量
if let Some(start) = c.name.rfind('(') {
if let Some(end) = c.name.rfind(')') {
c.name[start + 1..end].parse().unwrap_or(0)
} else {
0
}
} else {
0
}
})
.sum();
TreeNode {
id: org.id,
name: format!("{} ({})", org.name, total_count),
children: if all_children.is_empty() {
None
} else {
Some(all_children)
},
abstract_id: None,
imei: None,
status: None,
disabled: org.disabled,
address: None,
latitude: None,
longitude: None,
}
})
.collect()
}
// ---------------------------------------------------------------------------
// 设备列表筛选
// ---------------------------------------------------------------------------
/// 设备列表筛选参数
#[derive(Deserialize)]
pub struct CabinetFilterParams {
pub organization_id: Option<i64>,
pub project_id: Option<i64>,
pub page: Option<i64>,
pub page_size: Option<i64>,
pub bound: Option<String>,
}
/// 设备列表接口 — 根据组织/项目筛选
pub async fn filter_cabinets(
user: CurrentUser,
State(state): State<AppState>,
Query(params): Query<CabinetFilterParams>,
) -> Result<Json<Value>, AppError> {
auth::check_permission(&user, "device:view")?;
let db = &state.mysql;
let page = params.page.unwrap_or(1).max(1);
let page_size = params.page_size.unwrap_or(12).clamp(1, 100);
let offset = (page - 1) * page_size;
// 组织/项目过滤 + 已绑定/未绑定过滤
let (where_clause, binds) = build_filter_where(&params, &user);
// 总数量
let total = fetch_filter_total(db, &where_clause, &binds).await;
if total == 0 || offset >= total {
return Ok(Json(json!({ "code": 0, "data": { "total": 0, "data": [] }, "message": "ok" })));
}
// 分页数据
let sql_rows = fetch_filter_page(db, &where_clause, &binds, page_size, offset).await?;
// 转换为CabinetRow
let mut rows: Vec<CabinetRow> = sql_rows.into_iter().map(CabinetRow::from).collect();
// 从Redis获取实时状态(Pipeline批量HGET,1次网络往返)
attach_realtime_status(&state, &mut rows).await?;
// 查询柜子所属组织是否禁用(甲方禁用客户组织 → 前端显示徽标)
attach_org_disabled(db, &mut rows).await;
Ok(Json(json!({ "code": 0, "data": { "total": total, "data": rows }, "message": "ok" })))
}
/// 构造设备列表的 WHERE 子句及其绑定参数
///
/// **三道边界,全部取交集(书写顺序 = 占位符顺序 = 绑定值顺序)**:
/// 1. **显式筛选**(入参 `project_id` / `organization_id`)—— 只表达"用户想看什么",
/// **不构成归属边界**。历史缺陷:这两个分支只按入参过滤,客户账号传别人的
/// `project_id` / `organization_id` 即可越权取数(**组织隔离层面的漏洞**,
/// 与数据范围 `user_scopes` 无关 —— 未配置 scope 的用户同样被利用);
/// 2. **组织隔离**([`auth::org_condition_for_project`])—— **无条件叠加**:
/// 总管理员豁免(全量),客户侧收窄到 `users.organization_id`,
/// 无组织则 `1=0`(空集,**绝不退化成全量**);
/// 3. **数据范围**(`user_scopes` 项目白名单,`effective_scope()`)—— 未配置 = 不限制。
///
/// 「已绑定/未绑定」(`bound`)只是附加筛选,插在组织隔离之后、数据范围之前,
/// 不影响上面三道边界。注意:**这里不再 `JOIN projects`** —— 组织/范围条件统一用
/// `c.project_id IN (SELECT ...)` 子查询表达,避免 JOIN 与 count/分页 SQL 的耦合。
fn build_filter_where(
params: &CabinetFilterParams,
user: &CurrentUser,
) -> (String, Vec<String>) {
let mut conditions: Vec<String> = Vec::new();
let mut binds: Vec<String> = Vec::new();
// ① 显式筛选:项目 / 组织(组织含其直接子组织,沿用原口径)
if let Some(proj_id) = params.project_id {
conditions.push("c.project_id = ?".to_string());
binds.push(proj_id.to_string());
}
if let Some(org_id) = params.organization_id {
conditions.push(
"c.project_id IN (SELECT p.id FROM projects p WHERE p.organization_id IN \
(SELECT o.id FROM organizations o WHERE o.id = ? OR o.parent_id = ?))"
.to_string(),
);
binds.push(org_id.to_string());
binds.push(org_id.to_string());
}
// ② 组织隔离:无论有没有入参都必须叠加(这一步才是"能不能看"的边界)
let (org_cond, org_binds) = auth::org_condition_for_project(user, "c.project_id");
push_condition(&mut conditions, &mut binds, &org_cond, org_binds);
// 附加:已绑定/未绑定过滤(绑定 = 分配到项目)
match params.bound.as_deref() {
Some("1") => conditions.push("c.project_id IS NOT NULL".to_string()),
Some("0") => conditions.push("c.project_id IS NULL".to_string()),
_ => {}
}
// ③ 数据范围(项目级)收窄:用 `effective_scope()`(而不是 `user.scope`)——
// **总管理员恒为"不限制"**,即便别处构造的 CurrentUser 带着非空 scope 也不会把工厂自己收窄。
let (scope_cond, scope_binds) = user.effective_scope().project_condition("c.project_id");
push_condition(&mut conditions, &mut binds, &scope_cond, scope_binds);
let where_clause = if conditions.is_empty() {
String::new()
} else {
format!("WHERE {}", conditions.join(" AND "))
};
(where_clause, binds)
}
/// 追加 ` AND xxx` 形式的条件片段及其绑定值(片段为空 = 不限制,原样跳过)
///
/// **绑定值必须与片段一起追加**:占位符在 SQL 文本里的先后顺序即绑定顺序,
/// 分开追加必然错位(错位的结果是条件张冠李戴,比不加条件更危险)。
fn push_condition(
conditions: &mut Vec<String>,
binds: &mut Vec<String>,
fragment: &str,
fragment_binds: Vec<String>,
) {
if fragment.is_empty() {
return;
}
conditions.push(fragment.trim_start_matches(" AND ").to_string());
binds.extend(fragment_binds);
}
/// 统计符合条件的设备总数(查询失败按 0 处理,与原实现一致)
async fn fetch_filter_total(db: &sqlx::MySqlPool, where_clause: &str, binds: &[String]) -> i64 {
let count_sql = format!("SELECT COUNT(*) FROM cabinets c {}", where_clause);
let mut count_query = sqlx::query_scalar::<_, i64>(&count_sql);
for b in binds {
count_query = count_query.bind(b);
}
count_query.fetch_one(db).await.unwrap_or(0)
}
/// 查询当前页设备行(分页参数由 WHERE 之后的 LIMIT ? OFFSET ? 绑定)
async fn fetch_filter_page(
db: &sqlx::MySqlPool,
where_clause: &str,
binds: &[String],
page_size: i64,
offset: i64,
) -> Result<Vec<CabinetSqlRow>, AppError> {
let data_sql = format!("{} {} ORDER BY c.id LIMIT ? OFFSET ?", CABINET_FILTER_BASE_SQL, where_clause);
let mut data_query = sqlx::query_as::<_, CabinetSqlRow>(&data_sql);
for b in binds {
data_query = data_query.bind(b);
}
let rows: Vec<CabinetSqlRow> = data_query
.bind(page_size)
.bind(offset)
.fetch_all(db)
.await?;
Ok(rows)
}
// ---------------------------------------------------------------------------
// 设备在线判据 —— **全仓唯一权威实现**(纯函数 + 批量计数)
// ---------------------------------------------------------------------------
/// 设备级实时状态的 Redis key 前缀:`device:{imei}`(Hash,TTL 300 秒)。
const DEVICE_KEY_PREFIX: &str = "device:";
/// 设备级在线标记在 `device:{imei}` 里的字段名(`"1"` = 在线)。
///
/// **写入侧**:`workers/report_worker.rs::write_device_state`(`login` / `status_post`),
/// 每条写 `device:{imei}` 的路径都会跟一条 `EXPIRE 300`
/// ⇒ 「300 秒内无任意上行」即整键过期 ⇒ 判离线。
pub(crate) const DEVICE_ONLINE_FIELD: &str = "online";
/// 在线标记取值(协议约定 `"1"` = 在线;其余一律不算在线)。
const DEVICE_ONLINE_VALUE: &str = "1";
/// `device:{imei}` 的 key(通道级 key 见 [`channel_key`])。
pub(crate) fn device_key(imei: &str) -> String {
format!("{}{}", DEVICE_KEY_PREFIX, imei)
}
/// **设备在线判据的唯一权威纯函数**(全仓共用,禁止各处自写一套)。
///
/// # 为什么是它,而不是 `cabinets.status`
/// `cabinets.status` 是**启用/禁用配置态**(`2` = 禁用,其余 = 启用/false 语义),
/// 全仓**没有任何写入方会把它写成 0(离线)**(`routes/cabinets.rs::set_cabinet_status`
/// 只在 `2` / 其它值之间切)⇒ 拿它当"在线"会让管理端在**一台设备都没连**时
/// 依然恒定显示"全部在线"。实时在线一律以 Redis 为准(`AGENTS.md`「状态位展示约定」)。
///
/// # 三态("无数据"与"离线"必须区分)
/// - `Some("1")` → `Some(true)`:在线;
/// - `Some(其它)`(含 `"0"`)→ `Some(false)`:离线;
/// - `None`(字段缺失 / key 不存在 / Redis 不可用)→ `None`:**无数据**。
///
/// 计数场景请用 [`is_device_online`](无数据按"不在线"计,**绝不猜测在线**)。
pub(crate) fn device_online_flag(raw: Option<&str>) -> Option<bool> {
raw.map(|v| v == DEVICE_ONLINE_VALUE)
}
/// 在线判据的**布尔形态**(供计数使用):只有明确的 `online = "1"` 才算在线。
///
/// 与 [`device_online_flag`] **同源** —— 内部直接调用它,保证
/// 「设备列表 `is_online`」「能耗报表在线柜数」「H5 在线数/详情在线」四处
/// 永远同一口径、不可能分叉。
pub(crate) fn is_device_online(raw: Option<&str>) -> bool {
device_online_flag(raw) == Some(true)
}
/// **纯函数形态的在线计数**:把一批 `device:{imei}.online` 原始值折算成在线台数。
///
/// 抽成纯函数的目的:让"**0 台设备连接 ⇒ 在线数 0**"这条语义可以在单测里覆盖,
/// 且不需要连真实 Redis。所有在线计数(能耗总览 / H5 首页)最终都走它。
pub(crate) fn count_online_flags(raws: &[Option<String>]) -> i64 {
raws.iter()
.filter(|v| is_device_online(v.as_deref()))
.count() as i64
}
/// 批量统计一批 IMEI 里**明确在线**的台数(**一次 pipeline,零 N 次单查**)。
///
/// 判据与设备列表 `is_online` 完全同源([`is_device_online`] → [`count_online_flags`])。
/// Redis 连不上 / pipeline 失败 → 记 `ERROR` 并按 0 计(等价于"全部无数据",
/// **绝不猜测在线**,与 `derive_counts_by_cabinet` 的降级口径一致)。
pub(crate) async fn count_online_devices(state: &AppState, imeis: &[String]) -> i64 {
if imeis.is_empty() {
return 0;
}
let mut conn = match open_redis_conn(state).await {
Ok(c) => c,
Err(e) => {
tracing::error!("[online] Redis 连接失败,在线数按 0 计 err={}", e);
return 0;
}
};
// 每设备 1 次 `HGET device:{imei} online`,全部走一次 pipeline(1 次网络往返)
let mut pipe = redis::pipe();
for imei in imeis {
pipe.cmd("HGET")
.arg(device_key(imei))
.arg(DEVICE_ONLINE_FIELD);
}
let queried: Result<Vec<Option<String>>, redis::RedisError> = pipe.query_async(&mut conn).await;
match queried {
Ok(vals) => count_online_flags(&vals),
Err(e) => {
tracing::error!("[online] 在线状态 pipeline 读取失败,在线数按 0 计 err={}", e);
0
}
}
}
/// 设备列表 pipeline 中**每个设备**下发的 `HGET` 字段个数。
///
/// ⚠️ **命令顺序与下标一一对应**:`online`(0) / `rssi`(1) / `pow_fail_dc`(2) /
/// `pow_fail_ac`(3) / `cabinet_status1`(4) / `cabinet_status2`(5)。
/// 该常量是 [`attach_realtime_status`] 里 `base = i * DEVICE_LIST_FIELDS` 的唯一依据,
/// **新增字段必须同步改这里、命令顺序与解析下标**(只能追加在末尾)。
const DEVICE_LIST_FIELDS: usize = 6;
/// 柜级 hex 位域取值:`None` / 空串 / 纯空白都算"无数据"(→ 响应 `null`),
/// **绝不落成 `"00"`**(`"00"` 是"全部正常"的有效值,与"没数据"必须区分)。
/// 非空值**原样返回**(不解析、不改写大小写 —— 前端两端有统一的 hex 解析纯函数)。
fn pick_non_empty_hex(raw: Option<&Option<String>>) -> Option<String> {
raw.and_then(|v| v.as_ref())
.map(|v| v.trim())
.filter(|v| !v.is_empty())
.map(str::to_string)
}
/// 从 Redis 批量补齐实时状态(Pipeline HGET,1 次网络往返)
async fn attach_realtime_status(state: &AppState, rows: &mut [CabinetRow]) -> Result<(), AppError> {
// 使用多路复用连接(MultiplexedConnection),替代已废弃的 aio::Connection
let mut conn = open_redis_conn(state).await?;
// 用Pipeline一次发所有HGET命令(每个设备1次HGET取6个字段)
// ⚠️ 命令顺序与下面解析的下标一一对应:online/rssi/pow_fail_dc/pow_fail_ac/cabinet_status1/cabinet_status2
let mut pipe = redis::pipe();
for row in rows.iter() {
let key = device_key(&row.imei);
pipe.cmd("HGET").arg(&key).arg(DEVICE_ONLINE_FIELD);
pipe.cmd("HGET").arg(&key).arg("rssi");
pipe.cmd("HGET").arg(&key).arg("pow_fail_dc");
pipe.cmd("HGET").arg(&key).arg("pow_fail_ac");
// 柜级状态位(协议 §3.8):原样透出 hex,取不到 → null(不落 "00")
pipe.cmd("HGET").arg(&key).arg("cabinet_status1");
pipe.cmd("HGET").arg(&key).arg("cabinet_status2");
}
// 一次发送,获取所有结果
let results: Vec<Option<String>> = pipe.query_async(&mut conn).await.unwrap_or_default();
// 每6个一组解析
for (i, row) in rows.iter_mut().enumerate() {
let base = i * DEVICE_LIST_FIELDS;
// 在线判据**唯一权威纯函数**(与能耗在线柜数、H5 在线数同源;见 [`device_online_flag`])
row.is_online =
device_online_flag(results.get(base).and_then(|v| v.as_ref()).map(String::as_str));
row.rssi = results
.get(base + 1)
.and_then(|v| v.as_ref())
.and_then(|v| v.parse::<i32>().ok());
row.pow_fail_dc = results
.get(base + 2)
.and_then(|v| v.as_ref())
.map(|v| v == "1");
row.pow_fail_ac = results
.get(base + 3)
.and_then(|v| v.as_ref())
.map(|v| v == "1");
row.status1 = pick_non_empty_hex(results.get(base + 4));
row.status2 = pick_non_empty_hex(results.get(base + 5));
}
// 通道级:按柜读取通道状态 → 派生「充满数 / 故障数」并覆盖 SQL 的返回值
apply_derived_counts(state, rows).await;
Ok(())
}
/// 按柜派生「充满数 / 故障数」并**覆盖** SQL 的返回值
///
/// - 依据:通道实时状态 `batt_status` / `pow_status` → [`derive_channel_flags`],
/// 与设备实时详情接口**同一套唯一权威推导规则**(优先级 fault > full > charging > idle);
/// - **拿不到 Redis 数据时(设备离线 / 该通道无键 / Redis 不可用)无法判定,按 0 计**,
/// 绝不猜测、也不回落到 SQL 的占位值(那两个值在 DB 里恒为 0,回落没有意义);
/// - `idle_count` / `charging_count` **不动**:它们是 DB 的占用语义(0/1),本来就是对的。
async fn apply_derived_counts(state: &AppState, rows: &mut [CabinetRow]) {
let cabinet_ids: Vec<i64> = rows.iter().map(|r| r.id).collect();
let entries = match fetch_channel_keys(&state.mysql, &cabinet_ids).await {
Ok(entries) => entries,
Err(e) => {
tracing::error!(
"[cabinets] 通道 key 查询失败,充满/故障计数按 0 计(离线口径)err={}",
e
);
Vec::new()
}
};
let counts = derive_counts_by_cabinet(state, &entries).await;
for row in rows.iter_mut() {
// `ChannelCounts = (idle, charging, full, fault)`:只取后两位覆盖响应
let (_, _, full, fault) = counts.get(&row.id).copied().unwrap_or((0, 0, 0, 0));
row.full_count = Some(full);
row.fault_count = Some(fault);
}
}
/// 通道级 Redis key 查询行:`(柜子ID, IMEI, 仓控板号, 通道号)`
type ChannelKeyRow = (i64, String, i8, i8);
/// 通道状态 pipeline 每次 HMGET 的字段(下标顺序即 [`derive_channel_flags`] 的入参顺序)
const CHANNEL_FLAG_FIELDS: [&str; 2] = ["batt_status", "pow_status"];
/// 查询这批柜子的通道级 Redis key,返回 `(柜子ID, device:{imei}:board:{板号}:ch:{通道号})`
///
/// 为什么按 DB 的「板号 + 通道号」直接拼 key,而不是 `KEYS device:{imei}:board:*:ch:*`:
/// `KEYS` 是**全库扫描**,列表页 12~100 个柜就是 12~100 次全库扫描(Redis 单线程会被打满);
/// 而通道键的构成与设备上报写入侧完全一致(`workers/report_worker.rs::channel_status_key`:
/// 板号/通道号十进制不补零),按 DB 拼 key 精确且零扫描。
pub(crate) async fn fetch_channel_keys(
db: &sqlx::MySqlPool,
cabinet_ids: &[i64],
) -> Result<Vec<(i64, String)>, AppError> {
if cabinet_ids.is_empty() {
return Ok(Vec::new());
}
let placeholders = vec!["?"; cabinet_ids.len()].join(",");
let sql = format!(
r#"SELECT c.id, c.imei, b.board_id, comp.channel_id
FROM cabinets c
JOIN cabin_boards b ON b.cabinet_id = c.id
JOIN compartments comp ON comp.cabin_board_id = b.id
WHERE c.id IN ({})
ORDER BY c.id, b.board_id, comp.channel_id"#,
placeholders
);
let mut query = sqlx::query_as::<_, ChannelKeyRow>(&sql);
for id in cabinet_ids {
query = query.bind(id);
}
let rows: Vec<ChannelKeyRow> = query.fetch_all(db).await?;
Ok(rows
.into_iter()
// IMEI 为空的柜子拼不出有效通道 key(`device::board:1:ch:1`),直接跳过
.filter(|(_, imei, _, _)| !imei.is_empty())
.map(|(cabinet_id, imei, board_id, channel_id)| {
(
cabinet_id,
format!("device:{}:board:{}:ch:{}", imei, board_id, channel_id),
)
})
.collect())
}
/// 批量读取通道状态并按柜聚合出四计数,返回 `柜子ID → ChannelCounts`
///
/// **离线策略(产品口径)**:通道读不到数据(键不存在 / 字段缺失)时该通道按派生 idle 记账,
/// 因此设备离线、通道从未上报时 `full` / `fault` 恒为 0 —— **离线无法判定,不猜测**;
/// 表里没有的柜子同样按全 0 处理(调用方 `unwrap_or_default`)。
///
/// ⚠️ 通道级 Hash 由上报链路写入后**没有 TTL**,设备离线后仍会残留最后一次上报值,
/// 所以本口径是「最后一次上报的通道状态」而非严格「在线实时」;若要收紧为仅在线设备,
/// 需额外用 `device:{imei}`(TTL 300s)是否存在做闸门(本次未做,见报告待决策项)。
///
/// Redis 不可用 / pipeline 失败只记 ERROR 并返回空表(等价于全部无数据 → 充满/故障记 0),
/// 绝不因此让列表接口整体失败。
pub(crate) async fn derive_counts_by_cabinet(
state: &AppState,
entries: &[(i64, String)],
) -> HashMap<i64, ChannelCounts> {
if entries.is_empty() {
return HashMap::new();
}
let mut conn = match open_redis_conn(state).await {
Ok(conn) => conn,
Err(e) => {
tracing::error!("[cabinets] Redis 连接失败,充满/故障计数按 0 计 err={}", e);
return HashMap::new();
}
};
// 每个通道 1 次 HMGET(2 个字段),全部走一次 pipeline
let mut pipe = redis::pipe();
for (_, key) in entries {
pipe.cmd("HMGET").arg(key).arg(&CHANNEL_FLAG_FIELDS);
}
let results: Vec<Vec<Option<String>>> = match pipe.query_async(&mut conn).await {
Ok(results) => results,
Err(e) => {
tracing::error!("[cabinets] 通道状态读取失败,充满/故障计数按 0 计 err={}", e);
return HashMap::new();
}
};
let mut per_cabinet: HashMap<i64, Vec<ChannelFlags>> = HashMap::new();
for (idx, (cabinet_id, _)) in entries.iter().enumerate() {
let field = |i: usize| -> Option<&str> {
results
.get(idx)
.and_then(|vals| vals.get(i))
.and_then(|v| v.as_deref())
};
per_cabinet
.entry(*cabinet_id)
.or_default()
.push(derive_channel_flags(field(0), field(1)));
}
per_cabinet
.into_iter()
.map(|(cabinet_id, flags)| (cabinet_id, derive_channel_counts(&flags)))
.collect()
}
/// 标记柜子所属组织是否被禁用(甲方禁用客户组织 → 前端显示徽标)
async fn attach_org_disabled(db: &sqlx::MySqlPool, rows: &mut [CabinetRow]) {
let project_ids: Vec<i64> = rows.iter().filter_map(|r| r.project_id).collect();
if project_ids.is_empty() {
return;
}
let placeholders: Vec<String> = project_ids.iter().map(|_| "?".to_string()).collect();
let sql = format!(
r#"SELECT p.id, o.disabled FROM projects p
LEFT JOIN organizations o ON p.organization_id = o.id
WHERE p.id IN ({})"#,
placeholders.join(",")
);
let mut q = sqlx::query_as::<_, (i64, i8)>(&sql);
for pid in &project_ids {
q = q.bind(pid);
}
if let Ok(results) = q.fetch_all(db).await {
let disabled_map: HashMap<i64, bool> =
results.into_iter().map(|(pid, d)| (pid, d == 1)).collect();
for row in rows.iter_mut() {
row.org_disabled = row.project_id.and_then(|pid| disabled_map.get(&pid)).copied();
}
}
}
// ---------------------------------------------------------------------------
// 设备实时状态
// ---------------------------------------------------------------------------
/// 设备实时状态参数
#[derive(Deserialize)]
pub struct DeviceStatusParams {
pub imei: String,
}
/// 打开 Redis 多路复用连接(失败按原错误信息返回内部错误)
async fn open_redis_conn(state: &AppState) -> Result<redis::aio::MultiplexedConnection, AppError> {
let client = redis::Client::open(state.redis_url.as_str()).map_err(|e| {
AppError::Internal(format!("Redis客户端创建失败: {}", e))
})?;
// 使用多路复用连接(MultiplexedConnection),替代已废弃的 aio::Connection
client.get_multiplexed_async_connection().await.map_err(|e| {
AppError::Internal(format!("Redis连接失败: {}", e))
})
}
/// 设备实时状态接口 — 返回板级+通道级状态
pub async fn device_realtime_status(
user: CurrentUser,
State(state): State<AppState>,
Query(params): Query<DeviceStatusParams>,
) -> Result<Json<Value>, AppError> {
auth::check_permission(&user, "device:view")?;
// 校验设备归属
ensure_device_access(&state, &user, &params.imei).await?;
let mut conn = open_redis_conn(&state).await?;
let device_status = load_device_status(&mut conn, &params.imei).await;
// 板级+通道级:按 DB 拓扑拼 key + 一次 pipeline(**零 KEYS 扫描**,见函数注释)
let boards = load_boards_status(&state.mysql, &mut conn, &params.imei).await;
Ok(Json(json!({
"code": 0,
"data": {
"device": device_status,
"boards": boards,
},
"message": "ok"
})))
}
/// 设备归属校验:非总管理员必须能通过「设备 → 项目 → 组织」链路命中本组织,
/// 且(配了数据范围时)设备所属项目必须落在范围内
async fn ensure_device_access(
state: &AppState,
user: &CurrentUser,
imei: &str,
) -> Result<(), AppError> {
if user.role_level < 2 {
if let Some(org_id) = user.organization_id {
let (scope_cond, scope_binds) = user.effective_scope().project_condition("p.id");
let sql = DEVICE_ACCESS_SQL.replacen("{}", &scope_cond, 1);
let mut query = sqlx::query_scalar::<_, bool>(&sql).bind(imei).bind(org_id);
for bind in &scope_binds {
query = query.bind(bind);
}
let has_access: bool = query.fetch_one(&state.mysql).await.unwrap_or(false);
if !has_access {
return Err(AppError::Forbidden("无权访问该设备".into()));
}
} else {
return Err(AppError::Forbidden("无权访问该设备".into()));
}
}
Ok(())
}
/// 读取设备级字段(online / rssi / pow_fail_dc / pow_fail_ac)
async fn load_device_status(
conn: &mut redis::aio::MultiplexedConnection,
imei: &str,
) -> Value {
let device_key = format!("device:{}", imei);
// 获取设备级字段
let fields = vec!["online", "rssi", "pow_fail_dc", "pow_fail_ac"];
let values: Vec<Option<String>> = redis::cmd("HMGET")
.arg(&device_key)
.arg(&fields)
.query_async(&mut *conn)
.await
.unwrap_or_default();
json!({
"online": values.get(0).and_then(|v| v.as_ref()).map(|v| v == "1").unwrap_or(false),
"rssi": values.get(1).and_then(|v| v.as_ref()).and_then(|v| v.parse::<i32>().ok()),
"pow_fail_dc": values.get(2).and_then(|v| v.as_ref()).map(|v| v == "1").unwrap_or(false),
"pow_fail_ac": values.get(3).and_then(|v| v.as_ref()).map(|v| v == "1").unwrap_or(false),
})
}
/// 仓控板号量程:协议 §3.7「仓控板 1–15」
const BOARD_ID_MIN: i64 = 1;
const BOARD_ID_MAX: i64 = 15;
/// 通道号量程:协议 §3.7「通道 1–8」(不是 0–7)
const CHANNEL_ID_MIN: i64 = 1;
const CHANNEL_ID_MAX: i64 = 8;
/// 通道级 Hash 的**权威字段名**(与 `workers/report_worker.rs::write_channel_state` 一致)
const CHANNEL_STATUS_FIELDS: [&str; 6] =
["batt_status", "pow_status", "soc", "voltage", "current", "temp"];
/// 通道级 Redis key:`device:{imei}:board:{板号}:ch:{通道号}`(板号/通道号**十进制不补零**)。
///
/// ⚠️ 必须与上报写入侧(`workers/report_worker.rs::channel_status_key`)**逐字一致**,
/// 否则读不到设备写入的状态。
fn channel_redis_key(imei: &str, board_id: i64, channel_id: i64) -> String {
format!("device:{}:board:{}:ch:{}", imei, board_id, channel_id)
}
/// 该通道级 key 是否落在协议量程内(板号 1–15 / 通道号 1–8)。
///
/// 为什么要有这道闸门:本接口现在**按 DB 拓扑拼 key**,一旦 `compartments` / `cabin_boards`
/// 里存在越界脏数据(`board_id = 0` / `channel_id = 9`),拼出来的就是
/// `...:board:0:ch:9` 这类**永远不会被上报写入的垃圾 key**;挡掉它们同时也拦住了
/// `...:ch:{通道号}:bat`(电池台账)/ `...:ch:{通道号}:order`(电量累加器)内部键。
fn is_valid_channel_key_parts(board_id: i64, channel_id: i64) -> bool {
(BOARD_ID_MIN..=BOARD_ID_MAX).contains(&board_id)
&& (CHANNEL_ID_MIN..=CHANNEL_ID_MAX).contains(&channel_id)
}
/// 一条通道的原始字段快照;`None` = 该通道 key 在 Redis 里**不存在**。
type ChannelSnapshot = Option<HashMap<String, String>>;
/// 纯函数:把快照里的真实字段翻译成响应 JSON(字段名与旧实现完全一致)。
fn channel_view_from_snapshot(snapshot: &HashMap<String, String>) -> Value {
let field = |name: &str| -> Option<&str> { snapshot.get(name).map(String::as_str) };
let number = |name: &str| -> Option<i64> { field(name).and_then(|v| v.parse::<i64>().ok()) };
derive_channel_view(
field(CHANNEL_STATUS_FIELDS[0]),
field(CHANNEL_STATUS_FIELDS[1]),
number(CHANNEL_STATUS_FIELDS[2]),
number(CHANNEL_STATUS_FIELDS[3]),
number(CHANNEL_STATUS_FIELDS[4]),
number(CHANNEL_STATUS_FIELDS[5]),
)
}
/// 纯函数:把(DB 拓扑行 + 对应通道快照)组装成 `boards` 响应 JSON。
///
/// `snapshots` 与 `topology` **下标一一对应**:`None` 表示该通道在 Redis 里没有键
/// (设备从未上报 / 键已过期),此时**整条通道省略** —— 这与旧实现
/// `KEYS device:{imei}:board:*:ch:*`「只返回实际存在的键」语义一致,
/// 不会凭空多出"全空通道"。
///
/// 板号用 `BTreeMap` 保证升序输出(响应顺序稳定);`status_hex` 由各通道的
/// `batt_status + pow_status`(各 2 位 hex)按**通道号升序**拼回(板级 `status_hex`
/// 在新协议里不存在,该字段名保持不变,语义完全由真实字段解释)。
fn assemble_boards(topology: &[(i8, i8)], snapshots: &[ChannelSnapshot]) -> Value {
// 板号 → [(通道号, 快照)],仅收录 Redis 里真实存在的通道
let mut grouped: BTreeMap<i64, Vec<(i64, &HashMap<String, String>)>> = BTreeMap::new();
for ((board_id, channel_id), snapshot) in topology.iter().zip(snapshots.iter()) {
if let Some(snapshot) = snapshot {
grouped
.entry(i64::from(*board_id))
.or_default()
.push((i64::from(*channel_id), snapshot));
}
}
let mut boards = json!({});
for (board_no, entries) in grouped {
let mut channels = json!({});
let mut status_hex = String::new();
for (channel_no, snapshot) in entries {
let channel = channel_view_from_snapshot(snapshot);
if let Some(hex) = channel.get("hex").and_then(|v| v.as_str()) {
status_hex.push_str(hex);
}
channels[channel_no.to_string().as_str()] = channel;
}
boards[board_no.to_string().as_str()] = json!({
"status_hex": status_hex,
"channels": channels,
});
}
boards
}
/// 读取该柜的通道拓扑 `(板号, 通道号)`,按(板号, 通道号)升序。
///
/// 为什么用 DB 而不是扫 Redis:这是本接口**零扫描**的基础 —— 通道级 key 的构成
/// 完全由 DB 拓扑决定(板号/通道号十进制不补零),拼出来与设备上报写入的 key 逐字相同。
async fn fetch_channel_topology(
db: &sqlx::MySqlPool,
imei: &str,
) -> Result<Vec<(i8, i8)>, AppError> {
let rows: Vec<(i8, i8)> = sqlx::query_as(
"SELECT b.board_id, comp.channel_id
FROM cabinets c
JOIN cabin_boards b ON b.cabinet_id = c.id
JOIN compartments comp ON comp.cabin_board_id = b.id
WHERE c.imei = ?
ORDER BY b.board_id, comp.channel_id",
)
.bind(imei)
.fetch_all(db)
.await?;
Ok(rows)
}
/// 组装全部仓控板状态(板级 + 通道级)。
///
/// ⚠️ 设备上报链路**只写通道级 key**(`device:{imei}:board:{板号}:ch:{通道号}`,
/// 见 `workers/report_worker.rs::write_channel_state`),**新协议里根本没有板级 hash**
/// (老实现的 `device:{imei}:board:{板号}` + `status_hex` 已不复存在;生产 Redis 里若还能
/// 搜到这类键,那是**旧版本残留**,不是当前写入契约)。
///
/// # 为什么不再用 `KEYS`(2026-09 修复)
/// 旧实现是 `KEYS device:{imei}:board:*:ch:*`。`KEYS` 会**阻塞 Redis 单线程**地遍历
/// 整个 keyspace,而本项目的 Redis 实例是**与既有充电站生产平台(db1–4)共用的**:
/// 单柜详情一次 `KEYS` 就足以在规模化后拖慢/拖死别人的业务。
/// 现在改为:**按 DB 拓扑拼出候选 key(零扫描)→ 一次 pipeline `HGETALL`**。
/// `HGETALL` 返回空 map ⇒ 该 key 不存在 ⇒ 按"无数据"略过,
/// 因此**语义与 `KEYS` 完全一致**(只少了对"DB 里没有对应行的孤儿 key"的收录 ——
/// 那正是旧代码注释里要避免的"幽灵板/幽灵通道")。
///
/// Redis 不可用 / pipeline 失败只记 ERROR 并返回空对象(等价于全部无数据),
/// 绝不因此让详情接口整体失败。
async fn load_boards_status(
db: &sqlx::MySqlPool,
conn: &mut redis::aio::MultiplexedConnection,
imei: &str,
) -> Value {
let topology_rows = match fetch_channel_topology(db, imei).await {
Ok(rows) => rows,
Err(e) => {
tracing::error!(
"[realtime_status] 查询通道拓扑失败,板级状态降级为空 imei={} err={}",
imei,
e
);
return json!({});
}
};
// 量程闸门:越界脏数据(board_id=0 / channel_id=9)拼不出有效 key,直接跳过并记 WARN
let topology: Vec<(i8, i8)> = topology_rows
.into_iter()
.filter(|(board_id, channel_id)| {
let ok = is_valid_channel_key_parts(i64::from(*board_id), i64::from(*channel_id));
if !ok {
tracing::warn!(
"[realtime_status] 拓扑越界已跳过(协议 §3.7 允许板 1–15 / 通道 1–8)\
imei={} board={} ch={}",
imei,
board_id,
channel_id
);
}
ok
})
.collect();
if topology.is_empty() {
// 拓扑未配置(或全是越界脏数据):没有候选 key,直接返回空对象
return json!({});
}
// 一次 pipeline 拿全部通道(每通道 1 次 HGETALL),1 次网络往返
let mut pipe = redis::pipe();
for (board_id, channel_id) in &topology {
let key = channel_redis_key(imei, i64::from(*board_id), i64::from(*channel_id));
pipe.cmd("HGETALL").arg(&key);
}
let queried: Result<Vec<HashMap<String, String>>, redis::RedisError> =
pipe.query_async(&mut *conn).await;
let snapshots: Vec<ChannelSnapshot> = match queried {
Ok(values) => values
.into_iter()
// Redis 里不保留空 Hash ⇒ 空 map 即"键不存在"(与 KEYS 只返回存在键一致)
.map(|fields| if fields.is_empty() { None } else { Some(fields) })
.collect(),
Err(e) => {
tracing::error!(
"[realtime_status] 通道状态 pipeline 读取失败,板级状态降级为空 imei={} err={}",
imei,
e
);
return json!({});
}
};
assemble_boards(&topology, &snapshots)
}
/// 电池状态位(协议 §3.8 `batt_status`):bit0 充满 / bit1 正在充电 /
/// bit4 过欠压 / bit5 过流 / bit6 热失控 / bit7 有电池
const BATT_FULL_MASK: u8 = 0x01;
const BATT_CHARGING_MASK: u8 = 0x02;
const BATT_FAULT_MASK: u8 = 0x10 | 0x20 | 0x40;
/// 电源状态位(协议 §3.8 `pow_status`):bit0 电源故障
const POW_FAULT_MASK: u8 = 0x01;
/// 2 位(兼容 1 位)hex → u8;字段缺失 / 长度超 2 / 非 hex → `None`
/// (与 `web/src/utils/deviceStatus.ts::parseHexByte` 口径一致)
fn parse_hex_byte(raw: Option<&str>) -> Option<u8> {
let text = raw?.trim();
if text.is_empty() || text.len() > 2 {
return None;
}
u8::from_str_radix(text, 16).ok()
}
/// 通道派生结果(协议 §3.8):由 `batt_status` / `pow_status` 两个字节推导出的三态布尔
///
/// 这是**全仓唯一权威推导规则**:设备实时详情([`derive_channel_view`])、
/// 设备列表的充满/故障计数([`derive_channel_counts`])以及仓体展示状态
/// (`routes::cabinets::get_cabinet_detail` / `routes::h5::get_compartment_detail`)
/// **全部复用它,任何地方都不允许另写一套位判断**。
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct ChannelFlags {
/// `batt_status` bit1(0x02 正在充电)
pub on: bool,
/// `batt_status` bit0(0x01 充满)
pub full: bool,
/// `batt_status` bit4/5/6(过欠压 / 过流 / 热失控)**或** `pow_status` bit0(电源故障)
pub fault: bool,
/// Redis 里是否存在**可判定**的通道数据(`batt_status` / `pow_status` 至少一个合法)
///
/// 为 `false` 表示设备离线或该通道从未上报 —— 此时无法判定充满/故障,
/// 调用方必须按「无数据」处理(计数按 0 计 / 展示回落到 DB 占用值),**不得猜测**。
pub has_data: bool,
}
impl ChannelFlags {
/// 展示状态名:fault > full > charging > idle(唯一优先级顺序,与旧实现一致)
pub(crate) fn status_name(&self) -> &'static str {
if self.fault {
"fault"
} else if self.full {
"full"
} else if self.on {
"charging"
} else {
"idle"
}
}
/// 前端状态码:`0 空闲 / 1 充电中 / 2 充满 / 3 故障`
///
/// 与 `web/src/pages/devices/cabinet.tsx::STATUS_MAP`、
/// `h5-uniapp/src/pages/device-detail/index.vue::compClass`、
/// `h5-uniapp/src/pages/compartment-detail/index.vue::statusText` **四处口径必须一致**
/// (**没有"禁用"这一档**:仓体级禁用不存在)。
pub(crate) fn status_code(&self) -> i8 {
match self.status_name() {
"fault" => 3,
"full" => 2,
"charging" => 1,
_ => 0,
}
}
}
/// 纯函数:由通道真实字段推导 [`ChannelFlags`](协议 §3.8,位定义与
/// `software/web/src/utils/deviceStatus.ts` 完全一致)
pub(crate) fn derive_channel_flags(batt_raw: Option<&str>, pow_raw: Option<&str>) -> ChannelFlags {
let batt = parse_hex_byte(batt_raw);
let pow = parse_hex_byte(pow_raw);
ChannelFlags {
full: batt.is_some_and(|b| b & BATT_FULL_MASK != 0),
on: batt.is_some_and(|b| b & BATT_CHARGING_MASK != 0),
fault: batt.is_some_and(|b| b & BATT_FAULT_MASK != 0)
|| pow.is_some_and(|p| p & POW_FAULT_MASK != 0),
has_data: batt.is_some() || pow.is_some(),
}
}
/// 通道四计数(派生口径),顺序固定:`(idle, charging, full, fault)`
///
/// ⚠️ 生产中只有 `.2`(充满)/ `.3`(故障)会被写回响应:
/// `idle_count` / `charging_count` 一律用 DB 的**占用**值(`compartments.status` 的 0/1,
/// 那才是这两列的语义),派生出来的 idle/charging 分区**不覆盖**它们。
/// 保留这两个分区是为了让「全空闲 / 充电中 / 充满 / 故障 / 优先级冲突 / 离线无数据」
/// 这些边界能在同一个可测纯函数里被完整覆盖(见本文件单测)。
pub(crate) type ChannelCounts = (i64, i64, i64, i64);
/// 纯函数:按 [`ChannelFlags`] 的优先级(fault > full > charging > idle)把通道集合分区计数
///
/// 无数据(`has_data == false`)的通道按派生 idle 记账 —— 因此「离线 / 无该通道键」时
/// 充满数与故障数恒为 0(**离线无法判定,按 0 计**)。
pub(crate) fn derive_channel_counts(channels: &[ChannelFlags]) -> ChannelCounts {
let mut counts: ChannelCounts = (0, 0, 0, 0);
for flags in channels {
match flags.status_code() {
3 => counts.3 += 1,
2 => counts.2 += 1,
1 => counts.1 += 1,
_ => counts.0 += 1,
}
}
counts
}
/// 纯函数:把通道的**真实字段**组装成响应 JSON。
///
/// **响应字段名与旧实现完全一致**(前端在消费),但取值改为由真实字段推导
/// (位判断统一走 [`derive_channel_flags`],本函数不再自己算位):
/// - `on` = `batt_status` bit1(0x02 正在充电)
/// - `full` = `batt_status` bit0(0x01 充满)
/// - `fault` = `batt_status` bit4/5/6(过欠压 / 过流 / 热失控)**或** `pow_status` bit0(电源故障)
/// - `hex` = `batt_status` + `pow_status`(即通道 18 字符 hex 的前 4 个字符;缺任一字节则为空串)
/// - `status` = `fault` > `full` > `charging` > `idle`(见 [`ChannelFlags::status_name`])
/// - `temp` = 原样透传(单位 **K×100**,由前端换算,见 `web/.../cabinet.tsx`)
/// 另外**新增** `batt_status` / `pow_status` / `soc`(‰)/ `voltage`(mV)/ `current`(mA)
/// 便于管理端展示真实语义(新增字段不影响既有字段名)。
fn derive_channel_view(
batt_raw: Option<&str>,
pow_raw: Option<&str>,
soc: Option<i64>,
voltage: Option<i64>,
current: Option<i64>,
temp: Option<i64>,
) -> Value {
// hex 需要原始字节;位判断统一交给 `derive_channel_flags`(重复解析是无副作用的纯计算)
let batt = parse_hex_byte(batt_raw);
let pow = parse_hex_byte(pow_raw);
let flags = derive_channel_flags(batt_raw, pow_raw);
// 统一输出小写 2 位 hex(与 `report_worker` 的写入格式一致)
let batt_status = batt.map(|b| format!("{:02x}", b)).unwrap_or_default();
let pow_status = pow.map(|p| format!("{:02x}", p)).unwrap_or_default();
let hex = if batt_status.is_empty() || pow_status.is_empty() {
String::new()
} else {
format!("{}{}", batt_status, pow_status)
};
json!({
"status": flags.status_name(),
"on": flags.on,
"full": flags.full,
"fault": flags.fault,
"hex": hex,
"temp": temp,
"batt_status": batt_status,
"pow_status": pow_status,
"soc": soc,
"voltage": voltage,
"current": current,
})
}
// ---------------------------------------------------------------------------
// 单元测试:通道状态推导(协议 §3.8,口径必须与 web 端 deviceStatus.ts 一致)
//
// 只测纯函数(不连 Redis / DB):这一层是"设备实时详情"接口正确性的关键,
// 因为它负责把 `batt_status` / `pow_status` 翻译成前端消费的旧字段名,
// 以及把 DB 拓扑 + Redis 快照组装成 `boards` 响应。
// ---------------------------------------------------------------------------
#[cfg(test)]
mod tests {
use super::*;
// 设备列表测试要构造 `CurrentUser`(其 `scope` 字段是 `UserScope`),
// `super::*` 只带进来 `CurrentUser`,`UserScope` 需显式引入
use crate::middleware::auth::UserScope;
/// 交付文档 §5.3 示例 `828255152c01f40143`:有电池 + 正在充电、电源已打开
#[test]
fn test_derive_channel_view_charging_channel_sets_on_true() {
let view = derive_channel_view(Some("82"), Some("82"), Some(850), Some(54200), Some(5000), Some(32300));
assert_eq!(view["status"], "charging");
assert_eq!(view["on"], true);
assert_eq!(view["full"], false);
assert_eq!(view["fault"], false);
assert_eq!(view["hex"], "8282");
assert_eq!(view["batt_status"], "82");
assert_eq!(view["pow_status"], "82");
assert_eq!(view["soc"], 850);
assert_eq!(view["voltage"], 54200);
assert_eq!(view["current"], 5000);
assert_eq!(view["temp"], 32300);
}
/// 充满(`batt_status` bit0)+ 仍在报充电位 → `status` 取「充满」(优先级 full > charging)
#[test]
fn test_derive_channel_view_full_battery_prefers_full_over_charging() {
let view = derive_channel_view(Some("83"), Some("82"), None, None, None, None);
assert_eq!(view["full"], true);
assert_eq!(view["on"], true);
assert_eq!(view["status"], "full");
}
/// `batt_status` 的过欠压 / 过流 / 热失控位(bit4/5/6)任一置位 → 故障
#[test]
fn test_derive_channel_view_batt_fault_bits_map_to_fault() {
for (batt, other) in [("10", "00"), ("20", "00"), ("40", "00"), ("10", "82")] {
let view = derive_channel_view(Some(batt), Some(other), None, None, None, None);
assert_eq!(view["fault"], true, "batt={batt} pow={other}");
assert_eq!(view["status"], "fault", "batt={batt} pow={other}");
}
// 故障优先级最高:即使同时"充满 + 正在充电",仍是 fault
let view = derive_channel_view(Some("43"), Some("00"), None, None, None, None);
assert_eq!(view["fault"], true);
assert_eq!(view["status"], "fault");
}
/// `pow_status` bit0 = 电源故障 → 通道判故障(与电池状态相互独立)
#[test]
fn test_derive_channel_view_pow_fault_maps_to_fault() {
let view = derive_channel_view(Some("82"), Some("01"), None, None, None, None);
assert_eq!(view["fault"], true);
assert_eq!(view["status"], "fault");
assert_eq!(view["hex"], "8201");
}
/// **空仓**:无电池(batt `00`)+ 电源已打开(pow `82`)→ 语义仍是 idle,
/// 但电源状态照实返回(两维度独立,协议 §3.8)
#[test]
fn test_derive_channel_view_empty_channel_keeps_power_status() {
let view = derive_channel_view(Some("00"), Some("82"), Some(0), Some(0), Some(0), Some(0));
assert_eq!(view["status"], "idle");
assert_eq!(view["on"], false);
assert_eq!(view["full"], false);
assert_eq!(view["fault"], false);
assert_eq!(view["hex"], "0082");
assert_eq!(view["pow_status"], "82");
}
/// 字段整体缺失(老数据 / 通道刚建立)→ 全部落空,**不得 panic、不得拼出半个字节**
#[test]
fn test_derive_channel_view_missing_fields_yields_idle_and_empty_hex() {
let view = derive_channel_view(None, None, None, None, None, None);
assert_eq!(view["status"], "idle");
assert_eq!(view["on"], false);
assert_eq!(view["full"], false);
assert_eq!(view["fault"], false);
assert_eq!(view["hex"], "");
assert_eq!(view["batt_status"], "");
assert_eq!(view["pow_status"], "");
assert_eq!(view["soc"], Value::Null);
assert_eq!(view["temp"], Value::Null);
}
/// 只有一个字节(例如电源状态缺失)→ `hex` 必须为空串,而不是只拼一半
#[test]
fn test_derive_channel_view_single_byte_yields_empty_hex() {
let view = derive_channel_view(Some("82"), None, None, None, None, None);
assert_eq!(view["hex"], "");
assert_eq!(view["batt_status"], "82");
assert_eq!(view["status"], "charging");
}
/// hex 大小写不敏感 + 兼容 1 位写法,输出统一补零小写(与 web 端 `parseHexByte` 一致)
#[test]
fn test_derive_channel_view_normalizes_hex_case_and_padding() {
let view = derive_channel_view(Some("8A"), Some("A"), None, None, None, None);
assert_eq!(view["batt_status"], "8a");
assert_eq!(view["pow_status"], "0a");
assert_eq!(view["hex"], "8a0a");
}
// ---- ChannelFlags / derive_channel_counts:由通道状态派生「充满数 / 故障数」 ----
/// 全空闲:有数据但既没充满也没故障(`batt_status=00`,电源已开)→ 全部记 idle
#[test]
fn test_derive_channel_counts_all_idle_channels_yield_zero_full_fault() {
let channels: Vec<ChannelFlags> = ["00", "80"]
.iter()
.map(|batt| derive_channel_flags(Some(*batt), Some("82")))
.collect();
assert_eq!(derive_channel_counts(&channels), (2, 0, 0, 0));
}
/// 充电中(`batt_status` bit1)→ 记 charging,不算充满也不算故障
#[test]
fn test_derive_channel_counts_charging_channels_counted_as_charging() {
let channels = vec![
derive_channel_flags(Some("82"), Some("82")),
derive_channel_flags(Some("82"), Some("82")),
derive_channel_flags(Some("00"), Some("82")),
];
assert_eq!(derive_channel_counts(&channels), (1, 2, 0, 0));
}
/// 充满(`batt_status` bit0):即使同时带充电位(0x83)也只记充满
#[test]
fn test_derive_channel_counts_full_channels_counted_as_full() {
let channels = vec![
derive_channel_flags(Some("81"), Some("82")),
derive_channel_flags(Some("83"), Some("82")),
];
assert_eq!(derive_channel_counts(&channels), (0, 0, 2, 0));
}
/// 故障:`batt_status` bit4/5/6 与 `pow_status` bit0 都算故障(两个维度独立)
#[test]
fn test_derive_channel_counts_fault_channels_counted_as_fault() {
let channels = vec![
derive_channel_flags(Some("10"), Some("00")),
derive_channel_flags(Some("20"), Some("00")),
derive_channel_flags(Some("40"), Some("00")),
derive_channel_flags(Some("00"), Some("01")),
];
assert_eq!(derive_channel_counts(&channels), (0, 0, 0, 4));
}
/// **优先级冲突**:同时满足「充满」(bit0)与「故障」(bit4)时算故障,不算充满
#[test]
fn test_derive_channel_counts_fault_wins_over_full_on_conflict() {
// 0x11 = 充满 + 过欠压;0x13 = 充满 + 充电中 + 过欠压;0x41 = 充满 + 热失控
let channels = vec![
derive_channel_flags(Some("11"), Some("00")),
derive_channel_flags(Some("13"), Some("00")),
derive_channel_flags(Some("41"), Some("00")),
];
let (idle, charging, full, fault) = derive_channel_counts(&channels);
assert_eq!((idle, charging, full, fault), (0, 0, 0, 3));
}
/// **离线 / 无数据**:空集合与「字段全缺失」的通道 → 充满数故障数都必须为 0(不猜测)
#[test]
fn test_derive_channel_counts_offline_missing_data_yields_zero_full_fault() {
// 该柜在 Redis 里一个通道键都没有(设备从未上报)
assert_eq!(derive_channel_counts(&[]), (0, 0, 0, 0));
// 键存在但字段全缺失(老数据 / 通道刚建立)→ 无数据,同样不得计成充满或故障
let no_data = vec![
derive_channel_flags(None, None),
derive_channel_flags(None, None),
];
let (idle, charging, full, fault) = derive_channel_counts(&no_data);
assert_eq!((idle, charging, full, fault), (2, 0, 0, 0));
// 混合:一半有数据(一个充满 + 一个故障)、一半无数据 → 只统计有数据的那两个
let mixed = vec![
derive_channel_flags(Some("81"), Some("82")),
derive_channel_flags(Some("10"), Some("00")),
derive_channel_flags(None, None),
derive_channel_flags(None, None),
];
let (idle, charging, full, fault) = derive_channel_counts(&mixed);
assert_eq!((idle, charging, full, fault), (2, 0, 1, 1));
}
/// `has_data` / `status_code`:与前端 `0空闲/1充电中/2充满/3故障` 口径一致,无"禁用"档
#[test]
fn test_channel_flags_has_data_and_status_code_match_frontend_mapping() {
// 无数据(字段缺失)→ has_data=false,状态码回落 0(调用方据此改用 DB 占用值)
let none = derive_channel_flags(None, None);
assert!(!none.has_data);
assert_eq!(none.status_code(), 0);
// 只有电源状态、没有电池状态 → 数据存在,按电源维度判定(电源故障=3)
let pow_only = derive_channel_flags(None, Some("01"));
assert!(pow_only.has_data);
assert_eq!(pow_only.status_code(), 3);
assert_eq!(derive_channel_flags(Some("00"), Some("82")).status_code(), 0);
assert_eq!(derive_channel_flags(Some("82"), Some("82")).status_code(), 1);
assert_eq!(derive_channel_flags(Some("81"), Some("02")).status_code(), 2);
assert_eq!(derive_channel_flags(Some("10"), Some("00")).status_code(), 3);
// 非法 hex 与空串都按「无数据」处理(不 panic、不猜成充满/故障)
for bad in ["", " ", "zz", "123"] {
let flags = derive_channel_flags(Some(bad), Some(bad));
assert!(!flags.has_data, "非法 hex {bad:?} 必须按无数据处理");
assert_eq!(flags.status_code(), 0);
}
}
/// hex 解析:长度超 2 / 非 hex / 空白 → `None`
#[test]
fn test_parse_hex_byte_rejects_malformed_values() {
assert_eq!(parse_hex_byte(None), None);
assert_eq!(parse_hex_byte(Some("")), None);
assert_eq!(parse_hex_byte(Some(" ")), None);
assert_eq!(parse_hex_byte(Some("123")), None);
assert_eq!(parse_hex_byte(Some("zz")), None);
assert_eq!(parse_hex_byte(Some("82")), Some(0x82));
assert_eq!(parse_hex_byte(Some(" 82 ")), Some(0x82));
}
/// 设备列表 pipeline 的柜级 hex 位域取值(协议 §3.8):
/// `"00"`(全正常)必须原样保留,缺失/空串/纯空白必须为 `None`(响应 `null`)。
#[test]
fn test_pick_non_empty_hex_keeps_00_but_drops_missing() {
assert_eq!(
pick_non_empty_hex(Some(&Some("00".to_string()))).as_deref(),
Some("00"),
"\"00\" 是「全部正常」的有效值,不能当成无数据"
);
assert_eq!(
pick_non_empty_hex(Some(&Some("10".to_string()))).as_deref(),
Some("10"),
"柜门打开位必须原样透出"
);
assert_eq!(pick_non_empty_hex(Some(&Some(String::new()))), None);
assert_eq!(pick_non_empty_hex(Some(&Some(" ".to_string()))), None);
assert_eq!(pick_non_empty_hex(Some(&None)), None);
assert_eq!(pick_non_empty_hex(None), None);
}
/// 设备列表 pipeline 每设备字段数必须与命令顺序/解析下标同步
#[test]
fn test_device_list_fields_in_sync() {
assert_eq!(
DEVICE_LIST_FIELDS, 6,
"online/rssi/pow_fail_dc/pow_fail_ac/cabinet_status1/cabinet_status2"
);
}
// -----------------------------------------------------------------------
// 设备在线判据守门(防回归:不得再拿 `cabinets.status` 当"在线")
// -----------------------------------------------------------------------
/// 在线判据**三态**:只有明确的 `"1"` 才算在线;`"0"` 是明确离线;
/// `None`(无数据)必须与"离线"区分开 —— 三者混为一谈是上一轮 `cabinets.status` 错配的根源。
#[test]
fn test_device_online_flag_is_three_state() {
assert_eq!(device_online_flag(Some("1")), Some(true), "\"1\" = 在线");
assert_eq!(device_online_flag(Some("0")), Some(false), "\"0\" = 明确离线");
assert_eq!(device_online_flag(Some("")), Some(false), "空串不是在线");
assert_eq!(device_online_flag(Some("2")), Some(false), "只有 \"1\" 算在线");
assert_eq!(device_online_flag(None), None, "无数据 ≠ 离线,必须是 None");
}
/// **语义守门:0 台设备连接 ⇒ 在线数 0**(守卫"在线数退化成柜子总数"这条真实故障)。
///
/// 纯函数级覆盖,不需要真实 Redis:设备都没连时,`device:{imei}` 要么不存在
/// (`None`)要么是历史 `"0"`,两种都必须**不计入**在线。
#[test]
fn test_count_online_flags_is_zero_when_no_device_connected() {
// 68 台柜子,一台都没连:20 台键不存在(None)、48 台是明确的 "0"
let mut raws: Vec<Option<String>> = vec![Some("0".to_string()); 48];
raws.extend(vec![None; 20]);
assert_eq!(raws.len(), 68, "样本规模与生产现场一致");
assert_eq!(
count_online_flags(&raws),
0,
"无设备连接时必须为 0,绝不能等于柜子总数"
);
// 反向:全部 "1" ⇒ 全部在线(防止把判据写成"恒 0")
let all_online = vec![Some("1".to_string()); 68];
assert_eq!(count_online_flags(&all_online), 68);
// 混合:只有明确的 "1" 计入
let mixed = vec![
Some("1".to_string()),
Some("0".to_string()),
None,
Some(String::new()),
Some("2".to_string()),
Some("1".to_string()),
];
assert_eq!(count_online_flags(&mixed), 2);
// 空列表(无可见柜子)不 panic 且为 0
assert_eq!(count_online_flags(&[]), 0);
}
/// **口径同源**:`is_device_online`(布尔)与 `count_online_flags`(计数)
/// 必须与 `device_online_flag`(三态)逐值一致 —— 三者是同一判据的三种形态。
#[test]
fn test_online_predicate_forms_are_mutually_consistent() {
let samples: Vec<Option<String>> = vec![
Some("1".into()),
Some("0".into()),
Some("".into()),
Some("2".into()),
Some("true".into()),
None,
];
for raw in &samples {
assert_eq!(
is_device_online(raw.as_deref()),
device_online_flag(raw.as_deref()) == Some(true),
"布尔形态与三态形态不一致:{:?}",
raw
);
}
assert_eq!(count_online_flags(&samples), 1);
}
/// 在线取数的 Redis key / 字段名必须与上报写入侧(`device:{imei}` 的 `online`)**逐字一致**:
/// 取数来源写错就永远判离线(表现为"全部离线",静默无报错、无异常)。
#[test]
fn test_device_online_source_matches_report_writer_contract() {
assert_eq!(DEVICE_ONLINE_FIELD, "online", "必须与 report_worker 写入的字段名一致");
assert_eq!(DEVICE_ONLINE_VALUE, "1", "必须与 report_worker 写入的取值一致");
assert_eq!(device_key("864720562557021"), "device:864720562557021");
// 设备列表 pipeline 的**第 0 个** HGET 字段就是在线字段(`base + 0`)
assert_eq!(
DEVICE_ONLINE_FIELD, "online",
"attach_realtime_status 的 base+0 必须解析在线字段"
);
}
// ---- 通道级 key / 拓扑闸门(板号 1–15 / 通道号 1–8)-------
/// 协议量程内合法,越界一律拒绝(`board_id=0` / `channel_id=9` 拼不出有效 key)
#[test]
fn test_is_valid_channel_key_parts_accepts_protocol_range() {
assert!(is_valid_channel_key_parts(1, 1));
assert!(is_valid_channel_key_parts(6, 6));
assert!(is_valid_channel_key_parts(15, 8));
}
/// 越界与 0 值必须拒绝:老实现按 KEYS 末段取"板号"时,内部键
/// `...:ch:{通道号}:bat` / `...:ch:{通道号}:order` 会生成幽灵板/幽灵通道
#[test]
fn test_is_valid_channel_key_parts_rejects_out_of_range() {
assert!(!is_valid_channel_key_parts(0, 1));
assert!(!is_valid_channel_key_parts(16, 1));
assert!(!is_valid_channel_key_parts(1, 0));
assert!(!is_valid_channel_key_parts(1, 9));
}
/// 通道级 key 必须与上报写入侧逐字一致(板号/通道号十进制不补零)
#[test]
fn test_channel_redis_key_matches_writer_format() {
assert_eq!(
channel_redis_key("864720562557021", 1, 3),
"device:864720562557021:board:1:ch:3"
);
assert_eq!(
channel_redis_key("864720562557021", 15, 8),
"device:864720562557021:board:15:ch:8"
);
}
/// 组装:Redis 里**不存在**的通道必须整条省略(与旧 `KEYS` 只返回存在键语义一致)
#[test]
fn test_assemble_boards_omits_channels_absent_from_redis() {
let topology = vec![(1i8, 1i8), (1, 2), (2, 1)];
let present = HashMap::from([
("batt_status".to_string(), "82".to_string()),
("pow_status".to_string(), "82".to_string()),
("soc".to_string(), "850".to_string()),
("voltage".to_string(), "54000".to_string()),
("current".to_string(), "8000".to_string()),
("temp".to_string(), "29800".to_string()),
]);
let snapshots: Vec<ChannelSnapshot> = vec![Some(present), None, None];
let boards = assemble_boards(&topology, &snapshots);
// 板 1 只有通道 1(通道 2 无 key ⇒ 省略),板 2 整块不存在 ⇒ 不出现
assert_eq!(boards["1"]["channels"]["1"]["status"], "charging");
assert!(boards["1"]["channels"].get("2").is_none(), "缺 key 的通道不得出现");
assert!(boards.get("2").is_none(), "整块板无数据时不得生成幽灵板");
assert_eq!(boards["1"]["status_hex"], "8282");
}
/// 组装:`status_hex` 按**通道号升序**拼接(拓扑有序即输出有序)
#[test]
fn test_assemble_boards_status_hex_follows_topology_order() {
let topology = vec![(1i8, 1i8), (1, 2)];
let snap = |batt: &str, pow: &str| -> ChannelSnapshot {
Some(HashMap::from([
("batt_status".to_string(), batt.to_string()),
("pow_status".to_string(), pow.to_string()),
]))
};
// SQL 已 `ORDER BY board_id, channel_id`,故按拓扑顺序拼接即可得到通道升序
let boards = assemble_boards(&topology, &[snap("01", "82"), snap("03", "82")]);
assert_eq!(boards["1"]["status_hex"], "01820382");
}
/// 组装:空拓扑 / 全无数据 ⇒ 空对象(不得 panic、不得造板)
#[test]
fn test_assemble_boards_empty_inputs_return_empty_object() {
assert_eq!(assemble_boards(&[], &[]), json!({}));
let topology = vec![(1i8, 1i8)];
assert_eq!(assemble_boards(&topology, &[None]), json!({}));
}
/// 快照字段缺失时 `channel_view_from_snapshot` 不得 panic,按"无数据"落空值
#[test]
fn test_channel_view_from_snapshot_handles_missing_fields() {
let empty = HashMap::new();
let view = channel_view_from_snapshot(&empty);
assert_eq!(view["status"], "idle");
assert_eq!(view["hex"], "");
assert_eq!(view["soc"], Value::Null);
}
// -----------------------------------------------------------------------
// 设备列表:组织隔离 + 数据范围(user_scopes)叠加
// -----------------------------------------------------------------------
/// 构造一个客户侧用户(role=1,属组织 42),scope 由调用方给定
fn make_user(scope: UserScope) -> CurrentUser {
CurrentUser {
user_id: 9,
phone: "13800138000".into(),
permissions: std::collections::HashSet::new(),
role_level: 1,
organization_id: Some(42),
scope,
}
}
/// 无任何入参的客户账号 ⇒ **组织隔离条件必须出现**(未配置 scope 时不额外收窄)
#[test]
fn test_build_filter_where_without_scope_keeps_org_isolation() {
let params = CabinetFilterParams {
organization_id: None,
project_id: None,
page: None,
page_size: None,
bound: None,
};
let (clause, binds) = build_filter_where(&params, &make_user(UserScope::default()));
assert_eq!(
clause,
"WHERE c.project_id IN (SELECT pj.id FROM projects pj WHERE pj.organization_id = ?)"
);
assert_eq!(binds, vec!["42"]);
}
/// **核心越权回归**:客户传别人的 `project_id` / `organization_id`,
/// 组织隔离条件**必须仍然叠加** ⇒ 结果只可能是本组织数据(越权取数不可发生)
#[test]
fn test_build_filter_where_foreign_project_and_org_params_keep_org_isolation() {
let user = make_user(UserScope::default());
let foreign_project = CabinetFilterParams {
organization_id: None,
project_id: Some(9999),
page: None,
page_size: None,
bound: None,
};
let (clause, binds) = build_filter_where(&foreign_project, &user);
assert_eq!(
clause,
"WHERE c.project_id = ? AND c.project_id IN \
(SELECT pj.id FROM projects pj WHERE pj.organization_id = ?)"
);
assert_eq!(binds, vec!["9999", "42"], "占位符顺序 = 绑定顺序");
let foreign_org = CabinetFilterParams {
organization_id: Some(9999),
project_id: None,
page: None,
page_size: None,
bound: None,
};
let (clause, binds) = build_filter_where(&foreign_org, &user);
assert!(
clause.contains("o.id = ? OR o.parent_id = ?"),
"入参组织分支保持原口径:{}",
clause
);
assert!(
clause.ends_with("AND c.project_id IN (SELECT pj.id FROM projects pj WHERE pj.organization_id = ?)"),
"组织参数分支也必须叠加归属隔离:{}",
clause
);
assert_eq!(binds, vec!["9999", "9999", "42"]);
}
/// 无组织归属的客户侧账号 ⇒ `1=0` 空集(**绝不退化成全量**),
/// 且此时传任何入参都不能把自己"放开"
#[test]
fn test_build_filter_where_no_org_user_is_empty_even_with_params() {
let mut user = make_user(UserScope::default());
user.organization_id = None;
for project_id in [None, Some(9999)] {
let params = CabinetFilterParams {
organization_id: None,
project_id,
page: None,
page_size: None,
bound: None,
};
let (clause, _) = build_filter_where(&params, &user);
assert!(clause.contains("1=0"), "无组织必须空集:{}", clause);
}
}
/// **配置了项目范围**:组织条件之后必须再叠加 `c.project_id IN (...)`(取交集)
#[test]
fn test_build_filter_where_with_scope_appends_project_narrowing() {
let params = CabinetFilterParams {
organization_id: None,
project_id: None,
page: None,
page_size: None,
bound: None,
};
let user = make_user(UserScope { project_ids: vec![7, 9] });
let (clause, binds) = build_filter_where(&params, &user);
assert!(clause.contains("pj.organization_id = ?"), "组织隔离不能丢:{}", clause);
assert!(clause.contains("AND c.project_id IN (?,?)"), "数据范围必须叠加:{}", clause);
assert_eq!(binds, vec!["42", "7", "9"], "绑定顺序必须与占位符顺序一致");
}
/// 显式指定了范围外项目 / 其他组织时,范围条件仍然叠加 ⇒ 结果必然为空集(不泄露)
#[test]
fn test_build_filter_where_scope_applies_to_explicit_project_and_org_params() {
let user = make_user(UserScope { project_ids: vec![7] });
let by_project = CabinetFilterParams {
organization_id: None,
project_id: Some(999),
page: None,
page_size: None,
bound: None,
};
let (clause, binds) = build_filter_where(&by_project, &user);
assert!(clause.starts_with("WHERE c.project_id = ? AND c.project_id IN (SELECT pj.id"));
assert!(clause.ends_with("AND c.project_id IN (?)"), "{}", clause);
assert_eq!(binds, vec!["999", "42", "7"]);
let by_org = CabinetFilterParams {
organization_id: Some(43),
project_id: None,
page: None,
page_size: None,
bound: None,
};
let (clause, binds) = build_filter_where(&by_org, &user);
assert!(clause.contains("AND c.project_id IN (?)"), "组织参数分支也要叠加范围:{}", clause);
assert_eq!(binds, vec!["43", "43", "42", "7"]);
}
/// 已绑定/未绑定过滤与组织隔离、数据范围可以共存,且绑定值顺序正确
#[test]
fn test_build_filter_where_scope_combines_with_bound_filter() {
let params = CabinetFilterParams {
organization_id: None,
project_id: None,
page: None,
page_size: None,
bound: Some("1".into()),
};
let user = make_user(UserScope { project_ids: vec![7] });
let (clause, binds) = build_filter_where(&params, &user);
assert!(clause.contains("c.project_id IS NOT NULL"), "{}", clause);
assert!(clause.ends_with("AND c.project_id IN (?)"), "{}", clause);
assert_eq!(binds, vec!["42", "7"]);
}
/// 总管理员(role_level>=2)不受组织隔离与数据范围收窄(豁免);
/// 但**显式入参仍然生效**(只是纯筛选,不是边界)
#[test]
fn test_build_filter_where_super_admin_ignores_scope() {
let params = CabinetFilterParams {
organization_id: None,
project_id: None,
page: None,
page_size: None,
bound: None,
};
let mut user = make_user(UserScope { project_ids: vec![7] });
user.role_level = 2;
user.organization_id = None;
let (clause, binds) = build_filter_where(&params, &user);
assert!(clause.is_empty(), "总管理员无任何过滤:{}", clause);
assert!(binds.is_empty());
let by_project = CabinetFilterParams {
organization_id: None,
project_id: Some(5),
page: None,
page_size: None,
bound: None,
};
let (clause, binds) = build_filter_where(&by_project, &user);
assert_eq!(clause, "WHERE c.project_id = ?");
assert_eq!(binds, vec!["5"]);
}
// -----------------------------------------------------------------------
// 组织树:组织隔离 + 数据范围
// -----------------------------------------------------------------------
/// 组织树组织查询:工厂不限 / 客户仅本组织 / 无组织空集
#[test]
fn test_visible_orgs_sql_isolation() {
let (sql, binds) = visible_orgs_sql(&make_user(UserScope::default()));
assert!(sql.contains("WHERE 1=1 AND id = ?"), "{}", sql);
assert_eq!(binds, vec!["42"]);
let (sql, binds) = visible_orgs_sql(&{
let mut u = make_user(UserScope::default());
u.role_level = 2;
u.organization_id = None;
u
});
assert!(sql.contains("WHERE 1=1 ORDER BY id"), "总管理员全量:{}", sql);
assert!(binds.is_empty());
let (sql, _) = visible_orgs_sql(&{
let mut u = make_user(UserScope::default());
u.organization_id = None;
u
});
assert!(sql.contains("WHERE 1=1 AND 1=0"), "无组织必须空集:{}", sql);
}
/// 组织树项目查询:组织隔离 ∧ 数据范围(未配置 scope 时不额外收窄)
#[test]
fn test_visible_projects_sql_isolation_and_scope() {
let (sql, binds) = visible_projects_sql(&make_user(UserScope::default()));
assert!(sql.contains("p.organization_id = ?"), "{}", sql);
assert!(!sql.contains("p.id IN"), "未配置 scope 不得收窄:{}", sql);
assert_eq!(binds, vec!["42"]);
let (sql, binds) = visible_projects_sql(&make_user(UserScope { project_ids: vec![7, 9] }));
assert!(sql.contains("p.organization_id = ?"), "{}", sql);
assert!(sql.contains("AND p.id IN (?,?)"), "数据范围必须叠加:{}", sql);
assert_eq!(binds, vec!["42", "7", "9"], "占位符顺序 = 绑定顺序");
let (sql, binds) = visible_projects_sql(&{
let mut u = make_user(UserScope { project_ids: vec![7] });
u.role_level = 2;
u.organization_id = None;
u
});
assert!(!sql.contains("organization_id = ?"), "总管理员不限组织:{}", sql);
assert!(!sql.contains("p.id IN"), "总管理员不受范围收窄:{}", sql);
assert!(binds.is_empty());
}
/// 设备归属校验 SQL:未配置范围时不收窄,配置后必须叠加项目收窄
#[test]
fn test_device_access_sql_appends_scope_only_when_configured() {
let plain = DEVICE_ACCESS_SQL.replacen("{}", "", 1);
assert!(plain.contains("p.organization_id = ?)"));
assert!(!plain.contains("project_id IN"));
let (cond, binds) = UserScope { project_ids: vec![7, 9] }.project_condition("p.id");
let narrowed = DEVICE_ACCESS_SQL.replacen("{}", &cond, 1);
assert!(narrowed.contains("p.id IN (?,?)"), "{}", narrowed);
assert_eq!(binds, vec!["7", "9"]);
}
}