kenaito-conf/server/internal/handler/config_api.go

package handler

import (
	"fmt"
	"kenaito-conf/server/internal/dao"
	"kenaito-conf/server/internal/logger"
	"kenaito-conf/server/internal/model"
	"kenaito-conf/server/internal/service"
	"net"
	"net/http"
	"strconv"
	"strings"
	"time"

	"github.com/gin-gonic/gin"
)

// getClientIP 获取客户端 IPv4 地址，将 IPv6 环回地址转为 127.0.0.1
func getClientIP(c *gin.Context) string {
	ip := c.ClientIP()
	if ip == "" {
		return ""
	}
	parsed := net.ParseIP(ip)
	if parsed == nil {
		return ip
	}
	if parsed.IsLoopback() {
		return "127.0.0.1"
	}
	if v4 := parsed.To4(); v4 != nil {
		return v4.String()
	}
	return ip
}

type ConfigApiHandler struct {
	parseService *service.ConfigParseService
}

func NewConfigApiHandler(parse *service.ConfigParseService) *ConfigApiHandler {
	return &ConfigApiHandler{parseService: parse}
}

type FetchAllRequest struct {
	AppName      string `json:"appName" binding:"required"`
	Env          string `json:"env" binding:"required"`
	Version      string `json:"version"`
	IgnoreGlobal bool   `json:"ignoreGlobal"`
	Token        string `json:"token"`
}

func (h *ConfigApiHandler) FetchAllConfigs(c *gin.Context) {
	var req FetchAllRequest
	if err := c.ShouldBindJSON(&req); err != nil {
		c.JSON(http.StatusBadRequest, model.WrapperResponse{Code: 400, Msg: "参数错误"})
		return
	}

	version := req.Version
	if version == "" {
		version = "latest"
	}

	if req.AppName == "" || req.Env == "" {
		c.JSON(http.StatusBadRequest, model.WrapperResponse{Code: 400, Msg: "参数缺失"})
		return
	}

	// find app by code
	var app model.Application
	has, _ := dao.Engine.Where("code = ? AND enabled = 1", req.AppName).Get(&app)
	if !has {
		c.JSON(http.StatusOK, model.WrapperResponse{Code: 1001, Msg: "配置不存在"})
		return
	}

	// check if app's group is enabled
	if app.GroupId != nil {
		var group model.BusinessGroup
		has, _ := dao.Engine.ID(*app.GroupId).Get(&group)
		if !has || !group.Enabled {
			c.JSON(http.StatusForbidden, model.WrapperResponse{Code: 403, Msg: "业务组已禁用"})
			return
		}
	}

	// validate token if app has token configured
	// 兼容旧版客户端把 token 放在 URL query 中的行为
	token := req.Token
	if token == "" {
		token = c.Query("token")
	}
	if app.Token != "" && app.Token != token {
		c.JSON(http.StatusForbidden, model.WrapperResponse{Code: 403, Msg: "Token 无效"})
		return
	}

	// 游标上界取查询前的当前时间，保证查询期间发生的更新会在下次增量中返回
	upperBound := time.Now()

	result := make(map[string]interface{})
	var configIds []int

	// load global configs
	if !req.IgnoreGlobal && app.GroupId != nil {
		var globalConfigs []model.AppConfig
		dao.Engine.Where("env = ? AND version = ? AND enabled = 1 AND group_id = ? AND (app_id is null OR app_id = 0)",
			req.Env, version, *app.GroupId).Find(&globalConfigs)
		for _, cfg := range globalConfigs {
			kvMap, err := h.parseService.ParseConfigToKVMap(&cfg)
			if err != nil {
				logger.SugaredLogger.Errorf("[FetchAllConfigs] parse config %d (%s) error: %v", cfg.Id, cfg.Name, err)
				continue
			}
			configIds = append(configIds, cfg.Id)
			for k, v := range kvMap {
				result[k] = v
			}
		}
	}

	// load app configs
	var appConfigs []model.AppConfig
	dao.Engine.Where("env = ? AND version = ? AND enabled = 1 AND app_id = ?",
		req.Env, version, app.Id).Find(&appConfigs)
	for _, cfg := range appConfigs {
		kvMap, err := h.parseService.ParseConfigToKVMap(&cfg)
		if err != nil {
			logger.SugaredLogger.Errorf("[FetchAllConfigs] parse config %d (%s) error: %v", cfg.Id, cfg.Name, err)
			continue
		}
		configIds = append(configIds, cfg.Id)
		for k, v := range kvMap {
			result[k] = v
		}
	}

	if len(result) == 0 {
		c.JSON(http.StatusOK, model.WrapperResponse{Code: 1001, Msg: "配置不存在"})
		return
	}

	result["latest.fetch.timestamp"] = strconv.FormatInt(upperBound.UnixMilli(), 10)

	// record client instances asynchronously
	go h.recordClientInstances(getClientIP(c), configIds, req.AppName, req.Env, "fetchAll")

	c.JSON(http.StatusOK, result)
}

type FetchChangedRequest struct {
	AppName      string `json:"appName" binding:"required"`
	Env          string `json:"env" binding:"required"`
	Version      string `json:"version"`
	LastTime     int64  `json:"lastTime" binding:"required"`
	IgnoreGlobal bool   `json:"ignoreGlobal"`
	Token        string `json:"token"`
}

func (h *ConfigApiHandler) FetchChangedConfigs(c *gin.Context) {
	var req FetchChangedRequest
	if err := c.ShouldBindJSON(&req); err != nil {
		c.JSON(http.StatusBadRequest, model.WrapperResponse{Code: 400, Msg: "参数错误"})
		return
	}

	var app model.Application
	has, _ := dao.Engine.Where("code = ? AND enabled = 1", req.AppName).Get(&app)
	if !has {
		c.JSON(http.StatusOK, model.WrapperResponse{Code: 1001, Msg: "应用不存在"})
		return
	}

	// validate token if app has token configured
	// 兼容旧版客户端把 token 放在 URL query 中的行为
	token := req.Token
	if token == "" {
		token = c.Query("token")
	}
	if app.Token != "" && app.Token != token {
		c.JSON(http.StatusForbidden, model.WrapperResponse{Code: 403, Msg: "Token 无效"})
		return
	}

	version := req.Version
	if version == "" {
		version = "latest"
	}

	lastUpdateTime := time.UnixMilli(req.LastTime)

	// 以当前时间为本次查询的固定上界，避免查询与游标生成之间的竞态
	upperBound := time.Now()

	result := make(map[string]string)
	var configIds []int

	// 查询已删除的配置（enabled=0 且 updated_at > lastTime），返回 deletedKeys
	var deletedKeys []string
	if !req.IgnoreGlobal && app.GroupId != nil {
		var deletedGlobalConfigs []model.AppConfig
		dao.Engine.Where("env = ? AND version = ? AND enabled = 0 AND group_id = ? AND (app_id is null OR app_id = 0) AND updated_at > ? AND updated_at <= ?",
			req.Env, version, *app.GroupId, lastUpdateTime, upperBound).Find(&deletedGlobalConfigs)
		for _, cfg := range deletedGlobalConfigs {
			kvMap, err := h.parseService.ParseConfigToKVMap(&cfg)
			if err != nil {
				logger.SugaredLogger.Errorf("[FetchChangedConfigs] parse deleted config %d (%s) error: %v", cfg.Id, cfg.Name, err)
				continue
			}
			for k := range kvMap {
				deletedKeys = append(deletedKeys, k)
			}
		}
	}
	var deletedAppConfigs []model.AppConfig
	dao.Engine.Where("env = ? AND version = ? AND enabled = 0 AND app_id = ? AND updated_at > ? AND updated_at <= ?",
		req.Env, version, app.Id, lastUpdateTime, upperBound).Find(&deletedAppConfigs)
	for _, cfg := range deletedAppConfigs {
		kvMap, err := h.parseService.ParseConfigToKVMap(&cfg)
		if err != nil {
			logger.SugaredLogger.Errorf("[FetchChangedConfigs] parse deleted config %d (%s) error: %v", cfg.Id, cfg.Name, err)
			continue
		}
		for k := range kvMap {
			deletedKeys = append(deletedKeys, k)
		}
	}
	if len(deletedKeys) > 0 {
		result["deletedKeys"] = strings.Join(deletedKeys, ",")
	}

	// load global changed configs
	if !req.IgnoreGlobal && app.GroupId != nil {
		var globalConfigs []model.AppConfig
		dao.Engine.Where("env = ? AND version = ? AND enabled = 1 AND group_id = ? AND (app_id is null OR app_id = 0) AND updated_at > ? AND updated_at <= ?",
			req.Env, version, *app.GroupId, lastUpdateTime, upperBound).Find(&globalConfigs)
		for _, cfg := range globalConfigs {
			if h.addChangedConfigs(result, &cfg) {
				configIds = append(configIds, cfg.Id)
			}
		}
	}

	// load app changed configs
	var appConfigs1 []model.AppConfig
	dao.Engine.Where("env = ? AND version = ? AND enabled = 1 AND app_id = ? AND updated_at > ? AND updated_at <= ?",
		req.Env, version, app.Id, lastUpdateTime, upperBound).Find(&appConfigs1)
	for _, cfg := range appConfigs1 {
		if h.addChangedConfigs(result, &cfg) {
			configIds = append(configIds, cfg.Id)
		}
	}

	result["latest.fetch.timestamp"] = strconv.FormatInt(upperBound.UnixMilli(), 10)

	// query all config ids for this app to record client instances
	var allConfigIds []int
	if !req.IgnoreGlobal && app.GroupId != nil {
		var globalConfigs []model.AppConfig
		dao.Engine.Where("env = ? AND version = ? AND enabled = 1 AND group_id = ? AND (app_id is null OR app_id = 0)",
			req.Env, version, *app.GroupId).Find(&globalConfigs)
		for _, cfg := range globalConfigs {
			allConfigIds = append(allConfigIds, cfg.Id)
		}
	}
	var appConfigs2 []model.AppConfig
	dao.Engine.Where("env = ? AND version = ? AND enabled = 1 AND app_id = ?",
		req.Env, version, app.Id).Find(&appConfigs2)
	for _, cfg := range appConfigs2 {
		allConfigIds = append(allConfigIds, cfg.Id)
	}

	// record client instances asynchronously
	// only update last_fetch_at if there are actual changes
	fetchType := ""
	if len(configIds) > 0 {
		fetchType = "fetchChanged"
	}
	go h.recordClientInstances(getClientIP(c), allConfigIds, req.AppName, req.Env, fetchType)

	c.JSON(http.StatusOK, model.WrapperResponse{Code: 0, Data: result})
}

func (h *ConfigApiHandler) addChangedConfigs(result map[string]string, config *model.AppConfig) bool {
	kvMap, err := h.parseService.ParseConfigToKVMap(config)
	if err != nil {
		logger.SugaredLogger.Errorf("[FetchChangedConfigs] parse config %d (%s) error: %v", config.Id, config.Name, err)
		return false
	}
	for k, v := range kvMap {
		result[k] = fmt.Sprintf("%v", v)
	}
	return true
}

func (h *ConfigApiHandler) recordClientInstances(ip string, configIds []int, appName, env, fetchType string) {
	if len(configIds) == 0 || ip == "" {
		return
	}
	now := time.Now()

	// 批量查询已存在的记录
	var existingList []model.ClientInstance
	dao.Engine.Where("app_name = ? AND env = ? AND ip = ?", appName, env, ip).
		In("app_config_id", configIds).
		Find(&existingList)

	existingMap := make(map[int]*model.ClientInstance, len(existingList))
	for i := range existingList {
		existingMap[existingList[i].AppConfigId] = &existingList[i]
	}

	var toInsert []model.ClientInstance
	for _, configId := range configIds {
		if existing, ok := existingMap[configId]; ok {
			existing.LastPingAt = &now
			existing.UpdatedAt = &now
			if fetchType != "" {
				existing.LastFetchAt = &now
				existing.LastFetchType = fetchType
				dao.Engine.ID(existing.Id).Cols("last_ping_at", "last_fetch_at", "last_fetch_type", "updated_at").Update(existing)
			} else {
				dao.Engine.ID(existing.Id).Cols("last_ping_at", "updated_at").Update(existing)
			}
		} else {
			instance := model.ClientInstance{
				AppConfigId: configId,
				AppName:     appName,
				Env:         env,
				Ip:          ip,
				LastPingAt:  &now,
				CreatedAt:   &now,
				UpdatedAt:   &now,
			}
			if fetchType != "" {
				instance.LastFetchAt = &now
				instance.LastFetchType = fetchType
			}
			toInsert = append(toInsert, instance)
		}
	}

	if len(toInsert) > 0 {
		if _, err := dao.Engine.Insert(&toInsert); err != nil {
			logger.SugaredLogger.Errorf("[recordClientInstances] batch insert error: %v", err)
		}
	}
}

// func (h *ConfigApiHandler) Ping(c *gin.Context) {
// 	c.String(http.StatusOK, "200")
// }