Change Streams и реактивность

В прошлом уроке данные разъехались по шардам. Сегодня задача другая. Покупатель отменил заказ, и складу нужно об этом узнать: вернуть книгу на полку, а бухгалтерии сделать возврат денег. Можно раз в секунду спрашивать базу «что поменялось?», но есть способ лучше: база сама сообщает о каждом изменении.

Звонок вместо проверки почтового ящика

Ждёшь важное письмо и каждые пять минут спускаешься к ящику. Чаще всего он пустой, а если письмо пришло через минуту после проверки, ты узнаешь о нём только через четыре. Это опрос: много пустых походов и задержка.

Курьер, который звонит в дверь, устроен иначе. Ты ничего не проверяешь, а узнаёшь о посылке в момент доставки. Потоки изменений в MongoDB это такой курьер.

Откуда берутся события

В уроке про replica set мы видели oplog: журнал, куда первичный узел пишет каждое изменение. Поток изменений читает этот журнал и отдаёт приложению события в удобном виде. Отсюда главное ограничение: потоку нужен oplog, а значит replica set или шардированный кластер. На одиночном сервере команда падает:

Uncaught MongoServerError[Location40573]: The $changeStream stage is only supported on replica sets or mongos

Дальше все примеры сняты на replica set из прошлого урока с загруженной учебной базой магазина.

Первый поток

Поток открывается методом watch() у коллекции. Получим курсор, сделаем три изменения и прочитаем события по одному через tryNext():

const cs = db.orders.watch()
db.orders.insertOne({ _id: 10, customer_id: 4, created_at: ISODate('2026-10-10T09:00:00Z'), status: 'new', items: [] })
db.orders.updateOne({ _id: 10 }, { $set: { status: 'paid' } })
db.orders.deleteOne({ _id: 10 })
cs.tryNext()

Первое событие, вставка:

{
  _id: {
    _data: '826AC9BA6F000000012B042C0100296E5A1004C695FAA23E4D47F287672B6DC85E8C4C463C6F7065726174696F6E54797065003C696E736572740046646F63756D656E744B657900461E5F6964002B14000004'
  },
  operationType: 'insert',
  clusterTime: Timestamp({ t: 1791605359, i: 1 }),
  wallTime: ISODate('2026-10-10T04:09:19.322Z'),
  fullDocument: {
    _id: 10,
    customer_id: 4,
    created_at: ISODate('2026-10-10T09:00:00.000Z'),
    status: 'new',
    items: []
  },
  ns: { db: 'test', coll: 'orders' },
  documentKey: { _id: 10 }
}

Разберём поля события:

• operationType: что случилось. Бывают insert, update, replace, delete и служебные вроде drop.
• ns: в какой базе и коллекции.
• documentKey: какой документ, обычно его _id.
• fullDocument: сам документ. У вставки он есть всегда.
• _id: токен события, о нём ниже.

Второе событие, обновление, выглядит иначе (токен сокращён):

{
  _id: { _data: '826AC9BA6F00000002...' },
  operationType: 'update',
  clusterTime: Timestamp({ t: 1791605359, i: 2 }),
  wallTime: ISODate('2026-10-10T04:09:19.352Z'),
  ns: { db: 'test', coll: 'orders' },
  documentKey: { _id: 10 },
  updateDescription: {
    updatedFields: { status: 'paid' },
    removedFields: [],
    truncatedArrays: []
  }
}

Документа целиком здесь нет, только updateDescription: какие поля поменялись и какие удалены. У удаления нет ни документа, ни описания, только documentKey: { _id: 10 }. Когда события кончились, tryNext() возвращает null. Событию нужно мгновение, чтобы дойти до потока: если tryNext() сразу после изменения вернул null, повтори вызов через секунду.