batchedUpdate.js 3.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136
  1. const { ReadPreference, ObjectId } = require('mongodb')
  2. const { db, waitForDb } = require('../../app/src/infrastructure/mongodb')
  3. let BATCH_DESCENDING
  4. let BATCH_SIZE
  5. let VERBOSE_LOGGING
  6. let BATCH_LAST_ID
  7. let BATCH_RANGE_END
  8. refreshGlobalOptionsForBatchedUpdate()
  9. function refreshGlobalOptionsForBatchedUpdate(options = {}) {
  10. options = Object.assign({}, options, process.env)
  11. BATCH_DESCENDING = options.BATCH_DESCENDING === 'true'
  12. BATCH_SIZE = parseInt(options.BATCH_SIZE, 10) || 1000
  13. VERBOSE_LOGGING = options.VERBOSE_LOGGING === 'true'
  14. if (options.BATCH_LAST_ID) {
  15. BATCH_LAST_ID = ObjectId(options.BATCH_LAST_ID)
  16. } else if (options.BATCH_RANGE_START) {
  17. BATCH_LAST_ID = ObjectId(options.BATCH_RANGE_START)
  18. }
  19. if (options.BATCH_RANGE_END) {
  20. BATCH_RANGE_END = ObjectId(options.BATCH_RANGE_END)
  21. }
  22. }
  23. async function getNextBatch(collection, query, maxId, projection, options) {
  24. const queryIdField = {}
  25. maxId = maxId || BATCH_LAST_ID
  26. if (maxId) {
  27. if (BATCH_DESCENDING) {
  28. queryIdField.$lt = maxId
  29. } else {
  30. queryIdField.$gt = maxId
  31. }
  32. }
  33. if (BATCH_RANGE_END) {
  34. if (BATCH_DESCENDING) {
  35. queryIdField.$gt = BATCH_RANGE_END
  36. } else {
  37. queryIdField.$lt = BATCH_RANGE_END
  38. }
  39. }
  40. if (queryIdField.$gt || queryIdField.$lt) {
  41. query._id = queryIdField
  42. }
  43. const entries = await collection
  44. .find(query, options)
  45. .project(projection)
  46. .sort({ _id: BATCH_DESCENDING ? -1 : 1 })
  47. .limit(BATCH_SIZE)
  48. .toArray()
  49. return entries
  50. }
  51. async function performUpdate(collection, nextBatch, update) {
  52. return collection.updateMany(
  53. { _id: { $in: nextBatch.map(entry => entry._id) } },
  54. update
  55. )
  56. }
  57. async function batchedUpdate(
  58. collectionName,
  59. query,
  60. update,
  61. projection,
  62. findOptions,
  63. batchedUpdateOptions
  64. ) {
  65. refreshGlobalOptionsForBatchedUpdate(batchedUpdateOptions)
  66. await waitForDb()
  67. const collection = db[collectionName]
  68. findOptions = findOptions || {}
  69. findOptions.readPreference = ReadPreference.SECONDARY
  70. projection = projection || { _id: 1 }
  71. let nextBatch
  72. let updated = 0
  73. let maxId
  74. while (
  75. (nextBatch = await getNextBatch(
  76. collection,
  77. query,
  78. maxId,
  79. projection,
  80. findOptions
  81. )).length
  82. ) {
  83. maxId = nextBatch[nextBatch.length - 1]._id
  84. updated += nextBatch.length
  85. if (VERBOSE_LOGGING) {
  86. console.log(
  87. `Running update on batch with ids ${JSON.stringify(
  88. nextBatch.map(entry => entry._id)
  89. )}`
  90. )
  91. } else {
  92. console.error(`Running update on batch ending ${maxId}`)
  93. }
  94. if (typeof update === 'function') {
  95. await update(collection, nextBatch)
  96. } else {
  97. await performUpdate(collection, nextBatch, update)
  98. }
  99. console.error(`Completed batch ending ${maxId}`)
  100. }
  101. return updated
  102. }
  103. function batchedUpdateWithResultHandling(
  104. collection,
  105. query,
  106. update,
  107. projection,
  108. options
  109. ) {
  110. batchedUpdate(collection, query, update, projection, options)
  111. .then(updated => {
  112. console.error({ updated })
  113. process.exit(0)
  114. })
  115. .catch(error => {
  116. console.error({ error })
  117. process.exit(1)
  118. })
  119. }
  120. module.exports = {
  121. getNextBatch,
  122. batchedUpdate,
  123. batchedUpdateWithResultHandling,
  124. }