feat:新增MQTT报文管理功能

service/.env

@@ -35,6 +35,13 @@ FWH_TEST_MQTT_CA="./cert/test/AmazonRootCA1.pem"
 FWH_TEST_MQTT_CERT="./cert/test/01.certificate.pem.crt"
 FWH_TEST_MQTT_KEY="./cert/test/01.private.pem.key"
 
+# FWH ES (iotcore-stat Elasticsearch)
+FWH_ES_HOST="alb-prod.franklinwh.click:5603"
+FWH_ES_USER="iotcore_readonly"
+FWH_ES_PASSWORD="nv_psM2V&9&z"
+FWH_ES_INDEX_PREFIX="iotcore-stat"
+FWH_ES_INSECURE_SKIP_VERIFY=true
+
 # PLANE CONFIG
 PLANE_HOST="http://localhost:8600"
 PLANE_API_KEY="plane_api_xxxxx"

service/config/config.go

@@ -36,6 +36,20 @@ type FwhMqttConfig struct {
 	FwhMqttKeyUrl  string
 }
 
+// FwhEsConfig Elasticsearch(iotcore-stat) 连接配置
+type FwhEsConfig struct {
+	// EsHost 形如 "alb-prod.franklinwh.click:5603"（不含 scheme）
+	EsHost string
+	// EsUser / EsPassword basic auth 凭证
+	EsUser     string
+	EsPassword string
+	// EsIndexPrefix 索引前缀，形如 "iotcore-stat"，
+	// 实际索引名会拼接日期，如 "iotcore-stat-2026.09.07"
+	EsIndexPrefix string
+	// EsInsecureSkipVerify 是否跳过 TLS 证书校验（对应 curl -k）
+	EsInsecureSkipVerify bool
+}
+
 type PlaneApiConfig struct {
 	Host           string
 	ApiKey         string
@@ -77,6 +91,7 @@ type Config struct {
 	FwhTestApi	FwhApiConfig
 	FwhMqtt     FwhMqttConfig
 	FwhTestMqtt FwhMqttConfig
+	FwhEs       FwhEsConfig
 	PlaneApi    PlaneApiConfig
 	PlaneWebApi PlaneWebApiConfig
 	Kiro        KiroConfig
@@ -114,6 +129,9 @@ func LoadConfig(configPaths ...string) (*Config, error) {
 	v.SetDefault("FWH_OSS", "oss.franklinwh.com")
 	// FWH MQTT
 
+	// FWH ES(iotcore-stat)
+	v.SetDefault("FWH_ES_INDEX_PREFIX", "iotcore-stat")
+	v.SetDefault("FWH_ES_INSECURE_SKIP_VERIFY", true)
 	// 多文件路径支持：优先使用调用方传入的路径，否则回退默认
 	paths := configPaths
 	if len(paths) == 0 {
@@ -191,6 +209,13 @@ func LoadConfig(configPaths ...string) (*Config, error) {
 			FwhMqttCertUrl: v.GetString("FWH_TEST_MQTT_CERT"),
 			FwhMqttKeyUrl:  v.GetString("FWH_TEST_MQTT_KEY"),
 		},
+		FwhEs: FwhEsConfig{
+			EsHost:               v.GetString("FWH_ES_HOST"),
+			EsUser:               v.GetString("FWH_ES_USER"),
+			EsPassword:           v.GetString("FWH_ES_PASSWORD"),
+			EsIndexPrefix:        v.GetString("FWH_ES_INDEX_PREFIX"),
+			EsInsecureSkipVerify: v.GetBool("FWH_ES_INSECURE_SKIP_VERIFY"),
+		},
 		PlaneApi: PlaneApiConfig{
 			Host:           v.GetString("PLANE_HOST"),
 			ApiKey:         v.GetString("PLANE_API_KEY"),

service/controllers/mqttDataManageControllers.go

@@ -0,0 +1,199 @@
+/*
+ * MQTT 报文管理 API List（数据来自 iotcore-stat Elasticsearch，站点即 equipNo）
+ *
+ * 看板：
+ *   /mqttDataManage/total/daily      - GET 日总量      query: date=yyyy.MM.dd
+ *   /mqttDataManage/total/monthly    - GET 月总量      query: year, month
+ *   /mqttDataManage/total/dailyTrend - GET 月度每日总量 query: year, month
+ *   /mqttDataManage/site/daily       - GET 站点日总量  query: date, page, size
+ *   /mqttDataManage/site/cmdType     - GET 站点当日命令类型总量 query: equipNo, date, page, size
+ * 查询：
+ *   /mqttDataManage/messages         - GET 逐条报文    query: equipNo, cmdType, startTime, endTime, page, size
+ *
+ * 注：站点月总量列表(site/monthly)因查询非常慢、非常耗资源已下线，见 fwh_es_mqtt.GetSiteMonthlyTotals 备注。
+ *
+ * 时区：全部统一使用 UTC。date 参数(yyyy.MM.dd)对应 UTC 日索引；
+ *       startTime/endTime(yyyy-MM-dd HH:mm:ss)按 UTC 解析；聚合按 UTC 分桶。
+ */
+package controllers
+
+import (
+	"errors"
+	"fmt"
+	"fwh-toolbox-server/repositories"
+	"fwh-toolbox-server/services"
+	"regexp"
+	"strconv"
+	"time"
+
+	"github.com/gofiber/fiber/v2"
+	"go.uber.org/zap"
+)
+
+// 日期格式 yyyy.MM.dd（对应 ES 日索引）
+var esDateRegexp = regexp.MustCompile(`^\d{4}\.(0[1-9]|1[0-2])\.(0[1-9]|[12]\d|3[01])$`)
+
+// 时间格式 yyyy-MM-dd HH:mm:ss
+const mqttTimeLayout = "2006-01-02 15:04:05"
+
+// 解析分页参数（page 默认 1，size 默认 20）
+func parsePaging(c *fiber.Ctx) (int, int, error) {
+	page, err := strconv.Atoi(c.Query("page", "1"))
+	if err != nil || page <= 0 {
+		return 0, 0, err
+	}
+	size, err := strconv.Atoi(c.Query("size", "20"))
+	if err != nil || size <= 0 {
+		return 0, 0, err
+	}
+	return page, size, nil
+}
+
+// 解析年月
+func parseYearMonth(c *fiber.Ctx) (int, int, bool) {
+	year, err := strconv.Atoi(c.Query("year", ""))
+	if err != nil || year < 2000 || year > 9999 {
+		return 0, 0, false
+	}
+	month, err := strconv.Atoi(c.Query("month", ""))
+	if err != nil || month < 1 || month > 12 {
+		return 0, 0, false
+	}
+	return year, month, true
+}
+
+// GetMqttDailyTotal 日总量
+// /mqttDataManage/total/daily
+func GetMqttDailyTotal(c *fiber.Ctx) error {
+	date := c.Query("date", "")
+	if !esDateRegexp.MatchString(date) {
+		zap.L().Warn("date parameter is in the wrong format", zap.String("date", date))
+		return Fail(c, "date parameter is in the wrong format (yyyy.MM.dd)")
+	}
+	res, err := services.GetMqttDailyTotal(date)
+	if err != nil {
+		return Error(c, "Get MQTT daily total failed.")
+	}
+	return Success(c, res)
+}
+
+// GetMqttMonthlyTotal 月总量
+// /mqttDataManage/total/monthly
+func GetMqttMonthlyTotal(c *fiber.Ctx) error {
+	year, month, ok := parseYearMonth(c)
+	if !ok {
+		return Fail(c, "year/month parameter is invalid")
+	}
+	res, err := services.GetMqttMonthlyTotal(year, month)
+	if err != nil {
+		return Error(c, "Get MQTT monthly total failed.")
+	}
+	return Success(c, res)
+}
+
+// GetMqttMonthlyDailyTrend 月度每日总量（看板柱状图，整月补零）
+// /mqttDataManage/total/dailyTrend
+func GetMqttMonthlyDailyTrend(c *fiber.Ctx) error {
+	year, month, ok := parseYearMonth(c)
+	if !ok {
+		return Fail(c, "year/month parameter is invalid")
+	}
+	res, err := services.GetMqttMonthlyDailyTrend(year, month)
+	if err != nil {
+		return Error(c, "Get MQTT monthly daily trend failed.")
+	}
+	return Success(c, res)
+}
+
+// GetMqttSiteDailyTotals 站点日总量（分页）
+// /mqttDataManage/site/daily
+func GetMqttSiteDailyTotals(c *fiber.Ctx) error {
+	date := c.Query("date", "")
+	if !esDateRegexp.MatchString(date) {
+		return Fail(c, "date parameter is in the wrong format (yyyy.MM.dd)")
+	}
+	page, size, err := parsePaging(c)
+	if err != nil {
+		return Fail(c, "page/size parameter is invalid")
+	}
+	res, err := services.GetMqttSiteDailyTotals(date, page, size)
+	if err != nil {
+		return Error(c, "Get MQTT site daily totals failed.")
+	}
+	return Success(c, res)
+}
+
+// GetMqttSiteCmdTypeTotals 站点当日命令类型总量（下钻，分页）
+// /mqttDataManage/site/cmdType
+func GetMqttSiteCmdTypeTotals(c *fiber.Ctx) error {
+	equipNo := c.Query("equipNo", "")
+	if equipNo == "" {
+		return Fail(c, "equipNo parameter is empty")
+	}
+	date := c.Query("date", "")
+	if !esDateRegexp.MatchString(date) {
+		return Fail(c, "date parameter is in the wrong format (yyyy.MM.dd)")
+	}
+	page, size, err := parsePaging(c)
+	if err != nil {
+		return Fail(c, "page/size parameter is invalid")
+	}
+	res, err := services.GetMqttSiteCmdTypeTotals(equipNo, date, page, size)
+	if err != nil {
+		return Error(c, "Get MQTT site cmdType totals failed.")
+	}
+	return Success(c, res)
+}
+
+// QueryMqttMessages 逐条报文查询
+// /mqttDataManage/messages
+func QueryMqttMessages(c *fiber.Ctx) error {
+	// 可选：equipNo
+	var equipNoPtr *string
+	if v := c.Query("equipNo", ""); v != "" {
+		equipNoPtr = &v
+	}
+
+	// 可选：cmdType
+	var cmdTypePtr *int64
+	if v := c.Query("cmdType", ""); v != "" {
+		cmdType, err := strconv.ParseInt(v, 10, 64)
+		if err != nil {
+			return Fail(c, "cmdType parameter is invalid")
+		}
+		cmdTypePtr = &cmdType
+	}
+
+	// 可选：startTime / endTime（yyyy-MM-dd HH:mm:ss，按 UTC 解析）
+	var startPtr, endPtr *time.Time
+	if v := c.Query("startTime", ""); v != "" {
+		t, err := time.ParseInLocation(mqttTimeLayout, v, time.UTC)
+		if err != nil {
+			return Fail(c, "startTime parameter is in the wrong format (yyyy-MM-dd HH:mm:ss)")
+		}
+		startPtr = &t
+	}
+	if v := c.Query("endTime", ""); v != "" {
+		t, err := time.ParseInLocation(mqttTimeLayout, v, time.UTC)
+		if err != nil {
+			return Fail(c, "endTime parameter is in the wrong format (yyyy-MM-dd HH:mm:ss)")
+		}
+		endPtr = &t
+	}
+
+	page, size, err := parsePaging(c)
+	if err != nil {
+		return Fail(c, "page/size parameter is invalid")
+	}
+
+	res, err := services.QueryMqttMessages(equipNoPtr, cmdTypePtr, startPtr, endPtr, page, size)
+	if err != nil {
+		// 深分页超出 ES 窗口上限：返回 400 友好提示，而非 500
+		if errors.Is(err, repositories.ErrEsDeepPaging) {
+			return Fail(c, fmt.Sprintf("查询超出可翻页范围：最多可浏览前 %d 条，请缩小时间范围或增加设备号/命令类型等过滤条件。",
+				repositories.EsMaxMessagePageWindow))
+		}
+		return Error(c, "Query MQTT messages failed.")
+	}
+	return Success(c, res)
+}

service/domain/esMqttDomain.go

@@ -0,0 +1,157 @@
+package domain
+
+// Elasticsearch(iotcore-stat) MQTT 报文统计相关的响应结构。
+// 服务端遵循 Elasticsearch API 规范。索引名形如 iotcore-stat-<yyyy.MM.dd>，
+// 每条文档字段：@timestamp(date)、timeStamp(long,秒)、equipNo(keyword,站点)、cmdType(int)、event(object)。
+//
+// 说明：需求中的“站点”即 equipNo（设备号）。
+
+/*************** 通用 ES 结构 ***************/
+
+// EsShardInfo 分片信息
+type EsShardInfo struct {
+	Total      int `json:"total"`
+	Successful int `json:"successful"`
+	Skipped    int `json:"skipped"`
+	Failed     int `json:"failed"`
+}
+
+// EsCountResponse 对应 GET <index>/_count
+type EsCountResponse struct {
+	Count  int64        `json:"count"`
+	Shards *EsShardInfo `json:"_shards,omitempty"`
+}
+
+// EsBucket terms 聚合的单个桶（key 为字符串）
+type EsBucket struct {
+	Key      string `json:"key"`
+	DocCount int64  `json:"doc_count"`
+}
+
+// EsKeyedBucket terms 聚合的单个桶（key 为数值，如 cmdType）
+type EsKeyedBucket struct {
+	Key      int64 `json:"key"`
+	DocCount int64 `json:"doc_count"`
+}
+
+// EsTermsAggregation 字符串 key 的 terms 聚合结果
+type EsTermsAggregation struct {
+	DocCountErrorUpperBound int64      `json:"doc_count_error_upper_bound"`
+	SumOtherDocCount        int64      `json:"sum_other_doc_count"`
+	Buckets                 []EsBucket `json:"buckets"`
+}
+
+// EsNumericTermsAggregation 数值 key 的 terms 聚合结果
+type EsNumericTermsAggregation struct {
+	DocCountErrorUpperBound int64           `json:"doc_count_error_upper_bound"`
+	SumOtherDocCount        int64           `json:"sum_other_doc_count"`
+	Buckets                 []EsKeyedBucket `json:"buckets"`
+}
+
+// EsDateBucket date_histogram 的单个桶
+type EsDateBucket struct {
+	KeyAsString string `json:"key_as_string"`
+	Key         int64  `json:"key"` // epoch millis
+	DocCount    int64  `json:"doc_count"`
+}
+
+// EsDateHistogramAggregation date_histogram 聚合结果
+type EsDateHistogramAggregation struct {
+	Buckets []EsDateBucket `json:"buckets"`
+}
+
+// EsCardinality cardinality 聚合结果（去重计数）
+type EsCardinality struct {
+	Value int64 `json:"value"`
+}
+
+/*************** 看板：报文总量 ***************/
+
+// MessageTotal 报文总量（月/日通用）
+type MessageTotal struct {
+	Period string `json:"period"` // 统计周期，如 "2026.09" 或 "2026.09.08"
+	Total  int64  `json:"total"`  // 报文总量
+}
+
+/*************** 看板：站点总量（分页，按量降序） ***************/
+
+// SiteTotalItem 单个站点(equipNo)的报文总量
+type SiteTotalItem struct {
+	EquipNo string `json:"equipNo"`
+	Total   int64  `json:"total"`
+}
+
+// SiteTotalPage 站点总量分页结果
+type SiteTotalPage struct {
+	Page      int             `json:"page"`      // 当前页，从 1 开始
+	PageSize  int             `json:"pageSize"`  // 每页条数
+	SiteCount int64           `json:"siteCount"` // 站点总数（去重 equipNo 数量），用于计算总页数
+	Items     []SiteTotalItem `json:"items"`
+}
+
+/*************** 下钻：站点每日总量 ***************/
+
+// DailyTotal 某天的报文总量
+type DailyTotal struct {
+	Date  string `json:"date"`  // "2026-09-08"
+	Total int64  `json:"total"`
+}
+
+/*************** 下钻：站点每日命令类型总量（分页，按量降序） ***************/
+
+// CmdTypeTotalItem 单个命令类型的报文总量
+type CmdTypeTotalItem struct {
+	CmdType int64 `json:"cmdType"`
+	Total   int64 `json:"total"`
+}
+
+// CmdTypeTotalPage 命令类型总量分页结果
+type CmdTypeTotalPage struct {
+	Page         int                `json:"page"`
+	PageSize     int                `json:"pageSize"`
+	CmdTypeCount int64              `json:"cmdTypeCount"` // 命令类型总数（去重），用于计算总页数
+	Items        []CmdTypeTotalItem `json:"items"`
+}
+
+/*************** 查询：逐条原始报文 ***************/
+
+// MqttMessageItem 单条 MQTT 报文记录（对应 _source）
+type MqttMessageItem struct {
+	Timestamp string         `json:"@timestamp"` // ISO8601，如 2026-09-08T21:49:53.000Z
+	TimeStamp int64          `json:"timeStamp"`  // Unix 秒
+	EquipNo   string         `json:"equipNo"`
+	CmdType   int64          `json:"cmdType"`
+	Event     map[string]any `json:"event,omitempty"`
+}
+
+// MessagePage 逐条报文查询的分页结果
+type MessagePage struct {
+	Page     int               `json:"page"`
+	PageSize int               `json:"pageSize"`
+	Total    int64             `json:"total"` // 命中总数（ES track_total_hits，可能为下界）
+	Items    []MqttMessageItem `json:"items"`
+}
+
+/*************** 内部：解析 ES _search 响应用 ***************/
+
+// EsHitsTotal hits.total
+type EsHitsTotal struct {
+	Value    int64  `json:"value"`
+	Relation string `json:"relation"` // "eq" | "gte"
+}
+
+// EsMessageHit 单条命中
+type EsMessageHit struct {
+	Source MqttMessageItem `json:"_source"`
+}
+
+// EsMessageHits hits 部分
+type EsMessageHits struct {
+	Total EsHitsTotal    `json:"total"`
+	Hits  []EsMessageHit `json:"hits"`
+}
+
+// EsMessageSearchResponse 逐条查询 _search 的完整返回
+type EsMessageSearchResponse struct {
+	Hits *EsMessageHits `json:"hits,omitempty"`
+}

service/main.go

@@ -64,6 +64,9 @@ func main() {
 	// 初始化FwhTestApi（测试环境）
 	repositories.FwhTestHttpInit(&cfg.FwhTestApi)
 
+	// 初始化Fwh ES（iotcore-stat 报文统计）
+	repositories.EsHttpInit(&cfg.FwhEs)
+
 	// 初始化PlaneApi
 	repositories.PlaneHttpInit(&cfg.PlaneApi)
 

service/repositories/api/fwh_es_mqtt/esHttpClient.go

@@ -0,0 +1,140 @@
+package fwh_es_mqtt
+
+import (
+	"bytes"
+	"crypto/tls"
+	"encoding/json"
+	"fmt"
+	"fwh-toolbox-server/config"
+	"io"
+	"net/http"
+	"strings"
+	"time"
+
+	"go.uber.org/zap"
+)
+
+// 服务端是一个 Elasticsearch，接口遵循 Elasticsearch API 技术规范。
+// 该模块使用独立的 HTTP 客户端（basic auth + 可选 TLS 跳过校验），
+// 与项目通用 httpClient 解耦，原因：
+//  1. ES 使用 basic auth，通用客户端不支持；
+//  2. ES 走 HTTPS 自签证书（curl -k），需要跳过 TLS 校验；
+//  3. ES 的非 200(如 400/404) 返回体仍包含有用的错误信息，需要读取。
+
+const (
+	esTimeout time.Duration = 3 * time.Minute // ES 聚合查询可能较慢，超时 3min
+)
+
+var (
+	esHost        string // 形如 "alb-prod.franklinwh.click:5603"
+	esUser        string
+	esPassword    string
+	esIndexPrefix string // 形如 "iotcore-stat"
+	esClient      *http.Client
+)
+
+// EsHttpInit 由启动流程调用，注入 ES 连接配置。
+func EsHttpInit(cfg *config.FwhEsConfig) {
+	esHost = cfg.EsHost
+	esUser = cfg.EsUser
+	esPassword = cfg.EsPassword
+	esIndexPrefix = cfg.EsIndexPrefix
+
+	transport := &http.Transport{
+		TLSClientConfig: &tls.Config{
+			InsecureSkipVerify: cfg.EsInsecureSkipVerify, // 对应 curl -k
+		},
+	}
+	esClient = &http.Client{
+		Timeout:   esTimeout,
+		Transport: transport,
+	}
+}
+
+// getFullEsUrl 拼接完整的 ES 请求 URL。
+// relativePath 形如 "/iotcore-stat-2026.09.07/_search"
+func getFullEsUrl(relativePath string) string {
+	if !strings.HasPrefix(relativePath, "/") {
+		relativePath = "/" + relativePath
+	}
+	return "https://" + esHost + relativePath
+}
+
+// esDo 发送请求并读取响应体。
+// 非 200 时同样返回响应体（供上层解析 ES 错误信息），并附带 error。
+func esDo(method, relativePath string, body any) ([]byte, int, error) {
+	if esClient == nil {
+		return nil, 0, fmt.Errorf("[esDo] ES client 未初始化，请先调用 EsHttpInit")
+	}
+
+	var reader io.Reader
+	if body != nil {
+		jsonData, err := json.Marshal(body)
+		if err != nil {
+			zap.L().Error("[esDo] body marshal failed.", zap.Any("body", body), zap.String("err", err.Error()))
+			return nil, 0, err
+		}
+		reader = bytes.NewReader(jsonData)
+	}
+
+	url := getFullEsUrl(relativePath)
+	zap.L().Debug("[esDo]", zap.String("method", method), zap.String("url", url))
+
+	req, err := http.NewRequest(method, url, reader)
+	if err != nil {
+		zap.L().Error("[esDo] NewRequest failed.", zap.String("err", err.Error()))
+		return nil, 0, err
+	}
+
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("Accept", "application/json")
+	req.SetBasicAuth(esUser, esPassword)
+
+	resp, err := esClient.Do(req)
+	if err != nil {
+		zap.L().Error("[esDo] Do Send failed.", zap.String("err", err.Error()))
+		return nil, 0, err
+	}
+	defer resp.Body.Close()
+
+	respBody, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return nil, resp.StatusCode, err
+	}
+
+	if resp.StatusCode != http.StatusOK {
+		zap.L().Warn("[esDo] Response failed",
+			zap.String("status", resp.Status),
+			zap.String("body", string(respBody)))
+		return respBody, resp.StatusCode, fmt.Errorf("[esDo] ES response failed, status: %s", resp.Status)
+	}
+
+	return respBody, resp.StatusCode, nil
+}
+
+// esGet 发送 GET 请求（如 _count）。
+func esGet(relativePath string) ([]byte, int, error) {
+	return esDo(http.MethodGet, relativePath, nil)
+}
+
+// esSearch 向指定索引发送 _search POST 请求，body 为聚合查询 DSL。
+func esSearch(index string, query any) ([]byte, error) {
+	respBody, _, err := esDo(http.MethodPost, "/"+index+"/_search", query)
+	if err != nil {
+		return respBody, err
+	}
+	return respBody, nil
+}
+
+// esIndexName 根据日期(UTC 日期，格式 yyyy.MM.dd)拼接索引名。
+// 例如 date="2026.09.07" -> "iotcore-stat-2026.09.07"
+func esIndexName(date string) string {
+	return esIndexPrefix + "-" + date
+}
+
+// esMonthIndexPattern 根据年月拼接“该月所有日索引”的通配模式。
+// 例如 year=2026, month=9 -> "iotcore-stat-2026.09.*"
+// ES 支持索引名通配，可一次跨该月全部日索引做聚合。
+func esMonthIndexPattern(year, month int) string {
+	return fmt.Sprintf("%s-%04d.%02d.*", esIndexPrefix, year, month)
+}

service/repositories/api/fwh_es_mqtt/fwh_es_mqtt.go

@@ -0,0 +1,482 @@
+package fwh_es_mqtt
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"fwh-toolbox-server/domain"
+	"time"
+)
+
+// 本文件按「界面需求.txt」实现两大块接口：看板(Dashboard) + 查询(Query)。
+// 需求中的“站点”即 equipNo（设备号）。
+//
+// 看板：
+//   报文总量统计：月总量 / 日总量
+//   站点月总量（分页降序） -> 下钻：本月站点每日总量
+//   站点日总量（分页降序） -> 下钻：站点每日命令类型总量（分页降序）
+// 查询：
+//   按过滤条件查询逐条原始报文，默认当天、按时间降序。
+//
+// 约定：
+//   date 参数格式 "yyyy.MM.dd"（对应日索引 iotcore-stat-yyyy.MM.dd）。
+//   月度统计用通配索引 iotcore-stat-yyyy.MM.* 一次聚合。
+//   分页从 1 开始；聚合类分页用 ES bucket_sort（服务端分页）。
+//
+// 时区约定（重要）：全部统一使用 UTC。
+//   - 索引 iotcore-stat-yyyy.MM.dd 按 UTC 日期切分（实测某日索引覆盖该日 UTC 00:00:00~23:59:59）；
+//   - 记录 @timestamp 为 UTC 绝对时刻；
+//   - 因此 date 参数、date_histogram 分桶、时间范围过滤、“当天”判断均按 UTC 处理，
+//     不做本地时区/北京时间换算，保证与索引切分对齐、首尾天不丢数据。
+//     前端展示与传参也统一按 UTC。
+
+const (
+	// esTermsScanSize terms 聚合内部扫描桶数上限（需大于实际站点/命令类型总数）。
+	esTermsScanSize = 60000
+	// defaultPageSize 默认每页条数
+	defaultPageSize = 20
+	// MaxMessagePageWindow ES from+size 深分页窗口上限(index.max_result_window 默认 10000)。
+	// 逐条查询 from+size 超过该值会被拒绝（ES 硬限制，调大会拖垮集群）。
+	MaxMessagePageWindow = 10000
+)
+
+// ErrDeepPaging 逐条查询深分页超出 ES 窗口上限时返回，供上层识别并给出友好提示（400 而非 500）。
+var ErrDeepPaging = errors.New("deep paging beyond max result window")
+
+// normalizePaging 规整分页参数
+func normalizePaging(page, pageSize int) (int, int) {
+	if page <= 0 {
+		page = 1
+	}
+	if pageSize <= 0 {
+		pageSize = defaultPageSize
+	}
+	return page, pageSize
+}
+
+/*********************** 看板：报文总量 ***********************/
+
+// GetDailyMessageTotal 日总量：某天索引的文档总数。
+// 对应：GET iotcore-stat-<date>/_count
+func GetDailyMessageTotal(date string) (*domain.MessageTotal, error) {
+	index := esIndexName(date)
+	respBody, _, err := esGet("/" + index + "/_count")
+	if err != nil {
+		return nil, err
+	}
+	var res domain.EsCountResponse
+	if err := json.Unmarshal(respBody, &res); err != nil {
+		return nil, fmt.Errorf("[GetDailyMessageTotal] 解析响应失败: %w", err)
+	}
+	return &domain.MessageTotal{Period: date, Total: res.Count}, nil
+}
+
+// GetMonthlyMessageTotal 月总量：跨该月所有日索引的文档总数。
+// 对应：GET iotcore-stat-<year>.<month>.*/_count
+func GetMonthlyMessageTotal(year, month int) (*domain.MessageTotal, error) {
+	pattern := esMonthIndexPattern(year, month)
+	respBody, _, err := esGet("/" + pattern + "/_count")
+	if err != nil {
+		return nil, err
+	}
+	var res domain.EsCountResponse
+	if err := json.Unmarshal(respBody, &res); err != nil {
+		return nil, fmt.Errorf("[GetMonthlyMessageTotal] 解析响应失败: %w", err)
+	}
+	period := fmt.Sprintf("%04d.%02d", year, month)
+	return &domain.MessageTotal{Period: period, Total: res.Count}, nil
+}
+
+/*********************** 看板：站点总量（分页降序） ***********************/
+
+// buildSiteTotalsQuery 构造“各站点(equipNo)报文总量 + 分页降序 + 站点去重计数”的聚合查询。
+func buildSiteTotalsQuery(page, pageSize int) map[string]any {
+	from := (page - 1) * pageSize
+	return map[string]any{
+		"size": 0,
+		"aggs": map[string]any{
+			// 站点总数（去重 equipNo），用于前端计算总页数
+			"site_count": map[string]any{
+				"cardinality": map[string]any{
+					"field": "equipNo",
+				},
+			},
+			// 各站点总量
+			"by_site": map[string]any{
+				"terms": map[string]any{
+					"field":      "equipNo",
+					"size":       esTermsScanSize,
+					"shard_size": esTermsScanSize,
+					"order":      map[string]any{"_count": "desc"},
+				},
+				"aggs": map[string]any{
+					// 服务端分页：先按量降序，再截取当前页
+					"paged": map[string]any{
+						"bucket_sort": map[string]any{
+							"sort": []any{
+								map[string]any{"_count": map[string]any{"order": "desc"}},
+							},
+							"from": from,
+							"size": pageSize,
+						},
+					},
+				},
+			},
+		},
+	}
+}
+
+// siteTotalsAggResponse 解析站点总量聚合响应
+type siteTotalsAggResponse struct {
+	Aggregations *struct {
+		SiteCount *domain.EsCardinality      `json:"site_count,omitempty"`
+		BySite    *domain.EsTermsAggregation `json:"by_site,omitempty"`
+	} `json:"aggregations,omitempty"`
+}
+
+// parseSiteTotalPage 将聚合响应转为分页结果
+func parseSiteTotalPage(respBody []byte, page, pageSize int, fnName string) (*domain.SiteTotalPage, error) {
+	var res siteTotalsAggResponse
+	if err := json.Unmarshal(respBody, &res); err != nil {
+		return nil, fmt.Errorf("[%s] 解析响应失败: %w", fnName, err)
+	}
+	if res.Aggregations == nil || res.Aggregations.BySite == nil {
+		return nil, fmt.Errorf("[%s] 响应缺少 by_site 聚合", fnName)
+	}
+	items := make([]domain.SiteTotalItem, 0, len(res.Aggregations.BySite.Buckets))
+	for _, b := range res.Aggregations.BySite.Buckets {
+		items = append(items, domain.SiteTotalItem{EquipNo: b.Key, Total: b.DocCount})
+	}
+	var siteCount int64
+	if res.Aggregations.SiteCount != nil {
+		siteCount = res.Aggregations.SiteCount.Value
+	}
+	return &domain.SiteTotalPage{
+		Page:      page,
+		PageSize:  pageSize,
+		SiteCount: siteCount,
+		Items:     items,
+	}, nil
+}
+
+// GetSiteMonthlyTotals 本月各站点报文总量，分页、按量从大到小。
+//
+// ⚠️ 性能警告：此查询【非常慢、非常耗资源】，生产环境慎用！
+//   - 它对整月通配索引 iotcore-stat-<year>.<month>.*（数十亿级文档、数十个分片）
+//     做 equipNo 的 terms 聚合（size=60000）+ cardinality 去重 + bucket_sort 分页，
+//     等于全月全量扫描并在协调节点合并海量桶，耗时可达数十秒甚至更久，
+//     且会给 ES 集群带来很大内存/CPU 压力，可能影响其它查询。
+//   - 因该性能问题，前端“站点月总量列表”界面已下线，本函数当前无上层调用，
+//     仅作为库能力保留。若确需使用，务必：限制并发、加超时、错峰调用，
+//     或改用预聚合/离线统计（如按天预算后再累加），不要直接对整月实时聚合。
+//
+// 相比之下，按“单日索引”做同类聚合（GetSiteDailyTotals）代价可控，推荐优先使用。
+func GetSiteMonthlyTotals(year, month, page, pageSize int) (*domain.SiteTotalPage, error) {
+	page, pageSize = normalizePaging(page, pageSize)
+	pattern := esMonthIndexPattern(year, month)
+	respBody, err := esSearch(pattern, buildSiteTotalsQuery(page, pageSize))
+	if err != nil {
+		return nil, err
+	}
+	return parseSiteTotalPage(respBody, page, pageSize, "GetSiteMonthlyTotals")
+}
+
+// GetSiteDailyTotals 某天各站点报文总量，分页、按量从大到小。
+func GetSiteDailyTotals(date string, page, pageSize int) (*domain.SiteTotalPage, error) {
+	page, pageSize = normalizePaging(page, pageSize)
+	index := esIndexName(date)
+	respBody, err := esSearch(index, buildSiteTotalsQuery(page, pageSize))
+	if err != nil {
+		return nil, err
+	}
+	return parseSiteTotalPage(respBody, page, pageSize, "GetSiteDailyTotals")
+}
+
+/*********************** 下钻：站点本月每日总量 ***********************/
+
+// GetSiteDailyTotalsInMonth 某站点本月的每日报文总量（用于点击站点后的每日趋势）。
+// 用 date_histogram 按天(UTC)聚合，一次跨该月全部日索引。
+func GetSiteDailyTotalsInMonth(equipNo string, year, month int) ([]domain.DailyTotal, error) {
+	pattern := esMonthIndexPattern(year, month)
+	query := map[string]any{
+		"size": 0,
+		"query": map[string]any{
+			"term": map[string]any{"equipNo": equipNo},
+		},
+		"aggs": map[string]any{
+			"by_day": map[string]any{
+				"date_histogram": map[string]any{
+					"field":             "@timestamp",
+					"calendar_interval": "day",
+					// 统一使用 UTC：不设置 time_zone，ES 默认按 UTC 切分“每一天”，
+					// 与索引 iotcore-stat-yyyy.MM.dd 的 UTC 日期切分对齐。
+					"format":        "yyyy-MM-dd",
+					"min_doc_count": 0,
+				},
+			},
+		},
+	}
+	respBody, err := esSearch(pattern, query)
+	if err != nil {
+		return nil, err
+	}
+
+	var res struct {
+		Aggregations *struct {
+			ByDay *domain.EsDateHistogramAggregation `json:"by_day,omitempty"`
+		} `json:"aggregations,omitempty"`
+	}
+	if err := json.Unmarshal(respBody, &res); err != nil {
+		return nil, fmt.Errorf("[GetSiteDailyTotalsInMonth] 解析响应失败: %w", err)
+	}
+	if res.Aggregations == nil || res.Aggregations.ByDay == nil {
+		return nil, fmt.Errorf("[GetSiteDailyTotalsInMonth] 响应缺少 by_day 聚合")
+	}
+	out := make([]domain.DailyTotal, 0, len(res.Aggregations.ByDay.Buckets))
+	for _, b := range res.Aggregations.ByDay.Buckets {
+		out = append(out, domain.DailyTotal{Date: b.KeyAsString, Total: b.DocCount})
+	}
+	return out, nil
+}
+
+// GetMonthlyDailyTrend 全站本月每日报文总量（不区分设备）。
+// 用 date_histogram 按天(UTC)对整月通配索引聚合，返回有数据的每日桶。
+//
+// 性能说明：仅按“天”聚合总量、不做 equipNo 分组，代价可控（本质接近 _count 分桶），
+// 与已下线的“站点月总量列表”那类 equipNo 全量聚合不同，可放心用于看板。
+// 注：ES date_histogram 的 min_doc_count:0 只会补齐“已存在数据的时间范围内”的空天，
+// 整月 1~月末的完整补零由上层(service)负责。
+func GetMonthlyDailyTrend(year, month int) ([]domain.DailyTotal, error) {
+	pattern := esMonthIndexPattern(year, month)
+	query := map[string]any{
+		"size": 0,
+		"aggs": map[string]any{
+			"by_day": map[string]any{
+				"date_histogram": map[string]any{
+					"field":             "@timestamp",
+					"calendar_interval": "day",
+					// 统一使用 UTC：不设置 time_zone，ES 默认按 UTC 切分“每一天”，
+					// 与索引 iotcore-stat-yyyy.MM.dd 的 UTC 日期切分对齐。
+					"format":        "yyyy-MM-dd",
+					"min_doc_count": 0,
+				},
+			},
+		},
+	}
+	respBody, err := esSearch(pattern, query)
+	if err != nil {
+		return nil, err
+	}
+
+	var res struct {
+		Aggregations *struct {
+			ByDay *domain.EsDateHistogramAggregation `json:"by_day,omitempty"`
+		} `json:"aggregations,omitempty"`
+	}
+	if err := json.Unmarshal(respBody, &res); err != nil {
+		return nil, fmt.Errorf("[GetMonthlyDailyTrend] 解析响应失败: %w", err)
+	}
+	if res.Aggregations == nil || res.Aggregations.ByDay == nil {
+		return nil, fmt.Errorf("[GetMonthlyDailyTrend] 响应缺少 by_day 聚合")
+	}
+	out := make([]domain.DailyTotal, 0, len(res.Aggregations.ByDay.Buckets))
+	for _, b := range res.Aggregations.ByDay.Buckets {
+		out = append(out, domain.DailyTotal{Date: b.KeyAsString, Total: b.DocCount})
+	}
+	return out, nil
+}
+
+/*********************** 下钻：站点当日各命令类型总量（分页降序） ***********************/
+
+// GetSiteCmdTypeTotals 某站点某天各命令类型(cmdType)的报文总量，分页、按量从大到小。
+func GetSiteCmdTypeTotals(equipNo, date string, page, pageSize int) (*domain.CmdTypeTotalPage, error) {
+	page, pageSize = normalizePaging(page, pageSize)
+	index := esIndexName(date)
+	from := (page - 1) * pageSize
+
+	query := map[string]any{
+		"size": 0,
+		"query": map[string]any{
+			"term": map[string]any{"equipNo": equipNo},
+		},
+		"aggs": map[string]any{
+			"cmd_count": map[string]any{
+				"cardinality": map[string]any{"field": "cmdType"},
+			},
+			"by_cmd": map[string]any{
+				"terms": map[string]any{
+					"field":      "cmdType",
+					"size":       esTermsScanSize,
+					"shard_size": esTermsScanSize,
+					"order":      map[string]any{"_count": "desc"},
+				},
+				"aggs": map[string]any{
+					"paged": map[string]any{
+						"bucket_sort": map[string]any{
+							"sort": []any{
+								map[string]any{"_count": map[string]any{"order": "desc"}},
+							},
+							"from": from,
+							"size": pageSize,
+						},
+					},
+				},
+			},
+		},
+	}
+
+	respBody, err := esSearch(index, query)
+	if err != nil {
+		return nil, err
+	}
+
+	var res struct {
+		Aggregations *struct {
+			CmdCount *domain.EsCardinality             `json:"cmd_count,omitempty"`
+			ByCmd    *domain.EsNumericTermsAggregation `json:"by_cmd,omitempty"`
+		} `json:"aggregations,omitempty"`
+	}
+	if err := json.Unmarshal(respBody, &res); err != nil {
+		return nil, fmt.Errorf("[GetSiteCmdTypeTotals] 解析响应失败: %w", err)
+	}
+	if res.Aggregations == nil || res.Aggregations.ByCmd == nil {
+		return nil, fmt.Errorf("[GetSiteCmdTypeTotals] 响应缺少 by_cmd 聚合")
+	}
+	items := make([]domain.CmdTypeTotalItem, 0, len(res.Aggregations.ByCmd.Buckets))
+	for _, b := range res.Aggregations.ByCmd.Buckets {
+		items = append(items, domain.CmdTypeTotalItem{CmdType: b.Key, Total: b.DocCount})
+	}
+	var cmdTypeCount int64
+	if res.Aggregations.CmdCount != nil {
+		cmdTypeCount = res.Aggregations.CmdCount.Value
+	}
+	return &domain.CmdTypeTotalPage{
+		Page:         page,
+		PageSize:     pageSize,
+		CmdTypeCount: cmdTypeCount,
+		Items:        items,
+	}, nil
+}
+
+/*********************** 查询：逐条原始报文 ***********************/
+
+// QueryMessageParam 逐条报文查询的过滤条件。
+// 所有过滤字段可选；时间范围为空时默认查“当天”(UTC)。
+type QueryMessageParam struct {
+	EquipNo   *string    // 站点(设备号)
+	CmdType   *int64     // 命令类型
+	StartTime *time.Time // 起始时间(含)
+	EndTime   *time.Time // 结束时间(含)
+	Page      int        // 页码，从 1 开始
+	PageSize  int        // 每页条数
+}
+
+// QueryMessages 按过滤条件查询逐条原始报文。
+// 默认：当天数据，按 @timestamp 降序（最新在前）。
+//
+// 时区说明：统一使用 UTC。索引 iotcore-stat-yyyy.MM.dd 按 UTC 日期切分，
+// 记录 @timestamp 为 UTC 绝对时刻；因此“当天/时间范围”均以 UTC 计算，
+// 上层(controller)传入的 StartTime/EndTime 也应为 UTC 时刻。
+// 说明：ES from+size 深分页上限为 index.max_result_window(默认 10000)，
+// 超出范围会返回错误提示，避免深分页拖垮集群。
+func QueryMessages(param QueryMessageParam) (*domain.MessagePage, error) {
+	page, pageSize := normalizePaging(param.Page, param.PageSize)
+
+	// 时间范围默认当天(UTC)
+	var start, end time.Time
+	switch {
+	case param.StartTime != nil && param.EndTime != nil:
+		start, end = *param.StartTime, *param.EndTime
+	case param.StartTime != nil:
+		start = *param.StartTime
+		end = time.Now().UTC()
+	case param.EndTime != nil:
+		end = *param.EndTime
+		start = time.Date(end.Year(), end.Month(), end.Day(), 0, 0, 0, 0, time.UTC)
+	default:
+		now := time.Now().UTC()
+		start = time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, time.UTC)
+		end = now
+	}
+
+	from := (page - 1) * pageSize
+	if from+pageSize > MaxMessagePageWindow {
+		return nil, fmt.Errorf("[QueryMessages] 分页超出上限: from(%d)+size(%d) > %d: %w",
+			from, pageSize, MaxMessagePageWindow, ErrDeepPaging)
+	}
+
+	// 过滤条件
+	filters := []any{
+		map[string]any{
+			"range": map[string]any{
+				"@timestamp": map[string]any{
+					"gte": start.UTC().Format(time.RFC3339),
+					"lte": end.UTC().Format(time.RFC3339),
+				},
+			},
+		},
+	}
+	if param.EquipNo != nil {
+		filters = append(filters, map[string]any{"term": map[string]any{"equipNo": *param.EquipNo}})
+	}
+	if param.CmdType != nil {
+		filters = append(filters, map[string]any{"term": map[string]any{"cmdType": *param.CmdType}})
+	}
+
+	query := map[string]any{
+		"from":             from,
+		"size":             pageSize,
+		"track_total_hits": true,
+		"query": map[string]any{
+			"bool": map[string]any{"filter": filters},
+		},
+		"sort": []any{
+			map[string]any{"@timestamp": map[string]any{"order": "desc"}},
+		},
+	}
+
+	// 跨天查询用月/多日通配，这里以“当天所在日索引”为默认；
+	// 若时间跨月，改用更大范围通配以覆盖。
+	index := chooseIndexForRange(start, end)
+
+	respBody, err := esSearch(index, query)
+	if err != nil {
+		return nil, err
+	}
+
+	var res domain.EsMessageSearchResponse
+	if err := json.Unmarshal(respBody, &res); err != nil {
+		return nil, fmt.Errorf("[QueryMessages] 解析响应失败: %w", err)
+	}
+
+	out := &domain.MessagePage{Page: page, PageSize: pageSize}
+	if res.Hits != nil {
+		out.Total = res.Hits.Total.Value
+		out.Items = make([]domain.MqttMessageItem, 0, len(res.Hits.Hits))
+		for _, h := range res.Hits.Hits {
+			out.Items = append(out.Items, h.Source)
+		}
+	}
+	return out, nil
+}
+
+// chooseIndexForRange 根据查询时间范围选择索引：
+//   - 同一天：精确到日索引 iotcore-stat-yyyy.MM.dd
+//   - 同一月跨多天：该月通配 iotcore-stat-yyyy.MM.*
+//   - 跨月：年通配 iotcore-stat-yyyy.*
+//
+// 时间以 UTC 判断，与索引按 UTC 日期切分保持一致。
+func chooseIndexForRange(start, end time.Time) string {
+	s := start.UTC()
+	e := end.UTC()
+	if s.Year() == e.Year() && s.Month() == e.Month() && s.Day() == e.Day() {
+		return esIndexName(fmt.Sprintf("%04d.%02d.%02d", s.Year(), int(s.Month()), s.Day()))
+	}
+	if s.Year() == e.Year() && s.Month() == e.Month() {
+		return esMonthIndexPattern(s.Year(), int(s.Month()))
+	}
+	if s.Year() == e.Year() {
+		return fmt.Sprintf("%s-%04d.*", esIndexPrefix, s.Year())
+	}
+	return esIndexPrefix + "-*"
+}

service/repositories/repositories.go

@@ -4,6 +4,7 @@ import (
 	"fwh-toolbox-server/config"
 	"fwh-toolbox-server/domain"
 	"fwh-toolbox-server/repositories/api/acceleronix"
+	"fwh-toolbox-server/repositories/api/fwh_es_mqtt"
 	"fwh-toolbox-server/repositories/api/fwh_pro"
 	"fwh-toolbox-server/repositories/api/fwh_test"
 	"fwh-toolbox-server/repositories/api/plane_v1"
@@ -149,6 +150,34 @@ var QueryAcceleronixBatchSimCardDataInfo = acceleronix.QueryBatchSimCardDataInfo
 var QueryAcceleronixDataPoolList = acceleronix.QueryDataPoolList                 // 查询流量池列表
 var QueryAcceleronixDataPoolSimInfo = acceleronix.QueryDataPoolSimInfo           // 查询流量池SIM详情
 
+// ============ fwh_es_mqtt（iotcore-stat Elasticsearch 报文统计）============
+// 说明：站点即 equipNo（设备号）。看板 + 逐条查询。
+
+// 类型/参数导出
+type EsQueryMessageParam = fwh_es_mqtt.QueryMessageParam
+
+var EsHttpInit = fwh_es_mqtt.EsHttpInit // 初始化 ES HTTP 客户端（启动时调用）
+
+// 看板 - 报文总量
+var EsGetDailyMessageTotal = fwh_es_mqtt.GetDailyMessageTotal     // 日总量
+var EsGetMonthlyMessageTotal = fwh_es_mqtt.GetMonthlyMessageTotal // 月总量
+
+// 看板 - 站点总量（分页降序）
+var EsGetSiteMonthlyTotals = fwh_es_mqtt.GetSiteMonthlyTotals             // 站点月总量
+var EsGetSiteDailyTotals = fwh_es_mqtt.GetSiteDailyTotals                 // 站点日总量
+var EsGetSiteDailyTotalsInMonth = fwh_es_mqtt.GetSiteDailyTotalsInMonth   // 下钻：站点本月每日总量
+var EsGetMonthlyDailyTrend = fwh_es_mqtt.GetMonthlyDailyTrend             // 全站本月每日总量（看板月度趋势图）
+var EsGetSiteCmdTypeTotals = fwh_es_mqtt.GetSiteCmdTypeTotals             // 下钻：站点当日各命令类型总量
+
+// 查询 - 逐条原始报文
+var EsQueryMessages = fwh_es_mqtt.QueryMessages // 按过滤条件查询逐条报文（默认当天，按时间降序）
+
+// ErrEsDeepPaging 逐条查询深分页超出 ES 窗口上限的哨兵错误（供 controller 识别并返回 400 友好提示）
+var ErrEsDeepPaging = fwh_es_mqtt.ErrDeepPaging
+
+// EsMaxMessagePageWindow 逐条查询 from+size 上限（ES max_result_window）
+const EsMaxMessagePageWindow = fwh_es_mqtt.MaxMessagePageWindow
+
 // Plane_V1
 type PlaneConfig = plane_v1.PlaneConfig
 type PlaneIntakeIssueRequest = plane_v1.IntakeIssueRequest

service/routes/mqttDataManageRoutes.go

@@ -0,0 +1,36 @@
+/*
+ * MQTT 报文管理 API List（数据来自 iotcore-stat Elasticsearch，站点即 equipNo）
+ * 看板：
+ *   /mqttDataManage/total/daily       - GET 日总量
+ *   /mqttDataManage/total/monthly     - GET 月总量
+ *   /mqttDataManage/total/dailyTrend  - GET 月度每日总量（柱状图，整月补零）
+ *   /mqttDataManage/site/daily        - GET 站点日总量（分页）
+ *   /mqttDataManage/site/cmdType      - GET 站点当日命令类型总量（下钻，分页）
+ * 查询：
+ *   /mqttDataManage/messages          - GET 逐条报文查询
+ *
+ * 注：站点月总量列表(site/monthly)因查询非常慢、非常耗资源已下线。
+ */
+package routes
+
+import (
+	"fwh-toolbox-server/controllers"
+
+	"github.com/gofiber/fiber/v2"
+)
+
+func MqttDataManageSetup(api fiber.Router) {
+	// 报文总量
+	api.Get("/mqttDataManage/total/daily", controllers.GetMqttDailyTotal)
+	api.Get("/mqttDataManage/total/monthly", controllers.GetMqttMonthlyTotal)
+	api.Get("/mqttDataManage/total/dailyTrend", controllers.GetMqttMonthlyDailyTrend)
+
+	// 站点日总量（分页降序）
+	api.Get("/mqttDataManage/site/daily", controllers.GetMqttSiteDailyTotals)
+
+	// 下钻：站点当日命令类型总量
+	api.Get("/mqttDataManage/site/cmdType", controllers.GetMqttSiteCmdTypeTotals)
+
+	// 逐条报文查询
+	api.Get("/mqttDataManage/messages", controllers.QueryMqttMessages)
+}

service/routes/routes.go

@@ -33,6 +33,7 @@ func Setup(app *fiber.App) {
 	DespatchConfigSetup(api)
 	FaultRecordSetup(api)
 	SimCardManageSetup(api)
+	MqttDataManageSetup(api)
 	FeedbackSetup(api)
 	AiChatSetup(api)
 

service/services/mqttDataManageServices.go

@@ -0,0 +1,106 @@
+/*
+ * MQTT 报文管理 Service（数据来自 iotcore-stat Elasticsearch）
+ * 站点即 equipNo。看板统计 + 逐条查询。
+ */
+package services
+
+import (
+	"fmt"
+	"fwh-toolbox-server/domain"
+	"fwh-toolbox-server/repositories"
+	"time"
+
+	"go.uber.org/zap"
+)
+
+/*************** 看板：报文总量 ***************/
+
+// GetMqttDailyTotal 日报文总量
+func GetMqttDailyTotal(date string) (*domain.MessageTotal, error) {
+	res, err := repositories.EsGetDailyMessageTotal(date)
+	if err != nil {
+		zap.L().Error("EsGetDailyMessageTotal failed.", zap.String("err", err.Error()), zap.String("date", date))
+		return nil, err
+	}
+	return res, nil
+}
+
+// GetMqttMonthlyTotal 月报文总量（仅统计总数，代价可控；不涉及慢的站点月聚合）
+func GetMqttMonthlyTotal(year, month int) (*domain.MessageTotal, error) {
+	res, err := repositories.EsGetMonthlyMessageTotal(year, month)
+	if err != nil {
+		zap.L().Error("EsGetMonthlyMessageTotal failed.", zap.String("err", err.Error()), zap.Int("year", year), zap.Int("month", month))
+		return nil, err
+	}
+	return res, nil
+}
+
+// GetMqttMonthlyDailyTrend 全站本月每日报文总量（看板柱状图）。
+// 结果按整月 1~月末补齐：缺失的天补 0，保证横轴完整。
+func GetMqttMonthlyDailyTrend(year, month int) ([]domain.DailyTotal, error) {
+	list, err := repositories.EsGetMonthlyDailyTrend(year, month)
+	if err != nil {
+		zap.L().Error("EsGetMonthlyDailyTrend failed.", zap.String("err", err.Error()), zap.Int("year", year), zap.Int("month", month))
+		return nil, err
+	}
+
+	// 已有数据按日期建索引
+	totalByDate := make(map[string]int64, len(list))
+	for _, d := range list {
+		totalByDate[d.Date] = d.Total
+	}
+
+	// 生成整月 1~月末的完整日期序列（缺失补 0）
+	// 用下月第 0 天得到当月天数
+	daysInMonth := time.Date(year, time.Month(month)+1, 0, 0, 0, 0, 0, time.UTC).Day()
+	out := make([]domain.DailyTotal, 0, daysInMonth)
+	for day := 1; day <= daysInMonth; day++ {
+		date := fmt.Sprintf("%04d-%02d-%02d", year, month, day)
+		out = append(out, domain.DailyTotal{Date: date, Total: totalByDate[date]})
+	}
+	return out, nil
+}
+
+/*************** 看板：站点日总量（分页降序） ***************/
+
+// GetMqttSiteDailyTotals 站点日总量（分页）
+func GetMqttSiteDailyTotals(date string, page, pageSize int) (*domain.SiteTotalPage, error) {
+	res, err := repositories.EsGetSiteDailyTotals(date, page, pageSize)
+	if err != nil {
+		zap.L().Error("EsGetSiteDailyTotals failed.", zap.String("err", err.Error()), zap.String("date", date))
+		return nil, err
+	}
+	return res, nil
+}
+
+/*************** 下钻：站点当日命令类型总量 ***************/
+
+// GetMqttSiteCmdTypeTotals 某站点某天各命令类型总量（分页）
+func GetMqttSiteCmdTypeTotals(equipNo, date string, page, pageSize int) (*domain.CmdTypeTotalPage, error) {
+	res, err := repositories.EsGetSiteCmdTypeTotals(equipNo, date, page, pageSize)
+	if err != nil {
+		zap.L().Error("EsGetSiteCmdTypeTotals failed.", zap.String("err", err.Error()), zap.String("equipNo", equipNo), zap.String("date", date))
+		return nil, err
+	}
+	return res, nil
+}
+
+/*************** 查询：逐条原始报文 ***************/
+
+// QueryMqttMessages 按过滤条件查询逐条报文
+func QueryMqttMessages(equipNo *string, cmdType *int64, startTime, endTime *time.Time, page, pageSize int) (*domain.MessagePage, error) {
+	param := repositories.EsQueryMessageParam{
+		EquipNo:   equipNo,
+		CmdType:   cmdType,
+		StartTime: startTime,
+		EndTime:   endTime,
+		Page:      page,
+		PageSize:  pageSize,
+	}
+	res, err := repositories.EsQueryMessages(param)
+	if err != nil {
+		zap.L().Error("EsQueryMessages failed.", zap.String("err", err.Error()))
+		return nil, err
+	}
+	return res, nil
+}

src/router/record.ts

@@ -35,6 +35,39 @@ export const routeRecordRaw: MergeMenuMixedOptions[] = [
     },
     component: 'sim-card-manage/index',
   },
+  {
+    path: 'mqtt-data-manage',
+    key: 'mqtt-data-manage',
+    name: 'MqttDataManage',
+    icon: 'i-material-symbols:sms-outline-rounded',
+    label: 'MQTT报文管理',
+    children: [
+      {
+        path: 'mqtt-data-dashboard',
+        key: 'mqtt-data-dashboard',
+        name: 'MqttDataDashboard',
+        icon: 'i-mage:dashboard-chart',
+        label: 'MQTT数据看板',
+        meta: {
+          componentName: 'MqttDataDashboard',
+          showTab: true,
+        },
+        component: 'mqtt-data-manage/dashboard/index',
+      },
+      {
+        path: 'mqtt-data-list',
+        key: 'mqtt-data-list',
+        name: 'MqttDataList',
+        icon: 'i-boxicons:list-ul-square',
+        label: 'MQTT数据列表',
+        meta: {
+          componentName: 'MqttDataList',
+          showTab: true,
+        },
+        component: 'mqtt-data-manage/list/index',
+      },
+    ],
+  },
   {
     path: 'shutdown-manage',
     name: 'shutdownManage',

src/service/mqttDataManage/index.ts

@@ -0,0 +1,78 @@
+// MQTT 报文管理 API（站点即 equipNo，数据来自 iotcore-stat Elasticsearch）
+// 看板：
+//   /mqttDataManage/total/daily        - GET 日总量
+//   /mqttDataManage/total/monthly      - GET 月总量
+//   /mqttDataManage/total/dailyTrend   - GET 月度每日总量（柱状图，整月补零）
+//   /mqttDataManage/site/daily         - GET 站点日总量（分页）
+//   /mqttDataManage/site/cmdType       - GET 站点当日命令类型总量（下钻，分页）
+// 查询：
+//   /mqttDataManage/messages           - GET 逐条报文查询
+// 注：站点月总量列表因查询非常慢、非常耗资源已下线。
+//
+// 时区约定：全部统一使用 UTC。
+//   - date 参数格式 yyyy.MM.dd，对应 UTC 日索引 iotcore-stat-yyyy.MM.dd；
+//   - 逐条查询的 startTime/endTime(yyyy-MM-dd HH:mm:ss) 按 UTC 解释；
+//   - 记录 @timestamp/timeStamp 为 UTC 绝对时刻，前端按 UTC 展示。
+
+import http from '@/service/serverReq'
+import type {
+  MessageTotal,
+  DailyTotal,
+  SiteTotalPage,
+  CmdTypeTotalPage,
+  MessagePage,
+  QueryMessageParams,
+} from './interface'
+
+const API_List = {
+  totalDaily: '/mqttDataManage/total/daily',
+  totalMonthly: '/mqttDataManage/total/monthly',
+  totalDailyTrend: '/mqttDataManage/total/dailyTrend',
+  siteDaily: '/mqttDataManage/site/daily',
+  siteCmdType: '/mqttDataManage/site/cmdType',
+  messages: '/mqttDataManage/messages',
+}
+
+// 日总量，date: yyyy.MM.dd
+export const getMqttDailyTotal = async (date: string): Promise<MessageTotal> => {
+  return http.get<MessageTotal>(API_List.totalDaily, { params: { date } })
+}
+
+// 月总量
+export const getMqttMonthlyTotal = async (year: number, month: number): Promise<MessageTotal> => {
+  return http.get<MessageTotal>(API_List.totalMonthly, { params: { year, month } })
+}
+
+// 月度每日总量（整月补零，用于看板柱状图）
+export const getMqttMonthlyDailyTrend = async (
+  year: number,
+  month: number,
+): Promise<DailyTotal[]> => {
+  return http.get<DailyTotal[]>(API_List.totalDailyTrend, { params: { year, month } })
+}
+
+// 站点日总量（分页），date: yyyy.MM.dd
+export const getMqttSiteDailyTotals = async (
+  date: string,
+  page: number,
+  size: number,
+): Promise<SiteTotalPage> => {
+  return http.get<SiteTotalPage>(API_List.siteDaily, { params: { date, page, size } })
+}
+
+// 下钻：站点当日命令类型总量（分页），date: yyyy.MM.dd
+export const getMqttSiteCmdTypeTotals = async (
+  equipNo: string,
+  date: string,
+  page: number,
+  size: number,
+): Promise<CmdTypeTotalPage> => {
+  return http.get<CmdTypeTotalPage>(API_List.siteCmdType, {
+    params: { equipNo, date, page, size },
+  })
+}
+
+// 逐条报文查询
+export const queryMqttMessages = async (params: QueryMessageParams): Promise<MessagePage> => {
+  return http.get<MessagePage>(API_List.messages, { params })
+}

src/service/mqttDataManage/interface.ts

@@ -0,0 +1,68 @@
+// MQTT 报文管理相关类型（数据来自 iotcore-stat Elasticsearch，站点即 equipNo）
+
+// 报文总量（月/日通用）
+export interface MessageTotal {
+  period: string // "2026.09" 或 "2026.09.08"
+  total: number
+}
+
+// 单个站点总量
+export interface SiteTotalItem {
+  equipNo: string
+  total: number
+}
+
+// 站点总量分页
+export interface SiteTotalPage {
+  page: number
+  pageSize: number
+  siteCount: number // 站点去重总数，用于算总页数
+  items: SiteTotalItem[]
+}
+
+// 某天总量（月度每日总量趋势）
+export interface DailyTotal {
+  date: string // "2026-09-08"
+  total: number
+}
+
+// 单个命令类型总量
+export interface CmdTypeTotalItem {
+  cmdType: number
+  total: number
+}
+
+// 命令类型总量分页
+export interface CmdTypeTotalPage {
+  page: number
+  pageSize: number
+  cmdTypeCount: number
+  items: CmdTypeTotalItem[]
+}
+
+// 单条 MQTT 报文
+export interface MqttMessageItem {
+  '@timestamp': string // ISO8601
+  timeStamp: number // Unix 秒
+  equipNo: string
+  cmdType: number
+  event?: Record<string, unknown>
+}
+
+// 逐条报文查询分页
+export interface MessagePage {
+  page: number
+  pageSize: number
+  total: number
+  items: MqttMessageItem[]
+}
+
+// 逐条查询过滤条件
+export interface QueryMessageParams {
+  equipNo?: string
+  cmdType?: number
+  startTime?: string // "yyyy-MM-dd HH:mm:ss"
+  endTime?: string // "yyyy-MM-dd HH:mm:ss"
+  page: number
+  size: number
+}

src/views/mqtt-data-manage/dashboard/index.vue

@@ -0,0 +1,467 @@
+<template>
+  <NSpin :show="isInit">
+    <div class="p-4 space-y-4">
+      <!-- 顶部：日期选择 + 日总量卡片 -->
+      <NCard size="small">
+        <div class="flex flex-wrap gap-4 items-center">
+          <NForm inline label-placement="left" label-width="auto" :show-feedback="false">
+            <NFormItem label="统计日期">
+              <NDatePicker
+                v-model:value="selectedDateTs"
+                type="date"
+                :actions="null"
+                clearable
+              />
+            </NFormItem>
+            <NFormItem>
+              <NButton type="primary" secondary @click="refreshAll">刷新</NButton>
+            </NFormItem>
+          </NForm>
+          <NTag size="small" :bordered="false" type="info">时间均为 UTC</NTag>
+        </div>
+      </NCard>
+
+      <div class="gap-4 grid grid-cols-1 md:grid-cols-2">
+        <InfoCard :data="monthlyCard" />
+        <InfoCard :data="dailyCard" />
+      </div>
+
+      <!-- 月度每日总量柱状图 -->
+      <NCard :title="`月度每日总量（${monthLabel}）`" size="small">
+        <NSpin :show="loadingTrend">
+          <div ref="trendChartRef" style="height: 320px" />
+        </NSpin>
+      </NCard>
+
+      <!-- 站点日总量分页表 -->
+      <NCard :title="`站点日总量（${dateLabel}）`" size="small">
+        <template #header-extra>
+          <span class="text-xs text-neutral-400">共 {{ siteDailyPage.siteCount }} 站点</span>
+        </template>
+        <NDataTable
+          :remote="true"
+          size="small"
+          :columns="siteColumnsDaily"
+          :data="siteDailyPage.items"
+          :row-key="(row: SiteTotalItem) => row.equipNo"
+          :loading="loadingSiteDaily"
+          striped
+          :single-line="false"
+        />
+        <div class="mt-3 flex justify-end">
+          <NPagination
+            :page="siteDailyPagination.page"
+            :page-size="siteDailyPagination.pageSize"
+            :item-count="siteDailyPage.siteCount"
+            :page-sizes="[10, 20, 50]"
+            show-size-picker
+            @update:page="onSiteDailyPage"
+            @update:page-size="onSiteDailyPageSize"
+          />
+        </div>
+      </NCard>
+    </div>
+
+    <!-- 下钻：站点当日命令类型总量 -->
+    <NModal
+      v-model:show="showCmdType"
+      preset="card"
+      style="width: 640px"
+      :title="`站点命令类型总量 - ${drilldownEquipNo}（${dateLabel}）`"
+    >
+      <NDataTable
+        :remote="true"
+        size="small"
+        :columns="cmdTypeColumns"
+        :data="cmdTypePage.items"
+        :loading="loadingCmdType"
+        striped
+        :single-line="false"
+      />
+      <div class="mt-3 flex justify-end">
+        <NPagination
+          :page="cmdTypePagination.page"
+          :page-size="cmdTypePagination.pageSize"
+          :item-count="cmdTypePage.cmdTypeCount"
+          :page-sizes="[10, 20, 50]"
+          show-size-picker
+          @update:page="onCmdTypePage"
+          @update:page-size="onCmdTypePageSize"
+        />
+      </div>
+    </NModal>
+  </NSpin>
+</template>
+
+<script setup lang="tsx">
+import {
+  NSpin,
+  NCard,
+  NForm,
+  NFormItem,
+  NButton,
+  NDatePicker,
+  NDataTable,
+  NPagination,
+  NModal,
+  NTag,
+  useMessage,
+} from 'naive-ui'
+import type { DataTableColumns } from 'naive-ui'
+import { Chart } from '@antv/g2'
+import { usePersonalization } from '@/composable/usePersonalization'
+import InfoCard, { type InfoCardData } from '@/views/dashboard/component/infoCard.vue'
+import {
+  getMqttDailyTotal,
+  getMqttMonthlyTotal,
+  getMqttMonthlyDailyTrend,
+  getMqttSiteDailyTotals,
+  getMqttSiteCmdTypeTotals,
+} from '@/service/mqttDataManage'
+import type {
+  SiteTotalItem,
+  SiteTotalPage,
+  CmdTypeTotalPage,
+  CmdTypeTotalItem,
+  DailyTotal,
+} from '@/service/mqttDataManage/interface'
+
+defineOptions({ name: 'MqttDataDashboard' })
+
+const message = useMessage()
+const { isDark } = usePersonalization()
+
+// ------- 日期状态（默认今天，UTC） -------
+// 时区统一用 UTC：用户选择的“日历日”被当作 UTC 日处理。
+// 默认值取 UTC 今天对应的日历日（用本地 Date 承载该 y/m/d，保证选择器显示的就是 UTC 日期）。
+const nowUtc = new Date()
+const now = new Date(nowUtc.getUTCFullYear(), nowUtc.getUTCMonth(), nowUtc.getUTCDate())
+const selectedDateTs = ref<number | null>(now.getTime())
+
+const dateParts = computed(() => {
+  const d = selectedDateTs.value ? new Date(selectedDateTs.value) : now
+  // 选择器返回所选日历日的本地零点，用本地 getter 读到的正是用户点选的“日”，按 UTC 日语义使用
+  return { year: d.getFullYear(), month: d.getMonth() + 1, day: d.getDate() }
+})
+// 月份直接取“统计日期”所在的月，不再单独选择
+const monthParts = computed(() => ({ year: dateParts.value.year, month: dateParts.value.month }))
+
+// ES 日索引日期格式 yyyy.MM.dd
+const pad2 = (n: number) => String(n).padStart(2, '0')
+const esDate = computed(
+  () => `${dateParts.value.year}.${pad2(dateParts.value.month)}.${pad2(dateParts.value.day)}`,
+)
+const dateLabel = computed(
+  () => `${dateParts.value.year}-${pad2(dateParts.value.month)}-${pad2(dateParts.value.day)}`,
+)
+const monthLabel = computed(() => `${monthParts.value.year}-${pad2(monthParts.value.month)}`)
+// 当前选中日期（yyyy-MM-dd），用于在柱状图中高亮当天
+const selectedDateStr = computed(
+  () => `${dateParts.value.year}-${pad2(dateParts.value.month)}-${pad2(dateParts.value.day)}`,
+)
+// 今天（UTC）yyyy-MM-dd，用于判断“查询的是不是当天”
+const todayStr = `${now.getFullYear()}-${pad2(now.getMonth() + 1)}-${pad2(now.getDate())}`
+// 当前已加载月度数据(月总量+趋势图)对应的月份键 year*100+month，用于判断是否跨月
+let loadedMonthKey = 0
+
+// ------- 总量卡片（月总量 + 日总量） -------
+const monthlyTotal = ref(0)
+const dailyTotal = ref(0)
+const monthlyCard = computed<InfoCardData>(() => ({
+  title: `月总量（${monthLabel.value}）`,
+  value: monthlyTotal.value,
+  iconClass: 'i-ph:calendar-dots-bold',
+  backgroundColor: 'indigo',
+  precision: 0,
+}))
+const dailyCard = computed<InfoCardData>(() => ({
+  title: `日总量（${dateLabel.value}）`,
+  value: dailyTotal.value,
+  iconClass: 'i-material-symbols:sms-outline-rounded',
+  backgroundColor: 'blue',
+  precision: 0,
+}))
+
+// ------- 月度每日总量柱状图 -------
+const trendChartRef = ref<HTMLDivElement | null>(null)
+let trendChart: Chart | null = null
+let trendEventsBound = false // 确保图表点击事件只绑定一次，避免重复渲染叠加
+const loadingTrend = ref(false)
+const trendData = ref<DailyTotal[]>([])
+
+// 从柱状图选中某天：把 "yyyy-MM-dd" 转成与日期选择器一致的时间戳并回写，
+// 触发已有的 watch(selectedDateTs) 联动（刷新日维度数据 + 移动高亮）。
+const selectDateFromChart = (dateStr: string) => {
+  const [y, m, d] = dateStr.split('-').map(Number)
+  if (!y || !m || !d) return
+  const ts = new Date(y, m - 1, d).getTime()
+  if (ts !== selectedDateTs.value) {
+    selectedDateTs.value = ts
+  }
+}
+
+// Y 轴数值中文简写：>=1亿 用“亿”，>=1万 用“万”，否则原样（带千分位）
+const formatCnUnit = (v: number): string => {
+  const n = Number(v)
+  if (!Number.isFinite(n)) return String(v)
+  const abs = Math.abs(n)
+  if (abs >= 1e8) return `${+(n / 1e8).toFixed(2)}亿`
+  if (abs >= 1e4) return `${+(n / 1e4).toFixed(1)}万`
+  return n.toLocaleString()
+}
+
+// X 轴只取“日”（不带前导零），入参形如 "2026-09-01"
+const formatDayLabel = (date: string): string => {
+  const day = date.slice(8, 10)
+  return String(Number(day))
+}
+
+const renderTrendChart = () => {
+  if (!trendChartRef.value) return
+  if (!trendChart) {
+    trendChart = new Chart({ container: trendChartRef.value, autoFit: true })
+  }
+  trendChart.options({
+    type: 'interval', // 柱状图
+    autoFit: true,
+    theme: { type: isDark.value ? 'dark' : 'light' },
+    data: trendData.value,
+    encode: { x: 'date', y: 'total' },
+    style: {
+      // 选中当天稍微醒目：用主色高亮，其余用低饱和度的同色系，含蓄不刺眼
+      fill: (d: DailyTotal) => (d.date === selectedDateStr.value ? '#1668dc' : '#a3c2f0'),
+      cursor: 'pointer', // 提示柱子可点击
+    },
+    state: {
+      // 悬停高亮反馈
+      active: { fill: '#1668dc' },
+    },
+    interaction: {
+      elementHighlight: true, // 悬停高亮当前柱子
+    },
+    axis: {
+      // x 轴只显示“日”（不带前导零，如 1、2 … 28/29/30/31），月份见卡片标题
+      x: { title: false, labelFormatter: (v: string) => formatDayLabel(v) },
+      // y 轴用中文万/亿简写，减少 0 的数量
+      y: { title: false, labelFormatter: (v: number) => formatCnUnit(v) },
+    },
+    tooltip: {
+      title: (d: DailyTotal) => d.date, // 完整日期
+      items: [
+        // 悬停显示完整千分位精确数值
+        { field: 'total', name: '报文量', valueFormatter: (v: number) => v.toLocaleString() },
+      ],
+    },
+    animate: { enter: { type: 'scaleInY', duration: 600 } },
+  })
+
+  // 点击柱子切换日期（只绑一次，避免重复渲染叠加多个监听）
+  if (!trendEventsBound) {
+    trendEventsBound = true
+    trendChart.on('interval:click', (evt: { data?: { data?: DailyTotal } }) => {
+      const d = evt?.data?.data
+      if (d?.date) selectDateFromChart(d.date)
+    })
+  }
+
+  trendChart.render()
+}
+
+// ------- 站点日总量表 -------
+const emptySitePage = (): SiteTotalPage => ({ page: 1, pageSize: 20, siteCount: 0, items: [] })
+const siteDailyPage = ref<SiteTotalPage>(emptySitePage())
+const loadingSiteDaily = ref(false)
+const siteDailyPagination = reactive({ page: 1, pageSize: 20 })
+
+// ------- 下钻：命令类型 -------
+const drilldownEquipNo = ref('')
+const showCmdType = ref(false)
+const cmdTypePage = ref<CmdTypeTotalPage>({ page: 1, pageSize: 20, cmdTypeCount: 0, items: [] })
+const loadingCmdType = ref(false)
+const cmdTypePagination = reactive({ page: 1, pageSize: 20 })
+
+const isInit = ref(true)
+
+// ------- 列定义 -------
+const siteColumnsDaily = computed<DataTableColumns<SiteTotalItem>>(() => [
+  {
+    title: 'No.',
+    key: 'no',
+    width: 60,
+    render: (_, index) =>
+      `${(siteDailyPagination.page - 1) * siteDailyPagination.pageSize + index + 1}`,
+  },
+  {
+    title: '站点(equipNo)',
+    key: 'equipNo',
+    render: (row) => (
+      <NButton text type="primary" onClick={() => openCmdType(row.equipNo)}>
+        {row.equipNo}
+      </NButton>
+    ),
+  },
+  { title: '报文量', key: 'total', width: 140, render: (row) => row.total.toLocaleString() },
+])
+
+const cmdTypeColumns = computed<DataTableColumns<CmdTypeTotalItem>>(() => [
+  {
+    title: 'No.',
+    key: 'no',
+    width: 60,
+    render: (_, index) =>
+      `${(cmdTypePagination.page - 1) * cmdTypePagination.pageSize + index + 1}`,
+  },
+  { title: '命令类型(cmdType)', key: 'cmdType' },
+  { title: '报文量', key: 'total', width: 140, render: (row) => row.total.toLocaleString() },
+])
+
+// ------- 数据加载 -------
+const handleError = (err: unknown) => {
+  message.error(err instanceof Error ? err.message : String(err))
+}
+
+const loadDailyTotal = async () => {
+  try {
+    const d = await getMqttDailyTotal(esDate.value)
+    dailyTotal.value = d.total
+  } catch (err) {
+    handleError(err)
+  }
+}
+
+const loadMonthlyTotal = async () => {
+  try {
+    const m = await getMqttMonthlyTotal(monthParts.value.year, monthParts.value.month)
+    monthlyTotal.value = m.total
+  } catch (err) {
+    handleError(err)
+  }
+}
+
+const loadTrend = async () => {
+  loadingTrend.value = true
+  try {
+    trendData.value = await getMqttMonthlyDailyTrend(monthParts.value.year, monthParts.value.month)
+    loadedMonthKey = monthParts.value.year * 100 + monthParts.value.month
+    renderTrendChart()
+  } catch (err) {
+    handleError(err)
+  } finally {
+    loadingTrend.value = false
+  }
+}
+
+const loadSiteDaily = async () => {
+  loadingSiteDaily.value = true
+  try {
+    siteDailyPage.value = await getMqttSiteDailyTotals(
+      esDate.value,
+      siteDailyPagination.page,
+      siteDailyPagination.pageSize,
+    )
+  } catch (err) {
+    handleError(err)
+  } finally {
+    loadingSiteDaily.value = false
+  }
+}
+
+const loadCmdType = async () => {
+  loadingCmdType.value = true
+  try {
+    cmdTypePage.value = await getMqttSiteCmdTypeTotals(
+      drilldownEquipNo.value,
+      esDate.value,
+      cmdTypePagination.page,
+      cmdTypePagination.pageSize,
+    )
+  } catch (err) {
+    handleError(err)
+  } finally {
+    loadingCmdType.value = false
+  }
+}
+
+const openCmdType = async (equipNo: string) => {
+  // 释放触发按钮的焦点，避免 NModal 打开后背景 aria-hidden 内仍含聚焦元素（a11y 警告）
+  ;(document.activeElement as HTMLElement | null)?.blur()
+  drilldownEquipNo.value = equipNo
+  cmdTypePagination.page = 1
+  showCmdType.value = true
+  await loadCmdType()
+}
+
+// ------- 分页事件 -------
+const onSiteDailyPage = (p: number) => {
+  siteDailyPagination.page = p
+  loadSiteDaily()
+}
+const onSiteDailyPageSize = (s: number) => {
+  siteDailyPagination.pageSize = s
+  siteDailyPagination.page = 1
+  loadSiteDaily()
+}
+const onCmdTypePage = (p: number) => {
+  cmdTypePagination.page = p
+  loadCmdType()
+}
+const onCmdTypePageSize = (s: number) => {
+  cmdTypePagination.pageSize = s
+  cmdTypePagination.page = 1
+  loadCmdType()
+}
+
+// ------- 日期/月份变更 -------
+// 选中日期变化时的刷新策略：
+//  - 日总量、站点日总量：始终刷新（它们按“天”维度）。
+//  - 月总量、月度柱状图：仅在【跨月】或【查询的是当天】时刷新。
+//    · 同月内切换历史日期：不重新请求，只把柱状图高亮移到该天（本地重渲染）。
+//    · 查询当天：当天数据还在增长，需刷新月总量并更新柱状图当天柱子。
+watch(selectedDateTs, () => {
+  if (isInit.value) return
+
+  // 日维度：总是刷新
+  siteDailyPagination.page = 1
+  loadDailyTotal()
+  loadSiteDaily()
+
+  // 月维度：按需刷新
+  const curMonthKey = monthParts.value.year * 100 + monthParts.value.month
+  const monthChanged = curMonthKey !== loadedMonthKey
+  const isToday = selectedDateStr.value === todayStr
+
+  if (monthChanged || isToday) {
+    // 跨月 或 查当天：重新拉月总量 + 月度趋势
+    loadMonthlyTotal()
+    loadTrend()
+  } else {
+    // 同月历史日：数据无需变化，仅移动高亮
+    renderTrendChart()
+  }
+})
+
+const refreshAll = async () => {
+  await Promise.all([loadMonthlyTotal(), loadTrend(), loadDailyTotal(), loadSiteDaily()])
+}
+
+onMounted(async () => {
+  await refreshAll()
+  isInit.value = false
+})
+
+onUnmounted(() => {
+  if (trendChart) {
+    trendChart.destroy()
+    trendChart = null
+    trendEventsBound = false
+  }
+})
+
+// 主题切换时重渲染图表
+watch(isDark, () => {
+  if (trendChart) {
+    trendChart.theme({ type: isDark.value ? 'dark' : 'light' })
+    trendChart.render()
+  }
+})
+</script>

src/views/mqtt-data-manage/list/index.vue

@@ -0,0 +1,264 @@
+<template>
+  <div class="main-wrap p-4 flex flex-col gap-y-4">
+    <NCard
+      ref="mainCardRef"
+      class="h-full"
+      content-class="h-full flex flex-col"
+      :style="cardHeight"
+    >
+      <!-- 过滤条件 -->
+      <div class="flex gap-2 min-h-6 items-center">
+        <NForm
+          inline
+          label-placement="left"
+          label-width="auto"
+          :show-feedback="false"
+          class="flex-1 flex-wrap"
+        >
+          <NFormItem label="站点(equipNo)">
+            <NInput
+              v-model:value="filterEquipNo"
+              clearable
+              placeholder="设备号"
+              class="w-200px"
+              size="small"
+            />
+          </NFormItem>
+          <NFormItem label="命令类型">
+            <NInputNumber
+              v-model:value="filterCmdType"
+              clearable
+              placeholder="cmdType"
+              class="w-140px"
+              size="small"
+              :show-button="false"
+            />
+          </NFormItem>
+          <NFormItem label="时间范围(UTC)">
+            <NDatePicker
+              v-model:value="filterTimeRange"
+              type="datetimerange"
+              clearable
+              size="small"
+            />
+          </NFormItem>
+          <NFormItem>
+            <NButton type="primary" size="small" @click="clickSearch">
+              <template #icon><span class="i-mdi:search" /></template>
+              查询
+            </NButton>
+          </NFormItem>
+          <NFormItem>
+            <NButton size="small" @click="clickReset">重置</NButton>
+          </NFormItem>
+        </NForm>
+      </div>
+      <NDivider style="margin: 16px 0" />
+
+      <!-- 数据表 -->
+      <div class="flex flex-1 flex-col min-h-0">
+        <div ref="tableWrapper" class="flex-1 min-h-0 overflow-hidden">
+          <NDataTable
+            :remote="true"
+            size="small"
+            :columns="tableColumns"
+            :data="tableData"
+            :row-key="tableRowKey"
+            :loading="isRequestLoading"
+            striped
+            :single-line="false"
+            :style="{ height: `${tableHeight}px` }"
+            flex-height
+          />
+        </div>
+        <div class="mt-3 flex items-center justify-between">
+          <div class="text-xs text-neutral-400 flex flex-col gap-y-0.5">
+            <span>默认查询当天数据，按时间降序（最新在前）；时间均为 UTC</span>
+            <span>
+              共 <b>{{ realTotal.toLocaleString() }}</b> 条
+              <template v-if="isTruncated">
+                <span class="text-amber-500">
+                  （仅可浏览前 {{ MAX_WINDOW.toLocaleString() }} 条，请缩小时间范围或增加设备号/命令类型过滤）
+                </span>
+              </template>
+            </span>
+          </div>
+          <NPagination v-bind="tablePagination" />
+        </div>
+      </div>
+    </NCard>
+  </div>
+</template>
+
+<script setup lang="tsx">
+import {
+  NCard,
+  NForm,
+  NFormItem,
+  NInput,
+  NInputNumber,
+  NDatePicker,
+  NButton,
+  NDivider,
+  NDataTable,
+  NPagination,
+  useMessage,
+} from 'naive-ui'
+import type { PaginationProps, DataTableColumns } from 'naive-ui'
+import { useFullCard } from '@/composable/comp/useFullCard'
+import { queryMqttMessages } from '@/service/mqttDataManage'
+import type { MqttMessageItem, QueryMessageParams } from '@/service/mqttDataManage/interface'
+
+defineOptions({ name: 'MqttDataList' })
+
+const message = useMessage()
+
+const mainCardRef = useTemplateRef<InstanceType<typeof NCard>>('mainCardRef')
+const { height } = useFullCard(mainCardRef)
+const cardHeight = computed(() => ({
+  height: `${height.value === undefined || height.value === 0 ? 480 : height.value}px`,
+}))
+
+// ------- 过滤条件 -------
+const filterEquipNo = ref('')
+const filterCmdType = ref<number | null>(null)
+const filterTimeRange = ref<[number, number] | null>(null)
+
+// ------- 表格 -------
+const isRequestLoading = ref(false)
+const tableData = ref<MqttMessageItem[]>([])
+const tableWrapper = ref<HTMLDivElement | null>(null)
+const tableHeight = ref(100)
+
+// ES from+size 深分页窗口上限（与后端 MaxMessagePageWindow 一致）：
+// 分页控件最多翻到前 MAX_WINDOW 条，超出会被后端拒绝。
+const MAX_WINDOW = 10000
+// 真实命中总数（可能远大于 MAX_WINDOW），用于文字展示
+const realTotal = ref(0)
+// 是否被窗口截断（真实总数超过可翻页范围）
+const isTruncated = computed(() => realTotal.value > MAX_WINDOW)
+
+const tableRowKey = (row: MqttMessageItem) => `${row.equipNo}-${row.timeStamp}-${row.cmdType}`
+
+// 时区统一 UTC。
+// 记录展示：timeStamp(秒) 为 UTC 绝对时刻，用 UTC getter 显示其 UTC 墙钟时间。
+const pad2 = (n: number) => String(n).padStart(2, '0')
+const formatUtcFromSeconds = (sec: number) => {
+  const d = new Date(sec * 1000)
+  return (
+    `${d.getUTCFullYear()}-${pad2(d.getUTCMonth() + 1)}-${pad2(d.getUTCDate())} `
+    + `${pad2(d.getUTCHours())}:${pad2(d.getUTCMinutes())}:${pad2(d.getUTCSeconds())}`
+  )
+}
+// 查询范围：日期选择器返回所选“墙钟时间”的本地毫秒，用户按 UTC 语义理解，
+// 故用本地 getter 取出所选墙钟并原样作为 UTC 字符串传给后端（后端按 UTC 解析）。
+const formatPickedAsUtc = (ms: number) => {
+  const d = new Date(ms)
+  return (
+    `${d.getFullYear()}-${pad2(d.getMonth() + 1)}-${pad2(d.getDate())} `
+    + `${pad2(d.getHours())}:${pad2(d.getMinutes())}:${pad2(d.getSeconds())}`
+  )
+}
+
+const tableColumns = computed<DataTableColumns<MqttMessageItem>>(() => [
+  {
+    title: 'No.',
+    key: 'no',
+    width: 60,
+    render: (_, index) => {
+      const page = tablePagination.page ?? 1
+      const pageSize = tablePagination.pageSize ?? 20
+      return `${(page - 1) * pageSize + index + 1}`
+    },
+  },
+  { title: '站点(equipNo)', key: 'equipNo', width: 220 },
+  { title: '命令类型(cmdType)', key: 'cmdType', width: 150 },
+  {
+    title: '时间(UTC)',
+    key: 'timeStamp',
+    width: 200,
+    render: (row) => formatUtcFromSeconds(row.timeStamp),
+  },
+  {
+    title: 'event',
+    key: 'event',
+    render: (row) =>
+      row.event && Object.keys(row.event).length > 0 ? JSON.stringify(row.event) : '-',
+  },
+])
+
+const tablePagination = reactive<PaginationProps>({
+  page: 1,
+  pageSize: 20,
+  showSizePicker: true,
+  pageSizes: [20, 50, 100],
+  itemCount: 0,
+  showQuickJumper: true,
+  onUpdatePage: (page: number) => {
+    tablePagination.page = page
+    handleQuery()
+  },
+  onUpdatePageSize: (pageSize: number) => {
+    tablePagination.pageSize = pageSize
+    tablePagination.page = 1
+    handleQuery()
+  },
+})
+
+// ------- 查询 -------
+const handleQuery = async () => {
+  isRequestLoading.value = true
+  try {
+    const params: QueryMessageParams = {
+      page: tablePagination.page ?? 1,
+      size: tablePagination.pageSize ?? 20,
+    }
+    if (filterEquipNo.value.trim()) params.equipNo = filterEquipNo.value.trim()
+    if (filterCmdType.value !== null) params.cmdType = filterCmdType.value
+    if (filterTimeRange.value) {
+      params.startTime = formatPickedAsUtc(filterTimeRange.value[0])
+      params.endTime = formatPickedAsUtc(filterTimeRange.value[1])
+    }
+
+    const res = await queryMqttMessages(params)
+    tableData.value = res.items ?? []
+    realTotal.value = res.total
+    // ES from+size 深分页上限：分页控件最多允许翻到前 MAX_WINDOW 条，避免深分页请求失败。
+    // 真实总数另用文字展示。
+    tablePagination.itemCount = Math.min(res.total, MAX_WINDOW)
+  } catch (err) {
+    tableData.value = []
+    realTotal.value = 0
+    tablePagination.itemCount = 0
+    message.error(err instanceof Error ? err.message : String(err))
+  } finally {
+    isRequestLoading.value = false
+  }
+}
+
+const clickSearch = () => {
+  tablePagination.page = 1
+  handleQuery()
+}
+
+const clickReset = () => {
+  filterEquipNo.value = ''
+  filterCmdType.value = null
+  filterTimeRange.value = null
+  tablePagination.page = 1
+  handleQuery()
+}
+
+onMounted(async () => {
+  await handleQuery()
+
+  nextTick(() => {
+    const observer = new ResizeObserver((entries) => {
+      tableHeight.value = entries[0].contentRect.height
+    })
+    if (tableWrapper.value) {
+      observer.observe(tableWrapper.value)
+    }
+  })
+})
+</script>