From ec87b29286211362075a08f5db299af629b1bcb1 Mon Sep 17 00:00:00 2001 From: Thomas Trompette Date: Wed, 14 Jan 2026 13:29:12 +0100 Subject: [PATCH] Move query matching before publication (#17121) - new channel EVENT_STREAM_CHANNEL based on event stream id - on event, perform the matching and publish only to the right streams - store a list of active streams per workspace - store the user id along with the queries for each stream Bonus: - remove onSubscriptionMatch --- .../src/generated-metadata/graphql.ts | 23 --- .../twenty-front/src/generated/graphql.ts | 23 --- .../auth/types/auth-context.type.ts | 8 + .../services/cache-storage.service.ts | 34 ++++ .../constants/event-stream-ttl.constant.ts | 1 + .../dtos/subscription-matches.dto.ts | 18 -- .../subscriptions/dtos/subscription.input.ts | 15 -- .../enums/subscription-channel.enum.ts | 2 +- .../subscriptions/event-stream.service.ts | 191 ++++++++++++++---- .../subscriptions/subscription.service.ts | 122 ++++------- .../subscriptions/subscriptions.module.ts | 5 +- .../types/event-stream-data.type.ts | 10 + .../object-record-subscription-event.type.ts | 5 + .../get-channel-id-from-event-stream-id.ts | 7 + .../utils/wrap-async-iterator-with-cleanup.ts | 32 +++ .../workspace-event-emitter.module.ts | 4 +- .../workspace-event-emitter.resolver.ts | 177 ++++++---------- .../workspace-event-emitter.service.ts | 98 ++++++++- 18 files changed, 441 insertions(+), 334 deletions(-) create mode 100644 packages/twenty-server/src/engine/subscriptions/constants/event-stream-ttl.constant.ts delete mode 100644 packages/twenty-server/src/engine/subscriptions/dtos/subscription-matches.dto.ts delete mode 100644 packages/twenty-server/src/engine/subscriptions/dtos/subscription.input.ts create mode 100644 packages/twenty-server/src/engine/subscriptions/types/event-stream-data.type.ts create mode 100644 packages/twenty-server/src/engine/subscriptions/types/object-record-subscription-event.type.ts create mode 100644 packages/twenty-server/src/engine/workspace-event-emitter/utils/get-channel-id-from-event-stream-id.ts create mode 100644 packages/twenty-server/src/engine/workspace-event-emitter/utils/wrap-async-iterator-with-cleanup.ts diff --git a/packages/twenty-front/src/generated-metadata/graphql.ts b/packages/twenty-front/src/generated-metadata/graphql.ts index 9ef32e8fda..154b2be634 100644 --- a/packages/twenty-front/src/generated-metadata/graphql.ts +++ b/packages/twenty-front/src/generated-metadata/graphql.ts @@ -4219,7 +4219,6 @@ export type Subscription = { __typename?: 'Subscription'; onDbEvent: OnDbEvent; onEventSubscription?: Maybe; - onSubscriptionMatch?: Maybe; serverlessFunctionLogs: ServerlessFunctionLogs; }; @@ -4234,37 +4233,15 @@ export type SubscriptionOnEventSubscriptionArgs = { }; -export type SubscriptionOnSubscriptionMatchArgs = { - subscriptions: Array; -}; - - export type SubscriptionServerlessFunctionLogsArgs = { input: ServerlessFunctionLogsInput; }; -export type SubscriptionInput = { - id: Scalars['String']; - query: Scalars['String']; - selectedEventActions?: InputMaybe>; -}; - export enum SubscriptionInterval { Month = 'Month', Year = 'Year' } -export type SubscriptionMatch = { - __typename?: 'SubscriptionMatch'; - event: OnDbEvent; - subscriptionIds: Array; -}; - -export type SubscriptionMatches = { - __typename?: 'SubscriptionMatches'; - matches: Array; -}; - export enum SubscriptionStatus { Active = 'Active', Canceled = 'Canceled', diff --git a/packages/twenty-front/src/generated/graphql.ts b/packages/twenty-front/src/generated/graphql.ts index 9c4317463e..d0b8a03d04 100644 --- a/packages/twenty-front/src/generated/graphql.ts +++ b/packages/twenty-front/src/generated/graphql.ts @@ -4094,7 +4094,6 @@ export type Subscription = { __typename?: 'Subscription'; onDbEvent: OnDbEvent; onEventSubscription?: Maybe; - onSubscriptionMatch?: Maybe; serverlessFunctionLogs: ServerlessFunctionLogs; }; @@ -4109,37 +4108,15 @@ export type SubscriptionOnEventSubscriptionArgs = { }; -export type SubscriptionOnSubscriptionMatchArgs = { - subscriptions: Array; -}; - - export type SubscriptionServerlessFunctionLogsArgs = { input: ServerlessFunctionLogsInput; }; -export type SubscriptionInput = { - id: Scalars['String']; - query: Scalars['String']; - selectedEventActions?: InputMaybe>; -}; - export enum SubscriptionInterval { Month = 'Month', Year = 'Year' } -export type SubscriptionMatch = { - __typename?: 'SubscriptionMatch'; - event: OnDbEvent; - subscriptionIds: Array; -}; - -export type SubscriptionMatches = { - __typename?: 'SubscriptionMatches'; - matches: Array; -}; - export enum SubscriptionStatus { Active = 'Active', Canceled = 'Canceled', diff --git a/packages/twenty-server/src/engine/core-modules/auth/types/auth-context.type.ts b/packages/twenty-server/src/engine/core-modules/auth/types/auth-context.type.ts index c9690db876..e03cb008e7 100644 --- a/packages/twenty-server/src/engine/core-modules/auth/types/auth-context.type.ts +++ b/packages/twenty-server/src/engine/core-modules/auth/types/auth-context.type.ts @@ -22,6 +22,14 @@ export type AuthContext = { }; }; +export type SerializableAuthContext = { + userId?: string; + userWorkspaceId?: string; + workspaceMemberId?: string; + apiKeyId?: string; + applicationId?: string; +}; + export enum JwtTokenTypeEnum { ACCESS = 'ACCESS', REFRESH = 'REFRESH', diff --git a/packages/twenty-server/src/engine/core-modules/cache-storage/services/cache-storage.service.ts b/packages/twenty-server/src/engine/core-modules/cache-storage/services/cache-storage.service.ts index 65d96afbee..522b9b476d 100644 --- a/packages/twenty-server/src/engine/core-modules/cache-storage/services/cache-storage.service.ts +++ b/packages/twenty-server/src/engine/core-modules/cache-storage/services/cache-storage.service.ts @@ -106,6 +106,32 @@ export class CacheStorageService { }); } + async setRemove(key: string, values: string[]): Promise { + if (values.length === 0) { + return 0; + } + + if (this.isRedisCache()) { + return (this.cache as RedisCache).store.client.sRem( + this.getKey(key), + values, + ); + } + + const existing = await this.get(key); + + if (!existing) { + return 0; + } + + const filtered = existing.filter((v) => !values.includes(v)); + const removed = existing.length - filtered.length; + + await this.set(key, filtered); + + return removed; + } + async countAllSetMembers(cacheKeys: string[]) { return ( await Promise.all(cacheKeys.map((key) => this.getSetLength(key) || 0)) @@ -143,6 +169,14 @@ export class CacheStorageService { }); } + async setMembers(key: string): Promise { + if (this.isRedisCache()) { + return (this.cache as RedisCache).store.client.sMembers(this.getKey(key)); + } + + return (await this.get(key)) ?? []; + } + async flush() { return this.cache.reset(); } diff --git a/packages/twenty-server/src/engine/subscriptions/constants/event-stream-ttl.constant.ts b/packages/twenty-server/src/engine/subscriptions/constants/event-stream-ttl.constant.ts new file mode 100644 index 0000000000..2bf333ef60 --- /dev/null +++ b/packages/twenty-server/src/engine/subscriptions/constants/event-stream-ttl.constant.ts @@ -0,0 +1 @@ +export const EVENT_STREAM_TTL_MS = 30 * 60 * 1_000; // 30 minutes diff --git a/packages/twenty-server/src/engine/subscriptions/dtos/subscription-matches.dto.ts b/packages/twenty-server/src/engine/subscriptions/dtos/subscription-matches.dto.ts deleted file mode 100644 index b21a68e657..0000000000 --- a/packages/twenty-server/src/engine/subscriptions/dtos/subscription-matches.dto.ts +++ /dev/null @@ -1,18 +0,0 @@ -import { Field, ObjectType } from '@nestjs/graphql'; - -import { OnDbEventDTO } from './on-db-event.dto'; - -@ObjectType('SubscriptionMatch') -export class SubscriptionMatchDTO { - @Field(() => [String]) - subscriptionIds: string[]; - - @Field(() => OnDbEventDTO) - event: OnDbEventDTO; -} - -@ObjectType('SubscriptionMatches') -export class SubscriptionMatchesDTO { - @Field(() => [SubscriptionMatchDTO]) - matches: SubscriptionMatchDTO[]; -} diff --git a/packages/twenty-server/src/engine/subscriptions/dtos/subscription.input.ts b/packages/twenty-server/src/engine/subscriptions/dtos/subscription.input.ts deleted file mode 100644 index f0acbf77a8..0000000000 --- a/packages/twenty-server/src/engine/subscriptions/dtos/subscription.input.ts +++ /dev/null @@ -1,15 +0,0 @@ -import { Field, InputType } from '@nestjs/graphql'; - -import { DatabaseEventAction } from 'src/engine/api/graphql/graphql-query-runner/enums/database-event-action'; - -@InputType() -export class SubscriptionInput { - @Field() - id: string; - - @Field() - query: string; - - @Field(() => [DatabaseEventAction], { nullable: true }) - selectedEventActions?: DatabaseEventAction[]; -} diff --git a/packages/twenty-server/src/engine/subscriptions/enums/subscription-channel.enum.ts b/packages/twenty-server/src/engine/subscriptions/enums/subscription-channel.enum.ts index 6c5da5a697..07b9844914 100644 --- a/packages/twenty-server/src/engine/subscriptions/enums/subscription-channel.enum.ts +++ b/packages/twenty-server/src/engine/subscriptions/enums/subscription-channel.enum.ts @@ -1,5 +1,5 @@ export enum SubscriptionChannel { - WORKSPACE_EVENT_BATCH_CHANNEL = 'WORKSPACE_EVENT_BATCH_CHANNEL', DATABASE_EVENT_CHANNEL = 'DATABASE_EVENT_CHANNEL', SERVERLESS_FUNCTION_LOGS_CHANNEL = 'SERVERLESS_FUNCTION_LOGS_CHANNEL', + EVENT_STREAM_CHANNEL = 'EVENT_STREAM_CHANNEL', } diff --git a/packages/twenty-server/src/engine/subscriptions/event-stream.service.ts b/packages/twenty-server/src/engine/subscriptions/event-stream.service.ts index 538af3264e..cd4337fac6 100644 --- a/packages/twenty-server/src/engine/subscriptions/event-stream.service.ts +++ b/packages/twenty-server/src/engine/subscriptions/event-stream.service.ts @@ -1,73 +1,171 @@ import { Injectable } from '@nestjs/common'; -import { type ObjectRecordEvent } from 'twenty-shared/database-events'; import { type RecordGqlOperationSignature } from 'twenty-shared/types'; +import { isDefined } from 'twenty-shared/utils'; +import { type SerializableAuthContext } from 'src/engine/core-modules/auth/types/auth-context.type'; +import { CacheLockService } from 'src/engine/core-modules/cache-lock/cache-lock.service'; import { WithLock } from 'src/engine/core-modules/cache-lock/with-lock.decorator'; import { InjectCacheStorage } from 'src/engine/core-modules/cache-storage/decorators/cache-storage.decorator'; import { CacheStorageService } from 'src/engine/core-modules/cache-storage/services/cache-storage.service'; import { CacheStorageNamespace } from 'src/engine/core-modules/cache-storage/types/cache-storage-namespace.enum'; - -export type ObjectRecordSubscriptionEvent = ObjectRecordEvent & { - objectNameSingular: string; -}; +import { EVENT_STREAM_TTL_MS } from 'src/engine/subscriptions/constants/event-stream-ttl.constant'; +import { type EventStreamData } from 'src/engine/subscriptions/types/event-stream-data.type'; +import { type ObjectRecordSubscriptionEvent } from 'src/engine/subscriptions/types/object-record-subscription-event.type'; @Injectable() export class EventStreamService { constructor( @InjectCacheStorage(CacheStorageNamespace.EngineSubscriptions) private readonly cacheStorageService: CacheStorageService, + private readonly cacheLockService: CacheLockService, ) {} - private getEventStreamKey( - workspaceId: string, - eventStreamId: string, - ): string { - return `eventStream:${workspaceId}:${eventStreamId}`; + async createEventStream({ + workspaceId, + eventStreamChannelId, + authContext, + }: { + workspaceId: string; + eventStreamChannelId: string; + authContext: SerializableAuthContext; + }): Promise { + const key = this.getEventStreamKey(workspaceId, eventStreamChannelId); + const streamData: EventStreamData = { + authContext, + workspaceId, + queries: {}, + createdAt: Date.now(), + }; + + await this.cacheStorageService.set(key, streamData, EVENT_STREAM_TTL_MS); + + const activeStreamsKey = this.getActiveStreamsKey(workspaceId); + + await this.cacheLockService.withLock(async () => { + await this.cacheStorageService.setAdd( + activeStreamsKey, + [eventStreamChannelId], + EVENT_STREAM_TTL_MS, + ); + }, activeStreamsKey); } - @WithLock('eventStreamId') + async destroyEventStream({ + workspaceId, + eventStreamChannelId, + }: { + workspaceId: string; + eventStreamChannelId: string; + }): Promise { + const key = this.getEventStreamKey(workspaceId, eventStreamChannelId); + + await this.cacheStorageService.del(key); + + const activeStreamsKey = this.getActiveStreamsKey(workspaceId); + + await this.cacheLockService.withLock(async () => { + await this.cacheStorageService.setRemove(activeStreamsKey, [ + eventStreamChannelId, + ]); + }, activeStreamsKey); + } + + async getActiveStreamIds(workspaceId: string): Promise { + return this.cacheStorageService.setMembers( + this.getActiveStreamsKey(workspaceId), + ); + } + + async removeFromActiveStreams( + workspaceId: string, + streamIdsToRemove: string[], + ): Promise { + if (streamIdsToRemove.length === 0) { + return; + } + + const activeStreamsKey = this.getActiveStreamsKey(workspaceId); + + await this.cacheLockService.withLock(async () => { + await this.cacheStorageService.setRemove( + activeStreamsKey, + streamIdsToRemove, + ); + }, activeStreamsKey); + } + + async getStreamsData( + workspaceId: string, + streamChannelIds: string[], + ): Promise> { + if (streamChannelIds.length === 0) { + return new Map(); + } + + const keys = streamChannelIds.map((id) => + this.getEventStreamKey(workspaceId, id), + ); + const values = await this.cacheStorageService.mget(keys); + + const result = new Map(); + + streamChannelIds.forEach((id, index) => { + result.set(id, values[index]); + }); + + return result; + } + + async getStreamData( + workspaceId: string, + eventStreamChannelId: string, + ): Promise { + const key = this.getEventStreamKey(workspaceId, eventStreamChannelId); + + return this.cacheStorageService.get(key); + } + + @WithLock('eventStreamChannelId') async addQuery({ workspaceId, - eventStreamId, + eventStreamChannelId, queryId, operationSignature, }: { workspaceId: string; - eventStreamId: string; + eventStreamChannelId: string; queryId: string; operationSignature: RecordGqlOperationSignature; }): Promise { - const key = this.getEventStreamKey(workspaceId, eventStreamId); - const existing = - (await this.cacheStorageService.get< - Record - >(key)) || {}; + const key = this.getEventStreamKey(workspaceId, eventStreamChannelId); + const existing = await this.cacheStorageService.get(key); - existing[queryId] = operationSignature; + if (!isDefined(existing)) { + return; + } - await this.cacheStorageService.set(key, existing); + existing.queries[queryId] = operationSignature; + + await this.cacheStorageService.set(key, existing, EVENT_STREAM_TTL_MS); } - @WithLock('eventStreamId') + @WithLock('eventStreamChannelId') async removeQuery({ workspaceId, - eventStreamId, + eventStreamChannelId, queryId, }: { workspaceId: string; - eventStreamId: string; + eventStreamChannelId: string; queryId: string; }): Promise { - const key = this.getEventStreamKey(workspaceId, eventStreamId); - const existing = - await this.cacheStorageService.get< - Record - >(key); + const key = this.getEventStreamKey(workspaceId, eventStreamChannelId); + const existing = await this.cacheStorageService.get(key); - if (existing && existing[queryId]) { - delete existing[queryId]; - await this.cacheStorageService.set(key, existing); + if (isDefined(existing) && isDefined(existing.queries[queryId])) { + delete existing.queries[queryId]; + await this.cacheStorageService.set(key, existing, EVENT_STREAM_TTL_MS); } } @@ -75,22 +173,22 @@ export class EventStreamService { workspaceId: string, eventStreamId: string, ): Promise> { - const key = this.getEventStreamKey(workspaceId, eventStreamId); - const data = - await this.cacheStorageService.get< - Record - >(key); + const streamData = await this.getStreamData(workspaceId, eventStreamId); - return new Map(Object.entries(data || {})); + if (!isDefined(streamData)) { + return new Map(); + } + + return new Map(Object.entries(streamData.queries)); } - async matchQueriesWithEvent( - queries: Map, + matchQueriesWithEvent( + queries: Record, event: ObjectRecordSubscriptionEvent, - ): Promise { + ): string[] { const matchedQueryIds: string[] = []; - for (const [queryId, operationSignature] of queries.entries()) { + for (const [queryId, operationSignature] of Object.entries(queries)) { if (this.isQueryMatchingEvent(operationSignature, event)) { matchedQueryIds.push(queryId); } @@ -106,4 +204,15 @@ export class EventStreamService { // to be improved return operationSignature.objectNameSingular === event.objectNameSingular; } + + private getEventStreamKey( + workspaceId: string, + eventStreamId: string, + ): string { + return `eventStream:${workspaceId}:${eventStreamId}`; + } + + private getActiveStreamsKey(workspaceId: string): string { + return `workspace:${workspaceId}:activeStreams`; + } } diff --git a/packages/twenty-server/src/engine/subscriptions/subscription.service.ts b/packages/twenty-server/src/engine/subscriptions/subscription.service.ts index e0ed7dae62..75019ee7e0 100644 --- a/packages/twenty-server/src/engine/subscriptions/subscription.service.ts +++ b/packages/twenty-server/src/engine/subscriptions/subscription.service.ts @@ -1,20 +1,11 @@ import { Injectable } from '@nestjs/common'; -import { FieldNode, OperationDefinitionNode, parse } from 'graphql'; - import { RedisClientService } from 'src/engine/core-modules/redis-client/redis-client.service'; -import { WorkspaceManyOrAllFlatEntityMapsCacheService } from 'src/engine/metadata-modules/flat-entity/services/workspace-many-or-all-flat-entity-maps-cache.service'; -import { FlatObjectMetadata } from 'src/engine/metadata-modules/flat-object-metadata/types/flat-object-metadata.type'; -import { OnDbEventDTO } from 'src/engine/subscriptions/dtos/on-db-event.dto'; -import { SubscriptionInput } from 'src/engine/subscriptions/dtos/subscription.input'; import { SubscriptionChannel } from 'src/engine/subscriptions/enums/subscription-channel.enum'; @Injectable() export class SubscriptionService { - constructor( - private readonly redisClient: RedisClientService, - private readonly workspaceManyOrAllFlatEntityMapsCacheService: WorkspaceManyOrAllFlatEntityMapsCacheService, - ) {} + constructor(private readonly redisClient: RedisClientService) {} private getSubscriptionChannel({ channel, @@ -26,6 +17,16 @@ export class SubscriptionService { return `${channel}:${workspaceId}`; } + private getEventStreamChannel({ + workspaceId, + eventStreamChannelId, + }: { + workspaceId: string; + eventStreamChannelId: string; + }) { + return `${SubscriptionChannel.EVENT_STREAM_CHANNEL}:${workspaceId}:${eventStreamChannelId}`; + } + async subscribe({ channel, workspaceId, @@ -40,6 +41,20 @@ export class SubscriptionService { ); } + async subscribeToEventStream({ + workspaceId, + eventStreamChannelId, + }: { + workspaceId: string; + eventStreamChannelId: string; + }) { + const client = this.redisClient.getPubSubClient(); + + return client.asyncIterator( + this.getEventStreamChannel({ workspaceId, eventStreamChannelId }), + ); + } + async publish({ channel, payload, @@ -57,81 +72,20 @@ export class SubscriptionService { ); } - public async isSubscriptionMatchingEvent( - subscription: SubscriptionInput, - event: OnDbEventDTO, - workspaceId: string, - ): Promise { - const objectName = this.parseQueryObjectName(subscription.query); + async publishToEventStream({ + workspaceId, + eventStreamChannelId, + payload, + }: { + workspaceId: string; + eventStreamChannelId: string; + payload: T; + }): Promise { + const client = this.redisClient.getPubSubClient(); - if (!objectName) { - return false; - } - - if ( - subscription.selectedEventActions && - !subscription.selectedEventActions.includes(event.action) - ) { - return false; - } - - const { flatObjectMetadataMaps } = - await this.workspaceManyOrAllFlatEntityMapsCacheService.getOrRecomputeManyOrAllFlatEntityMaps( - { - workspaceId, - flatMapsKeys: ['flatObjectMetadataMaps'], - }, - ); - - const eventObjectMetadata = Object.values(flatObjectMetadataMaps.byId).find( - (metadata: FlatObjectMetadata) => - metadata.nameSingular === event.objectNameSingular, + await client.publish( + this.getEventStreamChannel({ workspaceId, eventStreamChannelId }), + payload, ); - - if (!eventObjectMetadata) { - return false; - } - - const queryObjectNameLower = objectName.toLowerCase(); - const eventNameSingularLower = - eventObjectMetadata.nameSingular.toLowerCase(); - const eventNamePluralLower = eventObjectMetadata.namePlural.toLowerCase(); - - return ( - queryObjectNameLower === eventNameSingularLower || - queryObjectNameLower === eventNamePluralLower - ); - } - - private parseQueryObjectName(queryString: string): string | null { - try { - const { query } = JSON.parse(queryString) as { - query: string; - variables?: Record; - }; - - const ast = parse(query); - - const firstOperation = ast.definitions.find( - (def): def is OperationDefinitionNode => - def.kind === 'OperationDefinition', - ); - - if (!firstOperation) { - return null; - } - - const rootSelection = firstOperation.selectionSet.selections[0]; - - if (rootSelection.kind !== 'Field') { - return null; - } - - const rootField = rootSelection as FieldNode; - - return rootField.name.value; - } catch { - return null; - } } } diff --git a/packages/twenty-server/src/engine/subscriptions/subscriptions.module.ts b/packages/twenty-server/src/engine/subscriptions/subscriptions.module.ts index 7ee227f1f0..6204ce820d 100644 --- a/packages/twenty-server/src/engine/subscriptions/subscriptions.module.ts +++ b/packages/twenty-server/src/engine/subscriptions/subscriptions.module.ts @@ -1,18 +1,19 @@ import { Module } from '@nestjs/common'; +import { TypeOrmModule } from '@nestjs/typeorm'; import { CacheLockModule } from 'src/engine/core-modules/cache-lock/cache-lock.module'; import { CacheStorageModule } from 'src/engine/core-modules/cache-storage/cache-storage.module'; import { RedisClientModule } from 'src/engine/core-modules/redis-client/redis-client.module'; -import { WorkspaceManyOrAllFlatEntityMapsCacheModule } from 'src/engine/metadata-modules/flat-entity/services/workspace-many-or-all-flat-entity-maps-cache.module'; +import { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity'; import { EventStreamService } from 'src/engine/subscriptions/event-stream.service'; import { SubscriptionService } from 'src/engine/subscriptions/subscription.service'; @Module({ imports: [ RedisClientModule, - WorkspaceManyOrAllFlatEntityMapsCacheModule, CacheStorageModule, CacheLockModule, + TypeOrmModule.forFeature([WorkspaceEntity]), ], providers: [SubscriptionService, EventStreamService], exports: [SubscriptionService, EventStreamService], diff --git a/packages/twenty-server/src/engine/subscriptions/types/event-stream-data.type.ts b/packages/twenty-server/src/engine/subscriptions/types/event-stream-data.type.ts new file mode 100644 index 0000000000..e24daa02eb --- /dev/null +++ b/packages/twenty-server/src/engine/subscriptions/types/event-stream-data.type.ts @@ -0,0 +1,10 @@ +import { type RecordGqlOperationSignature } from 'twenty-shared/types'; + +import { type SerializableAuthContext } from 'src/engine/core-modules/auth/types/auth-context.type'; + +export type EventStreamData = { + authContext: SerializableAuthContext; + workspaceId: string; + queries: Record; + createdAt: number; +}; diff --git a/packages/twenty-server/src/engine/subscriptions/types/object-record-subscription-event.type.ts b/packages/twenty-server/src/engine/subscriptions/types/object-record-subscription-event.type.ts new file mode 100644 index 0000000000..359c7fc594 --- /dev/null +++ b/packages/twenty-server/src/engine/subscriptions/types/object-record-subscription-event.type.ts @@ -0,0 +1,5 @@ +import { type ObjectRecordEvent } from 'twenty-shared/database-events'; + +export type ObjectRecordSubscriptionEvent = ObjectRecordEvent & { + objectNameSingular: string; +}; diff --git a/packages/twenty-server/src/engine/workspace-event-emitter/utils/get-channel-id-from-event-stream-id.ts b/packages/twenty-server/src/engine/workspace-event-emitter/utils/get-channel-id-from-event-stream-id.ts new file mode 100644 index 0000000000..f1beab0a36 --- /dev/null +++ b/packages/twenty-server/src/engine/workspace-event-emitter/utils/get-channel-id-from-event-stream-id.ts @@ -0,0 +1,7 @@ +import { v5 } from 'uuid'; + +const EVENT_STREAM_NAMESPACE = '6ba7b810-9dad-11d1-80b4-00c04fd430c8'; + +export const eventStreamIdToChannelId = (eventStreamId: string): string => { + return v5(eventStreamId, EVENT_STREAM_NAMESPACE); +}; diff --git a/packages/twenty-server/src/engine/workspace-event-emitter/utils/wrap-async-iterator-with-cleanup.ts b/packages/twenty-server/src/engine/workspace-event-emitter/utils/wrap-async-iterator-with-cleanup.ts new file mode 100644 index 0000000000..b2c53f1531 --- /dev/null +++ b/packages/twenty-server/src/engine/workspace-event-emitter/utils/wrap-async-iterator-with-cleanup.ts @@ -0,0 +1,32 @@ +export function wrapAsyncIteratorWithCleanup( + iterator: AsyncIterableIterator, + onClose: () => void | Promise, +): AsyncIterableIterator { + return { + next: () => iterator.next(), + return: async () => { + let result: IteratorResult; + + try { + await onClose(); + } finally { + result = (await iterator.return?.()) ?? { + done: true, + value: undefined, + }; + } + + return result; + }, + throw: async (error) => { + if (iterator.throw) { + return iterator.throw(error); + } + + throw error; + }, + [Symbol.asyncIterator]() { + return this; + }, + }; +} diff --git a/packages/twenty-server/src/engine/workspace-event-emitter/workspace-event-emitter.module.ts b/packages/twenty-server/src/engine/workspace-event-emitter/workspace-event-emitter.module.ts index f56f2a4a93..14dfaaf57c 100644 --- a/packages/twenty-server/src/engine/workspace-event-emitter/workspace-event-emitter.module.ts +++ b/packages/twenty-server/src/engine/workspace-event-emitter/workspace-event-emitter.module.ts @@ -1,9 +1,9 @@ import { Global, Module } from '@nestjs/common'; -import { WorkspaceEventEmitter } from 'src/engine/workspace-event-emitter/workspace-event-emitter'; -import { WorkspaceEventEmitterService } from 'src/engine/workspace-event-emitter/workspace-event-emitter.service'; import { SubscriptionsModule } from 'src/engine/subscriptions/subscriptions.module'; +import { WorkspaceEventEmitter } from 'src/engine/workspace-event-emitter/workspace-event-emitter'; import { WorkspaceEventEmitterResolver } from 'src/engine/workspace-event-emitter/workspace-event-emitter.resolver'; +import { WorkspaceEventEmitterService } from 'src/engine/workspace-event-emitter/workspace-event-emitter.service'; @Global() @Module({ diff --git a/packages/twenty-server/src/engine/workspace-event-emitter/workspace-event-emitter.resolver.ts b/packages/twenty-server/src/engine/workspace-event-emitter/workspace-event-emitter.resolver.ts index cb32369006..2b788eec89 100644 --- a/packages/twenty-server/src/engine/workspace-event-emitter/workspace-event-emitter.resolver.ts +++ b/packages/twenty-server/src/engine/workspace-event-emitter/workspace-event-emitter.resolver.ts @@ -1,29 +1,34 @@ import { UseFilters, UseGuards, UsePipes } from '@nestjs/common'; import { Args, Mutation, Resolver, Subscription } from '@nestjs/graphql'; -import { ObjectRecordEvent } from 'twenty-shared/database-events'; import { isDefined } from 'twenty-shared/utils'; +import { type ApiKeyEntity } from 'src/engine/core-modules/api-key/api-key.entity'; import { PreventNestToAutoLogGraphqlErrorsFilter } from 'src/engine/core-modules/graphql/filters/prevent-nest-to-auto-log-graphql-errors.filter'; import { ResolverValidationPipe } from 'src/engine/core-modules/graphql/pipes/resolver-validation.pipe'; +import { UserEntity } from 'src/engine/core-modules/user/user.entity'; import { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity'; +import { AuthApiKey } from 'src/engine/decorators/auth/auth-api-key.decorator'; +import { AuthUserWorkspaceId } from 'src/engine/decorators/auth/auth-user-workspace-id.decorator'; +import { AuthUser } from 'src/engine/decorators/auth/auth-user.decorator'; import { AuthWorkspace } from 'src/engine/decorators/auth/auth-workspace.decorator'; import { NoPermissionGuard } from 'src/engine/guards/no-permission.guard'; import { UserAuthGuard } from 'src/engine/guards/user-auth.guard'; import { WorkspaceAuthGuard } from 'src/engine/guards/workspace-auth.guard'; import { AddQuerySubscriptionInput } from 'src/engine/subscriptions/dtos/add-query-subscription.input'; -import { EventSubscriptionDTO } from 'src/engine/subscriptions/dtos/event-subscription.dto'; -import { ObjectRecordEventDTO } from 'src/engine/subscriptions/dtos/object-record-event.dto'; +import { + EventSubscriptionDTO, + EventWithQueryIdsDTO, +} from 'src/engine/subscriptions/dtos/event-subscription.dto'; import { OnDbEventDTO } from 'src/engine/subscriptions/dtos/on-db-event.dto'; import { OnDbEventInput } from 'src/engine/subscriptions/dtos/on-db-event.input'; import { RemoveQueryFromEventStreamInput } from 'src/engine/subscriptions/dtos/remove-query-subscription.input'; -import { SubscriptionMatchesDTO } from 'src/engine/subscriptions/dtos/subscription-matches.dto'; -import { SubscriptionInput } from 'src/engine/subscriptions/dtos/subscription.input'; import { SubscriptionChannel } from 'src/engine/subscriptions/enums/subscription-channel.enum'; import { EventStreamService } from 'src/engine/subscriptions/event-stream.service'; import { SubscriptionService } from 'src/engine/subscriptions/subscription.service'; -import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event-batch.type'; -import { parseEventNameOrThrow } from 'src/engine/workspace-event-emitter/utils/parse-event-name'; +import { wrapAsyncIteratorWithCleanup } from 'src/engine/workspace-event-emitter/utils/wrap-async-iterator-with-cleanup'; + +import { eventStreamIdToChannelId } from './utils/get-channel-id-from-event-stream-id'; @Resolver() @UseGuards(WorkspaceAuthGuard, UserAuthGuard, NoPermissionGuard) @@ -68,119 +73,57 @@ export class WorkspaceEventEmitterResolver { }); } - @Subscription(() => SubscriptionMatchesDTO, { - nullable: true, - resolve: async function ( - this: WorkspaceEventEmitterResolver, - payload: { onDbEvents: OnDbEventDTO[] }, - args: { subscriptions: SubscriptionInput[] }, - context: { req: { workspace: { id: string } } }, - ): Promise { - const workspaceId = context.req.workspace.id; - - const matches: { subscriptionIds: string[]; event: OnDbEventDTO }[] = []; - - for (const event of payload.onDbEvents) { - const matchedSubscriptionIds = await Promise.all( - args.subscriptions.map(async (subscription) => { - const isMatch = - await this.subscriptionService.isSubscriptionMatchingEvent( - subscription, - event, - workspaceId, - ); - - return isMatch ? subscription.id : null; - }), - ); - - const filteredIds = matchedSubscriptionIds.filter( - (id): id is string => id !== null, - ); - - if (filteredIds.length > 0) { - matches.push({ - subscriptionIds: filteredIds, - event, - }); - } - } - - return { matches }; - }, - }) - onSubscriptionMatch( - @Args('subscriptions', { type: () => [SubscriptionInput] }) - _: SubscriptionInput[], - @AuthWorkspace() workspace: WorkspaceEntity, - ) { - return this.subscriptionService.subscribe({ - channel: SubscriptionChannel.WORKSPACE_EVENT_BATCH_CHANNEL, - workspaceId: workspace.id, - }); - } - @Subscription(() => EventSubscriptionDTO, { nullable: true, - resolve: async function ( - this: WorkspaceEventEmitterResolver, - payload: { workspaceEventBatch: WorkspaceEventBatch }, - args: { eventStreamId: string }, - context: { req: { workspace: { id: string } } }, - ): Promise { - const workspaceId = context.req.workspace.id; - const { eventStreamId } = args; - - const queries = await this.eventStreamService.getQueries( - workspaceId, - eventStreamId, - ); - - const objectNameSingular = - payload.workspaceEventBatch.objectMetadata.nameSingular; - - const eventWithQueryIdsList: { - queryIds: string[]; - event: ObjectRecordEventDTO; - }[] = []; - - for (const event of payload.workspaceEventBatch.events) { - const eventName = parseEventNameOrThrow( - payload.workspaceEventBatch.name, - ); - - const action = eventName.action; - - const eventWithObjectName = { - ...event, - objectNameSingular, - action, - }; - - const matchedQueryIds = - await this.eventStreamService.matchQueriesWithEvent( - queries, - eventWithObjectName, - ); - - if (matchedQueryIds.length > 0) { - eventWithQueryIdsList.push({ - queryIds: matchedQueryIds, - event: eventWithObjectName, - }); - } - } - - return { eventStreamId, eventWithQueryIdsList }; + resolve: ( + payload: EventWithQueryIdsDTO[], + variables: { eventStreamId: string }, + ) => { + return { + eventStreamId: variables.eventStreamId, + eventWithQueryIdsList: payload, + }; }, }) - onEventSubscription( - @Args('eventStreamId') _: string, + async onEventSubscription( + @Args('eventStreamId') eventStreamId: string, @AuthWorkspace() workspace: WorkspaceEntity, + @AuthUser({ allowUndefined: true }) user: UserEntity | undefined, + @AuthUserWorkspaceId() userWorkspaceId: string | undefined, + @AuthApiKey() apiKey: ApiKeyEntity | undefined, ) { - return this.subscriptionService.subscribe({ - channel: SubscriptionChannel.WORKSPACE_EVENT_BATCH_CHANNEL, + const eventStreamChannelId = eventStreamIdToChannelId(eventStreamId); + + await this.eventStreamService.createEventStream({ workspaceId: workspace.id, + eventStreamChannelId, + authContext: { + userId: user?.id, + userWorkspaceId, + apiKeyId: apiKey?.id, + }, + }); + + let iterator: AsyncIterableIterator; + + try { + iterator = await this.subscriptionService.subscribeToEventStream({ + workspaceId: workspace.id, + eventStreamChannelId, + }); + } catch (error) { + await this.eventStreamService.destroyEventStream({ + workspaceId: workspace.id, + eventStreamChannelId, + }); + throw error; + } + + return wrapAsyncIteratorWithCleanup(iterator, async () => { + await this.eventStreamService.destroyEventStream({ + workspaceId: workspace.id, + eventStreamChannelId, + }); }); } @@ -189,9 +132,11 @@ export class WorkspaceEventEmitterResolver { @Args('input') input: AddQuerySubscriptionInput, @AuthWorkspace() workspace: WorkspaceEntity, ): Promise { + const eventStreamChannelId = eventStreamIdToChannelId(input.eventStreamId); + await this.eventStreamService.addQuery({ workspaceId: workspace.id, - eventStreamId: input.eventStreamId, + eventStreamChannelId, queryId: input.queryId, operationSignature: input.operationSignature, }); @@ -204,9 +149,11 @@ export class WorkspaceEventEmitterResolver { @Args('input') input: RemoveQueryFromEventStreamInput, @AuthWorkspace() workspace: WorkspaceEntity, ): Promise { + const eventStreamChannelId = eventStreamIdToChannelId(input.eventStreamId); + await this.eventStreamService.removeQuery({ workspaceId: workspace.id, - eventStreamId: input.eventStreamId, + eventStreamChannelId, queryId: input.queryId, }); diff --git a/packages/twenty-server/src/engine/workspace-event-emitter/workspace-event-emitter.service.ts b/packages/twenty-server/src/engine/workspace-event-emitter/workspace-event-emitter.service.ts index 522cd22589..78607021b8 100644 --- a/packages/twenty-server/src/engine/workspace-event-emitter/workspace-event-emitter.service.ts +++ b/packages/twenty-server/src/engine/workspace-event-emitter/workspace-event-emitter.service.ts @@ -1,23 +1,27 @@ import { Injectable } from '@nestjs/common'; import { type ObjectRecordEvent } from 'twenty-shared/database-events'; +import { isDefined } from 'twenty-shared/utils'; import { transformEventToWebhookEvent } from 'src/engine/core-modules/webhook/utils/transform-event-to-webhook-event'; import { SubscriptionChannel } from 'src/engine/subscriptions/enums/subscription-channel.enum'; +import { EventStreamService } from 'src/engine/subscriptions/event-stream.service'; import { SubscriptionService } from 'src/engine/subscriptions/subscription.service'; +import { type EventStreamData } from 'src/engine/subscriptions/types/event-stream-data.type'; import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event-batch.type'; @Injectable() export class WorkspaceEventEmitterService { - constructor(private readonly subscriptionService: SubscriptionService) {} + constructor( + private readonly subscriptionService: SubscriptionService, + private readonly eventStreamService: EventStreamService, + ) {} async publish( workspaceEventBatch: WorkspaceEventBatch, ): Promise { const [nameSingular, operation] = workspaceEventBatch.name.split('.'); - const batchEvents = []; - for (const eventData of workspaceEventBatch.events) { const { record, updatedFields } = transformEventToWebhookEvent({ eventName: workspaceEventBatch.name, @@ -32,8 +36,6 @@ export class WorkspaceEventEmitterService { ...(updatedFields && { updatedFields }), }; - batchEvents.push(event); - // Publish individual events to legacy channel (onDbEvent) await this.subscriptionService.publish({ channel: SubscriptionChannel.DATABASE_EVENT_CHANNEL, @@ -42,10 +44,86 @@ export class WorkspaceEventEmitterService { }); } - await this.subscriptionService.publish({ - channel: SubscriptionChannel.WORKSPACE_EVENT_BATCH_CHANNEL, - workspaceId: workspaceEventBatch.workspaceId, - payload: { workspaceEventBatch: workspaceEventBatch }, - }); + await this.publishToEventStreams(workspaceEventBatch); + } + + private async publishToEventStreams( + workspaceEventBatch: WorkspaceEventBatch, + ): Promise { + const workspaceId = workspaceEventBatch.workspaceId; + + const activeStreamIds = + await this.eventStreamService.getActiveStreamIds(workspaceId); + + if (activeStreamIds.length === 0) { + return; + } + + const streamsData = await this.eventStreamService.getStreamsData( + workspaceId, + activeStreamIds, + ); + + const streamIdsToRemove: string[] = []; + + for (const [streamChannelId, streamData] of streamsData) { + if (!isDefined(streamData)) { + streamIdsToRemove.push(streamChannelId); + continue; + } + + await this.processStreamEvents( + streamChannelId, + streamData, + workspaceEventBatch, + ); + } + + await this.eventStreamService.removeFromActiveStreams( + workspaceId, + streamIdsToRemove, + ); + } + + private async processStreamEvents( + streamChannelId: string, + streamData: EventStreamData, + workspaceEventBatch: WorkspaceEventBatch, + ): Promise { + const matchedEvents: { + queryIds: string[]; + event: ObjectRecordEvent & { objectNameSingular: string }; + }[] = []; + + const objectNameSingular = workspaceEventBatch.objectMetadata.nameSingular; + + for (const event of workspaceEventBatch.events) { + const eventWithObjectName = { + objectNameSingular, + ...event, + }; + + const matchedQueryIds = this.eventStreamService.matchQueriesWithEvent( + streamData.queries, + eventWithObjectName, + ); + + if (matchedQueryIds.length === 0) { + continue; + } + + matchedEvents.push({ + queryIds: matchedQueryIds, + event: eventWithObjectName, + }); + } + + if (matchedEvents.length > 0) { + await this.subscriptionService.publishToEventStream({ + workspaceId: workspaceEventBatch.workspaceId, + eventStreamChannelId: streamChannelId, + payload: matchedEvents, + }); + } } }