ez-api/internal/api/internal_handler.go

package api

import (
	"net/http"
	"strings"
	"time"

	"github.com/ez-api/ez-api/internal/model"
	"github.com/gin-gonic/gin"
	"gorm.io/gorm"
)

type InternalHandler struct {
	db *gorm.DB
}

func NewInternalHandler(db *gorm.DB) *InternalHandler {
	return &InternalHandler{db: db}
}

type statsFlushRequest struct {
	Keys []statsFlushEntry `json:"keys"`
}

type statsFlushEntry struct {
	TokenHash      string `json:"token_hash"`
	Requests       int64  `json:"requests"`
	Tokens         int64  `json:"tokens"`
	LastAccessedAt int64  `json:"last_accessed_at"`
}

type apiKeyStatsFlushRequest struct {
	Keys []apiKeyStatsFlushEntry `json:"keys"`
}

type apiKeyStatsFlushEntry struct {
	APIKeyID        uint  `json:"api_key_id"`
	Requests        int64 `json:"requests"`
	SuccessRequests int64 `json:"success_requests"`
}

// FlushStats godoc
// @Summary      Flush key stats
// @Description  Internal endpoint for flushing accumulated key usage stats from DP to CP database
// @Tags         internal
// @Accept       json
// @Produce      json
// @Param        request  body      statsFlushRequest  true  "Stats to flush"
// @Success      200      {object}  gin.H
// @Failure      400      {object}  gin.H
// @Failure      500      {object}  gin.H
// @Router       /internal/stats/flush [post]
func (h *InternalHandler) FlushStats(c *gin.Context) {
	if h == nil || h.db == nil {
		c.JSON(http.StatusInternalServerError, gin.H{"error": "database not configured"})
		return
	}

	var req statsFlushRequest
	if err := c.ShouldBindJSON(&req); err != nil {
		c.JSON(http.StatusBadRequest, gin.H{"error": "invalid request body"})
		return
	}
	if len(req.Keys) == 0 {
		c.JSON(http.StatusOK, gin.H{"updated": 0})
		return
	}

	updated := 0
	err := h.db.Transaction(func(tx *gorm.DB) error {
		for _, entry := range req.Keys {
			hash := strings.TrimSpace(entry.TokenHash)
			if hash == "" {
				continue
			}
			if entry.Requests < 0 || entry.Tokens < 0 || entry.LastAccessedAt < 0 {
				return gorm.ErrInvalidData
			}

			updates := map[string]any{}
			if entry.Requests > 0 {
				updates["request_count"] = gorm.Expr("request_count + ?", entry.Requests)
			}
			if entry.Tokens > 0 {
				updates["used_tokens"] = gorm.Expr("used_tokens + ?", entry.Tokens)
				updates["quota_used"] = gorm.Expr("CASE WHEN quota_limit >= 0 THEN quota_used + ? ELSE quota_used END", entry.Tokens)
			}
			if entry.LastAccessedAt > 0 {
				accessTime := time.Unix(entry.LastAccessedAt, 0).UTC()
				updates["last_accessed_at"] = gorm.Expr("CASE WHEN last_accessed_at IS NULL OR last_accessed_at < ? THEN ? ELSE last_accessed_at END", accessTime, accessTime)
			}

			if len(updates) == 0 {
				continue
			}

			res := tx.Model(&model.Key{}).Where("token_hash = ?", hash).Updates(updates)
			if res.Error != nil {
				return res.Error
			}
			if res.RowsAffected > 0 {
				updated++
			}
		}
		return nil
	})
	if err != nil {
		if err == gorm.ErrInvalidData {
			c.JSON(http.StatusBadRequest, gin.H{"error": "invalid stats payload"})
			return
		}
		c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to flush stats", "details": err.Error()})
		return
	}

	c.JSON(http.StatusOK, gin.H{"updated": updated})
}

// FlushAPIKeyStats godoc
// @Summary      Flush API key stats
// @Description  Internal endpoint for flushing accumulated APIKey stats from DP to CP database
// @Tags         internal
// @Accept       json
// @Produce      json
// @Param        request  body      apiKeyStatsFlushRequest  true  "Stats to flush"
// @Success      200      {object}  gin.H
// @Failure      400      {object}  gin.H
// @Failure      500      {object}  gin.H
// @Router       /internal/apikey-stats/flush [post]
func (h *InternalHandler) FlushAPIKeyStats(c *gin.Context) {
	if h == nil || h.db == nil {
		c.JSON(http.StatusInternalServerError, gin.H{"error": "database not configured"})
		return
	}

	var req apiKeyStatsFlushRequest
	if err := c.ShouldBindJSON(&req); err != nil {
		c.JSON(http.StatusBadRequest, gin.H{"error": "invalid request body"})
		return
	}
	if len(req.Keys) == 0 {
		c.JSON(http.StatusOK, gin.H{"updated": 0, "groups_updated": 0})
		return
	}

	type apiKeyDelta struct {
		requests int64
		success  int64
	}

	deltas := make(map[uint]apiKeyDelta, len(req.Keys))
	for _, entry := range req.Keys {
		if entry.APIKeyID == 0 {
			continue
		}
		if entry.Requests < 0 || entry.SuccessRequests < 0 || entry.SuccessRequests > entry.Requests {
			c.JSON(http.StatusBadRequest, gin.H{"error": "invalid stats payload"})
			return
		}
		if entry.Requests == 0 && entry.SuccessRequests == 0 {
			continue
		}
		delta := deltas[entry.APIKeyID]
		delta.requests += entry.Requests
		delta.success += entry.SuccessRequests
		deltas[entry.APIKeyID] = delta
	}

	if len(deltas) == 0 {
		c.JSON(http.StatusOK, gin.H{"updated": 0, "groups_updated": 0})
		return
	}

	ids := make([]uint, 0, len(deltas))
	for id := range deltas {
		ids = append(ids, id)
	}

	var apiKeys []model.APIKey
	if err := h.db.Model(&model.APIKey{}).Select("id, group_id").Where("id IN ?", ids).Find(&apiKeys).Error; err != nil {
		c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to load api keys", "details": err.Error()})
		return
	}
	groupByKey := make(map[uint]uint, len(apiKeys))
	for _, key := range apiKeys {
		groupByKey[key.ID] = key.GroupID
	}

	statsUpdates := func(requests, success int64) map[string]any {
		return map[string]any{
			"total_requests":   gorm.Expr("total_requests + ?", requests),
			"success_requests": gorm.Expr("success_requests + ?", success),
			"failure_requests": gorm.Expr("(total_requests + ?) - (success_requests + ?)", requests, success),
			"success_rate": gorm.Expr(
				"CASE WHEN (total_requests + ?) > 0 THEN (success_requests + ?) * 1.0 / (total_requests + ?) ELSE 0 END",
				requests, success, requests,
			),
			"failure_rate": gorm.Expr(
				"CASE WHEN (total_requests + ?) > 0 THEN ((total_requests + ?) - (success_requests + ?)) * 1.0 / (total_requests + ?) ELSE 0 END",
				requests, requests, success, requests,
			),
		}
	}

	updated := 0
	groupsUpdated := 0
	groupDeltas := make(map[uint]apiKeyDelta)

	err := h.db.Transaction(func(tx *gorm.DB) error {
		for id, delta := range deltas {
			groupID, ok := groupByKey[id]
			if !ok {
				continue
			}
			if delta.requests == 0 && delta.success == 0 {
				continue
			}
			res := tx.Model(&model.APIKey{}).Where("id = ?", id).Updates(statsUpdates(delta.requests, delta.success))
			if res.Error != nil {
				return res.Error
			}
			if res.RowsAffected > 0 {
				updated++
			}
			if groupID > 0 {
				groupDelta := groupDeltas[groupID]
				groupDelta.requests += delta.requests
				groupDelta.success += delta.success
				groupDeltas[groupID] = groupDelta
			}
		}

		for groupID, delta := range groupDeltas {
			if delta.requests == 0 && delta.success == 0 {
				continue
			}
			res := tx.Model(&model.ProviderGroup{}).Where("id = ?", groupID).Updates(statsUpdates(delta.requests, delta.success))
			if res.Error != nil {
				return res.Error
			}
			if res.RowsAffected > 0 {
				groupsUpdated++
			}
		}
		return nil
	})
	if err != nil {
		c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to flush api key stats", "details": err.Error()})
		return
	}

	c.JSON(http.StatusOK, gin.H{"updated": updated, "groups_updated": groupsUpdated})
}