Merge branch 'main' into fix/csp-ws-http-scheme-mapping

Author: Priyanshubhartistm

.changeset/fix-event-expiration-validation.md

@@ -0,0 +1,5 @@
+---
+"nostream": patch
+---
+
+fix: reject expiration timestamp 0 and millisecond-scale values, and accept safe-integer second-based timestamps up to the Postgres int4 max (2038-01-19T03:14:07Z) in getEventExpiration()

.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),

src/utils/event.ts

@@ -245,6 +245,11 @@ export const isExpiredEvent = (event: Event): boolean => {
   return expirationTime <= now
 }
 
+// Postgres int4 max (2038-01-19T03:14:07Z). events.expires_at is a signed 32-bit
+// integer column — Knex's .unsigned() is a no-op on Postgres — so anything beyond
+// this would fail the INSERT and get silently dropped instead of stored.
+const MAX_EXPIRATION_TIME = 2147483647
+
 export const getEventExpiration = (event: Event): number | undefined => {
   const [, rawExpirationTime] = event.tags.find((tag) => tag.length >= 2 && tag[0] === EventTags.Expiration) ?? []
   if (!rawExpirationTime) {
@@ -253,7 +258,7 @@ export const getEventExpiration = (event: Event): number | undefined => {
 
   const expirationTime = Number(rawExpirationTime)
 
-  if (Number.isSafeInteger(expirationTime) && Math.log10(expirationTime) < 10) {
+  if (Number.isSafeInteger(expirationTime) && expirationTime > 0 && expirationTime <= MAX_EXPIRATION_TIME) {
     return expirationTime
   }
 }

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', () => {

test/unit/utils/event.spec.ts

@@ -691,6 +691,31 @@ describe('NIP-40', () => {
       event.tags = [['expiration', 'a']]
       expect(getEventExpiration(event)).to.be.undefined
     })
+
+    it('returns false if expiration is 0', () => {
+      event.tags = [['expiration', '0']]
+      expect(getEventExpiration(event)).to.be.undefined
+    })
+
+    it('returns false if expiration exceeds the Postgres int4 column max', () => {
+      event.tags = [['expiration', '10000000000']]
+      expect(getEventExpiration(event)).to.be.undefined
+    })
+
+    it('returns false if expiration is the maximum representable Unix seconds timestamp', () => {
+      event.tags = [['expiration', '253402300799']]
+      expect(getEventExpiration(event)).to.be.undefined
+    })
+
+    it('returns true if expiration is the maximum value the expires_at column can store', () => {
+      event.tags = [['expiration', '2147483647']]
+      expect(getEventExpiration(event)).to.equal(2147483647)
+    })
+
+    it('returns false if expiration looks like a millisecond-scale timestamp', () => {
+      event.tags = [['expiration', '1700000000000']]
+      expect(getEventExpiration(event)).to.be.undefined
+    })
   })
 
   describe('isExpiredEvent', () => {