NestJS SSE 服务端实战：鉴权、断连取消与资源清理

Server-Sent Events（SSE）用一个长 HTTP 响应把服务端事件持续送给浏览器。它适合任务进度、通知和 AI 生成等“客户端发起，服务端持续返回”的工作流。接口能返回事件只是第一步；上线质量取决于授权、断连传播、上游取消、资源配额、代理配置和可观测性。

什么时候选 SSE

SSE 基于 HTTP 文本响应，服务端单向推送；浏览器原生 EventSource 管理连接和重连。若客户端需要 POST、JSON 请求体或自定义认证头，可让客户端用 fetch 读取 SSE 格式响应，由客户端解析事件帧。需要双向实时通信时再评估 WebSocket。无论何种方式，业务数据仍应落在数据库、对象存储或其他权威存储中，不能只存在连接内存里。

NestJS 的 @Sse() 路由返回 RxJS Observable<MessageEvent>（或解析为 Observable 的 Promise）。事件的 data 可为字符串或对象，type、id、retry、comment 对应 SSE 字段。事件对象和前端消费协议需要版本化，不要把内部异常堆栈原样发给客户端。

一个受限的进度流控制器

下面的例子假设 JobService.watch() 返回可取消的 Observable，且服务层负责校验任务属于当前用户。实际系统可按已有的守卫、策略和仓库接口调整：

import { Controller, Param, Sse, UseGuards } from '@nestjs/common';
import { finalize, map, Observable } from 'rxjs';

@Controller('jobs')
@UseGuards(JwtAuthGuard)
export class JobsController {
  constructor(private readonly jobs: JobService) {}

  @Sse(':jobId/events')
  events(@Param('jobId') jobId: string): Observable<MessageEvent> {
    return this.jobs.watchForCurrentUser(jobId).pipe(
      map((event) => ({
        id: String(event.sequence),
        type: 'progress',
        data: {
          version: 1,
          jobId: event.jobId,
          state: event.state,
          completed: event.completed,
          total: event.total,
        },
      })),
      finalize(() => this.jobs.releaseWatch(jobId)),
    );
  }
}

finalize 在 Observable 完成、报错或被取消订阅时执行。这里的服务层还应保证：只释放当前订阅持有的资源，而不是误关其他用户共享的连接。若数据源本身支持 teardown，应在 watchForCurrentUser 创建 Observable 时返回取消逻辑，让取消能一路传到 Redis 订阅、数据库游标、上游 HTTP 请求或模型 SDK。

不能仅依赖 finalize 做权限校验；授权必须在建立流之前执行。路径中的 jobId 只是用户输入，必须在服务端验证访问权。任何事件中的对象和字段也应只包含当前用户有权看到的数据。

让取消信号到达上游

finalize 是资源释放钩子，不会自动取消模型供应商的工作。若应用调用上游 HTTP API，应将每个流订阅与一个 AbortController 绑定，并确保上游 SDK 接受 signal。伪代码结构如下：

function createModelEventStream(prompt: string): Observable<MessageEvent> {
  return new Observable((subscriber) => {
    const controller = new AbortController();
    let upstreamReader: ReadableStreamDefaultReader<Uint8Array> | undefined;

    void (async () => {
      try {
        const response = await fetch(MODEL_ENDPOINT, {
          method: 'POST',
          headers: modelHeaders(),
          body: JSON.stringify({ prompt, stream: true }),
          signal: controller.signal,
        });
        if (!response.ok || !response.body) throw new Error('Upstream stream failed');

        upstreamReader = response.body.getReader();
        while (!subscriber.closed) {
          const { value, done } = await upstreamReader.read();
          if (done) break;
          for (const event of decodeProviderFrames(value)) {
            subscriber.next({ type: 'token', data: validateProviderEvent(event) });
          }
        }
        if (!subscriber.closed) subscriber.complete();
      } catch (error) {
        if (!controller.signal.aborted && !subscriber.closed) {
          subscriber.error(new Error('Generation failed'));
        }
      } finally {
        upstreamReader?.releaseLock();
      }
    })();

    return () => {
      controller.abort();
    };
  });
}

这是生命周期示意，不是完整的 provider SSE parser：decodeProviderFrames 必须处理 UTF-8 多字节字符跨 chunk、行尾跨 chunk、事件边界、多个 data: 字段和 provider 的结束标记。API key 只留在服务端；错误映射为稳定的公共错误码，不返回 endpoint、token 或堆栈。还要处理取消发生在上游响应头返回之前的竞态，并确认所用 SDK 的取消语义。

背压、并发和成本

为每用户、每 IP、每账户设置并发生成上限、速率限制、输入长度、最大输出、总耗时和上游费用预算。
一个用户建立多个连接时要单独计量；如果服务端使用进程内 Map 做限额，多实例部署后它不是全局限制，应使用共享限流存储或网关策略。
不要把无限增长的 token 队列放在内存中等待慢客户端。优先让上游读取随下游消费推进；否则设置队列上限，超限时取消任务并记录明确原因。
进度消息可以合并，只保留最新百分比；审计事件和用户消息不能静默丢弃，应写入持久化队列并以游标恢复。
重连不等于续传。若需续传，需定义事件 ID、服务端保留窗口、去重和过期游标行为；否则返回明确的“任务仍运行/结果可重新查询”状态。
Observable 的取消只覆盖本订阅。共享的上游任务要用引用计数或任务 owner 策略，不能因一个读者断线就误取消其他读者仍在看的工作。

反向代理与传输链路

代理缓冲可能让一切 SSE 事件等到响应结束才到达浏览器。应对具体流路由检查 Nginx、负载均衡、API 网关和 CDN 的 buffering、缓存、压缩、空闲超时和最大响应时长；不要为了 SSE 把全站响应缓冲都关闭。定期发送轻量心跳可以减少某些代理的空闲断开，但心跳间隔必须短于链路空闲超时，而且不能替代任务截止时间。

部署验收时同时观察：应用是否按时 next、代理是否转发、浏览器是否逐帧收到、客户端取消后上游是否停止、任务状态是否可查询。生产配置中应为流路由单独定义策略并在变更后验证普通 JSON/API 路由没有受到影响。

日志与指标

建议记录 requestId、任务 ID、用户 ID 的内部不可逆标识、建立时间、首事件耗时、输出 token 数、结束原因、客户端断连、上游取消结果和限流原因。提示词、完整模型输出、凭据和个人信息默认不进日志。分别监控活跃连接、每用户并发、断连率、首 token 延迟、总耗时、取消后的上游残留时间、网关 499/502/504 和上游错误率。

验收清单

未认证、已封禁用户、非任务所有者不能建立事件流。
客户端停止后，Observable teardown、上游请求和计费任务均按策略停止；共享任务不会被其他订阅的断开误杀。
上游 401/429/500、无 body、超时、畸形事件和无结束标记都有稳定终态。
代理环境下第一个事件能及时到达，心跳能维持连接且空闲时限会正确结束任务。
大量慢客户端不会导致无限缓冲、连接泄漏或内存持续增长。
客户端重连时按明确的续传策略处理重复和过期事件。
日志没有 API key、原始敏感提示词或不必要的响应正文。
多实例下的限流和配额测试使用共享状态，不误以为进程内变量可以全局计数。