FranklinWH-Toolbox/service/repositories/api/fwh_es_mqtt/fwh_es_mqtt.go

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 {
			item := h.Source
			// _id 仅在单个索引内唯一，跨日索引查询时需带上索引名
			item.DocID = h.Index + "/" + h.ID
			out.Items = append(out.Items, item)
		}
	}
	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 + "-*"
}