mongo_update_stream.js

import { Transform, Writable } from 'stream';
import { pipeline } from 'stream/promises';

export async function updateWithStream(
    model, filters = {}, 
    transformFunc, saveFunc, 
    options = {
        batchSize: 10
    },
) {
    const readStream = model.find(filters).cursor({
        batchSize: options.batchSize || 10,
    });

    const transformStream = new Transform({
        objectMode            : true,
        readableHighWaterMark : options.batchSize || 10,
        writableHighWaterMark : options.batchSize || 10,
        transform             : async (doc, encoding, callback) => {
            try {
                callback(null, await transformFunc(doc, encoding));
            }
            catch (e) {
                return callback(e);
            }
        },
    });

    const saveStream = new Writable({
        objectMode    : true,
        highWaterMark : options.batchSize || 10,
        write         : async (doc, encoding, callback) => {
            try {
                await saveFunc(doc);
                callback();
            }
            catch (e) {
                return callback(e);
            }
        },
    });

    await pipeline(readStream, transformStream, saveStream);
}