add_salesforce_data_to_subscriptions.mjs 5.3 KB

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