feat(nip66): add RelayMonitorWorker cluster worker and probe scheduler

.changeset/nip66-relay-monitor-worker.md

@@ -0,0 +1,5 @@
+---
+"nostream": minor
+---
+
+feat(nip66): add RelayMonitorWorker cluster worker and probe scheduler

.changeset/nip66-settings-foundation.md

@@ -0,0 +1,5 @@
+---
+"nostream": minor
+---
+
+Add NIP-66 relay monitor settings foundation with defaults for probe interval, timeouts, targets, monitor identity, and DNS cache TTL.

.env.example

@@ -55,6 +55,9 @@ WORKER_COUNT=2 # Defaults to CPU count. Use 1 or 2 for local testing.
 # --- RELAY PRIVATE KEY (Optional) ---
 # RELAY_PRIVATE_KEY=your_hex_private_key
 
+# --- NIP-66 MONITOR IDENTITY (Reserved; not used yet) ---
+# MONITOR_PRIVATE_KEY=your_hex_monitor_private_key
+
 # --- PAYMENTS (Only if enabled in settings.yaml) ---
 # ZEBEDEE_API_KEY=
 # NODELESS_API_KEY=

CONFIGURATION.md

@@ -194,6 +194,16 @@ The settings below are listed in alphabetical order by name. Please keep this ta
 | nip50.enabled                               | Enable or disable NIP-50 full-text search. Defaults to false. When enabled, clients can include a `search` field in REQ filters to perform text queries against event content. Requires the GIN full-text index migration. |
 | nip50.language                              | PostgreSQL text-search configuration name. Defaults to `simple` (language-agnostic tokenization). Set to `english`, `spanish`, etc. for stemming support. See [PostgreSQL text search configurations](https://www.postgresql.org/docs/current/textsearch-configuration.html). **Note:** The GIN index migration is built with the `simple` configuration. If you change this value, you must manually rebuild the index: `DROP INDEX CONCURRENTLY events_content_fts_idx; CREATE INDEX CONCURRENTLY events_content_fts_idx ON events USING gin (to_tsvector('<your_language>', event_content));` — otherwise the planner cannot use the index and queries fall back to sequential scans. |
 | nip50.maxQueryLength                        | Maximum length of the search query string. Queries exceeding this are truncated. Defaults to 256. |
+| nip66.dnsCacheTtlSeconds                    | DNS cache TTL in seconds for repeated probe lookups of the same hostname. Defaults to 300. |
+| nip66.enabled                               | Enable NIP-66 relay monitoring. When true, starts a `relay-monitor` cluster worker that probes targets on an interval and stores the latest snapshot in Redis. Defaults to false. |
+| nip66.monitorPrivateKey                     | Hex-encoded private key for the monitor identity that will sign kind 30166/10166 events. Reserved for a future event publisher. |
+| nip66.monitorPubkey                         | Hex-encoded public key for the monitor identity. Optional when `monitorPrivateKey` is configured. Reserved for a future event publisher. |
+| nip66.probeIntervalSeconds                  | Seconds between scheduled relay probe runs. Defaults to 3600. |
+| nip66.targets                               | Public WebSocket URLs to probe (for example `wss://relay.example.com`). When empty, defaults to `info.relay_url`. |
+| nip66.timeouts.dnsMs                        | DNS probe timeout in milliseconds. Defaults to 10000. |
+| nip66.timeouts.nip11Ms                      | NIP-11 fetch timeout in milliseconds. Defaults to 10000. |
+| nip66.timeouts.tlsMs                        | TLS probe timeout in milliseconds. Defaults to 10000. |
+| nip66.timeouts.wsRttMs                      | WebSocket open RTT probe timeout in milliseconds. Defaults to 10000. |
 | paymentProcessors.lnbits.baseURL            | Base URL of your Lnbits instance. |
 | paymentProcessors.lnbits.callbackBaseURL    | Public-facing Nostream's Lnbits Callback URL. (e.g. https://relay.your-domain.com/callbacks/lnbits) |
 | paymentProcessors.lnurl.invoiceURL          | [LUD-06 Pay Request](https://github.com/lnurl/luds/blob/luds/06.md) provider URL. (e.g. https://getalby.com/lnurlp/your-username) |

resources/default-settings.yaml

@@ -80,6 +80,23 @@ nip50:
   # 'simple' (no stemming) or a language name like 'english', 'spanish'
   language: simple
   maxQueryLength: 256
+nip66:
+  # NIP-66 relay liveness monitoring. Disabled by default.
+  # When enabled, a relay-monitor worker probes targets and stores the latest snapshot in Redis.
+  enabled: false
+  # Seconds between scheduled probe runs.
+  probeIntervalSeconds: 3600
+  timeouts:
+    dnsMs: 10000
+    tlsMs: 10000
+    wsRttMs: 10000
+    nip11Ms: 10000
+  # Public WebSocket URLs to probe. Empty list defaults to info.relay_url.
+  targets: []
+  # Optional monitor identity (reserved for a future event publisher).
+  # monitorPrivateKey: replace-with-monitor-private-key-in-hex
+  # monitorPubkey: replace-with-monitor-pubkey-in-hex
+  dnsCacheTtlSeconds: 300
 wot:
   # Web of Trust filtering. When enabled, only events from pubkeys within
   # the relay owner's 2-hop follow graph are accepted.

src/@types/relay-probe-snapshot.ts

@@ -0,0 +1,48 @@
+import {
+  DnsRecord,
+  Nip11Result,
+  ProbeCheckResult,
+  ProbeResult,
+  ProbeTarget,
+  WsRttResult,
+} from '../utils/relay-probe/types'
+
+export type RelayProbeRunStatus = 'ok' | 'partial' | 'failed'
+
+export interface StoredDnsResult {
+  hostname: string
+  records: DnsRecord[]
+  fromCache: boolean
+  cacheExpiresAt?: string
+}
+
+export interface StoredTlsResult {
+  valid: boolean
+  issuer?: string
+  subject?: string
+  expiresAt?: string
+  daysUntilExpiry?: number
+}
+
+export interface StoredProbeResult {
+  target: ProbeTarget
+  checkedAt: string
+  dns: ProbeCheckResult<StoredDnsResult>
+  tls: ProbeCheckResult<StoredTlsResult>
+  wsRtt: ProbeCheckResult<WsRttResult>
+  nip11: ProbeCheckResult<Nip11Result>
+}
+
+export interface RelayProbeRunSnapshot {
+  runAt: string
+  targets: string[]
+  results: StoredProbeResult[]
+  status: RelayProbeRunStatus
+}
+
+export interface IRelayProbeSnapshotStore {
+  saveLatest(snapshot: RelayProbeRunSnapshot, expirySeconds?: number): Promise<void>
+  getLatest(): Promise<RelayProbeRunSnapshot | null>
+}
+
+export type ProbeRunStatusInput = Pick<StoredProbeResult, 'wsRtt'>

src/@types/settings.ts

@@ -273,6 +273,50 @@ export interface Nip50Settings {
   maxQueryLength?: number
 }
 
+export interface Nip66ProbeTimeouts {
+  dnsMs: number
+  tlsMs: number
+  wsRttMs: number
+  nip11Ms: number
+}
+
+export interface Nip66Settings {
+  /**
+   * Enable NIP-66 relay monitoring. When true, the primary process starts a
+   * relay-monitor cluster worker that probes configured targets on an interval.
+   * Defaults to false.
+   */
+  enabled: boolean
+  /**
+   * Interval in seconds between probe runs. Defaults to 3600.
+   */
+  probeIntervalSeconds: number
+  /**
+   * Per-check probe timeouts in milliseconds.
+   */
+  timeouts: Nip66ProbeTimeouts
+  /**
+   * Public relay WebSocket URLs to probe (for example wss://relay.example.com).
+   * When empty, the monitor worker uses info.relay_url.
+   */
+  targets: string[]
+  /**
+   * Hex-encoded private key for the monitor identity that will sign kind 30166/10166 events.
+   * Reserved for a future event publisher worker.
+   */
+  monitorPrivateKey?: Secret
+  /**
+   * Hex-encoded public key for the monitor identity.
+   * Optional when monitorPrivateKey is set. Reserved for a future event publisher worker.
+   */
+  monitorPubkey?: Pubkey
+  /**
+   * DNS cache TTL in seconds for repeated probes of the same hostname.
+   * Defaults to 300.
+   */
+  dnsCacheTtlSeconds: number
+}
+
 export interface Nip05Settings {
   mode: Nip05Mode
   /**
@@ -351,5 +395,6 @@ export interface Settings {
   nip43?: Nip43Settings
   nip45?: Nip45Settings
   nip50?: Nip50Settings
+  nip66?: Nip66Settings
   wot?: WoTSettings
 }

src/app/app.ts

@@ -108,6 +108,13 @@ export class App implements IRunnable {
       logCentered(`${mirrors.length} static-mirroring worker started`, width)
     }
 
+    if (settings.nip66?.enabled) {
+      createWorker({
+        WORKER_TYPE: 'relay-monitor',
+      })
+      logCentered('1 relay-monitor worker started', width)
+    }
+
     const dvmWorkers = settings?.dvm?.workers
 
     if (Array.isArray(dvmWorkers) && dvmWorkers.length) {

src/app/relay-monitor-worker.ts

@@ -0,0 +1,161 @@
+import { IRunnable } from '../@types/base'
+import { IRelayProbeSnapshotStore, RelayProbeRunSnapshot } from '../@types/relay-probe-snapshot'
+import { Settings } from '../@types/settings'
+import { createLogger } from '../factories/logger-factory'
+import { shutdownMetricsTelemetry } from '../telemetry/metrics'
+import { filterValidProbeTargets, resolveProbeTargets } from '../utils/relay-probe-targets'
+import { deriveRelayProbeRunStatus, serializeProbeResults } from '../utils/relay-probe-snapshot'
+import { runProbe } from '../utils/relay-probe'
+import { ProbeOptions, ProbeResult } from '../utils/relay-probe/types'
+
+const logger = createLogger('relay-monitor-worker')
+
+const DEFAULT_PROBE_INTERVAL_SECONDS = 3600
+const MIN_PROBE_INTERVAL_SECONDS = 60
+
+export type RunProbeFn = (relayUrl: string, options?: ProbeOptions) => Promise<ProbeResult>
+
+export const buildProbeOptions = (settings: Settings): ProbeOptions => {
+  const nip66 = settings.nip66
+
+  return {
+    timeouts: nip66?.timeouts,
+    dnsCacheTtlSeconds: nip66?.dnsCacheTtlSeconds,
+  }
+}
+
+export const getProbeIntervalMs = (settings: Settings): number => {
+  const configured = settings.nip66?.probeIntervalSeconds ?? DEFAULT_PROBE_INTERVAL_SECONDS
+  const intervalSeconds = Math.max(configured, MIN_PROBE_INTERVAL_SECONDS)
+
+  return intervalSeconds * 1000
+}
+
+export class RelayMonitorWorker implements IRunnable {
+  private interval: NodeJS.Timeout | undefined
+  private isRunning = false
+
+  public constructor(
+    private readonly process: NodeJS.Process,
+    private readonly settings: () => Settings,
+    private readonly snapshotStore: IRelayProbeSnapshotStore,
+    private readonly probeRunner: RunProbeFn = runProbe,
+  ) {
+    this.process
+      .on('SIGINT', this.onExit.bind(this))
+      .on('SIGHUP', this.onExit.bind(this))
+      .on('SIGTERM', this.onExit.bind(this))
+      .on('uncaughtException', this.onError.bind(this))
+      .on('unhandledRejection', this.onError.bind(this))
+  }
+
+  public run(): void {
+    const currentSettings = this.settings()
+
+    if (!currentSettings.nip66?.enabled) {
+      logger('NIP-66 relay monitoring is disabled; worker exiting')
+      return
+    }
+
+    const intervalMs = getProbeIntervalMs(currentSettings)
+    logger('starting probe scheduler with interval %d ms', intervalMs)
+
+    void this.runScheduledProbes()
+
+    this.interval = setInterval(() => {
+      void this.runScheduledProbes()
+    }, intervalMs)
+  }
+
+  private async runScheduledProbes(): Promise<void> {
+    if (this.isRunning) {
+      logger('skipping scheduled probe run because previous run is still in progress')
+      return
+    }
+
+    this.isRunning = true
+
+    try {
+      await this.onSchedule()
+    } catch (error) {
+      this.onError(error as Error)
+    } finally {
+      this.isRunning = false
+    }
+  }
+
+  private async onSchedule(): Promise<void> {
+    const currentSettings = this.settings()
+
+    if (!currentSettings.nip66?.enabled) {
+      logger('NIP-66 relay monitoring disabled during scheduled run; stopping scheduler')
+      this.close()
+      return
+    }
+
+    const configuredTargets = resolveProbeTargets(currentSettings)
+    const { valid, invalid } = filterValidProbeTargets(configuredTargets)
+
+    for (const target of invalid) {
+      logger.warn('skipping invalid probe target: %s', target)
+    }
+
+    if (valid.length === 0) {
+      logger.warn('no valid probe targets configured; skipping probe run')
+      return
+    }
+
+    const probeOptions = buildProbeOptions(currentSettings)
+    const results: ProbeResult[] = []
+
+    for (const target of valid) {
+      try {
+        results.push(await this.probeRunner(target, probeOptions))
+      } catch (error) {
+        logger.error('probe run failed for %s: %o', target, error)
+      }
+    }
+
+    if (results.length === 0) {
+      logger.warn('probe run produced no results')
+      return
+    }
+
+    const snapshot: RelayProbeRunSnapshot = {
+      runAt: new Date().toISOString(),
+      targets: valid,
+      results: serializeProbeResults(results),
+      status: deriveRelayProbeRunStatus(results),
+    }
+
+    const expirySeconds = Math.max(
+      (currentSettings.nip66?.probeIntervalSeconds ?? DEFAULT_PROBE_INTERVAL_SECONDS) * 2,
+      MIN_PROBE_INTERVAL_SECONDS * 2,
+    )
+
+    await this.snapshotStore.saveLatest(snapshot, expirySeconds)
+    logger('saved probe snapshot for %d target(s) with status %s', valid.length, snapshot.status)
+  }
+
+  private onError(error: Error) {
+    logger('error: %o', error)
+    throw error
+  }
+
+  private onExit() {
+    logger('exiting')
+    void shutdownMetricsTelemetry().finally(() => {
+      this.close(() => {
+        this.process.exit(0)
+      })
+    })
+  }
+
+  public close(callback?: () => void) {
+    logger('closing')
+    clearInterval(this.interval)
+    if (typeof callback === 'function') {
+      callback()
+    }
+  }
+}

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

@@ -0,0 +1,11 @@
+import { RedisAdapter } from '../adapters/redis-adapter'
+import { RelayMonitorWorker } from '../app/relay-monitor-worker'
+import { getCacheClient } from '../cache/client'
+import { createSettings } from './settings-factory'
+import { RelayProbeSnapshotStore } from '../utils/relay-probe-snapshot'
+
+export const relayMonitorWorkerFactory = () => {
+  const snapshotStore = new RelayProbeSnapshotStore(new RedisAdapter(getCacheClient()))
+
+  return new RelayMonitorWorker(process, createSettings, snapshotStore)
+}

src/index.ts

@@ -3,6 +3,7 @@ import cluster from 'cluster'
 import { appFactory } from './factories/app-factory'
 import { dvmOrchestratorWorkerFactory } from './factories/dvm-orchestrator-worker-factory'
 import { maintenanceWorkerFactory } from './factories/maintenance-worker-factory'
+import { relayMonitorWorkerFactory } from './factories/relay-monitor-worker-factory'
 import { staticMirroringWorkerFactory } from './factories/static-mirroring.worker-factory'
 import { workerFactory } from './factories/worker-factory'
 import { initializeMetricsTelemetry } from './telemetry/metrics'
@@ -18,6 +19,8 @@ export const getRunner = () => {
         return maintenanceWorkerFactory()
       case 'static-mirroring':
         return staticMirroringWorkerFactory()
+      case 'relay-monitor':
+        return relayMonitorWorkerFactory()
       case 'dvm-orchestrator':
         return dvmOrchestratorWorkerFactory()
       default: