fix(handlers): abort in-flight subscription stream on client disconnect

.changeset/fix-subscribe-abort-controller.md

@@ -0,0 +1,5 @@
+---
+"nostream": patch
+---
+
+fix: abort in-flight streaming queries when a subscription is cancelled

src/handlers/subscribe-message-handler.ts

@@ -1,5 +1,5 @@
 import { anyPass, equals, isNil, map, omit, propSatisfies, uniqWith } from 'ramda'
-// import { addAbortSignal } from 'stream'
+import { addAbortSignal } from 'stream'
 import { pipeline } from 'stream/promises'
 
 import {
@@ -24,18 +24,18 @@ import { WebSocketAdapterEvent } from '../constants/adapter'
 const logger = createLogger('subscribe-message-handler')
 
 export class SubscribeMessageHandler implements IMessageHandler, IAbortable {
-  //private readonly abortController: AbortController
+  private readonly abortController: AbortController
 
   public constructor(
     private readonly webSocket: IWebSocketAdapter,
     private readonly eventRepository: IEventRepository,
     private readonly settings: () => Settings,
   ) {
-    //this.abortController = new AbortController()
+    this.abortController = new AbortController()
   }
 
   public abort(): void {
-    //this.abortController.abort()
+    this.abortController.abort()
   }
 
   public async handleMessage(message: SubscribeMessage): Promise<void> {
@@ -90,11 +90,11 @@ export class SubscribeMessageHandler implements IMessageHandler, IAbortable {
 
     const findEvents = this.eventRepository.findByFilters(filters).stream()
 
-    // const abortableFindEvents = addAbortSignal(this.abortController.signal, findEvents)
+    const abortableFindEvents = addAbortSignal(this.abortController.signal, findEvents)
 
     try {
       await pipeline(
-        findEvents,
+        abortableFindEvents,
         streamFilter(propSatisfies(isNil, 'deleted_at')),
         streamMap(toNostrEvent),
         streamFilter(isTagUnexpired),

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

@@ -357,6 +357,19 @@ describe('SubscribeMessageHandler', () => {
       await expect(promise).to.eventually.be.rejectedWith(error)
       expect(destroySpy).to.have.been.called
     })
+
+    it('aborts and destroys the event stream when abort() is called', async () => {
+      isClientSubscribedToEventStub.returns(always(true))
+
+      const destroySpy = sandbox.spy(stream, 'destroy')
+
+      const promise = (handler as any).fetchAndSend(subscriptionId, filters)
+
+      handler.abort()
+
+      await expect(promise).to.eventually.be.rejected
+      expect(destroySpy).to.have.been.called
+    })
   })
 
   describe('.isClientSubscribedToEvent', () => {