Fix: nip01 replaceable tiebreaker (#416)

* fix: update event upsert query to handle duplicate timestamps using event ID comparison

* test: add NIP-01 tie-breaker integration test for identical timestamps

Author: YashIIT0909

src/@types/adapters.ts

@@ -22,6 +22,7 @@ export interface ICacheAdapter {
   getKey(key: string): Promise<string>
   hasKey(key: string): Promise<boolean>
   setKey(key: string, value: string): Promise<boolean>
+  setKeyWithExpiry(key: string, value: string, ttl: number): Promise<boolean>
   addToSortedSet(key: string, set: Record<string, string> | Record<string, string>[]): Promise<number>
   removeRangeByScoreFromSortedSet(key: string, min: number, max: number): Promise<number>
   getRangeFromSortedSet(key: string, start: number, stop: number): Promise<string[]>

src/adapters/redis-adapter.ts

@@ -60,10 +60,16 @@ export class RedisAdapter implements ICacheAdapter {
 
   public async setKey(key: string, value: string): Promise<boolean> {
     await this.connection
-    debug('get %s key', key)
+    debug('set %s key', key)
     return 'OK' === await this.client.set(key, value)
   }
 
+  public async setKeyWithExpiry(key: string, value: string, ttl: number): Promise<boolean> {
+    await this.connection
+    debug('set %s key with expiry %d', key, ttl)
+    return 'OK' === await this.client.set(key, value, { EX: ttl })
+  }
+
   public async removeRangeByScoreFromSortedSet(key: string, min: number, max: number): Promise<number> {
     await this.connection
     debug('remove %d..%d range from sorted set %s', min, max, key)

src/constants/caching.ts

@@ -0,0 +1,8 @@
+/**
+ * States for user admission caching.
+ */
+export enum CacheAdmissionState {
+  ADMITTED = 'admitted',
+  NOT_ADMITTED = 'not-admitted',
+  INSUFFICIENT_BALANCE = 'insufficient-balance',
+}

src/factories/message-handler-factory.ts

@@ -1,13 +1,23 @@
+import { ICacheAdapter, IWebSocketAdapter } from '../@types/adapters'
 import { IEventRepository, IUserRepository } from '../@types/repositories'
 import { IncomingMessage, MessageType } from '../@types/messages'
 import { createSettings } from './settings-factory'
 import { EventMessageHandler } from '../handlers/event-message-handler'
 import { eventStrategyFactory } from './event-strategy-factory'
-import { IWebSocketAdapter } from '../@types/adapters'
+import { getCacheClient } from '../cache/client'
+import { RedisAdapter } from '../adapters/redis-adapter'
 import { slidingWindowRateLimiterFactory } from './rate-limiter-factory'
 import { SubscribeMessageHandler } from '../handlers/subscribe-message-handler'
 import { UnsubscribeMessageHandler } from '../handlers/unsubscribe-message-handler'
 
+let cacheAdapter: ICacheAdapter | undefined = undefined
+const getCache = (): ICacheAdapter => {
+  if (!cacheAdapter) {
+    cacheAdapter = new RedisAdapter(getCacheClient())
+  }
+  return cacheAdapter
+}
+
 export const messageHandlerFactory = (
   eventRepository: IEventRepository,
   userRepository: IUserRepository,
@@ -22,6 +32,7 @@ export const messageHandlerFactory = (
           userRepository,
           createSettings,
           slidingWindowRateLimiterFactory,
+          getCache(),
         )
       }
     case MessageType.REQ:

src/handlers/event-message-handler.ts

@@ -15,9 +15,11 @@ import {
 } from '../utils/event'
 import { IEventRepository, IUserRepository } from '../@types/repositories'
 import { IEventStrategy, IMessageHandler } from '../@types/message-handlers'
+import { CacheAdmissionState } from '../constants/caching'
 import { createCommandResult } from '../utils/messages'
 import { createLogger } from '../factories/logger-factory'
 import { Factory } from '../@types/base'
+import { ICacheAdapter } from '../@types/adapters'
 import { IncomingEventMessage } from '../@types/messages'
 import { IRateLimiter } from '../@types/utils'
 import { IWebSocketAdapter } from '../@types/adapters'
@@ -33,6 +35,7 @@ export class EventMessageHandler implements IMessageHandler {
     protected readonly userRepository: IUserRepository,
     private readonly settings: () => Settings,
     private readonly slidingWindowRateLimiter: Factory<IRateLimiter>,
+    private readonly cache: ICacheAdapter,
   ) {}
 
   public async handleMessage(message: IncomingEventMessage): Promise<void> {
@@ -313,17 +316,45 @@ export class EventMessageHandler implements IMessageHandler {
       return
     }
 
-    // const hasKey = await this.cache.hasKey(`${event.pubkey}:is-admitted`)
-    // TODO: use cache
+    const cacheKey = `${event.pubkey}:is-admitted`
+
+    try {
+      const cachedAdmission = await this.cache.getKey(cacheKey)
+      switch (cachedAdmission) {
+        case CacheAdmissionState.ADMITTED:
+          debug('cache hit for %s admission: admitted', event.pubkey)
+          return
+        case CacheAdmissionState.NOT_ADMITTED:
+          debug('cache hit for %s admission: blocked', event.pubkey)
+          return 'blocked: pubkey not admitted'
+        case CacheAdmissionState.INSUFFICIENT_BALANCE:
+          debug('cache hit for %s admission: blocked (insufficient balance)', event.pubkey)
+          return 'blocked: insufficient balance'
+        default:
+          break
+      }
+    } catch (error) {
+      debug('cache error for %s: %o', event.pubkey, error)
+    }
+
     const user = await this.userRepository.findByPubkey(event.pubkey)
     if (!user || !user.isAdmitted) {
+      this.cacheSet(cacheKey, CacheAdmissionState.NOT_ADMITTED, 60)
       return 'blocked: pubkey not admitted'
     }
 
     const minBalance = currentSettings.limits?.event?.pubkey?.minBalance ?? 0n
     if (minBalance > 0n && user.balance < minBalance) {
+      this.cacheSet(cacheKey, CacheAdmissionState.INSUFFICIENT_BALANCE, 60)
       return 'blocked: insufficient balance'
     }
+
+    this.cacheSet(cacheKey, CacheAdmissionState.ADMITTED, 300)
+  }
+
+  private cacheSet(key: string, value: string, ttl: number): void {
+    this.cache.setKeyWithExpiry(key, value, ttl)
+      .catch((error) => debug('unable to cache %s: %o', key, error))
   }
 
   protected addExpirationMetadata(event: Event): Event | ExpiringEvent {

src/repositories/event-repository.ts

@@ -227,7 +227,13 @@ export class EventRepository implements IEventRepository {
         )
       )
       .merge(omit(['event_pubkey', 'event_kind', 'event_deduplication'])(row))
-      .where('events.event_created_at', '<', row.event_created_at)
+      .where(function () {
+        this.where('events.event_created_at', '<', row.event_created_at)
+          .orWhere(function () {
+            this.where('events.event_created_at', '=', row.event_created_at)
+              .andWhere('events.event_id', '>', row.event_id)
+          })
+      })
 
     return {
       then: <T1, T2>(onfulfilled: (value: number) => T1 | PromiseLike<T1>, onrejected: (reason: any) => T2 | PromiseLike<T2>) => query.then(prop('rowCount') as () => number).then(onfulfilled, onrejected),

test/integration/features/nip-16/nip-16.feature

@@ -9,6 +9,12 @@ Feature: NIP-16 Event treatment
     When Alice subscribes to author Alice
     Then Alice receives 1 replaceable_event_0 event from Alice with content "updated" and EOSE
 
+  Scenario: Tie-breaker on Identical Timestamps
+    Given someone called Alice
+    When Alice sends two identically-timestamped replaceable_event_0 events where the second has a lower ID
+    And Alice subscribes to author Alice
+    Then Alice receives 1 replaceable_event_0 event from Alice matching the lower ID event and EOSE
+
   Scenario: Charlie sends an ephemeral event
     Given someone called Charlie
     Given someone called Alice

test/integration/features/nip-16/nip-16.feature.ts

@@ -5,7 +5,7 @@ import WebSocket from 'ws'
 import { createEvent, sendEvent, waitForEventCount, waitForNextEvent } from '../helpers'
 import { Event } from '../../../../src/@types/event'
 
-When(/^(\w+) sends a replaceable_event_0 event with content "([^"]+)"$/, async function(
+When(/^(\w+) sends a replaceable_event_0 event with content "([^"]+)"$/, async function (
   name: string,
   content: string,
 ) {
@@ -20,17 +20,17 @@ When(/^(\w+) sends a replaceable_event_0 event with content "([^"]+)"$/, async f
 
 Then(
   /(\w+) receives a replaceable_event_0 event from (\w+) with content "([^"]+?)"/,
-  async function(name: string, author: string, content: string) {
-  const ws = this.parameters.clients[name] as WebSocket
-  const subscription = this.parameters.subscriptions[name][this.parameters.subscriptions[name].length - 1]
-  const receivedEvent = await waitForNextEvent(ws, subscription.name)
+  async function (name: string, author: string, content: string) {
+    const ws = this.parameters.clients[name] as WebSocket
+    const subscription = this.parameters.subscriptions[name][this.parameters.subscriptions[name].length - 1]
+    const receivedEvent = await waitForNextEvent(ws, subscription.name)
 
-  expect(receivedEvent.kind).to.equal(10000)
-  expect(receivedEvent.pubkey).to.equal(this.parameters.identities[author].pubkey)
-  expect(receivedEvent.content).to.equal(content)
-})
+    expect(receivedEvent.kind).to.equal(10000)
+    expect(receivedEvent.pubkey).to.equal(this.parameters.identities[author].pubkey)
+    expect(receivedEvent.content).to.equal(content)
+  })
 
-Then(/(\w+) receives (\d+) replaceable_event_0 events? from (\w+) with content "([^"]+?)" and EOSE/, async function(
+Then(/(\w+) receives (\d+) replaceable_event_0 events? from (\w+) with content "([^"]+?)" and EOSE/, async function (
   name: string,
   count: string,
   author: string,
@@ -46,7 +46,7 @@ Then(/(\w+) receives (\d+) replaceable_event_0 events? from (\w+) with content "
   expect(events[0].content).to.equal(content)
 })
 
-When(/^(\w+) sends a ephemeral_event_0 event with content "([^"]+)"$/, async function(
+When(/^(\w+) sends a ephemeral_event_0 event with content "([^"]+)"$/, async function (
   name: string,
   content: string,
 ) {
@@ -61,23 +61,66 @@ When(/^(\w+) sends a ephemeral_event_0 event with content "([^"]+)"$/, async fun
 
 Then(
   /(\w+) receives a ephemeral_event_0 event from (\w+) with content "([^"]+?)"/,
-  async function(name: string, author: string, content: string) {
+  async function (name: string, author: string, content: string) {
+    const ws = this.parameters.clients[name] as WebSocket
+    const subscription = this.parameters.subscriptions[name][this.parameters.subscriptions[name].length - 1]
+    const receivedEvent = await waitForNextEvent(ws, subscription.name)
+
+    expect(receivedEvent.kind).to.equal(20000)
+    expect(receivedEvent.pubkey).to.equal(this.parameters.identities[author].pubkey)
+    expect(receivedEvent.content).to.equal(content)
+  })
+
+Then(/(\w+) receives (\d+) ephemeral_event_0 events? and EOSE/, async function (
+  name: string,
+  count: string,
+) {
   const ws = this.parameters.clients[name] as WebSocket
   const subscription = this.parameters.subscriptions[name][this.parameters.subscriptions[name].length - 1]
-  const receivedEvent = await waitForNextEvent(ws, subscription.name)
+  const events = await waitForEventCount(ws, subscription.name, Number(count), true)
 
-  expect(receivedEvent.kind).to.equal(20000)
-  expect(receivedEvent.pubkey).to.equal(this.parameters.identities[author].pubkey)
-  expect(receivedEvent.content).to.equal(content)
+  expect(events.length).to.equal(Number(count))
 })
 
-Then(/(\w+) receives (\d+) ephemeral_event_0 events? and EOSE/, async function(
+When(/^(\w+) sends two identically-timestamped replaceable_event_0 events where the second has a lower ID$/, async function (
+  name: string
+) {
+  const ws = this.parameters.clients[name] as WebSocket
+  const { pubkey, privkey } = this.parameters.identities[name]
+
+  const commonTimestamp = Math.floor(Date.now() / 1000)
+
+  const event1 = await createEvent({ pubkey, kind: 10000, content: 'first content', created_at: commonTimestamp }, privkey)
+
+  let nonce = 0
+  let event2: Event
+  for (; ;) {
+    event2 = await createEvent({ pubkey, kind: 10000, content: `second content ${nonce++}`, created_at: commonTimestamp }, privkey)
+
+    if (event2.id < event1.id) {
+      break
+    }
+  }
+
+  await sendEvent(ws, event1)
+  await sendEvent(ws, event2)
+
+  this.parameters.events[name].push(event1, event2)
+  this.parameters.lowerIdEventContent = event2.content
+})
+
+Then(/(\w+) receives (\d+) replaceable_event_0 event from (\w+) matching the lower ID event and EOSE/, async function (
   name: string,
   count: string,
+  author: string,
 ) {
   const ws = this.parameters.clients[name] as WebSocket
   const subscription = this.parameters.subscriptions[name][this.parameters.subscriptions[name].length - 1]
   const events = await waitForEventCount(ws, subscription.name, Number(count), true)
 
   expect(events.length).to.equal(Number(count))
+  expect(events[0].kind).to.equal(10000)
+  expect(events[0].pubkey).to.equal(this.parameters.identities[author].pubkey)
+  expect(events[0].content).to.equal(this.parameters.lowerIdEventContent)
 })
+

test/unit/handlers/event-message-handler.spec.ts

@@ -10,6 +10,7 @@ chai.use(chaiAsPromised)
 
 import { EventLimits, Settings } from '../../../src/@types/settings'
 import { IncomingEventMessage, MessageType } from '../../../src/@types/messages'
+import { CacheAdmissionState } from '../../../src/constants/caching'
 import { Event } from '../../../src/@types/event'
 import { EventKinds } from '../../../src/constants/base'
 import { EventMessageHandler } from '../../../src/handlers/event-message-handler'
@@ -88,7 +89,8 @@ describe('EventMessageHandler', () => {
         () => ({
           info: { relay_url: 'relay_url' },
         }) as any,
-        () => ({ hit: async () => false })
+        () => ({ hit: async () => false }),
+        { hasKey: async () => false, setKey: async () => true } as any,
       )
     })
 
@@ -262,7 +264,8 @@ describe('EventMessageHandler', () => {
         { hasActiveRequestToVanish: async () => false } as any,
         userRepository,
         () => settings,
-        () => ({ hit: async () => false })
+        () => ({ hit: async () => false }),
+        { hasKey: async () => false, setKey: async () => true } as any,
       )
     })
 
@@ -738,7 +741,8 @@ describe('EventMessageHandler', () => {
         { hasActiveRequestToVanish: async () => false } as any,
         userRepository,
         () => settings,
-        () => ({ hit: rateLimiterHitStub })
+        () => ({ hit: rateLimiterHitStub }),
+        { hasKey: async () => false, setKey: async () => true, setKeyWithExpiry: async () => true } as any,
       )
     })
 
@@ -953,6 +957,7 @@ describe('EventMessageHandler', () => {
     let webSocket: IWebSocketAdapter
     let getRelayPublicKeyStub: SinonStub
     let userRepositoryFindByPubkeyStub: SinonStub
+    let cacheGetKeyStub: SinonStub
 
     beforeEach(() => {
       settings = {
@@ -994,6 +999,7 @@ describe('EventMessageHandler', () => {
       getRelayPublicKeyStub = sandbox.stub(EventMessageHandler.prototype, 'getRelayPublicKey' as any)
       getClientAddressStub = sandbox.stub()
       userRepositoryFindByPubkeyStub = sandbox.stub()
+      cacheGetKeyStub = sandbox.stub().resolves(null)
       webSocket = {
         getClientAddress: getClientAddressStub,
       } as any
@@ -1006,11 +1012,18 @@ describe('EventMessageHandler', () => {
         { hasActiveRequestToVanish: async () => false } as any,
         userRepository,
         () => settings,
-        () => ({ hit: async () => false })
+        () => ({ hit: async () => false }),
+        {
+          hasKey: async () => false,
+          getKey: cacheGetKeyStub,
+          setKey: async () => true,
+          setKeyWithExpiry: async () => true,
+          setKeyExpiry: async () => undefined,
+        } as any,
       )
     })
 
-    it ('fulfills with undefined if payments are disabled', async () => {
+    it('fulfills with undefined if payments are disabled', async () => {
       settings.payments.enabled = false
 
       return expect((handler as any).isUserAdmitted(event)).to.eventually.be.undefined
@@ -1089,12 +1102,6 @@ describe('EventMessageHandler', () => {
       return expect((handler as any).isUserAdmitted(event)).to.eventually.equal('blocked: pubkey not admitted')
     })
 
-    it('fulfills with reason if user is not admitted', async () => {
-      userRepositoryFindByPubkeyStub.resolves({ isAdmitted: false })
-
-      return expect((handler as any).isUserAdmitted(event)).to.eventually.equal('blocked: pubkey not admitted')
-    })
-
     it('fulfills with reason if user does not meet minimum balance', async () => {
       settings.limits.event.pubkey.minBalance = 1000n
       userRepositoryFindByPubkeyStub.resolves({ isAdmitted: true, balance: 999n })
@@ -1108,5 +1115,23 @@ describe('EventMessageHandler', () => {
 
       return expect((handler as any).isUserAdmitted(event)).to.eventually.be.undefined
     })
+
+    it('fulfills with undefined if user is admitted in cache', async () => {
+      cacheGetKeyStub.resolves(CacheAdmissionState.ADMITTED)
+
+      return expect((handler as any).isUserAdmitted(event)).to.eventually.be.undefined
+    })
+
+    it('fulfills with reason if user is blocked in cache', async () => {
+      cacheGetKeyStub.resolves(CacheAdmissionState.NOT_ADMITTED)
+
+      return expect((handler as any).isUserAdmitted(event)).to.eventually.equal('blocked: pubkey not admitted')
+    })
+
+    it('fulfills with reason if user has insufficient balance in cache', async () => {
+      cacheGetKeyStub.resolves(CacheAdmissionState.INSUFFICIENT_BALANCE)
+
+      return expect((handler as any).isUserAdmitted(event)).to.eventually.equal('blocked: insufficient balance')
+    })
   })
 })