batchedUpdate.js 1.8 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071
  1. const { ReadPreference, ObjectId } = require('mongodb')
  2. const { db, waitForDb } = require('../../app/src/infrastructure/mongodb')
  3. const BATCH_SIZE = parseInt(process.env.BATCH_SIZE, 10) || 1000
  4. let BATCH_LAST_ID
  5. if (process.env.BATCH_LAST_ID) {
  6. BATCH_LAST_ID = ObjectId(process.env.BATCH_LAST_ID)
  7. }
  8. async function getNextBatch(collection, query, maxId, projection) {
  9. if (maxId) {
  10. query['_id'] = { $gt: maxId }
  11. }
  12. const entries = await collection
  13. .find(query)
  14. .project(projection)
  15. .sort({ _id: 1 })
  16. .limit(BATCH_SIZE)
  17. .setReadPreference(ReadPreference.SECONDARY)
  18. .toArray()
  19. return entries
  20. }
  21. async function performUpdate(collection, nextBatch, update) {
  22. return collection.updateMany(
  23. { _id: { $in: nextBatch.map(entry => entry._id) } },
  24. update
  25. )
  26. }
  27. async function batchedUpdate(collectionName, query, update, projection) {
  28. await waitForDb()
  29. const collection = db[collectionName]
  30. projection = projection || { _id: 1 }
  31. let nextBatch
  32. let updated = 0
  33. let maxId = BATCH_LAST_ID
  34. while (
  35. (nextBatch = await getNextBatch(collection, query, maxId, projection))
  36. .length
  37. ) {
  38. maxId = nextBatch[nextBatch.length - 1]._id
  39. updated += nextBatch.length
  40. console.log(JSON.stringify(nextBatch))
  41. if (typeof update === 'function') {
  42. await update(collection, nextBatch)
  43. } else {
  44. await performUpdate(collection, nextBatch, update)
  45. }
  46. }
  47. return updated
  48. }
  49. function batchedUpdateWithResultHandling(collection, query, update) {
  50. batchedUpdate(collection, query, update)
  51. .then(updated => {
  52. console.error({ updated })
  53. process.exit(0)
  54. })
  55. .catch(error => {
  56. console.error({ error })
  57. process.exit(1)
  58. })
  59. }
  60. module.exports = {
  61. batchedUpdate,
  62. batchedUpdateWithResultHandling
  63. }