batchedUpdate.js 1.4 KB

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