BPOS-Service-Gateway/test/integration/sse.test.ts

import { describe, expect, it, beforeAll, vi } from 'vitest';
import { Hono } from 'hono';
import { SseHub, isSubscribable } from '../../src/modules/sse/hub';
import type { Connection, SseEvent } from '../../src/modules/sse/hub';
import { sseRoutes } from '../../src/modules/sse/routes';
import { createHttpTaskAccess } from '../../src/modules/sse/task-access';
import { createHttpWorkflowAccess } from '../../src/modules/sse/workflow-access';
import { buildApp } from '../../src/app/app';
import { makeToken, setupTestJwks } from '../support/token';

function conn(id: string, staffSub: string, topics: string[] = []): { c: Connection; got: SseEvent[] } {
  const got: SseEvent[] = [];
  return { c: { id, staffSub, isAdmin: false, topics: new Set(topics), send: (e) => got.push(e) }, got };
}

/** 一律放行/一律拒绝的假口：两口分开传，才证明得了"这一拒来自哪一口" */
const ALLOW = { allowed: async (): Promise<boolean> => true };
const DENY = { allowed: async (): Promise<boolean> => false };

describe('SseHub 路由', () => {
  it('staff 精确投递、system 广播、task 按订阅集、回放窗口', () => {
    const hub = new SseHub(8);
    const a = conn('a', 'sub-a', ['touch.task.t1']);
    const b = conn('b', 'sub-b');
    hub.register(a.c);
    hub.register(b.c);
    hub.deliver('touch.system', '{}');
    hub.deliver('touch.staff.sub-a', '{}');
    hub.deliver('touch.task.t1', '{}');
    expect(a.got.map((e) => e.topic)).toEqual(['touch.system', 'touch.staff.sub-a', 'touch.task.t1']);
    expect(b.got.map((e) => e.topic)).toEqual(['touch.system']);
    // 断线重连回放：c 以 lastEventId=1 上线且订阅 t1 → 补 seq>1 的可见事件。
    // D11：register 只登记、replay 单独一步——路由要把 hello 插在两者之间。
    const c = conn('c', 'sub-c', ['touch.task.t1']);
    hub.register(c.c);
    hub.replay(c.c, 1);
    expect(c.got.map((e) => e.topic)).toEqual(['touch.task.t1']);
  });

  it('subscribe 授权：access 拒 → 不入订阅集', async () => {
    const hub = new SseHub();
    const a = conn('a', 'sub-a');
    hub.register(a.c);
    const res = await hub.subscribe('a', 'stream.task.t9', DENY, ALLOW);
    expect(res.ok).toBe(false);
    expect(a.c.topics.has('stream.task.t9')).toBe(false);
  });

  it('D3：形状校验排在连接查找之前，失败带判别式 reason 供路由分流', async () => {
    const hub = new SseHub();
    // 连接不存在 + 形状也不对 → 必须报形状（否则 HTTP 面永远回不出 422）
    expect(await hub.subscribe('nope', 'touch.staff.hack', DENY, DENY)).toMatchObject({
      ok: false,
      reason: 'bad-shape',
    });
    expect(await hub.subscribe('nope', 'touch.task.t1', DENY, DENY)).toMatchObject({
      ok: false,
      reason: 'no-conn',
    });
    const a = conn('a', 'sub-a');
    hub.register(a.c);
    expect(await hub.subscribe('a', 'stream.task.t9', DENY, DENY)).toMatchObject({
      ok: false,
      reason: 'denied',
    });
  });

  it('N3（原 I6 反转）：touch.workflow.* 放回可订阅集——规则到位了才收得进来', async () => {
    // I6 当初把它摘出去，是因为**当时没有规则**：形状一过就直接进订阅集，而鉴权分支只覆盖
    // touch.task./stream.task.，任何登录用户订 touch.workflow.<任意 id> 都能收别人流程的推送。
    // N3 补上归属口（参与人/审批人，admin 旁路）之后放回来——放形状与补鉴权是同一次改动，
    // 下面两条用例就是"规则确实生效"的那一半，缺了它这条断言等于把 I6 原样退回去。
    expect(isSubscribable('touch.workflow.w1')).toBe(true);
    expect(isSubscribable('touch.task.t1')).toBe(true);
    expect(isSubscribable('stream.task.t1')).toBe(true);
    // 放回的只是这一节：多一节的主题、以及别的域照旧不可订阅
    expect(isSubscribable('touch.workflow.w1.extra')).toBe(false);
    expect(isSubscribable('touch.workflow.')).toBe(false);
    expect(isSubscribable('touch.staff.hack')).toBe(false);
  });

  it('N3：工作流按归属口放行——拒则不入集不投递，准则入集且收得到', async () => {
    const hub = new SseHub();
    const a = conn('a', 'sub-a');
    hub.register(a.c);
    // 任务口一律放行：这样"被拒"只可能来自**工作流**那一口，而不是碰巧被任务授权挡住
    const denied = await hub.subscribe('a', 'touch.workflow.w1', ALLOW, DENY);
    expect(denied).toMatchObject({ ok: false, reason: 'denied' });
    expect(denied.ok === false && denied.message).toContain('参与人/审批人或 admin');
    expect(a.c.topics.size).toBe(0);
    hub.deliver('touch.workflow.w1', '{}');
    expect(a.got).toEqual([]); // 拒了就一条都不该扇过去

    const asked: string[][] = [];
    const ok = await hub.subscribe('a', 'touch.workflow.w1', ALLOW, {
      allowed: async (id: string, sub: string): Promise<boolean> => {
        asked.push([id, sub]);
        return true;
      },
    });
    expect(ok).toEqual({ ok: true });
    expect(asked).toEqual([['w1', 'sub-a']]); // 问的是主题末节＋**连接的** staffSub，不是调用方的
    hub.deliver('touch.workflow.w1', '{"n":1}');
    expect(a.got.map((e) => e.topic)).toEqual(['touch.workflow.w1']);
  });

  it('N3：admin 订工作流不问归属口——admin 全可见，白跑一次跨服务往返没意义', async () => {
    const hub = new SseHub();
    const a = conn('adm', 'sub-adm');
    hub.register({ ...a.c, isAdmin: true }); // topics/send 与 a.c 共用，断言照旧看 a
    let calls = 0;
    const counting = {
      allowed: async (): Promise<boolean> => {
        calls += 1;
        return false; // 真被问到就会拒——订上了即证明没问
      },
    };
    expect(await hub.subscribe('adm', 'touch.workflow.w9', counting, counting)).toEqual({ ok: true });
    expect(calls).toBe(0);
    expect(a.c.topics.has('touch.workflow.w9')).toBe(true);
    hub.deliver('touch.workflow.w9', '{}');
    expect(a.got.map((e) => e.topic)).toEqual(['touch.workflow.w9']);
  });

  it('D12：shutdown 逐条调 close 把手并摘除连接', () => {
    const hub = new SseHub();
    let closed = 0;
    const a = conn('a', 'sub-a');
    hub.register({ ...a.c, close: () => (closed += 1) });
    hub.shutdown();
    expect(closed).toBe(1);
    hub.deliver('touch.system', '{}');
    expect(a.got).toEqual([]); // 已摘除，不再收扇出
  });
});

describe('SSE HTTP 面', () => {
  beforeAll(async () => {
    await setupTestJwks();
  });

  it('subscribe 主题形状校验 422；无 token 401', async () => {
    const app = buildApp();
    expect((await app.request('/api/stream/subscribe', { method: 'POST' })).status).toBe(401);
    const token = await makeToken({ roles: ['user'], aud: 'api-gateway' });
    const bad = await app.request('/api/stream/subscribe', {
      method: 'POST',
      headers: { authorization: `Bearer ${token}`, 'content-type': 'application/json' },
      body: JSON.stringify({ connectionId: 'nope', add: ['touch.staff.hack'] }),
    });
    expect(bad.status).toBe(422);
  });

  it('F1：形状预检先于变更——整单 422 时一条订阅都不留', async () => {
    const hub = new SseHub();
    const a = conn('live-1', 'test-sub-1'); // token 的默认 sub，过得了 F2 的归属校验
    hub.register(a.c);
    // 直接挂 sseRoutes 并注入"一律放行"的授权口：走 buildApp 的话内建 access 会在
    // await token() 处就抛（测试环境没有 KEYCLOAK_CLIENT_SECRET），t1 到不了订阅集，
    // 这条用例也就证明不了"预检之前不会发生变更"这件事。
    let accessCalls = 0;
    const access = {
      allowed: async (): Promise<boolean> => {
        accessCalls += 1;
        return true;
      },
    };
    const app = new Hono().route('/api/stream', sseRoutes(hub, access, access));
    const token = await makeToken({ roles: ['user'], aud: 'api-gateway' });
    const res = await app.request('/api/stream/subscribe', {
      method: 'POST',
      headers: { authorization: `Bearer ${token}`, 'content-type': 'application/json' },
      body: JSON.stringify({ connectionId: 'live-1', add: ['touch.task.t1', 'bogus'] }),
    });
    expect(res.status).toBe(422);
    // 关键：合法的 t1 排在非法项**前面**，边订边查的写法会把它真订上
    expect(a.c.topics.has('touch.task.t1')).toBe(false);
    expect(a.c.topics.size).toBe(0);
    expect(accessCalls).toBe(0); // 且一次授权往返都不该白跑
  });

  it('I7：add/remove 不是数组 → 422（不是 500）', async () => {
    // `as string[]` 只是断言。字符串上没有 .find，此前直接 TypeError 冒到 500——
    // 请求写错却报成服务端故障，而 500 在网关这一层还会被客户端/重试层当成可重试。
    const hub = new SseHub();
    const a = conn('live-7', 'test-sub-1');
    hub.register(a.c);
    const app = new Hono().route('/api/stream', sseRoutes(hub, ALLOW, ALLOW));
    const token = await makeToken({ roles: ['user'], aud: 'api-gateway' });
    const post = async (payload: unknown): Promise<Response> =>
      app.request('/api/stream/subscribe', {
        method: 'POST',
        headers: { authorization: `Bearer ${token}`, 'content-type': 'application/json' },
        body: JSON.stringify(payload),
      });

    const badAdd = await post({ connectionId: 'live-7', add: 'touch.task.t1' });
    expect(badAdd.status).toBe(422);
    expect(await badAdd.json()).toMatchObject({ error: 'validation_failed' });
    const badRemove = await post({ connectionId: 'live-7', remove: { '0': 'touch.task.t1' } });
    expect(badRemove.status).toBe(422);
    expect(a.c.topics.size).toBe(0);                       // 422 零副作用
    // 缺省（undefined）仍然合法：只订不退、只退不订都是正常用法
    expect((await post({ connectionId: 'live-7', add: ['touch.task.t1'] })).status).toBe(200);
    expect(a.c.topics.has('touch.task.t1')).toBe(true);
  });

  it('F2：连接归属校验——他人 connectionId 一律 403（add 与 remove 同守）', async () => {
    const hub = new SseHub();
    const app = buildApp(hub);
    const owner = conn('own-1', 'test-sub-1');
    owner.c.topics.add('touch.task.t1');
    hub.register(owner.c);
    const post = async (tok: string, payload: unknown): Promise<Response> =>
      app.request('/api/stream/subscribe', {
        method: 'POST',
        headers: { authorization: `Bearer ${tok}`, 'content-type': 'application/json' },
        body: JSON.stringify(payload),
      });

    // 闯入者：合法 token，但连接不是他的 —— remove 此前完全不过任何校验
    const intruder = await makeToken({ sub: 'sub-intruder', roles: ['user'], aud: 'api-gateway' });
    const denied = await post(intruder, { connectionId: 'own-1', remove: ['touch.task.t1'] });
    expect(denied.status).toBe(403);
    expect(await denied.json()).toMatchObject({ error: 'forbidden' });
    expect(owner.c.topics.has('touch.task.t1')).toBe(true); // 没被别人退订掉

    // 主人自己来则照常放行——校验不是一刀切拒绝
    const self = await makeToken({ roles: ['user'], aud: 'api-gateway' }); // sub=test-sub-1
    const ok = await post(self, { connectionId: 'own-1', remove: ['touch.task.t1'] });
    expect(ok.status).toBe(200);
    expect(owner.c.topics.has('touch.task.t1')).toBe(false);
  });

  it('F3：task-access 非 200 → false（失败关闭）且 id 都转义进 URL', async () => {
    const warn = vi.spyOn(console, 'warn').mockImplementation(() => undefined);
    const calls: string[] = [];
    try {
      vi.stubGlobal('fetch', (url: string | URL) => {
        calls.push(String(url));
        return Promise.resolve(new Response('down', { status: 503 }));
      });
      const access = createHttpTaskAccess(async () => 'stub-token');
      expect(await access.allowed('t 1', 'sub/a')).toBe(false);
      expect(calls[0]).toContain('/task/t%201/access?staff=sub%2Fa');
      expect(warn).toHaveBeenCalled(); // 不出声的"人人被拒"无从诊断

      vi.stubGlobal('fetch', () =>
        Promise.resolve(
          new Response(JSON.stringify({ allowed: true }), {
            status: 200,
            headers: { 'content-type': 'application/json' },
          }),
        ),
      );
      expect(await access.allowed('t1', 's1')).toBe(true);
    } finally {
      vi.unstubAllGlobals();
      warn.mockRestore();
    }
  });

  it('workflow-access 与 task-access 同形：非 200 → false（失败关闭），打的是 /workflow/:id/access', async () => {
    const warn = vi.spyOn(console, 'warn').mockImplementation(() => undefined);
    const calls: string[] = [];
    try {
      vi.stubGlobal('fetch', (url: string | URL) => {
        calls.push(String(url));
        return Promise.resolve(new Response('down', { status: 503 }));
      });
      const wf = createHttpWorkflowAccess(async () => 'stub-token');
      expect(await wf.allowed('w 1', 'sub/a')).toBe(false);
      // 路径与任务口分开：任务侧是两条端点，判据也不同源
      expect(calls[0]).toContain('/workflow/w%201/access?staff=sub%2Fa');
      expect(warn).toHaveBeenCalled(); // 不出声的"人人被拒"无从诊断

      vi.stubGlobal('fetch', () =>
        Promise.resolve(
          new Response(JSON.stringify({ allowed: true }), {
            status: 200,
            headers: { 'content-type': 'application/json' },
          }),
        ),
      );
      expect(await wf.allowed('w1', 's1')).toBe(true);
    } finally {
      vi.unstubAllGlobals();
      warn.mockRestore();
    }
  });

  it('D11/D12：hello 先于回放；hub.shutdown() 能把在途长连收尾', async () => {
    const hub = new SseHub(8);
    const app = buildApp(hub);
    hub.deliver('touch.system', '{"n":1}'); // seq 1
    hub.deliver('touch.staff.test-sub-1', '{"n":2}'); // seq 2
    const token = await makeToken({ roles: ['user'], aud: 'api-gateway' });
    const res = await app.request(`/api/stream?access_token=${token}`, {
      headers: { 'last-event-id': '1' },
    });
    expect(res.status).toBe(200);
    const body = res.body;
    if (!body) throw new Error('SSE 响应缺流体');
    const reader = body.getReader();
    const decoder = new TextDecoder();
    let text = '';
    while (!text.includes('event: touch.staff.test-sub-1')) {
      const { value, done } = await reader.read();
      if (done) break;
      text += decoder.decode(value, { stream: true });
    }
    expect(text).toContain('event: hello');
    expect(text.indexOf('event: hello')).toBeLessThan(text.indexOf('event: touch.staff.test-sub-1'));
    expect(text).not.toContain('event: touch.system'); // seq 1 ≤ lastEventId，不补

    // D12：回调里挂住的 promise 必须能被 close 把手 settle——settle 了 streamSSE 才会
    // close 流，客户端这边才读得到 done。旧写法 `await new Promise(() => {})` 永不 settle：
    // 下面这次 read 会一直吊着（用例超时），线上则是 SIGTERM 关不掉服务。
    hub.shutdown();
    let ended = false;
    for (let i = 0; i < 4 && !ended; i += 1) ended = (await reader.read()).done;
    expect(ended).toBe(true);
  });
});