如何通过MongoDB变更流监听指定字段，实现posts变更实时推送前端

我在MongoDB数据库中有一个名为users的集合，每个用户文档都包含一个名为posts的字段。我希望在用户新增帖子或更新已有帖子时，实时获取该posts字段的变更数据，再通过socket服务端将更新后的值返回给前端。

单个用户文档的结构如下：

[  
    {
        "_id" : "613fb31a52ef63541d036fcd",
        "posts": ["613fb31a52ef63541d036fdf", "613fb31a52ef63541d036ff1"],
        "mobiles" : [ "01751150412"],
        "languages" : [ "English", "Bangla"],
     }
]

最开始我尝试使用代码 const userChangeStream = connection.collection("users").watch() 实现监听，但该写法会返回users集合中所有甚至未发生变更的字段，不符合我的需求。

之后我修改代码为：

const filter = 
 [
   {
      $match: 
      {
        $and: 
        [
          { "updateDescription.updatedFields.posts": { $exists: true } },
          { operationType: "update" }
        ]
      }
   }
]

const options = { fullDocument: 'updateLookup' };

const postsChangeStream =    connection.collection("users").watch(filter,options);

postsChangeStream.on("change", (change) => {
  console.log('change on', change);
});

但修改代码后，我在对posts字段新增内容时，监听不到任何响应，请问该如何解决该问题？

你当前的匹配规则仅能覆盖直接替换整个posts字段的场景。当你使用$push、$addToSet、$pull这类数组操作符修改posts元素时，MongoDB不会将整个posts字段放到updateDescription.updatedFields中，而是会生成带数组下标的键（例如posts.2代表posts数组的第三个元素），因此你的匹配条件无法命中这类变更。

修改变更流的过滤规则，同时覆盖全量替换posts和修改posts数组元素两类场景，示例代码如下：

const filter = [
  {
    $match: {
      // 覆盖更新、替换文档两类场景，需要新增用户带posts的通知可新增"insert"到数组中
      operationType: { $in: ["update", "replace"] },
      $or: [
        // 场景1：直接替换整个posts字段
        { "updateDescription.updatedFields.posts": { $exists: true } },
        // 场景2：修改posts数组内的元素
        {
          $expr: {
            $gt: [
              {
                $size: {
                  $filter: {
                    input: { $objectToArray: "$updateDescription.updatedFields" },
                    cond: { $regexMatch: { input: "$$this.k", regex: /^posts\./ } }
                  }
                }
              },
              0
            ]
          }
        }
      ]
    }
  }
]

const options = { fullDocument: 'updateLookup' };
const postsChangeStream = connection.collection("users").watch(filter, options);

postsChangeStream.on("change", (change) => {
  // 获取最新的posts数组
  const updatedPosts = change.fullDocument.posts;
  // 获取对应用户ID
  const userId = change.fullDocument._id;
  // 此处添加你的socket推送逻辑，将posts数据推送给对应用户的前端
  // io.to(`user:${userId}`).emit('posts:update', updatedPosts);
});

• 变更流仅支持MongoDB副本集、分片集群部署模式，单节点MongoDB无法使用该功能
• 该规则会过滤所有不涉及posts字段的变更，符合你的业务需求
• 若更新操作同时修改了posts和其他字段，该规则依然会正确触发

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