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

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)
}