WebSocket场景下Mongoose watch监听集合新增文档无返回值问题排查

已确认的代码层缺陷 #

• 连接监听使用once方法存在逻辑硬伤
  当前实现通过wss.once("connection", ...)绑定连接事件，Once意味着整个WebSocket服务只会响应第一个接入的客户端连接，后续所有新建立的连接都不会触发回调逻辑，也就不会初始化对应的watch监听。该缺陷完全符合“此前功能可正常运行，目前出现异常”的场景：早期单连接测试时逻辑可跑通，一旦出现连接重连、多用户接入、首个连接断开后重建的情况，监听就会完全失效。
• WebSocket发送方法语法不匹配
  当前实现使用了Socket.IO风格的socket.send('feed', data)双参数写法，但项目集成的是原生ws库的WebSocketServer，ws库的socket发送方法仅接收单个序列化参数，即便watch事件正常触发，这里也会抛出参数异常，无法向前端推送数据。
• Watch游标生命周期管理错误
  目前Comments.watch(...)的初始化逻辑写在socket.onmessage回调内部，存在两个隐患：
  1. 建立连接时的消息监听绑定存在时序差，若前端发送的用户ID消息在后端完成onmessage绑定前就抵达，会直接被丢弃，导致watch游标从未被初始化。
  2. 每次收到消息都会新建一个watch游标，未对旧游标做销毁处理，游标静默失效后不会触发任何错误回调，直接导致后续所有变更事件收不到。
• 连接关闭事件绑定位置错误
  目前将onclose事件直接绑定在wss服务端实例上，而不是单个socket连接实例上，该事件永远不会被正确触发，既无法感知单个连接的断开，也无法在连接关闭时销毁对应的watch游标。

变更流不触发的环境层排查项 #

• 确认MongoDB实例满足变更流前置要求：Mongoose的watch能力完全依赖MongoDB副本集架构（单节点测试环境也需要配置为单节点副本集），若当前连接的是独立运行的非副本集mongod实例，变更流功能会完全禁用，change回调永远不会触发。
• 确认当前数据库连接的权限配置：连接使用的数据库账号需要具备changeStream、read权限，否则watch初始化会静默失败。
• 临时移除watch的match过滤条件，测试是否是过滤规则不匹配导致insert事件被全部拦截。

参考修复实现 #

调整后的socket服务端代码：

const { Comments } = require('../db/models');
const { ObjectID } = require('mongodb');

module.exports = function(wss) {
    // 替换once为on，支持全量连接接入
    wss.on("connection", function(socket){
        let commentWatcher = null;

        // 连接建立后立刻绑定消息监听，避免消息时序差
        socket.onmessage = async(event) => {
            const uid = event.data;
            // 销毁旧游标，避免重复监听和内存泄漏
            if (commentWatcher) {
                await commentWatcher.close();
            }

            const filter = [
                { $match: { operationType: 'insert' } }
            ];
            const options = { fullDocument: 'updateLookup' };

            commentWatcher = Comments.watch(filter, options)
                .on('change', changeData => {
                    // 单参数发送，序列化结构化消息
                    socket.send(JSON.stringify({
                        type: 'feed',
                        payload: changeData
                    }));
                })
                .on('error', err => {
                    console.error(new Date().toISOString(), 'Watch运行异常', err);
                });
        };

        // 绑定单个连接的关闭事件，主动销毁游标
        socket.onclose = async() => {
            if (commentWatcher) {
                await commentWatcher.close();
            }
            console.log("前端连接已断开");
        };
    });
};

排查验证步骤：

1. 在watch初始化逻辑处加打印日志，确认游标真的被创建
2. 手动向Comments集合插入测试文档，确认事件流可以被监听
3. 确认前端WebSocket的onmessage回调可以正常接收到序列化后的JSON字符串

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