test: add unit tests for SubscribeMessageHandler

src/@types/adapters.ts

@@ -12,5 +12,5 @@ export interface IWebServerAdapter extends EventEmitter {
 
 
 export type IWebSocketAdapter = EventEmitter & {
-  getSubscriptions(): Map<string, Set<SubscriptionFilter>>
+  getSubscriptions(): Map<string, SubscriptionFilter[]>
 }

src/@types/messages.ts

@@ -29,6 +29,7 @@ export type SubscribeMessage = {
 } & Array<SubscriptionFilter>
 
 export type IncomingEventMessage = EventMessage & [MessageType.EVENT, Event]
+
 export interface EventMessage {
   0: MessageType.EVENT
   1: Event

src/adapters/web-socket-adapter.ts

@@ -18,7 +18,7 @@ export class WebSocketAdapter extends EventEmitter implements IWebSocketAdapter
   private id: string
   private clientAddress: string
   private alive: boolean
-  private subscriptions: Map<SubscriptionId, Set<SubscriptionFilter>>
+  private subscriptions: Map<SubscriptionId, SubscriptionFilter[]>
 
   private sent = 0
   private received = 0
@@ -54,7 +54,7 @@ export class WebSocketAdapter extends EventEmitter implements IWebSocketAdapter
     this.subscriptions.delete(subscriptionId)
   }
 
-  public onSubscribed(subscriptionId: string, filters: Set<SubscriptionFilter>): void {
+  public onSubscribed(subscriptionId: string, filters: SubscriptionFilter[]): void {
     this.subscriptions.set(subscriptionId, filters)
   }
 
@@ -69,7 +69,7 @@ export class WebSocketAdapter extends EventEmitter implements IWebSocketAdapter
   public onSendEvent(event: Event): void {
     this.subscriptions.forEach((filters, subscriptionId) => {
       if (
-        Array.from(filters).map(isEventMatchingFilter).some((Matches) => Matches(event))
+        filters.map(isEventMatchingFilter).some((Matches) => Matches(event))
       ) {
         this.sendMessage(createOutgoingEventMessage(subscriptionId, event))
       }
@@ -95,7 +95,7 @@ export class WebSocketAdapter extends EventEmitter implements IWebSocketAdapter
     this.client.ping()
   }
 
-  public getSubscriptions(): Map<string, Set<SubscriptionFilter>> {
+  public getSubscriptions(): Map<string, SubscriptionFilter[]> {
     return new Map(this.subscriptions)
   }
 

src/handlers/subscribe-message-handler.ts

@@ -38,47 +38,56 @@ export class SubscribeMessageHandler implements IMessageHandler, IAbortable {
       return
     }
 
-    this.webSocket.emit(WebSocketAdapterEvent.Subscribe, subscriptionId, new Set(filters))
+    this.webSocket.emit(WebSocketAdapterEvent.Subscribe, subscriptionId, filters)
 
+    return this.fetchAndSend(subscriptionId, filters)
+  }
+
+  private async fetchAndSend(subscriptionId: string, filters: SubscriptionFilter[]): Promise<void> {
     const sendEvent = (event: Event) =>
       this.webSocket.emit(WebSocketAdapterEvent.Message, createOutgoingEventMessage(subscriptionId, event))
     const sendEOSE = () =>
       this.webSocket.emit(WebSocketAdapterEvent.Message, createEndOfStoredEventsNoticeMessage(subscriptionId))
+    const isSubscribedToEvent = SubscribeMessageHandler.isClientSubscribedToEvent(filters)
 
     const findEvents = this.eventRepository.findByFilters(filters).stream()
+
     try {
       await pipeline(
         findEvents,
         streamMap(toNostrEvent),
-        streamFilter(anyPass(map(isEventMatchingFilter)(filters))),
+        streamFilter(isSubscribedToEvent),
         streamEach(sendEvent),
-        streamEnd(sendEOSE), // NIP-15: End of Stored Events Notice
+        streamEnd(sendEOSE),
         {
           signal: this.abortController.signal,
-        },
+        }
       )
     } catch (error) {
       if (error instanceof Error && error.name === 'AbortError') {
-        console.log('AbortError when finding events')
         findEvents.end()
       }
       throw error
     }
   }
 
-  private canSubscribe(subscriptionId: string, filters: SubscriptionFilter[]): string | undefined {
+  private static isClientSubscribedToEvent(filters: SubscriptionFilter[]): (event: Event) => boolean {
+    return anyPass(map(isEventMatchingFilter)(filters))
+  }
+
+  private canSubscribe(subscriptionId: SubscriptionId, filters: SubscriptionFilter[]): string | undefined {
     const maxSubscriptions = this.settings().limits.client.subscription.maxSubscriptions
     if (maxSubscriptions > 0) {
       const subscriptions = this.webSocket.getSubscriptions()
       if (!subscriptions.has(subscriptionId) && subscriptions.size + 1 > maxSubscriptions) {
-        return `Too many subscriptions: Number of subscriptions must be less than ${maxSubscriptions}`
+        return `Too many subscriptions: Number of subscriptions must be less than or equal to ${maxSubscriptions}`
       }
     }
 
     const maxFilters = this.settings().limits.client.subscription.maxFilters
     if (maxFilters > 0) {
       if (filters.length > maxFilters) {
-        return `Too many filters: Number of filters per susbscription must be less or equal to ${maxFilters}`
+        return `Too many filters: Number of filters per susbscription must be less then or equal to ${maxFilters}`
       }
     }
   }