batchedUpdate.js 2.2 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677
  1. const { promisify } = require('util')
  2. const { ReadPreference, ObjectId } = require('mongodb')
  3. const { getNativeDb } = require('../../app/src/infrastructure/Mongoose')
  4. const BATCH_SIZE = parseInt(process.env.BATCH_SIZE, 10) || 1000
  5. let BATCH_LAST_ID
  6. if (process.env.BATCH_LAST_ID) {
  7. BATCH_LAST_ID = ObjectId(process.env.BATCH_LAST_ID)
  8. }
  9. async function getNextBatch(collection, query, maxId, projection) {
  10. if (maxId) {
  11. query['_id'] = { $gt: maxId }
  12. }
  13. const entries = await collection
  14. .find(query, projection)
  15. .sort({ _id: 1 })
  16. .limit(BATCH_SIZE)
  17. .setReadPreference(ReadPreference.SECONDARY)
  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. // Apparently the mongo driver returns the connection too early.
  29. // Some secondary connections are not ready as it returns, leading to
  30. // failing cursor actions with a readPreference set to 'secondary'.
  31. // TODO(das7pad): revisit/remove this delay after the mongo-driver update.
  32. await Promise.all([getNativeDb(), promisify(setTimeout)(10 * 1000)])
  33. const db = await getNativeDb()
  34. const collection = db.collection(collectionName)
  35. projection = projection || { _id: 1 }
  36. let nextBatch
  37. let updated = 0
  38. let maxId = BATCH_LAST_ID
  39. while (
  40. (nextBatch = await getNextBatch(collection, query, maxId, projection))
  41. .length
  42. ) {
  43. maxId = nextBatch[nextBatch.length - 1]._id
  44. updated += nextBatch.length
  45. console.log(JSON.stringify(nextBatch))
  46. if (typeof update === 'function') {
  47. await update(collection, nextBatch)
  48. } else {
  49. await performUpdate(collection, nextBatch, update)
  50. }
  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. batchedUpdate,
  67. batchedUpdateWithResultHandling
  68. }