overleaf/services/web/scripts/add_marketing_details_to_csv.mjs

#!/usr/bin/env node
/* eslint-disable camelcase */

import { scriptRunner } from './lib/ScriptRunner.mjs'
import fs from 'node:fs'
import minimist from 'minimist'
import {
  db,
  ObjectId,
  READ_PREFERENCE_SECONDARY,
} from '../app/src/infrastructure/mongodb.mjs'
// https://github.com/import-js/eslint-plugin-import/issues/1810
// eslint-disable-next-line import/no-unresolved
import * as csv from 'csv/sync'

function usage() {
  console.log(
    `
    This script enriches price and subscription data with user emails and first names, given a csv containing user IDs, outputs to /tmp/output.csv

    Usage:
    - Locally:
        docker compose exec web bash
        node scripts/add_marketing_details_to_csv.mjs [--input=<path>] [--output=<path>] [--batchSize=<number>]
    - On the server:
        rake run:pod[staging,web]
        node scripts/add_marketing_details_to_csv.mjs [--input=<path>] [--output=<path>] [--batchSize=<number>]
        exit
        kubectl cp web-standalone-prod-XXXXX:/tmp/output.csv ~/output.csv

    Options:
    --help                     Show this screen

    --input=<path>             Input file path (default: input.csv)

    --output=<path>            Output file path (default: /tmp/output.csv)

    --batchSize=<number>       Number of users to be fetched in one query
    `
  )
}

function parseArgs() {
  const result = minimist(process.argv.slice(2), {
    string: ['input', 'output'],
    bool: ['help'],
    number: ['batchSize'],
    default: {
      help: false,
      input: 'input.csv',
      output: '/tmp/output.csv',
      batchSize: 1000,
    },
  })

  if (result.help) {
    usage()
    process.exit(0)
  }

  return result
}

async function processBatch(trackProgress, idBatch) {
  const result = {}
  try {
    const cursor = db.users.find(
      {
        _id: { $in: idBatch },
      },
      {
        projection: {
          _id: 1,
          first_name: 1,
          email: 1,
        },
        readPreference: READ_PREFERENCE_SECONDARY,
      }
    )

    for await (const doc of cursor) {
      result[doc._id.toString()] = {
        email: doc.email,
        first_name: doc.first_name,
      }
    }
  } catch (err) {
    await trackProgress(`ERROR Processing batch: ${err}`)
  }
  return result
}

async function enrichRecords(trackProgress, records, users) {
  const result = []

  for (const record of records) {
    const user = users[record.user_id]
    let email = ''
    let first_name = ''
    if (user) {
      if (user.email === '' || user.first_name === '') {
        await trackProgress(`WARNING Incomplete data for: ${record.user_id}`)
      }
      email = user.email
      first_name = user.first_name
    } else {
      await trackProgress(`WARNING Didn't find: ${record.user_id}`)
    }
    result.push({
      ...record,
      email,
      first_name,
    })
  }

  return result
}

async function main(trackProgress) {
  const args = parseArgs()
  const input = fs.readFileSync(args.input, 'utf8')
  const records = csv.parse(input, { columns: true, skipEmptyLines: true })
  await trackProgress(`INFO Starting to process ${records.length} records`)

  let objectIdBatch = []
  let recordBatch = []
  const outputRecords = []
  for (const record of records) {
    try {
      objectIdBatch.push(new ObjectId(record.user_id))
      recordBatch.push(record)
    } catch (e) {
      await trackProgress(`ERROR Skipping invalid user ID: ${record.user_id}`)
      outputRecords.push({ ...record, email: '', first_name: '' })
    }
    if (objectIdBatch.length >= args.batchSize) {
      const users = await processBatch(trackProgress, objectIdBatch)
      const enriched = await enrichRecords(trackProgress, recordBatch, users)
      outputRecords.push(...enriched)

      objectIdBatch = []
      recordBatch = []
    }
  }

  if (objectIdBatch.length > 0) {
    const users = await processBatch(trackProgress, objectIdBatch)
    const enriched = await enrichRecords(trackProgress, recordBatch, users)
    outputRecords.push(...enriched)
  }

  const output = csv.stringify(outputRecords, { header: true })
  fs.writeFileSync(args.output, output)
  await trackProgress(
    `INFO Finished processing of ${outputRecords.length} records`
  )
}

try {
  await scriptRunner(main)
  process.exit(0)
} catch (error) {
  console.error(error)
  process.exit(1)
}