BPOS-TMP-REPO/apps/api/src/app/routes/stream.ts

/**
 * SSE:前端唯一一条订阅连接(编码规范 §9.2)。
 *
 * 「SSE 事件到达 → 按信封目标失效对应 TanStack Query 缓存,由订阅层统一做;
 *   组件禁止自行轮询或手写缓存。」
 * 推送里**不带业务载荷**——只说「哪个目标动了」,画面仍由服务端投影重建。
 */
import type { Hono } from 'hono';
import { streamSSE } from 'hono/streaming';
import type { Broadcaster } from '../broadcast.ts';
import type { AppEnv } from '../http-env.ts';

export function registerStreamRoute(app: Hono<AppEnv>, broadcaster: Broadcaster): void {
  app.get('/api/stream', (c) =>
    streamSSE(c, async (stream) => {
      const queue: string[] = [];
      let wake: (() => void) | undefined;

      const unsubscribe = broadcaster.subscribe((touch) => {
        queue.push(JSON.stringify(touch));
        wake?.();
      });
      stream.onAbort(unsubscribe);

      await stream.writeSSE({ event: 'ready', data: '1' });

      // 心跳兼作断线探测:客户端 EventSource 自带重连,重连后前端整体失效一次
      const heartbeat = setInterval(() => {
        queue.push('');
        wake?.();
      }, 20_000);

      try {
        for (;;) {
          if (queue.length === 0) {
            await new Promise<void>((resolve) => {
              wake = resolve;
            });
            wake = undefined;
            continue;
          }
          const next = queue.shift()!;
          if (next === '') await stream.writeSSE({ event: 'ping', data: '1' });
          else await stream.writeSSE({ event: 'touch', data: next });
        }
      } finally {
        clearInterval(heartbeat);
        unsubscribe();
      }
    }),
  );
}