Blizzard
e5fa0ae36c
第 2 层网关持久层落地，遵守 sundynix_ 表名前缀 + AutoMigrate 约定。 - store: GORM(NamingStrategy 前缀 sundynix_/单数) → User=sundynix_user, Task=sundynix_task 启动 AutoMigrate；go-redis/v9 滑动窗口限流(Incr+Expire,按 IP) - 优雅降级：连不上库则 warn 继续(不 fatal)，保证无 Docker 的 make demo 仍跑通 - handler: SubmitTask 持久化任务(best-effort)，Billing 真实读库返回 tasks_submitted - main: OpenPostgres/OpenRedis 读 POSTGRES_DSN/REDIS_ADDR 环境变量 - 验证: 4 模块 build ✓；e2e 3 测试 PASS；live 双路径(真实库持久化 + 坏DSN降级)实测通过 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
98 lines
2.8 KiB
Go
98 lines
2.8 KiB
Go
// Package handler 实现网关的 HTTP 处理器。
|
||
package handler
|
||
|
||
import (
|
||
"encoding/json"
|
||
"io"
|
||
"log"
|
||
"net/http"
|
||
|
||
"github.com/gin-gonic/gin"
|
||
|
||
"github.com/sundynix/sundynix-gateway/internal/dsl"
|
||
"github.com/sundynix/sundynix-gateway/internal/nats"
|
||
"github.com/sundynix/sundynix-gateway/internal/store"
|
||
)
|
||
|
||
type Handler struct {
|
||
db *store.Postgres
|
||
cache *store.Redis
|
||
bus *nats.Bus
|
||
}
|
||
|
||
func New(db *store.Postgres, cache *store.Redis, bus *nats.Bus) *Handler {
|
||
return &Handler{db: db, cache: cache, bus: bus}
|
||
}
|
||
|
||
// SubmitTask: 解析客户端导出的 JSON DSL，组装为 Task，Publish 到 sundynix.tasks.*。
|
||
func (h *Handler) SubmitTask(c *gin.Context) {
|
||
var raw json.RawMessage
|
||
if err := c.ShouldBindJSON(&raw); err != nil {
|
||
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
|
||
return
|
||
}
|
||
task, err := dsl.ParseAndAssemble(raw) // Task DSL Parser & Assembly
|
||
if err != nil {
|
||
c.JSON(http.StatusUnprocessableEntity, gin.H{"error": err.Error()})
|
||
return
|
||
}
|
||
// 持久化任务提交（best-effort：降级模式下静默跳过，不阻断发布）。
|
||
if err := h.db.SaveTask(c.Request.Context(), task.ID, string(task.Graph)); err != nil {
|
||
log.Printf("[gateway] save task %s failed: %v", task.ID, err)
|
||
}
|
||
if err := h.bus.PublishTask(c.Request.Context(), task); err != nil {
|
||
c.JSON(http.StatusBadGateway, gin.H{"error": err.Error()})
|
||
return
|
||
}
|
||
c.JSON(http.StatusAccepted, gin.H{"task_id": task.ID})
|
||
}
|
||
|
||
// StreamTask: 订阅 sundynix.streams.<task_id>，以 SSE 把零拷贝 Token Stream 推给客户端。
|
||
func (h *Handler) StreamTask(c *gin.Context) {
|
||
taskID := c.Param("id")
|
||
c.Writer.Header().Set("Content-Type", "text/event-stream")
|
||
c.Writer.Header().Set("Cache-Control", "no-cache")
|
||
c.Writer.Header().Set("Connection", "keep-alive")
|
||
|
||
tokens := make(chan []byte, 256)
|
||
done := make(chan struct{})
|
||
unsub, err := h.bus.SubscribeTokens(taskID,
|
||
func(tok []byte) {
|
||
select {
|
||
case tokens <- tok:
|
||
default: // 背压保护：客户端过慢则丢弃，避免阻塞 NATS 回调
|
||
}
|
||
},
|
||
func() { close(done) },
|
||
)
|
||
if err != nil {
|
||
c.JSON(http.StatusBadGateway, gin.H{"error": err.Error()})
|
||
return
|
||
}
|
||
defer func() { _ = unsub() }()
|
||
|
||
// gin 的流式写：返回 false 即结束响应。
|
||
c.Stream(func(w io.Writer) bool {
|
||
select {
|
||
case tok := <-tokens:
|
||
c.SSEvent("token", string(tok))
|
||
return true
|
||
case <-done:
|
||
c.SSEvent("done", taskID)
|
||
return false
|
||
case <-c.Request.Context().Done():
|
||
return false
|
||
}
|
||
})
|
||
}
|
||
|
||
func (h *Handler) Billing(c *gin.Context) {
|
||
// TODO: 商业化与计费模块；暂以已提交任务计数演示真实读库。
|
||
n, err := h.db.CountTasks(c.Request.Context())
|
||
if err != nil {
|
||
c.JSON(http.StatusBadGateway, gin.H{"error": err.Error()})
|
||
return
|
||
}
|
||
c.JSON(http.StatusOK, gin.H{"status": "ok", "tasks_submitted": n, "persisted": h.db.Enabled()})
|
||
}
|