fix: added expired_at filter to message pipeline

Author: kushagra0902

src/handlers/subscribe-message-handler.ts

@@ -3,12 +3,12 @@ import { anyPass, equals, isNil, map, propSatisfies, uniqWith } from 'ramda'
 import { pipeline } from 'stream/promises'
 
 import { createEndOfStoredEventsNoticeMessage, createNoticeMessage, createOutgoingEventMessage } from '../utils/messages'
+import { DBEvent, Event } from '../@types/event'
 import { IAbortable, IMessageHandler } from '../@types/message-handlers'
 import { isEventMatchingFilter, toNostrEvent } from '../utils/event'
 import { streamEach, streamEnd, streamFilter, streamMap } from '../utils/stream'
 import { SubscriptionFilter, SubscriptionId } from '../@types/subscription'
 import { createLogger } from '../factories/logger-factory'
-import { Event } from '../@types/event'
 import { IEventRepository } from '../@types/repositories'
 import { IWebSocketAdapter } from '../@types/adapters'
 import { Settings } from '../@types/settings'
@@ -55,6 +55,8 @@ export class SubscribeMessageHandler implements IMessageHandler, IAbortable {
     const sendEOSE = () =>
       this.webSocket.emit(WebSocketAdapterEvent.Message, createEndOfStoredEventsNoticeMessage(subscriptionId))
     const isSubscribedToEvent = SubscribeMessageHandler.isClientSubscribedToEvent(filters)
+    const isNotExpired = (event: DBEvent) =>
+      typeof event.expires_at !== 'number' || event.expires_at > Math.floor(Date.now() / 1000)
 
     const findEvents = this.eventRepository.findByFilters(filters).stream()
 
@@ -64,6 +66,7 @@ export class SubscribeMessageHandler implements IMessageHandler, IAbortable {
       await pipeline(
         findEvents,
         streamFilter(propSatisfies(isNil, 'deleted_at')),
+        streamFilter(isNotExpired),
         streamMap(toNostrEvent),
         streamFilter(isSubscribedToEvent),
         streamEach(sendEvent),

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

@@ -17,14 +17,18 @@ import { WebSocketAdapterEvent } from '../../../src/constants/adapter'
 chai.use(chaiAsPromised)
 const { expect } = chai
 
-const toDbEvent = (event: Event) => ({
+const toDbEvent = (
+  event: Event,
+  metadata: { expires_at?: number, deleted_at?: Date | null } = {},
+) => ({
   event_id: Buffer.from(event.id, 'hex'),
   event_kind: event.kind,
   event_pubkey: Buffer.from(event.pubkey, 'hex'),
   event_created_at: event.created_at,
   event_content: event.content,
   event_tags: event.tags,
   event_signature: Buffer.from(event.sig, 'hex'),
+  ...metadata,
 })
 
 describe('SubscribeMessageHandler', () => {
@@ -165,6 +169,43 @@ describe('SubscribeMessageHandler', () => {
       )
     })
 
+    it('does not send expired events', async () => {
+      isClientSubscribedToEventStub.returns(always(true))
+
+      const now = Math.floor(Date.now() / 1000)
+      const promise = (handler as any).fetchAndSend(subscriptionId, filters)
+
+      stream.write(toDbEvent(event, { expires_at: now - 1 }))
+      stream.end()
+
+      await promise
+
+      expect(eventRepositoryFindByFiltersStub).to.have.been.calledOnceWithExactly(filters)
+      expect(webSocketOnMessageStub).to.have.been.calledOnceWithExactly(
+        ['EOSE', subscriptionId],
+      )
+    })
+
+    it('sends event if expiration is in the future', async () => {
+      isClientSubscribedToEventStub.returns(always(true))
+
+      const now = Math.floor(Date.now() / 1000)
+      const promise = (handler as any).fetchAndSend(subscriptionId, filters)
+
+      stream.write(toDbEvent(event, { expires_at: now + 60 }))
+      stream.end()
+
+      await promise
+
+      expect(eventRepositoryFindByFiltersStub).to.have.been.calledOnceWithExactly(filters)
+      expect(webSocketOnMessageStub).to.have.been.calledWithExactly(
+        ['EVENT', subscriptionId, event],
+      )
+      expect(webSocketOnMessageStub).to.have.been.calledWithExactly(
+        ['EOSE', subscriptionId],
+      )
+    })
+
     it('sends EOSE', async () => {
       const promise = (handler as any).fetchAndSend(subscriptionId, filters)