feat(nip77): add NEG-OPEN/NEG-MSG/NEG-CLOSE wire protocol

.changeset/nip77-wire-protocol.md

@@ -0,0 +1,5 @@
+---
+"nostream": minor
+---
+
+feat(nip77): serve NEG-OPEN, NEG-MSG and NEG-CLOSE negentropy sync for the PostgreSQL event store

.knip.json

@@ -15,7 +15,6 @@
   ],
   "ignore": [
     ".nostr/**",
-    "src/utils/negentropy.ts",
     "src/utils/relay-probe/**"
   ],
   "commitlint": false,

CONFIGURATION.md

@@ -232,6 +232,10 @@ The settings below are listed in alphabetical order by name. Please keep this ta
 | nip66.timeouts.nip11Ms                      | NIP-11 fetch timeout in milliseconds. Defaults to 10000. |
 | nip66.timeouts.tlsMs                        | TLS probe timeout in milliseconds. Defaults to 10000. |
 | nip66.timeouts.wsRttMs                      | WebSocket open RTT probe timeout in milliseconds. Defaults to 10000. |
+| nip77.enabled                               | Enable NIP-77 (Negentropy) set-reconciliation sync, so clients and other relays can find which events differ without downloading them all. Defaults to false. Only answers when `eventStore.backend` is `postgres`; other backends bring their own negentropy implementation. When `nip42.restrictedReads` is enabled, reconciliation would list the ids of restricted events, so a `NEG-OPEN` filter that could match one (it names a restricted kind or no kinds at all) is only accepted when scoped by `authors` or `#p` to pubkeys the client has authenticated as, and is re-checked on every `NEG-MSG`; otherwise the session is closed with `NEG-ERR auth-required:`. Each open session holds the ids of every event matching its filter in memory, so size the limits below for your hardware. |
+| nip77.maxEventsPerSession                   | Most events one session may cover. A `NEG-OPEN` whose filter matches more is rejected with `NEG-ERR blocked:` (clients should split the sync into `since`/`until` windows). The filter's `limit` is ignored, since reconciliation needs the whole set. Defaults to 100000. |
+| nip77.maxSessionsPerConnection              | Concurrent negentropy sessions one websocket connection may hold; opening an id that is already open replaces it. Further `NEG-OPEN`s are rejected with `NEG-ERR blocked:`. Defaults to 3. |
+| nip77.sessionIdleTimeoutMs                  | A session that receives no `NEG-MSG` for this many milliseconds is closed with `NEG-ERR closed:`. Defaults to 60000. |
 | paymentProcessors.lnbits.baseURL            | Base URL of your Lnbits instance. |
 | paymentProcessors.lnbits.callbackBaseURL    | Public-facing Nostream's Lnbits Callback URL. (e.g. https://relay.your-domain.com/callbacks/lnbits) |
 | paymentProcessors.lnurl.invoiceURL          | [LUD-06 Pay Request](https://github.com/lnurl/luds/blob/luds/06.md) provider URL. (e.g. https://getalby.com/lnurlp/your-username) |

resources/default-settings.yaml

@@ -148,6 +148,15 @@ nip56:
   # When true, events matching an actionable report are excluded from
   # REQ/COUNT results. Has no effect unless enabled is also true.
   hideActionableReports: false
+nip77:
+  # NIP-77 negentropy sync. Only answers when eventStore.backend is postgres.
+  enabled: false
+  # Most events a single session may hold in memory.
+  maxEventsPerSession: 100000
+  # Concurrent negentropy sessions per connection.
+  maxSessionsPerConnection: 3
+  # Close a session with no NEG-MSG for this long.
+  sessionIdleTimeoutMs: 60000
 network:
   maxPayloadSize: 524288
   # Uncomment only when using a trusted reverse proxy and configuring trustedProxies.

src/@types/adapters.ts

@@ -1,4 +1,5 @@
 import { EventEmitter } from 'node:stream'
+import { INegentropySessions } from './negentropy'
 import { SubscriptionFilter } from './subscription'
 
 export interface IWebSocketServerAdapter extends EventEmitter, IWebServerAdapter {
@@ -20,6 +21,8 @@ export type IWebSocketAdapter = EventEmitter & {
   /** Returns false if this AUTH event id was already accepted on this socket. */
   addAuthenticatedPubkey(pubkey: string, authEventId: string): boolean
   drainAndClose(reason?: string): void
+  /** NIP-77 sessions of this connection. */
+  getNegentropySessions(): INegentropySessions
 }
 
 export interface ICacheAdapter {

src/@types/messages.ts

@@ -13,13 +13,35 @@ export enum MessageType {
   COUNT = 'COUNT',
   CLOSED = 'CLOSED',
   AUTH = 'AUTH',
+  NEG_OPEN = 'NEG-OPEN',
+  NEG_MSG = 'NEG-MSG',
+  NEG_CLOSE = 'NEG-CLOSE',
+  NEG_ERR = 'NEG-ERR',
 }
 
-export type IncomingMessage = (SubscribeMessage | IncomingEventMessage | UnsubscribeMessage | CountMessage | AuthMessage) & {
+export type IncomingMessage = (
+  | SubscribeMessage
+  | IncomingEventMessage
+  | UnsubscribeMessage
+  | CountMessage
+  | AuthMessage
+  | NegOpenMessage
+  | NegMsgMessage
+  | NegCloseMessage
+) & {
   [ContextMetadataKey]?: ContextMetadata
 }
 
-export type OutgoingMessage = OutgoingEventMessage | EndOfStoredEventsNotice | NoticeMessage | CommandResult | CountResultMessage | ClosedMessage | AuthChallengeMessage
+export type OutgoingMessage =
+  | OutgoingEventMessage
+  | EndOfStoredEventsNotice
+  | NoticeMessage
+  | CommandResult
+  | CountResultMessage
+  | ClosedMessage
+  | AuthChallengeMessage
+  | NegMsgMessage
+  | NegErrMessage
 
 export type SubscribeMessage = {
   [index in Range<2, 100>]: SubscriptionFilter
@@ -103,3 +125,32 @@ export interface AuthChallengeMessage {
   0: MessageType.AUTH
   1: string
 }
+
+// NIP-77
+export interface NegOpenMessage {
+  0: MessageType.NEG_OPEN
+  1: SubscriptionId
+  2: SubscriptionFilter
+  /** Hex-encoded initial negentropy message. */
+  3: string
+}
+
+/** Sent in both directions; the payload is a hex-encoded negentropy message. */
+export interface NegMsgMessage {
+  0: MessageType.NEG_MSG
+  1: SubscriptionId
+  2: string
+}
+
+export interface NegCloseMessage {
+  0: MessageType.NEG_CLOSE
+  1: SubscriptionId
+}
+
+export interface NegErrMessage {
+  0: MessageType.NEG_ERR
+  1: SubscriptionId
+  2: string
+  /** For `blocked:` replies, the most records a session may hold. */
+  3?: number
+}

src/@types/negentropy.ts

@@ -0,0 +1,42 @@
+import { Negentropy } from '../utils/negentropy'
+import { SubscriptionFilter, SubscriptionId } from './subscription'
+
+/** Proves a caller still owns a subscription id after awaiting; see `INegentropySessions.reserve`. */
+export type NegentropySessionToken = symbol
+
+export interface NegentropySession {
+  negentropy: Negentropy
+  /** The filter the session was opened with, so access can be re-checked on every message. */
+  filter: SubscriptionFilter
+}
+
+export interface INegentropySessions {
+  /**
+   * Claims `subscriptionId` for a new session, closing any session already open under that id.
+   * Returns undefined when the connection already holds `maxLimit` other sessions, or already has
+   * `maxLimit` event reads in flight -- including reads of requests that have since been replaced,
+   * which keep running until `endRead` is called. The caller must call `endRead` for a token it got.
+   */
+  reserve(subscriptionId: SubscriptionId, maxLimit: number): NegentropySessionToken | undefined
+  /** Marks the event read started for a reservation as finished, successfully or not. */
+  endRead(token: NegentropySessionToken): void
+  /** Whether `token` is still the reservation for `subscriptionId`, i.e. nobody replaced or closed it. */
+  isCurrent(subscriptionId: SubscriptionId, token: NegentropySessionToken): boolean
+  /**
+   * Turns a reservation into a live session. Returns false when the reservation was closed or
+   * replaced while the caller was busy (e.g. reading events), in which case nothing is stored.
+   */
+  activate(
+    subscriptionId: SubscriptionId,
+    token: NegentropySessionToken,
+    session: NegentropySession,
+    idleTimeoutMs: number,
+  ): boolean
+  /** The live session for an id, with its idle timer restarted. */
+  touch(subscriptionId: SubscriptionId): NegentropySession | undefined
+  /** Drops a reservation that never became a session; a no-op if it has since been replaced. */
+  release(subscriptionId: SubscriptionId, token: NegentropySessionToken): void
+  close(subscriptionId: SubscriptionId): boolean
+  closeAll(): void
+  getSessionIds(): SubscriptionId[]
+}

src/@types/repositories.ts

@@ -44,7 +44,7 @@ export interface IEventRepository {
   upsertMany(events: Event[]): Promise<number>
   findByFilters(filters: SubscriptionFilter[]): IQueryResult<DBEvent[]>
   countByFilters(filters: SubscriptionFilter[]): Promise<number>
-  findReconciliationItems(filter: SubscriptionFilter): IQueryResult<ReconciliationItem[]>
+  findReconciliationItems(filter: SubscriptionFilter, limit?: number): IQueryResult<ReconciliationItem[]>
   deleteByPubkeyAndIds(pubkey: Pubkey, ids: EventId[]): Promise<number>
   deleteByPubkeyExceptKinds(pubkey: Pubkey, excludedKinds: number[]): Promise<number>
   hasActiveRequestToVanish(pubkey: Pubkey): Promise<boolean>

src/@types/settings.ts

@@ -314,6 +314,24 @@ export interface Nip45Settings {
   enabled?: boolean
 }
 
+export interface Nip77Settings {
+  /**
+   * NIP-77 negentropy sync. Defaults to false. Even when true it only answers
+   * if `eventStore.backend` is postgres: other backends (e.g. strfry) bring
+   * their own negentropy implementation.
+   */
+  enabled?: boolean
+  /**
+   * Most events one session may hold in memory; larger queries are rejected
+   * with `NEG-ERR blocked:`. Defaults to 100000.
+   */
+  maxEventsPerSession?: number
+  /** Concurrent negentropy sessions allowed on one connection. Defaults to 3. */
+  maxSessionsPerConnection?: number
+  /** A session with no `NEG-MSG` for this long is closed with `NEG-ERR closed:`. Defaults to 60000. */
+  sessionIdleTimeoutMs?: number
+}
+
 export interface Nip50Settings {
   enabled?: boolean
   language?: string
@@ -491,5 +509,6 @@ export interface Settings {
   nip50?: Nip50Settings
   nip56?: Nip56Settings
   nip66?: Nip66Settings
+  nip77?: Nip77Settings
   wot?: WoTSettings
 }

src/adapters/web-socket-adapter.ts

@@ -8,11 +8,14 @@ import { ContextMetadata, Factory } from '../@types/base'
 import {
   createAuthChallengeMessage,
   createClosedMessage,
+  createNegErrMessage,
   createNoticeMessage,
   createOutgoingEventMessage,
 } from '../utils/messages'
 import { IAbortable, IMessageHandler } from '../@types/message-handlers'
 import { IncomingMessage, MessageType, OutgoingMessage } from '../@types/messages'
+import { INegentropySessions } from '../@types/negentropy'
+import { NegentropySessions } from '../utils/negentropy-sessions'
 import { IWebSocketAdapter, IWebSocketServerAdapter } from '../@types/adapters'
 import { SubscriptionFilter, SubscriptionId } from '../@types/subscription'
 import { WebSocketAdapterEvent, WebSocketServerAdapterEvent } from '../constants/adapter'
@@ -42,6 +45,8 @@ export class WebSocketAdapter extends EventEmitter implements IWebSocketAdapter
   private alive: boolean
   private subscriptions: Map<SubscriptionId, SubscriptionFilter[]>
   private readonly session: Nip42SessionManager
+  // NIP-77
+  private readonly negentropySessions: NegentropySessions
 
   public constructor(
     private readonly client: WebSocket,
@@ -54,6 +59,9 @@ export class WebSocketAdapter extends EventEmitter implements IWebSocketAdapter
     super()
     this.alive = true
     this.subscriptions = new Map()
+    this.negentropySessions = new NegentropySessions((subscriptionId) =>
+      this.sendMessage(createNegErrMessage(subscriptionId, 'closed: negentropy session timed out')),
+    )
 
     this.clientId = Buffer.from(this.request.headers['sec-websocket-key'] as string, 'base64').toString('hex')
 
@@ -175,11 +183,21 @@ export class WebSocketAdapter extends EventEmitter implements IWebSocketAdapter
       this.sendMessage(createClosedMessage(subscriptionId, `error: ${reason}`))
     })
 
+    this.negentropySessions.getSessionIds().forEach((subscriptionId) => {
+      this.sendMessage(createNegErrMessage(subscriptionId, `closed: ${reason}`))
+    })
+    this.negentropySessions.closeAll()
+
     if (this.client.readyState === WebSocket.OPEN) {
       this.client.close(1001, reason)
     }
   }
 
+  // NIP-77
+  public getNegentropySessions(): INegentropySessions {
+    return this.negentropySessions
+  }
+
   // NIP-42
   public getChallenge(): string {
     return this.session.getChallenge()
@@ -349,6 +367,7 @@ export class WebSocketAdapter extends EventEmitter implements IWebSocketAdapter
     recordWebsocketConnectionClosed()
     this.alive = false
     this.subscriptions.clear()
+    this.negentropySessions.closeAll()
     this.session.clear()
 
     const handlers = abortableMessageHandlers.get(this.client)

src/factories/message-handler-factory.ts

@@ -13,6 +13,9 @@ import { getCacheClient } from '../cache/client'
 import { AuthMessageHandler } from '../handlers/auth-message-handler'
 import { CountMessageHandler } from '../handlers/count-message-handler'
 import { EventMessageHandler } from '../handlers/event-message-handler'
+import { NegCloseMessageHandler } from '../handlers/neg-close-message-handler'
+import { NegMsgMessageHandler } from '../handlers/neg-msg-message-handler'
+import { NegOpenMessageHandler } from '../handlers/neg-open-message-handler'
 import { SubscribeMessageHandler } from '../handlers/subscribe-message-handler'
 import { UnsubscribeMessageHandler } from '../handlers/unsubscribe-message-handler'
 import { eventStrategyFactory } from './event-strategy-factory'
@@ -74,6 +77,12 @@ export const messageHandlerFactory =
         return new CountMessageHandler(adapter, eventRepository, createSettings)
       case MessageType.AUTH:
         return new AuthMessageHandler(adapter, createSettings)
+      case MessageType.NEG_OPEN:
+        return new NegOpenMessageHandler(adapter, eventRepository, createSettings)
+      case MessageType.NEG_MSG:
+        return new NegMsgMessageHandler(adapter, createSettings)
+      case MessageType.NEG_CLOSE:
+        return new NegCloseMessageHandler(adapter)
       default:
         throw new Error(`Unknown message type: ${String(message[0]).substring(0, 64)}`)
     }

src/handlers/neg-close-message-handler.ts

@@ -0,0 +1,12 @@
+import { IMessageHandler } from '../@types/message-handlers'
+import { IWebSocketAdapter } from '../@types/adapters'
+import { NegCloseMessage } from '../@types/messages'
+
+export class NegCloseMessageHandler implements IMessageHandler {
+  public constructor(private readonly webSocket: IWebSocketAdapter) {}
+
+  // NIP-77 defines no reply to NEG-CLOSE, and closing an unknown id is harmless.
+  public async handleMessage(message: NegCloseMessage): Promise<void> {
+    this.webSocket.getNegentropySessions().close(message[1])
+  }
+}

src/handlers/neg-msg-message-handler.ts

@@ -0,0 +1,60 @@
+import { createNegErrMessage, createNegMsgMessage } from '../utils/messages'
+import { IMessageHandler } from '../@types/message-handlers'
+import { IWebSocketAdapter } from '../@types/adapters'
+import { NegMsgMessage, OutgoingMessage } from '../@types/messages'
+import { Settings } from '../@types/settings'
+import { WebSocketAdapterEvent } from '../constants/adapter'
+import { createLogger } from '../factories/logger-factory'
+import { isNip77Active } from '../utils/nip77'
+import { isReconciliationAuthorized } from '../utils/nip42'
+
+const logger = createLogger('neg-msg-message-handler')
+
+export class NegMsgMessageHandler implements IMessageHandler {
+  public constructor(
+    private readonly webSocket: IWebSocketAdapter,
+    private readonly settings: () => Settings,
+  ) {}
+
+  public async handleMessage(message: NegMsgMessage): Promise<void> {
+    const subscriptionId = message[1]
+    const sessions = this.webSocket.getNegentropySessions()
+
+    if (!isNip77Active(this.settings())) {
+      // Turned off while a session was open: end it instead of leaving it half-served.
+      sessions.close(subscriptionId)
+      this.reply(createNegErrMessage(subscriptionId, 'blocked: NIP-77 is disabled by relay configuration'))
+      return
+    }
+
+    const session = sessions.touch(subscriptionId)
+    if (!session) {
+      this.reply(createNegErrMessage(subscriptionId, 'closed: unknown negentropy subscription'))
+      return
+    }
+
+    // Access can lapse while a session is open (e.g. NIP-42 auth expiring), so check it every time.
+    if (!isReconciliationAuthorized(this.settings(), session.filter, () => this.webSocket.getAuthenticatedPubkeys())) {
+      sessions.close(subscriptionId)
+      this.reply(
+        createNegErrMessage(subscriptionId, 'auth-required: authentication is required to sync these event kinds'),
+      )
+      return
+    }
+
+    try {
+      const { output } = session.negentropy.reconcile(Buffer.from(message[2], 'hex'))
+
+      this.reply(createNegMsgMessage(subscriptionId, Buffer.from(output as Uint8Array).toString('hex')))
+    } catch (error) {
+      // NIP-77: after a NEG-ERR the subscription is closed.
+      sessions.close(subscriptionId)
+      logger('NEG-MSG %s sent a malformed negentropy message: %o', subscriptionId, error)
+      this.reply(createNegErrMessage(subscriptionId, 'invalid: malformed negentropy message'))
+    }
+  }
+
+  private reply(message: OutgoingMessage): void {
+    this.webSocket.emit(WebSocketAdapterEvent.Message, message)
+  }
+}