Tor / I2P & network probes
diff --git a/resources/default-settings.yaml b/resources/default-settings.yaml
index f53357362..888b28737 100755
--- a/resources/default-settings.yaml
+++ b/resources/default-settings.yaml
@@ -198,7 +198,9 @@ peering:
terms: []
# Peering peers trusted by their relay pubkey (hex or npub), never by URL
# alone. A connection that proves one of these keys with NIP-42 AUTH is
- # recognized as a peer; any other key gets no peer recognition.
+ # recognized as a peer; any other key gets no peer recognition. Enabled
+ # peers are also synced: the peering-sync worker reads each peer's NIP-11
+ # peering scopes, reconciles them over NIP-77, and keeps a live feed.
# Example:
# peers:
# - pubkey: replace-with-peer-relay-pubkey-in-hex
diff --git a/src/@types/peering.ts b/src/@types/peering.ts
new file mode 100644
index 000000000..fecb40cf2
--- /dev/null
+++ b/src/@types/peering.ts
@@ -0,0 +1,40 @@
+import { PeeringScope, PeeringTerm } from './settings'
+
+/** One scope advertised by a peer in its NIP-11 `peering` section. */
+export type AdvertisedPeeringScope = PeeringScope
+
+export interface PeeringAdvertisement {
+ enabled: boolean
+ scopes: AdvertisedPeeringScope[]
+ terms: PeeringTerm[]
+ /** The relay key the peer advertises as `self`, when it publishes one. */
+ self?: string
+ supportedNips: number[]
+}
+
+export type PeeringSyncStatus = 'idle' | 'connecting' | 'reconciling' | 'live' | 'error' | 'stopped'
+
+/**
+ * Per-peer sync state. Written by the peering-sync worker and read by the admin
+ * console; `lagSeconds` is how old the newest event we hold in a current scope
+ * is, so a healthy live feed stays near zero.
+ */
+export interface PeeringPeerSyncState {
+ pubkey: string
+ relayUrl: string
+ status: PeeringSyncStatus
+ scopeCount: number
+ updatedAt: string
+ lastSyncStartedAt?: string
+ lastSyncFinishedAt?: string
+ lastReconciledAt?: string
+ lastEventAt?: string
+ /** Newest `created_at` held in a scope that is still current (no past `until`). */
+ latestEventCreatedAt?: number
+ lagSeconds?: number
+ /** Events the last reconciliation found on the peer but not locally. */
+ missingEvents?: number
+ /** Events fetched from the peer during the last reconciliation. */
+ receivedEvents?: number
+ lastError?: string
+}
diff --git a/src/app/app.ts b/src/app/app.ts
index c11cb7939..e0da3e483 100644
--- a/src/app/app.ts
+++ b/src/app/app.ts
@@ -1,23 +1,21 @@
+import { Serializable } from 'child_process'
import { Cluster, Worker } from 'cluster'
+import { FSWatcher } from 'fs'
import { cpus, hostname } from 'os'
import { levels } from 'pino'
import { path, pathEq } from 'ramda'
-import { FSWatcher } from 'fs'
-
-import { addOnion } from '../tor/client'
-import { createLogger } from '../factories/logger-factory'
-import { EventStoreBackend } from '../constants/base'
-import { IRunnable } from '../@types/base'
import packageJson from '../../package.json'
-import { Serializable } from 'child_process'
-import { Settings } from '../@types/settings'
-import { SettingsStatic } from '../utils/settings'
-import { BANNER_WIDTH, centre, isStructuredLogFormat, RELAY_BANNER_LINES } from '../utils/banner'
-import { shutdownMetricsTelemetry } from '../telemetry/metrics'
+import { IRunnable } from '../@types/base'
import { OperatorNotificationEventType } from '../@types/operator-notifications'
+import { Settings } from '../@types/settings'
+import { EventStoreBackend } from '../constants/base'
+import { createLogger } from '../factories/logger-factory'
import { RedisRelayBroadcastFanout } from '../relay-broadcast/redis-relay-broadcast-fanout'
-import { enqueueOperatorNotification } from '../utils/operator-notification-enqueue'
+import { shutdownMetricsTelemetry } from '../telemetry/metrics'
+import { addOnion } from '../tor/client'
+import { BANNER_WIDTH, centre, isStructuredLogFormat, RELAY_BANNER_LINES } from '../utils/banner'
import { EVENT_STORE_BACKEND_ENV, getConfiguredEventStoreBackend } from '../utils/event-store'
+import { enqueueOperatorNotification } from '../utils/operator-notification-enqueue'
import { RelayBroadcastDeduplicator } from '../utils/relay-broadcast-deduplicator'
import {
isRelayBroadcastFanoutEnabled,
@@ -25,6 +23,7 @@ import {
RelayBroadcastMessage,
} from '../utils/relay-broadcast-message'
import { setRelayBroadcastFanoutReady } from '../utils/relay-broadcast-state'
+import { SettingsStatic } from '../utils/settings'
import { getPrimaryShutdownDeadlineMs } from '../utils/shutdown-state'
const logger = createLogger('app-primary')
@@ -113,22 +112,24 @@ export class App implements IRunnable {
// own so a pipeline reads this line like every other one.
if (structuredLogs) {
const dvmWorkerCount = Array.isArray(settings?.dvm?.workers) ? settings.dvm.workers.length : 0
- this.process.stdout.write(`${JSON.stringify({
- level: levels.values.info,
- time: Date.now(),
- pid: process.pid,
- hostname: hostname(),
- msg: 'relay starting',
- version: packageJson.version,
- supportedNips: packageJson.supportedNips,
- payments: paymentsEnabled ? 'enabled' : 'disabled',
- paymentsProvider: paymentsEnabled ? path(['payments', 'processor'], settings) : undefined,
- eventStore: eventStoreBackend,
- clientWorkers: workerCount,
- maintenanceWorkers: 1,
- relayMonitor: settings.nip66?.enabled === true,
- dvmWorkers: dvmWorkerCount,
- })}\n`)
+ this.process.stdout.write(
+ `${JSON.stringify({
+ level: levels.values.info,
+ time: Date.now(),
+ pid: process.pid,
+ hostname: hostname(),
+ msg: 'relay starting',
+ version: packageJson.version,
+ supportedNips: packageJson.supportedNips,
+ payments: paymentsEnabled ? 'enabled' : 'disabled',
+ paymentsProvider: paymentsEnabled ? path(['payments', 'processor'], settings) : undefined,
+ eventStore: eventStoreBackend,
+ clientWorkers: workerCount,
+ maintenanceWorkers: 1,
+ relayMonitor: settings.nip66?.enabled === true,
+ dvmWorkers: dvmWorkerCount,
+ })}\n`,
+ )
}
const createWorker = (env: Record
) => {
@@ -158,6 +159,14 @@ export class App implements IRunnable {
logCentered('mirroring.static is no longer supported (ignored)')
}
+ const peeringPeers = settings.peering?.peers ?? []
+ if (peeringPeers.length > 0) {
+ createWorker({
+ WORKER_TYPE: 'peering-sync',
+ })
+ logCentered(`1 peering-sync worker started (${peeringPeers.length} peer(s) configured)`)
+ }
+
if (settings.nip66?.enabled) {
createWorker({
WORKER_TYPE: 'relay-monitor',
diff --git a/src/app/peering-sync-worker.ts b/src/app/peering-sync-worker.ts
new file mode 100644
index 000000000..fee9253b0
--- /dev/null
+++ b/src/app/peering-sync-worker.ts
@@ -0,0 +1,811 @@
+import { FSWatcher } from 'fs'
+import { z } from 'zod'
+import { ICacheAdapter } from '../@types/adapters'
+import { IRunnable } from '../@types/base'
+import { Event } from '../@types/event'
+import { PeeringPeerSyncState } from '../@types/peering'
+import { IEventRepository, IUserRepository } from '../@types/repositories'
+import { PeeringPeer, PeeringScope, Settings } from '../@types/settings'
+import { SubscriptionFilter } from '../@types/subscription'
+import { createLogger } from '../factories/logger-factory'
+import { idSchema, pubkeySchema, signatureSchema, tagSchema } from '../schemas/base-schema'
+import { createEventBatchPersister } from '../services/event-import-service'
+import { ADMIN_OPS_HEARTBEAT_KEEPALIVE_MS, publishAdminOpsWorkerHeartbeatSafely } from '../utils/admin-ops-store'
+import {
+ broadcastEvent,
+ getRelayPrivateKey,
+ isEphemeralEvent,
+ isEventIdValid,
+ isEventMatchingFilter,
+ isEventSignatureValid,
+ isExpiredEvent,
+ isRequestToVanishEvent,
+} from '../utils/event'
+import { delayMs } from '../utils/misc'
+import { FetchPeeringAdvertisementOptions, fetchPeeringAdvertisement } from '../utils/peering-advertisement'
+import {
+ AuthenticatedPeerConnection,
+ authenticateToPeeringRelay,
+ listEnabledPeeringPeers,
+ parsePeeringPeerPubkey,
+} from '../utils/peering-auth'
+import { PeeringConnection, PeeringFetchOverflowError, PeeringSyncConnection } from '../utils/peering-connection'
+import {
+ isCurrentPeeringScope,
+ peeringScopesToLiveFilters,
+ peeringScopeToFilters,
+ splitPeeringFilter,
+} from '../utils/peering-scope'
+import { PeeringSyncStateStore } from '../utils/peering-sync-state'
+import { SettingsStatic } from '../utils/settings'
+
+const logger = createLogger('peering-sync-worker')
+
+export const PEERING_SYNC_RECONCILE_INTERVAL_MS = 15 * 60_000
+export const PEERING_SYNC_RECONNECT_DELAY_MS = 5_000
+export const PEERING_SYNC_EVENT_BATCH_SIZE = 500
+export const PEERING_SYNC_PEER_REFRESH_MS = 30_000
+export const PEERING_SYNC_STATE_HEARTBEAT_MS = 60_000
+export const PEERING_SYNC_MAX_PENDING_EVENTS = 500
+export const PEERING_SYNC_MAX_PENDING_EVENT_BYTES = 8 * 1024 * 1024
+export const PEERING_SYNC_MAX_SCOPE_SPLIT_DEPTH = 20
+
+const eventShapeSchema = z
+ .object({
+ id: idSchema,
+ pubkey: pubkeySchema,
+ created_at: z.number().int().nonnegative(),
+ kind: z.number().int(),
+ tags: z.array(tagSchema),
+ content: z.string(),
+ sig: signatureSchema,
+ })
+ .strict()
+
+type AuthenticatePeerFn = (options: {
+ relayUrl: string
+ privateKey: string
+ timeoutMs?: number
+}) => Promise
+
+type FetchAdvertisementFn = (
+ relayUrl: string,
+ options?: FetchPeeringAdvertisementOptions,
+) => ReturnType
+
+export interface PeeringSyncWorkerOptions {
+ authenticate?: AuthenticatePeerFn
+ fetchAdvertisement?: FetchAdvertisementFn
+ createConnection?: (socket: AuthenticatedPeerConnection['socket']) => PeeringSyncConnection
+ stateStore?: PeeringSyncStateStore
+ opsCache?: ICacheAdapter
+ userRepository?: Pick
+ privateKey?: () => string
+ reconnectDelayMs?: number
+ reconcileIntervalMs?: number
+ peerRefreshMs?: number
+ stateHeartbeatMs?: number
+ eventBatchSize?: number
+}
+
+interface SessionCallbacks {
+ onReconciled: (error?: Error) => void
+}
+
+const normalizePeer = (peer: PeeringPeer): PeeringPeer | undefined => {
+ const pubkey = parsePeeringPeerPubkey(peer.pubkey)
+ if (!pubkey || peer.enabled === false) {
+ return undefined
+ }
+
+ return { ...peer, pubkey }
+}
+
+const newestCreatedAt = (items: { event_created_at: number }[]): number | undefined =>
+ items.length ? items[items.length - 1]!.event_created_at : undefined
+
+const isOversizedReconciliationError = (error: unknown): boolean => {
+ const message = error instanceof Error ? error.message : String(error)
+
+ return /too big|too many events|exceeds .*events|maximum .*events/i.test(message)
+}
+
+class PeeringPeerSession {
+ private stopped = false
+ private connection: PeeringSyncConnection | undefined
+ private reconcileTimer: NodeJS.Timeout | undefined
+ private reconcileRunning = false
+ private liveFilters: SubscriptionFilter[] = []
+ private scopes: PeeringScope[] = []
+ private state: PeeringPeerSyncState
+ private stateTimer: NodeJS.Timeout | undefined
+ private stateDirty = false
+ private readonly pendingEvents: Event[] = []
+ private pendingEventBytes = 0
+ private drainingEvents = false
+ private readonly persistEvents: (events: Event[]) => Promise
+ private stateHeartbeatTimer: NodeJS.Timeout | undefined
+
+ public constructor(
+ private readonly peer: PeeringPeer,
+ private readonly settings: () => Settings,
+ private readonly eventRepository: IEventRepository,
+ private readonly options: Required<
+ Pick
+ > &
+ Pick<
+ PeeringSyncWorkerOptions,
+ | 'stateStore'
+ | 'reconnectDelayMs'
+ | 'reconcileIntervalMs'
+ | 'eventBatchSize'
+ | 'userRepository'
+ | 'stateHeartbeatMs'
+ >,
+ private readonly callbacks: SessionCallbacks,
+ ) {
+ this.state = {
+ pubkey: peer.pubkey.toLowerCase(),
+ relayUrl: peer.relayUrl,
+ status: 'idle',
+ scopeCount: 0,
+ updatedAt: new Date().toISOString(),
+ }
+ this.persistEvents = createEventBatchPersister(eventRepository, {
+ userRepository: options.userRepository,
+ relayUrl: settings().info?.relay_url,
+ })
+ }
+
+ public start(): void {
+ this.stateHeartbeatTimer = setInterval(
+ () => this.publishStateHeartbeat(),
+ this.options.stateHeartbeatMs ?? PEERING_SYNC_STATE_HEARTBEAT_MS,
+ )
+ this.stateHeartbeatTimer.unref()
+ void this.runLoop()
+ }
+
+ public get id(): string {
+ return `${this.peer.pubkey}|${this.peer.relayUrl}`
+ }
+
+ public stop(): void {
+ if (this.stopped) {
+ return
+ }
+
+ this.stopped = true
+ clearInterval(this.reconcileTimer)
+ clearInterval(this.stateHeartbeatTimer)
+ this.stateHeartbeatTimer = undefined
+ if (this.stateTimer) {
+ clearTimeout(this.stateTimer)
+ this.stateTimer = undefined
+ }
+ this.connection?.close()
+ this.connection = undefined
+ this.pendingEvents.length = 0
+ this.pendingEventBytes = 0
+ this.queueState({ status: 'stopped' })
+ void this.flushState()
+ }
+
+ private async runLoop(): Promise {
+ const reconnectDelayMs = this.options.reconnectDelayMs ?? PEERING_SYNC_RECONNECT_DELAY_MS
+
+ while (!this.stopped) {
+ if (!this.isStillConfigured()) {
+ logger('peer %s is no longer enabled in settings; stopping its session', this.peer.relayUrl)
+ return
+ }
+
+ try {
+ await this.connectAndFollow()
+ } catch (error) {
+ if (this.stopped) {
+ break
+ }
+
+ const message = error instanceof Error ? error.message : String(error)
+ logger.error('peer %s sync failed: %s', this.peer.relayUrl, message)
+ this.queueState({ status: 'error', lastError: message })
+ this.callbacks.onReconciled(error instanceof Error ? error : new Error(message))
+ } finally {
+ this.connection?.close()
+ this.connection = undefined
+ clearInterval(this.reconcileTimer)
+ this.reconcileTimer = undefined
+ }
+
+ if (!this.stopped) {
+ await delayMs(reconnectDelayMs)
+ }
+ }
+ }
+
+ private isStillConfigured(): boolean {
+ const normalized = parsePeeringPeerPubkey(this.peer.pubkey)
+
+ return listEnabledPeeringPeers(this.settings()).some(
+ (peer) => parsePeeringPeerPubkey(peer.pubkey) === normalized && peer.relayUrl === this.peer.relayUrl,
+ )
+ }
+
+ private async connectAndFollow(): Promise {
+ this.queueState({ status: 'connecting', lastError: undefined })
+ const advertisement = await this.options.fetchAdvertisement(this.peer.relayUrl)
+ if (!advertisement.enabled) {
+ throw new Error('peer does not advertise peering')
+ }
+
+ if (!advertisement.supportedNips.includes(77)) {
+ throw new Error('peer does not advertise NIP-77')
+ }
+
+ const configuredPubkey = parsePeeringPeerPubkey(this.peer.pubkey)
+ if (configuredPubkey && advertisement.self && advertisement.self !== configuredPubkey) {
+ throw new Error(`peer advertises relay key ${advertisement.self} but is configured as ${configuredPubkey}`)
+ }
+
+ this.queueState({ scopeCount: advertisement.scopes.length })
+
+ const authenticated = await this.options.authenticate({
+ relayUrl: this.peer.relayUrl,
+ privateKey: this.options.privateKey(),
+ })
+
+ if (this.stopped) {
+ authenticated.socket.close()
+ return
+ }
+
+ const connection = this.options.createConnection(authenticated.socket)
+ this.connection = connection
+ this.scopes = advertisement.scopes
+
+ // Start the live feed first: a long history reconciliation can outlast the
+ // overlap window, and events published after its snapshot must still be
+ // delivered. Reconciliation and live writes are both duplicate-safe.
+ this.liveFilters = peeringScopesToLiveFilters(advertisement.scopes, Math.floor(Date.now() / 1000))
+ if (this.liveFilters.length) {
+ await connection.subscribe(this.liveFilters, (event) => this.queueEvent(event))
+ }
+
+ if (this.stopped) {
+ return
+ }
+
+ await this.reconcile(connection, advertisement.scopes)
+
+ if (this.stopped) {
+ return
+ }
+
+ this.queueState({ status: 'live' })
+ this.startReconcileTimer(connection, advertisement.scopes)
+
+ await connection.waitForClose()
+
+ if (!this.stopped) {
+ throw new Error('peer connection closed')
+ }
+ }
+
+ private startReconcileTimer(connection: PeeringSyncConnection, scopes: PeeringScope[]): void {
+ clearInterval(this.reconcileTimer)
+ const intervalMs = this.options.reconcileIntervalMs ?? PEERING_SYNC_RECONCILE_INTERVAL_MS
+
+ this.reconcileTimer = setInterval(() => {
+ void this.reconcile(connection, scopes).catch((error) => {
+ const message = error instanceof Error ? error.message : String(error)
+ logger.error('periodic reconciliation with %s failed: %s', this.peer.relayUrl, message)
+ this.queueState({ status: 'error', lastError: message })
+ this.callbacks.onReconciled(error instanceof Error ? error : new Error(message))
+ })
+ }, intervalMs)
+ this.reconcileTimer.unref()
+ }
+
+ private async reconcile(connection: PeeringSyncConnection, scopes: PeeringScope[]): Promise {
+ if (this.reconcileRunning || this.stopped) {
+ return
+ }
+
+ this.reconcileRunning = true
+ const startedAt = new Date().toISOString()
+ const startedAtSeconds = Math.floor(Date.now() / 1000)
+ this.queueState({
+ status: 'reconciling',
+ lastSyncStartedAt: startedAt,
+ })
+
+ try {
+ const needed = new Map()
+ const currentFilters: SubscriptionFilter[] = []
+ let newestCurrent: number | undefined
+
+ for (const scope of scopes) {
+ if (this.stopped) {
+ return
+ }
+
+ const filters = peeringScopeToFilters(scope, startedAtSeconds)
+ const countCurrent = isCurrentPeeringScope(scope, startedAtSeconds)
+ if (countCurrent) {
+ currentFilters.push(...filters)
+ }
+
+ for (const filter of filters) {
+ newestCurrent = await this.reconcileFilter(
+ connection,
+ filter,
+ startedAtSeconds,
+ needed,
+ newestCurrent,
+ countCurrent,
+ )
+ }
+ }
+
+ const ids = [...needed.keys()]
+ const batchSize = this.options.eventBatchSize ?? PEERING_SYNC_EVENT_BATCH_SIZE
+ let received = 0
+
+ for (let offset = 0; offset < ids.length; offset += batchSize) {
+ if (this.stopped) {
+ return
+ }
+
+ const batch = ids.slice(offset, offset + batchSize)
+ const result = await this.fetchAndStoreBatch(connection, batch, needed)
+
+ received += result.stored
+ for (const event of result.events) {
+ if (currentFilters.some((filter) => isEventMatchingFilter(filter)(event))) {
+ newestCurrent = newestCurrent === undefined ? event.created_at : Math.max(newestCurrent, event.created_at)
+ }
+ }
+ }
+
+ const finishedAt = new Date().toISOString()
+ const finishedAtSeconds = Math.floor(Date.now() / 1000)
+ this.queueState({
+ status: 'live',
+ lastSyncFinishedAt: finishedAt,
+ lastReconciledAt: finishedAt,
+ lastError: undefined,
+ missingEvents: ids.length,
+ receivedEvents: received,
+ ...(newestCurrent !== undefined ? { latestEventCreatedAt: newestCurrent } : {}),
+ ...(newestCurrent !== undefined ? { lagSeconds: Math.max(0, finishedAtSeconds - newestCurrent) } : {}),
+ })
+ this.callbacks.onReconciled()
+ logger(
+ 'reconciled %s over %d scope(s): %d missing, %d received',
+ this.peer.relayUrl,
+ scopes.length,
+ ids.length,
+ received,
+ )
+ } finally {
+ this.reconcileRunning = false
+ }
+ }
+
+ private async reconcileFilter(
+ connection: PeeringSyncConnection,
+ filter: SubscriptionFilter,
+ nowSeconds: number,
+ needed: Map,
+ newestCurrent: number | undefined,
+ countCurrent: boolean,
+ depth = 0,
+ ): Promise {
+ if (this.stopped) {
+ return newestCurrent
+ }
+
+ const items = await this.eventRepository.findReconciliationItems(filter)
+ const newest = newestCreatedAt(items)
+
+ if (countCurrent && newest !== undefined) {
+ newestCurrent = newestCurrent === undefined ? newest : Math.max(newestCurrent, newest)
+ }
+
+ try {
+ const result = await connection.reconcile(filter, items)
+ for (const id of result.needIds) {
+ if (!needed.has(id)) {
+ needed.set(id, filter)
+ }
+ }
+
+ return newestCurrent
+ } catch (error) {
+ if (depth >= PEERING_SYNC_MAX_SCOPE_SPLIT_DEPTH || !isOversizedReconciliationError(error)) {
+ throw error
+ }
+
+ const split = splitPeeringFilter(filter, nowSeconds)
+ if (!split) {
+ throw error
+ }
+
+ logger.warn(
+ 'peer %s rejected a filter as too large; splitting it in time (depth %d)',
+ this.peer.relayUrl,
+ depth + 1,
+ )
+
+ let current = newestCurrent
+ for (const child of split) {
+ current = await this.reconcileFilter(connection, child, nowSeconds, needed, current, countCurrent, depth + 1)
+ }
+
+ return current
+ }
+ }
+
+ private queueEvent(event: Event): void {
+ const size = Buffer.byteLength(JSON.stringify(event))
+ if (
+ this.pendingEvents.length >= PEERING_SYNC_MAX_PENDING_EVENTS ||
+ this.pendingEventBytes + size > PEERING_SYNC_MAX_PENDING_EVENT_BYTES
+ ) {
+ logger.warn('peer %s live event backlog exceeded; reconnecting', this.peer.relayUrl)
+ this.connection?.close()
+ return
+ }
+
+ this.pendingEvents.push(event)
+ this.pendingEventBytes += size
+ void this.drainEvents()
+ }
+
+ private async drainEvents(): Promise {
+ if (this.drainingEvents) {
+ return
+ }
+
+ this.drainingEvents = true
+ try {
+ while (this.pendingEvents.length > 0 && !this.stopped) {
+ const event = this.pendingEvents.shift() as Event
+ this.pendingEventBytes = Math.max(0, this.pendingEventBytes - Buffer.byteLength(JSON.stringify(event)))
+
+ try {
+ await this.applyLiveEvent(event)
+ } catch (error) {
+ // One bad event must not poison the queue or become an unhandled
+ // rejection; the next reconciliation can retry it.
+ logger.error('failed to apply live event from %s: %o', this.peer.relayUrl, error)
+ }
+ }
+ } finally {
+ this.drainingEvents = false
+ }
+ }
+
+ private async applyLiveEvent(event: Event): Promise {
+ const nowSeconds = Math.floor(Date.now() / 1000)
+ // Persistence uses the scope's own bounds, not the subscription overlap:
+ // a late publish older than 60s is still valid for a scope without a
+ // rolling window.
+ const currentFilters = this.scopes
+ .filter((scope) => isCurrentPeeringScope(scope, nowSeconds))
+ .flatMap((scope) => peeringScopeToFilters(scope, nowSeconds))
+
+ if (!currentFilters.some((filter) => isEventMatchingFilter(filter)(event))) {
+ return
+ }
+
+ await this.acceptEvents([event], undefined, true)
+ const latestCreatedAt = Math.max(this.state.latestEventCreatedAt ?? 0, event.created_at)
+
+ this.queueState({
+ lastEventAt: new Date().toISOString(),
+ latestEventCreatedAt: latestCreatedAt,
+ lagSeconds: Math.max(0, nowSeconds - latestCreatedAt),
+ })
+ }
+
+ private async fetchAndStoreBatch(
+ connection: PeeringSyncConnection,
+ ids: string[],
+ needed: Map,
+ ): Promise<{ stored: number; events: Event[] }> {
+ try {
+ const events = await connection.fetchEvents(ids)
+ const accepted = await this.acceptEvents(events, needed)
+
+ return { stored: accepted.stored, events: accepted.events }
+ } catch (error) {
+ if (!(error instanceof PeeringFetchOverflowError)) {
+ throw error
+ }
+
+ if (ids.length <= 1) {
+ logger.warn(
+ 'peer %s sent an oversized event for %s; leaving it for the next reconciliation',
+ this.peer.relayUrl,
+ ids[0],
+ )
+ return { stored: 0, events: [] }
+ }
+
+ const midpoint = Math.ceil(ids.length / 2)
+ const left = await this.fetchAndStoreBatch(connection, ids.slice(0, midpoint), needed)
+ const right = await this.fetchAndStoreBatch(connection, ids.slice(midpoint), needed)
+
+ return {
+ stored: left.stored + right.stored,
+ events: [...left.events, ...right.events],
+ }
+ }
+ }
+
+ /**
+ * Accepts only events that were asked for and still match the scope, then
+ * verifies the event id and signature. A peer cannot inject an arbitrary
+ * event through a reconciliation reply, and a compromised transport cannot
+ * alter one that the peer did sign.
+ */
+ private async acceptEvents(
+ events: Event[],
+ scopeById?: Map,
+ broadcast = false,
+ ): Promise<{ events: Event[]; stored: number }> {
+ const accepted: Event[] = []
+
+ for (const event of events) {
+ if (scopeById && !scopeById.has(event.id)) {
+ continue
+ }
+
+ if (!eventShapeSchema.safeParse(event).success) {
+ continue
+ }
+
+ const scopeFilter = scopeById?.get(event.id)
+ if (scopeFilter && !isEventMatchingFilter(scopeFilter)(event)) {
+ continue
+ }
+
+ if (isExpiredEvent(event)) {
+ continue
+ }
+
+ if (!(await isEventIdValid(event)) || !(await isEventSignatureValid(event))) {
+ logger.warn('peer %s sent an event with an invalid id or signature: %s', this.peer.relayUrl, event.id)
+ continue
+ }
+
+ accepted.push(event)
+ }
+
+ if (accepted.length === 0) {
+ return { events: [], stored: 0 }
+ }
+
+ let stored = 0
+
+ if (broadcast) {
+ // Live traffic is one event at a time, so we can tell which write
+ // actually stored or changed something. Only those are rebroadcast:
+ // an overlap replay of an event both relays already hold must not echo.
+ for (const event of accepted) {
+ const count = await this.persistEvents([event])
+ stored += count
+
+ if ((count > 0 || isEphemeralEvent(event)) && !isRequestToVanishEvent(event)) {
+ void broadcastEvent(event).catch((error) => {
+ logger.error('failed to broadcast synced event %s: %o', event.id, error)
+ })
+ }
+ }
+
+ return { events: accepted, stored }
+ }
+
+ // Backfill is deliberately not broadcast: those events predate any live
+ // subscription and a full sync must not fan out a dump.
+ stored = await this.persistEvents(accepted)
+
+ return { events: accepted, stored }
+ }
+
+ private queueState(partial: Partial): void {
+ if (this.stopped && partial.status !== 'stopped') {
+ return
+ }
+
+ this.state = {
+ ...this.state,
+ ...partial,
+ updatedAt: new Date().toISOString(),
+ }
+ this.stateDirty = true
+
+ if (this.stateTimer) {
+ return
+ }
+
+ this.stateTimer = setTimeout(() => {
+ this.stateTimer = undefined
+ void this.flushState()
+ }, 250)
+ this.stateTimer.unref()
+ }
+
+ private async flushState(): Promise {
+ if (!this.options.stateStore || !this.stateDirty) {
+ return
+ }
+
+ this.stateDirty = false
+ try {
+ await this.options.stateStore.publish(this.state)
+ } catch (error) {
+ logger.warn('failed to record sync state for %s: %o', this.peer.relayUrl, error)
+ }
+ }
+
+ /** Keeps a long reconciliation from looking stale in the admin console. */
+ private publishStateHeartbeat(): void {
+ if (this.stopped || !this.options.stateStore) {
+ return
+ }
+
+ const nowSeconds = Math.floor(Date.now() / 1000)
+ this.state = {
+ ...this.state,
+ updatedAt: new Date().toISOString(),
+ ...(this.state.latestEventCreatedAt !== undefined
+ ? { lagSeconds: Math.max(0, nowSeconds - this.state.latestEventCreatedAt) }
+ : {}),
+ }
+ this.stateDirty = true
+ void this.flushState()
+ }
+}
+
+export class PeeringSyncWorker implements IRunnable {
+ private sessions = new Map()
+ private heartbeatTimer: NodeJS.Timeout | undefined
+ private peerRefreshTimer: NodeJS.Timeout | undefined
+ private watchers: FSWatcher[] | undefined
+ private lastRunFinishedAt: string | undefined
+ private lastError: string | undefined
+ private readonly startedAt = new Date().toISOString()
+
+ public constructor(
+ private readonly process: NodeJS.Process,
+ private readonly settings: () => Settings,
+ private readonly eventRepository: IEventRepository,
+ private readonly options: PeeringSyncWorkerOptions = {},
+ ) {
+ 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 {
+ this.watchers = SettingsStatic.watchSettings()
+ this.applyConfiguredPeers()
+
+ this.publishHeartbeat()
+ this.heartbeatTimer = setInterval(() => this.publishHeartbeat(), ADMIN_OPS_HEARTBEAT_KEEPALIVE_MS)
+ this.heartbeatTimer.unref()
+
+ // `watchSettings` invalidates the cached settings document; this periodic
+ // pass also replaces a session when only its relay URL changed, which a
+ // pubkey-only configuration check would miss.
+ this.peerRefreshTimer = setInterval(
+ () => this.applyConfiguredPeers(),
+ this.options.peerRefreshMs ?? PEERING_SYNC_PEER_REFRESH_MS,
+ )
+ this.peerRefreshTimer.unref()
+ }
+
+ private sessionKey(peer: PeeringPeer): string {
+ return `${parsePeeringPeerPubkey(peer.pubkey) ?? peer.pubkey}|${peer.relayUrl}`
+ }
+
+ private applyConfiguredPeers(): void {
+ const peers = listEnabledPeeringPeers(this.settings())
+ .map(normalizePeer)
+ .filter((peer): peer is PeeringPeer => peer !== undefined)
+ const desired = new Map(peers.map((peer) => [this.sessionKey(peer), peer]))
+
+ for (const [id, session] of this.sessions) {
+ if (!desired.has(id)) {
+ logger('stopping sync with removed or changed peer %s', session.id)
+ session.stop()
+ this.sessions.delete(id)
+ }
+ }
+
+ const commonOptions = {
+ authenticate: this.options.authenticate ?? authenticateToPeeringRelay,
+ fetchAdvertisement: this.options.fetchAdvertisement ?? fetchPeeringAdvertisement,
+ createConnection: this.options.createConnection ?? ((socket) => new PeeringConnection(socket)),
+ privateKey: this.options.privateKey ?? (() => getRelayPrivateKey(this.settings().info?.relay_url)),
+ stateStore: this.options.stateStore,
+ userRepository: this.options.userRepository,
+ reconnectDelayMs: this.options.reconnectDelayMs,
+ reconcileIntervalMs: this.options.reconcileIntervalMs,
+ eventBatchSize: this.options.eventBatchSize,
+ stateHeartbeatMs: this.options.stateHeartbeatMs,
+ }
+
+ for (const [id, peer] of desired) {
+ if (this.sessions.has(id)) {
+ continue
+ }
+
+ const session = new PeeringPeerSession(peer, this.settings, this.eventRepository, commonOptions, {
+ onReconciled: (error) => {
+ if (error) {
+ this.lastError = error.message
+ return
+ }
+ this.lastRunFinishedAt = new Date().toISOString()
+ this.lastError = undefined
+ },
+ })
+
+ logger('starting sync with peer %s', peer.relayUrl)
+ this.sessions.set(id, session)
+ session.start()
+ }
+
+ if (this.sessions.size === 0) {
+ logger('no enabled peering peers configured; worker idle')
+ }
+ }
+
+ private publishHeartbeat(): void {
+ void publishAdminOpsWorkerHeartbeatSafely(this.options.opsCache, {
+ type: 'peering-sync',
+ updatedAt: new Date().toISOString(),
+ lastRunStartedAt: this.startedAt,
+ lastRunFinishedAt: this.lastRunFinishedAt,
+ lastError: this.lastError,
+ })
+ }
+
+ private onError(error: Error): void {
+ logger('error: %o', error)
+ this.lastError = error.message
+ }
+
+ private onExit(): void {
+ logger('exiting')
+ this.close(() => {
+ this.process.exit(0)
+ })
+ }
+
+ public close(callback?: () => void): void {
+ logger('closing')
+ clearInterval(this.heartbeatTimer)
+ this.heartbeatTimer = undefined
+ clearInterval(this.peerRefreshTimer)
+ this.peerRefreshTimer = undefined
+ if (Array.isArray(this.watchers)) {
+ for (const watcher of this.watchers) {
+ watcher.close()
+ }
+ }
+ this.watchers = undefined
+ for (const session of this.sessions.values()) {
+ session.stop()
+ }
+ this.sessions.clear()
+ callback?.()
+ }
+}
diff --git a/src/controllers/admin/get-admin-peering-controller.ts b/src/controllers/admin/get-admin-peering-controller.ts
new file mode 100644
index 000000000..3ae4d470e
--- /dev/null
+++ b/src/controllers/admin/get-admin-peering-controller.ts
@@ -0,0 +1,21 @@
+import { Request, Response } from 'express'
+
+import { IController } from '../../@types/controllers'
+import { Settings } from '../../@types/settings'
+import { mergePeeringRows } from '../../utils/admin-peering'
+import { PeeringSyncStateStore } from '../../utils/peering-sync-state'
+import { loadMergedSettings } from '../../utils/settings-config'
+
+export class GetAdminPeeringController implements IController {
+ public constructor(
+ private readonly stateStore: PeeringSyncStateStore,
+ private readonly getSettings: () => Settings = loadMergedSettings,
+ ) {}
+
+ public async handleRequest(_request: Request, response: Response): Promise {
+ const [states, settings] = await Promise.all([this.stateStore.list(), this.getSettings()])
+ const peers = mergePeeringRows(settings, states)
+
+ response.status(200).setHeader('content-type', 'application/json').send({ peers })
+ }
+}
diff --git a/src/factories/controllers/get-admin-peering-controller-factory.ts b/src/factories/controllers/get-admin-peering-controller-factory.ts
new file mode 100644
index 000000000..bfc2bd47f
--- /dev/null
+++ b/src/factories/controllers/get-admin-peering-controller-factory.ts
@@ -0,0 +1,19 @@
+import { IController } from '../../@types/controllers'
+import { RedisAdapter } from '../../adapters/redis-adapter'
+import { getCacheClient } from '../../cache/client'
+import { GetAdminPeeringController } from '../../controllers/admin/get-admin-peering-controller'
+import { PeeringSyncStateStore } from '../../utils/peering-sync-state'
+
+export const createGetAdminPeeringController = (): IController => {
+ // The RedisAdapter constructor attaches connection listeners. Reusing one
+ // adapter keeps repeated dashboard polls from stacking listeners on the
+ // shared cache client.
+ if (!sharedController) {
+ const stateStore = new PeeringSyncStateStore(new RedisAdapter(getCacheClient()))
+ sharedController = new GetAdminPeeringController(stateStore)
+ }
+
+ return sharedController
+}
+
+let sharedController: IController | undefined
diff --git a/src/factories/peering-sync-worker-factory.ts b/src/factories/peering-sync-worker-factory.ts
new file mode 100644
index 000000000..a28f7771f
--- /dev/null
+++ b/src/factories/peering-sync-worker-factory.ts
@@ -0,0 +1,19 @@
+import { PeeringSyncWorker } from '../app/peering-sync-worker'
+import { getMasterDbClient, getReadReplicaDbClient } from '../database/client'
+import { UserRepository } from '../repositories/user-repository'
+import { PeeringSyncStateStore } from '../utils/peering-sync-state'
+import { createAdminOpsCacheAdapter } from './admin-ops-cache-adapter-factory'
+import { createEventRepository } from './event-repository-factory'
+import { createSettings } from './settings-factory'
+
+export const peeringSyncWorkerFactory = () => {
+ const cache = createAdminOpsCacheAdapter()
+ const dbClient = getMasterDbClient()
+ const eventRepository = createEventRepository(dbClient, getReadReplicaDbClient())
+ const userRepository = new UserRepository(dbClient, eventRepository)
+
+ return new PeeringSyncWorker(process, createSettings, eventRepository, {
+ ...(cache ? { opsCache: cache, stateStore: new PeeringSyncStateStore(cache) } : {}),
+ userRepository,
+ })
+}
diff --git a/src/index.ts b/src/index.ts
index b909447ca..87b86a66f 100644
--- a/src/index.ts
+++ b/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 { peeringSyncWorkerFactory } from './factories/peering-sync-worker-factory'
import { relayMonitorWorkerFactory } from './factories/relay-monitor-worker-factory'
import { workerFactory } from './factories/worker-factory'
import { initializeMetricsTelemetry } from './telemetry/metrics'
@@ -16,6 +17,8 @@ export const getRunner = () => {
return workerFactory()
case 'maintenance':
return maintenanceWorkerFactory()
+ case 'peering-sync':
+ return peeringSyncWorkerFactory()
case 'relay-monitor':
return relayMonitorWorkerFactory()
case 'dvm-orchestrator':
diff --git a/src/routes/admin/index.ts b/src/routes/admin/index.ts
index 8dd5317a5..58590d8e9 100644
--- a/src/routes/admin/index.ts
+++ b/src/routes/admin/index.ts
@@ -1,33 +1,33 @@
import express, { json, Router } from 'express'
-
-import { createGetAdminNotificationDeliveryLogController } from '../../factories/controllers/get-admin-notification-delivery-log-controller-factory'
-import { createGetAdminNotificationsController } from '../../factories/controllers/get-admin-notifications-controller-factory'
-import { createPatchAdminNotificationsController } from '../../factories/controllers/patch-admin-notifications-controller-factory'
+import { createGetAdminAuditLogController } from '../../factories/controllers/get-admin-audit-log-controller-factory'
+import { createGetAdminDiagnosticsExportController } from '../../factories/controllers/get-admin-diagnostics-export-controller-factory'
import { createGetAdminHealthController } from '../../factories/controllers/get-admin-health-controller-factory'
import { createGetAdminMetricsController } from '../../factories/controllers/get-admin-metrics-controller-factory'
+import { createGetAdminModerationLimitsController } from '../../factories/controllers/get-admin-moderation-limits-controller-factory'
+import { createGetAdminModerationStatsController } from '../../factories/controllers/get-admin-moderation-stats-controller-factory'
import { createGetAdminNetworkHealthController } from '../../factories/controllers/get-admin-network-health-controller-factory'
import { createGetAdminNetworkHealthHistoryController } from '../../factories/controllers/get-admin-network-health-history-controller-factory'
+import { createGetAdminNotificationDeliveryLogController } from '../../factories/controllers/get-admin-notification-delivery-log-controller-factory'
+import { createGetAdminNotificationsController } from '../../factories/controllers/get-admin-notifications-controller-factory'
+import { createGetAdminOpsNip05VerificationsController } from '../../factories/controllers/get-admin-ops-nip05-verifications-controller-factory'
+import { createGetAdminOpsStorageCountsController } from '../../factories/controllers/get-admin-ops-storage-counts-controller-factory'
+import { createGetAdminOpsWorkersController } from '../../factories/controllers/get-admin-ops-workers-controller-factory'
+import { createGetAdminPeeringController } from '../../factories/controllers/get-admin-peering-controller-factory'
import { createGetAdminSessionController } from '../../factories/controllers/get-admin-session-controller-factory'
import { createGetAdminSettingsBackupsController } from '../../factories/controllers/get-admin-settings-backups-controller-factory'
import { createGetAdminSettingsController } from '../../factories/controllers/get-admin-settings-controller-factory'
import { createGetAdminSettingsSchemaController } from '../../factories/controllers/get-admin-settings-schema-controller-factory'
+import { createPatchAdminModerationLimitsController } from '../../factories/controllers/patch-admin-moderation-limits-controller-factory'
+import { createPatchAdminNotificationsController } from '../../factories/controllers/patch-admin-notifications-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 { createPostAdminModerationActionsController } from '../../factories/controllers/post-admin-moderation-actions-controller-factory'
+import { createPostAdminNetworkHealthRunController } from '../../factories/controllers/post-admin-network-health-run-controller-factory'
+import { createPostAdminNotificationTestController } from '../../factories/controllers/post-admin-notification-test-controller-factory'
+import { createPostAdminOpsNip05ReverifyController } from '../../factories/controllers/post-admin-ops-nip05-reverify-controller-factory'
import { createPostAdminSettingsRestoreController } from '../../factories/controllers/post-admin-settings-restore-controller-factory'
import { createPostAdminSettingsValidateController } from '../../factories/controllers/post-admin-settings-validate-controller-factory'
-import { createGetAdminOpsWorkersController } from '../../factories/controllers/get-admin-ops-workers-controller-factory'
-import { createGetAdminOpsNip05VerificationsController } from '../../factories/controllers/get-admin-ops-nip05-verifications-controller-factory'
-import { createGetAdminOpsStorageCountsController } from '../../factories/controllers/get-admin-ops-storage-counts-controller-factory'
-import { createPostAdminOpsNip05ReverifyController } from '../../factories/controllers/post-admin-ops-nip05-reverify-controller-factory'
-import { createGetAdminAuditLogController } from '../../factories/controllers/get-admin-audit-log-controller-factory'
-import { createGetAdminDiagnosticsExportController } from '../../factories/controllers/get-admin-diagnostics-export-controller-factory'
-import { createPostAdminNetworkHealthRunController } from '../../factories/controllers/post-admin-network-health-run-controller-factory'
-import { createGetAdminModerationLimitsController } from '../../factories/controllers/get-admin-moderation-limits-controller-factory'
-import { createGetAdminModerationStatsController } from '../../factories/controllers/get-admin-moderation-stats-controller-factory'
-import { createPatchAdminModerationLimitsController } from '../../factories/controllers/patch-admin-moderation-limits-controller-factory'
-import { createPostAdminModerationActionsController } from '../../factories/controllers/post-admin-moderation-actions-controller-factory'
import { adminAuthGateMiddleware, adminAuthMiddleware } from '../../handlers/request-handlers/admin-auth-middleware'
import { adminEnabledMiddleware } from '../../handlers/request-handlers/admin-enabled-middleware'
import { adminJsonBodyMiddleware } from '../../handlers/request-handlers/admin-json-body-middleware'
@@ -206,6 +206,12 @@ router.get(
adminAuthMiddleware,
withAdminController(createGetAdminOpsWorkersController),
)
+router.get(
+ '/peering',
+ adminRateLimitMiddleware,
+ adminAuthMiddleware,
+ withAdminController(createGetAdminPeeringController),
+)
router.get(
'/ops/nip05-verifications',
adminRateLimitMiddleware,
diff --git a/src/services/event-import-service.ts b/src/services/event-import-service.ts
index 460ac4de9..7f498c0dc 100644
--- a/src/services/event-import-service.ts
+++ b/src/services/event-import-service.ts
@@ -6,6 +6,11 @@ const streamArrayModule = require('stream-json/streamers/stream-array.js') as {
}
const streamArray = streamArrayModule.default
+import { Event } from '../@types/event'
+import { IEventRepository, IUserRepository } from '../@types/repositories'
+
+import { EventDeduplicationMetadataKey, EventExpirationTimeMetadataKey, EventKinds, EventTags } from '../constants/base'
+import { eventSchema } from '../schemas/event-schema'
import {
getEventExpiration,
isDeleteEvent,
@@ -14,14 +19,10 @@ import {
isEventSignatureValid,
isParameterizedReplaceableEvent,
isReplaceableEvent,
+ isRequestToVanishEvent,
} from '../utils/event'
import { attemptValidation } from '../utils/validation'
-import { EventDeduplicationMetadataKey, EventExpirationTimeMetadataKey, EventTags } from '../constants/base'
-import { Event } from '../@types/event'
-import { eventSchema } from '../schemas/event-schema'
-import { IEventRepository } from '../@types/repositories'
-
type EventWithImportMetadata = Event & {
[EventDeduplicationMetadataKey]?: string[]
[EventExpirationTimeMetadataKey]?: number
@@ -88,8 +89,19 @@ const isDestroyableStream = (
return typeof candidate.destroy === 'function'
}
+export interface EventBatchPersisterOptions {
+ /**
+ * When present, events from vanished authors are skipped and a kind-62
+ * request applies the same deletion and vanish-state rules as the live
+ * event path. Import leaves this out and keeps its historical behavior.
+ */
+ userRepository?: Pick
+ /** This relay's URL, used to honor kind-62 `r` tag targeting. */
+ relayUrl?: string
+}
+
export const createEventBatchPersister =
- (eventRepository: IEventRepository) =>
+ (eventRepository: IEventRepository, options: EventBatchPersisterOptions = {}) =>
async (events: Event[]): Promise => {
if (!events.length) {
return 0
@@ -99,16 +111,51 @@ export const createEventBatchPersister =
const regularEvents: Event[] = []
const replaceableEvents: Event[] = []
+ // A batch often carries many events from the same author. Ask once per
+ // author instead of once per event.
+ const vanishedByAuthor = new Map>()
+ const isVanished = (pubkey: string): Promise => {
+ if (!options.userRepository) {
+ return Promise.resolve(false)
+ }
+
+ let result = vanishedByAuthor.get(pubkey)
+ if (!result) {
+ result = options.userRepository.isVanished(pubkey)
+ vanishedByAuthor.set(pubkey, result)
+ }
+
+ return result
+ }
+
+ const flush = async () => {
+ inserted += await eventRepository.createMany(regularEvents.splice(0))
+ inserted += await eventRepository.upsertMany(replaceableEvents.splice(0))
+ }
for (const event of events) {
+ if (options.userRepository && isRequestToVanishEvent(event, options.relayUrl)) {
+ // Flush pending writes before deleting, so the vanish applies after
+ // everything accepted ahead of it in this batch.
+ await flush()
+ await eventRepository.deleteByPubkeyExceptKinds(event.pubkey, [EventKinds.REQUEST_TO_VANISH])
+ inserted += await eventRepository.create(enrichEventMetadata(event))
+ await options.userRepository.setVanished(event.pubkey, true)
+ vanishedByAuthor.set(event.pubkey, Promise.resolve(true))
+ continue
+ }
+
+ if (await isVanished(event.pubkey)) {
+ continue
+ }
+
if (isEphemeralEvent(event)) {
continue
}
if (isDeleteEvent(event)) {
// flush pending batches before applying deletes
- inserted += await eventRepository.createMany(regularEvents.splice(0))
- inserted += await eventRepository.upsertMany(replaceableEvents.splice(0))
+ await flush()
const eventIdsToDelete = event.tags.reduce(
(ids, tag) =>
@@ -135,16 +182,13 @@ export const createEventBatchPersister =
}
// flush remaining
- inserted += await eventRepository.createMany(regularEvents)
- inserted += await eventRepository.upsertMany(replaceableEvents)
+ await flush()
return inserted
}
export class EventImportService {
- public constructor(
- private readonly persistBatch: (events: Event[]) => Promise,
- ) {}
+ public constructor(private readonly persistBatch: (events: Event[]) => Promise) {}
public async importFromReadable(
input: NodeJS.ReadableStream,
diff --git a/src/utils/admin-diagnostics-export.ts b/src/utils/admin-diagnostics-export.ts
index 0511eb3a1..ade762e62 100644
--- a/src/utils/admin-diagnostics-export.ts
+++ b/src/utils/admin-diagnostics-export.ts
@@ -4,9 +4,10 @@ import { join } from 'path'
import { Settings } from '../@types/settings'
import { RedisAdapter } from '../adapters/redis-adapter'
import { getCacheClient } from '../cache/client'
-import { AdminOpsStore } from './admin-ops-store'
import { collectAdminHealthSnapshot } from './admin-health'
import { getRedactedAdminNotifications } from './admin-notifications-settings'
+import { AdminOpsStore } from './admin-ops-store'
+import { PeeringSyncStateStore } from './peering-sync-state'
import { RelayProbeSnapshotStore } from './relay-probe-snapshot'
import { loadMergedSettings } from './settings-config'
import { redactSettingsSecrets } from './settings-redaction'
@@ -22,6 +23,7 @@ export interface AdminDiagnosticsExport {
}
operations: {
workers: Awaited>
+ peering: Awaited>
}
}
@@ -41,11 +43,13 @@ export const buildAdminDiagnosticsExport = async (
const cache = new RedisAdapter(getCacheClient())
const opsStore = new AdminOpsStore(cache)
const snapshotStore = new RelayProbeSnapshotStore(cache)
+ const peeringStore = new PeeringSyncStateStore(cache)
- const [health, workers, latest] = await Promise.all([
+ const [health, workers, latest, peering] = await Promise.all([
collectAdminHealthSnapshot(),
opsStore.listWorkerHeartbeats(),
snapshotStore.getLatest(),
+ peeringStore.list(),
])
const settings = redactSettingsSecrets(getSettings()) as Settings
@@ -57,6 +61,6 @@ export const buildAdminDiagnosticsExport = async (
settings,
notifications: getRedactedAdminNotifications(),
networkHealth: { latest },
- operations: { workers },
+ operations: { workers, peering },
}
}
diff --git a/src/utils/admin-ops-config.ts b/src/utils/admin-ops-config.ts
index 6fdaee741..c3369a49f 100644
--- a/src/utils/admin-ops-config.ts
+++ b/src/utils/admin-ops-config.ts
@@ -30,6 +30,14 @@ export const listConfiguredWorkers = (settings: Settings): AdminOpsConfiguredWor
label: 'Maintenance',
})
+ if ((settings.peering?.peers?.length ?? 0) > 0) {
+ workers.push({
+ type: 'peering-sync',
+ enabled: true,
+ label: 'Peering sync',
+ })
+ }
+
if (settings.nip66?.enabled) {
workers.push({
type: 'relay-monitor',
diff --git a/src/utils/admin-peering.ts b/src/utils/admin-peering.ts
new file mode 100644
index 000000000..ca37f8abd
--- /dev/null
+++ b/src/utils/admin-peering.ts
@@ -0,0 +1,63 @@
+import { PeeringPeerSyncState } from '../@types/peering'
+import { Settings } from '../@types/settings'
+import { parsePeeringPeerPubkey } from './peering-auth'
+import { isPeeringSyncStateStale, PEERING_SYNC_STATE_STALE_AFTER_SECONDS } from './peering-sync-state'
+
+export interface AdminPeeringRow {
+ pubkey: string
+ relayUrl: string
+ configured: boolean
+ enabled: boolean
+ stale: boolean
+ state?: PeeringPeerSyncState
+}
+
+export const mergePeeringRows = (
+ settings: Settings,
+ states: PeeringPeerSyncState[],
+ staleAfterSeconds = PEERING_SYNC_STATE_STALE_AFTER_SECONDS,
+): AdminPeeringRow[] => {
+ const statesByPubkey = new Map(states.map((state) => [state.pubkey.toLowerCase(), state]))
+ const rows: AdminPeeringRow[] = []
+ const matched = new Set()
+
+ for (const peer of settings.peering?.peers ?? []) {
+ const pubkey = parsePeeringPeerPubkey(peer.pubkey) ?? peer.pubkey
+ const normalized = pubkey.toLowerCase()
+ const stored = statesByPubkey.get(normalized)
+ const state = stored?.relayUrl === peer.relayUrl ? stored : undefined
+ const enabled = peer.enabled !== false
+
+ matched.add(normalized)
+ rows.push({
+ pubkey,
+ relayUrl: peer.relayUrl,
+ configured: true,
+ enabled,
+ stale: enabled && (!state || isPeeringSyncStateStale(state.updatedAt, staleAfterSeconds)),
+ state,
+ })
+ }
+
+ // A peer removed from settings can still have a stored state row until its TTL
+ // expires. Showing it makes a stale relationship obvious instead of hiding it.
+ for (const state of states) {
+ if (matched.has(state.pubkey.toLowerCase())) {
+ continue
+ }
+
+ rows.push({
+ pubkey: state.pubkey,
+ relayUrl: state.relayUrl,
+ configured: false,
+ enabled: false,
+ stale: isPeeringSyncStateStale(state.updatedAt, staleAfterSeconds),
+ state,
+ })
+ }
+
+ return rows.sort((left, right) => {
+ const relayCompare = left.relayUrl.localeCompare(right.relayUrl)
+ return relayCompare !== 0 ? relayCompare : left.pubkey.localeCompare(right.pubkey)
+ })
+}
diff --git a/src/utils/event.ts b/src/utils/event.ts
index fe4916661..1e2f28810 100644
--- a/src/utils/event.ts
+++ b/src/utils/event.ts
@@ -1,16 +1,16 @@
import * as secp256k1 from '@noble/secp256k1'
-import { ALL_RELAYS, EventKinds, EventTags } from '../constants/base'
+import cluster from 'cluster'
import { applySpec, pipe, prop } from 'ramda'
-import { CanonicalEvent, DBEvent, Event, UnidentifiedEvent, UnsignedEvent } from '../@types/event'
import { EventId, Pubkey, Tag } from '../@types/base'
-import cluster from 'cluster'
-import { deriveFromSecret } from './secret'
+import { CanonicalEvent, DBEvent, Event, UnidentifiedEvent, UnsignedEvent } from '../@types/event'
import { EventKindsRange } from '../@types/settings'
-import { fromBuffer } from './transform'
-import { getLeadingZeroBits } from './proof-of-work'
-import { isGenericTagQuery, isGeohashPrefixCriterion, stripGeohashPrefixWildcard } from './filter'
import { SubscriptionFilter } from '../@types/subscription'
import { WebSocketServerAdapterEvent } from '../constants/adapter'
+import { ALL_RELAYS, EventKinds, EventTags } from '../constants/base'
+import { isGenericTagQuery, isGeohashPrefixCriterion, stripGeohashPrefixWildcard } from './filter'
+import { getLeadingZeroBits } from './proof-of-work'
+import { deriveFromSecret } from './secret'
+import { fromBuffer } from './transform'
export const serializeEvent = (event: UnidentifiedEvent): CanonicalEvent => [
0,
@@ -238,7 +238,9 @@ export const isRequestToVanishEvent = (event: Event, relayUrl?: string): boolean
const relayTags = event.tags.filter((tag) => tag.length >= 2 && tag[0] === EventTags.Relay).map((tag) => tag[1])
- return relayTags.length > 0 && relayTags.every((relay) => relay === relayUrl || relay === ALL_RELAYS)
+ // NIP-62: a request may name several relays. It applies here when any target
+ // is this relay or ALL_RELAYS; it does not need every target to match.
+ return relayTags.some((relay) => relay === relayUrl || relay === ALL_RELAYS)
}
export const isExpiredEvent = (event: Event): boolean => {
diff --git a/src/utils/peering-advertisement.ts b/src/utils/peering-advertisement.ts
new file mode 100644
index 000000000..36ce2f706
--- /dev/null
+++ b/src/utils/peering-advertisement.ts
@@ -0,0 +1,178 @@
+import axios from 'axios'
+import { z } from 'zod'
+import { AdvertisedPeeringScope, PeeringAdvertisement } from '../@types/peering'
+import { EventKindsRange, PeeringTerm } from '../@types/settings'
+import { MAX_PEERING_KIND } from './peering-scope'
+import { isNip11FetchTargetSafe, safeNip11Lookup } from './relay-probe/nip11-probe'
+import { parseProbeTarget } from './relay-probe/target'
+
+const MAX_NIP11_RESPONSE_BYTES = 256 * 1024
+const DEFAULT_NIP11_TIMEOUT_MS = 10_000
+
+const peeringKindSchema = z.union([
+ z.number().int().min(0).max(MAX_PEERING_KIND),
+ z.tuple([z.number().int().min(0).max(MAX_PEERING_KIND), z.number().int().min(0).max(MAX_PEERING_KIND)]),
+])
+
+const advertisedScopeSchema = z
+ .object({
+ kinds: z.array(peeringKindSchema).min(1),
+ since: z.number().int().min(0).optional(),
+ until: z.number().int().min(0).optional(),
+ window_seconds: z.number().int().positive().optional(),
+ // Older or hand-written documents sometimes use camelCase; accept both.
+ windowSeconds: z.number().int().positive().optional(),
+ })
+ .passthrough()
+ .superRefine((scope, ctx) => {
+ const windowSeconds = scope.window_seconds ?? scope.windowSeconds
+ if (windowSeconds !== undefined && (scope.since !== undefined || scope.until !== undefined)) {
+ ctx.addIssue({
+ code: z.ZodIssueCode.custom,
+ message: 'window_seconds cannot be combined with since or until',
+ })
+ }
+
+ if (scope.since !== undefined && scope.until !== undefined && scope.since > scope.until) {
+ ctx.addIssue({
+ code: z.ZodIssueCode.custom,
+ message: 'since must be less than or equal to until',
+ })
+ }
+
+ for (const kind of scope.kinds) {
+ if (Array.isArray(kind) && kind[0] > kind[1]) {
+ ctx.addIssue({
+ code: z.ZodIssueCode.custom,
+ message: 'kind range start must be less than or equal to its end',
+ })
+ }
+ }
+ })
+
+const peeringTermSchema = z.object({
+ amount: z.number().finite().nonnegative(),
+ unit: z.string().min(1),
+ period: z.number().int().positive(),
+})
+
+const peeringDocumentSchema = z
+ .object({
+ enabled: z.boolean().optional(),
+ scopes: z.array(advertisedScopeSchema).optional(),
+ terms: z.array(peeringTermSchema).optional(),
+ })
+ .passthrough()
+
+const nip11Schema = z
+ .object({
+ self: z
+ .string()
+ .regex(/^[0-9a-fA-F]{64}$/)
+ .optional(),
+ supported_nips: z.array(z.number().int().positive()).optional(),
+ peering: peeringDocumentSchema.optional(),
+ })
+ .passthrough()
+
+export type Nip11PeeringDocument = z.infer
+
+export const parsePeeringAdvertisement = (document: unknown): PeeringAdvertisement => {
+ const parsed = nip11Schema.safeParse(document)
+ if (!parsed.success) {
+ const reason = parsed.error.issues.map((issue) => issue.message).join('; ')
+
+ throw new Error(`invalid NIP-11 document: ${reason}`)
+ }
+
+ const peering = parsed.data.peering
+ const enabled = peering?.enabled === true
+
+ if (!enabled) {
+ return {
+ enabled: false,
+ scopes: [],
+ terms: [],
+ self: parsed.data.self?.toLowerCase(),
+ supportedNips: parsed.data.supported_nips ?? [],
+ }
+ }
+
+ const scopes = peering?.scopes ?? []
+ if (scopes.length === 0) {
+ throw new Error('peer advertises peering but offers no scopes')
+ }
+
+ return {
+ enabled,
+ scopes: scopes.map(
+ (scope): AdvertisedPeeringScope => ({
+ kinds: scope.kinds as (number | EventKindsRange)[],
+ ...(scope.since !== undefined ? { since: scope.since } : {}),
+ ...(scope.until !== undefined ? { until: scope.until } : {}),
+ ...(scope.window_seconds !== undefined || scope.windowSeconds !== undefined
+ ? { windowSeconds: scope.window_seconds ?? scope.windowSeconds }
+ : {}),
+ }),
+ ),
+ terms: (peering?.terms ?? []).map(
+ (term): PeeringTerm => ({
+ amount: term.amount,
+ unit: term.unit,
+ period: term.period,
+ }),
+ ),
+ self: parsed.data.self?.toLowerCase(),
+ supportedNips: parsed.data.supported_nips ?? [],
+ }
+}
+
+export const nip11UrlForRelay = (relayUrl: string): string => parseProbeTarget(relayUrl).nip11Url
+
+export interface FetchPeeringAdvertisementOptions {
+ timeoutMs?: number
+}
+
+/**
+ * Reads a peer's NIP-11 document and returns the peering terms it advertises.
+ * Phases that act on those terms (sync here, settlement later) share this one
+ * reader instead of each re-implementing discovery.
+ */
+export const fetchPeeringAdvertisement = async (
+ relayUrl: string,
+ options: FetchPeeringAdvertisementOptions = {},
+): Promise => {
+ const timeoutMs = options.timeoutMs ?? DEFAULT_NIP11_TIMEOUT_MS
+ const url = nip11UrlForRelay(relayUrl)
+
+ if (!isNip11FetchTargetSafe(url)) {
+ throw new Error(`unable to read peering advertisement from ${relayUrl}: refused unsafe NIP-11 fetch target`)
+ }
+
+ try {
+ const response = await axios.get(url, {
+ timeout: timeoutMs,
+ headers: { Accept: 'application/nostr+json' },
+ responseType: 'json',
+ validateStatus: (status) => status === 200,
+ maxRedirects: 1,
+ maxContentLength: MAX_NIP11_RESPONSE_BYTES,
+ maxBodyLength: MAX_NIP11_RESPONSE_BYTES,
+ // Direct egress so safeNip11Lookup actually sees the peer host. A proxy
+ // would resolve the destination itself and bypass the private-address check.
+ proxy: false,
+ lookup: safeNip11Lookup,
+ beforeRedirect: (redirect: { href?: string; protocol?: string; hostname?: string }) => {
+ const href = redirect.href ?? `${redirect.protocol ?? ''}//${redirect.hostname ?? ''}`
+ if (!isNip11FetchTargetSafe(href)) {
+ throw new Error(`refused redirect to unsafe NIP-11 target: ${href}`)
+ }
+ },
+ })
+
+ return parsePeeringAdvertisement(response.data)
+ } catch (error) {
+ const message = error instanceof Error ? error.message : String(error)
+ throw new Error(`unable to read peering advertisement from ${relayUrl}: ${message}`)
+ }
+}
diff --git a/src/utils/peering-auth.ts b/src/utils/peering-auth.ts
index 623eb437a..d076eda4c 100644
--- a/src/utils/peering-auth.ts
+++ b/src/utils/peering-auth.ts
@@ -8,6 +8,8 @@ import { getPublicKey, identifyEvent, signEvent } from './event'
import { parseRelayPubkey } from './nip43-invites'
export const PEERING_AUTH_TIMEOUT_MS = 10_000
+/** One inbound frame from a peer relay. Matches the relay's default max payload with headroom. */
+export const PEERING_MAX_PAYLOAD_BYTES = 2 * 1024 * 1024
export interface PeeringAuthEventOptions {
relayUrl: string
@@ -21,6 +23,7 @@ export interface AuthenticatePeerOptions {
privateKey: string
timeoutMs?: number
createdAt?: number
+ maxPayloadBytes?: number
}
export interface AuthenticatedPeerConnection {
@@ -65,6 +68,16 @@ export const findConfiguredPeeringPeer = (
export const isConfiguredPeeringPeer = (settings: Pick | undefined, pubkey: string): boolean =>
typeof findConfiguredPeeringPeer(settings, pubkey) !== 'undefined'
+/** Enabled, minimally well-formed peers to sync from. Settings validation rejects the rest. */
+export const listEnabledPeeringPeers = (settings: Pick | undefined): PeeringPeer[] =>
+ (settings?.peering?.peers ?? []).filter(
+ (peer) =>
+ peer.enabled !== false &&
+ typeof peer.relayUrl === 'string' &&
+ peer.relayUrl.trim().length > 0 &&
+ parsePeeringPeerPubkey(peer.pubkey) !== undefined,
+ )
+
/**
* Builds the NIP-42 AUTH event that proves this relay's own identity to a
* source relay. Signed with the relay keypair, so the event's pubkey is the
@@ -112,12 +125,18 @@ const closeQuietly = (socket: WebSocket): void => {
* leaving the socket open for the caller (a live feed, negentropy session, ...).
*/
export const authenticateToPeeringRelay = (options: AuthenticatePeerOptions): Promise => {
- const { relayUrl, privateKey, timeoutMs = PEERING_AUTH_TIMEOUT_MS, createdAt } = options
+ const {
+ relayUrl,
+ privateKey,
+ timeoutMs = PEERING_AUTH_TIMEOUT_MS,
+ createdAt,
+ maxPayloadBytes = PEERING_MAX_PAYLOAD_BYTES,
+ } = options
return new Promise((resolve, reject) => {
let socket: WebSocket
try {
- socket = new WebSocket(relayUrl)
+ socket = new WebSocket(relayUrl, { maxPayload: maxPayloadBytes })
} catch (error) {
reject(error instanceof Error ? error : new Error('Unable to open the peering socket'))
return
diff --git a/src/utils/peering-connection.ts b/src/utils/peering-connection.ts
new file mode 100644
index 000000000..aae94873d
--- /dev/null
+++ b/src/utils/peering-connection.ts
@@ -0,0 +1,399 @@
+import WebSocket, { RawData } from 'ws'
+
+import { Event } from '../@types/event'
+import { ReconciliationItem } from '../@types/repositories'
+import { SubscriptionFilter } from '../@types/subscription'
+import { createLogger } from '../factories/logger-factory'
+import { Negentropy, NegentropyStorageVector, ReconcileResult } from './negentropy'
+import { NEGENTROPY_FRAME_SIZE_LIMIT } from './nip77'
+
+const logger = createLogger('peering-connection')
+
+const DEFAULT_REQUEST_TIMEOUT_MS = 30_000
+/** Upper bound for one fetched id batch's EVENT payloads, independent of count. */
+export const PEERING_FETCH_MAX_BYTES = 8 * 1024 * 1024
+
+/** A fetch reply exceeded its byte budget. The worker retries with smaller batches. */
+export class PeeringFetchOverflowError extends Error {
+ public constructor() {
+ super('peer sent more event data than this request allows')
+ this.name = 'PeeringFetchOverflowError'
+ }
+}
+
+export interface PeeringSyncConnection {
+ /** Reconciles one filter against the peer, returning the ids each side lacks. */
+ reconcile(filter: SubscriptionFilter, items: ReconciliationItem[], timeoutMs?: number): Promise
+ /** Fetches specific event ids over an ordinary REQ; resolves after EOSE. */
+ fetchEvents(ids: string[], timeoutMs?: number, maxBytes?: number): Promise
+ /** Opens a long-lived REQ and resolves after its EOSE. Events keep arriving afterwards. */
+ subscribe(filters: SubscriptionFilter[], onEvent: (event: Event) => void, timeoutMs?: number): Promise
+ /** Resolves when the socket closes, cleanly or not. */
+ waitForClose(): Promise
+ close(): void
+}
+
+interface NegWaiter {
+ resolve: (payload: string) => void
+ reject: (error: Error) => void
+ timer: NodeJS.Timeout
+}
+
+interface RequestState {
+ onEvent: (event: Event) => void
+ resolve?: () => void
+ reject?: (error: Error) => void
+ timer?: NodeJS.Timeout
+ closed: boolean
+ /** A live feed must end the connection on CLOSED so the worker reconnects. */
+ persistent?: boolean
+ /** Only ids still in this set are accepted; deletion makes fetches duplicate-safe. */
+ expectedIds?: Set
+ receivedBytes?: number
+ maxBytes?: number
+}
+
+const toHex = (bytes: Uint8Array): string => Buffer.from(bytes).toString('hex')
+
+const messageType = (message: unknown[]): string => (typeof message[0] === 'string' ? message[0] : '')
+
+const requestId = (message: unknown[]): string | undefined => (typeof message[1] === 'string' ? message[1] : undefined)
+
+const parseMessage = (raw: RawData): unknown[] | undefined => {
+ try {
+ const parsed = JSON.parse(raw.toString())
+
+ return Array.isArray(parsed) ? parsed : undefined
+ } catch {
+ return undefined
+ }
+}
+
+/**
+ * One authenticated outbound peering socket, speaking only the wire primitives
+ * the relay already implements for clients: AUTH (done before construction),
+ * NEG-OPEN/NEG-MSG/NEG-CLOSE for history and REQ/EVENT/EOSE for the live feed.
+ */
+export class PeeringConnection implements PeeringSyncConnection {
+ private readonly negWaiters = new Map()
+ private readonly requests = new Map()
+ private readonly closePromise: Promise
+ private closed = false
+ private requestCounter = 0
+
+ public constructor(private readonly socket: WebSocket) {
+ this.closePromise = new Promise((resolve) => {
+ socket.once('close', () => {
+ this.closed = true
+ this.failAll(new Error('peering connection closed'))
+ resolve()
+ })
+ })
+
+ socket.on('message', this.onMessage.bind(this))
+ socket.on('error', (error) => {
+ logger('peer socket error: %o', error)
+ })
+ }
+
+ public async reconcile(
+ filter: SubscriptionFilter,
+ items: ReconciliationItem[],
+ timeoutMs = DEFAULT_REQUEST_TIMEOUT_MS,
+ ): Promise {
+ const subscriptionId = this.nextRequestId('neg')
+ const storage = new NegentropyStorageVector()
+
+ for (const item of items) {
+ storage.insert(item.event_created_at, item.event_id)
+ }
+ storage.seal()
+
+ const negentropy = new Negentropy(storage, NEGENTROPY_FRAME_SIZE_LIMIT)
+ const haveIds: string[] = []
+ const needIds: string[] = []
+
+ let output: Uint8Array | null = negentropy.initiate()
+ let waiting = this.waitForNegMsg(subscriptionId, timeoutMs)
+
+ try {
+ this.send(['NEG-OPEN', subscriptionId, filter, toHex(output)])
+
+ while (output !== null) {
+ const payload = await waiting
+
+ const result = negentropy.reconcile(Buffer.from(payload, 'hex'))
+ haveIds.push(...result.haveIds)
+ needIds.push(...result.needIds)
+ output = result.output
+
+ if (output !== null) {
+ waiting = this.waitForNegMsg(subscriptionId, timeoutMs)
+ this.send(['NEG-MSG', subscriptionId, toHex(output)])
+ }
+ }
+ } finally {
+ this.clearNegWaiter(subscriptionId)
+ if (!this.closed) {
+ try {
+ this.send(['NEG-CLOSE', subscriptionId])
+ } catch {
+ // The socket may have closed while the final message was in flight.
+ }
+ }
+ }
+
+ return { output: null, haveIds, needIds }
+ }
+
+ public async fetchEvents(
+ ids: string[],
+ timeoutMs = DEFAULT_REQUEST_TIMEOUT_MS,
+ maxBytes = PEERING_FETCH_MAX_BYTES,
+ ): Promise {
+ if (ids.length === 0) {
+ return []
+ }
+
+ const subscriptionId = this.nextRequestId('fetch')
+ const events: Event[] = []
+
+ try {
+ await this.openSubscription(subscriptionId, [{ ids }], (event) => events.push(event), timeoutMs, false, {
+ expectedIds: new Set(ids),
+ maxBytes,
+ })
+
+ return events
+ } finally {
+ this.closeSubscription(subscriptionId)
+ }
+ }
+
+ public subscribe(
+ filters: SubscriptionFilter[],
+ onEvent: (event: Event) => void,
+ timeoutMs = DEFAULT_REQUEST_TIMEOUT_MS,
+ ): Promise {
+ const subscriptionId = this.nextRequestId('live')
+
+ return this.openSubscription(subscriptionId, filters, onEvent, timeoutMs, true)
+ }
+ public waitForClose(): Promise {
+ return this.closePromise
+ }
+
+ public close(): void {
+ if (this.closed) {
+ return
+ }
+
+ this.closed = true
+ this.failAll(new Error('peering connection closed'))
+ try {
+ this.socket.close()
+ } catch {
+ // Already closing; the close event will settle the promise.
+ }
+ }
+
+ private openSubscription(
+ subscriptionId: string,
+ filters: SubscriptionFilter[],
+ onEvent: (event: Event) => void,
+ timeoutMs: number,
+ persistent = false,
+ bounds: Pick = {},
+ ): Promise {
+ const state: RequestState = {
+ onEvent,
+ closed: false,
+ persistent,
+ ...bounds,
+ }
+
+ this.requests.set(subscriptionId, state)
+
+ return new Promise((resolve, reject) => {
+ state.resolve = resolve
+ state.reject = reject
+ state.timer = setTimeout(() => {
+ if (state.closed) {
+ return
+ }
+ state.closed = true
+ this.requests.delete(subscriptionId)
+ reject(new Error('timed out waiting for the peer to finish the stored query'))
+ }, timeoutMs)
+ state.timer.unref()
+
+ try {
+ this.send(['REQ', subscriptionId, ...filters])
+ } catch (error) {
+ this.clearRequest(state)
+ this.requests.delete(subscriptionId)
+ reject(error instanceof Error ? error : new Error(String(error)))
+ }
+ })
+ }
+
+ private closeSubscription(subscriptionId: string): void {
+ const state = this.requests.get(subscriptionId)
+ if (state) {
+ this.clearRequest(state)
+ this.requests.delete(subscriptionId)
+ }
+
+ if (!this.closed) {
+ try {
+ this.send(['CLOSE', subscriptionId])
+ } catch {
+ // The socket may already be closing.
+ }
+ }
+ }
+
+ private waitForNegMsg(subscriptionId: string, timeoutMs: number): Promise {
+ return new Promise((resolve, reject) => {
+ const timer = setTimeout(() => {
+ this.negWaiters.delete(subscriptionId)
+ reject(new Error('timed out waiting for a negentropy reply'))
+ }, timeoutMs)
+ timer.unref()
+
+ this.negWaiters.set(subscriptionId, { resolve, reject, timer })
+ })
+ }
+
+ private clearNegWaiter(subscriptionId: string): void {
+ const waiter = this.negWaiters.get(subscriptionId)
+ if (!waiter) {
+ return
+ }
+
+ clearTimeout(waiter.timer)
+ this.negWaiters.delete(subscriptionId)
+ }
+
+ private onMessage(raw: RawData): void {
+ const message = parseMessage(raw)
+ if (!message) {
+ return
+ }
+
+ const subscriptionId = requestId(message)
+ if (!subscriptionId) {
+ return
+ }
+
+ switch (messageType(message)) {
+ case 'NEG-MSG': {
+ const waiter = this.negWaiters.get(subscriptionId)
+ if (!waiter || typeof message[2] !== 'string') {
+ return
+ }
+ clearTimeout(waiter.timer)
+ this.negWaiters.delete(subscriptionId)
+ waiter.resolve(message[2])
+ return
+ }
+ case 'NEG-ERR': {
+ const waiter = this.negWaiters.get(subscriptionId)
+ if (!waiter) {
+ return
+ }
+ clearTimeout(waiter.timer)
+ this.negWaiters.delete(subscriptionId)
+ waiter.reject(new Error(typeof message[2] === 'string' ? message[2] : 'negentropy request rejected'))
+ return
+ }
+ case 'EVENT': {
+ const state = this.requests.get(subscriptionId)
+ const event = message[2]
+ if (!state || typeof event !== 'object' || event === null) {
+ return
+ }
+
+ const parsedEvent = event as Event
+ if (state.expectedIds && !state.expectedIds.delete(parsedEvent.id)) {
+ return
+ }
+
+ const size = Buffer.byteLength(JSON.stringify(parsedEvent))
+ state.receivedBytes = (state.receivedBytes ?? 0) + size
+ if (state.maxBytes !== undefined && state.receivedBytes > state.maxBytes) {
+ this.clearRequest(state)
+ this.requests.delete(subscriptionId)
+ state.reject?.(new PeeringFetchOverflowError())
+ return
+ }
+
+ state.onEvent(parsedEvent)
+ return
+ }
+ case 'EOSE': {
+ const state = this.requests.get(subscriptionId)
+ if (!state) {
+ return
+ }
+ if (state.timer) {
+ clearTimeout(state.timer)
+ }
+ state.resolve?.()
+ return
+ }
+ case 'CLOSED': {
+ const state = this.requests.get(subscriptionId)
+ if (!state) {
+ return
+ }
+ const reason = typeof message[2] === 'string' ? message[2] : 'peer closed the subscription'
+ this.clearRequest(state)
+ this.requests.delete(subscriptionId)
+ state.reject?.(new Error(reason))
+ if (state.persistent) {
+ logger.warn('peer closed the live subscription: %s', reason)
+ this.close()
+ }
+ return
+ }
+ default:
+ return
+ }
+ }
+
+ private clearRequest(state: RequestState): void {
+ state.closed = true
+ if (state.timer) {
+ clearTimeout(state.timer)
+ }
+ }
+
+ private failAll(error: Error): void {
+ for (const waiter of this.negWaiters.values()) {
+ clearTimeout(waiter.timer)
+ waiter.reject(error)
+ }
+ this.negWaiters.clear()
+
+ for (const state of this.requests.values()) {
+ this.clearRequest(state)
+ state.reject?.(error)
+ }
+ this.requests.clear()
+ }
+
+ private send(message: unknown[]): void {
+ if (this.closed) {
+ throw new Error('peering connection is closed')
+ }
+
+ this.socket.send(JSON.stringify(message))
+ }
+
+ private nextRequestId(prefix: string): string {
+ this.requestCounter += 1
+
+ return `peering-${prefix}-${this.requestCounter}`
+ }
+}
diff --git a/src/utils/peering-scope.ts b/src/utils/peering-scope.ts
new file mode 100644
index 000000000..ba1639dd6
--- /dev/null
+++ b/src/utils/peering-scope.ts
@@ -0,0 +1,145 @@
+import { EventKindsRange, PeeringScope } from '../@types/settings'
+import { SubscriptionFilter } from '../@types/subscription'
+import { EventKinds } from '../constants/base'
+
+export const MAX_PEERING_KIND = 65535
+
+/**
+ * Overlap used when the live subscription starts. NIP-01 REQ has no "future
+ * only" mode, so ask for a short tail as well: an event published while
+ * negentropy runs is delivered either by reconciliation or by the live
+ * subscription, and duplicate inserts are harmless.
+ */
+export const PEERING_LIVE_OVERLAP_SECONDS = 60
+
+/**
+ * Keep each filter under the default peer maxFilterValues (10,000) with
+ * headroom. A broad scope such as `[[0, 65535]]` must become several filters
+ * or the source rejects the whole request before any history is synced.
+ */
+export const PEERING_MAX_FILTER_KINDS = 5000
+
+/** Expand `[min, max]` scopes into the explicit kind list NIP-01 filters require. */
+export const expandPeeringKinds = (kinds: (EventKinds | EventKindsRange)[]): number[] => {
+ const expanded = new Set()
+
+ for (const kind of kinds) {
+ if (typeof kind === 'number') {
+ expanded.add(kind)
+ continue
+ }
+
+ for (let value = kind[0]; value <= kind[1]; value++) {
+ expanded.add(value)
+ }
+ }
+
+ return [...expanded].sort((left, right) => left - right)
+}
+
+const scopeKinds = (scope: PeeringScope): number[] => expandPeeringKinds(scope.kinds)
+
+const chunk = (values: T[], size: number): T[][] => {
+ const chunks: T[][] = []
+ for (let offset = 0; offset < values.length; offset += size) {
+ chunks.push(values.slice(offset, offset + size))
+ }
+
+ return chunks
+}
+
+const applyScopeTime = (filter: SubscriptionFilter, scope: PeeringScope, nowSeconds: number): SubscriptionFilter => {
+ if (typeof scope.windowSeconds === 'number') {
+ filter.since = Math.max(0, nowSeconds - scope.windowSeconds)
+ } else {
+ if (typeof scope.since === 'number') {
+ filter.since = scope.since
+ }
+
+ if (typeof scope.until === 'number') {
+ filter.until = scope.until
+ }
+ }
+
+ return filter
+}
+
+/**
+ * The filter used to reconcile one advertised scope when it fits in one
+ * request. A rolling window becomes an absolute `since` at the moment of the
+ * reconciliation; `limit` is never set because negentropy compares whole sets
+ * rather than pages.
+ */
+export const peeringScopeToFilter = (scope: PeeringScope, nowSeconds: number): SubscriptionFilter =>
+ applyScopeTime({ kinds: scopeKinds(scope) }, scope, nowSeconds)
+
+/**
+ * Reconcile filters for one scope. Wide kind ranges are split into chunks so
+ * each request stays within a peer's `maxFilterValues`.
+ */
+export const peeringScopeToFilters = (
+ scope: PeeringScope,
+ nowSeconds: number,
+ maxKinds = PEERING_MAX_FILTER_KINDS,
+): SubscriptionFilter[] =>
+ chunk(scopeKinds(scope), Math.max(1, maxKinds)).map((kinds) => applyScopeTime({ kinds }, scope, nowSeconds))
+
+const isCurrentScope = (scope: PeeringScope, nowSeconds: number): boolean =>
+ typeof scope.until !== 'number' || scope.until >= nowSeconds
+
+/**
+ * Filters for the long-lived subscription. A scope whose `until` already passed
+ * is historical only and gets no live subscription. The subscription starts at
+ * the recent overlap, but never earlier than the scope itself: a 5-second
+ * rolling window must not import the last 60 seconds by accident.
+ */
+export const peeringScopesToLiveFilters = (
+ scopes: PeeringScope[],
+ nowSeconds: number,
+ overlapSeconds = PEERING_LIVE_OVERLAP_SECONDS,
+): SubscriptionFilter[] =>
+ scopes
+ .filter((scope) => isCurrentScope(scope, nowSeconds))
+ .flatMap((scope) =>
+ peeringScopeToFilters(scope, nowSeconds).map((filter) => {
+ const overlapSince = Math.max(0, nowSeconds - overlapSeconds)
+
+ if (typeof scope.windowSeconds === 'number') {
+ // The scope's own cutoff is already the newer of the two for short
+ // windows; for long windows the overlap is the newer one.
+ filter.since = Math.max(filter.since ?? 0, overlapSince)
+ delete filter.until
+ } else {
+ filter.since = Math.max(filter.since ?? 0, overlapSince)
+ }
+
+ return filter
+ }),
+ )
+
+export const isCurrentPeeringScope = (scope: PeeringScope, nowSeconds: number): boolean =>
+ isCurrentScope(scope, nowSeconds)
+
+/**
+ * Splits one reconciliation filter at its midpoint in time when a peer rejects
+ * the set as too large. Returns undefined when the range cannot be split any
+ * further, so the caller reports the original rejection instead of looping
+ * forever.
+ */
+export const splitPeeringFilter = (
+ filter: SubscriptionFilter,
+ nowSeconds: number,
+): [SubscriptionFilter, SubscriptionFilter] | undefined => {
+ const since = filter.since ?? 0
+ const until = filter.until ?? nowSeconds
+ if (until <= since) {
+ return undefined
+ }
+
+ const midpoint = Math.floor((since + until) / 2)
+
+ return [
+ { ...filter, since, until: midpoint },
+ { ...filter, since: midpoint + 1, until },
+ ]
+}
diff --git a/src/utils/peering-sync-state.ts b/src/utils/peering-sync-state.ts
new file mode 100644
index 000000000..c22e6ff29
--- /dev/null
+++ b/src/utils/peering-sync-state.ts
@@ -0,0 +1,62 @@
+import { ICacheAdapter } from '../@types/adapters'
+import { PeeringPeerSyncState } from '../@types/peering'
+import { createLogger } from '../factories/logger-factory'
+import { sanitizeAdminOpsError } from './admin-ops-sanitize'
+
+const logger = createLogger('peering-sync-state')
+
+export const PEERING_SYNC_STATE_INDEX_KEY = 'admin:peering:sync:index'
+export const PEERING_SYNC_STATE_TTL_SECONDS = 7 * 24 * 60 * 60
+/** A peer whose state has not moved for two reconciliation intervals is stale. */
+export const PEERING_SYNC_STATE_STALE_AFTER_SECONDS = 30 * 60
+
+const stateKey = (pubkey: string): string => `admin:peering:sync:${pubkey.toLowerCase()}`
+
+export class PeeringSyncStateStore {
+ public constructor(private readonly cache: ICacheAdapter) {}
+
+ public async publish(state: PeeringPeerSyncState): Promise {
+ const sanitized: PeeringPeerSyncState = {
+ ...state,
+ pubkey: state.pubkey.toLowerCase(),
+ lastError: state.lastError !== undefined ? sanitizeAdminOpsError(state.lastError) : undefined,
+ }
+
+ await this.cache.setKey(stateKey(sanitized.pubkey), JSON.stringify(sanitized), PEERING_SYNC_STATE_TTL_SECONDS)
+ await this.cache.addToSet(PEERING_SYNC_STATE_INDEX_KEY, [sanitized.pubkey])
+ await this.cache.setKeyExpiry(PEERING_SYNC_STATE_INDEX_KEY, PEERING_SYNC_STATE_TTL_SECONDS)
+ }
+
+ public async list(): Promise {
+ const pubkeys = await this.cache.getSetMembers(PEERING_SYNC_STATE_INDEX_KEY)
+ const states: PeeringPeerSyncState[] = []
+
+ for (const pubkey of pubkeys) {
+ const raw = await this.cache.getKey(stateKey(pubkey))
+ if (!raw) {
+ continue
+ }
+
+ try {
+ states.push(JSON.parse(raw) as PeeringPeerSyncState)
+ } catch {
+ logger.warn('ignoring corrupt peering sync state for %s', pubkey)
+ }
+ }
+
+ return states.sort((left, right) => left.pubkey.localeCompare(right.pubkey))
+ }
+}
+
+export const isPeeringSyncStateStale = (updatedAtIso: string | undefined, staleAfterSeconds: number): boolean => {
+ if (!updatedAtIso) {
+ return true
+ }
+
+ const updatedAt = Date.parse(updatedAtIso)
+ if (!Number.isFinite(updatedAt)) {
+ return true
+ }
+
+ return Date.now() - updatedAt > staleAfterSeconds * 1000
+}
diff --git a/src/utils/relay-probe/nip11-probe.ts b/src/utils/relay-probe/nip11-probe.ts
index 306fa4661..36bfa7a13 100644
--- a/src/utils/relay-probe/nip11-probe.ts
+++ b/src/utils/relay-probe/nip11-probe.ts
@@ -1,3 +1,5 @@
+import { lookup as dnsLookup } from 'node:dns/promises'
+import { BlockList, isIP } from 'node:net'
import axios, { AxiosError } from 'axios'
import { z } from 'zod'
@@ -41,6 +43,69 @@ export interface Nip11Fetcher {
fetch(url: string, timeoutMs: number): Promise
}
+const blockedNip11Addresses = new BlockList()
+blockedNip11Addresses.addSubnet('0.0.0.0', 8, 'ipv4')
+blockedNip11Addresses.addSubnet('10.0.0.0', 8, 'ipv4')
+blockedNip11Addresses.addSubnet('100.64.0.0', 10, 'ipv4')
+blockedNip11Addresses.addSubnet('127.0.0.0', 8, 'ipv4')
+blockedNip11Addresses.addSubnet('169.254.0.0', 16, 'ipv4')
+blockedNip11Addresses.addSubnet('172.16.0.0', 12, 'ipv4')
+blockedNip11Addresses.addSubnet('192.0.0.0', 24, 'ipv4')
+blockedNip11Addresses.addSubnet('192.0.2.0', 24, 'ipv4')
+blockedNip11Addresses.addSubnet('192.168.0.0', 16, 'ipv4')
+blockedNip11Addresses.addSubnet('198.18.0.0', 15, 'ipv4')
+blockedNip11Addresses.addSubnet('198.51.100.0', 24, 'ipv4')
+blockedNip11Addresses.addSubnet('203.0.113.0', 24, 'ipv4')
+blockedNip11Addresses.addSubnet('224.0.0.0', 4, 'ipv4')
+blockedNip11Addresses.addSubnet('240.0.0.0', 4, 'ipv4')
+blockedNip11Addresses.addSubnet('::', 128, 'ipv6')
+blockedNip11Addresses.addSubnet('::1', 128, 'ipv6')
+blockedNip11Addresses.addSubnet('2001:db8::', 32, 'ipv6')
+blockedNip11Addresses.addSubnet('fc00::', 7, 'ipv6')
+blockedNip11Addresses.addSubnet('fe80::', 10, 'ipv6')
+blockedNip11Addresses.addSubnet('ff00::', 8, 'ipv6')
+
+const normalizeResolvedAddress = (address: string): string => {
+ const lower = address.toLowerCase()
+ const mapped = lower.startsWith('::ffff:') ? lower.slice('::ffff:'.length) : lower
+
+ return isIP(mapped) === 4 ? mapped : lower
+}
+
+export const isBlockedNip11Address = (address: string): boolean => {
+ const normalized = normalizeResolvedAddress(address)
+ const family = isIP(normalized)
+
+ if (family === 4) {
+ return blockedNip11Addresses.check(normalized, 'ipv4')
+ }
+
+ if (family === 6) {
+ return blockedNip11Addresses.check(normalized, 'ipv6')
+ }
+
+ return true
+}
+
+/**
+ * Resolve a NIP-11 host and refuse private/loopback/link-local results before
+ * the socket connects. String-only URL checks miss hostnames whose A/AAAA
+ * records point inward, including DNS rebinding after the literal check.
+ */
+export const safeNip11Lookup = async (
+ hostname: string,
+ options: object,
+): Promise<{ address: string; family: number }[]> => {
+ const result = await dnsLookup(hostname, { ...(options as any), all: true })
+ const addresses = Array.isArray(result) ? result : [result]
+
+ if (addresses.some((entry) => isBlockedNip11Address(entry.address))) {
+ throw new Error(`refused NIP-11 target ${hostname} resolving to a private address`)
+ }
+
+ return addresses
+}
+
/**
* Reject redirect targets that would turn relay probing into an SSRF primitive.
* Mirrors the NIP-05 verification guard in src/utils/nip05.ts.
@@ -101,6 +166,10 @@ export const createNodeNip11Fetcher = (): Nip11Fetcher => ({
maxRedirects: MAX_REDIRECTS,
maxContentLength: MAX_RESPONSE_BYTES,
maxBodyLength: MAX_RESPONSE_BYTES,
+ // Direct egress: a proxy would resolve/connect on our behalf and
+ // bypass safeNip11Lookup's private-address rejection.
+ proxy: false,
+ lookup: safeNip11Lookup,
beforeRedirect: (options: { href?: string; protocol?: string; hostname?: string }) => {
const href = options.href ?? `${options.protocol ?? ''}//${options.hostname ?? ''}`
if (!isNip11FetchTargetSafe(href)) {
diff --git a/test/integration/features/peering-sync/peering-sync.feature b/test/integration/features/peering-sync/peering-sync.feature
new file mode 100644
index 000000000..6849ffec4
--- /dev/null
+++ b/test/integration/features/peering-sync/peering-sync.feature
@@ -0,0 +1,15 @@
+Feature: Peering sync planning
+ Scenario: Wide peering scopes are chunked for a peer
+ When a peering scope covers kinds 0 through 7
+ Then the scope produces filters of at most 4 kinds
+
+ Scenario: Short rolling windows keep their own cutoff
+ When a peering scope uses a 5 second rolling window at 1000
+ Then its live filter starts at 995
+
+ Scenario: Oversized filter is split in time
+ When an unbounded peering filter is split at 1000
+ Then the split filters cover 0 to 500 and 501 to 1000
+
+ Scenario: NIP-11 private addresses are rejected
+ Then the NIP-11 address "127.0.0.1" is blocked
diff --git a/test/integration/features/peering-sync/peering-sync.feature.ts b/test/integration/features/peering-sync/peering-sync.feature.ts
new file mode 100644
index 000000000..6d1e27541
--- /dev/null
+++ b/test/integration/features/peering-sync/peering-sync.feature.ts
@@ -0,0 +1,59 @@
+import { Then, When, World } from '@cucumber/cucumber'
+import chai from 'chai'
+import { SubscriptionFilter } from '../../../../src/@types/subscription'
+import {
+ peeringScopesToLiveFilters,
+ peeringScopeToFilters,
+ splitPeeringFilter,
+} from '../../../../src/utils/peering-scope'
+import { isBlockedNip11Address } from '../../../../src/utils/relay-probe/nip11-probe'
+
+const { expect } = chai
+
+When(
+ 'a peering scope covers kinds {int} through {int}',
+ function (this: World>, from: number, to: number) {
+ this.parameters.peeringFilters = peeringScopeToFilters({ kinds: [[from, to] as any] }, 1_000, 4)
+ },
+)
+
+Then('the scope produces filters of at most {int} kinds', function (this: World>, limit: number) {
+ const filters = this.parameters.peeringFilters as SubscriptionFilter[]
+ expect(filters.length).to.be.greaterThan(0)
+ expect(filters.every((filter) => (filter.kinds?.length ?? 0) <= limit)).to.equal(true)
+})
+
+When(
+ 'a peering scope uses a {int} second rolling window at {int}',
+ function (this: World>, windowSeconds: number, now: number) {
+ this.parameters.peeringLiveFilters = peeringScopesToLiveFilters([{ kinds: [1], windowSeconds }], now, 60)
+ },
+)
+
+Then('its live filter starts at {int}', function (this: World>, since: number) {
+ const [filter] = this.parameters.peeringLiveFilters as SubscriptionFilter[]
+ expect(filter.since).to.equal(since)
+})
+
+When('an unbounded peering filter is split at {int}', function (this: World>, now: number) {
+ this.parameters.peeringSplit = splitPeeringFilter({ kinds: [1] }, now)
+})
+
+Then(
+ 'the split filters cover {int} to {int} and {int} to {int}',
+ function (
+ this: World>,
+ firstSince: number,
+ firstUntil: number,
+ secondSince: number,
+ secondUntil: number,
+ ) {
+ const [first, second] = this.parameters.peeringSplit as [SubscriptionFilter, SubscriptionFilter]
+ expect([first.since, first.until]).to.deep.equal([firstSince, firstUntil])
+ expect([second.since, second.until]).to.deep.equal([secondSince, secondUntil])
+ },
+)
+
+Then('the NIP-11 address {string} is blocked', function (this: World>, address: string) {
+ expect(isBlockedNip11Address(address)).to.equal(true)
+})
diff --git a/test/unit/app/app.spec.ts b/test/unit/app/app.spec.ts
index 7e8abb6dc..d8991a919 100644
--- a/test/unit/app/app.spec.ts
+++ b/test/unit/app/app.spec.ts
@@ -1,14 +1,12 @@
-import EventEmitter from 'events'
-
import chai from 'chai'
+import EventEmitter from 'events'
import Sinon from 'sinon'
import sinonChai from 'sinon-chai'
-
+import { Settings } from '../../../src/@types/settings'
import { App } from '../../../src/app/app'
import * as metricsTelemetry from '../../../src/telemetry/metrics'
-import * as operatorNotificationEnqueue from '../../../src/utils/operator-notification-enqueue'
import * as torClient from '../../../src/tor/client'
-import { Settings } from '../../../src/@types/settings'
+import * as operatorNotificationEnqueue from '../../../src/utils/operator-notification-enqueue'
import { SettingsStatic } from '../../../src/utils/settings'
chai.use(sinonChai)
@@ -168,7 +166,10 @@ describe('App', () => {
app.run()
- const written = fakeProcess.stdout.write.getCalls().map((call) => call.args[0]).join('')
+ const written = fakeProcess.stdout.write
+ .getCalls()
+ .map((call) => call.args[0])
+ .join('')
expect(written).to.include('███▄ █ ▒█████')
expect(written).to.match(/v\d+\.\d+\.\d+/)
@@ -196,7 +197,8 @@ describe('App', () => {
app.run()
- const lines = fakeProcess.stdout.write.getCalls()
+ const lines = fakeProcess.stdout.write
+ .getCalls()
.map((call) => String(call.args[0]).trim())
.filter(Boolean)
@@ -286,4 +288,41 @@ describe('App', () => {
expect(cluster.fork.getCalls().map((call) => call.args[0].WORKER_TYPE)).to.deep.equal(['worker', 'maintenance'])
})
+
+ it('starts one peering-sync worker when enabled peers are configured', () => {
+ const env = { ...process.env }
+ delete env.WORKER_COUNT
+ delete env.RELAY_BROADCAST_FANOUT
+ sandbox.stub(process, 'env').value(env)
+ sandbox.useFakeTimers()
+ sandbox.stub(SettingsStatic, 'watchSettings').returns([])
+ sandbox.stub(operatorNotificationEnqueue, 'enqueueOperatorNotification').resolves()
+ sandbox.stub(torClient, 'addOnion').rejects(new Error('tor is not configured'))
+ let pid = 1001
+ cluster.fork.callsFake(() => createWorker(String(pid), pid++))
+ const app = new App(
+ fakeProcess as any,
+ cluster as any,
+ sandbox.stub().returns({
+ payments: { enabled: false },
+ workers: { count: 1 },
+ eventStore: { backend: 'postgres' },
+ peering: {
+ enabled: false,
+ peers: [
+ { pubkey: 'a'.repeat(64), relayUrl: 'wss://peer.example.com', enabled: true },
+ { pubkey: 'b'.repeat(64), relayUrl: 'wss://disabled.example.com', enabled: false },
+ ],
+ },
+ } as unknown as Settings),
+ )
+
+ app.run()
+
+ expect(cluster.fork.getCalls().map((call) => call.args[0].WORKER_TYPE)).to.deep.equal([
+ 'worker',
+ 'maintenance',
+ 'peering-sync',
+ ])
+ })
})
diff --git a/test/unit/app/peering-sync-worker.spec.ts b/test/unit/app/peering-sync-worker.spec.ts
new file mode 100644
index 000000000..c97e5cbed
--- /dev/null
+++ b/test/unit/app/peering-sync-worker.spec.ts
@@ -0,0 +1,478 @@
+import { expect } from 'chai'
+import EventEmitter from 'events'
+import { Event } from '../../../src/@types/event'
+import { PeeringPeerSyncState } from '../../../src/@types/peering'
+import { IEventRepository, ReconciliationItem } from '../../../src/@types/repositories'
+import { Settings } from '../../../src/@types/settings'
+import { PeeringSyncWorker } from '../../../src/app/peering-sync-worker'
+import { getPublicKey, identifyEvent, isEventMatchingFilter, signEvent } from '../../../src/utils/event'
+import { PeeringFetchOverflowError } from '../../../src/utils/peering-connection'
+import { PeeringSyncStateStore } from '../../../src/utils/peering-sync-state'
+
+const privateKey = 'a'.repeat(64)
+const pubkey = getPublicKey(privateKey)
+const peerPubkey = getPublicKey('b'.repeat(64))
+
+const makeEvent = async (content: string, createdAt: number): Promise => {
+ const unsigned = await identifyEvent({
+ pubkey,
+ created_at: createdAt,
+ kind: 1,
+ tags: [],
+ content,
+ })
+
+ return signEvent(privateKey)(unsigned)
+}
+
+const waitFor = async (predicate: () => boolean, timeoutMs = 5_000): Promise => {
+ const startedAt = Date.now()
+ while (!predicate()) {
+ if (Date.now() - startedAt > timeoutMs) {
+ throw new Error('timed out waiting for condition')
+ }
+ await new Promise((resolve) => setTimeout(resolve, 10))
+ }
+}
+
+class FakeConnection {
+ public liveOnEvent?: (event: Event) => void
+ public closeCalls = 0
+ private readonly closed: Promise
+ private resolveClosed: (() => void) | undefined
+
+ public constructor(private readonly remoteEvents: Event[]) {
+ this.closed = new Promise((resolve) => {
+ this.resolveClosed = resolve
+ })
+ }
+
+ public async reconcile(
+ filter: Parameters[0],
+ items: ReconciliationItem[],
+ ): Promise<{ output: null; haveIds: string[]; needIds: string[] }> {
+ const localIds = new Set(items.map((item) => Buffer.from(item.event_id).toString('hex')))
+ const remoteIds = this.remoteEvents.filter((event) => isEventMatchingFilter(filter)(event)).map((event) => event.id)
+
+ return {
+ output: null,
+ haveIds: [],
+ needIds: remoteIds.filter((id) => !localIds.has(id)),
+ }
+ }
+
+ public async fetchEvents(ids: string[]): Promise {
+ const wanted = new Set(ids)
+
+ return this.remoteEvents.filter((event) => wanted.has(event.id))
+ }
+
+ public async subscribe(_filters: unknown[], onEvent: (event: Event) => void): Promise {
+ this.liveOnEvent = onEvent
+ }
+
+ public waitForClose(): Promise {
+ return this.closed
+ }
+
+ public close(): void {
+ this.closeCalls += 1
+ this.resolveClosed?.()
+ }
+}
+
+class FakeStateStore extends PeeringSyncStateStore {
+ public readonly published: PeeringPeerSyncState[] = []
+
+ public constructor() {
+ super({} as any)
+ }
+
+ public override async publish(state: PeeringPeerSyncState): Promise {
+ this.published.push({ ...state })
+ }
+}
+
+describe('PeeringSyncWorker', () => {
+ const now = Math.floor(Date.now() / 1000)
+ let fakeProcess: EventEmitter & { exit: (...args: unknown[]) => void }
+ let connection: FakeConnection
+ let stateStore: FakeStateStore
+ let localEvents: Event[]
+ let eventRepository: IEventRepository
+ let worker: PeeringSyncWorker
+ let settingsState: Settings
+
+ beforeEach(async () => {
+ fakeProcess = Object.assign(new EventEmitter(), {
+ exit: () => {},
+ })
+ localEvents = []
+ stateStore = new FakeStateStore()
+ settingsState = {
+ info: { relay_url: 'wss://relay.example.com' },
+ peering: {
+ enabled: true,
+ peers: [{ pubkey: peerPubkey, relayUrl: 'wss://peer.example.com', enabled: true }],
+ },
+ } as Settings
+
+ eventRepository = {
+ findReconciliationItems: async (filter: Parameters[0]) =>
+ localEvents
+ .filter((event) => isEventMatchingFilter(filter)(event))
+ .map((event) => ({ event_created_at: event.created_at, event_id: Buffer.from(event.id, 'hex') })),
+ create: async (event: Event) => {
+ if (localEvents.some((local) => local.id === event.id)) {
+ return 0
+ }
+ localEvents.push(event)
+ return 1
+ },
+ createMany: async (events: Event[]) => {
+ let inserted = 0
+ for (const event of events) {
+ if (!localEvents.some((local) => local.id === event.id)) {
+ localEvents.push(event)
+ inserted += 1
+ }
+ }
+ return inserted
+ },
+ upsert: async (event: Event) => {
+ const index = localEvents.findIndex((local) => local.id === event.id)
+ if (index >= 0) {
+ if (localEvents[index].created_at < event.created_at) {
+ localEvents[index] = event
+ return 1
+ }
+ return 0
+ }
+ localEvents.push(event)
+ return 1
+ },
+ upsertMany: async (events: Event[]) => {
+ let inserted = 0
+ for (const event of events) {
+ if (!localEvents.some((local) => local.id === event.id)) {
+ localEvents.push(event)
+ inserted += 1
+ }
+ }
+ return inserted
+ },
+ deleteByPubkeyAndIds: async (pubkey: string, ids: string[]) => {
+ const before = localEvents.length
+ const remaining = localEvents.filter((event) => !(event.pubkey === pubkey && ids.includes(event.id)))
+ localEvents.splice(0, localEvents.length, ...remaining)
+ return before - localEvents.length
+ },
+ deleteByPubkeyExceptKinds: async (pubkey: string, excludedKinds: number[]) => {
+ const before = localEvents.length
+ const remaining = localEvents.filter((event) => event.pubkey !== pubkey || excludedKinds.includes(event.kind))
+ localEvents.splice(0, localEvents.length, ...remaining)
+ return before - localEvents.length
+ },
+ } as unknown as IEventRepository
+ })
+
+ afterEach(() => {
+ worker?.close()
+ })
+
+ interface StartWorkerOptions {
+ scopes?: { kinds: number[]; windowSeconds?: number }[]
+ reconcileIntervalMs?: number
+ peerRefreshMs?: number
+ stateHeartbeatMs?: number
+ configureConnection?: (connection: FakeConnection) => void
+ userRepository?: {
+ isVanished: (pubkey: string) => Promise
+ setVanished: (pubkey: string) => Promise
+ }
+ }
+
+ const startWorker = (remoteEvents: Event[], options: StartWorkerOptions = {}): void => {
+ connection = new FakeConnection(remoteEvents)
+ options.configureConnection?.(connection)
+
+ worker = new PeeringSyncWorker(fakeProcess as unknown as NodeJS.Process, () => settingsState, eventRepository, {
+ authenticate: async () => ({ socket: { close: () => {} }, challenge: 'c', authEvent: {} }) as any,
+ fetchAdvertisement: async () => ({
+ enabled: true,
+ supportedNips: [1, 77],
+ self: peerPubkey,
+ terms: [],
+ scopes: options.scopes ?? [{ kinds: [1] }],
+ }),
+ createConnection: () => connection as any,
+ stateStore,
+ privateKey: () => privateKey,
+ reconcileIntervalMs: options.reconcileIntervalMs ?? 60_000,
+ peerRefreshMs: options.peerRefreshMs,
+ stateHeartbeatMs: options.stateHeartbeatMs,
+ reconnectDelayMs: 10,
+ userRepository: options.userRepository,
+ })
+
+ worker.run()
+ }
+
+ it('converges an empty store from negentropy and then applies live events', async () => {
+ const current = Math.floor(Date.now() / 1000)
+ const historical = await makeEvent('historical', current - 30)
+ startWorker([historical])
+
+ await waitFor(() => localEvents.length === 1)
+ expect(localEvents[0].id).to.equal(historical.id)
+ expect(connection.liveOnEvent).to.be.a('function')
+
+ await waitFor(() => stateStore.published.some((state) => state.status === 'live'))
+ const liveState = stateStore.published.filter((state) => state.status === 'live').at(-1)
+ expect(liveState?.lastReconciledAt).to.be.a('string')
+ expect(liveState?.missingEvents).to.equal(1)
+ expect(liveState?.receivedEvents).to.equal(1)
+
+ const live = await makeEvent('live', current)
+ connection.liveOnEvent?.(live)
+
+ await waitFor(() => localEvents.length === 2)
+ expect(localEvents.map((event) => event.id)).to.include(live.id)
+
+ const older = await makeEvent('older overlap replay', current - 20)
+ connection.liveOnEvent?.(older)
+
+ await waitFor(() => localEvents.length === 3)
+ await waitFor(() => stateStore.published.at(-1)?.latestEventCreatedAt === live.created_at)
+ const lag = stateStore.published.at(-1)?.lagSeconds ?? Infinity
+ expect(lag).to.be.at.least(0)
+ expect(lag).to.be.at.most(5)
+ })
+
+ it('delivers live events while history reconciliation is still running', async () => {
+ let releaseReconcile: (() => void) | undefined
+ const reconcileGate = new Promise((resolve) => {
+ releaseReconcile = resolve
+ })
+
+ startWorker([], {
+ configureConnection: (fake) => {
+ const reconcile = fake.reconcile.bind(fake)
+ fake.reconcile = async (filter, items) => {
+ await reconcileGate
+ return reconcile(filter as any, items)
+ }
+ },
+ })
+
+ await waitFor(() => connection.liveOnEvent !== undefined)
+
+ const live = await makeEvent('during history sync', Math.floor(Date.now() / 1000))
+ connection.liveOnEvent?.(live)
+
+ await waitFor(() => localEvents.length === 1)
+ expect(localEvents[0].id).to.equal(live.id)
+ releaseReconcile?.()
+ })
+
+ it('republishes sync state while a long reconciliation is running', async () => {
+ let releaseReconcile: (() => void) | undefined
+ const reconcileGate = new Promise((resolve) => {
+ releaseReconcile = resolve
+ })
+
+ startWorker([], {
+ stateHeartbeatMs: 20,
+ configureConnection: (fake) => {
+ const reconcile = fake.reconcile.bind(fake)
+ fake.reconcile = async (filter, items) => {
+ await reconcileGate
+ return reconcile(filter as any, items)
+ }
+ },
+ })
+
+ await waitFor(() => stateStore.published.filter((state) => state.status === 'reconciling').length >= 2)
+ releaseReconcile?.()
+ })
+
+ it('advances lag on state heartbeats between reconciliations', async () => {
+ const head = await makeEvent('head', Math.floor(Date.now() / 1000))
+ startWorker([head], { stateHeartbeatMs: 20 })
+
+ await waitFor(() => localEvents.length === 1)
+ await waitFor(() => stateStore.published.some((state) => (state.lagSeconds ?? 0) >= 1), 4_000)
+ })
+
+ it('retries an overflowing fetch batch in smaller pieces', async () => {
+ const first = await makeEvent('first', now - 30)
+ const second = await makeEvent('second', now - 20)
+ const fetchSizes: number[] = []
+
+ startWorker([first, second], {
+ configureConnection: (fake) => {
+ const fetchEvents = fake.fetchEvents.bind(fake)
+ fake.fetchEvents = async (ids) => {
+ fetchSizes.push(ids.length)
+ if (ids.length > 1) {
+ throw new PeeringFetchOverflowError()
+ }
+ return fetchEvents(ids)
+ }
+ },
+ })
+
+ await waitFor(() => localEvents.length === 2)
+ expect(fetchSizes).to.deep.equal([2, 1, 1])
+ })
+
+ it('re-checks live events against scope bounds, not the subscription overlap', async () => {
+ startWorker([])
+
+ await waitFor(() => connection.liveOnEvent !== undefined)
+
+ const late = await makeEvent('late publish', Math.floor(Date.now() / 1000) - 120)
+ connection.liveOnEvent?.(late)
+
+ await waitFor(() => localEvents.length === 1)
+ expect(localEvents[0].id).to.equal(late.id)
+ })
+
+ it('splits a scope the peer rejects as too large', async () => {
+ const historical = await makeEvent('historical', now - 30)
+ let calls = 0
+
+ startWorker([historical], {
+ configureConnection: (fake) => {
+ const reconcile = fake.reconcile.bind(fake)
+ fake.reconcile = async (filter, items) => {
+ calls += 1
+ if (calls === 1) {
+ throw new Error('blocked: this query is too big')
+ }
+ return reconcile(filter as any, items)
+ }
+ },
+ })
+
+ await waitFor(() => localEvents.length === 1)
+ expect(localEvents[0].id).to.equal(historical.id)
+ expect(calls).to.be.greaterThan(2)
+ await waitFor(() => stateStore.published.some((state) => state.status === 'live'))
+ expect(stateStore.published.filter((state) => state.status === 'live').at(-1)?.missingEvents).to.equal(1)
+ })
+
+ it('does not count a batch whose write stored nothing', async () => {
+ const historical = await makeEvent('historical', now - 30)
+ eventRepository.createMany = async () => 0
+ eventRepository.upsertMany = async () => 0
+ eventRepository.create = async () => 0
+ eventRepository.upsert = async () => 0
+
+ startWorker([historical])
+
+ await waitFor(() => stateStore.published.some((state) => state.status === 'live'))
+ const liveState = stateStore.published.filter((state) => state.status === 'live').at(-1)
+ expect(localEvents).to.have.length(0)
+ expect(liveState?.receivedEvents).to.equal(0)
+ })
+
+ it('applies the rolling window to live events using the current time', async () => {
+ startWorker([], { scopes: [{ kinds: [1], windowSeconds: 60 }] })
+
+ await waitFor(() => connection.liveOnEvent !== undefined)
+
+ const current = Math.floor(Date.now() / 1000)
+ const stale = await makeEvent('outside window', current - 120)
+ connection.liveOnEvent?.(stale)
+ await new Promise((resolve) => setTimeout(resolve, 50))
+ expect(localEvents).to.have.length(0)
+
+ const fresh = await makeEvent('inside window', current)
+ connection.liveOnEvent?.(fresh)
+ await waitFor(() => localEvents.length === 1)
+ expect(localEvents[0].id).to.equal(fresh.id)
+ })
+
+ it('stops a session when its peer is disabled in reloaded settings', async () => {
+ startWorker([], { peerRefreshMs: 20 })
+ await waitFor(() => connection.liveOnEvent !== undefined)
+
+ settingsState = {
+ ...settingsState,
+ peering: {
+ enabled: true,
+ peers: [],
+ },
+ } as Settings
+
+ await waitFor(() => connection.closeCalls > 0)
+ })
+
+ it('skips events from vanished authors', async () => {
+ const event = await makeEvent('from vanished author', now)
+ startWorker([event], {
+ userRepository: {
+ isVanished: async () => true,
+ setVanished: async () => 1,
+ },
+ })
+
+ await waitFor(() => stateStore.published.some((state) => state.status === 'live'))
+ expect(localEvents).to.have.length(0)
+ expect(stateStore.published.filter((state) => state.status === 'live').at(-1)?.receivedEvents).to.equal(0)
+ })
+
+ it('records a periodic reconciliation failure in peer state', async () => {
+ let calls = 0
+
+ startWorker([], {
+ reconcileIntervalMs: 500,
+ configureConnection: (fake) => {
+ const reconcile = fake.reconcile.bind(fake)
+ fake.reconcile = async (filter, items) => {
+ calls += 1
+ if (calls > 1) {
+ throw new Error('negentropy timeout')
+ }
+ return reconcile(filter as any, items)
+ }
+ },
+ })
+
+ await waitFor(() => stateStore.published.some((state) => state.status === 'live'))
+ await waitFor(() =>
+ stateStore.published.some((state) => state.status === 'error' && state.lastError === 'negentropy timeout'),
+ )
+ })
+
+ it('records an error when the peer has no peering advertisement', async () => {
+ connection = new FakeConnection([])
+ const settings = {
+ info: { relay_url: 'wss://relay.example.com' },
+ peering: {
+ enabled: true,
+ peers: [{ pubkey: peerPubkey, relayUrl: 'wss://peer.example.com', enabled: true }],
+ },
+ } as Settings
+
+ worker = new PeeringSyncWorker(fakeProcess as unknown as NodeJS.Process, () => settings, eventRepository, {
+ authenticate: async () => ({ socket: { close: () => {} }, challenge: 'c', authEvent: {} }) as any,
+ fetchAdvertisement: async () => ({
+ enabled: false,
+ supportedNips: [1],
+ scopes: [],
+ terms: [],
+ }),
+ createConnection: () => connection as any,
+ stateStore,
+ privateKey: () => privateKey,
+ reconnectDelayMs: 1_000,
+ })
+
+ worker.run()
+
+ await waitFor(() => stateStore.published.some((state) => state.status === 'error'))
+ expect(stateStore.published.some((state) => state.lastError === 'peer does not advertise peering')).to.equal(true)
+ })
+})
diff --git a/test/unit/controllers/admin/get-admin-peering-controller.spec.ts b/test/unit/controllers/admin/get-admin-peering-controller.spec.ts
new file mode 100644
index 000000000..3abcd102f
--- /dev/null
+++ b/test/unit/controllers/admin/get-admin-peering-controller.spec.ts
@@ -0,0 +1,50 @@
+import chai from 'chai'
+import Sinon from 'sinon'
+import sinonChai from 'sinon-chai'
+
+import type { PeeringPeerSyncState } from '../../../../src/@types/peering'
+import type { Settings } from '../../../../src/@types/settings'
+import { GetAdminPeeringController } from '../../../../src/controllers/admin/get-admin-peering-controller'
+
+chai.use(sinonChai)
+
+const { expect } = chai
+
+describe('GetAdminPeeringController', () => {
+ it('merges configured peers with stored sync state', async () => {
+ const state: PeeringPeerSyncState = {
+ pubkey: 'a'.repeat(64),
+ relayUrl: 'wss://peer.example.com',
+ status: 'live',
+ scopeCount: 1,
+ updatedAt: new Date().toISOString(),
+ lastReconciledAt: new Date().toISOString(),
+ lagSeconds: 3,
+ }
+ const stateStore = { list: Sinon.stub().resolves([state]) }
+ const settings = {
+ peering: {
+ enabled: true,
+ peers: [{ pubkey: state.pubkey, relayUrl: state.relayUrl }],
+ },
+ } as Settings
+ const controller = new GetAdminPeeringController(stateStore as any, () => settings)
+ const response = {
+ status: Sinon.stub().returnsThis(),
+ setHeader: Sinon.stub().returnsThis(),
+ send: Sinon.stub(),
+ }
+
+ await controller.handleRequest({} as any, response as any)
+
+ expect(response.status).to.have.been.calledWith(200)
+ const payload = response.send.firstCall.args[0]
+ expect(payload.peers).to.have.length(1)
+ expect(payload.peers[0]).to.include({
+ configured: true,
+ enabled: true,
+ stale: false,
+ })
+ expect(payload.peers[0].state.lagSeconds).to.equal(3)
+ })
+})
diff --git a/test/unit/factories/controllers/get-admin-peering-controller-factory.spec.ts b/test/unit/factories/controllers/get-admin-peering-controller-factory.spec.ts
new file mode 100644
index 000000000..44f48f374
--- /dev/null
+++ b/test/unit/factories/controllers/get-admin-peering-controller-factory.spec.ts
@@ -0,0 +1,32 @@
+import { expect } from 'chai'
+import Sinon from 'sinon'
+
+import * as cacheClient from '../../../../src/cache/client'
+import { createGetAdminPeeringController } from '../../../../src/factories/controllers/get-admin-peering-controller-factory'
+
+describe('get-admin-peering-controller factory', () => {
+ let sandbox: Sinon.SinonSandbox
+
+ beforeEach(() => {
+ sandbox = Sinon.createSandbox()
+ const client = {
+ isOpen: true,
+ on: function () {
+ return client
+ },
+ }
+ sandbox.stub(cacheClient, 'getCacheClient').returns(client as any)
+ })
+
+ afterEach(() => {
+ sandbox.restore()
+ })
+
+ it('creates one controller and reuses it across dashboard requests', () => {
+ const first = createGetAdminPeeringController()
+ const second = createGetAdminPeeringController()
+
+ expect(first).to.have.property('handleRequest')
+ expect(second).to.equal(first)
+ })
+})
diff --git a/test/unit/services/event-import-service.spec.ts b/test/unit/services/event-import-service.spec.ts
index 8aa91870f..7806f7a64 100644
--- a/test/unit/services/event-import-service.spec.ts
+++ b/test/unit/services/event-import-service.spec.ts
@@ -1,18 +1,17 @@
-import { join } from 'path'
+import { expect } from 'chai'
import fs from 'fs'
import os from 'os'
+import { join } from 'path'
import { Readable } from 'stream'
-
+import { Event } from '../../../src/@types/event'
+import { EventDeduplicationMetadataKey, EventKinds, EventTags } from '../../../src/constants/base'
import {
createEventBatchPersister,
EventImportLineError,
EventImportService,
EventImportStats,
} from '../../../src/services/event-import-service'
-import { EventDeduplicationMetadataKey, EventKinds, EventTags } from '../../../src/constants/base'
-import { Event } from '../../../src/@types/event'
-import { expect } from 'chai'
import { getEvents } from '../data/events'
describe('EventImportService', () => {
@@ -393,4 +392,201 @@ describe('EventImportService', () => {
expect(upsertedEvents).to.have.length(1)
expect((upsertedEvents[0] as any)[EventDeduplicationMetadataKey]).to.deep.equal(['one'])
})
+
+ it('skips vanished authors and applies kind-62 vanish rules when a user repository is configured', async () => {
+ const visible: Event = {
+ id: 'a'.repeat(64),
+ pubkey: 'b'.repeat(64),
+ created_at: 1,
+ kind: 1,
+ tags: [],
+ content: 'visible',
+ sig: 'c'.repeat(128),
+ }
+ const vanished: Event = {
+ ...visible,
+ id: 'd'.repeat(64),
+ pubkey: 'e'.repeat(64),
+ }
+ const vanish: Event = {
+ ...visible,
+ id: 'f'.repeat(64),
+ pubkey: '1'.repeat(64),
+ kind: EventKinds.REQUEST_TO_VANISH,
+ content: 'vanish',
+ }
+
+ const created: Event[] = []
+ const deletedExcept: string[][] = []
+ const vanishedPubkeys: string[] = []
+
+ const eventRepository = {
+ create: async (event: Event) => {
+ created.push(event)
+ return 1
+ },
+ createMany: async (events: Event[]) => {
+ created.push(...events)
+ return events.length
+ },
+ upsert: async () => 0,
+ upsertMany: async () => 0,
+ deleteByPubkeyAndIds: async () => 0,
+ deleteByPubkeyExceptKinds: async (pubkey: string, kinds: number[]) => {
+ deletedExcept.push([pubkey, ...kinds.map(String)])
+ return 1
+ },
+ } as any
+
+ const userRepository = {
+ isVanished: async (pubkey: string) => pubkey === vanished.pubkey,
+ setVanished: async (pubkey: string) => {
+ vanishedPubkeys.push(pubkey)
+ return 1
+ },
+ }
+
+ const persistBatch = createEventBatchPersister(eventRepository, { userRepository })
+ const inserted = await persistBatch([visible, vanished, vanish])
+
+ expect(inserted).to.equal(2)
+ expect(created.map((event) => event.id)).to.deep.equal([visible.id, vanish.id])
+ expect(deletedExcept).to.deep.equal([[vanish.pubkey, String(EventKinds.REQUEST_TO_VANISH)]])
+ expect(vanishedPubkeys).to.deep.equal([vanish.pubkey])
+ })
+
+ it('checks vanish state once per author within a batch', async () => {
+ const first: Event = {
+ id: 'a'.repeat(64),
+ pubkey: 'b'.repeat(64),
+ created_at: 1,
+ kind: 1,
+ tags: [],
+ content: 'one',
+ sig: 'c'.repeat(128),
+ }
+ const second: Event = { ...first, id: 'd'.repeat(64), content: 'two' }
+
+ let checks = 0
+ const eventRepository = {
+ create: async () => 1,
+ createMany: async (events: Event[]) => events.length,
+ upsert: async () => 0,
+ upsertMany: async (events: Event[]) => events.length,
+ deleteByPubkeyAndIds: async () => 0,
+ } as any
+
+ const persistBatch = createEventBatchPersister(eventRepository, {
+ userRepository: {
+ isVanished: async () => {
+ checks += 1
+ return false
+ },
+ setVanished: async () => 1,
+ },
+ })
+
+ const inserted = await persistBatch([first, second])
+
+ expect(inserted).to.equal(2)
+ expect(checks).to.equal(1)
+ })
+
+ it('does not apply a kind-62 request aimed at another relay', async () => {
+ const vanish: Event = {
+ id: 'a'.repeat(64),
+ pubkey: 'b'.repeat(64),
+ created_at: 1,
+ kind: EventKinds.REQUEST_TO_VANISH,
+ tags: [[EventTags.Relay, 'wss://other.example.com']],
+ content: 'vanish',
+ sig: 'c'.repeat(128),
+ }
+ const eventRepository = {
+ create: async () => 1,
+ createMany: async (events: Event[]) => events.length,
+ upsert: async () => 0,
+ upsertMany: async () => 0,
+ deleteByPubkeyAndIds: async () => 0,
+ deleteByPubkeyExceptKinds: async () => {
+ throw new Error('must not delete for another relay')
+ },
+ } as any
+ let vanished = 0
+ const persistBatch = createEventBatchPersister(eventRepository, {
+ relayUrl: 'wss://relay.example.com',
+ userRepository: {
+ isVanished: async () => false,
+ setVanished: async () => {
+ vanished += 1
+ return 1
+ },
+ },
+ })
+
+ const inserted = await persistBatch([vanish])
+
+ expect(inserted).to.equal(1)
+ expect(vanished).to.equal(0)
+ })
+
+ it('keeps import behavior for kind-62 events when no user repository is supplied', async () => {
+ const vanish: Event = {
+ id: 'a'.repeat(64),
+ pubkey: 'b'.repeat(64),
+ created_at: 1,
+ kind: EventKinds.REQUEST_TO_VANISH,
+ tags: [[EventTags.Relay, 'wss://other.example.com']],
+ content: 'vanish',
+ sig: 'c'.repeat(128),
+ }
+ const eventRepository = {
+ create: async () => 1,
+ createMany: async (events: Event[]) => events.length,
+ upsert: async () => 0,
+ upsertMany: async () => 0,
+ deleteByPubkeyAndIds: async () => 0,
+ deleteByPubkeyExceptKinds: async () => {
+ throw new Error('import must not apply vanish deletions')
+ },
+ } as any
+
+ const persistBatch = createEventBatchPersister(eventRepository)
+ const inserted = await persistBatch([vanish])
+
+ expect(inserted).to.equal(1)
+ })
+
+ it('drops ephemeral events without counting them as inserted', async () => {
+ const ephemeral: Event = {
+ id: 'a'.repeat(64),
+ pubkey: 'b'.repeat(64),
+ created_at: 1,
+ kind: EventKinds.EPHEMERAL_FIRST,
+ tags: [],
+ content: 'ephemeral',
+ sig: 'c'.repeat(128),
+ }
+
+ let createdManyCalls = 0
+ const eventRepository = {
+ create: async () => 1,
+ createMany: async (events: Event[]) => {
+ if (!events.length) {
+ return 0
+ }
+ createdManyCalls += 1
+ return events.length
+ },
+ upsert: async () => 0,
+ upsertMany: async (events: Event[]) => events.length,
+ deleteByPubkeyAndIds: async () => 0,
+ } as any
+
+ const persistBatch = createEventBatchPersister(eventRepository)
+ const inserted = await persistBatch([ephemeral])
+
+ expect(inserted).to.equal(0)
+ expect(createdManyCalls).to.equal(0)
+ })
})
diff --git a/test/unit/utils/admin-ops-config.spec.ts b/test/unit/utils/admin-ops-config.spec.ts
index 7694db13f..fbad276cf 100644
--- a/test/unit/utils/admin-ops-config.spec.ts
+++ b/test/unit/utils/admin-ops-config.spec.ts
@@ -1,15 +1,11 @@
import { expect } from 'chai'
-
-import { listConfiguredWorkers, mergeWorkerRows } from '../../../src/utils/admin-ops-config'
import { Settings } from '../../../src/@types/settings'
+import { listConfiguredWorkers, mergeWorkerRows } from '../../../src/utils/admin-ops-config'
import { ADMIN_OPS_HEARTBEAT_TTL_SECONDS, isAdminOpsHeartbeatStale } from '../../../src/utils/admin-ops-store'
describe('admin-ops-config', () => {
it('marks unmonitored workers as not stale without a heartbeat', () => {
- const rows = mergeWorkerRows(
- [{ type: 'worker', index: '0', enabled: true, label: 'Client worker 0' }],
- [],
- )
+ const rows = mergeWorkerRows([{ type: 'worker', index: '0', enabled: true, label: 'Client worker 0' }], [])
expect(rows[0].monitored).to.equal(false)
expect(rows[0].stale).to.equal(false)
@@ -40,4 +36,16 @@ describe('admin-ops-config', () => {
expect(workers.map((worker) => worker.type)).to.deep.equal(['worker', 'maintenance'])
})
+
+ it('lists the peering-sync worker when enabled peers are configured', () => {
+ const workers = listConfiguredWorkers({
+ workers: { count: 1 },
+ peering: {
+ enabled: false,
+ peers: [{ pubkey: 'a'.repeat(64), relayUrl: 'wss://peer.example.com' }],
+ },
+ } as unknown as Settings)
+
+ expect(workers.map((worker) => worker.type)).to.deep.equal(['worker', 'maintenance', 'peering-sync'])
+ })
})
diff --git a/test/unit/utils/admin-peering.spec.ts b/test/unit/utils/admin-peering.spec.ts
new file mode 100644
index 000000000..a5bf6e080
--- /dev/null
+++ b/test/unit/utils/admin-peering.spec.ts
@@ -0,0 +1,75 @@
+import { expect } from 'chai'
+import { PeeringPeerSyncState } from '../../../src/@types/peering'
+import { Settings } from '../../../src/@types/settings'
+import { mergePeeringRows } from '../../../src/utils/admin-peering'
+
+const state = (overrides: Partial = {}): PeeringPeerSyncState => ({
+ pubkey: 'a'.repeat(64),
+ relayUrl: 'wss://peer.example.com',
+ status: 'live',
+ scopeCount: 1,
+ updatedAt: new Date().toISOString(),
+ ...overrides,
+})
+
+describe('admin peering rows', () => {
+ it('merges configured peers with their stored sync state', () => {
+ const rows = mergePeeringRows(
+ {
+ peering: {
+ enabled: true,
+ peers: [{ pubkey: 'a'.repeat(64), relayUrl: 'wss://peer.example.com' }],
+ },
+ } as Settings,
+ [state()],
+ )
+
+ expect(rows).to.have.length(1)
+ expect(rows[0]).to.include({
+ configured: true,
+ enabled: true,
+ stale: false,
+ })
+ expect(rows[0].state?.status).to.equal('live')
+ })
+
+ it('marks an enabled peer stale when it has no state or the state is old', () => {
+ const rows = mergePeeringRows(
+ {
+ peering: {
+ enabled: true,
+ peers: [
+ { pubkey: 'a'.repeat(64), relayUrl: 'wss://new.example.com' },
+ { pubkey: 'b'.repeat(64), relayUrl: 'wss://old.example.com' },
+ ],
+ },
+ } as Settings,
+ [state({ pubkey: 'b'.repeat(64), updatedAt: '2020-01-01T00:00:00.000Z' })],
+ )
+
+ expect(rows.find((row) => row.relayUrl === 'wss://new.example.com')?.stale).to.equal(true)
+ expect(rows.find((row) => row.relayUrl === 'wss://old.example.com')?.stale).to.equal(true)
+ })
+
+ it('shows stored state for a peer that was removed from settings', () => {
+ const rows = mergePeeringRows({} as Settings, [state({ pubkey: 'c'.repeat(64) })])
+
+ expect(rows[0]).to.include({ configured: false, enabled: false })
+ })
+
+ it('does not attach state from an old relay URL to the configured peer', () => {
+ const rows = mergePeeringRows(
+ {
+ peering: {
+ enabled: true,
+ peers: [{ pubkey: 'a'.repeat(64), relayUrl: 'wss://new.example.com' }],
+ },
+ } as Settings,
+ [state({ relayUrl: 'wss://old.example.com' })],
+ )
+
+ expect(rows).to.have.length(1)
+ expect(rows[0].state).to.equal(undefined)
+ expect(rows[0].stale).to.equal(true)
+ })
+})
diff --git a/test/unit/utils/event.spec.ts b/test/unit/utils/event.spec.ts
index ec6d49723..9146199ab 100644
--- a/test/unit/utils/event.spec.ts
+++ b/test/unit/utils/event.spec.ts
@@ -1,5 +1,6 @@
-import { ALL_RELAYS, EventKinds, EventTags } from '../../../src/constants/base'
+import { expect } from 'chai'
import { CanonicalEvent, Event } from '../../../src/@types/event'
+import { ALL_RELAYS, EventKinds, EventTags } from '../../../src/constants/base'
import {
getEventExpiration,
isContactListEvent,
@@ -22,7 +23,6 @@ import {
isWelcomeRumorEvent,
serializeEvent,
} from '../../../src/utils/event'
-import { expect } from 'chai'
describe('NIP-01', () => {
describe('serializeEvent', () => {
@@ -661,6 +661,28 @@ describe('NIP-62', () => {
expect(isRequestToVanishEvent(event, 'relay_url')).to.be.false
})
+ it('returns true when any of several relay tags targets this relay', () => {
+ const event: Event = {
+ kind: 62,
+ tags: [
+ [EventTags.Relay, 'other_relay_url'],
+ [EventTags.Relay, 'relay_url'],
+ ],
+ } as any
+ expect(isRequestToVanishEvent(event, 'relay_url')).to.be.true
+ })
+
+ it('returns true when any of several relay tags is ALL_RELAYS', () => {
+ const event: Event = {
+ kind: 62,
+ tags: [
+ [EventTags.Relay, 'other_relay_url'],
+ [EventTags.Relay, ALL_RELAYS],
+ ],
+ } as any
+ expect(isRequestToVanishEvent(event, 'relay_url')).to.be.true
+ })
+
it('returns false when relay URL is provided for non-kind-62 event', () => {
const event: Event = {
kind: 1,
diff --git a/test/unit/utils/peering-advertisement.spec.ts b/test/unit/utils/peering-advertisement.spec.ts
new file mode 100644
index 000000000..bb7ca47e0
--- /dev/null
+++ b/test/unit/utils/peering-advertisement.spec.ts
@@ -0,0 +1,62 @@
+import { expect } from 'chai'
+
+import { fetchPeeringAdvertisement, parsePeeringAdvertisement } from '../../../src/utils/peering-advertisement'
+
+describe('peering advertisement', () => {
+ it('reads scopes and terms from the NIP-11 peering section', () => {
+ const advertisement = parsePeeringAdvertisement({
+ self: 'A'.repeat(64),
+ supported_nips: [1, 11, 77],
+ peering: {
+ enabled: true,
+ scopes: [
+ { kinds: [1, [6, 7]], since: 100 },
+ { kinds: [30023], window_seconds: 3600 },
+ ],
+ terms: [{ amount: 1000, unit: 'msats', period: 2592000 }],
+ },
+ })
+
+ expect(advertisement).to.deep.equal({
+ enabled: true,
+ self: 'a'.repeat(64),
+ supportedNips: [1, 11, 77],
+ scopes: [
+ { kinds: [1, [6, 7]], since: 100 },
+ { kinds: [30023], windowSeconds: 3600 },
+ ],
+ terms: [{ amount: 1000, unit: 'msats', period: 2592000 }],
+ })
+ })
+
+ it('reports peering as disabled when the section is absent', () => {
+ const advertisement = parsePeeringAdvertisement({ supported_nips: [1] })
+
+ expect(advertisement.enabled).to.equal(false)
+ expect(advertisement.scopes).to.deep.equal([])
+ })
+
+ it('rejects an enabled advertisement with no scopes', () => {
+ expect(() => parsePeeringAdvertisement({ peering: { enabled: true, scopes: [] } })).to.throw('offers no scopes')
+ })
+
+ it('rejects a scope that combines a window with an absolute range', () => {
+ expect(() =>
+ parsePeeringAdvertisement({
+ peering: {
+ enabled: true,
+ scopes: [{ kinds: [1], since: 100, window_seconds: 3600 }],
+ },
+ }),
+ ).to.throw('window_seconds cannot be combined')
+ })
+
+ it('refuses to fetch an internal NIP-11 target', async () => {
+ const error = await fetchPeeringAdvertisement('ws://127.0.0.1:8008').then(
+ () => undefined,
+ (err: Error) => err,
+ )
+
+ expect(error?.message).to.include('refused unsafe NIP-11 fetch target')
+ })
+})
diff --git a/test/unit/utils/peering-connection.spec.ts b/test/unit/utils/peering-connection.spec.ts
new file mode 100644
index 000000000..3a321153f
--- /dev/null
+++ b/test/unit/utils/peering-connection.spec.ts
@@ -0,0 +1,186 @@
+import { expect } from 'chai'
+import EventEmitter from 'events'
+import { Negentropy, NegentropyStorageVector } from '../../../src/utils/negentropy'
+import { PeeringConnection, PeeringFetchOverflowError } from '../../../src/utils/peering-connection'
+
+class FakeSocket extends EventEmitter {
+ public readonly sent: unknown[][] = []
+ public closed = false
+ private responder?: Negentropy
+ private readonly remoteEvents: {
+ id: string
+ created_at: number
+ content: string
+ pubkey: string
+ kind: number
+ tags: string[][]
+ sig: string
+ [key: string]: unknown
+ }[]
+
+ public constructor(
+ remoteItems: { timestamp: number; id: Uint8Array }[],
+ remoteEvents: FakeSocket['remoteEvents'] = [],
+ ) {
+ super()
+ const storage = new NegentropyStorageVector()
+ for (const item of remoteItems) {
+ storage.insert(item.timestamp, item.id)
+ }
+ storage.seal()
+ this.responder = new Negentropy(storage)
+ this.remoteEvents = remoteEvents
+ }
+
+ public send(raw: string): void {
+ const message = JSON.parse(raw) as unknown[]
+ this.sent.push(message)
+
+ switch (message[0]) {
+ case 'NEG-OPEN':
+ case 'NEG-MSG': {
+ const neg = this.responder as Negentropy
+ const { output } = neg.reconcile(Buffer.from((message[3] as string) ?? (message[2] as string), 'hex'))
+ if (output) {
+ this.deliver(['NEG-MSG', message[1], Buffer.from(output).toString('hex')])
+ }
+ return
+ }
+ case 'REQ': {
+ const subscriptionId = message[1]
+ for (const event of this.remoteEvents) {
+ this.deliver(['EVENT', subscriptionId, event])
+ }
+ this.deliver(['EOSE', subscriptionId])
+ return
+ }
+ case 'CLOSE':
+ case 'NEG-CLOSE':
+ return
+ default:
+ return
+ }
+ }
+
+ public close(): void {
+ this.closed = true
+ this.emit('close')
+ }
+
+ public deliver(message: unknown[]): void {
+ this.emit('message', Buffer.from(JSON.stringify(message)))
+ }
+}
+
+const id = (byte: number): Uint8Array => Uint8Array.from({ length: 32 }, () => byte)
+
+describe('PeeringConnection', () => {
+ it('reconciles ids in both directions over NEG-OPEN/NEG-MSG', async () => {
+ const localId = id(1)
+ const remoteId = id(2)
+ const socket = new FakeSocket([{ timestamp: 10, id: remoteId }])
+ const connection = new PeeringConnection(socket as any)
+
+ const result = await connection.reconcile(
+ { kinds: [1] },
+ [{ event_created_at: 5, event_id: Buffer.from(localId) }],
+ 1_000,
+ )
+
+ expect(result.needIds).to.deep.equal([Buffer.from(remoteId).toString('hex')])
+ expect(result.haveIds).to.deep.equal([Buffer.from(localId).toString('hex')])
+ expect(socket.sent.some((message) => message[0] === 'NEG-OPEN')).to.equal(true)
+ expect(socket.sent.some((message) => message[0] === 'NEG-CLOSE')).to.equal(true)
+ })
+
+ it('fetches requested events and resolves on EOSE', async () => {
+ const remoteId = id(2)
+ const event = {
+ id: Buffer.from(remoteId).toString('hex'),
+ pubkey: 'f'.repeat(64),
+ created_at: 10,
+ kind: 1,
+ tags: [],
+ content: 'hello',
+ sig: 'a'.repeat(128),
+ }
+ const socket = new FakeSocket([], [event])
+ const connection = new PeeringConnection(socket as any)
+
+ const events = await connection.fetchEvents([event.id], 1_000)
+
+ expect(events).to.deep.equal([event])
+ })
+
+ it('ignores unrequested and duplicate fetch replies', async () => {
+ const wanted = {
+ id: 'b'.repeat(64),
+ pubkey: 'f'.repeat(64),
+ created_at: 10,
+ kind: 1,
+ tags: [],
+ content: 'wanted',
+ sig: 'a'.repeat(128),
+ }
+ const extra = { ...wanted, id: 'c'.repeat(64), content: 'extra' }
+ const socket = new FakeSocket([], [wanted, wanted, extra])
+ const connection = new PeeringConnection(socket as any)
+
+ const events = await connection.fetchEvents([wanted.id], 1_000)
+
+ expect(events).to.deep.equal([wanted])
+ })
+
+ it('keeps a live subscription open after EOSE', async () => {
+ const socket = new FakeSocket([])
+ const connection = new PeeringConnection(socket as any)
+ const received: unknown[] = []
+
+ await connection.subscribe([{ kinds: [1] }], (event) => received.push(event), 1_000)
+
+ socket.deliver(['EVENT', socket.sent.find((message) => message[0] === 'REQ')?.[1], { id: 'a'.repeat(64), kind: 1 }])
+
+ expect(received).to.have.length(1)
+ })
+
+ it('rejects a fetch that exceeds its byte budget without closing the connection', async () => {
+ const event = {
+ id: 'b'.repeat(64),
+ pubkey: 'f'.repeat(64),
+ created_at: 10,
+ kind: 1,
+ tags: [],
+ content: 'x'.repeat(1_000),
+ sig: 'a'.repeat(128),
+ }
+ const socket = new FakeSocket([], [event])
+ const connection = new PeeringConnection(socket as any)
+
+ const error = await connection.fetchEvents([event.id], 1_000, 100).then(
+ () => undefined,
+ (err: Error) => err,
+ )
+
+ expect(error).to.be.instanceOf(PeeringFetchOverflowError)
+ expect(socket.closed).to.equal(false)
+
+ const events = await connection.fetchEvents([event.id], 1_000)
+ expect(events).to.deep.equal([event])
+ })
+
+ it('rejects when the peer returns NEG-ERR', async () => {
+ const socket = new FakeSocket([])
+ const connection = new PeeringConnection(socket as any)
+ socket.send = function (raw: string): void {
+ const message = JSON.parse(raw) as unknown[]
+ this.deliver(['NEG-ERR', message[1], 'blocked: too big'])
+ }
+
+ const error = await connection.reconcile({ kinds: [1] }, [], 1_000).then(
+ () => undefined,
+ (err: Error) => err,
+ )
+
+ expect(error?.message).to.equal('blocked: too big')
+ })
+})
diff --git a/test/unit/utils/peering-scope.spec.ts b/test/unit/utils/peering-scope.spec.ts
new file mode 100644
index 000000000..484e4fb5d
--- /dev/null
+++ b/test/unit/utils/peering-scope.spec.ts
@@ -0,0 +1,83 @@
+import { expect } from 'chai'
+
+import {
+ expandPeeringKinds,
+ peeringScopesToLiveFilters,
+ peeringScopeToFilter,
+ peeringScopeToFilters,
+ splitPeeringFilter,
+} from '../../../src/utils/peering-scope'
+
+describe('peering scope filters', () => {
+ it('expands kind ranges into the explicit list NIP-01 filters require', () => {
+ expect(expandPeeringKinds([1, [3, 5], 4])).to.deep.equal([1, 3, 4, 5])
+ })
+
+ it('turns an absolute scope into a filter without a limit', () => {
+ expect(peeringScopeToFilter({ kinds: [1, [6, 7]], since: 100, until: 200 }, 1_000)).to.deep.equal({
+ kinds: [1, 6, 7],
+ since: 100,
+ until: 200,
+ })
+ })
+
+ it('resolves a rolling window against the reconciliation time', () => {
+ expect(peeringScopeToFilter({ kinds: [1], windowSeconds: 300 }, 1_000)).to.deep.equal({
+ kinds: [1],
+ since: 700,
+ })
+ })
+
+ it('subscribes live from the recent overlap and skips scopes that already ended', () => {
+ const filters = peeringScopesToLiveFilters(
+ [
+ { kinds: [1], since: 100 },
+ { kinds: [7], until: 1_500 },
+ { kinds: [6], until: 900 },
+ { kinds: [30000], windowSeconds: 3600 },
+ ],
+ 1_000,
+ 60,
+ )
+
+ expect(filters).to.deep.equal([
+ { kinds: [1], since: 940 },
+ { kinds: [7], since: 940, until: 1_500 },
+ { kinds: [30000], since: 940 },
+ ])
+ })
+
+ it('splits wide kind ranges into filters that fit a peer limit', () => {
+ const filters = peeringScopeToFilters({ kinds: [0, [1, 7]], since: 100 }, 1_000, 4)
+
+ expect(filters.map((filter) => filter.kinds)).to.deep.equal([
+ [0, 1, 2, 3],
+ [4, 5, 6, 7],
+ ])
+ expect(filters.every((filter) => filter.since === 100)).to.equal(true)
+ })
+
+ it('never starts a short rolling window earlier than its own cutoff', () => {
+ const [filter] = peeringScopesToLiveFilters([{ kinds: [1], windowSeconds: 5 }], 1_000, 60)
+
+ expect(filter).to.deep.equal({ kinds: [1], since: 995 })
+ })
+
+ it('splits an oversized absolute range at its midpoint', () => {
+ expect(splitPeeringFilter({ kinds: [1], since: 100, until: 200 }, 1_000)).to.deep.equal([
+ { kinds: [1], since: 100, until: 150 },
+ { kinds: [1], since: 151, until: 200 },
+ ])
+ })
+
+ it('splits an unbounded filter between the beginning of time and now', () => {
+ expect(splitPeeringFilter({ kinds: [1] }, 1_000)).to.deep.equal([
+ { kinds: [1], since: 0, until: 500 },
+ { kinds: [1], since: 501, until: 1_000 },
+ ])
+ })
+
+ it('does not split a range that has no room left', () => {
+ expect(splitPeeringFilter({ kinds: [1], since: 500, until: 500 }, 1_000)).to.equal(undefined)
+ })
+})
diff --git a/test/unit/utils/peering-sync-state.spec.ts b/test/unit/utils/peering-sync-state.spec.ts
new file mode 100644
index 000000000..a0ece0710
--- /dev/null
+++ b/test/unit/utils/peering-sync-state.spec.ts
@@ -0,0 +1,66 @@
+import { expect } from 'chai'
+
+import { ICacheAdapter } from '../../../src/@types/adapters'
+import { PEERING_SYNC_STATE_INDEX_KEY, PeeringSyncStateStore } from '../../../src/utils/peering-sync-state'
+
+const createCache = (): ICacheAdapter & { values: Map; sets: Map } => {
+ const values = new Map()
+ const sets = new Map()
+
+ return {
+ values,
+ sets,
+ getKey: async (key) => values.get(key) ?? '',
+ hasKey: async (key) => values.has(key),
+ setKey: async (key, value) => {
+ values.set(key, value)
+ return true
+ },
+ setKeyIfNotExists: async (key, value) => {
+ if (values.has(key)) {
+ return false
+ }
+ values.set(key, value)
+ return true
+ },
+ addToSortedSet: async () => 0,
+ removeRangeByScoreFromSortedSet: async () => 0,
+ removeRangeByRankFromSortedSet: async () => 0,
+ getRangeFromSortedSet: async () => [],
+ setKeyExpiry: async () => {},
+ deleteKey: async () => 0,
+ getHKey: async () => '',
+ setHKey: async () => true,
+ addToSet: async (key, members) => {
+ const current = sets.get(key) ?? []
+ sets.set(key, [...new Set([...current, ...members])])
+ return members.length
+ },
+ getSetMembers: async (key) => sets.get(key) ?? [],
+ eval: async () => null,
+ }
+}
+
+describe('PeeringSyncStateStore', () => {
+ it('publishes and lists per-peer state', async () => {
+ const cache = createCache()
+ const store = new PeeringSyncStateStore(cache)
+
+ await store.publish({
+ pubkey: 'A'.repeat(64),
+ relayUrl: 'wss://peer.example.com',
+ status: 'live',
+ scopeCount: 1,
+ updatedAt: '2024-01-01T00:00:00.000Z',
+ lastReconciledAt: '2024-01-01T00:00:00.000Z',
+ lastError: 'failed to connect to wss://peer.example.com',
+ })
+
+ expect(cache.sets.get(PEERING_SYNC_STATE_INDEX_KEY)).to.deep.equal(['a'.repeat(64)])
+ const states = await store.list()
+
+ expect(states).to.have.length(1)
+ expect(states[0].pubkey).to.equal('a'.repeat(64))
+ expect(states[0].lastError).to.equal('failed to connect to [url]')
+ })
+})
diff --git a/test/unit/utils/safe-nip11-lookup.spec.ts b/test/unit/utils/safe-nip11-lookup.spec.ts
new file mode 100644
index 000000000..d3d49ab31
--- /dev/null
+++ b/test/unit/utils/safe-nip11-lookup.spec.ts
@@ -0,0 +1,37 @@
+import { expect } from 'chai'
+
+import { isBlockedNip11Address, safeNip11Lookup } from '../../../src/utils/relay-probe/nip11-probe'
+
+describe('safe NIP-11 lookup', () => {
+ it('blocks loopback, private, link-local, multicast, and mapped addresses', () => {
+ for (const address of [
+ '127.0.0.1',
+ '10.0.0.1',
+ '172.16.0.1',
+ '192.168.1.1',
+ '169.254.1.1',
+ '100.64.0.1',
+ '224.0.0.1',
+ '::1',
+ 'fd00::1',
+ 'fe80::1',
+ '::ffff:127.0.0.1',
+ ]) {
+ expect(isBlockedNip11Address(address), address).to.equal(true)
+ }
+ })
+
+ it('allows public addresses', () => {
+ expect(isBlockedNip11Address('8.8.8.8')).to.equal(false)
+ expect(isBlockedNip11Address('2606:4700:4700::1111')).to.equal(false)
+ })
+
+ it('rejects a literal private address before connecting', async () => {
+ const error = await safeNip11Lookup('127.0.0.1', {}).then(
+ () => undefined,
+ (err: Error) => err,
+ )
+
+ expect(error?.message).to.include('resolving to a private address')
+ })
+})