From 117e2db66ebc4f47bd8117dc8b8b9afbc49d2c84 Mon Sep 17 00:00:00 2001 From: anshumancanrock Date: Fri, 9 Oct 2026 20:39:41 +0530 Subject: [PATCH 1/4] feat(config): add eventStore.backend setting --- CONFIGURATION.md | 1 + resources/default-settings.yaml | 3 + src/@types/settings.ts | 8 +- src/constants/base.ts | 4 + src/utils/event-store.ts | 32 +++++++ src/utils/settings-config.ts | 11 ++- test/unit/utils/event-store.spec.ts | 112 ++++++++++++++++++++++++ test/unit/utils/settings-config.spec.ts | 30 +++++++ 8 files changed, 199 insertions(+), 2 deletions(-) create mode 100644 src/utils/event-store.ts create mode 100644 test/unit/utils/event-store.spec.ts diff --git a/CONFIGURATION.md b/CONFIGURATION.md index 2570fd8a..0cdd8360 100644 --- a/CONFIGURATION.md +++ b/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) | diff --git a/resources/default-settings.yaml b/resources/default-settings.yaml index c91d3c4f..115ea845 100755 --- a/resources/default-settings.yaml +++ b/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: diff --git a/src/@types/settings.ts b/src/@types/settings.ts index 874c696c..885ecdb1 100644 --- a/src/@types/settings.ts +++ b/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 diff --git a/src/constants/base.ts b/src/constants/base.ts index 34490531..02e1706f 100644 --- a/src/constants/base.ts +++ b/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') diff --git a/src/utils/event-store.ts b/src/utils/event-store.ts new file mode 100644 index 00000000..727d8844 --- /dev/null +++ b/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) +} diff --git a/src/utils/settings-config.ts b/src/utils/settings-config.ts index f746616f..f3937b6a 100644 --- a/src/utils/settings-config.ts +++ b/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)) { diff --git a/test/unit/utils/event-store.spec.ts b/test/unit/utils/event-store.spec.ts new file mode 100644 index 00000000..f6712f79 --- /dev/null +++ b/test/unit/utils/event-store.spec.ts @@ -0,0 +1,112 @@ +import { expect } from 'chai' +import cluster from 'cluster' +import Sinon from 'sinon' + +import { + EVENT_STORE_BACKEND_ENV, + EVENT_STORE_BACKENDS, + getConfiguredEventStoreBackend, + getEventStoreBackend, + isEventStoreBackend, +} from '../../../src/utils/event-store' +import { EventStoreBackend } from '../../../src/constants/base' +import { Settings } from '../../../src/@types/settings' + +describe('event-store', () => { + const postgresSettings = { eventStore: { backend: 'postgres' } } as Settings + const unsupportedSettings = { eventStore: { backend: 'mongodb' } } as unknown as Settings + + describe('getConfiguredEventStoreBackend', () => { + it('defaults to postgres when eventStore is unset', () => { + expect(getConfiguredEventStoreBackend({} as Settings)).to.equal(EventStoreBackend.POSTGRES) + }) + + it('defaults to postgres when eventStore.backend is unset', () => { + expect(getConfiguredEventStoreBackend({ eventStore: {} } as Settings)).to.equal(EventStoreBackend.POSTGRES) + }) + + it('returns the configured backend', () => { + expect(getConfiguredEventStoreBackend(postgresSettings)).to.equal(EventStoreBackend.POSTGRES) + }) + + it('throws on an unsupported backend and names the supported ones', () => { + expect(() => getConfiguredEventStoreBackend(unsupportedSettings)).to.throw( + Error, + 'Unsupported eventStore.backend "mongodb". Supported backends: postgres', + ) + }) + + it('does not accept a backend in the wrong case', () => { + const settings = { eventStore: { backend: 'Postgres' } } as unknown as Settings + + expect(() => getConfiguredEventStoreBackend(settings)).to.throw( + Error, + 'Unsupported eventStore.backend "Postgres"', + ) + }) + }) + + describe('getEventStoreBackend', () => { + let inheritedBackend: string | undefined + + beforeEach(() => { + inheritedBackend = process.env[EVENT_STORE_BACKEND_ENV] + }) + + afterEach(() => { + Sinon.restore() + if (inheritedBackend === undefined) { + delete process.env[EVENT_STORE_BACKEND_ENV] + } else { + process.env[EVENT_STORE_BACKEND_ENV] = inheritedBackend + } + }) + + it('reads the setting outside a cluster worker, ignoring the inherited variable', () => { + Sinon.stub(cluster, 'isWorker').value(false) + process.env[EVENT_STORE_BACKEND_ENV] = 'mongodb' + + expect(getEventStoreBackend(postgresSettings)).to.equal(EventStoreBackend.POSTGRES) + }) + + it('keeps the backend the primary handed a worker even after the setting changes', () => { + Sinon.stub(cluster, 'isWorker').value(true) + process.env[EVENT_STORE_BACKEND_ENV] = 'postgres' + + expect(getEventStoreBackend(unsupportedSettings)).to.equal(EventStoreBackend.POSTGRES) + }) + + it('reads the setting in a worker the primary handed no backend', () => { + Sinon.stub(cluster, 'isWorker').value(true) + delete process.env[EVENT_STORE_BACKEND_ENV] + + expect(() => getEventStoreBackend(unsupportedSettings)).to.throw( + Error, + 'Unsupported eventStore.backend "mongodb"', + ) + }) + + it('rejects an unsupported inherited backend', () => { + Sinon.stub(cluster, 'isWorker').value(true) + process.env[EVENT_STORE_BACKEND_ENV] = 'mongodb' + + expect(() => getEventStoreBackend(postgresSettings)).to.throw(Error, 'Unsupported EVENT_STORE_BACKEND "mongodb"') + }) + }) + + describe('isEventStoreBackend', () => { + it('accepts every supported backend', () => { + EVENT_STORE_BACKENDS.forEach((backend) => { + expect(isEventStoreBackend(backend)).to.equal(true) + }) + }) + + it('rejects anything else', () => { + expect(isEventStoreBackend('strfry')).to.equal(false) + expect(isEventStoreBackend('')).to.equal(false) + expect(isEventStoreBackend(undefined)).to.equal(false) + expect(isEventStoreBackend(null)).to.equal(false) + expect(isEventStoreBackend(1)).to.equal(false) + }) + }) +}) diff --git a/test/unit/utils/settings-config.spec.ts b/test/unit/utils/settings-config.spec.ts index 9d336ca7..bf50ff94 100644 --- a/test/unit/utils/settings-config.spec.ts +++ b/test/unit/utils/settings-config.spec.ts @@ -9,6 +9,7 @@ import { toCategoryLabel, getByPath, getTopLevelSettingCategories, + loadDefaults, parseTypedValue, parseValue, setByPath, @@ -220,6 +221,35 @@ describe('settings-config', () => { }) }) + describe('event store settings', () => { + const baseSettings = (): any => ({ + info: { relay_url: 'wss://test.relay', name: 'test' }, + network: {}, + }) + + it('ships postgres as the default backend', () => { + expect(loadDefaults().eventStore?.backend).to.equal('postgres') + }) + + it('accepts settings without an eventStore section', () => { + const issues = validateSettings(baseSettings()) + + expect(issues.some((issue) => issue.path.startsWith('eventStore'))).to.equal(false) + }) + + it('accepts the postgres backend', () => { + const issues = validateSettings({ ...baseSettings(), eventStore: { backend: 'postgres' } }) + + expect(issues.some((issue) => issue.path.startsWith('eventStore'))).to.equal(false) + }) + + it('rejects an unsupported backend', () => { + const issues = validateSettings({ ...baseSettings(), eventStore: { backend: 'mongodb' } }) + + expect(issues).to.deep.include({ path: 'eventStore.backend', message: 'backend must be one of: postgres' }) + }) + }) + it('formats setting category labels', () => { expect(toCategoryLabel('payments_processors')).to.equal('Payments Processors') expect(toCategoryLabel('rate_limiter')).to.equal('Rate Limiter') From 59d8f33f30038e05aea44ddbe6cc877b6135d97f Mon Sep 17 00:00:00 2001 From: anshumancanrock Date: Fri, 9 Oct 2026 20:41:16 +0530 Subject: [PATCH 2/4] refactor: build every event repository through one factory --- .../get-admission-check-controller-factory.ts | 4 +- .../post-invoice-controller-factory.ts | 4 +- .../dvm-orchestrator-worker-factory.ts | 4 +- src/factories/event-repository-factory.ts | 25 +++++ src/factories/maintenance-service-factory.ts | 4 +- src/factories/payments-service-factory.ts | 4 +- src/factories/relay-monitor-worker-factory.ts | 4 +- .../static-mirroring.worker-factory.ts | 4 +- src/factories/worker-factory.ts | 4 +- src/import-events.ts | 4 +- .../event-repository-factory.spec.ts | 94 +++++++++++++++++++ 11 files changed, 137 insertions(+), 18 deletions(-) create mode 100644 src/factories/event-repository-factory.ts create mode 100644 test/unit/factories/event-repository-factory.spec.ts diff --git a/src/factories/controllers/get-admission-check-controller-factory.ts b/src/factories/controllers/get-admission-check-controller-factory.ts index 08703428..d12463bd 100644 --- a/src/factories/controllers/get-admission-check-controller-factory.ts +++ b/src/factories/controllers/get-admission-check-controller-factory.ts @@ -1,6 +1,6 @@ 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' @@ -8,7 +8,7 @@ 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) diff --git a/src/factories/controllers/post-invoice-controller-factory.ts b/src/factories/controllers/post-invoice-controller-factory.ts index 966d3027..9985f982 100644 --- a/src/factories/controllers/post-invoice-controller-factory.ts +++ b/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() diff --git a/src/factories/dvm-orchestrator-worker-factory.ts b/src/factories/dvm-orchestrator-worker-factory.ts index 1ca5aa69..1a530b69 100644 --- a/src/factories/dvm-orchestrator-worker-factory.ts +++ b/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) } diff --git a/src/factories/event-repository-factory.ts b/src/factories/event-repository-factory.ts new file mode 100644 index 00000000..07d9da65 --- /dev/null +++ b/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}"`) + } + } +} diff --git a/src/factories/maintenance-service-factory.ts b/src/factories/maintenance-service-factory.ts index bb7af596..7028a9e1 100644 --- a/src/factories/maintenance-service-factory.ts +++ b/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) } diff --git a/src/factories/payments-service-factory.ts b/src/factories/payments-service-factory.ts index 3ee22e74..c16ebb16 100644 --- a/src/factories/payments-service-factory.ts +++ b/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() diff --git a/src/factories/relay-monitor-worker-factory.ts b/src/factories/relay-monitor-worker-factory.ts index 403bb434..d2f0afe2 100644 --- a/src/factories/relay-monitor-worker-factory.ts +++ b/src/factories/relay-monitor-worker-factory.ts @@ -2,8 +2,8 @@ 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' @@ -11,7 +11,7 @@ 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) diff --git a/src/factories/static-mirroring.worker-factory.ts b/src/factories/static-mirroring.worker-factory.ts index 85df95d4..5516848c 100644 --- a/src/factories/static-mirroring.worker-factory.ts +++ b/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) diff --git a/src/factories/worker-factory.ts b/src/factories/worker-factory.ts index 131a6627..0049188e 100644 --- a/src/factories/worker-factory.ts +++ b/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) diff --git a/src/import-events.ts b/src/import-events.ts index 8288bd7f..df2de128 100644 --- a/src/import-events.ts +++ b/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 diff --git a/test/unit/factories/event-repository-factory.spec.ts b/test/unit/factories/event-repository-factory.spec.ts new file mode 100644 index 00000000..880441a4 --- /dev/null +++ b/test/unit/factories/event-repository-factory.spec.ts @@ -0,0 +1,94 @@ +import { expect } from 'chai' +import cluster from 'cluster' +import fs from 'fs' +import knex from 'knex' +import path from 'path' +import Sinon from 'sinon' + +import { createEventRepository } from '../../../src/factories/event-repository-factory' +import { DatabaseClient } from '../../../src/@types/base' +import { EVENT_STORE_BACKEND_ENV } from '../../../src/utils/event-store' +import { EventRepository } from '../../../src/repositories/event-repository' +import { Settings } from '../../../src/@types/settings' +import { SettingsStatic } from '../../../src/utils/settings' + +describe('createEventRepository', () => { + let dbClient: DatabaseClient + let rrDbClient: DatabaseClient + let inheritedBackend: string | undefined + + beforeEach(() => { + dbClient = knex({ client: 'pg' }) + rrDbClient = knex({ client: 'pg' }) + inheritedBackend = process.env[EVENT_STORE_BACKEND_ENV] + }) + + afterEach(async () => { + Sinon.restore() + if (inheritedBackend === undefined) { + delete process.env[EVENT_STORE_BACKEND_ENV] + } else { + process.env[EVENT_STORE_BACKEND_ENV] = inheritedBackend + } + await dbClient.destroy() + await rrDbClient.destroy() + }) + + it('returns a PostgreSQL event repository when eventStore is unset', () => { + const repository = createEventRepository(dbClient, rrDbClient, () => ({}) as Settings) + + expect(repository).to.be.an.instanceOf(EventRepository) + }) + + it('returns a PostgreSQL event repository when eventStore.backend is postgres', () => { + const settings = () => ({ eventStore: { backend: 'postgres' } }) as Settings + + expect(createEventRepository(dbClient, rrDbClient, settings)).to.be.an.instanceOf(EventRepository) + }) + + it('hands the repository the same settings source', () => { + const search = [{ search: 'nostr' }] + const withSearch = createEventRepository(dbClient, rrDbClient, () => ({ nip50: { enabled: true } }) as Settings) + const withoutSearch = createEventRepository(dbClient, rrDbClient, () => ({ nip50: { enabled: false } }) as Settings) + + expect(withSearch.findByFilters(search).toString()).to.include('plainto_tsquery') + expect(withoutSearch.findByFilters(search).toString()).not.to.include('plainto_tsquery') + }) + + it('throws on an unsupported backend instead of falling back to postgres', () => { + const settings = () => ({ eventStore: { backend: 'mongodb' } }) as unknown as Settings + + expect(() => createEventRepository(dbClient, rrDbClient, settings)).to.throw( + Error, + 'Unsupported eventStore.backend "mongodb"', + ) + }) + + it('reads the relay settings when no settings source is given', () => { + const createSettingsStub = Sinon.stub(SettingsStatic, 'createSettings').returns({ + eventStore: { backend: 'mongodb' }, + } as unknown as Settings) + + expect(() => createEventRepository(dbClient, rrDbClient)).to.throw(Error, 'Unsupported eventStore.backend') + expect(createSettingsStub.called).to.equal(true) + }) + + it('in a worker, builds the store the primary chose at startup rather than the current setting', () => { + Sinon.stub(cluster, 'isWorker').value(true) + process.env[EVENT_STORE_BACKEND_ENV] = 'postgres' + const editedSettings = () => ({ eventStore: { backend: 'mongodb' } }) as unknown as Settings + + expect(createEventRepository(dbClient, rrDbClient, editedSettings)).to.be.an.instanceOf(EventRepository) + }) + + it('is the only place an EventRepository is constructed', () => { + const srcDir = path.join(process.cwd(), 'src') + const factoryFile = path.join('factories', 'event-repository-factory.ts') + + const offenders = (fs.readdirSync(srcDir, { recursive: true }) as string[]) + .filter((file) => file.endsWith('.ts') && file !== factoryFile) + .filter((file) => /new\s+EventRepository\s*\(/.test(fs.readFileSync(path.join(srcDir, file), 'utf-8'))) + + expect(offenders, 'construct event repositories through createEventRepository').to.deep.equal([]) + }) +}) From 028c1e371e2e9791eb4938629282c3ca4a24993b Mon Sep 17 00:00:00 2001 From: anshumancanrock Date: Fri, 9 Oct 2026 20:42:36 +0530 Subject: [PATCH 3/4] feat(app): pick the event store once at startup and hand it to workers --- src/app/app.ts | 18 +++++++++-- test/unit/app/app.spec.ts | 64 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 80 insertions(+), 2 deletions(-) diff --git a/src/app/app.ts b/src/app/app.ts index 8739077d..b3e505eb 100644 --- a/src/app/app.ts +++ b/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) => { - 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++) { diff --git a/test/unit/app/app.spec.ts b/test/unit/app/app.spec.ts index 5daa30b2..356312de 100644 --- a/test/unit/app/app.spec.ts +++ b/test/unit/app/app.spec.ts @@ -6,7 +6,10 @@ import sinonChai from 'sinon-chai' import { App } from '../../../src/app/app' import * as metricsTelemetry from '../../../src/telemetry/metrics' +import * as operatorNotificationEnqueue from '../../../src/utils/operator-notification-enqueue' +import * as torClient from '../../../src/tor/client' import { Settings } from '../../../src/@types/settings' +import { SettingsStatic } from '../../../src/utils/settings' chai.use(sinonChai) @@ -114,4 +117,65 @@ describe('App', () => { expect(fakeProcess.exit).to.have.been.calledOnceWithExactly(0) }) + + it('exits before forking any worker when eventStore.backend is unsupported', () => { + sandbox.stub(SettingsStatic, 'watchSettings').returns([]) + const app = new App( + fakeProcess as any, + cluster as any, + sandbox.stub().returns({ + payments: { enabled: false }, + workers: { count: 1 }, + eventStore: { backend: 'mongodb' }, + } as unknown as Settings), + ) + + app.run() + + expect(fakeProcess.exit).to.have.been.calledOnceWithExactly(1) + expect(cluster.fork).not.to.have.been.called + }) + + it('hands every worker, re-forked ones included, the event store resolved at startup', async () => { + // run() reads both from process.env + const env = { ...process.env } + delete env.WORKER_COUNT + delete env.RELAY_BROADCAST_FANOUT + sandbox.stub(process, 'env').value(env) + sandbox.useFakeTimers() + sandbox.stub(SettingsStatic, 'watchSettings').returns([]) + sandbox.stub(operatorNotificationEnqueue, 'enqueueOperatorNotification').resolves() + sandbox.stub(torClient, 'addOnion').rejects(new Error('tor is not configured')) + let pid = 1001 + cluster.fork.callsFake(() => createWorker(String(pid), pid++)) + const app = new App( + fakeProcess as any, + cluster as any, + sandbox.stub().returns({ + payments: { enabled: false }, + workers: { count: 1 }, + eventStore: { backend: 'postgres' }, + } as Settings), + ) + + app.run() + + expect(cluster.fork).to.have.been.calledTwice + expect(cluster.fork.firstCall.args[0]).to.deep.equal({ + WORKER_TYPE: 'worker', + WORKER_INDEX: '0', + EVENT_STORE_BACKEND: 'postgres', + }) + expect(cluster.fork.secondCall.args[0]).to.deep.equal({ + WORKER_TYPE: 'maintenance', + EVENT_STORE_BACKEND: 'postgres', + }) + + const exitHandlers = cluster.on.getCalls().filter((call) => call.args[0] === 'exit') + exitHandlers[exitHandlers.length - 1].args[1](cluster.fork.firstCall.returnValue, 1, null) + await sandbox.clock.tickAsync(10_000) + + expect(cluster.fork).to.have.been.calledThrice + expect(cluster.fork.thirdCall.args[0]).to.deep.equal(cluster.fork.firstCall.args[0]) + }) }) From f30021ef1d262e77ef0f512eadd392fb055b80a0 Mon Sep 17 00:00:00 2001 From: anshumancanrock Date: Fri, 9 Oct 2026 20:43:57 +0530 Subject: [PATCH 4/4] chore: add changeset for eventStore.backend --- .changeset/event-store-backend-setting.md | 7 +++++++ 1 file changed, 7 insertions(+) create mode 100644 .changeset/event-store-backend-setting.md diff --git a/.changeset/event-store-backend-setting.md b/.changeset/event-store-backend-setting.md new file mode 100644 index 00000000..da7b205b --- /dev/null +++ b/.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.