feat: implement caching for user admission check

Author: Mohit-Davar

package-lock.json

@@ -95,7 +95,7 @@
     "@types/express": "4.17.21",
     "@types/js-yaml": "4.0.5",
     "@types/mocha": "^9.1.1",
-    "@types/node": "^24.0.0",
+    "@types/node": "^24.12.2",
     "@types/pg": "^8.6.5",
     "@types/ramda": "^0.28.13",
     "@types/sinon": "^10.0.11",

package.json

@@ -21,7 +21,7 @@ export type IWebSocketAdapter = EventEmitter & {
 export interface ICacheAdapter {
   getKey(key: string): Promise<string>
   hasKey(key: string): Promise<boolean>
-  setKey(key: string, value: string): Promise<boolean>
+  setKey(key: string, value: string, expirySeconds?: 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/@types/adapters.ts

@@ -42,7 +42,7 @@ export class RedisAdapter implements ICacheAdapter {
   }
 
   private onClientError(error: Error) {
-    console.error('Unable to connect to Redis.', error)
+    debug('Unable to connect to Redis.', error)
     // throw error
   }
 
@@ -58,9 +58,12 @@ export class RedisAdapter implements ICacheAdapter {
     return this.client.get(key)
   }
 
-  public async setKey(key: string, value: string): Promise<boolean> {
+  public async setKey(key: string, value: string, expirySeconds?: number): Promise<boolean> {
     await this.connection
-    debug('get %s key', key)
+    debug('set %s key', key)
+    if (typeof expirySeconds === 'number') {
+      return 'OK' === await this.client.set(key, value, { EX: expirySeconds })
+    }
     return 'OK' === await this.client.set(key, value)
   }
 

src/adapters/redis-adapter.ts

@@ -0,0 +1,5 @@
+export enum CacheAdmissionState {
+  ADMITTED = 'admitted',
+  BLOCKED_NOT_ADMITTED = 'blocked_not_admitted',
+  BLOCKED_INSUFFICIENT_BALANCE = 'blocked_insufficient_balance',
+}

src/constants/caching.ts

@@ -1,13 +1,23 @@
+import { ICacheAdapter, IWebSocketAdapter } from '../@types/adapters'
 import { IEventRepository, INip05VerificationRepository, 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,
@@ -24,6 +34,7 @@ export const messageHandlerFactory = (
           createSettings,
           slidingWindowRateLimiterFactory,
           nip05VerificationRepository,
+          getCache(),
         )
       }
     case MessageType.REQ:

src/factories/message-handler-factory.ts

@@ -26,9 +26,11 @@ import {
 } from '../utils/event'
 import { IEventRepository, INip05VerificationRepository, 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'
@@ -46,6 +48,7 @@ export class EventMessageHandler implements IMessageHandler {
     private readonly settings: () => Settings,
     private readonly slidingWindowRateLimiter: Factory<IRateLimiter>,
     private readonly nip05VerificationRepository: INip05VerificationRepository,
+    private readonly cache: ICacheAdapter,
   ) {}
 
   public async handleMessage(message: IncomingEventMessage): Promise<void> {
@@ -112,8 +115,7 @@ export class EventMessageHandler implements IMessageHandler {
     try {
       await strategy.execute(event)
       this.processNip05Metadata(event)
-    } catch (error) {
-      console.error('error handling message', message, error)
+    } catch (_error) {
       this.webSocket.emit(WebSocketAdapterEvent.Message, createCommandResult(event.id, false, 'error: unable to process event'))
     }
   }
@@ -341,17 +343,44 @@ 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 cachedValue = await this.cache.getKey(cacheKey)
+      if (cachedValue === CacheAdmissionState.ADMITTED) {
+        debug('cache hit for %s admission: admitted', event.pubkey)
+        return
+      }
+      if (cachedValue === CacheAdmissionState.BLOCKED_NOT_ADMITTED) {
+        debug('cache hit for %s admission: blocked', event.pubkey)
+        return 'blocked: pubkey not admitted'
+      }
+      if (cachedValue === CacheAdmissionState.BLOCKED_INSUFFICIENT_BALANCE) {
+        debug('cache hit for %s admission: insufficient balance', event.pubkey)
+        return 'blocked: insufficient balance'
+      }
+    } 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.BLOCKED_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.BLOCKED_INSUFFICIENT_BALANCE, 60)
       return 'blocked: insufficient balance'
     }
+
+    this.cacheSet(cacheKey, CacheAdmissionState.ADMITTED, 300)
+  }
+
+  private cacheSet(key: string, value: string, ttl: number): void {
+    this.cache.setKey(key, value, ttl)
+      .catch((error) => debug('unable to cache %s: %o', key, error))
   }
 
   protected addExpirationMetadata(event: Event): Event | ExpiringEvent {