BPOS-TMP-REPO/apps/api/src/app/broadcast-bridge.ts

/**
 * 广播的**跨进程桥** —— 别的后端动了同一张单,这边的浏览器也得知道。
 *
 * 起因(2026-08-24 用户报的第三件事):「一直卡在 AI 推进中,刷新后才发现是已经推完了、
 * 需要人类确认了」。查下来推送本身是好的 —— 可 `createBroadcaster()` 是**进程内**的
 * 一个 Set:谁推进的,只有谁自己的 SSE 客户端收得到。而那会儿本机与局域网另一台
 * 各跑着一个后端连同一个库(同一批症状的另一面,见 `workers/task-claim.ts`):
 * 单子被那一台推着走,状态在库里早变了,**这一台的浏览器一个字都没收到** ——
 * 界面就停在「AI 推进中」,直到人手动刷新。
 *
 * 桥怎么搭:Postgres 自带的 `LISTEN/NOTIFY` —— 不引消息队列、不加部署件。
 *   · 本进程 publish 一条 → 顺手 `NOTIFY`(带上进程标识);
 *   · 一条**专用连接**常年 `LISTEN`,收到别的进程发来的就在本地 publish 一次。
 * **自己发的自己不再收**(靠进程标识认),不然一条推送在两台之间来回弹。
 * **转发进来的不再发出去**(2026-08-26):从前收到别人的就 `publish` 到本地 broadcaster,
 * 而本桥自己也订阅着它 —— 那条转发被当成「自己推的」再 NOTIFY 一遍(from 换成自己),
 * 对面收到不是它发的、再转发、再 NOTIFY……只挡「自己发的自己不收」这一跳,两台接力就是永动机:
 * 实测两台后端连同一个库,一次 `POST /api/gov/process/validate` 引出 45 秒 19 万条广播,
 * 工作台三把查询被打成连续重取(用户看到的就是「疯狂轮询三个接口」)。
 * 转发的那一条在本地 publish 前打上记号,「本地 → 库」那一侧认出记号就不再发。
 *
 * 桥断了不影响业务:推送本就是「告诉在线浏览器去重取」的提示,丢一条最坏是
 * 那一屏晚两秒(前端还有轮询兜底)。所以这里所有错误都只记一行,不往上抛。
 */
import type { Pool, PoolClient } from 'pg';
import type { Broadcaster, Touch } from './broadcast.ts';

/** 频道名 —— 全库一个就够,载荷里带着 kind/id,订阅方自己分 */
export const TOUCH_CHANNEL = 'jcai_touch';

/** 这一次进程的身份 —— 用来认「这条是我自己发的」,不必稳定、不必持久 */
const SELF = `${process.pid}-${Math.random().toString(36).slice(2, 8)}`;

type Wire = Touch & { readonly from: string };

/**
 * 把本地广播接到库上。返回 `stop()`;起不来不抛 —— 单进程照常跑,只是没有跨进程那一段。
 *
 * `notify` 用参数化的 `pg_notify` 而不是拼 `NOTIFY` 语句:载荷里有中文与引号,
 * 拼字符串迟早出事(而这条路上的字符串全来自业务)。
 */
export function bridgeBroadcaster(pool: Pool, broadcaster: Broadcaster, self: string = SELF): { stop(): void } {
  let listener: PoolClient | undefined;
  let stopped = false;
  /** 从库上转发进来的那几条 —— 它们不是本进程推的,不许再 NOTIFY 回去(按对象身份认,publish 同步派发) */
  const relayed = new WeakSet<Touch>();

  /** 本地 → 库:自己推的顺手 NOTIFY 一条(不 await,广播不该拖慢业务) */
  const unsubscribe = broadcaster.subscribe((touch) => {
    if (stopped || relayed.has(touch)) return;
    const wire: Wire = { ...touch, from: self };
    void pool
      .query('select pg_notify($1, $2)', [TOUCH_CHANNEL, JSON.stringify(wire)])
      .catch((error: unknown) => {
        process.stderr.write(`[broadcast-bridge] notify 失败:${error instanceof Error ? error.message : String(error)}\n`);
      });
  });

  /** 库 → 本地:一条专用连接常年 LISTEN,收到别人的就在本地 publish */
  const connect = async (): Promise<void> => {
    if (stopped) return;
    try {
      const client = await pool.connect();
      listener = client;
      client.on('notification', (message) => {
        if (message.channel !== TOUCH_CHANNEL || message.payload === undefined) return;
        try {
          const parsed: unknown = JSON.parse(message.payload);
          /*
           * **形状不对就丢** —— 频道是全库公用的,谁都能往里 NOTIFY 一句话。
           * 不校形状的话,一条 `"hello"` 会被当成 touch 广播出去:
           * 前端拿到 `kind: undefined` 谁也失效不了,还白跑一趟(界面不撒谎那一条,
           * 对推送同样算数)。
           */
          if (typeof parsed !== 'object' || parsed === null) return;
          const wire = parsed as Partial<Wire>;
          if (typeof wire.kind !== 'string' || typeof wire.id !== 'string') return;
          // **自己发的自己不收** —— 否则一条推送在进程之间来回弹
          if (wire.from === self) return;
          const touch: Touch = { kind: wire.kind as Touch['kind'], id: wire.id, reason: wire.reason ?? 'remote' };
          relayed.add(touch);
          broadcaster.publish(touch);
        } catch {
          /* 载荷根本不是 JSON:丢这一条,不拖垮连接 */
        }
      });
      client.on('error', (error) => {
        process.stderr.write(`[broadcast-bridge] 连接断了,5 秒后重连:${error.message}\n`);
        client.release();
        listener = undefined;
        setTimeout(() => void connect(), 5000).unref();
      });
      await client.query(`listen ${TOUCH_CHANNEL}`);
    } catch (error) {
      process.stderr.write(`[broadcast-bridge] 接不上,5 秒后重试:${error instanceof Error ? error.message : String(error)}\n`);
      setTimeout(() => void connect(), 5000).unref();
    }
  };
  void connect();

  return {
    stop() {
      stopped = true;
      unsubscribe();
      listener?.release();
      listener = undefined;
    },
  };
}