diff --git a/src/@types/adapters.ts b/src/@types/adapters.ts index 08556a9..5520341 100644 --- a/src/@types/adapters.ts +++ b/src/@types/adapters.ts @@ -1,4 +1,5 @@ import { EventEmitter } from 'node:stream' +import { SubscriptionFilter } from './subscription' export interface IWebSocketServerAdapter extends EventEmitter { getConnectedClients(): number @@ -10,4 +11,6 @@ export interface IWebServerAdapter extends EventEmitter { } -export type IWebSocketAdapter = EventEmitter +export type IWebSocketAdapter = EventEmitter & { + getSubscriptions(): Map> +} diff --git a/src/@types/messages.ts b/src/@types/messages.ts index 85202c8..94a9701 100644 --- a/src/@types/messages.ts +++ b/src/@types/messages.ts @@ -14,11 +14,12 @@ export type IncomingMessage = | SubscribeMessage | IncomingEventMessage | UnsubscribeMessage - | Notice + export type OutgoingMessage = | OutgoingEventMessage | EndOfStoredEventsNotice + | NoticeMessage export type SubscribeMessage = { [index in Range<2, 100>]: SubscriptionFilter @@ -44,7 +45,7 @@ export interface UnsubscribeMessage { 1: SubscriptionId } -export interface Notice { +export interface NoticeMessage { 0: MessageType.NOTICE 1: string } diff --git a/src/handlers/subscribe-message-handler.ts b/src/handlers/subscribe-message-handler.ts index ed71978..9ec5965 100644 --- a/src/handlers/subscribe-message-handler.ts +++ b/src/handlers/subscribe-message-handler.ts @@ -1,7 +1,7 @@ -import { anyPass, map } from 'ramda' +import { anyPass, equals, map, uniqWith } from 'ramda' import { pipeline } from 'stream/promises' -import { createEndOfStoredEventsNoticeMessage, createOutgoingEventMessage } from '../utils/messages' +import { createEndOfStoredEventsNoticeMessage, createNoticeMessage, createOutgoingEventMessage } from '../utils/messages' import { IAbortable, IMessageHandler } from '../@types/message-handlers' import { isEventMatchingFilter, toNostrEvent } from '../utils/event' import { streamEach, streamEnd, streamFilter, streamMap } from '../utils/stream' @@ -9,10 +9,10 @@ import { SubscriptionFilter, SubscriptionId } from '../@types/subscription' import { Event } from '../@types/event' import { IEventRepository } from '../@types/repositories' import { IWebSocketAdapter } from '../@types/adapters' +import { Settings } from '../utils/settings' import { SubscribeMessage } from '../@types/messages' import { WebSocketAdapterEvent } from '../constants/adapter' - export class SubscribeMessageHandler implements IMessageHandler, IAbortable { private readonly abortController: AbortController @@ -29,7 +29,13 @@ export class SubscribeMessageHandler implements IMessageHandler, IAbortable { public async handleMessage(message: SubscribeMessage): Promise { const subscriptionId = message[1] as SubscriptionId - const filters = message.slice(2) as SubscriptionFilter[] + const filters = uniqWith(equals, message.slice(2)) as SubscriptionFilter[] + + const reason = this.canSubscribe(subscriptionId, filters) + if (reason) { + this.webSocket.emit(WebSocketAdapterEvent.Message, createNoticeMessage(`Subscription request rejected: ${reason}`)) + return + } this.webSocket.emit(WebSocketAdapterEvent.Subscribe, subscriptionId, new Set(filters)) @@ -59,4 +65,20 @@ export class SubscribeMessageHandler implements IMessageHandler, IAbortable { } } + private canSubscribe(subscriptionId: string, filters: SubscriptionFilter[]): string | undefined { + const maxSubscriptions = Settings.limits.client.subscription.maxSubscriptions + if (maxSubscriptions > 0) { + const subscriptions = this.webSocket.getSubscriptions() + if (!subscriptions.has(subscriptionId) && subscriptions.size + 1 > maxSubscriptions) { + return `Too many subscriptions: Number of subscriptions must be less than ${maxSubscriptions}` + } + } + + const maxFilters = Settings.limits.client.subscription.maxFilters + if (maxFilters > 0) { + if (filters.length > maxFilters) { + return `Too many filters: Number of filters per susbscription must be less or equal to ${maxFilters}` + } + } + } } diff --git a/src/utils/messages.ts b/src/utils/messages.ts index 3d5db6e..536028a 100644 --- a/src/utils/messages.ts +++ b/src/utils/messages.ts @@ -1,13 +1,13 @@ import { EndOfStoredEventsNotice, MessageType, - Notice, + NoticeMessage, OutgoingMessage, } from '../@types/messages' import { Event } from '../@types/event' import { SubscriptionId } from '../@types/subscription' -export const createNotice = (notice: string): Notice => { +export const createNoticeMessage = (notice: string): NoticeMessage => { return [MessageType.NOTICE, notice] } diff --git a/test/unit/utils/messages.spec.ts b/test/unit/utils/messages.spec.ts index 015ffbf..461a991 100644 --- a/test/unit/utils/messages.spec.ts +++ b/test/unit/utils/messages.spec.ts @@ -1,12 +1,12 @@ import { expect } from 'chai' -import { createEndOfStoredEventsNoticeMessage, createNotice, createOutgoingEventMessage } from '../../../src/utils/messages' +import { createEndOfStoredEventsNoticeMessage, createNoticeMessage, createOutgoingEventMessage } from '../../../src/utils/messages' import { Event } from '../../../src/@types/event' import { MessageType } from '../../../src/@types/messages' describe('createNotice', () => { it('returns a notice message', () => { - expect(createNotice('some notice')).to.deep.equal([MessageType.NOTICE, 'some notice']) + expect(createNoticeMessage('some notice')).to.deep.equal([MessageType.NOTICE, 'some notice']) }) })