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

packages/twenty-front/src/generated-metadata/graphql.ts

@@ -4219,7 +4219,6 @@ export type Subscription = {
   __typename?: 'Subscription';
   onDbEvent: OnDbEvent;
   onEventSubscription?: Maybe<EventSubscription>;
-  onSubscriptionMatch?: Maybe<SubscriptionMatches>;
   serverlessFunctionLogs: ServerlessFunctionLogs;
 };
 
@@ -4234,37 +4233,15 @@ export type SubscriptionOnEventSubscriptionArgs = {
 };
 
 
-export type SubscriptionOnSubscriptionMatchArgs = {
-  subscriptions: Array<SubscriptionInput>;
-};
-
-
 export type SubscriptionServerlessFunctionLogsArgs = {
   input: ServerlessFunctionLogsInput;
 };
 
-export type SubscriptionInput = {
-  id: Scalars['String'];
-  query: Scalars['String'];
-  selectedEventActions?: InputMaybe<Array<DatabaseEventAction>>;
-};
-
 export enum SubscriptionInterval {
   Month = 'Month',
   Year = 'Year'
 }
 
-export type SubscriptionMatch = {
-  __typename?: 'SubscriptionMatch';
-  event: OnDbEvent;
-  subscriptionIds: Array<Scalars['String']>;
-};
-
-export type SubscriptionMatches = {
-  __typename?: 'SubscriptionMatches';
-  matches: Array<SubscriptionMatch>;
-};
-
 export enum SubscriptionStatus {
   Active = 'Active',
   Canceled = 'Canceled',

packages/twenty-front/src/generated/graphql.ts

@@ -4094,7 +4094,6 @@ export type Subscription = {
   __typename?: 'Subscription';
   onDbEvent: OnDbEvent;
   onEventSubscription?: Maybe<EventSubscription>;
-  onSubscriptionMatch?: Maybe<SubscriptionMatches>;
   serverlessFunctionLogs: ServerlessFunctionLogs;
 };
 
@@ -4109,37 +4108,15 @@ export type SubscriptionOnEventSubscriptionArgs = {
 };
 
 
-export type SubscriptionOnSubscriptionMatchArgs = {
-  subscriptions: Array<SubscriptionInput>;
-};
-
-
 export type SubscriptionServerlessFunctionLogsArgs = {
   input: ServerlessFunctionLogsInput;
 };
 
-export type SubscriptionInput = {
-  id: Scalars['String'];
-  query: Scalars['String'];
-  selectedEventActions?: InputMaybe<Array<DatabaseEventAction>>;
-};
-
 export enum SubscriptionInterval {
   Month = 'Month',
   Year = 'Year'
 }
 
-export type SubscriptionMatch = {
-  __typename?: 'SubscriptionMatch';
-  event: OnDbEvent;
-  subscriptionIds: Array<Scalars['String']>;
-};
-
-export type SubscriptionMatches = {
-  __typename?: 'SubscriptionMatches';
-  matches: Array<SubscriptionMatch>;
-};
-
 export enum SubscriptionStatus {
   Active = 'Active',
   Canceled = 'Canceled',

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',

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<number> {
+    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<string[]>(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<string[]> {
+    if (this.isRedisCache()) {
+      return (this.cache as RedisCache).store.client.sMembers(this.getKey(key));
+    }
+
+    return (await this.get<string[]>(key)) ?? [];
+  }
+
   async flush() {
     return this.cache.reset();
   }

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

packages/twenty-server/src/engine/subscriptions/dtos/subscription-matches.dto.ts

@@ -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[];
-}

packages/twenty-server/src/engine/subscriptions/dtos/subscription.input.ts

@@ -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[];
-}

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',
 }

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<void> {
+    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<void> {
+    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<string[]> {
+    return this.cacheStorageService.setMembers(
+      this.getActiveStreamsKey(workspaceId),
+    );
+  }
+
+  async removeFromActiveStreams(
+    workspaceId: string,
+    streamIdsToRemove: string[],
+  ): Promise<void> {
+    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<Map<string, EventStreamData | undefined>> {
+    if (streamChannelIds.length === 0) {
+      return new Map();
+    }
+
+    const keys = streamChannelIds.map((id) =>
+      this.getEventStreamKey(workspaceId, id),
+    );
+    const values = await this.cacheStorageService.mget<EventStreamData>(keys);
+
+    const result = new Map<string, EventStreamData | undefined>();
+
+    streamChannelIds.forEach((id, index) => {
+      result.set(id, values[index]);
+    });
+
+    return result;
+  }
+
+  async getStreamData(
+    workspaceId: string,
+    eventStreamChannelId: string,
+  ): Promise<EventStreamData | undefined> {
+    const key = this.getEventStreamKey(workspaceId, eventStreamChannelId);
+
+    return this.cacheStorageService.get<EventStreamData>(key);
+  }
+
+  @WithLock('eventStreamChannelId')
   async addQuery({
     workspaceId,
-    eventStreamId,
+    eventStreamChannelId,
     queryId,
     operationSignature,
   }: {
     workspaceId: string;
-    eventStreamId: string;
+    eventStreamChannelId: string;
     queryId: string;
     operationSignature: RecordGqlOperationSignature;
   }): Promise<void> {
-    const key = this.getEventStreamKey(workspaceId, eventStreamId);
-    const existing =
-      (await this.cacheStorageService.get<
-        Record<string, RecordGqlOperationSignature>
-      >(key)) || {};
+    const key = this.getEventStreamKey(workspaceId, eventStreamChannelId);
+    const existing = await this.cacheStorageService.get<EventStreamData>(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<void> {
-    const key = this.getEventStreamKey(workspaceId, eventStreamId);
-    const existing =
-      await this.cacheStorageService.get<
-        Record<string, RecordGqlOperationSignature>
-      >(key);
+    const key = this.getEventStreamKey(workspaceId, eventStreamChannelId);
+    const existing = await this.cacheStorageService.get<EventStreamData>(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<Map<string, RecordGqlOperationSignature>> {
-    const key = this.getEventStreamKey(workspaceId, eventStreamId);
-    const data =
-      await this.cacheStorageService.get<
-        Record<string, RecordGqlOperationSignature>
-      >(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<string, RecordGqlOperationSignature>,
+  matchQueriesWithEvent(
+    queries: Record<string, RecordGqlOperationSignature>,
     event: ObjectRecordSubscriptionEvent,
-  ): Promise<string[]> {
+  ): 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`;
+  }
 }

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<T>({
     channel,
     payload,
@@ -57,81 +72,20 @@ export class SubscriptionService {
     );
   }
 
-  public async isSubscriptionMatchingEvent(
-    subscription: SubscriptionInput,
-    event: OnDbEventDTO,
-    workspaceId: string,
-  ): Promise<boolean> {
-    const objectName = this.parseQueryObjectName(subscription.query);
+  async publishToEventStream<T>({
+    workspaceId,
+    eventStreamChannelId,
+    payload,
+  }: {
+    workspaceId: string;
+    eventStreamChannelId: string;
+    payload: T;
+  }): Promise<void> {
+    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<string, unknown>;
-      };
-
-      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;
-    }
   }
 }

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],

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<string, RecordGqlOperationSignature>;
+  createdAt: number;
+};

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;
+};

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);
+};

packages/twenty-server/src/engine/workspace-event-emitter/utils/wrap-async-iterator-with-cleanup.ts

@@ -0,0 +1,32 @@
+export function wrapAsyncIteratorWithCleanup<T>(
+  iterator: AsyncIterableIterator<T>,
+  onClose: () => void | Promise<void>,
+): AsyncIterableIterator<T> {
+  return {
+    next: () => iterator.next(),
+    return: async () => {
+      let result: IteratorResult<T>;
+
+      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;
+    },
+  };
+}

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({

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<SubscriptionMatchesDTO> {
-      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<ObjectRecordEvent> },
-      args: { eventStreamId: string },
-      context: { req: { workspace: { id: string } } },
-    ): Promise<EventSubscriptionDTO> {
-      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<EventWithQueryIdsDTO[]>;
+
+    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<boolean> {
+    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<boolean> {
+    const eventStreamChannelId = eventStreamIdToChannelId(input.eventStreamId);
+
     await this.eventStreamService.removeQuery({
       workspaceId: workspace.id,
-      eventStreamId: input.eventStreamId,
+      eventStreamChannelId,
       queryId: input.queryId,
     });
 

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<ObjectRecordEvent>,
   ): Promise<void> {
     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<ObjectRecordEvent>,
+  ): Promise<void> {
+    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<ObjectRecordEvent>,
+  ): Promise<void> {
+    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,
+      });
+    }
   }
 }