跨Node项目触发TypeORM订阅者：A如何响应B的数据库写入？

TypeORM的实体订阅者是进程内的事件监听机制，仅会捕获当前应用进程中通过自身TypeORM连接执行的数据库操作所触发的ORM生命周期事件（如afterInsert）。项目B的写入操作属于独立进程的数据库请求，完全不在项目A的进程上下文内，因此A的订阅者无法感知到这类跨进程的数据库变更。

方案1：数据库触发器 + 消息队列（最可靠） #

通过数据库层面的触发器捕获数据变更，再借助消息队列通知项目A执行推送逻辑，适用于任何写入方式（不止项目B，其他外部写入也能触发）。

• 步骤：
  1. 给目标表创建INSERT触发器，当有新数据插入时，将变更数据发送到消息队列/数据库通知频道。
  2. 项目A启动消息队列消费者，监听对应频道，收到消息后执行WebSocket推送。
• 示例（PostgreSQL + Redis）：
  • 创建PostgreSQL触发器函数：

    CREATE OR REPLACE FUNCTION notify_new_record()
    RETURNS TRIGGER AS $$
    BEGIN
      -- 将新记录转为JSON发送到PostgreSQL内置通知频道
      PERFORM pg_notify('new_record_channel', row_to_json(NEW)::text);
      RETURN NEW;
    END;
    $$ LANGUAGE plpgsql;
    
    -- 给目标表绑定触发器
    CREATE TRIGGER trigger_new_record
    AFTER INSERT ON your_target_table
    FOR EACH ROW EXECUTE FUNCTION notify_new_record();
    

  • 项目A中监听通知并推送：

    import { Client } from 'pg';
    import WebSocket from 'ws';
    
    const pgClient = new Client({ /* 数据库配置 */ });
    await pgClient.connect();
    await pgClient.query('LISTEN new_record_channel');
    
    const wsServer = new WebSocket.Server({ port: 8080 });
    
    pgClient.on('notification', async (msg) => {
      const newRecord = JSON.parse(msg.payload);
      // 给所有在线WebSocket客户端推送
      wsServer.clients.forEach(client => {
        if (client.readyState === WebSocket.OPEN) {
          client.send(JSON.stringify({ type: 'new_record', data: newRecord }));
        }
      });
    });
    

方案2：项目B主动调用项目A的API #

项目B完成数据写入后，直接调用项目A暴露的专用接口，触发推送逻辑，实现简单可控。

• 示例：
  • 项目A新增鉴权后的推送触发接口：

    import express from 'express';
    import WebSocket from 'ws';
    
    const app = express();
    app.use(express.json());
    const wsServer = new WebSocket.Server({ port: 8080 });
    
    // 用API密钥做简单鉴权
    const VALID_API_KEY = 'your-secret-key';
    app.post('/api/trigger-push', (req, res) => {
      if (req.headers['x-api-key'] !== VALID_API_KEY) {
        return res.status(403).send('Invalid API key');
      }
      const newRecord = req.body;
      // 执行WebSocket推送
      wsServer.clients.forEach(client => {
        if (client.readyState === WebSocket.OPEN) {
          client.send(JSON.stringify({ type: 'new_record', data: newRecord }));
        }
      });
      res.status(200).send({ success: true });
    });
    
    app.listen(3000);
    

  • 项目B写入后调用接口：

    import axios from 'axios';
    
    // 写入数据库后触发通知
    const newRecord = await yourRepository.save(yourEntity);
    await axios.post('http://project-a-domain:3000/api/trigger-push', newRecord, {
      headers: { 'x-api-key': 'your-secret-key' }
    });
    

方案3：应用层事件总线（基于Redis/MQ） #

项目A和B共享同一个消息队列，项目B写入后主动发布事件，项目A监听事件并处理，无需依赖数据库触发器。

• 示例（Redis Pub/Sub）：
  • 项目B发布事件：

    import Redis from 'ioredis';
    const publisher = new Redis({ /* Redis配置 */ });
    
    // 写入后发布事件
    await yourRepository.save(yourEntity);
    await publisher.publish('new_record_event', JSON.stringify(newRecord));
    

  • 项目A监听事件并推送：

    import Redis from 'ioredis';
    import WebSocket from 'ws';
    
    const subscriber = new Redis({ /* Redis配置 */ });
    const wsServer = new WebSocket.Server({ port: 8080 });
    
    subscriber.subscribe('new_record_event', (err) => {
      if (err) console.error('订阅事件失败:', err);
    });
    
    subscriber.on('message', (_, msg) => {
      const newRecord = JSON.parse(msg);
      wsServer.clients.forEach(client => {
        if (client.readyState === WebSocket.OPEN) {
          client.send(JSON.stringify({ type: 'new_record', data: newRecord }));
        }
      });
    });
    

内容的提问来源于stack exchange，提问作者Maxime