diff --git a/software/api-server/src/db/migrate.rs b/software/api-server/src/db/migrate.rs index 049e3da..68721c8 100644 --- a/software/api-server/src/db/migrate.rs +++ b/software/api-server/src/db/migrate.rs @@ -240,6 +240,15 @@ const SCHEMA_SQL: &str = r#" FOREIGN KEY (cabinet_id) REFERENCES cabinets(id) ON DELETE CASCADE ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; + CREATE TABLE IF NOT EXISTS cabinet_meter_readings ( + id BIGINT AUTO_INCREMENT PRIMARY KEY, + cabinet_id BIGINT NOT NULL COMMENT '柜子ID', + reading_wh BIGINT NOT NULL COMMENT '电表读数(Wh;协议平台单位,硬件最小刻度为 0.1 度电=100Wh,设备已换算后上报)', + collected_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '平台收到该读数的时刻', + FOREIGN KEY (cabinet_id) REFERENCES cabinets(id) ON DELETE CASCADE, + KEY idx_cabinet_meter_readings_cabinet (cabinet_id, id) + ) 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', @@ -820,6 +829,15 @@ async fn add_columns_if_not_exists(pool: &MySqlPool) -> Result<(), sqlx::Error> ensure_column(pool, "cabinets", "latitude", "DECIMAL(10,7) NULL COMMENT '纬度'").await?; ensure_column(pool, "cabinets", "longitude", "DECIMAL(10,7) NULL COMMENT '经度'").await?; + // cabinets:**最新电表读数**(柜级累计电能)。协议 §5.3 的 `status_post.cabinet.meter` + // 明确要求「平台原样存库、不再 ×100」,此前只写了 Redis(TTL 300s)⇒ 上报被清掉、 + // 历史无法回溯。这里落一份"最新值"便于列表/详情廉价读取; + // **变化留痕**在 `cabinet_meter_readings`(只在读数变化时追加,见 `workers/meter.rs`)。 + ensure_column(pool, "cabinets", "meter_wh", "BIGINT NULL COMMENT '最新电表读数(Wh);NULL=从未上报过'") + .await?; + ensure_column(pool, "cabinets", "meter_at", "DATETIME NULL COMMENT '最新电表读数的收到时刻'") + .await?; + // cabinets:设备认证 / 拓扑 / 工作模式(线上曾缺失,必须补齐) ensure_column( pool, @@ -1490,7 +1508,9 @@ mod tests { /// 外键引用的关键字(用于从建表语句中抽取被引用表名) const REFERENCES_MARKER: &str = "REFERENCES "; /// 期望的建表数量(表数量变化时该断言会提醒同步更新) - const EXPECTED_TABLE_COUNT: usize = 21; + /// 建表数量(**新增/删除表时必须同步**,由 est_schema_sql_table_order_has_forward_reference_returns_none 守门)。 +/// 21 → 22(2026-09-20 新增 cabinet_meter_readings:电表读数变化留痕,见 workers/meter.rs)。 +const EXPECTED_TABLE_COUNT: usize = 22; /// 抽取建表语句中所有 `REFERENCES <表名>` 的被引用表名 fn referenced_tables(statement: &str) -> Vec<&str> { diff --git a/software/api-server/src/workers/meter.rs b/software/api-server/src/workers/meter.rs new file mode 100644 index 0000000..e6d407b --- /dev/null +++ b/software/api-server/src/workers/meter.rs @@ -0,0 +1,184 @@ +//! 柜级电表读数(`status_post.cabinet.meter`)的落库与**变化留痕**。 +//! +//! # 为什么需要它 +//! +//! 协议 §5.3 的 `status_post` **本来就在上报** `cabinet.meter`(电表读数,单位 **Wh**, +//! 硬件 0.1kWh 读数 ×100,示例 `125500` = 125.5 kWh),字段表还明确写着 +//! 「**平台原样存库、不再 ×100**」(`docs/protocol/柜控板固件对接文档.md` L565)。 +//! +//! 但实现侧此前**只把它写进 Redis**(`device:{imei}` 的 `meter` 字段,TTL 300 秒), +//! **没有落库** ⇒ 每 60 秒上报一次、5 分钟后被清掉,**整柜累计电能在平台侧等于不存在、 +//! 历史也无法回溯**。本模块补上这一步(属**履约**,不是新增协议内容)。 +//! +//! # 与「订单电量」的区别(⛔ 别混) +//! +//! | | 含义 | 来源 | +//! |---|---|---| +//! | **`meter`**(本模块) | **整柜累计电能**(电表读数,天然是累计值) | 协议已有,硬件上报 | +//! | `energy` | **单订单充电量** | 协议**没有**该字段(待确认第 13 项),平台按 `V×I×Δt` 积分(见 [`super::energy`]) | +//! +//! ⛔ **整柜累计电量绝不能由订单电量相加得到**(含非充电损耗差异、且订单电量本身是估算)。 +//! +//! # 为什么"只记变化" +//! +//! `status_post` 固定每 60 秒一次 ⇒ 每柜每天 1440 条。若无条件落库,68 柜即 **≈9.8 万行/天**, +//! 而电表读数**只在有电量消耗时才变**。因此只在**读数相对上次发生变化**时追加一行 —— +//! 写入量与真实用电事件成正比,且"变化"本身就是我们要的那个信息。 +//! +//! ⚠️ 读数**变小也要记**(换表 / 电表清零 / 计数器回绕都属真实事件,必须留痕), +//! 但不在此处做任何"补差"推算 —— 口径是**宁可少算、不可虚增**(与 [`super::energy`] 同)。 + +use sqlx::MySqlPool; + +/// 更新 `cabinets` 上的**最新读数**(`meter_wh` / `meter_at`)。 +/// +/// `WHERE ... (meter_wh IS NULL OR meter_wh <> ?)` 是**原子闸门**: +/// `rows_affected == 1` 表示"本次读数确实发生了变化,且由本语句认领"。 +/// 并发下同一柜的两条 `status_post` 也只会有一条认领同一个新值(另一条 `<>` 不成立)。 +const UPDATE_CABINET_METER_SQL: &str = "UPDATE cabinets SET meter_wh = ?, meter_at = NOW() \ + WHERE imei = ? AND (meter_wh IS NULL OR meter_wh <> ?)"; + +/// 追加一条**变化留痕**。`cabinet_id` 由 `imei` 反查,省掉一次 `SELECT` 往返。 +const INSERT_METER_READING_SQL: &str = + "INSERT INTO cabinet_meter_readings (cabinet_id, reading_wh) SELECT id, ? FROM cabinets WHERE imei = ?"; + +/// 读上次读数(`imei` 是唯一索引,成本可忽略)。 +/// +/// 为什么先读一次:让 [`should_record`] 这个**纯函数判据真正成为生产判据** +/// (否则它只被单测使用 = 空壳,判据与实现会各走各的)。 +const SELECT_CABINET_METER_SQL: &str = "SELECT meter_wh FROM cabinets WHERE imei = ?"; + +/// **判据(纯函数,可单测)**:本次读数是否应当落库留痕。 +/// +/// - `current == None`:报文**没有** `meter` 字段(或值非法)⇒ **不记录** +/// (⛔ 关键:绝不能把"缺失"当成 `0` 写进去 —— 那会在序列里插进假的零值, +/// 后续算区间用量时就是一个凭空的巨大差值。历史实现用 `unwrap_or(0)` 正是这个毛病); +/// - `prev == None`:库中尚无记录(首次上报新柜 / 新加列)⇒ **记录**,作为基准; +/// - 有变化(含**变小**)⇒ **记录**; +/// - 完全相等 ⇒ 不记录(这是绝大多数情况,60 秒一次但读数常常不变)。 +pub fn should_record(prev: Option, current: Option) -> bool { + match current { + None => false, + Some(_) => prev != current, + } +} + +/// 落库一次读数(**只有变化时才写**)。 +/// +/// 返回 `Ok(true)` 表示本次确实记录了变化(`UPDATE` 认领成功且已追加留痕行)。 +/// +/// 失败**只向上抛给调用方记日志**:电表留痕是增值信息,绝不能因为库抖动影响 +/// `status_post` 的主流程(在线状态、通道实时值)。 +pub async fn record_if_changed( + pool: &MySqlPool, + imei: &str, + meter_wh: Option, +) -> Result { + let Some(current) = meter_wh else { + // 报文没带 meter(或值非法)⇒ 什么都不做(见 should_record 的说明) + return Ok(false); + }; + + // ① **判据**:读上次读数,交给纯函数 [`should_record`] 决定要不要留痕。 + // 绝大多数 `status_post`(读数没变)在这里就返回,连 UPDATE 都不发。 + let prev: Option> = sqlx::query_scalar(SELECT_CABINET_METER_SQL) + .bind(imei) + .fetch_optional(pool) + .await?; + // 三条路径要分清:柜子不存在(None)/ 该列从未写过(Some(None))/ 有旧值(Some(Some(v))) + let prev = prev.flatten(); + if !should_record(prev, Some(current)) { + return Ok(false); + } + + // ② **并发认领**:`should_record` 是"读后判",与写之间存在竞态窗口 + // (同一柜两条 status_post 并发)。真正的闸门是这条 SQL 的 `WHERE`: + // 只有"库里当下仍与传入值不同"的那一条能 `rows_affected == 1`。 + let changed = sqlx::query(UPDATE_CABINET_METER_SQL) + .bind(current) + .bind(imei) + .bind(current) + .execute(pool) + .await? + .rows_affected(); + + if changed == 0 { + return Ok(false); + } + + sqlx::query(INSERT_METER_READING_SQL) + .bind(current) + .bind(imei) + .execute(pool) + .await?; + + Ok(true) +} + +#[cfg(test)] +mod tests { + use super::*; + + /// ⛔ **最关键的一条**:字段缺失(`None`)绝不能落库成 `0`。 + /// 若把缺失当 0,序列里会出现凭空的"骤降到 0 再骤升回去",区间用量会被算成巨大的假值。 + #[test] + fn test_missing_meter_is_never_recorded() { + assert!(!should_record(None, None), "一直缺字段 ⇒ 永不记录"); + assert!(!should_record(Some(125_500), None), "本次缺字段 ⇒ 不记录(不能当 0)"); + } + + /// 首次上报(库中无值)⇒ 建立基准,必须记录。 + #[test] + fn test_first_reading_is_recorded_as_baseline() { + assert!(should_record(None, Some(0)), "首次上报 0 也是合法基准"); + assert!(should_record(None, Some(125_500)), "首次上报真实读数"); + } + + /// 读数**未变化** ⇒ 不记录(这是绝大多数情况;60 秒一次但读数常常不变)。 + #[test] + fn test_unchanged_reading_is_not_recorded() { + assert!(!should_record(Some(125_500), Some(125_500))); + assert!(!should_record(Some(0), Some(0))); + } + + /// 读数**变化**(含**变小**:换表 / 清零 / 回绕)⇒ 必须留痕。 + /// 变小不在这里做任何补差推算 —— 口径是"宁可少算、不可虚增"。 + #[test] + fn test_changed_reading_is_recorded_including_decrease() { + assert!(should_record(Some(125_500), Some(125_600)), "正常增长"); + assert!(should_record(Some(125_500), Some(1_000)), "变小也要留痕(换表/清零/回绕)"); + assert!(should_record(Some(125_500), Some(0)), "归零也要留痕"); + assert!(should_record(Some(0), Some(1)), "从 0 起算的变化"); + } + + /// 反证:`should_record` 对"缺失 vs 0"的处置必须**不同** —— + /// 若哪天有人把它简化成 `prev != current`,这条会红。 + #[test] + fn test_absent_and_zero_are_treated_differently() { + // 首次(无 prev)时:Some(0) 记录、None 不记录 + assert_ne!( + should_record(None, Some(0)), + should_record(None, None), + "『首次读到 0』与『没读到』必须区分 —— 否则缺失会被写成 0" + ); + } + + /// SQL 守门:`WHERE` 子句里的"变化闸门"与绑定占位符个数不得被改坏。 + #[test] + fn test_update_guard_sql_shape() { + assert!( + UPDATE_CABINET_METER_SQL.contains("meter_wh IS NULL OR meter_wh <> ?"), + "变化闸门不得移除:它同时承担『首次(IS NULL)』与『有变化(<> ?)』两种情形。当前:{}", + UPDATE_CABINET_METER_SQL + ); + assert_eq!( + UPDATE_CABINET_METER_SQL.matches('?').count(), + 3, + "占位符个数必须与 bind 顺序一致(新值 / imei / 比较值)" + ); + assert!( + INSERT_METER_READING_SQL.contains("SELECT id, ? FROM cabinets WHERE imei = ?"), + "留痕插入靠 imei 反查 cabinet_id,别改成需要额外 SELECT" + ); + } +} diff --git a/software/api-server/src/workers/mod.rs b/software/api-server/src/workers/mod.rs index 6fa2090..8e17999 100644 --- a/software/api-server/src/workers/mod.rs +++ b/software/api-server/src/workers/mod.rs @@ -1,6 +1,7 @@ //! 后台协程 — 处理 Redis 队列中的设备消息 pub mod energy; +pub mod meter; pub mod log_retention; pub mod node_checker; pub mod order_timeout; diff --git a/software/api-server/src/workers/report_worker.rs b/software/api-server/src/workers/report_worker.rs index 5c2b600..029a54e 100644 --- a/software/api-server/src/workers/report_worker.rs +++ b/software/api-server/src/workers/report_worker.rs @@ -1036,7 +1036,11 @@ async fn handle_status_post( let cabinet = msg.get("cabinet"); let rssi = msg.get("rssi").and_then(|v| v.as_i64()).unwrap_or(0); let temp = cabinet.and_then(|c| c.get("temp")).and_then(|v| v.as_i64()).unwrap_or(0); - let meter = cabinet.and_then(|c| c.get("meter")).and_then(|v| v.as_i64()).unwrap_or(0); + // ⚠️ 用 `Option` **保留「字段缺失」与「值为 0」的区别**: + // 缺失绝不能当成 0 落库(否则电表序列里会插进凭空的零值,见 `workers::meter`)。 + // Redis 那条路径沿用原口径(缺失即 0),不改既有实时状态契约。 + let meter_raw = cabinet.and_then(|c| c.get("meter")).and_then(|v| v.as_i64()); + let meter = meter_raw.unwrap_or(0); let raw_status1 = cabinet.and_then(|c| c.get("status1")).and_then(|v| v.as_str()); let raw_status2 = cabinet.and_then(|c| c.get("status2")).and_then(|v| v.as_str()); @@ -1054,6 +1058,24 @@ async fn handle_status_post( // 1) 设备级 Hash:在线标记 + 信号 + 柜体状态(TTL 300 秒) write_device_state(redis, dev_id, rssi, status1.as_deref(), status2.as_deref(), temp, meter).await; + // 1b) **电表读数落库 + 变化留痕**(协议 §5.3 字段表要求「平台原样存库」; + // 此前只写 Redis,5 分钟后被 TTL 清掉 ⇒ 整柜累计电能留不下来)。 + // 只在读数**变化**时写(60 秒一次但读数常常不变); + // 失败只记 WARN:留痕是增值信息,绝不能影响 status_post 的主流程。 + match crate::workers::meter::record_if_changed(mysql, dev_id, meter_raw).await { + Ok(true) => tracing::info!( + "[status_post] 电表读数变化已留痕 dev_id={} meter={}Wh", + dev_id, + meter_raw.unwrap_or(0) + ), + Ok(false) => {} + Err(e) => tracing::warn!( + "[status_post] 电表读数留痕失败(不影响状态上报)dev_id={} err={}", + dev_id, + e + ), + } + // 2) 供电恢复即清除停电标记:`cabinet.status1` bit0 = 0(正常)→ pow_fail_dc/ac 置 0; // bit0 = 1 时**不动**(停电事件由 pow_fail_dc / pow_fail_ac 单独置 1,协议 §5.4) clear_power_fail_if_normal(redis, dev_id, raw_status1).await;