# 嘉谷低空智控平台 — Go 后端设计文档 > 版本：v1.0 > 参考项目：pilot-train-server > MQTT 协议版本：v1（dock-edge） --- ## 1. 整体架构风格 **单体服务，内部分模块**，对齐 pilot-train-server 的代码风格： - **单一入口**：一个 `main.go`，一个 `go.mod`，内部分模块组织代码 - **Controller → Service → Model/VO** 分层，Service 直操 `common.DB`，无独立 Repository 层 - **包级单例**：`common.DB`、`common.RedisPool`、`service.DefaultXxxService` - **本地 YAML 配置**：一个 `config.yaml`，Viper 加载 - **JWT + Casbin RBAC**：`gorm-adapter` 持久化，Redis 缓存权限。admin 全量数据，user 按 `user_id` 过滤 - **统一响应** `{code, msg, data}` + 中文验证错误翻译 - **Redigo Redis Pool**（非 go-redis） --- ## 2. 技术选型 | 层面 | 选型 | 说明 | |---|---|---| | Web 框架 | **Gin v1.10** | 轻量高性能，完全对齐 pilot-train-server | | ORM | **GORM v1.25 + MySQL 驱动** | 完全对齐 pilot-train-server | | 数据库 | **MySQL 8.0** | 业务数据，完全对齐 pilot-train-server | | 时序数据库 | **TDengine 3.x** | 遥测/轨迹等时序数据，对齐 pilot-data 的用法 | | 缓存 | **Redis 7+（Redigo Pool）** | 设备实时状态、Session、权限、流量余额 | | MQTT Broker | **EMQX** | 设备通信，由甲方/边缘侧部署，后台只作为客户端连接 | | 对象存储 | **MinIO / 阿里云 OSS** | S3 兼容 API，视频和升级包存储 | | 直播 | **阿里云直播 + SRT 推流** | 后台生成签名推流地址，通过 MQTT 下发给工控机 | | 配置 | **Viper + config.yaml** | 本地配置文件 | | 认证 | **JWT HS256 + Casbin** | 无状态认证 + RBAC，7 天 access / 30 天 refresh | | 日志 | **Zap + Lumberjack** | 结构化 + 日志轮转，对齐 pilot-train-server | | 验证 | **go-playground/validator** | 中文翻译(`zh`) | | ID 生成 | **Snowflake** | `tool/snowflake.go` | | WebSocket | **gorilla/websocket** | 前端实时推送 | | 定时任务 | **robfig/cron** | 任务计划调度 | --- ## 3. 项目结构 ``` laic-backend/ ├── main.go # 入口，初始化 → MQTT + WS + HTTP ├── go.mod / go.sum ├── casbin.model ├── config.yaml # 本地配置文件 ├── Makefile │ ├── common/ # 基础设施 │ ├── gin.go # gin 引擎、统一响应 R、路由组 │ ├── mysql.go # GORM + OrgPlugin 多租户、Scopes │ ├── tdengine.go # TDengine 连接池 │ ├── redis.go # Redigo Pool、Lua 原子操作 │ ├── config.go # Viper 加载 config.yaml │ ├── casbin.go # Casbin 初始化 + Redis 缓存适配器 │ ├── busi_error.go # BusiError + 错误码常量 │ ├── auth_util.go # admin 全量 / user 按 user_id 过滤 │ ├── pageutil.go # 分页 │ └── validator.go # 自定义验证器（phone, password 等） │ ├── middleware/ # HTTP 中间件 │ ├── auth.go # JWT 校验 + Casbin 鉴权 │ ├── cros.go # CORS │ ├── trace.go # X-Trace-Id │ ├── default_log.go # 访问日志 │ ├── rate_limit.go # 限流 │ └── validate.go # 验证错误翻译 │ ├── cache/ # Redis Key 管理 │ ├── cache_key.go # 所有 Redis Key 集中定义 │ └── cache_value.go # 缓存值 DTO │ ├── token/ # JWT │ └── token.go # 生成/校验/刷新 │ ├── logger/ # 日志 │ ├── logger.go # 自定义 logger │ └── gorm_log.go # GORM logger 适配 │ ├── tool/ # 工具 │ ├── snowflake.go # 分布式 ID │ ├── geo.go # 地理计算（Haversine 距离） │ └── str.go / time.go # 字符串/时间工具 │ ├── model/ # GORM 数据模型 │ ├── user.go # User │ ├── dock.go # Dock, Drone │ ├── alarm.go # AlarmCode, Alarm │ ├── command.go # DeviceCommandLog │ ├── firmware.go # Firmware │ ├── task.go # TaskPlan, TaskExecution, WorkflowState │ ├── route.go # Route, RouteWaypoint │ ├── media.go # LiveSession, Video, DownloadLog │ └── billing.go # TrafficOrder, TrafficUsageLog, SimCard │ ├── vo/ # 请求/响应 DTO │ ├── user_vo.go │ ├── dock_vo.go │ ├── alarm_vo.go │ ├── task_vo.go │ ├── route_vo.go │ └── media_vo.go │ ├── handler/ # HTTP 处理器（Controller） │ ├── auth_handler.go # 登录/注册/短信/刷新 │ ├── user_handler.go # 用户 CRUD │ ├── account_handler.go # 账户中心/流量/SIM卡 │ ├── dock_handler.go # 机巢 CRUD/状态/指令/固件 │ ├── drone_handler.go # 无人机 CRUD/遥测 │ ├── alarm_handler.go # 告警列表/确认/关闭 │ ├── firmware_handler.go # 固件管理（admin） │ ├── task_handler.go # 任务 CRUD/执行 │ ├── execution_handler.go # 执行记录/轨迹 │ ├── route_handler.go # 航线 CRUD │ ├── live_handler.go # 直播管理 │ ├── video_handler.go # 视频/下载 │ └── system_handler.go # 系统管理 │ ├── service/ # 业务逻辑（struct + Default 单例） │ ├── auth_service.go │ ├── user_service.go │ ├── account_service.go │ ├── dock_service.go │ ├── drone_service.go │ ├── command_service.go # 指令构建/幂等/重试/超时 │ ├── alarm_service.go # alarmCodes diff → 告警生命周期 │ ├── firmware_service.go # 固件 CRUD + OTA 下发 │ ├── task_service.go │ ├── execution_service.go │ ├── route_service.go │ ├── live_service.go # 阿里云直播签名 + 推流会话 │ ├── video_service.go # OSS 签名下载 │ └── billing_service.go # 流量扣减 │ ├── mqtt/ # MQTT 客户端 │ ├── client.go # 连接管理（paho.mqtt.golang） │ ├── subscriber.go # 订阅设备上报 → Redis + 告警 diff │ └── publisher.go # 发布指令到设备 │ ├── websocket/ # WebSocket │ ├── hub.go # 连接池 + 广播 │ └── client.go # 单连接读写 │ ├── route/ # 路由注册 │ └── route.go # InitRouter │ └── sql/ # 数据库 DDL ├── 001_schema.sql # 表结构 ├── 002_alarm_codes.sql # 告警码初始化数据 └── 003_seed.sql # 种子数据 ``` --- ## 4. MQTT 接口设计 ### 4.1 Topic 结构（由边缘端协议定义） 基础前缀：`dock-edge/v1/dock/{dockId}` **上行（边缘 → 后台）：** | Topic | QoS | Retain | 说明 | |---|---:|---:|---| | `status/online` | 1 | 是 | 在线/离线/软件版本/bootId | | `state/dock` | 1 | 是 | 机巢当前完整状态 + alarmCodes | | `state/drone` | 1 | 是 | 无人机当前完整状态 + alarmCodes | | `state/workflow` | 1 | 是 | 任务/一键流程执行步骤 | | `state/video` | 1 | 是 | 图传和推流状态 | | `telemetry` | 0 | 否 | 1Hz 高频遥测（位置/姿态/电池/GPS） | | `command/ack` | 1 | 否 | 指令应答 | | `ota/reported` | 1 | 是 | 升级进度 | **下行（后台 → 边缘）：** | Topic | QoS | Retain | 说明 | |---|---:|---:|---| | `command` | 1 | 否 | 机巢/无人机/工作流/视频控制指令 | | `ota/desired` | 1 | 是 | 升级任务 | ### 4.2 后台 MQTT 客户端职责 ``` 订阅： dock-edge/v1/dock/+/status/online → 发现新设备 + 感知上下线 dock-edge/v1/dock/+/state/dock → Redis 状态 + alarmCodes diff dock-edge/v1/dock/+/state/drone → Redis 状态 + alarmCodes diff dock-edge/v1/dock/+/state/workflow → 任务步骤跟踪 dock-edge/v1/dock/+/state/video → 直播状态跟踪 dock-edge/v1/dock/+/telemetry → Redis 最新遥测 + TDengine 批量写入（500条/10s） dock-edge/v1/dock/+/command/ack → 更新 device_command_log dock-edge/v1/dock/+/ota/reported → 升级进度 发布： dock-edge/v1/dock/{dockId}/command → 设备控制指令 dock-edge/v1/dock/{dockId}/ota/desired → 升级任务 ``` ### 4.3 指令下发流程 ``` POST /v1/docks/:id/command {type:"workflow.one_key_takeoff"} → command_service: 1. 生成 commandId（Snowflake） 2. 生成 requestId（UUID） 3. 校验设备在线 + 安全条件 4. INSERT device_command_log(id=commandId, status='sent') 5. MQTT Publish → EMQX → 工控机 → 工控机: 校验 → command/ack {accepted:true, resultCode:"OK"} → mqtt/subscriber 收到 ACK → 更新 device_command_log(status='acked') → 工控机: 逐步骤发布 state/workflow → mqtt/subscriber → WS 推送前端 超时重试（30s 未 ACK）： → command_service: 生成新 requestId，复用 commandId → 工控机: commandId 幂等，不重复执行，返回原 ACK ``` ### 4.4 告警处理流程 ``` 工控机 state/dock {alarmCodes:["LEFT_DOOR_POWER_FAULT","DOCK_EMERGENCY_STOP"]} → mqtt/subscriber: 解析 → Redis HMSET dock:{dockId}:alarm_codes → alarm_service: 1. 对比上次 alarmCodes（从 Redis 读取） 2. 新增 code → INSERT alarm(status='active') 3. 移除 code → UPDATE alarm SET status='resolved', resolved_at=now() 4. WS 推送告警变更给前端 ``` --- ## 5. 数据模型 ### 5.1 用户 两个角色：**管理员（admin）** 看全部数据，**普通用户（user）** 只看自己的机巢。 ```sql CREATE TABLE user ( id BIGINT PRIMARY KEY, name VARCHAR(32) NOT NULL, phone VARCHAR(32) NOT NULL, email VARCHAR(64), password VARCHAR(64) NOT NULL, role VARCHAR(16) DEFAULT 'user', -- admin / user traffic_balance BIGINT DEFAULT 0, -- 云媒体剩余流量(字节) status TINYINT DEFAULT 1, last_login DATETIME, created_at DATETIME, updated_at DATETIME, UNIQUE KEY idx_phone (phone) ); ``` - 用户自行注册，默认角色 `user` - 管理员在系统初始化时手动指定，不需要组织/租户概念 - Casbin 策略：`admin` 全量数据，`user` 按 `user_id` 过滤自己的机巢 ### 5.2 机巢 & 无人机 机巢与无人机是**默认一对一绑定**的关系。工控机上报机巢状态时携带当前无人机信息，后台自动关联。 ```sql CREATE TABLE dock ( id BIGINT PRIMARY KEY, user_id BIGINT NOT NULL, -- 所属用户 dock_id VARCHAR(64) NOT NULL, -- MQTT dockId (dock-xxx) name VARCHAR(64) DEFAULT '', code VARCHAR(32) DEFAULT '', sn VARCHAR(64) DEFAULT '', iccid VARCHAR(32) DEFAULT '', -- 冗余，权威见 sim_card.iccid longitude DOUBLE DEFAULT 0, latitude DOUBLE DEFAULT 0, altitude DOUBLE DEFAULT 0, location VARCHAR(128) DEFAULT '', status VARCHAR(32) DEFAULT 'offline', register_status VARCHAR(16) DEFAULT 'pending', -- pending/registered dock_id_source VARCHAR(32) DEFAULT '', software_ver VARCHAR(16) DEFAULT '', protocol_ver VARCHAR(8) DEFAULT '1.0', created_at DATETIME, updated_at DATETIME, UNIQUE KEY idx_dock_id (dock_id) ); CREATE TABLE drone ( id BIGINT PRIMARY KEY, user_id BIGINT NOT NULL, -- 所属用户（继承 dock 的 user_id） drone_sn VARCHAR(32) NOT NULL, -- 飞控 HW_SN_NUM dock_id VARCHAR(64) NOT NULL, -- 绑定的机巢 dock_id（1:1） name VARCHAR(64) DEFAULT '', code VARCHAR(32) DEFAULT '', model VARCHAR(32) DEFAULT '', status VARCHAR(32) DEFAULT 'offline', battery INT DEFAULT 0, firmware_ver VARCHAR(16) DEFAULT '', created_at DATETIME, updated_at DATETIME, UNIQUE KEY idx_drone_sn (drone_sn), UNIQUE KEY idx_drone_dock (dock_id) ); ``` > 机巢上报 `state/drone` 时携带 `droneSn`，后台据此自动创建或更新无人机记录。`dock_id` 上唯一约束保证一对一绑定。换机巢时新 dock 上报不同 droneSn 即建立新关系，无需人工操作。 ### 5.3 告警 通过 `dock_id` 关联到用户，不单独存 `user_id`。 ```sql CREATE TABLE alarm_code ( id BIGINT PRIMARY KEY, code VARCHAR(64) NOT NULL, category VARCHAR(32) NOT NULL, -- dock / drone message_cn VARCHAR(256) NOT NULL, level VARCHAR(16) DEFAULT 'critical', source VARCHAR(128) DEFAULT '', created_at DATETIME, updated_at DATETIME, UNIQUE KEY idx_code (code) ); CREATE TABLE alarm ( id BIGINT PRIMARY KEY, dock_id VARCHAR(64) NOT NULL, code VARCHAR(64) NOT NULL, device_type VARCHAR(8) NOT NULL, message_cn VARCHAR(256) NOT NULL, level VARCHAR(16) NOT NULL, status VARCHAR(16) DEFAULT 'active', triggered_at DATETIME, resolved_at DATETIME, created_at DATETIME, KEY idx_alarm_dock (dock_id, status) ); ``` ### 5.4 指令 & 工作流 ```sql CREATE TABLE device_command_log ( id BIGINT PRIMARY KEY, -- = commandId (Snowflake) dock_id VARCHAR(64) NOT NULL, command_type VARCHAR(64) NOT NULL, -- dock.open / drone.takeoff / workflow.start_task params JSON, drone_sn VARCHAR(32), request_id VARCHAR(64), ttl_ms INT DEFAULT 30000, ack_accepted TINYINT, ack_result_code VARCHAR(64), status VARCHAR(16) DEFAULT 'sent', -- sent / acked / timeout / terminal retry_count INT DEFAULT 0, sent_at DATETIME, acked_at DATETIME, created_at DATETIME, KEY idx_cmd_log_dock (dock_id, created_at) ); CREATE TABLE workflow_state ( id BIGINT PRIMARY KEY, dock_id VARCHAR(64) NOT NULL, command_id VARCHAR(64), type VARCHAR(64), task_id VARCHAR(128), mission_id VARCHAR(128), state VARCHAR(16) NOT NULL, -- idle / running / succeeded / failed / cancelled step VARCHAR(64) NOT NULL, result_code VARCHAR(64), updated_at DATETIME ); ``` ### 5.5 固件管理 管理员上传和管理固件版本，用户选择版本下发 OTA 升级指令。 ```sql CREATE TABLE firmware ( id BIGINT PRIMARY KEY, component VARCHAR(32) NOT NULL, -- 固定 dock-edge-agent version VARCHAR(32) NOT NULL, -- 1.3.0 description VARCHAR(512) DEFAULT '', -- 版本说明/更新日志 file_url VARCHAR(512) NOT NULL, -- 升级包下载地址 sha256 VARCHAR(128) NOT NULL, -- SHA-256 摘要 signature TEXT NOT NULL, -- 签名 Base64 file_size BIGINT DEFAULT 0, -- 文件大小(字节) mandatory TINYINT DEFAULT 0, -- 是否强制升级 status VARCHAR(16) DEFAULT 'draft', -- draft/released/deprecated created_by BIGINT NOT NULL, -- 上传者(admin) created_at DATETIME, updated_at DATETIME, UNIQUE KEY idx_component_ver (component, version) ); ``` ### 5.6 任务 & 航线 ```sql CREATE TABLE task_plan ( id VARCHAR(128) PRIMARY KEY, user_id BIGINT NOT NULL, name VARCHAR(128) NOT NULL, dock_id VARCHAR(64) NOT NULL, route_id BIGINT, schedule_type VARCHAR(16) DEFAULT 'once', schedule_cron VARCHAR(32), video_policy VARCHAR(16) DEFAULT 'raw', status VARCHAR(16) DEFAULT 'draft', created_by BIGINT, created_at DATETIME, updated_at DATETIME ); CREATE TABLE task_execution ( id BIGINT PRIMARY KEY, task_id VARCHAR(128), command_id VARCHAR(64), dock_id VARCHAR(64) NOT NULL, drone_sn VARCHAR(32), start_time DATETIME, end_time DATETIME, status VARCHAR(16) DEFAULT 'pending', trajectory_json JSON, created_at DATETIME ); CREATE TABLE route ( id BIGINT PRIMARY KEY, user_id BIGINT NOT NULL, name VARCHAR(128) NOT NULL, description VARCHAR(256), created_at DATETIME, updated_at DATETIME ); CREATE TABLE route_waypoint ( id BIGINT PRIMARY KEY, route_id BIGINT NOT NULL, seq INT NOT NULL, longitude DOUBLE, latitude DOUBLE, altitude DOUBLE, speed DOUBLE, yaw DOUBLE, hold_sec INT DEFAULT 0, created_at DATETIME, UNIQUE KEY uk_route_seq (route_id, seq) ); ``` ### 5.7 媒体 ```sql CREATE TABLE live_session ( id VARCHAR(128) PRIMARY KEY, -- streamSessionId dock_id VARCHAR(64) NOT NULL, provider VARCHAR(16) DEFAULT 'aliyun', stream_name VARCHAR(128) NOT NULL, -- 直播流名（每次会话唯一，生成推流/播放地址的入参） push_url_hash VARCHAR(128), expires_at BIGINT NOT NULL, max_bitrate_bps BIGINT DEFAULT 1500000, phase VARCHAR(16) DEFAULT 'idle', stop_reason VARCHAR(32), error_code VARCHAR(64), started_at DATETIME, stopped_at DATETIME, created_at DATETIME ); CREATE TABLE video ( id BIGINT PRIMARY KEY, user_id BIGINT NOT NULL, drone_sn VARCHAR(32), execution_id BIGINT, file_name VARCHAR(256), file_size BIGINT, duration INT, oss_key VARCHAR(256), oss_bucket VARCHAR(128), -- 预签名上传的目标 bucket thumbnail_key VARCHAR(256), status VARCHAR(16) DEFAULT 'uploading', upload_expire_at DATETIME, -- 预签名 PUT URL 过期时间 created_at DATETIME ); CREATE TABLE download_log ( id BIGINT PRIMARY KEY, video_id BIGINT NOT NULL, user_id BIGINT NOT NULL, bytes BIGINT NOT NULL, created_at DATETIME ); ``` ### 5.8 计费 **商业模式：中间商赚差价。** - **云媒体流量**：平台自己有一个阿里云直播账号和流量包，所有用户共用。用户向平台订购流量（GB），平台在内部给该用户记账。用户消耗流量时从自己的余额扣，平台只需关注所有用户的**总消耗**，快用完时再向阿里云续购。平台赚取用户售价与阿里云成本之间的差价。 - **4G 卡流量**：用户为机巢的 SIM 卡订购套餐，付款后平台调用运营商接口给该卡充值，差价同上。 两类计费相互独立，付款成功后的处理： - **云媒体流量**：付款 → `user.traffic_balance` 加字节数，用户即刻可用 - **4G 卡流量**：付款 → 调运营商接口充值 → 更新 `sim_card.plan_gb` ```sql -- user 表流量字段 -- traffic_balance BIGINT DEFAULT 0 -- 云媒体剩余流量（字节） CREATE TABLE traffic_order ( id BIGINT PRIMARY KEY, user_id BIGINT NOT NULL, amount_gb INT NOT NULL, -- 购买流量(GB) unit_price DECIMAL(6,2), -- 平台售价(元/GB) total_price DECIMAL(10,2), -- 用户实付(元) pay_status VARCHAR(16) DEFAULT 'unpaid', -- unpaid/paid/cancelled created_at DATETIME, paid_at DATETIME ); -- 云媒体消费明细（直播/回放/下载） CREATE TABLE traffic_usage_log ( id BIGINT PRIMARY KEY, user_id BIGINT NOT NULL, source_type VARCHAR(16), -- live / replay / download source_id BIGINT, bytes_used BIGINT, balance_before BIGINT, -- 扣减前余额(字节) balance_after BIGINT, -- 扣减后余额(字节) created_at DATETIME ); -- SIM 卡（用户为机巢的 4G 卡订购套餐） CREATE TABLE sim_card ( id BIGINT PRIMARY KEY, user_id BIGINT NOT NULL, dock_id VARCHAR(64), carrier VARCHAR(16), -- 移动/联通/电信 phone VARCHAR(16), iccid VARCHAR(32), -- 权威字段（查询/充值的入参） plan_gb INT, -- 当前套餐总流量(GB) used_gb DECIMAL(10,4) DEFAULT 0, -- 已用流量(GB，由同步任务回填) expired_at DATE, status VARCHAR(16) DEFAULT 'active', -- 平台业务态: active/expired carrier_status VARCHAR(16) DEFAULT 'normal', -- 运营商侧状态: normal/suspended/arrears/cancelled last_sync_at DATETIME, -- 上次从运营商同步用量时间 created_at DATETIME, updated_at DATETIME ); -- SIM 卡充值记录 CREATE TABLE sim_recharge_log ( id BIGINT PRIMARY KEY, sim_card_id BIGINT NOT NULL, user_id BIGINT NOT NULL, amount_gb INT NOT NULL, unit_price DECIMAL(6,2), -- 平台售价(元/GB) total_price DECIMAL(10,2), -- 用户实付(元) pay_status VARCHAR(16) DEFAULT 'paid', recharge_status VARCHAR(16) DEFAULT 'pending', -- pending/success/failed recharge_msg VARCHAR(256) DEFAULT '', -- 运营商返回信息 idempotent_key VARCHAR(64) NOT NULL, -- 平台充值单号，超时重试去重 created_at DATETIME, UNIQUE KEY uk_recharge_idem (idempotent_key) ); -- SIM 卡用量快照（每次从运营商同步留痕，可审计/画曲线/对账） CREATE TABLE sim_usage_record ( id BIGINT PRIMARY KEY, sim_card_id BIGINT NOT NULL, plan_gb INT, -- 同步时套餐总量(GB) used_gb DECIMAL(10,4), -- 已用(GB) remain_gb DECIMAL(10,4), -- 剩余(GB) carrier_status VARCHAR(16), synced_at DATETIME, -- 同步时间 KEY idx_record_sim (sim_card_id, synced_at) ); ``` > 平台采购成本（从阿里云/运营商的拿货价）不在本系统记录，由财务线下管理。系统只记录面向用户的售价和消费。 > **余额一致性与账本模型**：流量余额以 Redis 为唯一实时真相（source of truth），MySQL 的 `user.traffic_balance` 仅作为 30 分钟刷一次的慢快照。`traffic_usage_log` 流水是**账本**（ledger），每次扣减同步写 MySQL、不可变、可重放；余额是**派生值**（derived balance），可由 `快照 + 重放流水 + 订单` 精确重建。因此 Redis 丢失最多损失 30 分钟余额变动，但账本仍在，可对账、可审计、可重建。 > **SIM 卡用量来源**：工控机不上报网络用量，`used_gb` / `expired_at` / `carrier_status` 均由后台定时任务（每 1 小时）调用运营商查询接口回填。`sim_card.iccid` 为权威字段（查询与充值的入参），`dock.iccid`（§5.2）降级为冗余。运营商接口抽象为 `CarrierAPI`（`QueryUsage` / `Recharge`），暂定一家运营商，先做「接口 + 单实现」，换卡商时不改业务层。 ### 5.9 TDengine 时序数据 遥测和轨迹等高频时序数据存入 TDengine，与 pilot-data 的用法保持一致。 **遥测超级表：** ```sql CREATE STABLE device_telemetry ( ts TIMESTAMP, longitude DOUBLE, latitude DOUBLE, altitude DOUBLE, ground_speed DOUBLE, roll DOUBLE, pitch DOUBLE, yaw DOUBLE, battery_pct INT, battery_v DOUBLE, satellites INT, gps_quality VARCHAR(16), link_quality INT, flight_mode VARCHAR(16), armed TINYINT ) TAGS ( dock_id VARCHAR(64), drone_sn VARCHAR(32) ); ``` **轨迹超级表（任务回放用）：** ```sql CREATE STABLE task_trajectory ( ts TIMESTAMP, longitude DOUBLE, latitude DOUBLE, altitude DOUBLE, ground_speed DOUBLE, yaw DOUBLE, battery_pct INT ) TAGS ( execution_id BIGINT, dock_id VARCHAR(64), drone_sn VARCHAR(32) ); ``` > pilot-data 使用 `TrackService` 缓冲（`sync.Map`）达到 500 条或 10 秒后批量写入 TDengine。本项目对齐此模式。 --- ## 6. Redis 缓存设计 ``` # 设备实时状态 device:dock:{dockId}:status → Hash (所有 state/dock 字段) device:dock:{dockId}:alarm_codes → Set (当前告警编码集合) device:drone:{dockId}:telemetry → Hash (所有 state/drone 字段) device:drone:{dockId}:alarm_codes → Set (当前告警编码集合) device:workflow:{dockId} → Hash (当前工作流状态) device:video:{dockId} → Hash (当前视频推流状态) # Token & Session laic:token:{user}:{shortId} → token 会话标记（支持踢人） # 权限缓存 laic:permission:{role} → 角色权限列表 # 云媒体流量余额（以 Redis 为主，user.traffic_balance 为 30 分钟快照） laic:traffic:{userId} → String (INT 值，可原子 DECRBY/INCRBY) # 在线设备集合 laic:devices:online:dock → Set {dockId...} laic:devices:online:drone → Set {droneSn...} ``` --- ## 7. API 路由 所有端点统一注册在一个 Gin router 上，`api-port` 默认 `8080`。 ### 7.1 认证（公开） ``` POST /v1/auth/login POST /v1/auth/logout POST /v1/auth/refresh POST /v1/auth/sms-code POST /v1/auth/register POST /v1/auth/reset-password ``` ### 7.2 账户中心（需登录） ``` GET /v1/account/profile PUT /v1/account/profile GET /v1/account/traffic/balance GET /v1/account/traffic/usage GET /v1/account/traffic/orders POST /v1/account/traffic/orders GET /v1/account/sim-cards POST /v1/account/sim-cards/:id/recharge ``` ### 7.3 机巢 & 无人机 ``` GET /v1/docks # 列表（支持筛选/分页） POST /v1/docks # 完善待登记机巢信息 GET /v1/docks/:id # 详情（含实时状态 from Redis） PUT /v1/docks/:id DELETE /v1/docks/:id GET /v1/docks/:id/status # 纯实时状态 POST /v1/docks/:id/command # 下发指令 POST /v1/docks/:id/firmware/upgrade # body: {firmware_id} GET /v1/drones GET /v1/drones/:id PUT /v1/drones/:id DELETE /v1/drones/:id GET /v1/drones/:id/telemetry ``` ### 7.4 告警 & 指令日志 ``` GET /v1/alarms # 告警列表 PUT /v1/alarms/:id/acknowledge PUT /v1/alarms/:id/resolve GET /v1/commands # 指令历史 GET /v1/commands/:id ``` ### 7.5 任务 & 航线 ``` GET /v1/tasks POST /v1/tasks GET /v1/tasks/:id PUT /v1/tasks/:id DELETE /v1/tasks/:id POST /v1/tasks/:id/execute GET /v1/executions GET /v1/executions/:id GET /v1/executions/:id/trajectory GET /v1/routes POST /v1/routes GET /v1/routes/:id PUT /v1/routes/:id DELETE /v1/routes/:id ``` ### 7.6 媒体 ``` GET /v1/live POST /v1/live/:dockId/start POST /v1/live/:dockId/stop GET /v1/live/:dockId/play-url POST /v1/videos # 申请上传（返回 OSS 预签名 PUT URL） POST /v1/videos/:id/complete # 上传完成确认（边缘侧回调） GET /v1/videos GET /v1/videos/:id POST /v1/videos/:id/download ``` ### 7.7 系统管理（需 admin） ``` GET /v1/users POST /v1/users PUT /v1/users/:id DELETE /v1/users/:id GET /v1/roles GET /v1/logs/operation ``` ### 7.8 固件管理（需 admin） ``` GET /v1/firmwares # 固件版本列表 POST /v1/firmwares # 上传新固件 PUT /v1/firmwares/:id # 编辑（状态/强制标记等） DELETE /v1/firmwares/:id # 删除 ``` ### 7.9 WebSocket ``` WS /v1/ws/monitor # 前端实时监控（设备状态 + 告警推送） ``` ### 7.10 统一响应格式 ```json { "code": 200, "msg": "ok", "data": {} } ``` --- ## 8. 核心数据流 ### 8.1 设备自发现 & 登记 ``` 工控机首次启动 → 生成 dockId → 连接 EMQX → 发布 retained status/online → mqtt/subscriber: 1. 查询 dock 表 WHERE dock_id = {dockId} 2. 不存在 → INSERT dock(dock_id, register_status='pending', user_id=0) 3. 更新 Redis: device:dock:{dockId}:status 4. WS 推送: 新设备上线 → 用户在前端认领或管理员分配: PUT /v1/docks/:id {user_id, name, code, ...} ``` ### 8.2 无人机自动关联（默认 1:1 绑定） ``` 工控机上报 state/drone {droneSn:"23001582", online:true, ...} → mqtt/subscriber: 1. 查询 drone 表 WHERE drone_sn = "23001582" OR dock_id = {上报dockId} 2. 都不存在 → INSERT drone(drone_sn, dock_id, user_id=同dock的user_id) 3. 存在 → UPDATE drone SET drone_sn, dock_id（替换绑定） 4. Redis 更新状态 5. WS 推送 ``` ### 8.3 固件升级 ``` 管理员 POST /v1/firmwares {version, file_url, sha256, signature, ...} → 上传并保存固件版本 用户 POST /v1/docks/:id/firmware/upgrade {firmware_id} → dock_service: 1. 校验机巢属于当前用户 2. 查询 firmware 表获取版本信息 3. MQTT Publish → dock-edge/v1/dock/{dockId}/ota/desired {updateId, component, version, url, sha256, signature, mandatory, issuedAt} → 工控机: ota/reported 上报升级进度 → mqtt/subscriber → 更新固件升级状态 → WS 推送 ``` ### 8.4 云媒体流量计费 **扣减场景**：直播推流（按推流时长 × 码率估算）、回放、原始视频下载。 ``` POST /v1/videos/:id/download → video_service: 1. 查询 video → 获取 file_size(字节), user_id 2. ensureBalance(userId): key 不存在则从 MySQL 读余额 SetNX 初始化（兜底；启动时 WarmUp 已预热） 3. Redis Lua 脚本: GET 余额 → 校验 ≥ file_size → DECRBY 余额不足 → 返回错误 "流量不足，请充值" 4. 同步 INSERT traffic_usage_log(bytes_used, balance_before, balance_after) ← 账本，必须落库 5. 生成 OSS 签名 URL（有效期 300s） 6. INSERT download_log 7. 返回签名 URL 直播扣减（后台定时任务）: → 每 60 秒扫描进行中的 live_session → 按 elapsed × max_bitrate_bps / 8 估算流量 → 批量扣减 Redis 余额（Lua），并同步写 usage_log → 余额耗尽 → MQTT video.stop_stream 停止推流 余额快照刷新（后台定时任务，30 分钟）: → MGET 批量读 Redis 余额 → 批量 UPDATE user.traffic_balance（CASE WHEN 逐行赋值，避免逐条 roundtrip） → 不清 Redis（Redis 仍是唯一实时真相） → 优雅退出时强制 flush 一次，避免丢失最后窗口 ``` ### 8.5 充值与发放 ``` 用户 POST /v1/account/traffic/orders {amount_gb, ...} → 生成 traffic_order(status=unpaid) → 用户完成付款（对接支付接口，当前先人工标记 paid） 付款成功后 — 云媒体流量: → account_service: 1. Redis INCRBY user.traffic_balance (amount_gb × 1024³) ← 充值走同一 Redis 真相 2. UPDATE traffic_order SET pay_status='paid', paid_at=now() ← 订单实时落库，可重放 3. 用户立即可用 付款成功后 — 4G 卡: → account_service: 1. 生成 idempotent_key（= 平台充值单号） 2. 调用运营商充值 API(iccid, amount_gb, idempotent_key) 3. 运营商返回成功 → UPDATE sim_card SET plan_gb = plan_gb + amount_gb 4. INSERT sim_recharge_log(idempotent_key, recharge_status='success') 5. 运营商返回失败 → recharge_status='failed'，告警管理员人工处理（不同步更新 plan_gb） 6. 超时重试 → 复用同一 idempotent_key，运营商侧去重，不重复充值 ``` ### 8.6 SIM 卡用量同步 ``` 定时任务（robfig/cron，每 1 小时）: → 分页扫描 status='active' 的 sim_card → 按 carrier 路由到 CarrierAPI.QueryUsage(iccid) → 回填 sim_card: used_gb / expired_at / carrier_status / last_sync_at → INSERT sim_usage_record(plan_gb, used_gb, remain_gb, carrier_status, synced_at) ← 留痕 → 阈值检查: remain_gb < plan_gb * 10% → 提醒用户充值 expired_at 距今 < 3 天 → 提醒续费 carrier_status = suspended/arrears → 告警管理员 carrier_status = cancelled 或已过期 → 置 status='expired' 限流: 串行 + 可控并发 + 失败退避，避免打爆运营商查询接口 ``` ### 8.7 直播地址生成与按需拉流 **按需拉流** = 控制推流起停：用户不点播放不推流（省流量/省电），点了才开始推。 ``` 用户点「播放」 → GET /v1/live/:dockId/start → live_service: 1. 生成唯一 stream_name（= live_session.id，Snowflake/UUID） 2. 用阿里云直播 Go SDK 在服务端计算推流地址（RTMP/SRT + 鉴权，有效期 2h） 3. INSERT live_session(stream_name, phase='starting') 4. MQTT 下发 video.start_stream {push_url} 给工控机 → 工控机: 用 push_url 向阿里云推流 → MQTT 上报 state/video {streaming:true} → live_service: 计算播放地址（HLS/FLV + URL 鉴权）→ 返回 {play_url} → 前端: Aliplayer / hls.js / flv.js 拉 play_url 播放 用户点「关闭」 → POST /v1/live/:dockId/stop → live_service: MQTT 下发 video.stop_stream → 工控机停止推流 → 更新 live_session(phase='stopped') ``` **地址格式（阿里云直播）：** - 推流：`rtmp://{推流域名}/live/{AppName}/{stream_name}?auth_key=...`（或 SRT） - 播放：HLS `http://{播流域名}/live/{AppName}/{stream_name}.m3u8` / FLV `.flv`（低延迟） **关键约束：** - 推流域名、播流域名需在阿里云控制台备案 + CNAME，提前申请 - AccessKey 只在后台，设备只拿「这一次」的推流地址，绝不下发长期凭证 - 网页播放推荐 HLS（兼容最好），对延迟敏感再上 HTTP-FLV - 服务端集成用**阿里云直播 Go SDK**（视频直播 OpenAPI + 鉴权计算），前端用阿里云播放器 SDK（Aliplayer） --- ## 9. 启动流程 ```go func main() { // 1. 初始化日志 logger.InitCustomLog(logger.Level_D, nil, "laic-backend") // 2. 加载 config.yaml conf := &AppConfig{} common.LoadConfig("config.yaml", conf) // 3. 初始化基础设施 common.InitConnection(&conf.Mysql) // GORM + MySQL common.InitTDengine(&conf.TDengine) // TDengine 连接池 common.InitRedis(conf.Redis) // Redis Pool common.InitCasbinEnforcer() // Casbin RBAC // 4. 预热缓存 go cache.WarmUp() // 5. 连接 MQTT Broker go mqtt.Connect(conf.MQTT) // 6. 启动 WebSocket Hub go websocket.Hub.Run() // 7. 启动 HTTP route.InitRouter(conf.ApiPort) common.ServerRun(conf.ApiPort) // 8. 优雅退出 quit := make(chan os.Signal, 1) signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM) <-quit } ``` `config.yaml` 示例： ```yaml api-port: 8080 mysql: url: root:password@tcp(127.0.0.1:3306)/laic?charset=utf8mb4&parseTime=True&loc=Local idle: 10 max-conn: 50 max-wait: 3600 tdengine: dsn: root:taosdata@tcp(127.0.0.1:6030)/laic redis: redis://127.0.0.1:6379/0 mqtt: broker: tcp://127.0.0.1:1883 client-id: laic-backend username: laic password: "" log: level: debug path: ./logs/ ``` --- ## 10. 与 pilot-train-server 的主要差异 | 维度 | pilot-train-server | laic-backend | |---|---|---| | 架构 | 多服务单体仓库（5 个 main.go） | **单服务分模块（1 个 main.go）** | | 数据库 | MySQL | MySQL（相同） | | 时序数据 | TDengine | TDengine（相同） | | 设备协议 | MAVLink over TCP | **MQTT**（EMQX broker） | | 对象存储 | 华为云 OBS | **MinIO / 阿里云 OSS**（S3 兼容） | | 直播 | 无 | **阿里云直播 SRT 推流** | | 计费 | 无 | **流量余额 + 订单 + 消费明细** | | 告警 | 无 | **alarmCodes diff 引擎 + 告警生命周期** | | 指令体系 | 简单 TCP 指令 | **commandId 幂等 + TTL + ACK + 重试** | | 配置 | Consul | **本地 config.yaml** | | 服务间通信 | HTTP + Consul | **内部函数调用** | --- ## 11. 待确认项 1. **TDengine 部署**：需确认 TDengine 版本（pilot-train-server 用的 3.x），以及遥测数据保留周期 2. **告警码初始化**：`alarm_code` 表需要根据 MQTT 文档 7.6 节全部枚举值初始化，共约 90+ 条 3. **SIM 卡流量**：MQTT 文档明确工控机不上报网络用量，需确认运营商 API 或人工录入 4. **视频上传通路**：MQTT 协议只定义了推流，原始视频如何从工控机到达 OSS 需确认（可能由边缘侧直接上传 OSS，通过 state/video 或独立接口通知后台）