batchedUpdate.js 2.2 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697
  1. const { ReadPreference, ObjectId } = require('mongodb')
  2. const { db, waitForDb } = require('../../app/src/infrastructure/mongodb')
  3. const BATCH_DESCENDING = process.env.BATCH_DESCENDING === 'true'
  4. const BATCH_SIZE = parseInt(process.env.BATCH_SIZE, 10) || 1000
  5. let BATCH_LAST_ID
  6. if (process.env.BATCH_LAST_ID) {
  7. BATCH_LAST_ID = ObjectId(process.env.BATCH_LAST_ID)
  8. }
  9. async function getNextBatch(collection, query, maxId, projection, options) {
  10. maxId = maxId || BATCH_LAST_ID
  11. if (maxId) {
  12. if (BATCH_DESCENDING) {
  13. query._id = { $lt: maxId }
  14. } else {
  15. query._id = { $gt: maxId }
  16. }
  17. }
  18. const entries = await collection
  19. .find(query, options)
  20. .project(projection)
  21. .sort({ _id: BATCH_DESCENDING ? -1 : 1 })
  22. .limit(BATCH_SIZE)
  23. .toArray()
  24. return entries
  25. }
  26. async function performUpdate(collection, nextBatch, update) {
  27. return collection.updateMany(
  28. { _id: { $in: nextBatch.map(entry => entry._id) } },
  29. update
  30. )
  31. }
  32. async function batchedUpdate(
  33. collectionName,
  34. query,
  35. update,
  36. projection,
  37. options
  38. ) {
  39. await waitForDb()
  40. const collection = db[collectionName]
  41. options = options || {}
  42. options.readPreference = ReadPreference.SECONDARY
  43. projection = projection || { _id: 1 }
  44. let nextBatch
  45. let updated = 0
  46. let maxId
  47. while (
  48. (nextBatch = await getNextBatch(
  49. collection,
  50. query,
  51. maxId,
  52. projection,
  53. options
  54. )).length
  55. ) {
  56. maxId = nextBatch[nextBatch.length - 1]._id
  57. updated += nextBatch.length
  58. console.log(
  59. `Running update on batch with ids ${JSON.stringify(
  60. nextBatch.map(entry => entry._id)
  61. )}`
  62. )
  63. if (typeof update === 'function') {
  64. await update(collection, nextBatch)
  65. } else {
  66. await performUpdate(collection, nextBatch, update)
  67. }
  68. console.error(`Completed batch ending ${maxId}`)
  69. }
  70. return updated
  71. }
  72. function batchedUpdateWithResultHandling(collection, query, update) {
  73. batchedUpdate(collection, query, update)
  74. .then(updated => {
  75. console.error({ updated })
  76. process.exit(0)
  77. })
  78. .catch(error => {
  79. console.error({ error })
  80. process.exit(1)
  81. })
  82. }
  83. module.exports = {
  84. getNextBatch,
  85. batchedUpdate,
  86. batchedUpdateWithResultHandling,
  87. }