migrate_audit_logs.js 4.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145
  1. const { batchedUpdate } = require('./helpers/batchedUpdate')
  2. const { promiseMapWithLimit, promisify } = require('../app/src/util/promises')
  3. const { db, ObjectId, waitForDb } = require('../app/src/infrastructure/mongodb')
  4. const sleep = promisify(setTimeout)
  5. const _ = require('lodash')
  6. async function main(options) {
  7. if (!options) {
  8. options = {}
  9. }
  10. _.defaults(options, {
  11. dryRun: process.env.DRY_RUN !== 'false',
  12. projectId: process.env.PROJECT_ID,
  13. userId: process.env.USER_ID,
  14. writeConcurrency: parseInt(process.env.WRITE_CONCURRENCY, 10) || 10,
  15. letUserDoubleCheckInputsFor: parseInt(
  16. process.env.LET_USER_DOUBLE_CHECK_INPUTS_FOR || 10 * 1000,
  17. 10
  18. ),
  19. })
  20. await letUserDoubleCheckInputs(options)
  21. if (options.projectId) {
  22. console.log('migrating projectId=' + options.projectId)
  23. const project = await db.projects.findOne(
  24. { _id: ObjectId(options.projectId) },
  25. { _id: 1, auditLog: 1 }
  26. )
  27. await processProject(project, options)
  28. } else if (options.userId) {
  29. console.log('migrating userId=' + options.userId)
  30. const user = await db.users.findOne(
  31. { _id: ObjectId(options.userId) },
  32. { _id: 1, auditLog: 1 }
  33. )
  34. await processUser(user, options)
  35. } else {
  36. await batchedUpdate(
  37. 'users',
  38. { auditLog: { $exists: true } },
  39. async (_, users) => {
  40. await processUsersBatch(users, options)
  41. },
  42. { _id: 1, auditLog: 1 }
  43. )
  44. // most projects are processed after its owner has been processed, but only those
  45. // users with an existing `auditLog` have been taken into consideration, leaving
  46. // some projects orphan. This batched update processes all remaining projects.
  47. await batchedUpdate(
  48. 'projects',
  49. { auditLog: { $exists: true } },
  50. async (x, projects) => {
  51. await processProjectsBatch(projects, options)
  52. },
  53. { _id: 1, auditLog: 1 }
  54. )
  55. }
  56. }
  57. async function processUsersBatch(users, options) {
  58. await promiseMapWithLimit(options.writeConcurrency, users, async user => {
  59. await processUser(user, options)
  60. })
  61. }
  62. async function processUser(user, options) {
  63. const entries = user.auditLog.map(log => ({ ...log, userId: user._id }))
  64. if (!options.dryRun && entries?.length > 0) {
  65. await db.userAuditLogEntries.insertMany(entries)
  66. }
  67. if (!options.dryRun) {
  68. await db.users.updateOne({ _id: user._id }, { $unset: { auditLog: 1 } })
  69. }
  70. const projects = await db.projects.find(
  71. { owner_ref: user._id, auditLog: { $exists: true } },
  72. { _id: 1, auditLog: 1 }
  73. )
  74. projects.forEach(project => processProject(project, options))
  75. }
  76. async function processProjectsBatch(projects, options) {
  77. await promiseMapWithLimit(
  78. options.writeConcurrency,
  79. projects,
  80. async project => {
  81. await processProject(project, options)
  82. }
  83. )
  84. }
  85. async function processProject(project, options) {
  86. const entries = project.auditLog.map(log => ({
  87. ...log,
  88. projectId: project._id,
  89. }))
  90. if (!options.dryRun && entries?.length > 0) {
  91. await db.projectAuditLogEntries.insertMany(entries)
  92. }
  93. if (!options.dryRun) {
  94. await db.projects.updateOne(
  95. { _id: project._id },
  96. { $unset: { auditLog: 1 } }
  97. )
  98. }
  99. }
  100. async function letUserDoubleCheckInputs(options) {
  101. const allOptions = {
  102. ...options,
  103. // batchedUpdate() environment variables
  104. BATCH_DESCENDING: process.env.BATCH_DESCENDING,
  105. BATCH_SIZE: process.env.BATCH_SIZE,
  106. VERBOSE_LOGGING: process.env.VERBOSE_LOGGING,
  107. BATCH_LAST_ID: process.env.BATCH_LAST_ID,
  108. BATCH_RANGE_END: process.env.BATCH_RANGE_END,
  109. }
  110. console.error('Options:', JSON.stringify(allOptions, null, 2))
  111. console.error(
  112. 'Waiting for you to double check inputs for',
  113. options.letUserDoubleCheckInputsFor,
  114. 'ms'
  115. )
  116. await sleep(options.letUserDoubleCheckInputsFor)
  117. }
  118. module.exports = main
  119. if (require.main === module) {
  120. waitForDb()
  121. .then(main)
  122. .then(() => {
  123. console.log('Done.')
  124. process.exit(0)
  125. })
  126. .catch(error => {
  127. console.error({ error })
  128. process.exit(1)
  129. })
  130. }