batchedUpdate.js 1.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778
  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. maxId = maxId || BATCH_LAST_ID
  10. if (maxId) {
  11. query._id = { $gt: maxId }
  12. }
  13. const entries = await collection
  14. .find(query, { readPreference: ReadPreference.SECONDARY })
  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(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
  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(
  41. `Running update on batch with ids ${JSON.stringify(
  42. nextBatch.map(entry => entry._id)
  43. )}`
  44. )
  45. if (typeof update === 'function') {
  46. await update(collection, nextBatch)
  47. } else {
  48. await performUpdate(collection, nextBatch, update)
  49. }
  50. console.error(`Completed batch ending ${maxId}`)
  51. }
  52. return updated
  53. }
  54. function batchedUpdateWithResultHandling(collection, query, update) {
  55. batchedUpdate(collection, query, update)
  56. .then(updated => {
  57. console.error({ updated })
  58. process.exit(0)
  59. })
  60. .catch(error => {
  61. console.error({ error })
  62. process.exit(1)
  63. })
  64. }
  65. module.exports = {
  66. getNextBatch,
  67. batchedUpdate,
  68. batchedUpdateWithResultHandling,
  69. }