746 lines
31 KiB
Go
746 lines
31 KiB
Go
package cmd
|
||
|
||
import (
|
||
"context"
|
||
"crypto/sha256"
|
||
"encoding/hex"
|
||
"fmt"
|
||
"io/fs"
|
||
"net/http"
|
||
"net/http/pprof"
|
||
"strings"
|
||
"time"
|
||
|
||
"github.com/gogf/gf/v2/frame/g"
|
||
"github.com/gogf/gf/v2/net/ghttp"
|
||
"github.com/gogf/gf/v2/os/gcmd"
|
||
"github.com/gogf/gf/v2/os/gctx"
|
||
|
||
"github.com/qianfree/team-api/internal/consts"
|
||
adminController "github.com/qianfree/team-api/internal/controller/admin"
|
||
captchaController "github.com/qianfree/team-api/internal/controller/captcha"
|
||
docsController "github.com/qianfree/team-api/internal/controller/docs"
|
||
openController "github.com/qianfree/team-api/internal/controller/open"
|
||
settingsController "github.com/qianfree/team-api/internal/controller/settings"
|
||
tenantController "github.com/qianfree/team-api/internal/controller/tenant"
|
||
"github.com/qianfree/team-api/internal/dispatchadapter"
|
||
"github.com/qianfree/team-api/internal/logic/admin"
|
||
"github.com/qianfree/team-api/internal/logic/billing"
|
||
"github.com/qianfree/team-api/internal/logic/common"
|
||
"github.com/qianfree/team-api/internal/logic/monitor"
|
||
relayLogic "github.com/qianfree/team-api/internal/logic/relay"
|
||
"github.com/qianfree/team-api/internal/logic/task"
|
||
"github.com/qianfree/team-api/internal/logic/tenant"
|
||
"github.com/qianfree/team-api/internal/logic/update"
|
||
"github.com/qianfree/team-api/internal/middleware"
|
||
"github.com/qianfree/team-api/internal/response"
|
||
"github.com/qianfree/team-api/internal/utility/crypto"
|
||
relaycommon "github.com/qianfree/team-api/relay/common"
|
||
|
||
adminHandler "github.com/qianfree/team-api/internal/handler/admin"
|
||
"github.com/qianfree/team-api/internal/handler/landing"
|
||
"github.com/qianfree/team-api/internal/handler/public"
|
||
"github.com/qianfree/team-api/internal/handler/relay"
|
||
setupHandler "github.com/qianfree/team-api/internal/handler/setup"
|
||
"github.com/qianfree/team-api/internal/plugin"
|
||
"github.com/qianfree/team-api/web"
|
||
)
|
||
|
||
var (
|
||
Main = gcmd.Command{
|
||
Name: "main",
|
||
Usage: "main",
|
||
Brief: "start http server of team-api",
|
||
Func: func(ctx context.Context, parser *gcmd.Parser) (err error) {
|
||
s := g.Server()
|
||
|
||
printBanner()
|
||
|
||
// Auto-migrate database schema
|
||
if err := runAutoMigrate(ctx); err != nil {
|
||
return err
|
||
}
|
||
|
||
// Health check: Redis
|
||
if err := checkRedis(ctx); err != nil {
|
||
g.Log().Fatalf(ctx, "Redis 连接检查失败,程序退出: %v", err)
|
||
}
|
||
g.Log().Info(ctx, "Redis 连接正常")
|
||
|
||
// Initialize JWT secret
|
||
common.InitJWTSecret(ctx)
|
||
|
||
// 启动阶段校验加密密钥,避免运行时请求路径 panic(#3)
|
||
if hexKey := g.Cfg().MustGet(ctx, "crypto.encryptionKey").String(); hexKey == "" {
|
||
g.Log().Fatal(ctx, "crypto.encryptionKey 未配置,程序退出(运行 openssl rand -hex 32 生成)")
|
||
} else if _, keyErr := crypto.GetEncryptionKey(hexKey); keyErr != nil {
|
||
g.Log().Fatalf(ctx, "crypto.encryptionKey 配置无效,程序退出: %v", keyErr)
|
||
}
|
||
g.Log().Info(ctx, "加密密钥校验通过")
|
||
|
||
// System initialization: env auto-init or detect setup mode
|
||
autoInit, initErr := admin.AutoInitAdmin(ctx)
|
||
if initErr != nil {
|
||
g.Log().Errorf(ctx, "auto init admin: %v", initErr)
|
||
}
|
||
if !autoInit {
|
||
exists, checkErr := admin.AdminExists(ctx)
|
||
if checkErr != nil {
|
||
g.Log().Errorf(ctx, "check admin existence: %v", checkErr)
|
||
} else if !exists {
|
||
setupHandler.SetSetupMode(true)
|
||
g.Log().Info(ctx, "系统未初始化,进入设置模式 — 请访问 /setup 完成初始化")
|
||
}
|
||
}
|
||
|
||
// Initialize config service
|
||
common.Config().Warmup(ctx)
|
||
common.Config().StartSubscriber(ctx)
|
||
|
||
// C1: 通用 Cache 的跨实例 L1 失效订阅(cache:invalidate)
|
||
common.StartCacheInvalidationSubscriber(ctx)
|
||
|
||
// Initialize content filter engine
|
||
common.InitContentFilter(ctx)
|
||
|
||
// Initialize monitoring collector
|
||
monitor.InitCollector(ctx)
|
||
monitor.InitRequestTracker()
|
||
monitor.InitRelaykitTracker()
|
||
monitor.InitDispatchTracker()
|
||
dispatchadapter.SetBreakerOpenHook(monitor.TrackDispatchBreakerOpen)
|
||
|
||
// Ensure partitioned tables have current+future partitions
|
||
if partitionErr := common.EnsurePartitions(ctx); partitionErr != nil {
|
||
g.Log().Errorf(ctx, "ensure partitions: %v", partitionErr)
|
||
}
|
||
|
||
// Initialize async usage log writer
|
||
common.InitUsageLogWriter()
|
||
|
||
// Initialize async channel error writer
|
||
common.InitChannelErrorWriter()
|
||
|
||
// Initialize async audit log writer
|
||
common.InitAuditLogWriter(ctx)
|
||
|
||
// Initialize async channel debug log writer + 注入 relay 层提交钩子
|
||
common.InitChannelDebugLogWriter()
|
||
relayLogic.InitDebugLogRecorder()
|
||
|
||
// Initialize async error log writer
|
||
response.InitErrorLogWriter()
|
||
|
||
// 按配置拉起 pprof 调试端口(「性能配置-调试」):仅监听本机回环,禁止公网暴露;
|
||
// 端口在进程启动时读取,变更后需重启生效
|
||
if port := common.Config().GetInt(ctx, "pprof_port"); port > 0 && port <= 65535 {
|
||
go func() {
|
||
mux := http.NewServeMux()
|
||
mux.HandleFunc("/debug/pprof/", pprof.Index)
|
||
mux.HandleFunc("/debug/pprof/cmdline", pprof.Cmdline)
|
||
mux.HandleFunc("/debug/pprof/profile", pprof.Profile)
|
||
mux.HandleFunc("/debug/pprof/symbol", pprof.Symbol)
|
||
mux.HandleFunc("/debug/pprof/trace", pprof.Trace)
|
||
if listenErr := http.ListenAndServe(fmt.Sprintf("127.0.0.1:%d", port), mux); listenErr != nil {
|
||
g.Log().Warningf(ctx, "pprof server exited: %v", listenErr)
|
||
}
|
||
}()
|
||
g.Log().Infof(ctx, "pprof listening on 127.0.0.1:%d", port)
|
||
}
|
||
|
||
// Register cron jobs
|
||
common.InitCronScheduler()
|
||
cs := common.GetCronScheduler()
|
||
registerCronJobs(cs)
|
||
cs.StartBackground(ctx)
|
||
|
||
// relay 层运行时配置注入(全局超时兜底/系统代理;渠道级 settings 仍优先)
|
||
syncGlobalRelaySettings(ctx)
|
||
|
||
// 累计消费基线种子:为存量钱包 hash 补种 total_consumed 历史基线(幂等),
|
||
// 必须在 relay 流量进入前完成,防止结算 HINCRBY 从 0 起算丢历史
|
||
billing.SeedWalletTotalConsumed(ctx)
|
||
|
||
// 启动钱包物化器(boot goroutine,秒级刷新 Redis 权威钱包状态到 DB;不走 cron)
|
||
billing.StartWalletMaterializer(ctx)
|
||
|
||
// Initialize update manager
|
||
update.InitManager(ctx)
|
||
update.CheckPendingVerification(ctx)
|
||
|
||
// Initialize plugin system (must be after CronScheduler init)
|
||
pluginApp := &plugin.App{
|
||
Server: s,
|
||
DB: g.DB(),
|
||
Redis: g.Redis(),
|
||
Hook: plugin.GlobalEmitter(),
|
||
}
|
||
if pluginErr := plugin.Bootstrap(ctx, pluginApp); pluginErr != nil {
|
||
g.Log().Errorf(ctx, "plugin bootstrap: %v", pluginErr)
|
||
}
|
||
|
||
// Global middleware
|
||
s.Use(middleware.Recovery)
|
||
s.Use(middleware.RequestId)
|
||
s.Use(middleware.ServiceName)
|
||
|
||
// Setup mode guard: block all requests until initialization is complete
|
||
s.Use(func(r *ghttp.Request) {
|
||
if !setupHandler.IsSetupMode() {
|
||
r.Middleware.Next()
|
||
return
|
||
}
|
||
// Double-check DB in case setup was completed by another request/instance
|
||
exists, _ := admin.AdminExists(r.Context())
|
||
if exists {
|
||
setupHandler.SetSetupComplete()
|
||
r.Middleware.Next()
|
||
return
|
||
}
|
||
// Allow setup and health endpoints
|
||
if r.URL.Path == "/setup" || r.URL.Path == "/api/setup/initialize" || r.URL.Path == "/api/setup/status" || r.URL.Path == "/api/health" {
|
||
r.Middleware.Next()
|
||
return
|
||
}
|
||
// Block everything else
|
||
accept := r.Header.Get("Accept")
|
||
if len(accept) >= 4 && accept[:4] == "text" {
|
||
r.Response.RedirectTo("/setup")
|
||
} else {
|
||
r.Response.WriteJson(g.Map{
|
||
"code": consts.CodeSetupNotInitialized,
|
||
"message": consts.MsgSetupNotInitialized,
|
||
"data": nil,
|
||
"request_id": r.GetCtxVar("RequestId"),
|
||
})
|
||
}
|
||
r.Exit()
|
||
})
|
||
|
||
// Setup routes (always registered, guard decides if accessible)
|
||
s.Group("/", func(group *ghttp.RouterGroup) {
|
||
group.GET("/setup", setupHandler.HandleSetupPage)
|
||
})
|
||
s.Group("/api/setup", func(group *ghttp.RouterGroup) {
|
||
group.Middleware(middleware.ErrorHandler)
|
||
group.GET("/status", setupHandler.HandleSetupStatus)
|
||
group.POST("/initialize", setupHandler.HandleSetupInitialize)
|
||
})
|
||
|
||
// Health check endpoint (always accessible, used by Docker/K8s probes)
|
||
s.Group("/api", func(group *ghttp.RouterGroup) {
|
||
group.GET("/health", func(r *ghttp.Request) {
|
||
r.Response.WriteJson(g.Map{
|
||
"status": "ok",
|
||
"version": consts.Version,
|
||
})
|
||
})
|
||
})
|
||
|
||
// Register route groups
|
||
s.Group("/api", func(group *ghttp.RouterGroup) {
|
||
group.Middleware(middleware.MiddlewareHandlerResponse)
|
||
|
||
// Admin — public endpoints use g.Meta middleware:"-" to skip auth
|
||
group.Group("/admin", func(g *ghttp.RouterGroup) {
|
||
g.Middleware(middleware.DemoMode, middleware.AdminAuth, middleware.AdminPermissionGuard, middleware.OperationLog)
|
||
g.Bind(adminController.NewV1())
|
||
})
|
||
|
||
// Tenant — public endpoints use g.Meta middleware:"-" to skip auth
|
||
group.Group("/tenant", func(g *ghttp.RouterGroup) {
|
||
// OperationLog 与管理后台共用:中间件按 ctx 的 userType/tenantId 落
|
||
// aud_operation_logs,租户写操作(成员/Key/工单等)此前完全无留痕,
|
||
// 且租户端"操作日志"页因 tenant_id 恒为 0 永远为空,挂上后一并修复。
|
||
g.Middleware(middleware.DemoMode, middleware.MaintenanceMode, middleware.TenantAuth, middleware.Idempotency, middleware.OperationLog)
|
||
g.Bind(tenantController.NewV1())
|
||
})
|
||
|
||
// Payment callbacks — manual registration (raw string response, not JSON)
|
||
group.Middleware(middleware.ErrorHandler)
|
||
registerPaymentCallbacks(group)
|
||
|
||
// Captcha — public, no auth required (shared by admin + tenant)
|
||
group.Group("/captcha", func(g *ghttp.RouterGroup) {
|
||
g.Bind(captchaController.NewV1())
|
||
})
|
||
// Settings — public settings, no auth required
|
||
group.Group("/settings", func(g *ghttp.RouterGroup) {
|
||
g.Bind(settingsController.NewV1())
|
||
})
|
||
|
||
// Docs — public OpenAPI spec
|
||
group.Group("/docs", func(g *ghttp.RouterGroup) {
|
||
g.Bind(docsController.NewV1())
|
||
})
|
||
})
|
||
|
||
// File streaming serve — admin auth required, bypasses JSON response wrapper.
|
||
// 流式返回二进制(local 存储模式 + 未来应用层代理下载),不能走统一 JSON 响应格式,
|
||
// 故独立注册路由组(不套 MiddlewareHandlerResponse),只套鉴权与审计中间件。
|
||
s.Group("/api/admin/files", func(group *ghttp.RouterGroup) {
|
||
group.Middleware(middleware.AdminAuth, middleware.AdminPermissionGuard, middleware.OperationLog)
|
||
group.GET("/{id}/serve", adminHandler.HandleFileServe)
|
||
})
|
||
|
||
// Open Platform API — HMAC-SHA256 authentication
|
||
// 中间件与 /api/tenant 对齐:演示模式同样要拦住写操作(否则演示部署可经开放平台
|
||
// 真实改数据),维护模式头同样要下发,幂等同样要生效 —— 开放平台是对外的、
|
||
// 天然会重试的接口,没有幂等语义比控制台更危险。
|
||
// Idempotency 必须排在 OpenPlatformAuth 之后:它按认证主体隔离幂等键。
|
||
s.Group("/api/open", func(group *ghttp.RouterGroup) {
|
||
group.Middleware(middleware.MiddlewareHandlerResponse)
|
||
// OperationLog 以应用身份(open_app + 应用ID)记录写操作,与管理/租户端共用
|
||
group.Middleware(middleware.DemoMode, middleware.MaintenanceMode, middleware.OpenPlatformAuth, middleware.Idempotency, middleware.OperationLog)
|
||
group.Bind(openController.NewV1())
|
||
})
|
||
|
||
// AI proxy endpoints (OpenAI compatible, /v1/xxx)
|
||
registerRelayRoutes(s)
|
||
|
||
// Register plugin routes
|
||
plugin.RegisterAllRoutes(ctx, s)
|
||
|
||
// Embedded frontend SPA serving
|
||
registerFrontendRoutes(s)
|
||
|
||
// Landing page for backend root when frontend is not embedded
|
||
registerLandingPage(s)
|
||
|
||
// Migrate encryption key (legacy default → configured key)
|
||
relayLogic.MigrateEncryptionKey(ctx)
|
||
|
||
// Initialize active task count and start polling
|
||
task.InitActiveCount(ctx)
|
||
task.StartAsyncPolling(ctx)
|
||
|
||
// Start sync-image async worker pool (wraps synchronous image providers as async tasks)
|
||
task.StartSyncImageWorkers(ctx)
|
||
|
||
// Start webhook dispatcher (event-driven delivery)
|
||
tenant.InitWebhookDispatcher(ctx)
|
||
|
||
// 注入 webhook 发布函数到 billing 包(解耦循环依赖)
|
||
billing.SetPublishWebhookEventFn(tenant.PublishWebhookEvent)
|
||
|
||
// Graceful shutdown order (defers run LIFO after s.Run returns):
|
||
// 1) 先排空任务池(StopSyncImageWorkers / StopAsyncPolling)——它们收尾时会写
|
||
// 用量日志 / 审计 / 结算;此时异步 Writer 必须仍存活。
|
||
// 2) 再 flush 关闭异步 Writer。
|
||
// 3) 最后关 webhook / 调度后台任务 / plugin。
|
||
// 因此把两个 task.Stop* 注册在最后(最先执行),Writer 关闭注册在前(后执行)。
|
||
// 0) MaybeExecRestart 注册在最前、最后执行:在线更新/回滚换壳后,用
|
||
// syscall.Exec 原地换壳为新版本(同 PID 继续运行,不依赖外部进程管理器);
|
||
// 正常退出时是空操作。必须在所有任务池/Writer 排空之后执行。
|
||
defer update.MaybeExecRestart(ctx)
|
||
defer plugin.Shutdown(ctx)
|
||
defer dispatchadapter.Shutdown()
|
||
defer tenant.ShutdownWebhookDispatcher()
|
||
defer common.CloseAuditLogWriter()
|
||
defer common.CloseChannelErrorWriter()
|
||
defer common.CloseUsageLogWriter()
|
||
defer response.CloseErrorLogWriter()
|
||
defer task.StopAsyncPolling()
|
||
defer task.StopSyncImageWorkers()
|
||
|
||
s.Run()
|
||
return nil
|
||
},
|
||
}
|
||
)
|
||
|
||
// printBanner prints the startup banner with copyright information.
|
||
func printBanner() {
|
||
cyan := "\x1b[36;1m"
|
||
green := "\x1b[32m"
|
||
dim := "\x1b[2m"
|
||
reset := "\x1b[0m"
|
||
|
||
fmt.Println()
|
||
fmt.Printf(" %s████████╗███████╗ █████╗ ███╗ ███╗ █████╗ ██████╗ ██╗%s\n", cyan, reset)
|
||
fmt.Printf(" %s╚══██╔══╝██╔════╝██╔══██╗████╗ ████║ ██╔══██╗██╔══██╗██║%s\n", cyan, reset)
|
||
fmt.Printf(" %s ██║ █████╗ ███████║██╔████╔██║█████╗███████║██████╔╝██║%s\n", cyan, reset)
|
||
fmt.Printf(" %s ██║ ██╔══╝ ██╔══██║██║╚██╔╝██║╚════╝██╔══██║██╔═══╝ ██║%s\n", cyan, reset)
|
||
fmt.Printf(" %s ██║ ███████╗██║ ██║██║ ╚═╝ ██║ ██║ ██║██║ ██║%s\n", cyan, reset)
|
||
fmt.Printf(" %s ╚═╝ ╚══════╝╚═╝ ╚═╝╚═╝ ╚═╝ ╚═╝ ╚═╝╚═╝ ╚═╝%s\n", cyan, reset)
|
||
fmt.Printf(" %sTeam-API%s %s%s%s %s| %s%s企业级大模型 API 网关系统%s\n", cyan, reset, green, consts.Version, reset, dim, reset, dim, reset)
|
||
fmt.Printf(" %shttps://github.com/qianfree/team-api%s\n", dim, reset)
|
||
fmt.Println()
|
||
fmt.Printf(" %sAGPL-3.0-or-later | Copyright © 2026 qianfree%s\n", dim, reset)
|
||
fmt.Println()
|
||
}
|
||
|
||
// channelAutoTestSchedule 按配置的探测间隔生成 cron 表达式。
|
||
//
|
||
// 用 @every 而非 */N:*/N 只在 N < 60 时有意义(*/90 在分钟位上等价于 */30,
|
||
// 静默给出与配置不符的频率),而 @every 支持任意时长。副作用是触发时刻相对进程
|
||
// 启动时间而非整点对齐——多实例部署时反而更好,不会所有实例卡在同一分钟同时探测。
|
||
func channelAutoTestSchedule(ctx context.Context) string {
|
||
minutes := common.Config().GetInt(ctx, "channel_auto_test_interval_minutes")
|
||
// 配置缺失(GetInt 返回 0)或越界时回落默认值,避免非法表达式让任务注册失败
|
||
if minutes < 1 || minutes > 1440 {
|
||
minutes = 5
|
||
}
|
||
return fmt.Sprintf("@every %dm", minutes)
|
||
}
|
||
|
||
// watchChannelAutoTestInterval 探测间隔配置变更时重排 cron,无需重启。
|
||
// 回调经 Redis pub/sub 触发,多实例会各自重排自己的调度器。
|
||
func watchChannelAutoTestInterval(cs *common.CronScheduler) {
|
||
common.OnSettingsChanged(func(ctx context.Context, key string) {
|
||
if key != "channel_auto_test_interval_minutes" {
|
||
return
|
||
}
|
||
if err := cs.Reschedule(ctx, "channel_auto_test", channelAutoTestSchedule(ctx)); err != nil {
|
||
g.Log().Errorf(ctx, "渠道自动探测重排失败,沿用原间隔: %v", err)
|
||
}
|
||
})
|
||
}
|
||
|
||
// syncGlobalRelaySettings 将系统设置中 relay 层的运行时配置注入 relay 层,并在配置变更时动态刷新(免重启):
|
||
// - 全局超时:优先级 渠道级 settings.timeout_seconds > 此处注入的全局值 > relay 层内置默认(180s/300s)
|
||
// - 系统代理:channel_proxy_url,渠道编辑中开启「使用代理」的请求经此代理转发
|
||
//
|
||
// relay 包不反向 import internal(会循环依赖),故由本函数桥接注入。
|
||
func syncGlobalRelaySettings(ctx context.Context) {
|
||
apply := func(ctx context.Context) {
|
||
relaycommon.SetGlobalRequestTimeoutSeconds(common.Config().GetInt(ctx, "request_timeout_seconds"))
|
||
relaycommon.SetGlobalStreamIdleTimeoutSeconds(common.Config().GetInt(ctx, "streaming_timeout_seconds"))
|
||
relaycommon.SetSystemProxyURL(common.Config().GetString(ctx, "channel_proxy_url"))
|
||
}
|
||
apply(ctx)
|
||
common.OnSettingsChanged(func(ctx context.Context, key string) {
|
||
if key != "request_timeout_seconds" && key != "streaming_timeout_seconds" && key != "channel_proxy_url" {
|
||
return
|
||
}
|
||
apply(ctx)
|
||
// 代理 URL 可能含认证凭据,只记录是否已配置,不落日志
|
||
g.Log().Infof(ctx, "全局 relay 配置已刷新: request=%ds stream_idle=%ds proxy_configured=%v",
|
||
relaycommon.GlobalRequestTimeoutSeconds(), relaycommon.GlobalStreamIdleTimeoutSeconds(),
|
||
relaycommon.GetSystemProxyURL() != "")
|
||
})
|
||
}
|
||
|
||
// registerCronJobs 集中注册所有定时任务,避免散落在主启动流程中。
|
||
// 新增 cron 任务时在此追加一项即可;任务名(用于分布式锁 key)需保持唯一且稳定。
|
||
func registerCronJobs(cs *common.CronScheduler) {
|
||
watchChannelAutoTestInterval(cs)
|
||
cs.Register("ops_system_collector", "系统指标采集", "* * * * *", func(ctx context.Context) error {
|
||
return monitor.CollectSystemMetrics(ctx)
|
||
})
|
||
cs.Register("ops_alert_detector", "告警检测", "* * * * *", func(ctx context.Context) error {
|
||
return monitor.RunAlertDetection(ctx)
|
||
})
|
||
cs.Register("ops_metrics_cleanup", "指标数据清理", "0 3 * * *", func(ctx context.Context) error {
|
||
return monitor.CleanupOldMetrics(ctx)
|
||
})
|
||
cs.Register("partition_ensure", "分区维护", "0 2 * * *", func(ctx context.Context) error {
|
||
return common.EnsurePartitions(ctx)
|
||
})
|
||
cs.Register("health_snapshot", "渠道健康快照", "*/5 * * * *", func(ctx context.Context) error {
|
||
return task.SnapshotHealthScores(ctx)
|
||
})
|
||
cs.Register("channel_auto_test", "渠道自动测试", channelAutoTestSchedule(gctx.New()), func(ctx context.Context) error {
|
||
if common.Config().GetBool(ctx, "channel_auto_test_enabled") {
|
||
task.AutoTestChannels(ctx)
|
||
}
|
||
return nil
|
||
})
|
||
cs.Register("model_sunset_check", "模型下线检查", "0 0 * * *", func(ctx context.Context) error {
|
||
return task.CheckModelSunset(ctx)
|
||
})
|
||
cs.Register("data_cleanup", "过期数据清理", "0 3 * * *", func(ctx context.Context) error {
|
||
return admin.CleanupExpiredData(ctx)
|
||
})
|
||
cs.Register("export_file_cleanup", "导出文件清理", "0 4 * * *", func(ctx context.Context) error {
|
||
return admin.CleanupExpiredExportFiles(ctx)
|
||
})
|
||
cs.Register("file_retention_check", "文件保留检查", "0 5 * * *", func(ctx context.Context) error {
|
||
return admin.CheckFileRetention(ctx)
|
||
})
|
||
cs.Register("task_timeout_check", "任务超时检查", "*/10 * * * *", func(ctx context.Context) error {
|
||
return admin.MarkStuckTasksFailed(ctx)
|
||
})
|
||
cs.Register("task_executor", "异步任务执行", "*/1 * * * *", func(ctx context.Context) error {
|
||
task.RunPendingTasks(ctx)
|
||
return nil
|
||
})
|
||
cs.Register("project_budget_check", "项目预算检查", "*/5 * * * *", func(ctx context.Context) error {
|
||
return tenant.CheckBudgetExhausted(ctx)
|
||
})
|
||
cs.Register("usage_log_cleanup", "用量日志清理", "0 3 * * *", func(ctx context.Context) error {
|
||
retentionDays := common.Config().GetInt(ctx, "usage_log_retention_days")
|
||
if retentionDays == 0 {
|
||
retentionDays = 90
|
||
}
|
||
return task.ScheduleAutoCleanup(ctx, retentionDays)
|
||
})
|
||
cs.Register("usage_daily_aggregate", "用量日报表聚合", "0 1 * * *", func(ctx context.Context) error {
|
||
// 用量日维度聚合:将 bil_usage_logs 聚合进 bil_usage_daily,供流量桑基图与趋势分析。
|
||
// 自愈:每次重算最近 3 个完整天,覆盖短暂宕机;ON CONFLICT 保证幂等。
|
||
// end 取今天 00:00(开区间),永不聚合当天进行中的数据。
|
||
end := time.Now().Format("2006-01-02")
|
||
start := time.Now().AddDate(0, 0, -3).Format("2006-01-02")
|
||
return task.AggregateUsageRange(ctx, start, end)
|
||
})
|
||
cs.Register("oauth_token_refresh", "OAuth 令牌刷新", "*/10 * * * *", func(ctx context.Context) error {
|
||
return task.RefreshExpiringOAuthTokens(ctx)
|
||
})
|
||
cs.Register("prededuct_sweep", "预扣清扫", "*/2 * * * *", func(ctx context.Context) error {
|
||
billing.PredeductSweep(ctx)
|
||
return nil
|
||
})
|
||
cs.Register("billing_daily_reconciliation", "计费日对账", "20 5 * * *", func(ctx context.Context) error {
|
||
// 日对账:聚合对账(bil_records vs bil_transactions)+ 交叉对账(usage_logs 反连接
|
||
// 发现漏结算免单请求)+ 冻结余额一致性校验,结果经日志告警
|
||
_, err := billing.RunDailyReconciliation(ctx)
|
||
return err
|
||
})
|
||
cs.Register("update_check", "系统更新检查", "0 */6 * * *", func(ctx context.Context) error {
|
||
if common.Config().GetBool(ctx, "update_auto_check_enabled") {
|
||
return update.BackgroundCheck(ctx)
|
||
}
|
||
return nil
|
||
})
|
||
cs.Register("order_expiration", "订单过期处理", "*/5 * * * *", func(ctx context.Context) error {
|
||
return task.ExpirePendingOrders(ctx)
|
||
})
|
||
}
|
||
|
||
// registerPaymentCallbacks registers payment callback routes.
|
||
func registerPaymentCallbacks(group *ghttp.RouterGroup) {
|
||
group.Group("/payment", func(g *ghttp.RouterGroup) {
|
||
g.POST("/callback/{channel}", public.HandlePaymentCallback)
|
||
g.GET("/callback/{channel}", public.HandlePaymentCallback)
|
||
g.GET("/epay/return", public.HandlePaymentEpayReturn)
|
||
g.POST("/epay/return", public.HandlePaymentEpayReturn)
|
||
})
|
||
}
|
||
|
||
// registerRelayRoutes registers AI proxy routes (/v1/xxx).
|
||
func registerRelayRoutes(server *ghttp.Server) {
|
||
server.Group("/v1", func(group *ghttp.RouterGroup) {
|
||
group.Middleware(middleware.ApiMaintenance, middleware.MaintenanceMode, middleware.ApiKeyAuth, middleware.ContentFilter)
|
||
|
||
group.POST("/chat/completions", relay.HandleChatCompletions)
|
||
group.GET("/models", relay.HandleModels)
|
||
group.GET("/models/{model_id}", relay.HandleModelDetail)
|
||
group.POST("/embeddings", relay.HandleEmbeddings)
|
||
group.POST("/images/generations", relay.HandleImagesGenerations)
|
||
group.POST("/completions", relay.HandleCompletions)
|
||
group.POST("/responses", relay.HandleResponses)
|
||
group.POST("/responses/compact", relay.HandleResponses)
|
||
// Responses 生命周期端点(Redis 路由还原渠道后透传上游)
|
||
group.GET("/responses/{id}", relay.HandleResponsesRetrieve)
|
||
group.POST("/responses/{id}/cancel", relay.HandleResponsesCancel)
|
||
group.DELETE("/responses/{id}", relay.HandleResponsesDelete)
|
||
group.POST("/messages", relay.HandleMessages)
|
||
group.POST("/messages/count_tokens", relay.HandleCountTokens)
|
||
group.POST("/audio/speech", relay.HandleAudioSpeech)
|
||
group.POST("/audio/transcriptions", relay.HandleAudioTranscription)
|
||
group.POST("/audio/translations", relay.HandleAudioTranslation)
|
||
group.POST("/rerank", relay.HandleRerank)
|
||
group.POST("/moderations", relay.HandleModerations)
|
||
group.POST("/images/edits", relay.HandleImagesEdits)
|
||
group.GET("/realtime", relay.HandleRealtime)
|
||
// OpenAI Videos 协议端点(官方 SDK 兼容,对接第三方视频模型)
|
||
group.POST("/videos", relay.HandleVideoCreate)
|
||
group.GET("/videos/{video_id}", relay.HandleVideoRetrieve)
|
||
group.GET("/videos/{video_id}/content", relay.HandleVideoContent)
|
||
group.DELETE("/videos/{video_id}", relay.HandleVideoDelete)
|
||
})
|
||
|
||
// Gemini 兼容路由(/v1beta/models/{model}:generateContent)
|
||
server.Group("/v1beta", func(group *ghttp.RouterGroup) {
|
||
group.Middleware(middleware.ApiMaintenance, middleware.MaintenanceMode, middleware.ApiKeyAuth, middleware.ContentFilter)
|
||
group.GET("/models", relay.HandleGeminiModels)
|
||
group.GET("/models/{model}", relay.HandleGeminiModelDetail)
|
||
group.POST("/models/{model}", relay.HandleGeminiGenerateContent)
|
||
})
|
||
|
||
// 异步任务端点(视频/音乐生成)
|
||
server.Group("/v1", func(group *ghttp.RouterGroup) {
|
||
group.Middleware(middleware.ApiMaintenance, middleware.MaintenanceMode, middleware.ApiKeyAuth, middleware.ContentFilter)
|
||
group.POST("/video/generations", relay.HandleTaskSubmit)
|
||
group.GET("/video/generations/{task_id}", relay.HandleTaskFetch)
|
||
})
|
||
|
||
// 异步图片生成端点(阿里云 DashScope 等)
|
||
server.Group("/v1", func(group *ghttp.RouterGroup) {
|
||
group.Middleware(middleware.ApiMaintenance, middleware.MaintenanceMode, middleware.ApiKeyAuth, middleware.ContentFilter)
|
||
group.POST("/images/generations/async", relay.HandleAliImageSubmit)
|
||
group.GET("/images/generations/async/{task_id}", relay.HandleTaskFetch)
|
||
})
|
||
|
||
// Suno 端点
|
||
server.Group("/suno", func(group *ghttp.RouterGroup) {
|
||
group.Middleware(middleware.ApiMaintenance, middleware.MaintenanceMode, middleware.ApiKeyAuth, middleware.ContentFilter)
|
||
group.POST("/submit/{action}", relay.HandleTaskSubmit)
|
||
group.POST("/fetch", relay.HandleSunoFetchBatch)
|
||
group.GET("/fetch/{task_id}", relay.HandleTaskFetch)
|
||
})
|
||
|
||
// MiniMax 官方视频协议端点(H3 v2,官方 SDK 换 base_url 直连;协议文档 docs/modeldocs/minimax/video/)
|
||
server.Group("/v2", func(group *ghttp.RouterGroup) {
|
||
group.Middleware(middleware.ApiMaintenance, middleware.MaintenanceMode, middleware.ApiKeyAuth, middleware.ContentFilter)
|
||
group.POST("/video_generation", relay.HandleMiniMaxVideoSubmit)
|
||
group.GET("/query/video_generation/{task_id}", relay.HandleMiniMaxVideoRetrieve)
|
||
group.DELETE("/video_generation/{task_id}", relay.HandleMiniMaxVideoCancel)
|
||
})
|
||
|
||
// MiniMax 官方视频协议端点(Hailuo 系列 v1,扁平字段协议;查询走 query 参数,
|
||
// 成品经 files/retrieve 二跳由网关轮询时代取,官方无取消/删除端点)
|
||
server.Group("/v1", func(group *ghttp.RouterGroup) {
|
||
group.Middleware(middleware.ApiMaintenance, middleware.MaintenanceMode, middleware.ApiKeyAuth, middleware.ContentFilter)
|
||
group.POST("/video_generation", relay.HandleMiniMaxVideoV1Submit)
|
||
group.GET("/query/video_generation", relay.HandleMiniMaxVideoV1Retrieve)
|
||
})
|
||
}
|
||
|
||
// registerLandingPage 在未内嵌前端(非 embedweb 构建)时,为后端根路径注册项目介绍页,
|
||
// 避免浏览器直接访问后端地址时显示 404。内嵌前端时根路径由租户 SPA 承接,此页不注册。
|
||
func registerLandingPage(s *ghttp.Server) {
|
||
if web.Enabled {
|
||
return
|
||
}
|
||
s.Group("/", func(group *ghttp.RouterGroup) {
|
||
group.GET("/", landing.HandleLanding)
|
||
group.GET("/index.html", landing.HandleLanding)
|
||
})
|
||
}
|
||
|
||
// registerFrontendRoutes serves embedded frontend SPA assets.
|
||
// Admin console at /admin, tenant console at / (catch-all).
|
||
// Existing API routes take priority over these wildcard routes.
|
||
// When built without the "embedweb" tag, this is a no-op.
|
||
func registerFrontendRoutes(s *ghttp.Server) {
|
||
if !web.Enabled {
|
||
return
|
||
}
|
||
|
||
adminSub, _ := fs.Sub(web.AdminFS, "admin/dist")
|
||
tenantSub, _ := fs.Sub(web.TenantFS, "tenant/dist")
|
||
|
||
// Admin SPA: /admin/* → web/admin/dist/
|
||
s.Group("/admin", func(group *ghttp.RouterGroup) {
|
||
group.ALL("/*any", ghttp.WrapF(spaHandler(adminSub, "/admin")))
|
||
})
|
||
|
||
// Tenant SPA: /* → web/tenant/dist/ (lowest priority catch-all)
|
||
s.Group("/", func(group *ghttp.RouterGroup) {
|
||
group.ALL("/*any", ghttp.WrapF(spaHandler(tenantSub, "")))
|
||
})
|
||
}
|
||
|
||
// staticAssetExts 视为静态资源的扩展名集合:此类请求未命中文件时应返回 404
|
||
// 而非回退 index.html,避免把 HTML 当 JS/JSON 返回触发浏览器严格 MIME 校验报错。
|
||
var staticAssetExts = map[string]struct{}{
|
||
".js": {}, ".mjs": {}, ".css": {}, ".wasm": {}, ".map": {},
|
||
".jpg": {}, ".jpeg": {}, ".png": {}, ".gif": {}, ".webp": {}, ".avif": {},
|
||
".svg": {}, ".ico": {}, ".woff": {}, ".woff2": {}, ".ttf": {}, ".otf": {}, ".eot": {},
|
||
".mp4": {}, ".webm": {}, ".mp3": {}, ".wav": {}, ".json": {}, ".xml": {}, ".txt": {},
|
||
}
|
||
|
||
// isAssetPath 判断请求路径是否具有静态资源特征(assets 目录或带资源扩展名)。
|
||
func isAssetPath(path string) bool {
|
||
if strings.HasPrefix(path, "assets/") {
|
||
return true
|
||
}
|
||
ext := strings.ToLower(path[strings.LastIndex(path, "/")+1:])
|
||
dot := strings.LastIndex(ext, ".")
|
||
if dot <= 0 {
|
||
return false
|
||
}
|
||
_, ok := staticAssetExts[ext[dot:]]
|
||
return ok
|
||
}
|
||
|
||
// setStaticCacheHeader 按资源类型设置缓存策略:
|
||
// Vite 构建产物(assets/ 下 hash 文件名)永久强缓存;入口文件(index.html 等)要求每次进站重新校验,
|
||
// 保证发版后用户总能拿到新入口。
|
||
func setStaticCacheHeader(w http.ResponseWriter, path string) {
|
||
if strings.HasPrefix(path, "assets/") {
|
||
w.Header().Set("Cache-Control", "public, max-age=31536000, immutable")
|
||
return
|
||
}
|
||
w.Header().Set("Cache-Control", "no-cache")
|
||
}
|
||
|
||
// matchesEtag 判断请求头 If-None-Match 是否包含当前指纹(值可能是逗号分隔列表或通配 *)。
|
||
func matchesEtag(header, tag string) bool {
|
||
if header == "" {
|
||
return false
|
||
}
|
||
for _, part := range strings.Split(header, ",") {
|
||
part = strings.TrimSpace(part)
|
||
if part == "*" || part == tag {
|
||
return true
|
||
}
|
||
}
|
||
return false
|
||
}
|
||
|
||
// spaHandler returns an http.HandlerFunc that serves static files from the
|
||
// given filesystem, falling back to index.html for SPA client-side routing.
|
||
func spaHandler(root fs.FS, prefix string) http.HandlerFunc {
|
||
fileServer := http.FileServer(http.FS(root))
|
||
if prefix != "" {
|
||
fileServer = http.StripPrefix(prefix, fileServer)
|
||
}
|
||
|
||
// 启动时预读入口文件并预计算 SHA-256 指纹作为 ETag:
|
||
// index.html 走 no-cache 协商缓存,未重新发布的重复访问返回 304 空响应,
|
||
// 既保证发版即生效、又不产生重复下载开销。
|
||
indexBytes, indexErr := fs.ReadFile(root, "index.html")
|
||
indexTag := ""
|
||
if indexErr == nil {
|
||
sum := sha256.Sum256(indexBytes)
|
||
indexTag = `"` + hex.EncodeToString(sum[:]) + `"`
|
||
}
|
||
|
||
serveIndex := func(w http.ResponseWriter, r *http.Request) {
|
||
if indexErr != nil {
|
||
http.NotFound(w, r)
|
||
return
|
||
}
|
||
w.Header().Set("Content-Type", "text/html; charset=utf-8")
|
||
w.Header().Set("Cache-Control", "no-cache")
|
||
w.Header().Set("ETag", indexTag)
|
||
if matchesEtag(r.Header.Get("If-None-Match"), indexTag) {
|
||
w.WriteHeader(http.StatusNotModified)
|
||
return
|
||
}
|
||
w.Write(indexBytes)
|
||
}
|
||
|
||
return func(w http.ResponseWriter, r *http.Request) {
|
||
// Resolve the file path within the embedded FS
|
||
path := strings.TrimPrefix(r.URL.Path, prefix)
|
||
path = strings.TrimPrefix(path, "/")
|
||
|
||
// 入口文件统一走自建响应(fileServer 无法为 embed FS 提供 Last-Modified/ETag)
|
||
if path == "" || path == "index.html" {
|
||
serveIndex(w, r)
|
||
return
|
||
}
|
||
|
||
// Try to open the file; if it exists, serve it directly
|
||
if f, err := root.Open(path); err == nil {
|
||
f.Close()
|
||
setStaticCacheHeader(w, path)
|
||
fileServer.ServeHTTP(w, r)
|
||
return
|
||
}
|
||
|
||
// 带静态资源特征的请求(如发版后旧 hash chunk)缺失时返回 404,
|
||
// 不回退 SPA,方便前端自愈逻辑识别并整页刷新到新版本。
|
||
if isAssetPath(path) {
|
||
http.NotFound(w, r)
|
||
return
|
||
}
|
||
|
||
// Route not found — serve index.html (SPA fallback)
|
||
serveIndex(w, r)
|
||
}
|
||
}
|