test(nip77): cover negentropy wire protocol and sessions

Signed-off-by: Priyanshubhartistm <bhartipriyanshustm@gmail.com>

Author: Priyanshubhartistm

test/unit/adapters/web-socket-adapter.spec.ts

@@ -13,6 +13,7 @@ const { expect } = chai
 
 import { WebSocketAdapterEvent, WebSocketServerAdapterEvent } from '../../../src/constants/adapter'
 import { IWebSocketServerAdapter } from '../../../src/@types/adapters'
+import { Negentropy, NegentropyStorageVector } from '../../../src/utils/negentropy'
 import { WebSocketAdapter } from '../../../src/adapters/web-socket-adapter'
 import { markActionableTarget, resetHiddenContentCache } from '../../../src/utils/hidden-content-cache'
 
@@ -976,4 +977,86 @@ describe('WebSocketAdapter', () => {
       adapter2.removeAllListeners()
     })
   })
+
+  describe('NIP-77 negentropy sessions', () => {
+    const emptyStorage = () => {
+      const storage = new NegentropyStorageVector()
+      storage.seal()
+
+      return new Negentropy(storage)
+    }
+
+    const openSession = (id: string, idleTimeoutMs = 60000) => {
+      const sessions = adapter.getNegentropySessions()
+      sessions.activate(id, sessions.reserve(id, 3) as symbol, emptyStorage(), idleTimeoutMs)
+    }
+
+    const sentMessages = () => client.send.getCalls().map((call: any) => JSON.parse(call.args[0]))
+
+    const triggerMessage = async (payload: unknown) => {
+      const call = client.on.getCalls().find((c: any) => c.args[0] === 'message')
+      await call.args[1](Buffer.from(JSON.stringify(payload)))
+    }
+
+    it('hands the same sessions to every caller', () => {
+      expect(adapter.getNegentropySessions()).to.equal(adapter.getNegentropySessions())
+    })
+
+    it('dispatches NEG-OPEN, NEG-MSG and NEG-CLOSE to a message handler', async () => {
+      const handleMessage = sandbox.stub().resolves()
+      createMessageHandler.returns({ handleMessage })
+
+      await triggerMessage(['NEG-OPEN', 'sync1', { kinds: [1] }, '6100000200'])
+      await triggerMessage(['NEG-MSG', 'sync1', '6100'])
+      await triggerMessage(['NEG-CLOSE', 'sync1'])
+
+      expect(handleMessage.getCalls().map((call: any) => call.args[0][0])).to.deep.equal([
+        'NEG-OPEN',
+        'NEG-MSG',
+        'NEG-CLOSE',
+      ])
+    })
+
+    it('replies with a NOTICE to a NEG-OPEN whose initial message is not hex', async () => {
+      await triggerMessage(['NEG-OPEN', 'sync1', {}, 'not hex'])
+
+      expect(createMessageHandler).to.not.have.been.called
+      expect(sentMessages()).to.have.lengthOf(1)
+      expect(sentMessages()[0][0]).to.equal('NOTICE')
+      expect(sentMessages()[0][1]).to.match(/^invalid:/)
+    })
+
+    it('tells the client when a session times out', () => {
+      const clock = sandbox.useFakeTimers()
+      openSession('sync1', 1000)
+
+      clock.tick(1000)
+
+      expect(sentMessages()).to.deep.include(['NEG-ERR', 'sync1', 'closed: negentropy session timed out'])
+      expect(adapter.getNegentropySessions().getSessionIds()).to.deep.equal([])
+    })
+
+    it('drops every session and its timer when the client disconnects', () => {
+      const clock = sandbox.useFakeTimers()
+      openSession('sync1', 1000)
+      const onClose = client.on.getCalls().find((call: any) => call.args[0] === 'close').args[1]
+
+      onClose()
+      clock.tick(5000)
+
+      expect(adapter.getNegentropySessions().getSessionIds()).to.deep.equal([])
+      expect(sentMessages().filter((message: any[]) => message[0] === 'NEG-ERR')).to.deep.equal([])
+    })
+
+    it('tells the client its sessions are closed when the relay drains', () => {
+      openSession('a')
+      openSession('b')
+
+      adapter.drainAndClose('relay shutting down')
+
+      expect(sentMessages()).to.deep.include(['NEG-ERR', 'a', 'closed: relay shutting down'])
+      expect(sentMessages()).to.deep.include(['NEG-ERR', 'b', 'closed: relay shutting down'])
+      expect(adapter.getNegentropySessions().getSessionIds()).to.deep.equal([])
+    })
+  })
 })

test/unit/factories/message-handler-factory.spec.ts

@@ -15,6 +15,9 @@ import { EventMessageHandler } from '../../../src/handlers/event-message-handler
 import { IWebSocketAdapter } from '../../../src/@types/adapters'
 import { messageHandlerFactory } from '../../../src/factories/message-handler-factory'
 import { CountMessageHandler } from '../../../src/handlers/count-message-handler'
+import { NegCloseMessageHandler } from '../../../src/handlers/neg-close-message-handler'
+import { NegMsgMessageHandler } from '../../../src/handlers/neg-msg-message-handler'
+import { NegOpenMessageHandler } from '../../../src/handlers/neg-open-message-handler'
 import { SubscribeMessageHandler } from '../../../src/handlers/subscribe-message-handler'
 import { UnsubscribeMessageHandler } from '../../../src/handlers/unsubscribe-message-handler'
 import * as cacheModule from '../../../src/cache/client'
@@ -101,6 +104,24 @@ describe('messageHandlerFactory', () => {
     expect(factory([message, adapter])).to.be.an.instanceOf(AuthMessageHandler)
   })
 
+  it('returns NegOpenMessageHandler when given a NEG-OPEN message', () => {
+    message = [MessageType.NEG_OPEN, 'sync1', {}, '6100000200'] as any
+
+    expect(factory([message, adapter])).to.be.an.instanceOf(NegOpenMessageHandler)
+  })
+
+  it('returns NegMsgMessageHandler when given a NEG-MSG message', () => {
+    message = [MessageType.NEG_MSG, 'sync1', '6100'] as any
+
+    expect(factory([message, adapter])).to.be.an.instanceOf(NegMsgMessageHandler)
+  })
+
+  it('returns NegCloseMessageHandler when given a NEG-CLOSE message', () => {
+    message = [MessageType.NEG_CLOSE, 'sync1'] as any
+
+    expect(factory([message, adapter])).to.be.an.instanceOf(NegCloseMessageHandler)
+  })
+
   it('throws when given an invalid message', () => {
     message = [] as any
 

test/unit/handlers/neg-close-message-handler.spec.ts

@@ -0,0 +1,60 @@
+import { expect } from 'chai'
+import EventEmitter from 'events'
+
+import { IWebSocketAdapter } from '../../../src/@types/adapters'
+import { MessageType } from '../../../src/@types/messages'
+import { NegCloseMessageHandler } from '../../../src/handlers/neg-close-message-handler'
+import { Negentropy, NegentropyStorageVector } from '../../../src/utils/negentropy'
+import { NegentropySessions } from '../../../src/utils/negentropy-sessions'
+import { WebSocketAdapterEvent } from '../../../src/constants/adapter'
+
+describe('NegCloseMessageHandler', () => {
+  let sessions: NegentropySessions
+  let webSocket: IWebSocketAdapter
+  let handler: NegCloseMessageHandler
+  let emitted: unknown[][]
+
+  const openSession = (id: string) => {
+    const storage = new NegentropyStorageVector()
+    storage.seal()
+    sessions.activate(id, sessions.reserve(id, 3) as symbol, new Negentropy(storage), 60000)
+  }
+
+  beforeEach(() => {
+    sessions = new NegentropySessions(() => undefined)
+    webSocket = new EventEmitter() as any
+    webSocket.getNegentropySessions = () => sessions
+    emitted = []
+    webSocket.on(WebSocketAdapterEvent.Message, (...args) => emitted.push(args))
+    handler = new NegCloseMessageHandler(webSocket)
+  })
+
+  afterEach(() => {
+    sessions.closeAll()
+    webSocket.removeAllListeners()
+  })
+
+  it('closes the session without replying', async () => {
+    openSession('sync1')
+
+    await handler.handleMessage([MessageType.NEG_CLOSE, 'sync1'] as any)
+
+    expect(sessions.getSessionIds()).to.deep.equal([])
+    expect(emitted).to.deep.equal([])
+  })
+
+  it('only closes the named session', async () => {
+    openSession('a')
+    openSession('b')
+
+    await handler.handleMessage([MessageType.NEG_CLOSE, 'a'] as any)
+
+    expect(sessions.getSessionIds()).to.deep.equal(['b'])
+  })
+
+  it('ignores an unknown id', async () => {
+    await handler.handleMessage([MessageType.NEG_CLOSE, 'nope'] as any)
+
+    expect(emitted).to.deep.equal([])
+  })
+})

test/unit/handlers/neg-msg-message-handler.spec.ts

@@ -0,0 +1,138 @@
+import chai from 'chai'
+import { createHash } from 'crypto'
+import EventEmitter from 'events'
+import Sinon from 'sinon'
+import sinonChai from 'sinon-chai'
+
+import { IWebSocketAdapter } from '../../../src/@types/adapters'
+import { MessageType } from '../../../src/@types/messages'
+import { Settings } from '../../../src/@types/settings'
+import { WebSocketAdapterEvent } from '../../../src/constants/adapter'
+import { NegMsgMessageHandler } from '../../../src/handlers/neg-msg-message-handler'
+import { Negentropy, NegentropyStorageVector } from '../../../src/utils/negentropy'
+import { NegentropySessions } from '../../../src/utils/negentropy-sessions'
+
+chai.use(sinonChai)
+const { expect } = chai
+
+const storageOf = (count: number): NegentropyStorageVector => {
+  const storage = new NegentropyStorageVector()
+  for (let i = 0; i < count; i++) {
+    storage.insert(1700000000 + i, createHash('sha256').update(`neg-msg-${i}`).digest())
+  }
+  storage.seal()
+
+  return storage
+}
+
+describe('NegMsgMessageHandler', () => {
+  let sandbox: Sinon.SinonSandbox
+  let webSocket: IWebSocketAdapter
+  let sessions: NegentropySessions
+  let settings: Settings
+  let emitted: Sinon.SinonStub
+  let handler: NegMsgMessageHandler
+
+  const send = (id: string, hex: string) => handler.handleMessage([MessageType.NEG_MSG, id, hex] as any)
+
+  const openSession = (id: string, itemCount = 5) => {
+    const token = sessions.reserve(id, 3) as symbol
+    sessions.activate(id, token, new Negentropy(storageOf(itemCount)), 60000)
+  }
+
+  const validMessageHex = Buffer.from(new Negentropy(storageOf(0)).initiate()).toString('hex')
+
+  beforeEach(() => {
+    sandbox = Sinon.createSandbox()
+    sessions = new NegentropySessions(() => undefined)
+    settings = { nip77: { enabled: true } } as Settings
+
+    webSocket = new EventEmitter() as any
+    webSocket.getNegentropySessions = () => sessions
+
+    emitted = sandbox.stub()
+    webSocket.on(WebSocketAdapterEvent.Message, emitted)
+
+    handler = new NegMsgMessageHandler(webSocket, () => settings)
+  })
+
+  afterEach(() => {
+    sessions.closeAll()
+    webSocket.removeAllListeners()
+    sandbox.restore()
+  })
+
+  it('answers a message on an open session with a NEG-MSG', async () => {
+    openSession('sync1')
+
+    await send('sync1', validMessageHex)
+
+    expect(emitted).to.have.been.calledOnce
+    const [type, id, payload] = emitted.firstCall.args[0]
+    expect(type).to.equal(MessageType.NEG_MSG)
+    expect(id).to.equal('sync1')
+    expect(payload).to.match(/^61(?:[0-9a-f]{2})+$/)
+    expect(sessions.getSessionIds()).to.deep.equal(['sync1'])
+  })
+
+  it('replies NEG-ERR closed for an unknown subscription id', async () => {
+    await send('nope', validMessageHex)
+
+    expect(emitted).to.have.been.calledOnceWithExactly([
+      MessageType.NEG_ERR,
+      'nope',
+      'closed: unknown negentropy subscription',
+    ])
+  })
+
+  it('does not answer for a REQ subscription id: the namespaces are separate', async () => {
+    // A REQ subscription named "sync1" lives in the adapter's subscriptions, not in the negentropy sessions.
+    await send('sync1', validMessageHex)
+
+    expect(emitted.firstCall.args[0][0]).to.equal(MessageType.NEG_ERR)
+  })
+
+  it('closes the session and replies NEG-ERR when the message is malformed', async () => {
+    openSession('sync1')
+
+    await send('sync1', '6100')
+
+    expect(emitted).to.have.been.calledOnceWithExactly([
+      MessageType.NEG_ERR,
+      'sync1',
+      'invalid: malformed negentropy message',
+    ])
+    expect(sessions.getSessionIds()).to.deep.equal([])
+  })
+
+  it('keeps other sessions when one is closed by an error', async () => {
+    openSession('a')
+    openSession('b')
+
+    await send('a', '6100')
+
+    expect(sessions.getSessionIds()).to.deep.equal(['b'])
+  })
+
+  it('ends the session with NEG-ERR blocked when NIP-77 is turned off while it is open', async () => {
+    openSession('sync1')
+    settings = { nip77: { enabled: false } } as Settings
+
+    await send('sync1', validMessageHex)
+
+    expect(emitted).to.have.been.calledOnceWithExactly([
+      MessageType.NEG_ERR,
+      'sync1',
+      'blocked: NIP-77 is disabled by relay configuration',
+    ])
+    expect(sessions.getSessionIds()).to.deep.equal([])
+  })
+
+  it('answers an unsupported protocol version with just the supported version', async () => {
+    openSession('sync1')
+
+    await send('sync1', '62')
+
+    expect(emitted).to.have.been.calledOnceWithExactly([MessageType.NEG_MSG, 'sync1', '61'])
+  })
+})