feat: Import .jsonl file with events into the events table (#414)

* feat: add event import service with batch persistence

* feat: add CLI entrypoint for jsonl import

* test: add unit tests for event import service

* feat: add npm run import script

* docs: add jsonl import usage to README

* refactor: reuse EventRepository instead of raw queries in import service

* fix: enrich event metadata before persistence in import service

* fix: remove double onProgress call at end of import

Author: Anshumancanrock

README.md

@@ -230,6 +230,29 @@ Print the Tor hostname:
   ./scripts/print_tor_hostname
   ```
 
+### Importing events from JSON Lines
+
+You can import NIP-01 events from a `.jsonl` file directly into the relay database.
+
+Basic import:
+  ```
+  npm run import -- ./events.jsonl
+  ```
+
+Set a custom batch size (default: `1000`):
+  ```
+  npm run import -- ./events.jsonl --batch-size 500
+  ```
+
+The importer:
+
+- Processes the file line-by-line to keep memory usage bounded.
+- Validates NIP-01 schema, event id hash, and Schnorr signature before insertion.
+- Inserts in database transactions per batch.
+- Skips duplicates without failing the whole import.
+- Prints progress in the format:
+  `[Processed: 50,000 | Inserted: 45,000 | Skipped: 5,000 | Errors: 0]`
+
 ### Running as a Service
 
 By default this server will run continuously until you stop it with Ctrl+C or until the system restarts.

package.json

@@ -31,6 +31,7 @@
     "lint": "eslint --ext .ts ./src ./test",
     "lint:report": "eslint -o .lint-reports/eslint.json -f json --ext .ts ./src ./test",
     "lint:fix": "npm run lint -- --fix",
+    "import": "node -r ts-node/register src/import-events.ts",
     "db:migrate": "knex migrate:latest",
     "db:migrate:rollback": "knex migrate:rollback",
     "db:seed": "knex seed:run",

src/@types/repositories.ts

@@ -14,7 +14,9 @@ export interface IQueryResult<T> extends Pick<Promise<T>, keyof Promise<T> & Exp
 
 export interface IEventRepository {
   create(event: Event): Promise<number>
+  createMany(events: Event[]): Promise<number>
   upsert(event: Event): Promise<number>
+  upsertMany(events: Event[]): Promise<number>
   findByFilters(filters: SubscriptionFilter[]): IQueryResult<DBEvent[]>
   deleteByPubkeyAndIds(pubkey: Pubkey, ids: EventId[]): Promise<number>
   deleteByPubkeyExceptKinds(pubkey: Pubkey, excludedKinds: number[]): Promise<number>

src/import-events.ts

@@ -0,0 +1,183 @@
+import { extname, resolve } from 'path'
+
+import fs from 'fs'
+
+import dotenv from 'dotenv'
+
+dotenv.config()
+
+import {
+  createEventBatchPersister,
+  EventImportLineError,
+  EventImportService,
+  EventImportStats,
+} from './services/event-import-service'
+import { EventRepository } from './repositories/event-repository'
+import { getMasterDbClient } from './database/client'
+
+interface CliOptions {
+  batchSize: number
+  filePath: string
+  showHelp: boolean
+}
+
+const DEFAULT_BATCH_SIZE = 1000
+const MAX_ERROR_LOGS = 20
+
+const formatNumber = (value: number): string => value.toLocaleString('en-US')
+
+const formatProgress = (stats: EventImportStats): string => {
+  return `[Processed: ${formatNumber(stats.processed)} | Inserted: ${formatNumber(stats.inserted)} | Skipped: ${formatNumber(stats.skipped)} | Errors: ${formatNumber(stats.errors)}]`
+}
+
+const printUsage = (): void => {
+  console.log('Usage: npm run import -- <file.jsonl> [--batch-size <number>]')
+  console.log('Example: npm run import -- ./events.jsonl --batch-size 1000')
+}
+
+const parseBatchSize = (value: string): number => {
+  const parsedValue = Number(value)
+
+  if (!Number.isInteger(parsedValue) || parsedValue <= 0) {
+    throw new Error(`Invalid --batch-size value: ${value}`)
+  }
+
+  return parsedValue
+}
+
+const parseCliArgs = (args: string[]): CliOptions => {
+  let batchSize = DEFAULT_BATCH_SIZE
+  let filePath: string | undefined
+
+  if (args.includes('--help') || args.includes('-h')) {
+    return {
+      batchSize,
+      filePath: '',
+      showHelp: true,
+    }
+  }
+
+  for (let i = 0; i < args.length; i++) {
+    const arg = args[i]
+
+    if (arg === '--batch-size') {
+      const nextArg = args[i + 1]
+      if (typeof nextArg !== 'string') {
+        throw new Error('Missing value for --batch-size')
+      }
+
+      batchSize = parseBatchSize(nextArg)
+      i += 1
+      continue
+    }
+
+    if (arg.startsWith('--batch-size=')) {
+      batchSize = parseBatchSize(arg.split('=', 2)[1])
+      continue
+    }
+
+    if (arg.startsWith('--')) {
+      throw new Error(`Unknown option: ${arg}`)
+    }
+
+    if (filePath) {
+      throw new Error(`Unexpected extra argument: ${arg}`)
+    }
+
+    filePath = arg
+  }
+
+  if (!filePath) {
+    throw new Error('Missing path to .jsonl file')
+  }
+
+  return {
+    batchSize,
+    filePath,
+    showHelp: false,
+  }
+}
+
+const ensureValidInputFile = (filePath: string): string => {
+  const absolutePath = resolve(process.cwd(), filePath)
+
+  if (extname(absolutePath).toLowerCase() !== '.jsonl') {
+    throw new Error('Input file must have a .jsonl extension')
+  }
+
+  if (!fs.existsSync(absolutePath)) {
+    throw new Error(`Input file does not exist: ${absolutePath}`)
+  }
+
+  const stats = fs.statSync(absolutePath)
+  if (!stats.isFile()) {
+    throw new Error(`Input path is not a file: ${absolutePath}`)
+  }
+
+  return absolutePath
+}
+
+const run = async (): Promise<void> => {
+  const options = parseCliArgs(process.argv.slice(2))
+
+  if (options.showHelp) {
+    printUsage()
+    return
+  }
+
+  const absoluteFilePath = ensureValidInputFile(options.filePath)
+
+  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 {
+    const stats = await importer.importFromJsonl(absoluteFilePath, {
+      batchSize: options.batchSize,
+      onLineError,
+      onProgress,
+    })
+
+    if (suppressedErrors > 0) {
+      console.warn(`Suppressed ${formatNumber(suppressedErrors)} additional line errors`)
+    }
+
+    const elapsedSeconds = ((Date.now() - startedAt) / 1000).toFixed(2)
+
+    console.log(`Import completed in ${elapsedSeconds}s`)
+    console.log(formatProgress(stats))
+  } finally {
+    await dbClient.destroy()
+  }
+}
+
+if (require.main === module) {
+  run().catch((error: unknown) => {
+    if (error instanceof Error) {
+      console.error(`Import failed: ${error.message}`)
+    } else {
+      console.error('Import failed with unknown error')
+    }
+
+    process.exit(1)
+  })
+}

src/repositories/event-repository.ts

@@ -170,9 +170,22 @@ export class EventRepository implements IEventRepository {
     return this.insert(event).then(prop('rowCount') as () => number, () => 0)
   }
 
-  private insert(event: Event) {
-    debug('inserting event: %o', event)
-    const row = applySpec({
+  public async createMany(events: Event[]): Promise<number> {
+    if (!events.length) {
+      return 0
+    }
+
+    const rows = events.map((event) => this.toInsertRow(event))
+
+    return this.masterDbClient('events')
+      .insert(rows)
+      .onConflict()
+      .ignore()
+      .then(prop('rowCount') as () => number, () => 0)
+  }
+
+  private toInsertRow(event: Event) {
+    return applySpec({
       event_id: pipe(prop('id'), toBuffer),
       event_pubkey: pipe(prop('pubkey'), toBuffer),
       event_created_at: prop('created_at'),
@@ -187,6 +200,11 @@ export class EventRepository implements IEventRepository {
         always(null),
       ),
     })(event)
+  }
+
+  private insert(event: Event) {
+    debug('inserting event: %o', event)
+    const row = this.toInsertRow(event)
 
     return this.masterDbClient('events')
       .insert(row)
@@ -197,9 +215,50 @@ export class EventRepository implements IEventRepository {
   public upsert(event: Event): Promise<number> {
     debug('upserting event: %o', event)
 
+    const row = this.toUpsertRow(event)
+
+    const query = this.masterDbClient('events')
+      .insert(row)
+      // NIP-16: Replaceable Events
+      // NIP-33: Parameterized Replaceable Events
+      .onConflict(
+        this.masterDbClient.raw(
+          '(event_pubkey, event_kind, event_deduplication) WHERE (event_kind = 0 OR event_kind = 3 OR event_kind = 41 OR (event_kind >= 10000 AND event_kind < 20000)) OR (event_kind >= 30000 AND event_kind < 40000)'
+        )
+      )
+      .merge(omit(['event_pubkey', 'event_kind', 'event_deduplication'])(row))
+      .where('events.event_created_at', '<', row.event_created_at)
+
+    return {
+      then: <T1, T2>(onfulfilled: (value: number) => T1 | PromiseLike<T1>, onrejected: (reason: any) => T2 | PromiseLike<T2>) => query.then(prop('rowCount') as () => number).then(onfulfilled, onrejected),
+      catch: <T>(onrejected: (reason: any) => T | PromiseLike<T>) => query.catch(onrejected),
+      toString: (): string => query.toString(),
+    } as Promise<number>
+  }
+
+  public async upsertMany(events: Event[]): Promise<number> {
+    if (!events.length) {
+      return 0
+    }
+
+    const rows = events.map((event) => this.toUpsertRow(event))
+
+    return this.masterDbClient('events')
+      .insert(rows)
+      .onConflict(
+        this.masterDbClient.raw(
+          '(event_pubkey, event_kind, event_deduplication) WHERE (event_kind = 0 OR event_kind = 3 OR event_kind = 41 OR (event_kind >= 10000 AND event_kind < 20000)) OR (event_kind >= 30000 AND event_kind < 40000)'
+        )
+      )
+      .merge(['deleted_at', 'event_content', 'event_created_at', 'event_id', 'event_signature', 'event_tags', 'expires_at'])
+      .whereRaw('"events"."event_created_at" < "excluded"."event_created_at"')
+      .then(prop('rowCount') as () => number, () => 0)
+  }
+
+  private toUpsertRow(event: Event) {
     const toJSON = (input: any) => JSON.stringify(input)
 
-    const row = applySpec<DBEvent>({
+    return applySpec<DBEvent>({
       event_id: pipe(prop('id'), toBuffer),
       event_pubkey: pipe(prop('pubkey'), toBuffer),
       event_created_at: prop('created_at'),
@@ -220,24 +279,6 @@ export class EventRepository implements IEventRepository {
       ),
       deleted_at: always(null),
     })(event)
-
-    const query = this.masterDbClient('events')
-      .insert(row)
-      // NIP-16: Replaceable Events
-      // NIP-33: Parameterized Replaceable Events
-      .onConflict(
-        this.masterDbClient.raw(
-          '(event_pubkey, event_kind, event_deduplication) WHERE (event_kind = 0 OR event_kind = 3 OR event_kind = 41 OR (event_kind >= 10000 AND event_kind < 20000)) OR (event_kind >= 30000 AND event_kind < 40000)'
-        )
-      )
-      .merge(omit(['event_pubkey', 'event_kind', 'event_deduplication'])(row))
-      .where('events.event_created_at', '<', row.event_created_at)
-
-    return {
-      then: <T1, T2>(onfulfilled: (value: number) => T1 | PromiseLike<T1>, onrejected: (reason: any) => T2 | PromiseLike<T2>) => query.then(prop('rowCount') as () => number).then(onfulfilled, onrejected),
-      catch: <T>(onrejected: (reason: any) => T | PromiseLike<T>) => query.catch(onrejected),
-      toString: (): string => query.toString(),
-    } as Promise<number>
   }
 
   public deleteByPubkeyAndIds(pubkey: string, eventIdsToDelete: EventId[]): Promise<number> {