Concurrently transform a MongoDB cursor into a stream of bulk MongoDB updates.

each_async_update.js

// Concurrently transform a MongoDB cursor into a stream of bulk MongoDB updates.
//
// Usage:
//
// const users         = db.collection('users');
// const usersToUpdate = users.find({ email: null });
//
// await eachAsyncUpdate(users, usersToUpdate, async function(user) {
//   const userEmail = await findUserEmail(user);
//
//   return {
//     find:      { _id: user._id },
//     updateOne: { $set: { email: userEmail } }
//   };
// });
//

const highland = require('highland');

async function eachAsyncUpdate(collection, query, updateFn) {
  await highland(query)
    .map(updateFn)
    .map(highland)
    .mergeWithLimit(100)
    .compact()
    .batch(1000)
    .map(bulkUpdateBatch)
    .map(highland)
    .series()
    .compact()
    .toPromise(Promise);


  async function bulkUpdateBatch(operations) {
    const bulk = collection.initializeUnorderedBulkOp();

    for (const operation of operations) {
      bulk.find(operation.find)
        .updateOne(operation.updateOne);
    }

    await bulk.execute();
  }
}