Function runImportEvents

src/import-events.ts:203–288

(
  args: string[] = process.argv.slice(2),
  runOptions: RunImportOptions = {},
)

}

export const runImportEvents = async (
  args: string[] = process.argv.slice(2),
  runOptions: RunImportOptions = {},
): Promise<number> => {
  const options = parseCliArgs(args)

  if (options.showHelp) {
    printUsage()
    return 0
  }

  const inputFile = await ensureValidInputFile(options.filePath)

  if (inputFile.compressionFormat && inputFile.format === 'json') {
    throw new Error('Compressed JSON array import is not supported. Use .json (uncompressed) or .jsonl.gz/.jsonl.xz.')
  }

  const dbClient = getMasterDbClient()
  const eventRepository = new EventRepository(dbClient, dbClient)
  const importer = new EventImportService(createEventBatchPersister(eventRepository))

  let loggedErrors = 0
  let suppressedErrors = 0

  const onLineError = ({ lineNumber, reason }: EventImportLineError) => {
    if (loggedErrors < MAX_ERROR_LOGS) {
      console.warn(`[line ${lineNumber}] ${reason}`)
      loggedErrors += 1
      return
    }

    suppressedErrors += 1
  }

  const onProgress = (stats: EventImportStats) => {
    console.log(formatProgress(stats))
  }

  const startedAt = Date.now()

  try {
    if (inputFile.compressionFormat) {
      console.log(`Detected ${inputFile.compressionFormat} compression. Decompressing on-the-fly...`)
    }

    const stats =
      inputFile.format === 'json'
        ? await importer.importFromJsonArray(inputFile.absolutePath, {
            batchSize: options.batchSize,
            onLineError,
            onProgress,
          })
        : await importer.importFromReadable(createImportStream(inputFile), {
            batchSize: options.batchSize,
            onLineError,
            onProgress,
          })