add_salesforce_data_to_subscriptions.mjs 5.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207
  1. import fs from 'node:fs'
  2. import minimist from 'minimist'
  3. import { parse } from 'csv'
  4. import Stream from 'node:stream/promises'
  5. import { ObjectId } from '../app/src/infrastructure/mongodb.js'
  6. import { Subscription } from '../app/src/models/Subscription.js'
  7. import { scriptRunner } from './lib/ScriptRunner.mjs'
  8. function usage() {
  9. console.log(
  10. 'Usage: node add_salesforce_data_to_subscriptions.mjs -f <filename> [options]'
  11. )
  12. console.log(
  13. 'Updates the subscriptions collection with external IDs for determining the Salesforce account that goes with the subscription. The file should be a CSV and have columns account_id, v1_id and subscription_id. The account_id column is the Salesforce account ID, the v1_id column is the V1 account ID, and the subscription_id column is the subscription ID.'
  14. )
  15. console.log('Options:')
  16. console.log(
  17. ' --commit, -c Commit changes to the database'
  18. )
  19. console.log(
  20. ' --emptyFieldValue <value> The value to treat as an empty field (default: NA)'
  21. )
  22. console.log(
  23. ' -f, --filename <filename> The path to the file to read data from'
  24. )
  25. console.log(' -h, --help Show this help message')
  26. console.log(' -v, --verbose Produces more detailed logs')
  27. process.exit(0)
  28. }
  29. const { commit, emptyFieldValue, filename, help, verbose } = minimist(
  30. process.argv.slice(2),
  31. {
  32. string: ['emptyFieldValue', 'filename'],
  33. boolean: ['commit', 'help', 'verbose'],
  34. alias: {
  35. commit: 'c',
  36. filename: 'f',
  37. help: 'h',
  38. verbose: 'v',
  39. },
  40. default: {
  41. commit: false,
  42. emptyFieldValue: 'NA',
  43. help: false,
  44. verbose: false,
  45. },
  46. }
  47. )
  48. const SUBSCRIPTION_ID_FIELD = 'subscription_id'
  49. const SALESFORCE_ID_FIELD = 'account_id'
  50. const V1_ID_FIELD = 'v1_id'
  51. if (help) {
  52. usage()
  53. process.exit(0)
  54. }
  55. if (!filename) {
  56. console.error('No filename provided')
  57. usage()
  58. process.exit(1)
  59. }
  60. const stats = {
  61. totalRows: 0,
  62. subscriptionIDMissing: 0,
  63. usedV1ID: 0,
  64. usedSalesforceID: 0,
  65. processedRows: 0,
  66. db: {
  67. errors: 0,
  68. matched: 0,
  69. updateAttempted: 0,
  70. updated: 0,
  71. },
  72. }
  73. function generateStats() {
  74. return `Stats:
  75. Total rows: ${stats.totalRows}
  76. Processed rows: ${stats.processedRows}
  77. Skipped (no subscription ID): ${stats.subscriptionIDMissing}
  78. Used V1 ID: ${stats.usedV1ID}
  79. Used Salesforce ID: ${stats.usedSalesforceID}${
  80. commit
  81. ? `
  82. Database operations:
  83. Errors: ${stats.db.errors}
  84. Matched: ${stats.db.matched}
  85. Updated: ${stats.db.updated}
  86. Update attempted: ${stats.db.updateAttempted}`
  87. : ''
  88. }`
  89. }
  90. function pickRelevantColumns(row) {
  91. const newRow = {
  92. salesforceId: row[SALESFORCE_ID_FIELD],
  93. }
  94. if (row[V1_ID_FIELD] && row[V1_ID_FIELD] !== emptyFieldValue) {
  95. newRow.v1Id = row[V1_ID_FIELD]
  96. }
  97. if (
  98. row[SUBSCRIPTION_ID_FIELD] &&
  99. row[SUBSCRIPTION_ID_FIELD] !== emptyFieldValue
  100. ) {
  101. newRow.subscriptionId = row[SUBSCRIPTION_ID_FIELD]
  102. }
  103. return newRow
  104. }
  105. async function processRows(rows) {
  106. for await (const row of rows) {
  107. const { v1Id, salesforceId, subscriptionId } = row
  108. const update = {}
  109. if (v1Id) {
  110. stats.usedV1ID++
  111. update.v1_id = v1Id
  112. } else {
  113. stats.usedSalesforceID++
  114. update.salesforce_id = salesforceId
  115. }
  116. // Useful for logging later.
  117. const updateString = Object.entries(update).flatMap(([k, v]) => `${k}=${v}`)
  118. if (commit) {
  119. try {
  120. const result = await Subscription.updateOne(
  121. { _id: new ObjectId(subscriptionId) },
  122. update,
  123. { upsert: false }
  124. )
  125. if (result.matchedCount) {
  126. stats.db.matched++
  127. }
  128. if (result.modifiedCount) {
  129. stats.db.updated++
  130. if (verbose) {
  131. console.log(
  132. `Updated subscription ${subscriptionId} to set ${updateString}`
  133. )
  134. }
  135. }
  136. } catch (error) {
  137. stats.db.errors++
  138. if (verbose) {
  139. console.error(
  140. `Error updating subscription ${subscriptionId}: ${error}`
  141. )
  142. }
  143. } finally {
  144. stats.db.updateAttempted++
  145. }
  146. } else if (verbose) {
  147. console.log(`Would set ${updateString} on subscription ${subscriptionId}`)
  148. }
  149. }
  150. }
  151. async function main(trackProgress) {
  152. await Stream.pipeline(
  153. fs.createReadStream(filename),
  154. parse({
  155. columns: true,
  156. cast: function (value, context) {
  157. if (context.column === V1_ID_FIELD && value !== emptyFieldValue) {
  158. return parseInt(value)
  159. }
  160. return value
  161. },
  162. on_record: function (record, context) {
  163. stats.totalRows++
  164. const row = pickRelevantColumns(record)
  165. // Cannot process records without a Subscription ID
  166. if (!row.subscriptionId) {
  167. if (verbose) {
  168. console.log(
  169. `No subscription id found for ${row.salesforceId}, skipping...`
  170. )
  171. }
  172. stats.subscriptionIDMissing++
  173. return null
  174. }
  175. stats.processedRows++
  176. return row
  177. },
  178. }),
  179. processRows
  180. )
  181. await trackProgress(generateStats())
  182. }
  183. if (!commit) {
  184. console.log('Dry run')
  185. } else {
  186. console.log('Committing changes to the database')
  187. }
  188. await scriptRunner(main)
  189. process.exit()