From 1d6e08fbda8527e37d8dcb6be4fcca28d3825db9 Mon Sep 17 00:00:00 2001 From: ddCat <2695092045@qq.com> Date: Thu, 6 Aug 2026 11:02:52 +0800 Subject: [PATCH] =?UTF-8?q?docs(deploy):=20=E6=96=87=E6=9C=AC=E4=BA=8B?= =?UTF-8?q?=E4=BB=B6=E6=B5=81=20-=20=E8=A1=A5=E5=85=85=20SSE=20=E5=8F=8D?= =?UTF-8?q?=E5=90=91=E4=BB=A3=E7=90=86=E4=B8=8E=E6=96=AD=E7=BA=BF=E6=81=A2?= =?UTF-8?q?=E5=A4=8D=E6=8C=87=E5=8D=97=20(#143)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fixes #143 --- README.md | 76 ++++++++++++++++ backend/internal/handler/routes.go | 91 ++++++++++++++++++++ backend/internal/handler/text_events_test.go | 53 ++++++++++++ nginx.conf | 16 ++++ 4 files changed, 236 insertions(+) create mode 100644 backend/internal/handler/text_events_test.go diff --git a/README.md b/README.md index 5c2e74365..22756e085 100644 --- a/README.md +++ b/README.md @@ -111,6 +111,82 @@ curl -fsSL https://raw.githubusercontent.com/ddcat-ai/open-ai-canvas/main/script 部署配置和 PostgreSQL 密码保存在 `/opt/open-ai-canvas/.env`，不要发送给他人，也不要删除 `backend-data`、`postgres-data` 和 `redis-data` 数据卷。数据卷持久化不等于备份，请定期备份 PostgreSQL 和上传文件。直接使用 IP 访问仅适合首次配置；公网长期使用必须绑定域名并配置 HTTPS。 +## 生产环境文本 SSE + +文本任务事件流是登录态接口 `GET /api/tasks/:id/text-events`。它只发送当前用户有权限访问的文本任务增量，响应类型为 `text/event-stream`；事件 `delta` 的 `id` 是单调递增的文本序号，`terminal` 表示任务已经成功、失败或取消。生产反向代理必须对这一条路径关闭响应缓冲和缓存，并允许长时间读取；不要把这些设置复制到所有 `/api/` 请求上。 + +### Nginx + +如果 Docker Compose 的网页容器暴露在本机 `3000` 端口，在 HTTPS server 中加入以下规则，并放在通用 `/api/` 规则之前。`proxy_pass` 也可以按你的部署拓扑改成实际网页入口；不要把后端 `8080` 直接暴露到公网。 + +```nginx +location ~ ^/api/tasks/[^/]+/text-events$ { + proxy_pass http://127.0.0.1:3000; + proxy_http_version 1.1; + proxy_set_header Host $host; + proxy_set_header X-Real-IP $remote_addr; + proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + proxy_set_header X-Forwarded-Proto $scheme; + proxy_set_header X-Forwarded-Host $host; + proxy_buffering off; + proxy_cache off; + gzip off; + proxy_read_timeout 3600s; + proxy_send_timeout 3600s; +} +``` + +项目镜像内的 `nginx.conf` 已包含同等规则，并只对 `/api/tasks//text-events` 关闭缓冲。外层 Nginx 和镜像内 Nginx 都存在时，两层都要保留该路径的流式设置；任一层重新缓冲都会让浏览器看起来直到任务结束才收到增量。 + +### Caddy + +Caddy 终止 HTTPS 后，将网页入口转发到 Compose 暴露的 `3000` 端口。`flush_interval -1` 让事件立即下发，长 `read_timeout` 防止长文本任务被网关主动断开： + +```caddyfile +canvas.example.com { + @textEvents path_regexp textEvents ^/api/tasks/[^/]+/text-events$ + reverse_proxy @textEvents 127.0.0.1:3000 { + flush_interval -1 + transport http { + read_timeout 1h + } + header_up X-Forwarded-Proto {scheme} + header_down Cache-Control "no-cache, no-transform" + } + + reverse_proxy 127.0.0.1:3000 +} +``` + +HTTPS 不是可选项：事件流使用登录 Cookie，用户配置的模型凭据也会参与请求；公网浏览器到反向代理的链路必须使用 HTTPS，并保留 `Host`、`X-Forwarded-For` 和 `X-Forwarded-Proto`。反向代理到网页容器可以使用隔离的 Compose 内网 HTTP。不要通过放宽 CORS、关闭鉴权或把 `Last-Event-ID` 放进 URL 来解决断线问题。 + +### Docker 与健康检查 + +`docker-compose.deploy.yml` / `docker-compose.server.yml` 只需要对外暴露网页容器的 `3000`；后端 `8080` 留在 Compose 网络内。部署后先确认容器健康： + +```bash +sudo docker compose --env-file .env -f docker-compose.deploy.yml -f docker-compose.build.yml ps +curl -fsS https://canvas.example.com/api/health +``` + +网页容器和后端镜像都带健康检查：网页检查 `/`，后端检查 `/api/health`。健康检查只能证明 HTTP 入口可用，不能证明代理正在逐事件转发；SSE 仍需按下面的命令验证。 + +### 断线恢复与排障 + +客户端重连时可以发送 `Last-Event-ID: <最后收到的序号>`，也可以使用 `?after=<序号>`；查询参数优先。服务端只返回 `sequence > after` 的 `delta`，因此同一段文本不会因为重连重复拼接。游标来自 SSE `id`，不是任务 ID。成功任务的增量保留 24 小时，失败或取消任务保留 7 天；超出保留期时应读取任务详情中的最终正文或失败草稿。 + +在已登录浏览器中复制 `open_ai_canvas_session` Cookie 后，可用 `curl -N` 检查首字节和增量是否在任务运行期间到达： + +```bash +curl -N --http1.1 \ + -H 'Accept: text/event-stream' \ + -H 'Last-Event-ID: 0' \ + -b 'open_ai_canvas_session=<登录 Cookie>' \ + 'https://canvas.example.com/api/tasks//text-events' +``` + +正常输出会先出现 `: connected`，随后是带 `id` 的 `event: delta`，最后是 `event: terminal`。如果只在任务结束后一次性出现全部内容，检查每一层是否仍有 `proxy_buffering on`、缓存规则或 gzip；如果长时间没有任何字节，检查 `proxy_read_timeout`、HTTPS 连接和后端容器日志。该排障命令只针对文本事件流，其他 API 仍使用默认缓存、安全和超时策略。 + ## 本地开发 需要 Bun、Go 和可用的 OpenAI 兼容模型渠道。 diff --git a/backend/internal/handler/routes.go b/backend/internal/handler/routes.go index 731d2847a..417344da8 100644 --- a/backend/internal/handler/routes.go +++ b/backend/internal/handler/routes.go @@ -1,6 +1,9 @@ package handler import ( + "encoding/json" + "errors" + "fmt" "net/http" "strconv" "time" @@ -112,6 +115,24 @@ func RegisterTaskRoutes(r *gin.RouterGroup, svc *service.Service) { } ok(c, result) }) + r.GET("/tasks/:id/text-events", func(c *gin.Context) { + user, err := currentUser(c, svc) + if err != nil { + failService(c, err) + return + } + after, err := taskTextEventCursor(c) + if err != nil { + fail(c, http.StatusBadRequest, err) + return + } + initial, err := svc.TaskTextReplay(user.ID, c.Param("id"), after) + if err != nil { + failService(c, err) + return + } + streamTaskTextEvents(c, svc, user.ID, c.Param("id"), after, initial) + }) r.POST("/tasks/:id/retry", func(c *gin.Context) { user, err := currentUser(c, svc) if err != nil { @@ -166,6 +187,76 @@ func RegisterTaskRoutes(r *gin.RouterGroup, svc *service.Service) { }) } +func taskTextEventCursor(c *gin.Context) (int64, error) { + raw := c.Query("after") + if raw == "" { + raw = c.GetHeader("Last-Event-ID") + } + if raw == "" { + return 0, nil + } + after, err := strconv.ParseInt(raw, 10, 64) + if err != nil || after < 0 { + return 0, errors.New("after 或 Last-Event-ID 必须是非负整数") + } + return after, nil +} + +func streamTaskTextEvents(c *gin.Context, svc *service.Service, userID string, taskID string, after int64, replay *service.TextReplayResult) { + c.Header("Content-Type", "text/event-stream; charset=utf-8") + c.Header("Cache-Control", "no-cache, no-transform") + c.Header("Connection", "keep-alive") + c.Header("X-Accel-Buffering", "no") + c.Status(http.StatusOK) + if _, err := fmt.Fprint(c.Writer, ": connected\n\n"); err != nil { + return + } + c.Writer.Flush() + + pollTicker := time.NewTicker(750 * time.Millisecond) + heartbeatTicker := time.NewTicker(15 * time.Second) + defer pollTicker.Stop() + defer heartbeatTicker.Stop() + for { + for _, delta := range replay.Deltas { + writeTaskTextSSE(c, "delta", delta.Sequence, map[string]any{"sequence": delta.Sequence, "content": delta.Content}) + after = delta.Sequence + } + if replay.Complete { + writeTaskTextSSE(c, "terminal", 0, replay) + return + } + select { + case <-c.Request.Context().Done(): + return + case <-heartbeatTicker.C: + if _, err := fmt.Fprint(c.Writer, ": heartbeat\n\n"); err != nil { + return + } + c.Writer.Flush() + case <-pollTicker.C: + } + next, err := svc.TaskTextReplay(userID, taskID, after) + if err != nil { + writeTaskTextSSE(c, "error", 0, map[string]string{"message": "任务文本流不可用"}) + return + } + replay = next + } +} + +func writeTaskTextSSE(c *gin.Context, event string, id int64, value any) { + data, err := json.Marshal(value) + if err != nil { + return + } + if id > 0 { + _, _ = fmt.Fprintf(c.Writer, "id: %d\n", id) + } + _, _ = fmt.Fprintf(c.Writer, "event: %s\ndata: %s\n\n", event, data) + c.Writer.Flush() +} + func RegisterSessionRoutes(r *gin.RouterGroup, svc *service.Service) { createSession := func(c *gin.Context) { user, err := currentUser(c, svc) diff --git a/backend/internal/handler/text_events_test.go b/backend/internal/handler/text_events_test.go new file mode 100644 index 000000000..34e9b6448 --- /dev/null +++ b/backend/internal/handler/text_events_test.go @@ -0,0 +1,53 @@ +package handler + +import ( + "net/http/httptest" + "strings" + "testing" + + "infinite-canvas/backend/internal/model" + "infinite-canvas/backend/internal/service" + + "github.com/gin-gonic/gin" +) + +func TestTaskTextEventCursorPrefersQueryAndSupportsLastEventID(t *testing.T) { + gin.SetMode(gin.TestMode) + request := httptest.NewRequest("GET", "/api/tasks/task-1/text-events?after=7", nil) + request.Header.Set("Last-Event-ID", "3") + context, _ := gin.CreateTestContext(httptest.NewRecorder()) + context.Request = request + after, err := taskTextEventCursor(context) + if err != nil || after != 7 { + t.Fatalf("cursor = %d, err = %v", after, err) + } + + request = httptest.NewRequest("GET", "/api/tasks/task-1/text-events", nil) + request.Header.Set("Last-Event-ID", "3") + context, _ = gin.CreateTestContext(httptest.NewRecorder()) + context.Request = request + after, err = taskTextEventCursor(context) + if err != nil || after != 3 { + t.Fatalf("Last-Event-ID cursor = %d, err = %v", after, err) + } +} + +func TestStreamTaskTextEventsWritesSequenceAndTerminalEvent(t *testing.T) { + gin.SetMode(gin.TestMode) + response := httptest.NewRecorder() + context, _ := gin.CreateTestContext(response) + context.Request = httptest.NewRequest("GET", "/api/tasks/task-1/text-events", nil) + replay := &service.TextReplayResult{ + Deltas: []model.TaskTextDelta{{Sequence: 8, Content: "增量"}}, + Complete: true, + FinalText: "增量", + } + streamTaskTextEvents(context, nil, "user-1", "task-1", 7, replay) + if response.Header().Get("Content-Type") != "text/event-stream; charset=utf-8" { + t.Fatalf("content type = %q", response.Header().Get("Content-Type")) + } + body := response.Body.String() + if !strings.Contains(body, "id: 8\nevent: delta") || !strings.Contains(body, "event: terminal") { + t.Fatalf("unexpected SSE body: %q", body) + } +} diff --git a/nginx.conf b/nginx.conf index 28350f239..0841e6152 100644 --- a/nginx.conf +++ b/nginx.conf @@ -35,6 +35,22 @@ server { proxy_read_timeout 3600s; } + # 文本任务事件流必须逐条转发；该规则只作用于 SSE，不改变其他 API 的缓存和超时策略。 + location ~ ^/api/tasks/[^/]+/text-events$ { + proxy_pass http://backend:8080; + proxy_http_version 1.1; + proxy_set_header Host $host; + proxy_set_header X-Real-IP $remote_addr; + proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + proxy_set_header X-Forwarded-Proto $canvas_forwarded_proto; + proxy_set_header X-Forwarded-Host $host; + proxy_buffering off; + proxy_cache off; + gzip off; + proxy_read_timeout 3600s; + proxy_send_timeout 3600s; + } + location /api/ { proxy_pass http://backend:8080; proxy_http_version 1.1;