refactor: reuse EventRepository instead of raw queries in import service

Author: Anshumancanrock

src/import-events.ts

@@ -13,6 +13,7 @@ import {
   EventImportStats,
 } from './services/event-import-service'
 import { getMasterDbClient } from './database/client'
+import { EventRepository } from './repositories/event-repository'
 
 interface CliOptions {
   batchSize: number
@@ -127,7 +128,8 @@ const run = async (): Promise<void> => {
   const absoluteFilePath = ensureValidInputFile(options.filePath)
 
   const dbClient = getMasterDbClient()
-  const importer = new EventImportService(createEventBatchPersister(dbClient))
+  const eventRepository = new EventRepository(dbClient, dbClient)
+  const importer = new EventImportService(createEventBatchPersister(eventRepository))
 
   let loggedErrors = 0
   let suppressedErrors = 0

src/services/event-import-service.ts

@@ -1,46 +1,23 @@
 import fs from 'fs'
 import readline from 'readline'
 
-import { Knex } from 'knex'
-
-import { DatabaseClient, EventId } from '../@types/base'
 import {
-  getEventExpiration,
   isDeleteEvent,
   isEphemeralEvent,
   isEventIdValid,
   isEventSignatureValid,
   isParameterizedReplaceableEvent,
   isReplaceableEvent,
 } from '../utils/event'
-import { toBuffer, toJSON } from '../utils/transform'
 import { attemptValidation } from '../utils/validation'
 
 import { Event } from '../@types/event'
 import { eventSchema } from '../schemas/event-schema'
 import { EventTags } from '../constants/base'
+import { IEventRepository } from '../@types/repositories'
 
 const DEFAULT_BATCH_SIZE = 1000
 
-const REPLACEABLE_EVENT_CONFLICT_TARGET =
-  '(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)'
-
-interface ImportEventRow {
-  deleted_at: null
-  event_content: string
-  event_created_at: number
-  event_deduplication: string | null
-  event_id: Buffer
-  event_kind: number
-  event_pubkey: Buffer
-  event_signature: Buffer
-  event_tags: string
-  expires_at: number | null
-}
-
 export interface EventImportStats {
   errors: number
   inserted: number
@@ -67,276 +44,48 @@ const getErrorMessage = (error: unknown): string => {
   return String(error)
 }
 
-const getAffectedRowCount = (result: unknown): number => {
-  if (Array.isArray(result)) {
-    return result.length
-  }
-
-  if (
-    typeof result === 'object'
-    && result !== null
-    && 'rowCount' in result
-    && typeof (result as { rowCount: unknown }).rowCount === 'number'
-  ) {
-    return Number((result as { rowCount: number }).rowCount)
-  }
-
-  return 0
-}
-
-const isEventIdUniqueViolation = (error: unknown): boolean => {
-  if (typeof error !== 'object' || error === null) {
-    return false
-  }
-
-  const dbError = error as {
-    code?: string
-    constraint?: string
-    message?: string
-  }
-
-  return dbError.code === '23505'
-    && (
-      dbError.constraint === 'events_event_id_unique'
-      || dbError.message?.includes('events_event_id_unique') === true
-    )
-}
-
-const isValidDeleteTag = (tag: string[]): boolean => {
-  return tag.length >= 2
-    && tag[0] === EventTags.Event
-    && /^[0-9a-f]{64}$/.test(tag[1])
-}
-
-const getDeleteTargetEventIds = (event: Event): EventId[] => {
-  return event.tags.reduce((eventIds, tag) => {
-    if (isValidDeleteTag(tag)) {
-      eventIds.push(tag[1])
-    }
-
-    return eventIds
-  }, [] as EventId[])
-}
-
-const isEventReplaceableForStorage = (event: Event): boolean => {
-  return isReplaceableEvent(event) || isParameterizedReplaceableEvent(event)
-}
-
-const getReplaceableEventDeduplication = (event: Event): string => {
-  if (isParameterizedReplaceableEvent(event)) {
-    const [, ...deduplication] = event.tags.find(
-      (tag) => tag.length >= 2 && tag[0] === EventTags.Deduplication,
-    ) ?? [null, '']
-
-    return toJSON(deduplication)
-  }
-
-  return toJSON([event.pubkey, event.kind])
-}
-
-const getReplaceableEventKey = (event: Event): string => {
-  return `${event.pubkey}:${event.kind}:${getReplaceableEventDeduplication(event)}`
-}
-
-const toImportEventRow = (event: Event): ImportEventRow => {
-  const expiresAt = getEventExpiration(event)
-
-  return {
-    deleted_at: null,
-    event_content: event.content,
-    event_created_at: event.created_at,
-    event_deduplication: (
-      isReplaceableEvent(event) || isParameterizedReplaceableEvent(event)
-        ? getReplaceableEventDeduplication(event)
-        : null
-    ),
-    event_id: toBuffer(event.id),
-    event_kind: event.kind,
-    event_pubkey: toBuffer(event.pubkey),
-    event_signature: toBuffer(event.sig),
-    event_tags: toJSON(event.tags),
-    expires_at: typeof expiresAt === 'number' ? expiresAt : null,
-  }
-}
-
-const applyDeleteEvents = async (
-  transaction: Knex.Transaction,
-  deleteEvent: Event,
-): Promise<void> => {
-  const eventIds = getDeleteTargetEventIds(deleteEvent)
-  if (!eventIds.length) {
-    return
-  }
-
-  await transaction('events')
-    .where('event_pubkey', toBuffer(deleteEvent.pubkey))
-    .whereIn('event_id', eventIds.map(toBuffer))
-    .whereNull('deleted_at')
-    .update({
-      deleted_at: transaction.raw('now()'),
-    })
-}
-
-const insertRegularEvents = async (
-  transaction: Knex.Transaction,
-  events: Event[],
-): Promise<number> => {
-  if (!events.length) {
-    return 0
-  }
-
-  const rows = events.map(toImportEventRow)
-
-  const result = await transaction('events')
-    .insert(rows)
-    .onConflict()
-    .ignore()
-    .returning('event_id')
-
-  return getAffectedRowCount(result)
-}
-
-const filterOutExistingEventIds = async (
-  transaction: Knex.Transaction,
-  events: Event[],
-): Promise<Event[]> => {
-  if (!events.length) {
-    return []
-  }
-
-  const existingRows = await transaction('events')
-    .select('event_id')
-    .whereIn('event_id', events.map((event) => toBuffer(event.id))) as Array<{ event_id: Buffer }>
-
-  const existingEventIds = new Set(existingRows.map((row) => row.event_id.toString('hex')))
-
-  return events.filter((event) => !existingEventIds.has(event.id))
-}
-
-const upsertReplaceableEvents = async (
-  transaction: Knex.Transaction,
-  events: Event[],
-): Promise<number> => {
-  if (!events.length) {
-    return 0
-  }
-
-  let pendingEvents = events
-
-  while (pendingEvents.length) {
-    const deduplicatedByEventId = new Map<string, Event>()
-    for (const event of pendingEvents) {
-      deduplicatedByEventId.set(event.id, event)
-    }
-
-    pendingEvents = Array.from(deduplicatedByEventId.values())
-
-    const rows = pendingEvents.map(toImportEventRow)
-
-    try {
-      const result = await transaction('events')
-        .insert(rows)
-        .onConflict(transaction.raw(REPLACEABLE_EVENT_CONFLICT_TARGET))
-        .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"')
-        .returning('event_id')
-
-      return getAffectedRowCount(result)
-    } catch (error) {
-      if (!isEventIdUniqueViolation(error)) {
-        throw error
-      }
-
-      const filteredEvents = await filterOutExistingEventIds(transaction, pendingEvents)
-
-      if (filteredEvents.length === pendingEvents.length) {
-        throw error
-      }
-
-      pendingEvents = filteredEvents
-    }
-  }
-
-  return 0
-}
-
 export const createEventBatchPersister =
-  (dbClient: DatabaseClient) =>
+  (eventRepository: IEventRepository) =>
     async (events: Event[]): Promise<number> => {
       if (!events.length) {
         return 0
       }
 
-      return dbClient.transaction(async (transaction) => {
-        let inserted = 0
+      let inserted = 0
 
-        let nonDeleteSegment: Event[] = []
-
-        const flushNonDeleteSegment = async () => {
-          if (!nonDeleteSegment.length) {
-            return
-          }
-
-          const regularEvents: Event[] = []
-          const replaceableEventsByKey = new Map<string, Event>()
-
-          for (const event of nonDeleteSegment) {
-            if (isEventReplaceableForStorage(event)) {
-              const deduplicationKey = getReplaceableEventKey(event)
-              const existingEvent = replaceableEventsByKey.get(deduplicationKey)
-
-              if (!existingEvent || existingEvent.created_at < event.created_at) {
-                replaceableEventsByKey.set(deduplicationKey, event)
-              }
-
-              continue
-            }
-
-            regularEvents.push(event)
-          }
-
-          inserted += await insertRegularEvents(transaction, regularEvents)
+      for (const event of events) {
+        if (isEphemeralEvent(event)) {
+          continue
+        }
 
-          const upsertEvents = await filterOutExistingEventIds(
-            transaction,
-            Array.from(replaceableEventsByKey.values()),
+        if (isDeleteEvent(event)) {
+          const eventIdsToDelete = event.tags.reduce(
+            (ids, tag) =>
+              tag.length >= 2
+              && tag[0] === EventTags.Event
+              && /^[0-9a-f]{64}$/.test(tag[1])
+                ? [...ids, tag[1]]
+                : ids,
+            [] as string[]
           )
 
-          inserted += await upsertReplaceableEvents(transaction, upsertEvents)
-
-          nonDeleteSegment = []
-        }
-
-        for (const event of events) {
-          if (isEphemeralEvent(event)) {
-            continue
+          if (eventIdsToDelete.length) {
+            await eventRepository.deleteByPubkeyAndIds(event.pubkey, eventIdsToDelete)
           }
 
-          if (isDeleteEvent(event)) {
-            await flushNonDeleteSegment()
-
-            await applyDeleteEvents(transaction, event)
-
-            inserted += await insertRegularEvents(transaction, [event])
-
-            continue
-          }
+          inserted += await eventRepository.create(event)
+          continue
+        }
 
-          nonDeleteSegment.push(event)
+        if (isReplaceableEvent(event) || isParameterizedReplaceableEvent(event)) {
+          inserted += await eventRepository.upsert(event)
+          continue
         }
 
-        await flushNonDeleteSegment()
+        inserted += await eventRepository.create(event)
+      }
 
-        return inserted
-      })
+      return inserted
     }
 
 export class EventImportService {