docs(deploy): 文本事件流 - 补充 SSE 反向代理与断线恢复指南 (#143)

Fixes #143
2026-08-06 11:02:52 +08:00
parent 670595a3aa
commit 10a0aa772f
4 changed files with 236 additions and 0 deletions
--- 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/<id>/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/<task-id>/text-events'
+```
+
+正常输出会先出现 `: connected`，随后是带 `id` 的 `event: delta`，最后是 `event: terminal`。如果只在任务结束后一次性出现全部内容，检查每一层是否仍有 `proxy_buffering on`、缓存规则或 gzip；如果长时间没有任何字节，检查 `proxy_read_timeout`、HTTPS 连接和后端容器日志。该排障命令只针对文本事件流，其他 API 仍使用默认缓存、安全和超时策略。
+
 ## 本地开发

 需要 Bun、Go 和可用的 OpenAI 兼容模型渠道。
--- 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)
--- a/backend/internal/handler/text_events_test.go
+++ 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)
+	}
+}
--- 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;