batchedUpdate.js 1.9 KB

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