SSE 业务建模：从协议到生产中的前后端核心链路

浏览器                                  服务端
   ─── GET /jobs/job_001/events ──────────→
   ←── 200，Content-Type: text/event-stream
   ←── progress：正在分析资料
   ←── content：第一段正文
   ←── content：第二段正文
   ←── done：生成完成

id: 3
event: content
data: {"text":"第一段正文"}
 

const source = new EventSource("/jobs/job_001/events");
 
source.addEventListener("content", (event) => {
  const { text } = JSON.parse(event.data);
  console.log(text);
});
 
source.addEventListener("done", () => source.close());
source.addEventListener("failed", () => source.close());

浏览器 ── POST /jobs ──→ API ──→ jobs + outbox（同一事务）
                                  │
                             Dispatcher
                                  ↓
                                Queue
                                  ↓
                               Worker
                                  │
                       保存 job_events 和结果
                                  ↓
浏览器 ←── SSE Endpoint ←── 按游标读取已提交事件

type Job = {
  id: string;
  ownerId: string;
  status: "queued" | "running" | "completed" | "failed";
  lastSeq: number;
  resultId?: string;
};
 
type EventBody =
  | { type: "progress"; payload: { percent: number } }
  | { type: "content"; payload: { text: string } }
  | { type: "done"; payload: { resultId: string } }
  | { type: "failed"; payload: { message: string } };
 
type JobEvent = EventBody & {
  jobId: string;
  seq: number;
};
 
type ClientState = {
  jobId: string;
  text: string;
  progress: number;
  lastSeq: number;
  status: "watching" | "completed" | "failed";
  resultId?: string;
  error?: string;
};

POST /jobs
Content-Type: application/json
Idempotency-Key: <本次提交生成并在重试时复用的键>
 
{"prompt":"写一篇关于 SSE 的文章"}

HTTP/1.1 202 Accepted
Content-Type: application/json
 
{"jobId":"job_001"}

async function createJob(ownerId: string, key: string, prompt: string) {
  return db.transaction(async (tx) => {
    // (ownerId, key) 有唯一约束；同键异参应拒绝。
    // insertOrGet 必须在并发请求下也只创建一个 Job。
    const { job, created } = await tx.jobs.insertOrGet({
      ownerId, key, prompt, status: "queued", lastSeq: 0,
    });
 
    if (created) {
      await tx.outbox.insert({
        key: `enqueue:${job.id}`,
        type: "job.created",
        payload: { jobId: job.id },
      });
    }
    return { jobId: job.id };
  });
}

async function runJob(jobId: string) {
  const lease = await jobs.claim(jobId);
  if (!lease) return; // 已完成，或正在由其他 Worker 执行
 
  // withLease 负责续租；丢失租约时中止执行。
  await jobs.withLease(lease, async (signal) => {
    await events.append(lease, {
      type: "progress", payload: { percent: 10 },
    });
 
    let text = "";
    for await (const chunk of generateArticle(lease.prompt, signal)) {
      text += chunk;
      await events.append(lease, {
        type: "content", payload: { text: chunk },
      });
    }
 
    await jobs.completeWithResultAndEvent(lease, text);
  });
}

BEGIN;
SELECT id FROM jobs WHERE id = $1 FOR UPDATE;
-- 此处校验租约有效，且任务仍允许写入。
UPDATE jobs SET last_seq = last_seq + 1
WHERE id = $1 RETURNING last_seq;
INSERT INTO job_events (job_id, seq, type, payload)
VALUES ($1, $2, $3, $4); -- $2 是上一步返回的 last_seq
COMMIT;

SELECT job_id, seq, type, payload
FROM job_events
WHERE job_id = $1 AND seq > $2
ORDER BY seq ASC
LIMIT 100;

function encodeSse(event: JobEvent): string {
  return `id: ${event.seq}\nevent: ${event.type}\ndata: ${JSON.stringify(event.payload)}\n\n`;
}

async function streamJob(
  jobId: string,
  after: number,
  sink: { write(frame: string): Promise<void> },
  signal: AbortSignal,
) {
  let cursor = after;
  let heartbeatAt = Date.now();
 
  while (!signal.aborted) {
    const batch = await events.listAfter(jobId, cursor, 100);
 
    for (const event of batch) {
      await sink.write(encodeSse(event));
      cursor = event.seq;
      if (event.type === "done" || event.type === "failed") return;
    }
 
    if (batch.length > 0) continue;
 
    // 必须先查事件再查终态；终态事务提交后，lastSeq 才确定。
    const job = await jobs.get(jobId);
    if (isTerminal(job.status) && cursor >= job.lastSeq) return;
 
    if (Date.now() - heartbeatAt >= 15_000) {
      await sink.write(": heartbeat\n\n");
      heartbeatAt = Date.now();
    }
    await sleep(500, signal); // 可取消的等待
  }
}

async function* readEvents(body: ReadableStream<Uint8Array>) {
  const reader = body.getReader();
  const decoder = new TextDecoder();
  let buffer = "";
 
  try {
    while (true) {
      const { value, done } = await reader.read();
      if (done) return; // 未完成的帧不应用，重连时由日志补回
      buffer += decoder.decode(value, { stream: true });
 
      let boundary: number;
      while ((boundary = buffer.indexOf("\n\n")) >= 0) {
        const frame = buffer.slice(0, boundary);
        buffer = buffer.slice(boundary + 2);
        if (frame.startsWith(":")) continue;
 
        const fields = Object.fromEntries(
          frame.split("\n").map((line) => {
            const i = line.indexOf(":");
            return [line.slice(0, i), line.slice(i + 1).replace(/^ /, "")];
          }),
        );
        // 运行时校验 seq 为正安全整数、type 和 payload 匹配；失败就抛错。
        yield validateWireEvent({
          seq: Number(fields.id),
          type: fields.event,
          payload: JSON.parse(fields.data),
        });
      }
    }
  } finally {
    try { await reader.cancel(); }
    finally { reader.releaseLock(); }
  }
}

function applyEvent(state: ClientState, event: JobEvent): ClientState {
  if (event.jobId !== state.jobId) throw new Error("任务不匹配");
  if (event.seq <= state.lastSeq) return state;
  if (event.seq !== state.lastSeq + 1) throw new Error("事件存在缺口");
 
  const next = { ...state, lastSeq: event.seq };
  switch (event.type) {
    case "progress":
      return { ...next, progress: event.payload.percent };
    case "content":
      return { ...next, text: state.text + event.payload.text };
    case "done":
      return { ...next, status: "completed", resultId: event.payload.resultId };
    case "failed":
      return { ...next, status: "failed", error: event.payload.message };
  }
}

async function consumeOnce(state: ClientState, signal: AbortSignal) {
  const response = await fetch(
    `/jobs/${encodeURIComponent(state.jobId)}/events?after=${state.lastSeq}`,
    { signal, headers: { Accept: "text/event-stream" }, cache: "no-store" },
  );
  if (!response.ok) throw new HttpError(response.status);
  if (!response.headers.get("content-type")?.startsWith("text/event-stream")) {
    throw new Error("响应类型错误");
  }
  if (!response.body) throw new Error("响应体为空");
 
  for await (const wire of readEvents(response.body)) {
    const next = applyEvent(state, { ...wire, jobId: state.jobId });
    // 一个原子检查点：同时保存 text、status、lastSeq 等完整状态。
    // 例如在一个 IndexedDB 事务里写入同一条记录。
    await checkpoints.save(next);
    state = next;
    render(state);
    if (state.status !== "watching") return;
  }
  throw new Error("终态事件到达前，连接结束");
}

客户端成功保存：正文前半部分 + lastSeq = 3
                       ↓ 断网
Worker 继续保存：4 content、5 done
                       ↓ 重连 after=3
服务端查询：seq > 3，依次发送 4、5
                       ↓
客户端补上正文，保存 completed 状态，停止订阅