feat(admin): operator notification backend (outbox, channels, delivery log)

.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.

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')
+}

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')
+}

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

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<string, unknown>
+
+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
+}

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<string, unknown>
+}
+
+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
+}

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,40 @@ export interface IReportRepository {
   findByEventId(eventId: EventId): Promise<Report[]>
   findActionable(limit?: number): Promise<Report[]>
 }
+
+export interface INotificationDeliveryLogRepository {
+  append(
+    entry: {
+      outboxId: string | null
+      eventType: string
+      targetId: string
+      targetType: OperatorNotificationChannelType
+      status: NotificationDeliveryStatus
+      attemptNumber: number
+      errorSnippet: string | null
+    },
+    client?: DatabaseClient,
+  ): Promise<void>
+  findRecent(limit?: number, client?: DatabaseClient): Promise<NotificationDeliveryLogEntry[]>
+  findSuccessfulTargetIds(outboxId: string, client?: DatabaseClient): Promise<string[]>
+  deleteOlderThan(cutoff: Date, client?: DatabaseClient): Promise<number>
+}
+
+export interface INotificationOutboxRepository {
+  enqueue(
+    eventType: string,
+    payload: NotificationOutboxPayload,
+    client?: DatabaseClient,
+  ): Promise<NotificationOutboxMessage>
+  claimBatch(limit: number, client?: DatabaseClient): Promise<NotificationOutboxMessage[]>
+  markDelivered(id: string, client?: DatabaseClient): Promise<void>
+  markFailed(
+    id: string,
+    error: string,
+    attemptCount: number,
+    maxAttempts: number,
+    baseDelayMs: number,
+    client?: DatabaseClient,
+  ): Promise<void>
+  deleteTerminalOlderThan(cutoff: Date, client?: DatabaseClient): Promise<number>
+}

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<void>
@@ -27,3 +28,26 @@ export interface IPaymentsService {
   sendInvoiceUpdateNotification(invoice: Invoice): Promise<void>
   getPendingInvoices(offset?: number): Promise<Invoice[]>
 }
+
+export interface NotificationDispatchContext {
+  outboxId?: string
+  attemptNumber?: number
+}
+
+export interface INotificationDispatcher {
+  dispatch(
+    eventType: string,
+    payload: NotificationOutboxPayload,
+    context?: NotificationDispatchContext,
+  ): Promise<void>
+}
+
+export interface IOperatorNotificationService extends INotificationDispatcher {
+  dispatchTestTarget(targetId: string): Promise<void>
+  getMaxAttempts(): number
+  getBaseDelayMs(): number
+}
+
+export interface INotificationOutboxService {
+  processBatch(limit?: number): Promise<number>
+}

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

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) => {

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,22 @@ export class MaintenanceWorker implements IRunnable {
     await clearOldEventsPromise
   }
 
+  private async processNotificationOutbox(): Promise<void> {
+    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)
+      await this.notificationOutboxRepository.deleteTerminalOlderThan(cutoff)
+    } catch (error) {
+      logger.error('Unable to process notification outbox', error)
+    }
+  }
+
   private async processNip05Reverifications(currentSettings: Settings): Promise<void> {
     const nip05Settings = currentSettings.nip05
     if (!nip05Settings || nip05Settings.mode === 'disabled') {