fix double handler subscription and pass tests

Author: Bewinxed

src/subscribe/client.ts

@@ -1,13 +1,20 @@
-// src/subscribe/client.ts (modified)
+// src/subscribe/client.ts (fixed with multiple handlers per channel)
 import postgres from 'postgres';
 import { SubscribeOptions } from '../types/core';
 
+type NotificationHandler<T> = (payload: T) => void | Promise<void>;
+
+interface ChannelSubscription {
+  listenRequest: postgres.ListenRequest;
+  handlers: Set<NotificationHandler<any>>;
+}
+
 /**
  * Client for subscribing to PostgreSQL notifications
  */
 class SubscriptionClient<Client> {
   private sql: postgres.Sql;
-  private activeSubscriptions: Map<string, postgres.ListenRequest> = new Map();
+  private activeChannels: Map<string, ChannelSubscription> = new Map();
 
   /**
    * Creates a new SubscriptionClient instance
@@ -37,13 +44,8 @@ class SubscriptionClient<Client> {
       onError = (error) => console.error('Subscription error:', error)
     } = options;
 
-    // Check if already subscribed
-    if (this.activeSubscriptions.has(channel)) {
-      throw new Error(`Already subscribed to channel: ${channel}`);
-    }
-
-    // Create the subscription
-    const listenRequest = this.sql.listen(channel, async (payload) => {
+    // Create a typed handler function
+    const handler: NotificationHandler<T> = async (payload: any) => {
       try {
         // Parse the payload
         const parsedPayload = parser(payload);
@@ -58,31 +60,79 @@ class SubscriptionClient<Client> {
       } catch (error) {
         onError(error as Error, payload);
       }
-    });
+    };
+
+    // Check if channel already has a postgres LISTEN
+    const existingSubscription = this.activeChannels.get(channel);
 
-    // Store the subscription
-    this.activeSubscriptions.set(channel, listenRequest);
+    if (existingSubscription) {
+      // Channel already exists - just add this handler
+      existingSubscription.handlers.add(handler);
+    } else {
+      // New channel - create postgres LISTEN and handler set
+      const listenRequest = this.sql.listen(
+        channel,
+        async (payload: string) => {
+          // Get current handlers for this channel
+          const subscription = this.activeChannels.get(channel);
+          if (subscription) {
+            // Call all handlers for this channel
+            const handlerPromises = Array.from(subscription.handlers).map((h) =>
+              Promise.resolve(h(payload)).catch((error) =>
+                onError(error as Error, payload)
+              )
+            );
+            await Promise.allSettled(handlerPromises);
+          }
+        }
+      );
+
+      // Store the channel subscription
+      this.activeChannels.set(channel, {
+        listenRequest,
+        handlers: new Set([handler])
+      });
+    }
   }
 
   /**
-   * Unsubscribes from notifications on a specific channel
+   * Unsubscribes a specific handler from notifications on a channel
    *
    * @param channel - The notification channel to unsubscribe from
+   * @param handler - The specific handler to remove (optional - removes all if not provided)
    * @returns A promise that resolves when unsubscribed
    */
-  public async unsubscribe(channel: string): Promise<void> {
-    const listenRequest = this.activeSubscriptions.get(channel);
+  public async unsubscribe(
+    channel: string,
+    handler?: NotificationHandler<any>
+  ): Promise<void> {
+    const subscription = this.activeChannels.get(channel);
+
+    if (!subscription) {
+      console.warn(`Not subscribed to channel: ${channel}`);
+      return;
+    }
 
-    if (!listenRequest) {
-      throw new Error(`Not subscribed to channel: ${channel}`);
+    if (handler) {
+      // Remove specific handler
+      subscription.handlers.delete(handler);
+    } else {
+      // Remove all handlers for this channel
+      subscription.handlers.clear();
     }
 
-    // Unlisten
-    const meta = await listenRequest;
-    await meta.unlisten();
+    // If no handlers left, unlisten from postgres
+    if (subscription.handlers.size === 0) {
+      try {
+        const meta = await subscription.listenRequest;
+        await meta.unlisten();
+      } catch (error) {
+        console.warn(`Error unlistening from ${channel}:`, error);
+      }
 
-    // Remove from active subscriptions
-    this.activeSubscriptions.delete(channel);
+      // Remove channel completely
+      this.activeChannels.delete(channel);
+    }
   }
 
   /**
@@ -91,14 +141,29 @@ class SubscriptionClient<Client> {
    * @returns A promise that resolves when all unsubscriptions are complete
    */
   public async unsubscribeAll(): Promise<void> {
-    const unsubscribePromises = Array.from(this.activeSubscriptions.keys()).map(
+    const unsubscribePromises = Array.from(this.activeChannels.keys()).map(
       async (channel) => {
         await this.unsubscribe(channel);
       }
     );
 
     await Promise.all(unsubscribePromises);
   }
+
+  /**
+   * Get debug info about active subscriptions
+   */
+  public getActiveChannels(): string[] {
+    return Array.from(this.activeChannels.keys());
+  }
+
+  /**
+   * Get handler count for a specific channel
+   */
+  public getHandlerCount(channel: string): number {
+    const subscription = this.activeChannels.get(channel);
+    return subscription ? subscription.handlers.size : 0;
+  }
 }
 
 // Export the class

src/trigger/registry.ts

@@ -214,34 +214,45 @@ export class Registry<Client> {
   async startListening(): Promise<void> {
     if (this.isListening) return;
 
+    const subscribedChannels = new Set<string>();
+
     // Listen to model channels
     for (const [, channelName] of this.modelChannels) {
-      await this.subscriptionClient.subscribe(channelName, {
-        onNotification: (payload: any) => {
-          this.handleNotification(channelName, payload);
-        }
-      });
+      if (!subscribedChannels.has(channelName)) {
+        await this.subscriptionClient.subscribe(channelName, {
+          onNotification: (payload: any) => {
+            this.handleNotification(channelName, payload);
+          }
+        });
+        subscribedChannels.add(channelName);
+      }
     }
 
     // Listen to custom channels
     for (const [channelName] of this.customChannels) {
-      await this.subscriptionClient.subscribe(channelName, {
-        onNotification: (payload: any) => {
-          this.handleNotification(channelName, payload);
-        }
-      });
+      if (!subscribedChannels.has(channelName)) {
+        await this.subscriptionClient.subscribe(channelName, {
+          onNotification: (payload: any) => {
+            this.handleNotification(channelName, payload);
+          }
+        });
+        subscribedChannels.add(channelName);
+      }
     }
 
     // Listen to custom trigger channels
     for (const [triggerName] of this.modelTriggers) {
       if (triggerName.includes('_custom_')) {
         const [, , triggerShortName] = triggerName.split('_');
         const channelName = `${triggerShortName}_events`;
-        await this.subscriptionClient.subscribe(channelName, {
-          onNotification: (payload: any) => {
-            this.handleNotification(channelName, payload);
-          }
-        });
+        if (!subscribedChannels.has(channelName)) {
+          await this.subscriptionClient.subscribe(channelName, {
+            onNotification: (payload: any) => {
+              this.handleNotification(channelName, payload);
+            }
+          });
+          subscribedChannels.add(channelName);
+        }
       }
     }