batchedUpdate.js 1.8 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061
  1. const { promisify } = require('util')
  2. const { ReadPreference } = require('mongodb')
  3. const { getNativeDb } = require('../../app/src/infrastructure/Mongoose')
  4. const BATCH_SIZE = parseInt(process.env.BATCH_SIZE, 10) || 1000
  5. async function getNextBatch(collection, query, maxId) {
  6. if (maxId) {
  7. query['_id'] = { $gt: maxId }
  8. }
  9. const entries = await collection
  10. .find(query, { _id: 1 })
  11. .sort({ _id: 1 })
  12. .limit(BATCH_SIZE)
  13. .setReadPreference(ReadPreference.SECONDARY)
  14. .toArray()
  15. return entries.map(entry => entry._id)
  16. }
  17. async function performUpdate(collection, nextBatch, update) {
  18. return collection.updateMany({ _id: { $in: nextBatch } }, update)
  19. }
  20. async function batchedUpdate(collectionName, query, update) {
  21. // Apparently the mongo driver returns the connection too early.
  22. // Some secondary connections are not ready as it returns, leading to
  23. // failing cursor actions with a readPreference set to 'secondary'.
  24. // TODO(das7pad): revisit/remove this delay after the mongo-driver update.
  25. await Promise.all([getNativeDb(), promisify(setTimeout)(10 * 1000)])
  26. const db = await getNativeDb()
  27. const collection = db.collection(collectionName)
  28. let nextBatch
  29. let updated = 0
  30. let maxId
  31. while ((nextBatch = await getNextBatch(collection, query, maxId)).length) {
  32. maxId = nextBatch[nextBatch.length - 1]
  33. updated += nextBatch.length
  34. console.log(JSON.stringify(nextBatch))
  35. await performUpdate(collection, nextBatch, update)
  36. }
  37. return updated
  38. }
  39. function batchedUpdateWithResultHandling(collection, query, update) {
  40. batchedUpdate(collection, query, update)
  41. .then(updated => {
  42. console.error({ updated })
  43. process.exit(0)
  44. })
  45. .catch(error => {
  46. console.error({ error })
  47. process.exit(1)
  48. })
  49. }
  50. module.exports = {
  51. batchedUpdate,
  52. batchedUpdateWithResultHandling
  53. }