charging-cabinet/software/device-server/src/main.rs

//! 充电柜矩阵系统 — 设备服务入口
//!
//! 轻量化 TCP 服务，只处理设备长连接通信和指令解析。
//! 不存储 DB、不验证业务逻辑，所有设备上报通过 Redis 转发给 API 服务。
//!
//! 启动流程：
//! 1. 连接 Redis
//! 2. 通过 Redis 注册到 API 服务（节点信息写入 Redis）
//! 3. 启动心跳协程（定期更新 Redis 中的心跳时间戳）
//! 4. 启动指令消费协程（从 Redis 读取平台指令并发送到设备）
//! 5. 启动 TCP 监听

mod config;
mod redis_conn;
mod tcp;

use std::time::Duration;

use config::Config;

#[tokio::main]
async fn main() {
    tracing_subscriber::fmt()
        .with_env_filter(
            tracing_subscriber::EnvFilter::try_from_default_env()
                .unwrap_or_else(|_| "info".into()),
        )
        .init();

    // 加载配置
    let cfg = Config::from_env();
    tracing::info!("配置加载完成 node_id={}", cfg.node_id);

    // 连接 Redis
    let redis_conn = redis_conn::init_connection(&cfg.redis_url)
        .await
        .expect("Redis 连接失败");
    tracing::info!("Redis 连接成功");

    // 初始化设备连接池
    let device_pool = tcp::connection::ConnectionPool::new();

    // 通过 Redis 注册节点信息
    register_node(&redis_conn, &cfg).await;

    // 启动心跳协程（通过 Redis 更新心跳时间戳）
    let hb_redis = redis_conn.clone();
    let hb_node_id = cfg.node_id.clone();
    let hb_interval = cfg.heartbeat_interval;
    let hb_ttl = cfg.register_ttl;
    tokio::spawn(async move {
        let mut interval = tokio::time::interval(Duration::from_secs(hb_interval));
        loop {
            interval.tick().await;
            heartbeat(&hb_redis, &hb_node_id, hb_ttl).await;
        }
    });

    // 启动指令消费协程（从 Redis 读取平台指令并发送到设备）
    let cmd_redis = redis_conn.clone();
    let cmd_pool = device_pool.clone();
    tokio::spawn(async move {
        consume_commands(cmd_redis, cmd_pool).await;
    });

    // 启动 TCP 监听
    let tcp_addr = format!("{}:{}", cfg.host, cfg.tcp_port);
    tracing::info!("TCP 设备服务启动于 {}", tcp_addr);
    tcp::server::run_tcp_server(tcp_addr, device_pool, redis_conn).await;
}

/// 通过 Redis 注册节点信息（HSET hash 结构，与 heartbeat 保持一致）
async fn register_node(redis: &redis::aio::ConnectionManager, cfg: &Config) {
    let key = format!("node:{}", cfg.node_id);
    let now = chrono::Utc::now().to_rfc3339();
    let mut conn = redis.clone();
    let result: Result<(), _> = redis::cmd("HSET")
        .arg(&key)
        .arg("node_id")
        .arg(&cfg.node_id)
        .arg("ip")
        .arg(&cfg.host)
        .arg("tcp_port")
        .arg(cfg.tcp_port.to_string())
        .arg("status")
        .arg("online")
        .arg("registered_at")
        .arg(&now)
        .arg("last_heartbeat")
        .arg(&now)
        .query_async(&mut conn)
        .await;

    // ponytail: HSET 后单独设置 TTL（redis-rs 的 HSET 不支持 EX 选项）
    let _: Result<(), _> = redis::cmd("EXPIRE")
        .arg(&key)
        .arg(cfg.register_ttl)
        .query_async(&mut conn)
        .await;

    match result {
        Ok(_) => tracing::info!("已通过 Redis 注册节点 node_id={}", cfg.node_id),
        Err(e) => tracing::error!("节点注册失败: {}", e),
    }
}

/// 通过 Redis 更新心跳时间戳，同时刷新 TTL
async fn heartbeat(redis: &redis::aio::ConnectionManager, node_id: &str, register_ttl: u64) {
    let key = format!("node:{}", node_id);
    let mut conn = redis.clone();
    let result: Result<(), _> = redis::cmd("HSET")
        .arg(&key)
        .arg("last_heartbeat")
        .arg(chrono::Utc::now().to_rfc3339())
        .query_async(&mut conn)
        .await;

    // ponytail: 心跳同时刷新 TTL，避免节点被过期清理
    let _: Result<(), _> = redis::cmd("EXPIRE")
        .arg(&key)
        .arg(register_ttl)
        .query_async(&mut conn)
        .await;

    match result {
        Ok(_) => tracing::debug!("心跳已更新 node_id={}", node_id),
        Err(e) => tracing::warn!("心跳更新失败: {}", e),
    }
}

/// 消费 Redis 指令队列 — 从统一队列 `device:commands` 读取平台指令并发送到设备
async fn consume_commands(
    redis: redis::aio::ConnectionManager,
    pool: tcp::connection::ConnectionPool,
) {
    tracing::info!("[cmd_consumer] 启动");

    loop {
        let mut conn = redis.clone();
        let result: Result<(String, String), redis::RedisError> = redis::cmd("BRPOP")
            .arg("device:commands")
            .arg(0)
            .query_async(&mut conn)
            .await;

        match result {
            Ok((_key, cmd_json)) => {
                // 从 JSON 中提取 dev_id 用于路由
                let dev_id = serde_json::from_str::<serde_json::Value>(&cmd_json)
                    .ok()
                    .and_then(|v| v.get("dev_id").and_then(|d| d.as_str()).map(String::from));

                let dev_id = match dev_id {
                    Some(id) if !id.is_empty() => id,
                    _ => {
                        tracing::warn!("[cmd_consumer] 指令缺少 dev_id，丢弃");
                        continue;
                    }
                };

                // 发送到设备
                match pool.send_command(&dev_id, &cmd_json).await {
                    Ok(()) => {
                        tracing::debug!("[cmd] 已发送到设备 dev_id={}", dev_id);
                    }
                    Err(e) => {
                        tracing::warn!("[cmd] 发送失败 dev_id={} err={}", dev_id, e);
                    }
                }
            }
            Err(e) => {
                tracing::error!("[cmd_consumer] BRPOP 错误: {}", e);
                tokio::time::sleep(Duration::from_secs(1)).await;
            }
        }
    }
}