变更流 Change Streams

// 传统轮询
setInterval(async () => {
  const docs = await db.posts.find({ updatedAt: { $gt: lastCheck } }).toArray()
  lastCheck = new Date()
  await syncToES(docs)
}, 5000)

       应用写入
          │
          ▼
   ┌─────────────┐
   │   primary   │
   │             │
   │  oplog ─────┼──── 复制给 secondary
   │    │        │
   └────┼────────┘
        │
        ▼  Change Streams 订阅 oplog（经过封装和过滤）
   ┌──────────────┐
   │  下游消费者   │  → 同步到 ES / 刷缓存 / 发通知 / 数据仓库
   └──────────────┘

// 集合级
const cs = db.posts.watch()
 
// 数据库级（该库所有集合）
const cs = db.watch()
 
// 部署级（所有库，需要 admin 权限）
const cs = db.getMongo().watch()

while (cs.hasNext()) {
  printjson(cs.next())
}

db.posts.insertOne({ _id: 500, title: "新帖子", authorId: "alice", status: "draft" })

{
  "_id": { "_data": "82665F1A2B000000012B022C0100296E5A100..." },
  "operationType": "insert",
  "clusterTime": Timestamp({ t: 1717495851, i: 1 }),
  "wallTime": ISODate("2024-06-04T09:30:51.234Z"),
  "ns": { "db": "community", "coll": "posts" },
  "documentKey": { "_id": 500 },
  "fullDocument": {
    "_id": 500, "title": "新帖子", "authorId": "alice", "status": "draft"
  }
}

db.posts.updateOne({ _id: 500 }, { $set: { status: "published" }, $inc: { "stats.views": 1 } })

{
  "_id": { "_data": "82665F1A2C..." },
  "operationType": "update",
  "clusterTime": Timestamp({ t: 1717495852, i: 1 }),
  "ns": { "db": "community", "coll": "posts" },
  "documentKey": { "_id": 500 },
  "updateDescription": {
    "updatedFields": { "status": "published", "stats.views": 1 },
    "removedFields": [],
    "truncatedArrays": []
  }
}

{
  "operationType": "delete",
  "ns": { "db": "community", "coll": "posts" },
  "documentKey": { "_id": 500 }
}

db.posts.watch([], { fullDocument: "updateLookup" })

db.runCommand({ collMod: "posts", changeStreamPreAndPostImages: { enabled: true } })
 
db.posts.watch([], {
  fullDocumentBeforeChange: "whenAvailable",
  fullDocument: "whenAvailable"
})

{
  "operationType": "update",
  "fullDocumentBeforeChange": { "_id": 500, "status": "draft",     "stats": { "views": 0 } },
  "fullDocument":             { "_id": 500, "status": "published", "stats": { "views": 1 } }
}

db.posts.watch([], {
  batchSize: 100,          // 每批最多返回多少个事件
  maxAwaitTimeMS: 1000     // 没有新事件时最多等多久（影响空闲时的响应延迟）
})

db.posts.watch([
  { $match: {
      operationType: { $in: ["insert", "update"] },
      "fullDocument.status": "published"
  } },
  { $project: {
      _id: 1,                        // resume token 必须保留！
      operationType: 1,
      documentKey: 1,
      "fullDocument.title": 1,
      "fullDocument.author": 1,
      "updateDescription.updatedFields": 1
  } }
], { fullDocument: "updateLookup" })

db.users.watch([
  { $match: {
      operationType: "update",
      $or: [
        { "updateDescription.updatedFields.username": { $exists: true } },
        { "updateDescription.updatedFields.profile.avatar": { $exists: true } }
      ]
  } }
])

let resumeToken = loadTokenFromStorage()     // 从 Redis/文件/集合里读
 
const options = { fullDocument: "updateLookup" }
if (resumeToken) options.resumeAfter = resumeToken
 
const cs = db.posts.watch([], options)
 
while (cs.hasNext()) {
  const change = cs.next()
 
  processChange(change)                       // 业务处理
 
  resumeToken = change._id
  saveTokenToStorage(resumeToken)             // 处理成功后再保存
}

先保存 token，再处理业务：
  处理失败 → token 已推进 → 事件丢失      → at-most-once
 
先处理业务，再保存 token：
  保存失败 → token 未推进 → 事件重放      → at-least-once  ✔ 推荐

// 从 10 分钟前开始重放
db.posts.watch([], {
  startAtOperationTime: Timestamp({ t: Math.floor(Date.now() / 1000) - 600, i: 1 })
})

// 查看 oplog 窗口有多长
db.getSiblingDB("local").oplog.rs.stats().maxSize / 1024 / 1024 / 1024   // GB
rs.printReplicationInfo()

configured oplog size:   51200MB
log length start to end: 86400secs (24hrs)
oplog first event time:  Mon Jun 03 2024 09:00:00
oplog last event time:   Tue Jun 04 2024 09:00:00
now:                     Tue Jun 04 2024 09:00:05

db.getSiblingDB("local").oplog.rs.find().sort({ $natural: -1 }).limit(1)

{
  "op": "u",
  "ns": "community.posts",
  "ui": UUID("..."),
  "o": { "$v": 2, "diff": { "u": { "status": "published" } } },
  "o2": { "_id": 500 },
  "ts": Timestamp({ t: 1717495852, i: 1 }),
  "t": NumberLong(1),
  "wall": ISODate("2024-06-04T09:30:52.001Z")
}

const cs = db.users.watch([
  { $match: {
      operationType: "update",
      "updateDescription.updatedFields.username": { $exists: true }
  } }
], { fullDocument: "updateLookup" })
 
while (cs.hasNext()) {
  const change = cs.next()
  const userId = change.documentKey._id
  const newName = change.fullDocument.username
 
  db.posts.updateMany(
    { "author._id": userId },
    { $set: { "author.username": newName } }
  )
  db.comments.updateMany(
    { "author._id": userId },
    { $set: { "author.username": newName } }
  )
 
  saveToken(change._id)
}

const cs = db.posts.watch([
  { $match: { "ns.coll": "posts" } }
], { fullDocument: "updateLookup" })
 
while (cs.hasNext()) {
  const c = cs.next()
  const id = c.documentKey._id.toString()
 
  switch (c.operationType) {
    case "insert":
    case "replace":
    case "update":
      if (c.fullDocument) {
        await es.index({ index: "posts", id, document: {
          title: c.fullDocument.title,
          body: c.fullDocument.body,
          tags: c.fullDocument.tags
        } })
      }
      break
    case "delete":
      await es.delete({ index: "posts", id }).catch(() => {})
      break
  }
  saveToken(c._id)
}

db.posts.watch([
  { $match: { operationType: { $in: ["update", "replace", "delete"] } } }
]).forEach(c => {
  redis.del(`post:${c.documentKey._id}`)
  saveToken(c._id)
})