close-change-stream

// @ts-check
import { logger } from './log';

/**
 *
 * @param {import('mongodb').ChangeStream} changeStream
 * @param {number} [timeInMs] default value 60000
 * @returns
 */
function closeChangeStream(changeStream, timeInMs = 60000) {
    return new Promise((resolve) => {
        setTimeout(() => {
            logger('info', 'db', {
                message: 'Closing the change stream',
            });
            changeStream.close();
            resolve();
        }, timeInMs);
    });
}

export { closeChangeStream };


//////////////////////////// in model /////////////////////////////////

const userChangeStream = UserModel.watch();

userChangeStream.on('change', async (doc) => {
    if (doc.operationType === 'update') {
        // async works
    }

    await closeChangeStream(userChangeStream).catch(console.error);
});