From 2310e24b16af6ab3ce6473c850a332c17247e3b3 Mon Sep 17 00:00:00 2001 From: ABHAY PANDEY Date: Sat, 19 Sep 2026 22:25:45 +0530 Subject: [PATCH 1/3] feat(admin): operator notification backend with outbox, channels, and delivery log --- ...120000_create_notification_outbox_table.js | 24 +++ ..._create_notification_delivery_log_table.js | 21 +++ resources/default-settings.yaml | 13 ++ src/@types/notification-outbox.ts | 40 +++++ src/@types/operator-notifications.ts | 75 ++++++++++ src/@types/repositories.ts | 40 +++++ src/@types/services.ts | 23 +++ src/@types/settings.ts | 2 + src/app/app.ts | 8 + src/app/maintenance-worker.ts | 37 ++++- ...et-notification-delivery-log-controller.ts | 27 ++++ .../admin/patch-settings-controller.ts | 19 +++ .../post-notification-test-controller.ts | 32 ++++ ...ication-delivery-log-controller-factory.ts | 8 + ...patch-admin-settings-controller-factory.ts | 5 +- ...in-notification-test-controller-factory.ts | 7 + src/factories/maintenance-worker-factory.ts | 8 + .../notification-outbox-service-factory.ts | 21 +++ src/factories/payments-service-factory.ts | 3 + .../notification-delivery-log-repository.ts | 72 +++++++++ .../notification-outbox-repository.ts | 133 +++++++++++++++++ src/routes/admin/index.ts | 16 ++ src/services/notification-channels.ts | 134 +++++++++++++++++ src/services/notification-outbox-service.ts | 48 ++++++ src/services/operator-notification-service.ts | 139 ++++++++++++++++++ src/services/payments-service.ts | 27 +++- src/utils/operator-notification-enqueue.ts | 31 ++++ src/utils/settings-config.ts | 55 +++++++ src/utils/settings-guided-schema.ts | 5 + src/utils/settings-redaction.ts | 2 +- test/unit/app/maintenance-worker.spec.ts | 20 ++- test/unit/routes/admin-settings.spec.ts | 20 +++ .../notification-outbox-service.spec.ts | 79 ++++++++++ .../operator-notification-service.spec.ts | 85 +++++++++++ test/unit/services/payments-service.spec.ts | 25 ++++ 35 files changed, 1298 insertions(+), 6 deletions(-) create mode 100644 migrations/20260919_120000_create_notification_outbox_table.js create mode 100644 migrations/20260919_130000_create_notification_delivery_log_table.js create mode 100644 src/@types/notification-outbox.ts create mode 100644 src/@types/operator-notifications.ts create mode 100644 src/controllers/admin/get-notification-delivery-log-controller.ts create mode 100644 src/controllers/admin/post-notification-test-controller.ts create mode 100644 src/factories/controllers/get-admin-notification-delivery-log-controller-factory.ts create mode 100644 src/factories/controllers/post-admin-notification-test-controller-factory.ts create mode 100644 src/factories/notification-outbox-service-factory.ts create mode 100644 src/repositories/notification-delivery-log-repository.ts create mode 100644 src/repositories/notification-outbox-repository.ts create mode 100644 src/services/notification-channels.ts create mode 100644 src/services/notification-outbox-service.ts create mode 100644 src/services/operator-notification-service.ts create mode 100644 src/utils/operator-notification-enqueue.ts create mode 100644 test/unit/services/notification-outbox-service.spec.ts create mode 100644 test/unit/services/operator-notification-service.spec.ts diff --git a/migrations/20260919_120000_create_notification_outbox_table.js b/migrations/20260919_120000_create_notification_outbox_table.js new file mode 100644 index 000000000..feb26712d --- /dev/null +++ b/migrations/20260919_120000_create_notification_outbox_table.js @@ -0,0 +1,24 @@ +exports.up = function (knex) { + return knex.schema.createTable('notification_outbox', (table) => { + table.uuid('id').primary().defaultTo(knex.raw('uuid_generate_v4()')) + table.text('event_type').notNullable() + table.jsonb('payload').notNullable() + table + .enum('status', ['pending', 'processing', 'delivered', 'dead']) + .notNullable() + .defaultTo('pending') + table.integer('attempt_count').unsigned().notNullable().defaultTo(0) + table.timestamp('available_at', { useTz: true }).notNullable().defaultTo(knex.fn.now()) + table.text('last_error').nullable() + table.timestamp('delivered_at', { useTz: true }).nullable() + table.timestamp('created_at', { useTz: true }).notNullable().defaultTo(knex.fn.now()) + table.timestamp('updated_at', { useTz: true }).notNullable().defaultTo(knex.fn.now()) + + table.index(['status', 'available_at', 'created_at'], 'idx_notification_outbox_dispatch') + table.index(['event_type', 'created_at'], 'idx_notification_outbox_event_type_created_at') + }) +} + +exports.down = function (knex) { + return knex.schema.dropTable('notification_outbox') +} diff --git a/migrations/20260919_130000_create_notification_delivery_log_table.js b/migrations/20260919_130000_create_notification_delivery_log_table.js new file mode 100644 index 000000000..db3e6807d --- /dev/null +++ b/migrations/20260919_130000_create_notification_delivery_log_table.js @@ -0,0 +1,21 @@ +exports.up = function (knex) { + return knex.schema.createTable('notification_delivery_log', (table) => { + table.uuid('id').primary().defaultTo(knex.raw('uuid_generate_v4()')) + table.uuid('outbox_id').nullable() + table.text('event_type').notNullable() + table.text('target_id').notNullable() + table.enum('target_type', ['http', 'discord', 'slack', 'telegram']).notNullable() + table.enum('status', ['success', 'failed']).notNullable() + table.integer('attempt_number').unsigned().notNullable() + table.text('error_snippet').nullable() + table.timestamp('created_at', { useTz: true }).notNullable().defaultTo(knex.fn.now()) + + table.index(['created_at'], 'idx_notification_delivery_log_created_at') + table.index(['event_type', 'created_at'], 'idx_notification_delivery_log_event_created_at') + table.foreign('outbox_id').references('id').inTable('notification_outbox').onDelete('SET NULL') + }) +} + +exports.down = function (knex) { + return knex.schema.dropTable('notification_delivery_log') +} diff --git a/resources/default-settings.yaml b/resources/default-settings.yaml index 3a8e63e2c..973a1417b 100755 --- a/resources/default-settings.yaml +++ b/resources/default-settings.yaml @@ -340,3 +340,16 @@ admin: enabled: false allowedPubkeys: [] maxSkewSeconds: 60 + notifications: + enabled: false + targets: [] + events: + admission.invoice.created: true + admission.invoice.paid: true + admission.invoice.failed: true + settings.changed: true + relay.restarted: false + retry: + maxAttempts: 5 + baseDelayMs: 1000 + deliveryLogRetentionDays: 30 diff --git a/src/@types/notification-outbox.ts b/src/@types/notification-outbox.ts new file mode 100644 index 000000000..b02ae89b3 --- /dev/null +++ b/src/@types/notification-outbox.ts @@ -0,0 +1,40 @@ +export enum NotificationOutboxStatus { + PENDING = 'pending', + PROCESSING = 'processing', + DELIVERED = 'delivered', + DEAD = 'dead', +} + +/** @deprecated Use OperatorNotificationEventType from operator-notifications.ts */ +export enum NotificationOutboxEventType { + OPERATOR_SETTINGS_UPDATED = 'settings.changed', + OPERATOR_INVOICE_PAID = 'admission.invoice.paid', +} + +export type NotificationOutboxPayload = Record + +export interface NotificationOutboxMessage { + id: string + eventType: NotificationOutboxEventType | string + payload: NotificationOutboxPayload + status: NotificationOutboxStatus + attemptCount: number + availableAt: Date + lastError: string | null + deliveredAt: Date | null + createdAt: Date + updatedAt: Date +} + +export interface DBNotificationOutboxMessage { + id: string + event_type: string + payload: NotificationOutboxPayload + status: NotificationOutboxStatus + attempt_count: number + available_at: Date + last_error: string | null + delivered_at: Date | null + created_at: Date + updated_at: Date +} diff --git a/src/@types/operator-notifications.ts b/src/@types/operator-notifications.ts new file mode 100644 index 000000000..037963b82 --- /dev/null +++ b/src/@types/operator-notifications.ts @@ -0,0 +1,75 @@ +export enum OperatorNotificationEventType { + ADMISSION_INVOICE_CREATED = 'admission.invoice.created', + ADMISSION_INVOICE_PAID = 'admission.invoice.paid', + ADMISSION_INVOICE_FAILED = 'admission.invoice.failed', + SETTINGS_CHANGED = 'settings.changed', + RELAY_RESTARTED = 'relay.restarted', +} + +export type OperatorNotificationChannelType = 'http' | 'discord' | 'slack' | 'telegram' + +export interface OperatorNotificationTarget { + id: string + type: OperatorNotificationChannelType + enabled: boolean + url?: string + botToken?: string + chatId?: string +} + +export interface OperatorNotificationEventToggles { + 'admission.invoice.created'?: boolean + 'admission.invoice.paid'?: boolean + 'admission.invoice.failed'?: boolean + 'settings.changed'?: boolean + 'relay.restarted'?: boolean +} + +export interface OperatorNotificationRetrySettings { + maxAttempts: number + baseDelayMs: number +} + +export interface AdminNotificationsSettings { + enabled: boolean + targets: OperatorNotificationTarget[] + events: OperatorNotificationEventToggles + retry: OperatorNotificationRetrySettings + deliveryLogRetentionDays?: number +} + +export interface OperatorNotificationEnvelope { + event: string + relay: string + timestamp: string + data: Record +} + +export enum NotificationDeliveryStatus { + SUCCESS = 'success', + FAILED = 'failed', +} + +export interface NotificationDeliveryLogEntry { + id: string + outboxId: string | null + eventType: string + targetId: string + targetType: OperatorNotificationChannelType + status: NotificationDeliveryStatus + attemptNumber: number + errorSnippet: string | null + createdAt: Date +} + +export interface DBNotificationDeliveryLogEntry { + id: string + outbox_id: string | null + event_type: string + target_id: string + target_type: OperatorNotificationChannelType + status: NotificationDeliveryStatus + attempt_number: number + error_snippet: string | null + created_at: Date +} diff --git a/src/@types/repositories.ts b/src/@types/repositories.ts index d17257f51..922796ee9 100644 --- a/src/@types/repositories.ts +++ b/src/@types/repositories.ts @@ -6,6 +6,12 @@ import { DBEvent, Event } from './event' import { CreateInviteCodeOptions, InviteCode } from './invite-code' import { Invoice } from './invoice' import { Nip05Verification } from './nip05' +import { + NotificationDeliveryLogEntry, + NotificationDeliveryStatus, + OperatorNotificationChannelType, +} from './operator-notifications' +import { NotificationOutboxMessage, NotificationOutboxPayload } from './notification-outbox' import { Report } from './report' import { EventKindsRange } from './settings' import { SubscriptionFilter } from './subscription' @@ -90,3 +96,37 @@ export interface IReportRepository { findByEventId(eventId: EventId): Promise findActionable(limit?: number): Promise } + +export interface INotificationDeliveryLogRepository { + append( + entry: { + outboxId: string | null + eventType: string + targetId: string + targetType: OperatorNotificationChannelType + status: NotificationDeliveryStatus + attemptNumber: number + errorSnippet: string | null + }, + client?: DatabaseClient, + ): Promise + findRecent(limit?: number, client?: DatabaseClient): Promise + deleteOlderThan(cutoff: Date, client?: DatabaseClient): Promise +} + +export interface INotificationOutboxRepository { + enqueue( + eventType: string, + payload: NotificationOutboxPayload, + client?: DatabaseClient, + ): Promise + claimBatch(limit: number, client?: DatabaseClient): Promise + markDelivered(id: string, client?: DatabaseClient): Promise + markFailed( + id: string, + error: string, + attemptCount: number, + maxAttempts: number, + client?: DatabaseClient, + ): Promise +} diff --git a/src/@types/services.ts b/src/@types/services.ts index c20808cf8..91dd14796 100644 --- a/src/@types/services.ts +++ b/src/@types/services.ts @@ -1,5 +1,6 @@ import { Invoice } from './invoice' import { Pubkey } from './base' +import { NotificationOutboxPayload } from './notification-outbox' export interface IMaintenanceService { clearOldEvents(): Promise @@ -27,3 +28,25 @@ export interface IPaymentsService { sendInvoiceUpdateNotification(invoice: Invoice): Promise getPendingInvoices(offset?: number): Promise } + +export interface NotificationDispatchContext { + outboxId?: string + attemptNumber?: number +} + +export interface INotificationDispatcher { + dispatch( + eventType: string, + payload: NotificationOutboxPayload, + context?: NotificationDispatchContext, + ): Promise +} + +export interface IOperatorNotificationService extends INotificationDispatcher { + dispatchTestTarget(targetId: string): Promise + getMaxAttempts(): number +} + +export interface INotificationOutboxService { + processBatch(limit?: number): Promise +} diff --git a/src/@types/settings.ts b/src/@types/settings.ts index c104ab1a1..3ee1f51d0 100644 --- a/src/@types/settings.ts +++ b/src/@types/settings.ts @@ -1,6 +1,7 @@ import { EventKinds } from '../constants/base' import { Pubkey, Secret } from './base' import { MessageType } from './messages' +import { AdminNotificationsSettings } from './operator-notifications' import { SubscriptionFilter } from './subscription' export interface Info { @@ -363,6 +364,7 @@ export interface AdminSettings { passwordHash?: string sessionTtlSeconds?: number nip98?: AdminNip98Settings + notifications?: AdminNotificationsSettings } export interface WoTSettings { enabled: boolean diff --git a/src/app/app.ts b/src/app/app.ts index 15b84a46a..d1b81ad8a 100644 --- a/src/app/app.ts +++ b/src/app/app.ts @@ -11,6 +11,8 @@ import { Serializable } from 'child_process' import { Settings } from '../@types/settings' import { SettingsStatic } from '../utils/settings' import { shutdownMetricsTelemetry } from '../telemetry/metrics' +import { OperatorNotificationEventType } from '../@types/operator-notifications' +import { enqueueOperatorNotification } from '../utils/operator-notification-enqueue' import { getPrimaryShutdownDeadlineMs } from '../utils/shutdown-state' const logger = createLogger('app-primary') @@ -131,6 +133,12 @@ export class App implements IRunnable { logger('settings: %O', settings) + // Primary-only: one outbox event per process start (not per client worker). + void enqueueOperatorNotification(OperatorNotificationEventType.RELAY_RESTARTED, { + version: packageJson.version, + relayPort: port, + }) + const host = `${hostname()}:${port}` addOnion(torHiddenServicePort, host).then( (value) => { diff --git a/src/app/maintenance-worker.ts b/src/app/maintenance-worker.ts index 9d1dde24a..a16d31ed3 100644 --- a/src/app/maintenance-worker.ts +++ b/src/app/maintenance-worker.ts @@ -4,14 +4,19 @@ import { Nip05VerificationOutcome, verifyNip05Identifier, } from '../utils/nip05' -import { IMaintenanceService, IPaymentsService } from '../@types/services' +import { IMaintenanceService, INotificationOutboxService, IPaymentsService } from '../@types/services' import { mergeDeepLeft, path, pipe } from 'ramda' import { IRunnable } from '../@types/base' import { createLogger } from '../factories/logger-factory' import { PENDING_INVOICE_PAGE_SIZE } from '../services/payments-service' import { delayMs } from '../utils/misc' -import { INip05VerificationRepository } from '../@types/repositories' +import { OperatorNotificationEventType } from '../@types/operator-notifications' +import { + INip05VerificationRepository, + INotificationDeliveryLogRepository, + INotificationOutboxRepository, +} from '../@types/repositories' import { InvoiceStatus } from '../@types/invoice' import { isExpiredInvoice } from '../utils/invoice' import { Nip05Verification } from '../@types/nip05' @@ -86,6 +91,9 @@ export class MaintenanceWorker implements IRunnable { private readonly maintenanceService: IMaintenanceService, private readonly settings: () => Settings, private readonly nip05VerificationRepository: INip05VerificationRepository, + private readonly notificationOutboxService: INotificationOutboxService, + private readonly notificationDeliveryLogRepository: INotificationDeliveryLogRepository, + private readonly notificationOutboxRepository: INotificationOutboxRepository, ) { this.process .on('SIGINT', this.onExit.bind(this)) @@ -131,6 +139,7 @@ export class MaintenanceWorker implements IRunnable { const clearOldEventsPromise = this.clearOldEventsSafely() await this.processNip05Reverifications(currentSettings) + await this.processNotificationOutbox() if (!path(['payments', 'enabled'], currentSettings)) { await clearOldEventsPromise @@ -188,6 +197,15 @@ export class MaintenanceWorker implements IRunnable { id: invoice.id, status: InvoiceStatus.EXPIRED, }) + try { + await this.notificationOutboxRepository.enqueue(OperatorNotificationEventType.ADMISSION_INVOICE_FAILED, { + invoiceId: invoice.id, + pubkey: invoice.pubkey, + reason: 'expired', + }) + } catch (error) { + logger.error('Unable to enqueue admission.invoice.failed notification', error) + } successful++ continue } @@ -201,6 +219,21 @@ export class MaintenanceWorker implements IRunnable { await clearOldEventsPromise } + private async processNotificationOutbox(): Promise { + try { + const delivered = await this.notificationOutboxService.processBatch() + if (delivered > 0) { + logger('delivered %d notification outbox message(s)', delivered) + } + + const retentionDays = this.settings().admin?.notifications?.deliveryLogRetentionDays ?? 30 + const cutoff = new Date(Date.now() - retentionDays * 86_400_000) + await this.notificationDeliveryLogRepository.deleteOlderThan(cutoff) + } catch (error) { + logger.error('Unable to process notification outbox', error) + } + } + private async processNip05Reverifications(currentSettings: Settings): Promise { const nip05Settings = currentSettings.nip05 if (!nip05Settings || nip05Settings.mode === 'disabled') { diff --git a/src/controllers/admin/get-notification-delivery-log-controller.ts b/src/controllers/admin/get-notification-delivery-log-controller.ts new file mode 100644 index 000000000..26ba96386 --- /dev/null +++ b/src/controllers/admin/get-notification-delivery-log-controller.ts @@ -0,0 +1,27 @@ +import { Request, Response } from 'express' + +import { IController } from '../../@types/controllers' +import { INotificationDeliveryLogRepository } from '../../@types/repositories' + +export class GetAdminNotificationDeliveryLogController implements IController { + public constructor(private readonly deliveryLogRepository: INotificationDeliveryLogRepository) {} + + public async handleRequest(_request: Request, response: Response): Promise { + const limit = Math.min(Number(_request.query.limit) || 50, 200) + const entries = await this.deliveryLogRepository.findRecent(limit) + + response.status(200).setHeader('content-type', 'application/json').send({ + entries: entries.map((entry) => ({ + id: entry.id, + outboxId: entry.outboxId, + eventType: entry.eventType, + targetId: entry.targetId, + targetType: entry.targetType, + status: entry.status, + attemptNumber: entry.attemptNumber, + errorSnippet: entry.errorSnippet, + createdAt: entry.createdAt.toISOString(), + })), + }) + } +} diff --git a/src/controllers/admin/patch-settings-controller.ts b/src/controllers/admin/patch-settings-controller.ts index 7914d3b8e..804af2215 100644 --- a/src/controllers/admin/patch-settings-controller.ts +++ b/src/controllers/admin/patch-settings-controller.ts @@ -18,8 +18,16 @@ import { redactSettingsValue, } from '../../utils/settings-redaction' import { validateSchema } from '../../utils/validation' +import { INotificationOutboxRepository } from '../../@types/repositories' +import { OperatorNotificationEventType } from '../../@types/operator-notifications' +import { redactSettingsSecrets } from '../../utils/settings-redaction' +import { createLogger } from '../../factories/logger-factory' + +const logger = createLogger('patch-admin-settings-controller') export class PatchAdminSettingsController implements IController { + public constructor(private readonly notificationOutboxRepository: INotificationOutboxRepository) {} + public async handleRequest(request: Request, response: Response): Promise { const validation = validateSchema(adminSettingsPatchBodySchema)(request.body) if (validation.error) { @@ -73,6 +81,17 @@ export class PatchAdminSettingsController implements IController { remoteAddress: request.ip, }) + try { + await this.notificationOutboxRepository.enqueue(OperatorNotificationEventType.SETTINGS_CHANGED, { + changes: redactSettingsSecrets( + updatedChanges.map(({ path, value, reload }) => ({ path, value, reload })), + ), + remoteAddress: request.ip, + }) + } catch (error) { + logger.error('Unable to enqueue settings notification outbox event', error) + } + if (changes.length === 1 && !('changes' in validation.value)) { const [change] = updatedChanges response.status(200).setHeader('content-type', 'application/json').send({ ok: true, ...change }) diff --git a/src/controllers/admin/post-notification-test-controller.ts b/src/controllers/admin/post-notification-test-controller.ts new file mode 100644 index 000000000..30126f5fe --- /dev/null +++ b/src/controllers/admin/post-notification-test-controller.ts @@ -0,0 +1,32 @@ +import { Request, Response } from 'express' +import { z } from 'zod' + +import { IController } from '../../@types/controllers' +import { IOperatorNotificationService } from '../../@types/services' +import { validateSchema } from '../../utils/validation' + +const bodySchema = z + .object({ + targetId: z.string().min(1), + }) + .strict() + +export class PostAdminNotificationTestController implements IController { + public constructor(private readonly operatorNotificationService: IOperatorNotificationService) {} + + public async handleRequest(request: Request, response: Response): Promise { + const validation = validateSchema(bodySchema)(request.body) + if (validation.error) { + response.status(400).setHeader('content-type', 'application/json').send({ error: 'Invalid request' }) + return + } + + try { + await this.operatorNotificationService.dispatchTestTarget(validation.value.targetId) + response.status(200).setHeader('content-type', 'application/json').send({ ok: true }) + } catch (error) { + const message = error instanceof Error ? error.message : 'Delivery failed' + response.status(502).setHeader('content-type', 'application/json').send({ error: message }) + } + } +} diff --git a/src/factories/controllers/get-admin-notification-delivery-log-controller-factory.ts b/src/factories/controllers/get-admin-notification-delivery-log-controller-factory.ts new file mode 100644 index 000000000..fbfd79043 --- /dev/null +++ b/src/factories/controllers/get-admin-notification-delivery-log-controller-factory.ts @@ -0,0 +1,8 @@ +import { GetAdminNotificationDeliveryLogController } from '../../controllers/admin/get-notification-delivery-log-controller' +import { IController } from '../../@types/controllers' +import { getMasterDbClient } from '../../database/client' +import { NotificationDeliveryLogRepository } from '../../repositories/notification-delivery-log-repository' + +export const createGetAdminNotificationDeliveryLogController = (): IController => { + return new GetAdminNotificationDeliveryLogController(new NotificationDeliveryLogRepository(getMasterDbClient())) +} diff --git a/src/factories/controllers/patch-admin-settings-controller-factory.ts b/src/factories/controllers/patch-admin-settings-controller-factory.ts index 09838d5b0..7a17f16f6 100644 --- a/src/factories/controllers/patch-admin-settings-controller-factory.ts +++ b/src/factories/controllers/patch-admin-settings-controller-factory.ts @@ -1,6 +1,9 @@ import { PatchAdminSettingsController } from '../../controllers/admin/patch-settings-controller' import { IController } from '../../@types/controllers' +import { getMasterDbClient } from '../../database/client' +import { NotificationOutboxRepository } from '../../repositories/notification-outbox-repository' export const createPatchAdminSettingsController = (): IController => { - return new PatchAdminSettingsController() + const notificationOutboxRepository = new NotificationOutboxRepository(getMasterDbClient()) + return new PatchAdminSettingsController(notificationOutboxRepository) } diff --git a/src/factories/controllers/post-admin-notification-test-controller-factory.ts b/src/factories/controllers/post-admin-notification-test-controller-factory.ts new file mode 100644 index 000000000..9ff891146 --- /dev/null +++ b/src/factories/controllers/post-admin-notification-test-controller-factory.ts @@ -0,0 +1,7 @@ +import { PostAdminNotificationTestController } from '../../controllers/admin/post-notification-test-controller' +import { IController } from '../../@types/controllers' +import { createOperatorNotificationService } from '../notification-outbox-service-factory' + +export const createPostAdminNotificationTestController = (): IController => { + return new PostAdminNotificationTestController(createOperatorNotificationService()) +} diff --git a/src/factories/maintenance-worker-factory.ts b/src/factories/maintenance-worker-factory.ts index 94197b7cc..27fd53115 100644 --- a/src/factories/maintenance-worker-factory.ts +++ b/src/factories/maintenance-worker-factory.ts @@ -1,18 +1,26 @@ import { createMaintenanceService } from './maintenance-service-factory' +import { createNotificationOutboxService } from './notification-outbox-service-factory' import { createPaymentsService } from './payments-service-factory' import { createSettings } from './settings-factory' import { getMasterDbClient } from '../database/client' import { MaintenanceWorker } from '../app/maintenance-worker' +import { NotificationDeliveryLogRepository } from '../repositories/notification-delivery-log-repository' +import { NotificationOutboxRepository } from '../repositories/notification-outbox-repository' import { Nip05VerificationRepository } from '../repositories/nip05-verification-repository' export const maintenanceWorkerFactory = () => { const dbClient = getMasterDbClient() const nip05VerificationRepository = new Nip05VerificationRepository(dbClient) + const notificationDeliveryLogRepository = new NotificationDeliveryLogRepository(dbClient) + const notificationOutboxRepository = new NotificationOutboxRepository(dbClient) return new MaintenanceWorker( process, createPaymentsService(), createMaintenanceService(), createSettings, nip05VerificationRepository, + createNotificationOutboxService(), + notificationDeliveryLogRepository, + notificationOutboxRepository, ) } diff --git a/src/factories/notification-outbox-service-factory.ts b/src/factories/notification-outbox-service-factory.ts new file mode 100644 index 000000000..b9fe27e88 --- /dev/null +++ b/src/factories/notification-outbox-service-factory.ts @@ -0,0 +1,21 @@ +import { createSettings } from './settings-factory' +import { getMasterDbClient } from '../database/client' +import { NotificationDeliveryLogRepository } from '../repositories/notification-delivery-log-repository' +import { NotificationOutboxRepository } from '../repositories/notification-outbox-repository' +import { NotificationOutboxService } from '../services/notification-outbox-service' +import { OperatorNotificationService } from '../services/operator-notification-service' + +export const createOperatorNotificationService = () => { + const deliveryLogRepository = new NotificationDeliveryLogRepository(getMasterDbClient()) + return new OperatorNotificationService(createSettings, deliveryLogRepository) +} + +export const createNotificationOutboxService = () => { + const outboxRepository = new NotificationOutboxRepository(getMasterDbClient()) + const operatorNotificationService = createOperatorNotificationService() + return new NotificationOutboxService( + outboxRepository, + operatorNotificationService, + () => operatorNotificationService.getMaxAttempts(), + ) +} diff --git a/src/factories/payments-service-factory.ts b/src/factories/payments-service-factory.ts index bbf3e595d..3ee22e747 100644 --- a/src/factories/payments-service-factory.ts +++ b/src/factories/payments-service-factory.ts @@ -5,6 +5,7 @@ import { EventRepository } from '../repositories/event-repository' import { InvoiceRepository } from '../repositories/invoice-repository' import { PaymentsService } from '../services/payments-service' import { UserRepository } from '../repositories/user-repository' +import { NotificationOutboxRepository } from '../repositories/notification-outbox-repository' export const createPaymentsService = () => { const dbClient = getMasterDbClient() @@ -12,6 +13,7 @@ export const createPaymentsService = () => { const invoiceRepository = new InvoiceRepository(dbClient) const eventRepository = new EventRepository(dbClient, rrDbClient) const userRepository = new UserRepository(dbClient, eventRepository) + const notificationOutboxRepository = new NotificationOutboxRepository(dbClient) const paymentsProcessor = createPaymentsProcessor() return new PaymentsService( @@ -21,5 +23,6 @@ export const createPaymentsService = () => { invoiceRepository, eventRepository, createSettings, + notificationOutboxRepository, ) } diff --git a/src/repositories/notification-delivery-log-repository.ts b/src/repositories/notification-delivery-log-repository.ts new file mode 100644 index 000000000..d2cedf300 --- /dev/null +++ b/src/repositories/notification-delivery-log-repository.ts @@ -0,0 +1,72 @@ +import { DatabaseClient } from '../@types/base' +import { + DBNotificationDeliveryLogEntry, + NotificationDeliveryLogEntry, + NotificationDeliveryStatus, + OperatorNotificationChannelType, +} from '../@types/operator-notifications' +import { INotificationDeliveryLogRepository } from '../@types/repositories' +import { createLogger } from '../factories/logger-factory' + +const logger = createLogger('notification-delivery-log-repository') + +function fromDB(row: DBNotificationDeliveryLogEntry): NotificationDeliveryLogEntry { + return { + id: row.id, + outboxId: row.outbox_id, + eventType: row.event_type, + targetId: row.target_id, + targetType: row.target_type, + status: row.status, + attemptNumber: row.attempt_number, + errorSnippet: row.error_snippet, + createdAt: row.created_at, + } +} + +export class NotificationDeliveryLogRepository implements INotificationDeliveryLogRepository { + public constructor(private readonly dbClient: DatabaseClient) {} + + public async append( + entry: { + outboxId: string | null + eventType: string + targetId: string + targetType: OperatorNotificationChannelType + status: NotificationDeliveryStatus + attemptNumber: number + errorSnippet: string | null + }, + client: DatabaseClient = this.dbClient, + ): Promise { + logger('append delivery log %s %s %s', entry.eventType, entry.targetId, entry.status) + + await client('notification_delivery_log').insert({ + outbox_id: entry.outboxId, + event_type: entry.eventType, + target_id: entry.targetId, + target_type: entry.targetType, + status: entry.status, + attempt_number: entry.attemptNumber, + error_snippet: entry.errorSnippet, + created_at: new Date(), + }) + } + + public async findRecent(limit = 50, client: DatabaseClient = this.dbClient): Promise { + const rows = await client('notification_delivery_log') + .orderBy('created_at', 'desc') + .limit(limit) + .select('*') + + return rows.map(fromDB) + } + + public async deleteOlderThan(cutoff: Date, client: DatabaseClient = this.dbClient): Promise { + const result = await client('notification_delivery_log') + .where('created_at', '<', cutoff) + .delete() + + return typeof result === 'number' ? result : 0 + } +} diff --git a/src/repositories/notification-outbox-repository.ts b/src/repositories/notification-outbox-repository.ts new file mode 100644 index 000000000..8788f5a26 --- /dev/null +++ b/src/repositories/notification-outbox-repository.ts @@ -0,0 +1,133 @@ +import { randomUUID } from 'crypto' + +import { DatabaseClient } from '../@types/base' +import { + DBNotificationOutboxMessage, + NotificationOutboxMessage, + NotificationOutboxPayload, + NotificationOutboxStatus, +} from '../@types/notification-outbox' +import { INotificationOutboxRepository } from '../@types/repositories' +import { createLogger } from '../factories/logger-factory' + +const logger = createLogger('notification-outbox-repository') + +function fromDB(row: DBNotificationOutboxMessage): NotificationOutboxMessage { + return { + id: row.id, + eventType: row.event_type, + payload: row.payload, + status: row.status, + attemptCount: row.attempt_count, + availableAt: row.available_at, + lastError: row.last_error, + deliveredAt: row.delivered_at, + createdAt: row.created_at, + updatedAt: row.updated_at, + } +} + +export class NotificationOutboxRepository implements INotificationOutboxRepository { + public constructor(private readonly dbClient: DatabaseClient) {} + + public async enqueue( + eventType: string, + payload: NotificationOutboxPayload, + client: DatabaseClient = this.dbClient, + ): Promise { + logger('enqueue notification outbox event %s', eventType) + + const now = new Date() + const row: DBNotificationOutboxMessage = { + id: randomUUID(), + event_type: eventType, + payload, + status: NotificationOutboxStatus.PENDING, + attempt_count: 0, + available_at: now, + last_error: null, + delivered_at: null, + created_at: now, + updated_at: now, + } + + await client('notification_outbox').insert(row) + + return fromDB(row) + } + + public async claimBatch( + limit: number, + client: DatabaseClient = this.dbClient, + ): Promise { + logger('claim notification outbox batch (limit %d)', limit) + + return client.transaction(async (trx) => { + const rows = await trx('notification_outbox') + .where('status', NotificationOutboxStatus.PENDING) + .where('available_at', '<=', trx.fn.now()) + .orderBy('created_at', 'asc') + .limit(limit) + .forUpdate() + .skipLocked() + .select('*') + + if (!rows.length) { + return [] + } + + const now = new Date() + const ids = rows.map((row) => row.id) + + await trx('notification_outbox').whereIn('id', ids).update({ + status: NotificationOutboxStatus.PROCESSING, + updated_at: now, + }) + + return rows.map((row) => + fromDB({ + ...row, + status: NotificationOutboxStatus.PROCESSING, + updated_at: now, + }), + ) + }) + } + + public async markDelivered(id: string, client: DatabaseClient = this.dbClient): Promise { + logger('mark notification outbox delivered %s', id) + + const now = new Date() + await client('notification_outbox').where('id', id).update({ + status: NotificationOutboxStatus.DELIVERED, + delivered_at: now, + last_error: null, + updated_at: now, + }) + } + + public async markFailed( + id: string, + error: string, + attemptCount: number, + maxAttempts: number, + client: DatabaseClient = this.dbClient, + ): Promise { + logger('mark notification outbox failed %s (attempt %d)', id, attemptCount) + + const now = new Date() + const isDead = attemptCount >= maxAttempts + const backoffMs = Math.min(60_000, 1000 * 2 ** Math.max(0, attemptCount - 1)) + const availableAt = new Date(now.getTime() + backoffMs) + + await client('notification_outbox') + .where('id', id) + .update({ + status: isDead ? NotificationOutboxStatus.DEAD : NotificationOutboxStatus.PENDING, + attempt_count: attemptCount, + available_at: isDead ? now : availableAt, + last_error: error.slice(0, 2000), + updated_at: now, + }) + } +} diff --git a/src/routes/admin/index.ts b/src/routes/admin/index.ts index c52f149d4..b320c32dc 100644 --- a/src/routes/admin/index.ts +++ b/src/routes/admin/index.ts @@ -1,5 +1,6 @@ import express, { json, Router } from 'express' +import { createGetAdminNotificationDeliveryLogController } from '../../factories/controllers/get-admin-notification-delivery-log-controller-factory' import { createGetAdminHealthController } from '../../factories/controllers/get-admin-health-controller-factory' import { createGetAdminMetricsController } from '../../factories/controllers/get-admin-metrics-controller-factory' import { createGetAdminSessionController } from '../../factories/controllers/get-admin-session-controller-factory' @@ -7,6 +8,7 @@ import { createGetAdminSettingsBackupsController } from '../../factories/control import { createGetAdminSettingsController } from '../../factories/controllers/get-admin-settings-controller-factory' import { createGetAdminSettingsSchemaController } from '../../factories/controllers/get-admin-settings-schema-controller-factory' import { createPatchAdminSettingsController } from '../../factories/controllers/patch-admin-settings-controller-factory' +import { createPostAdminNotificationTestController } from '../../factories/controllers/post-admin-notification-test-controller-factory' import { createPostAdminLoginController } from '../../factories/controllers/post-admin-login-controller-factory' import { createPostAdminLogoutController } from '../../factories/controllers/post-admin-logout-controller-factory' import { createPostAdminSettingsRestoreController } from '../../factories/controllers/post-admin-settings-restore-controller-factory' @@ -97,5 +99,19 @@ router.post( adminAuthMiddleware, withAdminController(createPostAdminSettingsRestoreController), ) +router.get( + '/notifications/delivery-log', + adminRateLimitMiddleware, + adminAuthMiddleware, + withAdminController(createGetAdminNotificationDeliveryLogController), +) +router.post( + '/notifications/test', + adminRateLimitMiddleware, + adminAuthGateMiddleware, + adminJsonBodyMiddleware, + adminAuthMiddleware, + withAdminController(createPostAdminNotificationTestController), +) export default router diff --git a/src/services/notification-channels.ts b/src/services/notification-channels.ts new file mode 100644 index 000000000..dde8cc4cc --- /dev/null +++ b/src/services/notification-channels.ts @@ -0,0 +1,134 @@ +import axios from 'axios' + +import { + OperatorNotificationChannelType, + OperatorNotificationEnvelope, + OperatorNotificationTarget, +} from '../@types/operator-notifications' + +const REQUEST_TIMEOUT_MS = 15_000 + +export const deliverToTarget = async ( + target: OperatorNotificationTarget, + envelope: OperatorNotificationEnvelope, +): Promise => { + switch (target.type) { + case 'http': + return deliverHttp(target, envelope) + case 'discord': + return deliverDiscord(target, envelope) + case 'slack': + return deliverSlack(target, envelope) + case 'telegram': + return deliverTelegram(target, envelope) + default: + throw new Error(`Unsupported notification target type: ${target.type as string}`) + } +} + +const deliverHttp = async (target: OperatorNotificationTarget, envelope: OperatorNotificationEnvelope): Promise => { + if (!target.url) { + throw new Error('HTTP notification target requires url') + } + + await axios.post(target.url, envelope, { + timeout: REQUEST_TIMEOUT_MS, + headers: { 'content-type': 'application/json' }, + validateStatus: (status) => status >= 200 && status < 300, + }) +} + +const deliverDiscord = async ( + target: OperatorNotificationTarget, + envelope: OperatorNotificationEnvelope, +): Promise => { + if (!target.url) { + throw new Error('Discord notification target requires webhook url') + } + + const content = `[${envelope.event}] ${envelope.relay}\n\`\`\`json\n${JSON.stringify(envelope.data, null, 2).slice(0, 1800)}\n\`\`\`` + + await axios.post( + target.url, + { content }, + { + timeout: REQUEST_TIMEOUT_MS, + validateStatus: (status) => status >= 200 && status < 300, + }, + ) +} + +const deliverSlack = async (target: OperatorNotificationTarget, envelope: OperatorNotificationEnvelope): Promise => { + if (!target.url) { + throw new Error('Slack notification target requires webhook url') + } + + await axios.post( + target.url, + { + text: `[${envelope.event}] ${envelope.relay}`, + blocks: [ + { + type: 'section', + text: { + type: 'mrkdwn', + text: `*${envelope.event}* on \`${envelope.relay}\`\n\`\`\`${JSON.stringify(envelope.data).slice(0, 2800)}\`\`\``, + }, + }, + ], + }, + { + timeout: REQUEST_TIMEOUT_MS, + validateStatus: (status) => status >= 200 && status < 300, + }, + ) +} + +const deliverTelegram = async ( + target: OperatorNotificationTarget, + envelope: OperatorNotificationEnvelope, +): Promise => { + if (!target.botToken || !target.chatId) { + throw new Error('Telegram notification target requires botToken and chatId') + } + + const text = `[${envelope.event}] ${envelope.relay}\n${JSON.stringify(envelope.data).slice(0, 3500)}` + const url = `https://api.telegram.org/bot${target.botToken}/sendMessage` + + await axios.post( + url, + { + chat_id: target.chatId, + text, + disable_web_page_preview: true, + }, + { + timeout: REQUEST_TIMEOUT_MS, + validateStatus: (status) => status >= 200 && status < 300, + }, + ) +} + +export const validateTargetConfig = (target: OperatorNotificationTarget): string | undefined => { + if (!target.id?.trim()) { + return 'target id is required' + } + + if (target.type === 'http' || target.type === 'discord' || target.type === 'slack') { + if (!target.url?.trim()) { + return `${target.type} target requires url` + } + } + + if (target.type === 'telegram') { + if (!target.botToken?.trim() || !target.chatId?.trim()) { + return 'telegram target requires botToken and chatId' + } + } + + return undefined +} + +export const maskTargetForLog = (target: OperatorNotificationTarget): OperatorNotificationChannelType | string => { + return `${target.type}:${target.id}` +} diff --git a/src/services/notification-outbox-service.ts b/src/services/notification-outbox-service.ts new file mode 100644 index 000000000..170b71da8 --- /dev/null +++ b/src/services/notification-outbox-service.ts @@ -0,0 +1,48 @@ +import { INotificationOutboxRepository } from '../@types/repositories' +import { INotificationDispatcher, INotificationOutboxService } from '../@types/services' +import { createLogger } from '../factories/logger-factory' + +const logger = createLogger('notification-outbox-service') + +export const NOTIFICATION_OUTBOX_BATCH_SIZE = 20 +export const NOTIFICATION_OUTBOX_MAX_ATTEMPTS = 5 + +export class NotificationOutboxService implements INotificationOutboxService { + public constructor( + private readonly outboxRepository: INotificationOutboxRepository, + private readonly dispatcher: INotificationDispatcher, + private readonly maxAttempts: () => number = () => NOTIFICATION_OUTBOX_MAX_ATTEMPTS, + ) {} + + public async processBatch(limit = NOTIFICATION_OUTBOX_BATCH_SIZE): Promise { + const messages = await this.outboxRepository.claimBatch(limit) + + if (!messages.length) { + return 0 + } + + let delivered = 0 + + for (const message of messages) { + try { + await this.dispatcher.dispatch(message.eventType, message.payload, { + outboxId: message.id, + attemptNumber: message.attemptCount + 1, + }) + await this.outboxRepository.markDelivered(message.id) + delivered++ + } catch (error) { + const reason = error instanceof Error ? error.message : String(error) + logger.error('notification outbox delivery failed for %s: %s', message.id, reason) + await this.outboxRepository.markFailed( + message.id, + reason, + message.attemptCount + 1, + this.maxAttempts(), + ) + } + } + + return delivered + } +} diff --git a/src/services/operator-notification-service.ts b/src/services/operator-notification-service.ts new file mode 100644 index 000000000..80e414ea5 --- /dev/null +++ b/src/services/operator-notification-service.ts @@ -0,0 +1,139 @@ +import { Settings } from '../@types/settings' +import { + AdminNotificationsSettings, + NotificationDeliveryStatus, + OperatorNotificationEnvelope, + OperatorNotificationEventType, + OperatorNotificationTarget, +} from '../@types/operator-notifications' +import { NotificationOutboxPayload } from '../@types/notification-outbox' +import { INotificationDeliveryLogRepository } from '../@types/repositories' +import { INotificationDispatcher } from '../@types/services' +import { createLogger } from '../factories/logger-factory' +import { loadDefaults } from '../utils/settings-config' +import { deliverToTarget, maskTargetForLog } from './notification-channels' + +const logger = createLogger('operator-notification-service') + +export interface OperatorNotificationDispatchContext { + outboxId?: string + attemptNumber?: number +} + +export class OperatorNotificationService implements INotificationDispatcher { + public constructor( + private readonly settings: () => Settings, + private readonly deliveryLogRepository: INotificationDeliveryLogRepository, + ) {} + + public async dispatch( + eventType: string, + payload: NotificationOutboxPayload, + context: OperatorNotificationDispatchContext = {}, + ): Promise { + const config = this.getNotificationsConfig() + if (!config.enabled) { + return + } + + if (!this.isEventEnabled(config, eventType)) { + return + } + + const targets = config.targets.filter((target) => target.enabled) + if (!targets.length) { + return + } + + const envelope = this.buildEnvelope(eventType, payload) + const attemptNumber = context.attemptNumber ?? 1 + const failures: string[] = [] + + for (const target of targets) { + try { + await deliverToTarget(target, envelope) + await this.deliveryLogRepository.append({ + outboxId: context.outboxId ?? null, + eventType, + targetId: target.id, + targetType: target.type, + status: NotificationDeliveryStatus.SUCCESS, + attemptNumber, + errorSnippet: null, + }) + } catch (error) { + const message = error instanceof Error ? error.message : String(error) + logger.error('delivery failed for %s: %s', maskTargetForLog(target), message) + failures.push(`${target.id}: ${message}`) + await this.deliveryLogRepository.append({ + outboxId: context.outboxId ?? null, + eventType, + targetId: target.id, + targetType: target.type, + status: NotificationDeliveryStatus.FAILED, + attemptNumber, + errorSnippet: message.slice(0, 2000), + }) + } + } + + if (failures.length) { + throw new Error(failures.join('; ')) + } + } + + /** Sends a one-off test message to a single configured target (admin API). */ + public async dispatchTestTarget(targetId: string): Promise { + const config = this.getNotificationsConfig() + const target = config.targets.find((entry) => entry.id === targetId) + if (!target) { + throw new Error(`Unknown notification target: ${targetId}`) + } + + const envelope = this.buildEnvelope(OperatorNotificationEventType.RELAY_RESTARTED, { + test: true, + message: 'Operator notification test delivery', + }) + + await deliverToTarget(target, envelope) + } + + public getMaxAttempts(): number { + return this.getNotificationsConfig().retry.maxAttempts + } + + private getNotificationsConfig(): AdminNotificationsSettings { + const defaults = loadDefaults().admin?.notifications + const configured = this.settings().admin?.notifications + + return { + enabled: configured?.enabled ?? defaults?.enabled ?? false, + targets: configured?.targets ?? defaults?.targets ?? [], + events: { ...defaults?.events, ...configured?.events }, + retry: { + maxAttempts: configured?.retry?.maxAttempts ?? defaults?.retry?.maxAttempts ?? 5, + baseDelayMs: configured?.retry?.baseDelayMs ?? defaults?.retry?.baseDelayMs ?? 1000, + }, + deliveryLogRetentionDays: + configured?.deliveryLogRetentionDays ?? defaults?.deliveryLogRetentionDays ?? 30, + } + } + + private isEventEnabled(config: AdminNotificationsSettings, eventType: string): boolean { + const toggles = config.events as Record + return toggles[eventType] !== false + } + + private buildEnvelope(eventType: string, payload: NotificationOutboxPayload): OperatorNotificationEnvelope { + return { + event: eventType, + relay: this.settings().info.relay_url, + timestamp: new Date().toISOString(), + data: payload, + } + } +} + +export const listEnabledTargets = (targets: OperatorNotificationTarget[]): OperatorNotificationTarget[] => { + return targets.filter((target) => target.enabled) +} diff --git a/src/services/payments-service.ts b/src/services/payments-service.ts index e3f6e8439..a1fb91bb9 100644 --- a/src/services/payments-service.ts +++ b/src/services/payments-service.ts @@ -2,7 +2,8 @@ import { andThen, otherwise, pipe } from 'ramda' import { broadcastEvent, getPublicKey, getRelayPrivateKey, identifyEvent, signEvent } from '../utils/event' import { DatabaseClient, Pubkey } from '../@types/base' import { FeeSchedule, Settings } from '../@types/settings' -import { IEventRepository, IInvoiceRepository, IUserRepository } from '../@types/repositories' +import { IEventRepository, IInvoiceRepository, INotificationOutboxRepository, IUserRepository } from '../@types/repositories' +import { OperatorNotificationEventType } from '../@types/operator-notifications' import { Invoice, InvoiceStatus, InvoiceUnit } from '../@types/invoice' import { Event, ExpiringEvent, UnidentifiedEvent } from '../@types/event' @@ -26,6 +27,7 @@ export class PaymentsService implements IPaymentsService { private readonly invoiceRepository: IInvoiceRepository, private readonly eventRepository: IEventRepository, private readonly settings: () => Settings, + private readonly notificationOutboxRepository: INotificationOutboxRepository, ) {} public async getPendingInvoices(offset = 0): Promise { @@ -90,6 +92,17 @@ export class PaymentsService implements IPaymentsService { transaction.transaction, ) + await this.notificationOutboxRepository.enqueue( + OperatorNotificationEventType.ADMISSION_INVOICE_CREATED, + { + invoiceId: invoiceResponse.id, + pubkey, + amountRequested: invoiceResponse.amountRequested.toString(), + unit: invoiceResponse.unit, + }, + transaction.transaction, + ) + await transaction.commit() return { @@ -184,6 +197,18 @@ export class PaymentsService implements IPaymentsService { await this.userRepository.admitUser(invoice.pubkey, date, transaction.transaction) } + await this.notificationOutboxRepository.enqueue( + OperatorNotificationEventType.ADMISSION_INVOICE_PAID, + { + invoiceId: invoice.id, + pubkey: invoice.pubkey, + amountPaid: invoice.amountPaid.toString(), + unit: invoice.unit, + confirmedAt: invoice.confirmedAt.toISOString(), + }, + transaction.transaction, + ) + await transaction.commit() } catch (error) { logger.error('Unable to confirm invoice. Reason:', error) diff --git a/src/utils/operator-notification-enqueue.ts b/src/utils/operator-notification-enqueue.ts new file mode 100644 index 000000000..f570153aa --- /dev/null +++ b/src/utils/operator-notification-enqueue.ts @@ -0,0 +1,31 @@ +import { DatabaseClient } from '../@types/base' +import { NotificationOutboxPayload } from '../@types/notification-outbox' +import { OperatorNotificationEventType } from '../@types/operator-notifications' +import { INotificationOutboxRepository } from '../@types/repositories' +import { getMasterDbClient } from '../database/client' +import { createLogger } from '../factories/logger-factory' +import { NotificationOutboxRepository } from '../repositories/notification-outbox-repository' + +const logger = createLogger('operator-notification-enqueue') + +let outboxRepository: INotificationOutboxRepository | undefined + +const getOutboxRepository = (): INotificationOutboxRepository => { + if (!outboxRepository) { + outboxRepository = new NotificationOutboxRepository(getMasterDbClient()) + } + + return outboxRepository +} + +export const enqueueOperatorNotification = async ( + eventType: OperatorNotificationEventType | string, + payload: NotificationOutboxPayload, + client?: DatabaseClient, +): Promise => { + try { + await getOutboxRepository().enqueue(eventType, payload, client) + } catch (error) { + logger.error('Unable to enqueue operator notification %s', eventType, error) + } +} diff --git a/src/utils/settings-config.ts b/src/utils/settings-config.ts index c2131e968..b6c22b745 100644 --- a/src/utils/settings-config.ts +++ b/src/utils/settings-config.ts @@ -613,6 +613,61 @@ export const validateSettings = (settings: Settings): ValidationIssue[] => { } validateShape(loadDefaults(), settings, [], issues) + issues.push(...validateAdminNotifications(settings)) + + return issues +} + +const validateAdminNotifications = (settings: Settings): ValidationIssue[] => { + const issues: ValidationIssue[] = [] + const notifications = settings.admin?.notifications + if (!notifications) { + return issues + } + + if (notifications.retry?.maxAttempts !== undefined && notifications.retry.maxAttempts < 1) { + issues.push({ path: 'admin.notifications.retry.maxAttempts', message: 'maxAttempts must be at least 1' }) + } + + if (notifications.retry?.baseDelayMs !== undefined && notifications.retry.baseDelayMs < 0) { + issues.push({ path: 'admin.notifications.retry.baseDelayMs', message: 'baseDelayMs must be >= 0' }) + } + + if (notifications.deliveryLogRetentionDays !== undefined && notifications.deliveryLogRetentionDays < 1) { + issues.push({ + path: 'admin.notifications.deliveryLogRetentionDays', + message: 'deliveryLogRetentionDays must be at least 1', + }) + } + + const targetIds = new Set() + for (const [index, target] of (notifications.targets ?? []).entries()) { + const prefix = `admin.notifications.targets[${index}]` + if (!target.id?.trim()) { + issues.push({ path: `${prefix}.id`, message: 'target id is required' }) + } else if (targetIds.has(target.id)) { + issues.push({ path: `${prefix}.id`, message: 'target id must be unique' }) + } else { + targetIds.add(target.id) + } + + if (!['http', 'discord', 'slack', 'telegram'].includes(target.type)) { + issues.push({ path: `${prefix}.type`, message: 'type must be http, discord, slack, or telegram' }) + } + + if (target.type === 'telegram') { + if (!target.botToken?.trim()) { + issues.push({ path: `${prefix}.botToken`, message: 'botToken is required for telegram targets' }) + } + if (!target.chatId?.trim()) { + issues.push({ path: `${prefix}.chatId`, message: 'chatId is required for telegram targets' }) + } + } else if (target.type === 'http' || target.type === 'discord' || target.type === 'slack') { + if (!target.url?.trim()) { + issues.push({ path: `${prefix}.url`, message: 'url is required for webhook targets' }) + } + } + } return issues } diff --git a/src/utils/settings-guided-schema.ts b/src/utils/settings-guided-schema.ts index 054a130f7..bb6b65adf 100644 --- a/src/utils/settings-guided-schema.ts +++ b/src/utils/settings-guided-schema.ts @@ -262,6 +262,11 @@ export const guidedSettingCategories: GuidedSettingCategory[] = [ type: 'number', validate: requireSafeNonNegativeIntegerSettingValue, }, + { + label: 'Enable operator notifications', + path: 'admin.notifications.enabled', + type: 'boolean', + }, ], }, ] diff --git a/src/utils/settings-redaction.ts b/src/utils/settings-redaction.ts index 3546786de..05d4e77fe 100644 --- a/src/utils/settings-redaction.ts +++ b/src/utils/settings-redaction.ts @@ -1,4 +1,4 @@ -const SENSITIVE_SETTING_KEYS = new Set(['passwordHash', 'secret']) +const SENSITIVE_SETTING_KEYS = new Set(['passwordHash', 'secret', 'botToken']) const isPlainObject = (value: unknown): value is Record => { return typeof value === 'object' && value !== null && !Array.isArray(value) diff --git a/test/unit/app/maintenance-worker.spec.ts b/test/unit/app/maintenance-worker.spec.ts index c42fba4da..24e747000 100644 --- a/test/unit/app/maintenance-worker.spec.ts +++ b/test/unit/app/maintenance-worker.spec.ts @@ -7,7 +7,7 @@ import sinonChai from 'sinon-chai' import { InvoiceStatus, InvoiceUnit } from '../../../src/@types/invoice' import { Nip05Verification } from '../../../src/@types/nip05' -import { IMaintenanceService, IPaymentsService } from '../../../src/@types/services' +import { IMaintenanceService, INotificationOutboxService, IPaymentsService } from '../../../src/@types/services' import { Settings } from '../../../src/@types/settings' import { applyReverificationOutcome, MaintenanceWorker } from '../../../src/app/maintenance-worker' import * as metricsTelemetry from '../../../src/telemetry/metrics' @@ -28,6 +28,9 @@ describe('MaintenanceWorker', () => { let settings: Sinon.SinonStub let settingsState: Settings let nip05VerificationRepository: any + let notificationOutboxService: Sinon.SinonStubbedInstance + let notificationDeliveryLogRepository: any + let notificationOutboxRepository: any let verifyStub: Sinon.SinonStub const pendingInvoice = { @@ -91,6 +94,18 @@ describe('MaintenanceWorker', () => { clearOldEvents: sandbox.stub().resolves(), } as any + notificationOutboxService = { + processBatch: sandbox.stub().resolves(0), + } as any + + notificationDeliveryLogRepository = { + deleteOlderThan: sandbox.stub().resolves(0), + } + + notificationOutboxRepository = { + enqueue: sandbox.stub().resolves(), + } + // Prevent real timeouts and randomized per-invoice delays. sandbox.stub(misc, 'delayMs').resolves() @@ -100,6 +115,9 @@ describe('MaintenanceWorker', () => { maintenanceService, settings as any, nip05VerificationRepository, + notificationOutboxService, + notificationDeliveryLogRepository, + notificationOutboxRepository, ) }) diff --git a/test/unit/routes/admin-settings.spec.ts b/test/unit/routes/admin-settings.spec.ts index 04b95df65..2d4f1f085 100644 --- a/test/unit/routes/admin-settings.spec.ts +++ b/test/unit/routes/admin-settings.spec.ts @@ -9,6 +9,7 @@ import Sinon from 'sinon' import { hashAdminPassword } from '../../../src/utils/admin-password' import * as adminRateLimitMiddleware from '../../../src/handlers/request-handlers/admin-rate-limit-middleware' import * as rateLimiterMiddleware from '../../../src/handlers/request-handlers/rate-limiter-middleware' +import { PatchAdminSettingsController } from '../../../src/controllers/admin/patch-settings-controller' import * as settingsFactory from '../../../src/factories/settings-factory' import { getSettingsAuditLogPath, @@ -27,6 +28,7 @@ describe('admin settings API', () => { let rateLimiterMiddlewareStub: Sinon.SinonStub let adminRateLimitMiddlewareStub: Sinon.SinonStub let adminLoginRateLimitMiddlewareStub: Sinon.SinonStub + let createPatchAdminSettingsControllerStub: Sinon.SinonStub let server: any const loadAdminRouter = () => { @@ -50,6 +52,21 @@ describe('admin settings API', () => { adminLoginRateLimitMiddlewareStub = Sinon.stub(adminRateLimitMiddleware, 'adminLoginRateLimitMiddleware').callsFake( passthrough, ) + const patchFactoryPath = require.resolve( + '../../../src/factories/controllers/patch-admin-settings-controller-factory', + ) + delete require.cache[patchFactoryPath] + // eslint-disable-next-line @typescript-eslint/no-var-requires + const patchAdminSettingsControllerFactory = require(patchFactoryPath) + createPatchAdminSettingsControllerStub = Sinon.stub( + patchAdminSettingsControllerFactory, + 'createPatchAdminSettingsController', + ).callsFake( + () => + new PatchAdminSettingsController({ + enqueue: Sinon.stub().resolves(), + } as any), + ) const router = loadAdminRouter() const app = express() app.use('/admin', router) @@ -66,7 +83,10 @@ describe('admin settings API', () => { rateLimiterMiddlewareStub?.restore() adminRateLimitMiddlewareStub?.restore() adminLoginRateLimitMiddlewareStub?.restore() + createPatchAdminSettingsControllerStub?.restore() delete require.cache[require.resolve('../../../src/routes/admin/index')] + delete require.cache[require.resolve('../../../src/factories/controllers/patch-admin-settings-controller-factory')] + delete require.cache[require.resolve('../../../src/controllers/admin/patch-settings-controller')] delete require.cache[require.resolve('../../../src/routes/admin')] if (server) { diff --git a/test/unit/services/notification-outbox-service.spec.ts b/test/unit/services/notification-outbox-service.spec.ts new file mode 100644 index 000000000..7fc6d8956 --- /dev/null +++ b/test/unit/services/notification-outbox-service.spec.ts @@ -0,0 +1,79 @@ +import chai from 'chai' +import chaiAsPromised from 'chai-as-promised' +import Sinon from 'sinon' +import sinonChai from 'sinon-chai' + +import { NotificationOutboxEventType, NotificationOutboxStatus } from '../../../src/@types/notification-outbox' +import { NotificationOutboxService } from '../../../src/services/notification-outbox-service' + +chai.use(sinonChai) +chai.use(chaiAsPromised) + +const { expect } = chai + +describe('NotificationOutboxService', () => { + let sandbox: Sinon.SinonSandbox + let outboxRepository: any + let dispatcher: any + let service: NotificationOutboxService + + const message = { + id: 'msg-1', + eventType: NotificationOutboxEventType.OPERATOR_INVOICE_PAID, + payload: { invoiceId: 'inv-1' }, + status: NotificationOutboxStatus.PROCESSING, + attemptCount: 0, + availableAt: new Date(), + lastError: null, + deliveredAt: null, + createdAt: new Date(), + updatedAt: new Date(), + } + + beforeEach(() => { + sandbox = Sinon.createSandbox() + outboxRepository = { + claimBatch: sandbox.stub().resolves([]), + markDelivered: sandbox.stub().resolves(), + markFailed: sandbox.stub().resolves(), + } + dispatcher = { + dispatch: sandbox.stub().resolves(), + } + service = new NotificationOutboxService(outboxRepository, dispatcher) + }) + + afterEach(() => { + sandbox.restore() + }) + + it('returns 0 when the outbox is empty', async () => { + await expect(service.processBatch()).to.eventually.equal(0) + }) + + it('marks a message delivered after dispatch succeeds', async () => { + outboxRepository.claimBatch.resolves([message]) + + await expect(service.processBatch()).to.eventually.equal(1) + + expect(dispatcher.dispatch).to.have.been.calledOnceWith(message.eventType, message.payload, { + outboxId: message.id, + attemptNumber: 1, + }) + expect(outboxRepository.markDelivered).to.have.been.calledOnceWith(message.id) + }) + + it('schedules a retry when dispatch fails', async () => { + outboxRepository.claimBatch.resolves([message]) + dispatcher.dispatch.rejects(new Error('network down')) + + await expect(service.processBatch()).to.eventually.equal(0) + + expect(outboxRepository.markFailed).to.have.been.calledOnceWith( + message.id, + 'network down', + 1, + 5, + ) + }) +}) diff --git a/test/unit/services/operator-notification-service.spec.ts b/test/unit/services/operator-notification-service.spec.ts new file mode 100644 index 000000000..61e17cbed --- /dev/null +++ b/test/unit/services/operator-notification-service.spec.ts @@ -0,0 +1,85 @@ +import axios from 'axios' +import chai from 'chai' +import chaiAsPromised from 'chai-as-promised' +import Sinon from 'sinon' +import sinonChai from 'sinon-chai' + +import { OperatorNotificationEventType } from '../../../src/@types/operator-notifications' +import { OperatorNotificationService } from '../../../src/services/operator-notification-service' + +chai.use(sinonChai) +chai.use(chaiAsPromised) + +const { expect } = chai + +describe('OperatorNotificationService', () => { + let sandbox: Sinon.SinonSandbox + let deliveryLogRepository: any + let service: OperatorNotificationService + + beforeEach(() => { + sandbox = Sinon.createSandbox() + deliveryLogRepository = { + append: sandbox.stub().resolves(), + } + service = new OperatorNotificationService( + () => + ({ + info: { relay_url: 'wss://relay.example' }, + admin: { + notifications: { + enabled: true, + targets: [ + { + id: 'discord-main', + type: 'discord', + enabled: true, + url: 'https://discord.com/api/webhooks/test', + }, + ], + events: { + 'admission.invoice.paid': true, + }, + retry: { maxAttempts: 5, baseDelayMs: 1000 }, + }, + }, + }) as any, + deliveryLogRepository, + ) + sandbox.stub(axios, 'post').resolves({ status: 204, data: {} }) + }) + + afterEach(() => { + sandbox.restore() + }) + + it('skips dispatch when notifications are disabled', async () => { + service = new OperatorNotificationService( + () => + ({ + info: { relay_url: 'wss://relay.example' }, + admin: { notifications: { enabled: false, targets: [], events: {}, retry: { maxAttempts: 5, baseDelayMs: 1000 } } }, + }) as any, + deliveryLogRepository, + ) + + await service.dispatch(OperatorNotificationEventType.ADMISSION_INVOICE_PAID, { invoiceId: 'x' }) + + expect(axios.post).to.not.have.been.called + }) + + it('delivers to enabled targets and logs success', async () => { + await service.dispatch(OperatorNotificationEventType.ADMISSION_INVOICE_PAID, { invoiceId: 'inv-1' }, { outboxId: 'ob-1' }) + + expect(axios.post).to.have.been.calledOnce + expect(deliveryLogRepository.append).to.have.been.calledOnce + }) + + it('throws when a target delivery fails', async () => { + ;(axios.post as Sinon.SinonStub).rejects(new Error('network down')) + + await expect( + service.dispatch(OperatorNotificationEventType.ADMISSION_INVOICE_PAID, { invoiceId: 'inv-1' }), + ).to.be.rejectedWith('network down') + }) +}) diff --git a/test/unit/services/payments-service.spec.ts b/test/unit/services/payments-service.spec.ts index 4f5d8467a..c32fbc6fb 100644 --- a/test/unit/services/payments-service.spec.ts +++ b/test/unit/services/payments-service.spec.ts @@ -8,6 +8,7 @@ chai.use(chaiAsPromised) import * as eventUtils from '../../../src/utils/event' import { Invoice, InvoiceStatus, InvoiceUnit } from '../../../src/@types/invoice' +import { OperatorNotificationEventType } from '../../../src/@types/operator-notifications' import { PaymentsService } from '../../../src/services/payments-service' const { expect } = chai @@ -21,6 +22,7 @@ describe('PaymentsService', () => { let userRepository: any let invoiceRepository: any let eventRepository: any + let notificationOutboxRepository: any let settings: Sinon.SinonStub const stubInvoice = (overrides: Partial = {}): Invoice => ({ @@ -71,6 +73,10 @@ describe('PaymentsService', () => { create: sandbox.stub().resolves(), } + notificationOutboxRepository = { + enqueue: sandbox.stub().resolves(), + } + settings = sandbox.stub() // Stub module-level utilities used inside PaymentsService @@ -104,6 +110,7 @@ describe('PaymentsService', () => { invoiceRepository, eventRepository, settings, + notificationOutboxRepository, ) }) @@ -342,6 +349,24 @@ describe('PaymentsService', () => { ).to.be.rejectedWith('Unable to get transaction: transaction not started.') }) + it('enqueues operator invoice paid in the confirmation transaction', async () => { + const invoice = makeCompletedInvoice() + + await service.confirmInvoice(invoice) + + expect(notificationOutboxRepository.enqueue).to.have.been.calledOnceWithExactly( + OperatorNotificationEventType.ADMISSION_INVOICE_PAID, + { + invoiceId: invoice.id, + pubkey: invoice.pubkey, + amountPaid: invoice.amountPaid!.toString(), + unit: invoice.unit, + confirmedAt: invoice.confirmedAt!.toISOString(), + }, + mockTrx, + ) + }) + it('converts SATS to msats before comparing against the fee', async () => { // 2 sats = 2000 msats; fee = 1000 msats → should admit settings.returns(makeSettings([{ enabled: true, amount: 1000n }])) From e3800e7a903e31b1194f23776f0701474742727c Mon Sep 17 00:00:00 2001 From: ABHAY PANDEY Date: Sat, 19 Sep 2026 23:52:17 +0530 Subject: [PATCH 2/3] chore: add changeset for operator notification backend --- .changeset/operator-notification-backend.md | 7 +++++++ 1 file changed, 7 insertions(+) create mode 100644 .changeset/operator-notification-backend.md diff --git a/.changeset/operator-notification-backend.md b/.changeset/operator-notification-backend.md new file mode 100644 index 000000000..674346a18 --- /dev/null +++ b/.changeset/operator-notification-backend.md @@ -0,0 +1,7 @@ +--- +"nostream": minor +--- + +feat(admin): operator notification backend with Postgres outbox + +Adds transactional outbox dispatch for operator alerts (HTTP, Discord, Slack, Telegram), delivery log, `admin.notifications` settings, event hooks for admission invoices and settings changes, and admin test/history endpoints. Closes #759. From 0e1007ed141e0bb65af595597f1813d4c517c5d6 Mon Sep 17 00:00:00 2001 From: ABHAY PANDEY Date: Sun, 20 Sep 2026 07:55:30 +0530 Subject: [PATCH 3/3] fix(admin): address notification outbox PR review --- src/@types/repositories.ts | 3 + src/@types/services.ts | 1 + src/app/maintenance-worker.ts | 1 + ...et-notification-delivery-log-controller.ts | 13 +++- .../notification-outbox-service-factory.ts | 1 + .../notification-delivery-log-repository.ts | 11 +++ .../notification-outbox-repository.ts | 37 +++++++++- src/routes/admin/index.ts | 2 +- src/services/notification-outbox-service.ts | 2 + src/services/operator-notification-service.ts | 16 +++- src/utils/settings-config.ts | 39 ++++++++-- src/utils/settings-redaction.ts | 20 +++-- test/unit/app/maintenance-worker.spec.ts | 1 + .../services/notification-channels.spec.ts | 74 +++++++++++++++++++ .../notification-outbox-service.spec.ts | 1 + .../operator-notification-service.spec.ts | 44 +++++++++++ test/unit/utils/settings-redaction.spec.ts | 22 ++++++ 17 files changed, 270 insertions(+), 18 deletions(-) create mode 100644 test/unit/services/notification-channels.spec.ts diff --git a/src/@types/repositories.ts b/src/@types/repositories.ts index 922796ee9..b119ebb06 100644 --- a/src/@types/repositories.ts +++ b/src/@types/repositories.ts @@ -111,6 +111,7 @@ export interface INotificationDeliveryLogRepository { client?: DatabaseClient, ): Promise findRecent(limit?: number, client?: DatabaseClient): Promise + findSuccessfulTargetIds(outboxId: string, client?: DatabaseClient): Promise deleteOlderThan(cutoff: Date, client?: DatabaseClient): Promise } @@ -127,6 +128,8 @@ export interface INotificationOutboxRepository { error: string, attemptCount: number, maxAttempts: number, + baseDelayMs: number, client?: DatabaseClient, ): Promise + deleteTerminalOlderThan(cutoff: Date, client?: DatabaseClient): Promise } diff --git a/src/@types/services.ts b/src/@types/services.ts index 91dd14796..bb7c124be 100644 --- a/src/@types/services.ts +++ b/src/@types/services.ts @@ -45,6 +45,7 @@ export interface INotificationDispatcher { export interface IOperatorNotificationService extends INotificationDispatcher { dispatchTestTarget(targetId: string): Promise getMaxAttempts(): number + getBaseDelayMs(): number } export interface INotificationOutboxService { diff --git a/src/app/maintenance-worker.ts b/src/app/maintenance-worker.ts index a16d31ed3..d5ed3fdd8 100644 --- a/src/app/maintenance-worker.ts +++ b/src/app/maintenance-worker.ts @@ -229,6 +229,7 @@ export class MaintenanceWorker implements IRunnable { const retentionDays = this.settings().admin?.notifications?.deliveryLogRetentionDays ?? 30 const cutoff = new Date(Date.now() - retentionDays * 86_400_000) await this.notificationDeliveryLogRepository.deleteOlderThan(cutoff) + await this.notificationOutboxRepository.deleteTerminalOlderThan(cutoff) } catch (error) { logger.error('Unable to process notification outbox', error) } diff --git a/src/controllers/admin/get-notification-delivery-log-controller.ts b/src/controllers/admin/get-notification-delivery-log-controller.ts index 26ba96386..9873776f5 100644 --- a/src/controllers/admin/get-notification-delivery-log-controller.ts +++ b/src/controllers/admin/get-notification-delivery-log-controller.ts @@ -7,7 +7,18 @@ export class GetAdminNotificationDeliveryLogController implements IController { public constructor(private readonly deliveryLogRepository: INotificationDeliveryLogRepository) {} public async handleRequest(_request: Request, response: Response): Promise { - const limit = Math.min(Number(_request.query.limit) || 50, 200) + let limit = 50 + if (_request.query.limit !== undefined) { + const parsed = Number(_request.query.limit) + if (!Number.isInteger(parsed) || parsed <= 0) { + response.status(400).setHeader('content-type', 'application/json').send({ + error: 'limit must be a positive integer', + }) + return + } + limit = Math.min(parsed, 200) + } + const entries = await this.deliveryLogRepository.findRecent(limit) response.status(200).setHeader('content-type', 'application/json').send({ diff --git a/src/factories/notification-outbox-service-factory.ts b/src/factories/notification-outbox-service-factory.ts index b9fe27e88..c499f8bcd 100644 --- a/src/factories/notification-outbox-service-factory.ts +++ b/src/factories/notification-outbox-service-factory.ts @@ -17,5 +17,6 @@ export const createNotificationOutboxService = () => { outboxRepository, operatorNotificationService, () => operatorNotificationService.getMaxAttempts(), + () => operatorNotificationService.getBaseDelayMs(), ) } diff --git a/src/repositories/notification-delivery-log-repository.ts b/src/repositories/notification-delivery-log-repository.ts index d2cedf300..b83bd6ea9 100644 --- a/src/repositories/notification-delivery-log-repository.ts +++ b/src/repositories/notification-delivery-log-repository.ts @@ -53,6 +53,17 @@ export class NotificationDeliveryLogRepository implements INotificationDeliveryL }) } + public async findSuccessfulTargetIds( + outboxId: string, + client: DatabaseClient = this.dbClient, + ): Promise { + return client('notification_delivery_log') + .where('outbox_id', outboxId) + .where('status', NotificationDeliveryStatus.SUCCESS) + .distinct('target_id') + .pluck('target_id') + } + public async findRecent(limit = 50, client: DatabaseClient = this.dbClient): Promise { const rows = await client('notification_delivery_log') .orderBy('created_at', 'desc') diff --git a/src/repositories/notification-outbox-repository.ts b/src/repositories/notification-outbox-repository.ts index 8788f5a26..6e29d416d 100644 --- a/src/repositories/notification-outbox-repository.ts +++ b/src/repositories/notification-outbox-repository.ts @@ -12,6 +12,9 @@ import { createLogger } from '../factories/logger-factory' const logger = createLogger('notification-outbox-repository') +/** Reclaim PROCESSING rows when a worker dies after claim. */ +export const NOTIFICATION_OUTBOX_PROCESSING_LEASE_MS = 5 * 60 * 1000 + function fromDB(row: DBNotificationOutboxMessage): NotificationOutboxMessage { return { id: row.id, @@ -63,9 +66,23 @@ export class NotificationOutboxRepository implements INotificationOutboxReposito logger('claim notification outbox batch (limit %d)', limit) return client.transaction(async (trx) => { + const now = new Date() + const staleBefore = new Date(now.getTime() - NOTIFICATION_OUTBOX_PROCESSING_LEASE_MS) + const rows = await trx('notification_outbox') - .where('status', NotificationOutboxStatus.PENDING) - .where('available_at', '<=', trx.fn.now()) + .where((builder) => { + builder + .where((pending) => { + pending + .where('status', NotificationOutboxStatus.PENDING) + .where('available_at', '<=', trx.fn.now()) + }) + .orWhere((processing) => { + processing + .where('status', NotificationOutboxStatus.PROCESSING) + .where('updated_at', '<', staleBefore) + }) + }) .orderBy('created_at', 'asc') .limit(limit) .forUpdate() @@ -76,7 +93,6 @@ export class NotificationOutboxRepository implements INotificationOutboxReposito return [] } - const now = new Date() const ids = rows.map((row) => row.id) await trx('notification_outbox').whereIn('id', ids).update({ @@ -111,13 +127,15 @@ export class NotificationOutboxRepository implements INotificationOutboxReposito error: string, attemptCount: number, maxAttempts: number, + baseDelayMs: number, client: DatabaseClient = this.dbClient, ): Promise { logger('mark notification outbox failed %s (attempt %d)', id, attemptCount) const now = new Date() const isDead = attemptCount >= maxAttempts - const backoffMs = Math.min(60_000, 1000 * 2 ** Math.max(0, attemptCount - 1)) + const delayBase = Math.max(0, baseDelayMs) + const backoffMs = Math.min(60_000, delayBase * 2 ** Math.max(0, attemptCount - 1)) const availableAt = new Date(now.getTime() + backoffMs) await client('notification_outbox') @@ -130,4 +148,15 @@ export class NotificationOutboxRepository implements INotificationOutboxReposito updated_at: now, }) } + + public async deleteTerminalOlderThan(cutoff: Date, client: DatabaseClient = this.dbClient): Promise { + logger('delete terminal notification outbox rows older than %s', cutoff.toISOString()) + + const result = await client('notification_outbox') + .whereIn('status', [NotificationOutboxStatus.DELIVERED, NotificationOutboxStatus.DEAD]) + .where('updated_at', '<', cutoff) + .delete() + + return typeof result === 'number' ? result : 0 + } } diff --git a/src/routes/admin/index.ts b/src/routes/admin/index.ts index b320c32dc..0d9e019ec 100644 --- a/src/routes/admin/index.ts +++ b/src/routes/admin/index.ts @@ -100,7 +100,7 @@ router.post( withAdminController(createPostAdminSettingsRestoreController), ) router.get( - '/notifications/delivery-log', + '/notifications/deliveries', adminRateLimitMiddleware, adminAuthMiddleware, withAdminController(createGetAdminNotificationDeliveryLogController), diff --git a/src/services/notification-outbox-service.ts b/src/services/notification-outbox-service.ts index 170b71da8..22b54129c 100644 --- a/src/services/notification-outbox-service.ts +++ b/src/services/notification-outbox-service.ts @@ -12,6 +12,7 @@ export class NotificationOutboxService implements INotificationOutboxService { private readonly outboxRepository: INotificationOutboxRepository, private readonly dispatcher: INotificationDispatcher, private readonly maxAttempts: () => number = () => NOTIFICATION_OUTBOX_MAX_ATTEMPTS, + private readonly baseDelayMs: () => number = () => 1000, ) {} public async processBatch(limit = NOTIFICATION_OUTBOX_BATCH_SIZE): Promise { @@ -39,6 +40,7 @@ export class NotificationOutboxService implements INotificationOutboxService { reason, message.attemptCount + 1, this.maxAttempts(), + this.baseDelayMs(), ) } } diff --git a/src/services/operator-notification-service.ts b/src/services/operator-notification-service.ts index 80e414ea5..3a9a5576c 100644 --- a/src/services/operator-notification-service.ts +++ b/src/services/operator-notification-service.ts @@ -40,7 +40,17 @@ export class OperatorNotificationService implements INotificationDispatcher { return } - const targets = config.targets.filter((target) => target.enabled) + const enabledTargets = config.targets.filter((target) => target.enabled) + if (!enabledTargets.length) { + return + } + + const alreadyDelivered = new Set( + context.outboxId + ? await this.deliveryLogRepository.findSuccessfulTargetIds(context.outboxId) + : [], + ) + const targets = enabledTargets.filter((target) => !alreadyDelivered.has(target.id)) if (!targets.length) { return } @@ -102,6 +112,10 @@ export class OperatorNotificationService implements INotificationDispatcher { return this.getNotificationsConfig().retry.maxAttempts } + public getBaseDelayMs(): number { + return this.getNotificationsConfig().retry.baseDelayMs + } + private getNotificationsConfig(): AdminNotificationsSettings { const defaults = loadDefaults().admin?.notifications const configured = this.settings().admin?.notifications diff --git a/src/utils/settings-config.ts b/src/utils/settings-config.ts index b6c22b745..9b338dbd6 100644 --- a/src/utils/settings-config.ts +++ b/src/utils/settings-config.ts @@ -640,10 +640,21 @@ const validateAdminNotifications = (settings: Settings): ValidationIssue[] => { }) } + const rawTargets = notifications.targets + if (rawTargets !== undefined && !Array.isArray(rawTargets)) { + issues.push({ path: 'admin.notifications.targets', message: 'targets must be an array' }) + return issues + } + const targetIds = new Set() - for (const [index, target] of (notifications.targets ?? []).entries()) { + for (const [index, target] of (rawTargets ?? []).entries()) { const prefix = `admin.notifications.targets[${index}]` - if (!target.id?.trim()) { + if (!target || typeof target !== 'object' || Array.isArray(target)) { + issues.push({ path: prefix, message: 'target must be an object' }) + continue + } + + if (typeof target.id !== 'string' || !target.id.trim()) { issues.push({ path: `${prefix}.id`, message: 'target id is required' }) } else if (targetIds.has(target.id)) { issues.push({ path: `${prefix}.id`, message: 'target id must be unique' }) @@ -651,20 +662,36 @@ const validateAdminNotifications = (settings: Settings): ValidationIssue[] => { targetIds.add(target.id) } - if (!['http', 'discord', 'slack', 'telegram'].includes(target.type)) { + if ( + typeof target.type !== 'string' || + !['http', 'discord', 'slack', 'telegram'].includes(target.type) + ) { issues.push({ path: `${prefix}.type`, message: 'type must be http, discord, slack, or telegram' }) } + if (target.enabled !== undefined && typeof target.enabled !== 'boolean') { + issues.push({ path: `${prefix}.enabled`, message: 'enabled must be a boolean' }) + } + if (target.type === 'telegram') { - if (!target.botToken?.trim()) { + if (typeof target.botToken !== 'string' || !target.botToken.trim()) { issues.push({ path: `${prefix}.botToken`, message: 'botToken is required for telegram targets' }) } - if (!target.chatId?.trim()) { + if (typeof target.chatId !== 'string' || !target.chatId.trim()) { issues.push({ path: `${prefix}.chatId`, message: 'chatId is required for telegram targets' }) } } else if (target.type === 'http' || target.type === 'discord' || target.type === 'slack') { - if (!target.url?.trim()) { + if (typeof target.url !== 'string' || !target.url.trim()) { issues.push({ path: `${prefix}.url`, message: 'url is required for webhook targets' }) + } else { + try { + const parsed = new URL(target.url) + if (parsed.protocol !== 'http:' && parsed.protocol !== 'https:') { + issues.push({ path: `${prefix}.url`, message: 'url must use http or https' }) + } + } catch { + issues.push({ path: `${prefix}.url`, message: 'url must be a valid URL' }) + } } } } diff --git a/src/utils/settings-redaction.ts b/src/utils/settings-redaction.ts index 05d4e77fe..a3edf8f7d 100644 --- a/src/utils/settings-redaction.ts +++ b/src/utils/settings-redaction.ts @@ -9,7 +9,16 @@ export const isSensitiveSettingsPath = (path: string): boolean => { const lastSegment = segments[segments.length - 1] ?? '' const key = lastSegment.replace(/\[\d+\]$/, '') - return SENSITIVE_SETTING_KEYS.has(key) + if (SENSITIVE_SETTING_KEYS.has(key)) { + return true + } + + // Webhook URLs embed secrets (Discord/Slack tokens, signed HTTP endpoints). + if (key === 'url' && path.startsWith('admin.notifications.targets')) { + return true + } + + return false } export const isWriteProtectedSettingsPath = (path: string): boolean => { @@ -25,9 +34,9 @@ export const redactSettingsValue = (path: string, value: unknown): unknown => { } export const redactSettingsSecrets = (settings: T): T => { - const redactWalk = (value: unknown): unknown => { + const redactWalk = (value: unknown, path = ''): unknown => { if (Array.isArray(value)) { - return value.map(redactWalk) + return value.map((entry, index) => redactWalk(entry, `${path}[${index}]`)) } if (!isPlainObject(value)) { @@ -37,12 +46,13 @@ export const redactSettingsSecrets = (settings: T): T => { const result: Record = {} for (const [key, entry] of Object.entries(value)) { - if (SENSITIVE_SETTING_KEYS.has(key) && typeof entry === 'string' && entry.length > 0) { + const childPath = path ? `${path}.${key}` : key + if (isSensitiveSettingsPath(childPath) && typeof entry === 'string' && entry.length > 0) { result[key] = '***' continue } - result[key] = redactWalk(entry) + result[key] = redactWalk(entry, childPath) } return result diff --git a/test/unit/app/maintenance-worker.spec.ts b/test/unit/app/maintenance-worker.spec.ts index 24e747000..da3ff4814 100644 --- a/test/unit/app/maintenance-worker.spec.ts +++ b/test/unit/app/maintenance-worker.spec.ts @@ -104,6 +104,7 @@ describe('MaintenanceWorker', () => { notificationOutboxRepository = { enqueue: sandbox.stub().resolves(), + deleteTerminalOlderThan: sandbox.stub().resolves(0), } // Prevent real timeouts and randomized per-invoice delays. diff --git a/test/unit/services/notification-channels.spec.ts b/test/unit/services/notification-channels.spec.ts new file mode 100644 index 000000000..3213f6296 --- /dev/null +++ b/test/unit/services/notification-channels.spec.ts @@ -0,0 +1,74 @@ +import axios from 'axios' +import chai from 'chai' +import Sinon from 'sinon' +import sinonChai from 'sinon-chai' + +import { OperatorNotificationEventType } from '../../../src/@types/operator-notifications' +import { deliverToTarget, validateTargetConfig } from '../../../src/services/notification-channels' + +chai.use(sinonChai) + +const { expect } = chai + +const envelope = { + event: OperatorNotificationEventType.ADMISSION_INVOICE_PAID, + relay: 'wss://relay.example', + timestamp: new Date().toISOString(), + data: { invoiceId: 'inv-1' }, +} + +describe('notification-channels', () => { + let sandbox: Sinon.SinonSandbox + + beforeEach(() => { + sandbox = Sinon.createSandbox() + sandbox.stub(axios, 'post').resolves({ status: 200, data: {} }) + }) + + afterEach(() => { + sandbox.restore() + }) + + it('delivers HTTP webhook payloads', async () => { + await deliverToTarget( + { id: 'http-1', type: 'http', enabled: true, url: 'https://example.com/hook' }, + envelope, + ) + + expect(axios.post).to.have.been.calledOnceWith('https://example.com/hook', envelope, Sinon.match.object) + }) + + it('delivers Slack webhook payloads', async () => { + await deliverToTarget( + { id: 'slack-1', type: 'slack', enabled: true, url: 'https://hooks.slack.com/services/test' }, + envelope, + ) + + expect(axios.post).to.have.been.calledOnce + }) + + it('delivers Telegram bot messages', async () => { + await deliverToTarget( + { + id: 'telegram-1', + type: 'telegram', + enabled: true, + botToken: '123:abc', + chatId: '-100123', + }, + envelope, + ) + + expect(axios.post).to.have.been.calledOnceWith( + 'https://api.telegram.org/bot123:abc/sendMessage', + Sinon.match.object, + Sinon.match.object, + ) + }) + + it('validates required target fields', () => { + expect(validateTargetConfig({ id: '', type: 'http', enabled: true, url: 'https://x' })).to.match(/id/) + expect(validateTargetConfig({ id: 'x', type: 'slack', enabled: true })).to.match(/url/) + expect(validateTargetConfig({ id: 'x', type: 'telegram', enabled: true, botToken: 't' })).to.match(/chatId/) + }) +}) diff --git a/test/unit/services/notification-outbox-service.spec.ts b/test/unit/services/notification-outbox-service.spec.ts index 7fc6d8956..b4c6cd4b5 100644 --- a/test/unit/services/notification-outbox-service.spec.ts +++ b/test/unit/services/notification-outbox-service.spec.ts @@ -74,6 +74,7 @@ describe('NotificationOutboxService', () => { 'network down', 1, 5, + 1000, ) }) }) diff --git a/test/unit/services/operator-notification-service.spec.ts b/test/unit/services/operator-notification-service.spec.ts index 61e17cbed..2d9dba7f6 100644 --- a/test/unit/services/operator-notification-service.spec.ts +++ b/test/unit/services/operator-notification-service.spec.ts @@ -21,6 +21,7 @@ describe('OperatorNotificationService', () => { sandbox = Sinon.createSandbox() deliveryLogRepository = { append: sandbox.stub().resolves(), + findSuccessfulTargetIds: sandbox.stub().resolves([]), } service = new OperatorNotificationService( () => @@ -82,4 +83,47 @@ describe('OperatorNotificationService', () => { service.dispatch(OperatorNotificationEventType.ADMISSION_INVOICE_PAID, { invoiceId: 'inv-1' }), ).to.be.rejectedWith('network down') }) + + it('skips targets that already succeeded for the same outbox message', async () => { + service = new OperatorNotificationService( + () => + ({ + info: { relay_url: 'wss://relay.example' }, + admin: { + notifications: { + enabled: true, + targets: [ + { + id: 'discord-main', + type: 'discord', + enabled: true, + url: 'https://discord.com/api/webhooks/test', + }, + { + id: 'slack-ops', + type: 'slack', + enabled: true, + url: 'https://hooks.slack.com/services/test', + }, + ], + events: { 'admission.invoice.paid': true }, + retry: { maxAttempts: 5, baseDelayMs: 1000 }, + }, + }, + }) as any, + deliveryLogRepository, + ) + deliveryLogRepository.findSuccessfulTargetIds.resolves(['discord-main']) + ;(axios.post as Sinon.SinonStub).rejects(new Error('slack down')) + + await expect( + service.dispatch( + OperatorNotificationEventType.ADMISSION_INVOICE_PAID, + { invoiceId: 'inv-1' }, + { outboxId: 'ob-1', attemptNumber: 2 }, + ), + ).to.be.rejectedWith('slack-ops: slack down') + + expect(axios.post).to.have.been.calledOnce + }) }) diff --git a/test/unit/utils/settings-redaction.spec.ts b/test/unit/utils/settings-redaction.spec.ts index e140f9037..cef8c78fb 100644 --- a/test/unit/utils/settings-redaction.spec.ts +++ b/test/unit/utils/settings-redaction.spec.ts @@ -56,5 +56,27 @@ describe('settings-redaction', () => { it('redacts single values by path', () => { expect(redactSettingsValue('mirroring.static[0].secret', 'top-secret')).to.equal('***') expect(redactSettingsValue('payments.enabled', true)).to.equal(true) + expect(redactSettingsValue('admin.notifications.targets[0].url', 'https://discord.com/api/webhooks/secret')).to.equal( + '***', + ) + }) + + it('redacts notification webhook urls in nested settings', () => { + const input = { + admin: { + notifications: { + targets: [ + { + id: 'slack-ops', + type: 'slack', + enabled: true, + url: 'https://hooks.slack.com/services/T00/B00/xxxxx', + }, + ], + }, + }, + } + + expect(redactSettingsSecrets(input).admin.notifications.targets[0].url).to.equal('***') }) })