batchedUpdate.js 1.8 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273
  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)
  15. .project(projection)
  16. .sort({ _id: 1 })
  17. .limit(BATCH_SIZE)
  18. .setReadPreference(ReadPreference.SECONDARY)
  19. .toArray()
  20. return entries
  21. }
  22. async function performUpdate(collection, nextBatch, update) {
  23. return collection.updateMany(
  24. { _id: { $in: nextBatch.map(entry => entry._id) } },
  25. update
  26. )
  27. }
  28. async function batchedUpdate(collectionName, query, update, projection) {
  29. await waitForDb()
  30. const collection = db[collectionName]
  31. projection = projection || { _id: 1 }
  32. let nextBatch
  33. let updated = 0
  34. let maxId
  35. while (
  36. (nextBatch = await getNextBatch(collection, query, maxId, projection))
  37. .length
  38. ) {
  39. maxId = nextBatch[nextBatch.length - 1]._id
  40. updated += nextBatch.length
  41. console.log(JSON.stringify(nextBatch))
  42. if (typeof update === 'function') {
  43. await update(collection, nextBatch)
  44. } else {
  45. await performUpdate(collection, nextBatch, update)
  46. }
  47. }
  48. return updated
  49. }
  50. function batchedUpdateWithResultHandling(collection, query, update) {
  51. batchedUpdate(collection, query, update)
  52. .then(updated => {
  53. console.error({ updated })
  54. process.exit(0)
  55. })
  56. .catch(error => {
  57. console.error({ error })
  58. process.exit(1)
  59. })
  60. }
  61. module.exports = {
  62. getNextBatch,
  63. batchedUpdate,
  64. batchedUpdateWithResultHandling
  65. }