charging-cabinet/software/device-server/src/tcp/handler.rs

//! 设备消息处理逻辑（轻量化）。
//!
//! 设备服务只做**协议透传**，不处理业务逻辑：
//! - `status_post` 的通道数据在设备侧已是 **18 字符 hex 明文**，本服务**原样透传**，
//!   不再做任何拆解/位域解析（解析归 api-server 的业务层）
//! - 所有设备上报消息统一转发到 Redis `device:reports` 队列
//! - 不写 DB、不验证签名、不处理业务规则
//! - auth_str/login 等协议级消息也由 API 服务处理后通过 `device:commands` 回复

use super::connection::ConnectionPool;
use super::protocol::DeviceMessage;

/// 上报队列 `device:reports` 的**长度上限**（超过就裁剪最旧的）。
///
/// **为什么必须有上限**：本服务只 `LPUSH`，api-server 是唯一消费者；一旦 api-server 长时间不消费
/// （宕机/卡死/上线顺序颠倒），队列会无界增长，而 Redis 实例 `maxmemory=1GB`、
/// `maxmemory-policy=volatile-lru` 且实例上绝大多数 key 没有 TTL（可淘汰集合≈1）⇒
/// **写命令 OOM 会连累同实例上既有的充电站生产平台**。
///
/// **为什么取 200000**（估算依据）：
/// - 单条 `status_post` 序列化后约 **1.1~1.2KB**（含 36 通道 18 字符 hex + 柜级字段）；
/// - 68 柜规模：每柜每天 1440 条（协议 §5.3 固定 60 秒一次）⇒ 约 **9.8 万条/天 ≈ 115MB/天**，
///   20 万条可容纳 **约 2 天**的完全积压（远超"重启窗口 + 正常积压"）；
/// - 1000 柜规模：约 **144 万条/天**，20 万条 ≈ **3.3 小时** 的缓冲；
/// - 内存占用上限 ≈ 20 万 × 1.15KB ≈ **230MB**，只占 1GB 实例的约 23%，
///   给同实例其他业务留足余量（这正是设上限的目的）。
///
/// ⚠️ 不改队列名、不改报文格式（协议冻结）。
const REPORTS_QUEUE_MAX: i64 = 200_000;

/// 分发消息：补写设备身份后转发到 Redis 上报队列。
///
/// `current_dev_id` = 本 TCP 连接绑定的设备身份（由 `auth_str`/`login` 建立）。
/// 为什么需要它：新协议里业务上行**不带 `dev_id`**（设备身份由连接确定），
/// 而 api-server 依赖 `dev_id` 识别设备，故转发前必须补上。
///
/// 本服务**不直接回复设备**：应答由 api-server 经 Redis `device:commands` 下发，
/// 因此这里没有返回值（历史上曾返回 `Option<ServerResponse>` 但恒为 `None`，
/// 那个结构体副本已删除，避免与 api-server 的真实定义漂移）。
pub async fn handle_message(
    mut msg: DeviceMessage,
    current_dev_id: Option<&str>,
    _pool: &ConnectionPool,
    redis: &redis::aio::ConnectionManager,
) {
    attach_dev_id(&mut msg, current_dev_id);

    // 所有消息统一转发到 Redis，由 API 服务的 report_worker 处理
    forward_to_redis(redis, &msg).await;

    // 不直接回复设备，等待 API 服务通过 device:commands 队列回复
}

/// 把连接绑定的设备身份写入待转发报文。
///
/// 以**连接**为准（而非报文自带值）：设备身份由 TCP 连接确定，
/// 这样即使设备漏带/错带 `dev_id`，api-server 侧也不会认错设备。
fn attach_dev_id(msg: &mut DeviceMessage, current_dev_id: Option<&str>) {
    match current_dev_id {
        Some(dev_id) if !dev_id.is_empty() => msg.dev_id = Some(dev_id.to_string()),
        // 理论上不会出现：未登录连接的业务报文已被 server.rs 的门控丢弃。
        // 此时仍照常转发（不丢数据），仅记 WARN 便于排查。
        _ => tracing::warn!(
            "[forward] 连接未绑定 dev_id，转发报文将不含设备身份, act={}",
            msg.act
        ),
    }
}

/// 将设备消息转发到 Redis 统一上报队列 `device:reports`
///
/// `LPUSH` 后用同一条 pipeline 做 `LLEN → LTRIM → LLEN`：
/// - 保证队列**有界**（上限见 [`REPORTS_QUEUE_MAX`]）；
/// - 裁剪发生时记 **WARN** 并给出被丢弃条数 —— 运维必须知道"丢了上报"，
///   而不是让 Redis 在无界增长里悄悄 OOM。
async fn forward_to_redis(redis: &redis::aio::ConnectionManager, msg: &DeviceMessage) {
    let json_str = match serde_json::to_string(msg) {
        Ok(json) => json,
        Err(e) => {
            // 序列化失败说明报文结构异常，记 ERROR 并丢弃（不推送半截数据）
            tracing::error!("[forward] 消息序列化失败 act={} err={}", msg.act, e);
            return;
        }
    };

    let mut conn = redis.clone();
    // 一条 pipeline 完成：入队 → 取裁剪前长度 → 裁剪到上限 → 取裁剪后长度
    // （4 条命令 1 次往返；`ignore()` 的两条不占返回位）
    let mut pipe = redis::pipe();
    pipe.cmd("LPUSH").arg("device:reports").arg(&json_str).ignore();
    pipe.cmd("LLEN").arg("device:reports");
    pipe.cmd("LTRIM")
        .arg("device:reports")
        .arg(0)
        .arg(REPORTS_QUEUE_MAX - 1)
        .ignore();
    pipe.cmd("LLEN").arg("device:reports");

    let result: Result<(i64, i64), redis::RedisError> = pipe.query_async(&mut conn).await;

    match result {
        Ok((len_before_trim, len_after_trim)) => {
            if len_before_trim > len_after_trim {
                // 越界裁剪：明确告知丢弃了多少条（这些上报已不可恢复，需要人工关注消费端状态）
                tracing::warn!(
                    "[forward] device:reports 超过上限 {}，已裁剪最旧的 {} 条上报（长度 {} → {}）",
                    REPORTS_QUEUE_MAX,
                    len_before_trim - len_after_trim,
                    len_before_trim,
                    len_after_trim
                );
            } else {
                tracing::debug!(
                    "[forward] 已推送 dev_id={} act={} queue_len={}",
                    msg.dev_id.as_deref().unwrap_or("?"),
                    msg.act,
                    len_after_trim
                );
            }
        }
        Err(e) => tracing::error!("[forward] Redis LPUSH/LTRIM 失败: {}", e),
    }
}