mirror of
https://github.com/EZ-Api/ez-api.git
synced 2026-01-13 17:47:51 +00:00
- Introduce `SyncOutboxService` and model to retry failed CP-to-Redis sync operations - Update `SyncService` to handle sync failures by enqueuing tasks to the outbox - Centralize provider group and API key validation logic into `ProviderGroupManager` - Refactor API handlers to utilize the new manager and robust sync methods - Add configuration options for sync outbox (interval, batch size, retries)
319 lines
9.2 KiB
Go
319 lines
9.2 KiB
Go
package api
|
|
|
|
import (
|
|
"context"
|
|
"net/http"
|
|
"strconv"
|
|
"strings"
|
|
|
|
"github.com/ez-api/ez-api/internal/dto"
|
|
"github.com/ez-api/ez-api/internal/model"
|
|
"github.com/ez-api/ez-api/internal/service"
|
|
"github.com/gin-gonic/gin"
|
|
"github.com/redis/go-redis/v9"
|
|
"gorm.io/gorm"
|
|
)
|
|
|
|
type Handler struct {
|
|
db *gorm.DB
|
|
logDB *gorm.DB
|
|
sync *service.SyncService
|
|
logger *service.LogWriter
|
|
rdb *redis.Client
|
|
logWebhook *service.LogWebhookService
|
|
logPartitioner *service.LogPartitioner
|
|
groupManager *service.ProviderGroupManager
|
|
}
|
|
|
|
func NewHandler(db *gorm.DB, logDB *gorm.DB, sync *service.SyncService, logger *service.LogWriter, rdb *redis.Client, partitioner *service.LogPartitioner) *Handler {
|
|
if logDB == nil {
|
|
logDB = db
|
|
}
|
|
return &Handler{
|
|
db: db,
|
|
logDB: logDB,
|
|
sync: sync,
|
|
logger: logger,
|
|
rdb: rdb,
|
|
logWebhook: service.NewLogWebhookService(rdb),
|
|
logPartitioner: partitioner,
|
|
groupManager: service.NewProviderGroupManager(),
|
|
}
|
|
}
|
|
|
|
func (h *Handler) logDBConn() *gorm.DB {
|
|
if h == nil || h.logDB == nil {
|
|
return h.db
|
|
}
|
|
return h.logDB
|
|
}
|
|
|
|
func (h *Handler) logBaseQuery() *gorm.DB {
|
|
return logBaseQuery(h.logDBConn(), h.logPartitioner)
|
|
}
|
|
|
|
// CreateKey is now handled by MasterHandler
|
|
|
|
// CreateModel godoc
|
|
// @Summary Register a new model
|
|
// @Description Register a supported model with its capabilities
|
|
// @Tags admin
|
|
// @Accept json
|
|
// @Produce json
|
|
// @Security AdminAuth
|
|
// @Param model body dto.ModelDTO true "Model Info"
|
|
// @Success 201 {object} model.Model
|
|
// @Failure 400 {object} gin.H
|
|
// @Failure 500 {object} gin.H
|
|
// @Router /admin/models [post]
|
|
func (h *Handler) CreateModel(c *gin.Context) {
|
|
var req dto.ModelDTO
|
|
if err := c.ShouldBindJSON(&req); err != nil {
|
|
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
|
|
return
|
|
}
|
|
|
|
name := strings.TrimSpace(req.Name)
|
|
if name == "" {
|
|
c.JSON(http.StatusBadRequest, gin.H{"error": "name required"})
|
|
return
|
|
}
|
|
kind := strings.ToLower(strings.TrimSpace(req.Kind))
|
|
if kind == "" {
|
|
kind = "chat"
|
|
}
|
|
switch kind {
|
|
case "chat", "embedding", "rerank", "other":
|
|
default:
|
|
c.JSON(http.StatusBadRequest, gin.H{"error": "invalid kind"})
|
|
return
|
|
}
|
|
|
|
modelReq := model.Model{
|
|
Name: name,
|
|
Kind: kind,
|
|
ContextWindow: req.ContextWindow,
|
|
CostPerToken: req.CostPerToken,
|
|
SupportsVision: req.SupportsVision,
|
|
SupportsFunctions: req.SupportsFunctions,
|
|
SupportsToolChoice: req.SupportsToolChoice,
|
|
SupportsFIM: req.SupportsFIM,
|
|
MaxOutputTokens: req.MaxOutputTokens,
|
|
}
|
|
|
|
if err := h.db.Create(&modelReq).Error; err != nil {
|
|
c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to create model", "details": err.Error()})
|
|
return
|
|
}
|
|
|
|
if err := h.sync.SyncModel(&modelReq); err != nil {
|
|
c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to sync model", "details": err.Error()})
|
|
return
|
|
}
|
|
|
|
c.JSON(http.StatusCreated, modelReq)
|
|
}
|
|
|
|
// ListModels godoc
|
|
// @Summary List all models
|
|
// @Description Get a list of all registered models
|
|
// @Tags admin
|
|
// @Produce json
|
|
// @Security AdminAuth
|
|
// @Param page query int false "page (1-based)"
|
|
// @Param limit query int false "limit (default 50, max 200)"
|
|
// @Param search query string false "search by name/kind"
|
|
// @Success 200 {array} model.Model
|
|
// @Failure 500 {object} gin.H
|
|
// @Router /admin/models [get]
|
|
func (h *Handler) ListModels(c *gin.Context) {
|
|
var models []model.Model
|
|
q := h.db.Model(&model.Model{}).Order("id desc")
|
|
query := parseListQuery(c)
|
|
q = applyListSearch(q, query.Search, "name", "kind")
|
|
q = applyListPagination(q, query)
|
|
if err := q.Find(&models).Error; err != nil {
|
|
c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to list models", "details": err.Error()})
|
|
return
|
|
}
|
|
c.JSON(http.StatusOK, models)
|
|
}
|
|
|
|
// UpdateModel godoc
|
|
// @Summary Update a model
|
|
// @Description Update an existing model's configuration
|
|
// @Tags admin
|
|
// @Accept json
|
|
// @Produce json
|
|
// @Security AdminAuth
|
|
// @Param id path int true "Model ID"
|
|
// @Param model body dto.ModelDTO true "Model Info"
|
|
// @Success 200 {object} model.Model
|
|
// @Failure 400 {object} gin.H
|
|
// @Failure 404 {object} gin.H
|
|
// @Failure 500 {object} gin.H
|
|
// @Router /admin/models/{id} [put]
|
|
func (h *Handler) UpdateModel(c *gin.Context) {
|
|
idParam := c.Param("id")
|
|
id, err := strconv.Atoi(idParam)
|
|
if err != nil {
|
|
c.JSON(http.StatusBadRequest, gin.H{"error": "invalid id"})
|
|
return
|
|
}
|
|
|
|
var req dto.ModelDTO
|
|
if err := c.ShouldBindJSON(&req); err != nil {
|
|
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
|
|
return
|
|
}
|
|
|
|
var existing model.Model
|
|
if err := h.db.First(&existing, id).Error; err != nil {
|
|
c.JSON(http.StatusNotFound, gin.H{"error": "model not found"})
|
|
return
|
|
}
|
|
|
|
name := strings.TrimSpace(req.Name)
|
|
if name == "" {
|
|
c.JSON(http.StatusBadRequest, gin.H{"error": "name required"})
|
|
return
|
|
}
|
|
kind := strings.ToLower(strings.TrimSpace(req.Kind))
|
|
if kind == "" {
|
|
kind = strings.ToLower(strings.TrimSpace(existing.Kind))
|
|
}
|
|
if kind == "" {
|
|
kind = "chat"
|
|
}
|
|
switch kind {
|
|
case "chat", "embedding", "rerank", "other":
|
|
default:
|
|
c.JSON(http.StatusBadRequest, gin.H{"error": "invalid kind"})
|
|
return
|
|
}
|
|
|
|
existing.Name = name
|
|
existing.Kind = kind
|
|
existing.ContextWindow = req.ContextWindow
|
|
existing.CostPerToken = req.CostPerToken
|
|
existing.SupportsVision = req.SupportsVision
|
|
existing.SupportsFunctions = req.SupportsFunctions
|
|
existing.SupportsToolChoice = req.SupportsToolChoice
|
|
existing.SupportsFIM = req.SupportsFIM
|
|
existing.MaxOutputTokens = req.MaxOutputTokens
|
|
|
|
if err := h.db.Save(&existing).Error; err != nil {
|
|
c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to update model", "details": err.Error()})
|
|
return
|
|
}
|
|
|
|
if err := h.sync.SyncModel(&existing); err != nil {
|
|
c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to sync model", "details": err.Error()})
|
|
return
|
|
}
|
|
|
|
c.JSON(http.StatusOK, existing)
|
|
}
|
|
|
|
// DeleteModel godoc
|
|
// @Summary Delete a model
|
|
// @Description Delete a model by id
|
|
// @Tags admin
|
|
// @Produce json
|
|
// @Security AdminAuth
|
|
// @Param id path int true "Model ID"
|
|
// @Success 200 {object} gin.H
|
|
// @Failure 400 {object} gin.H
|
|
// @Failure 404 {object} gin.H
|
|
// @Failure 500 {object} gin.H
|
|
// @Router /admin/models/{id} [delete]
|
|
func (h *Handler) DeleteModel(c *gin.Context) {
|
|
idParam := c.Param("id")
|
|
id, err := strconv.Atoi(idParam)
|
|
if err != nil || id <= 0 {
|
|
c.JSON(http.StatusBadRequest, gin.H{"error": "invalid id"})
|
|
return
|
|
}
|
|
|
|
var existing model.Model
|
|
if err := h.db.First(&existing, id).Error; err != nil {
|
|
c.JSON(http.StatusNotFound, gin.H{"error": "model not found"})
|
|
return
|
|
}
|
|
|
|
if err := h.db.Delete(&existing).Error; err != nil {
|
|
c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to delete model", "details": err.Error()})
|
|
return
|
|
}
|
|
|
|
if err := h.sync.SyncModelDelete(&existing); err != nil {
|
|
c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to sync model delete", "details": err.Error()})
|
|
return
|
|
}
|
|
|
|
c.JSON(http.StatusOK, gin.H{"status": "deleted"})
|
|
}
|
|
|
|
// SyncSnapshot godoc
|
|
// @Summary Force sync snapshot
|
|
// @Description Force full synchronization of DB state to Redis
|
|
// @Tags admin
|
|
// @Produce json
|
|
// @Security AdminAuth
|
|
// @Success 200 {object} gin.H
|
|
// @Failure 500 {object} gin.H
|
|
// @Router /admin/sync/snapshot [post]
|
|
func (h *Handler) SyncSnapshot(c *gin.Context) {
|
|
if err := h.sync.SyncAll(h.db); err != nil {
|
|
c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to sync snapshots", "details": err.Error()})
|
|
return
|
|
}
|
|
c.JSON(http.StatusOK, gin.H{"status": "synced"})
|
|
}
|
|
|
|
// IngestLog accepts log records from data plane or other services.
|
|
// @Summary Ingest logs
|
|
// @Description Internal endpoint for ingesting logs from Balancer
|
|
// @Tags system
|
|
// @Accept json
|
|
// @Produce json
|
|
// @Param log body model.LogRecord true "Log Record"
|
|
// @Success 202 {object} gin.H
|
|
// @Failure 400 {object} gin.H
|
|
// @Router /logs [post]
|
|
func (h *Handler) IngestLog(c *gin.Context) {
|
|
var rec model.LogRecord
|
|
if err := c.ShouldBindJSON(&rec); err != nil {
|
|
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
|
|
return
|
|
}
|
|
|
|
if !h.logRequestBodyEnabled(c.Request.Context()) {
|
|
rec.RequestBody = ""
|
|
}
|
|
|
|
// By default, only metadata is expected; payload fields may be empty.
|
|
if h.logger != nil {
|
|
h.logger.Write(rec)
|
|
}
|
|
if h.logWebhook != nil {
|
|
recCopy := rec
|
|
go h.logWebhook.NotifyIfNeeded(context.Background(), recCopy)
|
|
}
|
|
c.JSON(http.StatusAccepted, gin.H{"status": "queued"})
|
|
}
|
|
|
|
func (h *Handler) logRequestBodyEnabled(ctx context.Context) bool {
|
|
if h == nil || h.rdb == nil {
|
|
return true
|
|
}
|
|
raw, err := h.rdb.HGet(ctx, featuresKey, logRequestBodyFeatureKey).Result()
|
|
if err == redis.Nil || strings.TrimSpace(raw) == "" {
|
|
return true
|
|
}
|
|
if err != nil {
|
|
return true
|
|
}
|
|
return strings.EqualFold(raw, "true") || strings.EqualFold(raw, "1")
|
|
}
|