feat(storage): add eventStore.backend setting and a single event repository factory

.changeset/event-store-backend-setting.md

@@ -0,0 +1,7 @@
+---
+"nostream": minor
+---
+
+feat(storage): add an `eventStore.backend` setting and build every event repository through one factory
+
+Defaults to `postgres`, the only backend for now. The backend is read once at startup and passed to every worker.

CONFIGURATION.md

@@ -145,6 +145,7 @@ The settings below are listed in alphabetical order by name. Please keep this ta
 | dvm.workers[].command                       | Command to spawn for this DVM worker (e.g. an interpreter or executable path). |
 | dvm.workers[].kinds                         | NIP-90 job request kinds (5000-5999) this worker accepts. Optional. |
 | dvm.workers[].timeoutMs                     | Max time in ms to wait for a job result before considering it timed out. Optional. |
+| eventStore.backend                          | Where events are stored. Only `postgres` for now. Read once at startup, so changing it needs a restart; an unsupported value stops the relay. Defaults to `postgres`. |
 | info.banner                                 | Public banner image URL for the relay information document. |
 | info.contact                                | Relay operator's contact. (e.g. mailto:operator@relay-your-domain.com) |
 | info.description                            | Public description of your relay. (e.g. Toronto Bitcoin Group Public Relay) |

resources/default-settings.yaml

@@ -162,6 +162,9 @@ network:
     - "::ffff:10.10.10.1"
 workers:
   count: 0
+eventStore:
+  # Only postgres for now. Changing it needs a restart.
+  backend: postgres
 mirroring:
   static: []
 dvm:

src/@types/settings.ts

@@ -1,4 +1,4 @@
-import { EventKinds } from '../constants/base'
+import { EventKinds, EventStoreBackend } from '../constants/base'
 import { Pubkey, Secret } from './base'
 import { MessageType } from './messages'
 import { AdminNotificationsSettings } from './operator-notifications'
@@ -202,6 +202,11 @@ export interface Worker {
   count: number
 }
 
+export interface EventStoreSettings {
+  /** Read once at startup, so changing it needs a restart. Defaults to postgres. */
+  backend: EventStoreBackend
+}
+
 export interface FeeScheduleWhitelists {
   pubkeys?: Pubkey[]
   event_kinds?: (EventKinds | [EventKinds, EventKinds])[]
@@ -475,6 +480,7 @@ export interface Settings {
   paymentsProcessors?: PaymentsProcessors
   network: Network
   workers?: Worker
+  eventStore?: EventStoreSettings
   limits?: Limits
   mirroring?: Mirroring
   dvm?: Dvm

src/app/app.ts

@@ -5,6 +5,7 @@ import { FSWatcher } from 'fs'
 
 import { addOnion } from '../tor/client'
 import { createLogger } from '../factories/logger-factory'
+import { EventStoreBackend } from '../constants/base'
 import { IRunnable } from '../@types/base'
 import packageJson from '../../package.json'
 import { Serializable } from 'child_process'
@@ -14,6 +15,7 @@ import { shutdownMetricsTelemetry } from '../telemetry/metrics'
 import { OperatorNotificationEventType } from '../@types/operator-notifications'
 import { RedisRelayBroadcastFanout } from '../relay-broadcast/redis-relay-broadcast-fanout'
 import { enqueueOperatorNotification } from '../utils/operator-notification-enqueue'
+import { EVENT_STORE_BACKEND_ENV, getConfiguredEventStoreBackend } from '../utils/event-store'
 import { RelayBroadcastDeduplicator } from '../utils/relay-broadcast-deduplicator'
 import {
   isRelayBroadcastFanoutEnabled,
@@ -87,13 +89,25 @@ export class App implements IRunnable {
       this.process.exit(1)
     }
 
+    // Resolve once so every worker shares it and a bad value fails before forking.
+    let eventStoreBackend: EventStoreBackend
+    try {
+      eventStoreBackend = getConfiguredEventStoreBackend(settings)
+    } catch (error) {
+      logger.error((error as Error).message)
+      this.process.exit(1)
+      return
+    }
+    logCentered(`Event store: ${eventStoreBackend}`, width)
+
     const workerCount = process.env.WORKER_COUNT
       ? Number(process.env.WORKER_COUNT)
       : this.settings().workers?.count || cpus().length
 
     const createWorker = (env: Record<string, string>) => {
-      const worker = this.cluster.fork(env)
-      this.workers.set(worker, env)
+      const workerEnv = { ...env, [EVENT_STORE_BACKEND_ENV]: eventStoreBackend }
+      const worker = this.cluster.fork(workerEnv)
+      this.workers.set(worker, workerEnv)
     }
 
     for (let i = 0; i < workerCount; i++) {

src/constants/base.ts

@@ -127,6 +127,10 @@ export enum PaymentsProcessors {
   LNBITS = 'lnbits',
 }
 
+export enum EventStoreBackend {
+  POSTGRES = 'postgres',
+}
+
 export const EventDeduplicationMetadataKey = Symbol('Deduplication')
 export const ContextMetadataKey = Symbol('Context')
 export const EventExpirationTimeMetadataKey = Symbol('Expiration')

src/factories/controllers/get-admission-check-controller-factory.ts

@@ -1,14 +1,14 @@
 import { getMasterDbClient, getReadReplicaDbClient } from '../../database/client'
+import { createEventRepository } from '../event-repository-factory'
 import { createSettings } from '../settings-factory'
-import { EventRepository } from '../../repositories/event-repository'
 import { GetSubmissionCheckController } from '../../controllers/admission/get-admission-check-controller'
 import { rateLimiterFactory } from '../rate-limiter-factory'
 import { UserRepository } from '../../repositories/user-repository'
 
 export const createGetAdmissionCheckController = () => {
   const dbClient = getMasterDbClient()
   const readReplicaDbClient = getReadReplicaDbClient()
-  const eventRepository = new EventRepository(dbClient, readReplicaDbClient)
+  const eventRepository = createEventRepository(dbClient, readReplicaDbClient)
   const userRepository = new UserRepository(dbClient, eventRepository)
 
   return new GetSubmissionCheckController(userRepository, createSettings, rateLimiterFactory)

src/factories/controllers/post-invoice-controller-factory.ts

@@ -1,7 +1,7 @@
 import { getMasterDbClient, getReadReplicaDbClient } from '../../database/client'
+import { createEventRepository } from '../event-repository-factory'
 import { createPaymentsService } from '../payments-service-factory'
 import { createSettings } from '../settings-factory'
-import { EventRepository } from '../../repositories/event-repository'
 import { IController } from '../../@types/controllers'
 import { PostInvoiceController } from '../../controllers/invoices/post-invoice-controller'
 import { rateLimiterFactory } from '../rate-limiter-factory'
@@ -10,7 +10,7 @@ import { UserRepository } from '../../repositories/user-repository'
 export const createPostInvoiceController = (): IController => {
   const dbClient = getMasterDbClient()
   const readReplicaDbClient = getReadReplicaDbClient()
-  const eventRepository = new EventRepository(dbClient, readReplicaDbClient)
+  const eventRepository = createEventRepository(dbClient, readReplicaDbClient)
   const userRepository = new UserRepository(dbClient, eventRepository)
   const paymentsService = createPaymentsService()
 

src/factories/dvm-orchestrator-worker-factory.ts

@@ -1,16 +1,16 @@
 import process from 'process'
 
 import { getMasterDbClient, getReadReplicaDbClient } from '../database/client'
+import { createEventRepository } from './event-repository-factory'
 import { createSettings } from './settings-factory'
 import { DvmJobRepository } from '../repositories/dvm-job-repository'
 import { DvmOrchestratorWorker } from '../app/dvm-orchestrator-worker'
-import { EventRepository } from '../repositories/event-repository'
 
 export const dvmOrchestratorWorkerFactory = () => {
   const dbClient = getMasterDbClient()
   const readReplicaDbClient = getReadReplicaDbClient()
   const dvmJobRepository = new DvmJobRepository(dbClient)
-  const eventRepository = new EventRepository(dbClient, readReplicaDbClient, createSettings)
+  const eventRepository = createEventRepository(dbClient, readReplicaDbClient)
 
   return new DvmOrchestratorWorker(process, createSettings, dvmJobRepository, eventRepository)
 }

src/factories/event-repository-factory.ts

@@ -0,0 +1,25 @@
+import { createSettings } from './settings-factory'
+import { DatabaseClient } from '../@types/base'
+import { EventRepository } from '../repositories/event-repository'
+import { EventStoreBackend } from '../constants/base'
+import { getEventStoreBackend } from '../utils/event-store'
+import { IEventRepository } from '../@types/repositories'
+import { Settings } from '../@types/settings'
+
+export const createEventRepository = (
+  masterDbClient: DatabaseClient,
+  readReplicaDbClient: DatabaseClient,
+  settings: () => Settings = createSettings,
+): IEventRepository => {
+  const backend = getEventStoreBackend(settings())
+
+  switch (backend) {
+    case EventStoreBackend.POSTGRES:
+      return new EventRepository(masterDbClient, readReplicaDbClient, settings)
+    default: {
+      // exhaustiveness check
+      const unhandled: never = backend
+      throw new Error(`No event repository is available for eventStore.backend "${unhandled}"`)
+    }
+  }
+}

src/factories/maintenance-service-factory.ts

@@ -1,8 +1,8 @@
 import { getMasterDbClient, getReadReplicaDbClient } from '../database/client'
+import { createEventRepository } from './event-repository-factory'
 import { createSettings } from './settings-factory'
-import { EventRepository } from '../repositories/event-repository'
 import { MaintenanceService } from '../services/maintenance-service'
 
 export const createMaintenanceService = () => {
-  return new MaintenanceService(new EventRepository(getMasterDbClient(), getReadReplicaDbClient()), createSettings)
+  return new MaintenanceService(createEventRepository(getMasterDbClient(), getReadReplicaDbClient()), createSettings)
 }

src/factories/payments-service-factory.ts

@@ -1,7 +1,7 @@
 import { getMasterDbClient, getReadReplicaDbClient } from '../database/client'
+import { createEventRepository } from './event-repository-factory'
 import { createPaymentsProcessor } from './payments-processor-factory'
 import { createSettings } from './settings-factory'
-import { EventRepository } from '../repositories/event-repository'
 import { InvoiceRepository } from '../repositories/invoice-repository'
 import { PaymentsService } from '../services/payments-service'
 import { UserRepository } from '../repositories/user-repository'
@@ -11,7 +11,7 @@ export const createPaymentsService = () => {
   const dbClient = getMasterDbClient()
   const rrDbClient = getReadReplicaDbClient()
   const invoiceRepository = new InvoiceRepository(dbClient)
-  const eventRepository = new EventRepository(dbClient, rrDbClient)
+  const eventRepository = createEventRepository(dbClient, rrDbClient)
   const userRepository = new UserRepository(dbClient, eventRepository)
   const notificationOutboxRepository = new NotificationOutboxRepository(dbClient)
   const paymentsProcessor = createPaymentsProcessor()

src/factories/relay-monitor-worker-factory.ts

@@ -2,16 +2,16 @@ import { RedisAdapter } from '../adapters/redis-adapter'
 import { RelayMonitorWorker } from '../app/relay-monitor-worker'
 import { getCacheClient } from '../cache/client'
 import { getMasterDbClient, getReadReplicaDbClient } from '../database/client'
+import { createEventRepository } from './event-repository-factory'
 import { createSettings } from './settings-factory'
-import { EventRepository } from '../repositories/event-repository'
 import { Nip66EventPublisher } from '../services/nip66-event-publisher'
 import { RelayProbeSnapshotStore } from '../utils/relay-probe-snapshot'
 import { runProbe } from '../utils/relay-probe'
 
 export const relayMonitorWorkerFactory = () => {
   const cache = new RedisAdapter(getCacheClient())
   const snapshotStore = new RelayProbeSnapshotStore(cache)
-  const eventRepository = new EventRepository(getMasterDbClient(), getReadReplicaDbClient(), createSettings)
+  const eventRepository = createEventRepository(getMasterDbClient(), getReadReplicaDbClient())
   const eventPublisher = new Nip66EventPublisher(eventRepository, cache)
 
   return new RelayMonitorWorker(process, createSettings, snapshotStore, runProbe, eventPublisher)

src/factories/static-mirroring.worker-factory.ts

@@ -1,13 +1,13 @@
 import { getMasterDbClient, getReadReplicaDbClient } from '../database/client'
+import { createEventRepository } from './event-repository-factory'
 import { createSettings } from './settings-factory'
-import { EventRepository } from '../repositories/event-repository'
 import { StaticMirroringWorker } from '../app/static-mirroring-worker'
 import { UserRepository } from '../repositories/user-repository'
 
 export const staticMirroringWorkerFactory = () => {
   const dbClient = getMasterDbClient()
   const readReplicaDbClient = getReadReplicaDbClient()
-  const eventRepository = new EventRepository(dbClient, readReplicaDbClient)
+  const eventRepository = createEventRepository(dbClient, readReplicaDbClient)
   const userRepository = new UserRepository(dbClient, eventRepository)
 
   return new StaticMirroringWorker(eventRepository, userRepository, process, createSettings)

src/factories/worker-factory.ts

@@ -7,12 +7,12 @@ import { AppWorker } from '../app/worker'
 import { getMasterDbClient, getReadReplicaDbClient } from '../database/client'
 import { createSettings } from '../factories/settings-factory'
 import { DvmJobRepository } from '../repositories/dvm-job-repository'
-import { EventRepository } from '../repositories/event-repository'
 import { InviteCodeRepository } from '../repositories/invite-code-repository'
 import { Nip05VerificationRepository } from '../repositories/nip05-verification-repository'
 import { ReportRepository } from '../repositories/report-repository'
 import { UserRepository } from '../repositories/user-repository'
 import { startHiddenContentCache } from '../utils/hidden-content-cache'
+import { createEventRepository } from './event-repository-factory'
 import { createLogger } from './logger-factory'
 import { getCache } from './message-handler-factory'
 import { createWebApp } from './web-app-factory'
@@ -24,7 +24,7 @@ const logger = createLogger('worker-factory')
 export const workerFactory = (): AppWorker => {
   const dbClient = getMasterDbClient()
   const readReplicaDbClient = getReadReplicaDbClient()
-  const eventRepository = new EventRepository(dbClient, readReplicaDbClient, createSettings)
+  const eventRepository = createEventRepository(dbClient, readReplicaDbClient)
   const userRepository = new UserRepository(dbClient, eventRepository)
   const nip05VerificationRepository = new Nip05VerificationRepository(dbClient)
   const inviteCodeRepository = new InviteCodeRepository(dbClient)

src/import-events.ts

@@ -9,7 +9,7 @@ import {
   EventImportService,
   EventImportStats,
 } from './services/event-import-service'
-import { EventRepository } from './repositories/event-repository'
+import { createEventRepository } from './factories/event-repository-factory'
 import { getMasterDbClient } from './database/client'
 
 interface CliOptions {
@@ -218,7 +218,7 @@ export const runImportEvents = async (
   }
 
   const dbClient = getMasterDbClient()
-  const eventRepository = new EventRepository(dbClient, dbClient)
+  const eventRepository = createEventRepository(dbClient, dbClient)
   const importer = new EventImportService(createEventBatchPersister(eventRepository))
 
   let loggedErrors = 0

src/utils/event-store.ts

@@ -0,0 +1,32 @@
+import cluster from 'cluster'
+
+import { EventStoreBackend } from '../constants/base'
+import { Settings } from '../@types/settings'
+
+// Set by the primary on every worker it forks.
+export const EVENT_STORE_BACKEND_ENV = 'EVENT_STORE_BACKEND'
+
+export const EVENT_STORE_BACKENDS: readonly EventStoreBackend[] = Object.values(EventStoreBackend)
+
+export const isEventStoreBackend = (value: unknown): value is EventStoreBackend =>
+  EVENT_STORE_BACKENDS.includes(value as EventStoreBackend)
+
+const toEventStoreBackend = (value: unknown, source: string): EventStoreBackend => {
+  if (!isEventStoreBackend(value)) {
+    throw new Error(`Unsupported ${source} "${String(value)}". Supported backends: ${EVENT_STORE_BACKENDS.join(', ')}`)
+  }
+
+  return value
+}
+
+export const getConfiguredEventStoreBackend = (settings: Settings): EventStoreBackend =>
+  toEventStoreBackend(settings.eventStore?.backend ?? EventStoreBackend.POSTGRES, 'eventStore.backend')
+
+// Workers use the backend the primary resolved at startup; other processes read settings.
+export const getEventStoreBackend = (settings: Settings): EventStoreBackend => {
+  const inherited = cluster.isWorker ? process.env[EVENT_STORE_BACKEND_ENV] : undefined
+
+  return inherited === undefined
+    ? getConfiguredEventStoreBackend(settings)
+    : toEventStoreBackend(inherited, EVENT_STORE_BACKEND_ENV)
+}

src/utils/settings-config.ts

@@ -1,9 +1,10 @@
 import fs from 'fs'
 import { join } from 'path'
 import yaml from 'js-yaml'
-import { mergeDeepRight } from 'ramda'
+import { isNil, mergeDeepRight } from 'ramda'
 
 import { createLogger } from '../factories/logger-factory'
+import { EVENT_STORE_BACKENDS, isEventStoreBackend } from './event-store'
 import { Settings } from '../@types/settings'
 import {
   getConfigBaseDir,
@@ -593,6 +594,14 @@ export const validateSettings = (settings: Settings): ValidationIssue[] => {
     issues.push({ path: 'limits.rateLimiter.strategy', message: 'strategy must be ewma or sliding_window' })
   }
 
+  const eventStoreBackend = settings.eventStore?.backend
+  if (!isNil(eventStoreBackend) && !isEventStoreBackend(eventStoreBackend)) {
+    issues.push({
+      path: 'eventStore.backend',
+      message: `backend must be one of: ${EVENT_STORE_BACKENDS.join(', ')}`,
+    })
+  }
+
   const pow = settings.limits?.event?.pow
   if (pow?.enabled) {
     if (!(pow.floorBits >= 0) || !(pow.floorBits <= pow.ceilingBits)) {