batchedUpdate.js 2.0 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192
  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, options) {
  9. maxId = maxId || BATCH_LAST_ID
  10. if (maxId) {
  11. query._id = { $gt: maxId }
  12. }
  13. const entries = await collection
  14. .find(query, options)
  15. .project(projection)
  16. .sort({ _id: 1 })
  17. .limit(BATCH_SIZE)
  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(
  28. collectionName,
  29. query,
  30. update,
  31. projection,
  32. options
  33. ) {
  34. await waitForDb()
  35. const collection = db[collectionName]
  36. options = options || {}
  37. options.readPreference = ReadPreference.SECONDARY
  38. projection = projection || { _id: 1 }
  39. let nextBatch
  40. let updated = 0
  41. let maxId
  42. while (
  43. (nextBatch = await getNextBatch(
  44. collection,
  45. query,
  46. maxId,
  47. projection,
  48. options
  49. )).length
  50. ) {
  51. maxId = nextBatch[nextBatch.length - 1]._id
  52. updated += nextBatch.length
  53. console.log(
  54. `Running update on batch with ids ${JSON.stringify(
  55. nextBatch.map(entry => entry._id)
  56. )}`
  57. )
  58. if (typeof update === 'function') {
  59. await update(collection, nextBatch)
  60. } else {
  61. await performUpdate(collection, nextBatch, update)
  62. }
  63. console.error(`Completed batch ending ${maxId}`)
  64. }
  65. return updated
  66. }
  67. function batchedUpdateWithResultHandling(collection, query, update) {
  68. batchedUpdate(collection, query, update)
  69. .then(updated => {
  70. console.error({ updated })
  71. process.exit(0)
  72. })
  73. .catch(error => {
  74. console.error({ error })
  75. process.exit(1)
  76. })
  77. }
  78. module.exports = {
  79. getNextBatch,
  80. batchedUpdate,
  81. batchedUpdateWithResultHandling,
  82. }