From 406de8992cad492980cf78573abc6b9990c75494 Mon Sep 17 00:00:00 2001
From: Priyanshubhartistm <bhartipriyanshustm@gmail.com>
Date: Tue, 4 Aug 2026 22:47:24 +0530
Subject: [PATCH] fix(handlers): abort in-flight subscription stream on client
 disconnect

Signed-off-by: Priyanshubhartistm <bhartipriyanshustm@gmail.com>
---
 .changeset/fix-subscribe-abort-controller.md        |  5 +++++
 src/handlers/subscribe-message-handler.ts           | 12 ++++++------
 .../unit/handlers/subscribe-message-handler.spec.ts | 13 +++++++++++++
 3 files changed, 24 insertions(+), 6 deletions(-)
 create mode 100644 .changeset/fix-subscribe-abort-controller.md

diff --git a/.changeset/fix-subscribe-abort-controller.md b/.changeset/fix-subscribe-abort-controller.md
new file mode 100644
index 00000000..8f95d9ef
--- /dev/null
+++ b/.changeset/fix-subscribe-abort-controller.md
@@ -0,0 +1,5 @@
+---
+"nostream": patch
+---
+
+fix: abort in-flight streaming queries when a subscription is cancelled
diff --git a/src/handlers/subscribe-message-handler.ts b/src/handlers/subscribe-message-handler.ts
index e9fac71a..3d5afb45 100644
--- a/src/handlers/subscribe-message-handler.ts
+++ b/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 {
@@ -22,18 +22,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> {
@@ -72,11 +72,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),
diff --git a/test/unit/handlers/subscribe-message-handler.spec.ts b/test/unit/handlers/subscribe-message-handler.spec.ts
index 35f1bb2b..b30c4bf8 100644
--- a/test/unit/handlers/subscribe-message-handler.spec.ts
+++ b/test/unit/handlers/subscribe-message-handler.spec.ts
@@ -252,6 +252,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', () => {