Fix subscription cross tenant issue (#16670)

As title

packages/twenty-server/src/engine/api/graphql/workspace-query-runner/listeners/entity-events-to-db.listener.ts

@@ -18,10 +18,10 @@ import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queu
 import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service';
 import { CallWebhookJobsJob } from 'src/engine/core-modules/webhook/jobs/call-webhook-jobs.job';
 import { CallDatabaseEventTriggerJobsJob } from 'src/engine/metadata-modules/database-event-trigger/jobs/call-database-event-trigger-jobs.job';
-import { SubscriptionsService } from 'src/engine/subscriptions/subscriptions.service';
 import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event-batch.type';
 import { UpsertTimelineActivityFromInternalEvent } from 'src/modules/timeline/jobs/upsert-timeline-activity-from-internal-event.job';
 import { WorkspaceEventBatchForWebhook } from 'src/engine/core-modules/webhook/types/workspace-event-batch-for-webhook.type';
+import { WorkspaceEventEmitterService } from 'src/engine/workspace-event-emitter/workspace-event-emitter.service';
 
 @Injectable()
 export class EntityEventsToDbListener {
@@ -32,7 +32,7 @@ export class EntityEventsToDbListener {
     private readonly webhookQueueService: MessageQueueService,
     @InjectMessageQueue(MessageQueue.triggerQueue)
     private readonly triggerQueueService: MessageQueueService,
-    private readonly subscriptionsService: SubscriptionsService,
+    private readonly workspaceEventEmitterService: WorkspaceEventEmitterService,
   ) {}
 
   @OnDatabaseBatchEvent('*', DatabaseEventAction.CREATED)
@@ -79,7 +79,7 @@ export class EntityEventsToDbListener {
     };
 
     const promises = [
-      this.subscriptionsService.publish(batchEvent),
+      this.workspaceEventEmitterService.publish(batchEvent),
       this.webhookQueueService.add<WorkspaceEventBatchForWebhook<T>>(
         CallWebhookJobsJob.name,
         batchEventForWebhook,

packages/twenty-server/src/engine/core-modules/redis-client/redis-client.service.ts

@@ -2,6 +2,7 @@ import { Injectable, type OnModuleDestroy } from '@nestjs/common';
 
 import IORedis from 'ioredis';
 import { isDefined } from 'twenty-shared/utils';
+import { RedisPubSub } from 'graphql-redis-subscriptions';
 
 import { TwentyConfigService } from 'src/engine/core-modules/twenty-config/twenty-config.service';
 
@@ -9,6 +10,7 @@ import { TwentyConfigService } from 'src/engine/core-modules/twenty-config/twent
 export class RedisClientService implements OnModuleDestroy {
   private redisClient: IORedis | null = null;
   private redisQueueClient: IORedis | null = null;
+  private redisPubSubClient: RedisPubSub | null = null;
 
   constructor(private readonly twentyConfigService: TwentyConfigService) {}
 
@@ -46,6 +48,19 @@ export class RedisClientService implements OnModuleDestroy {
     return this.redisClient;
   }
 
+  getPubSubClient() {
+    if (!this.redisPubSubClient) {
+      const redisClient = this.getClient();
+
+      this.redisPubSubClient = new RedisPubSub({
+        publisher: redisClient.duplicate(),
+        subscriber: redisClient.duplicate(),
+      });
+    }
+
+    return this.redisPubSubClient;
+  }
+
   async onModuleDestroy() {
     if (isDefined(this.redisQueueClient)) {
       await this.redisQueueClient.quit();
@@ -55,5 +70,9 @@ export class RedisClientService implements OnModuleDestroy {
       await this.redisClient.quit();
       this.redisClient = null;
     }
+    if (isDefined(this.redisPubSubClient)) {
+      await this.redisPubSubClient.close();
+      this.redisPubSubClient = null;
+    }
   }
 }

packages/twenty-server/src/engine/metadata-modules/serverless-function/constants/serverless-function-logs-trigger.ts

@@ -1 +0,0 @@
-export const SERVERLESS_FUNCTION_LOGS_TRIGGER = 'serverlessFunctionLogs';

packages/twenty-server/src/engine/metadata-modules/serverless-function/serverless-function.resolver.ts

@@ -1,10 +1,9 @@
-import { Inject, UseFilters, UseGuards, UsePipes } from '@nestjs/common';
+import { UseFilters, UseGuards, UsePipes } from '@nestjs/common';
 import { Args, Mutation, Query, Resolver, Subscription } from '@nestjs/graphql';
 import { InjectRepository } from '@nestjs/typeorm';
 
 import graphqlTypeJson from 'graphql-type-json';
 import { Repository } from 'typeorm';
-import { RedisPubSub } from 'graphql-redis-subscriptions';
 import { isDefined } from 'twenty-shared/utils';
 import { PermissionFlagType } from 'twenty-shared/constants';
 
@@ -28,7 +27,8 @@ import { ServerlessFunctionService } from 'src/engine/metadata-modules/serverles
 import { serverlessFunctionGraphQLApiExceptionHandler } from 'src/engine/metadata-modules/serverless-function/utils/serverless-function-graphql-api-exception-handler.utils';
 import { ServerlessFunctionLogsDTO } from 'src/engine/metadata-modules/serverless-function/dtos/serverless-function-logs.dto';
 import { ServerlessFunctionLogsInput } from 'src/engine/metadata-modules/serverless-function/dtos/serverless-function-logs.input';
-import { SERVERLESS_FUNCTION_LOGS_TRIGGER } from 'src/engine/metadata-modules/serverless-function/constants/serverless-function-logs-trigger';
+import { SubscriptionChannel } from 'src/engine/subscriptions/enums/subscription-channel.enum';
+import { SubscriptionService } from 'src/engine/subscriptions/subscription.service';
 
 @UseGuards(
   WorkspaceAuthGuard,
@@ -43,8 +43,7 @@ export class ServerlessFunctionResolver {
     private readonly serverlessFunctionService: ServerlessFunctionService,
     @InjectRepository(ServerlessFunctionEntity)
     private readonly serverlessFunctionRepository: Repository<ServerlessFunctionEntity>,
-    @Inject('PUB_SUB')
-    private readonly pubSub: RedisPubSub,
+    private readonly subscriptionService: SubscriptionService,
   ) {}
 
   @Query(() => ServerlessFunctionDTO)
@@ -226,7 +225,13 @@ export class ServerlessFunctionResolver {
       );
     },
   })
-  serverlessFunctionLogs(@Args('input') _: ServerlessFunctionLogsInput) {
-    return this.pubSub.asyncIterator(SERVERLESS_FUNCTION_LOGS_TRIGGER);
+  serverlessFunctionLogs(
+    @Args('input') _: ServerlessFunctionLogsInput,
+    @AuthWorkspace() workspace: WorkspaceEntity,
+  ) {
+    return this.subscriptionService.subscribe({
+      channel: SubscriptionChannel.SERVERLESS_FUNCTION_LOGS_CHANNEL,
+      workspaceId: workspace.id,
+    });
   }
 }

packages/twenty-server/src/engine/metadata-modules/serverless-function/serverless-function.service.ts

@@ -1,4 +1,4 @@
-import { Inject, Injectable } from '@nestjs/common';
+import { Injectable } from '@nestjs/common';
 import { InjectRepository } from '@nestjs/typeorm';
 
 import { join } from 'path';
@@ -6,11 +6,10 @@ import { join } from 'path';
 import deepEqual from 'deep-equal';
 import { isDefined } from 'twenty-shared/utils';
 import { IsNull, Not, Repository } from 'typeorm';
-import { RedisPubSub } from 'graphql-redis-subscriptions';
 import { Sources } from 'twenty-shared/types';
 import {
-  DEFAULT_API_URL_NAME,
   DEFAULT_API_KEY_NAME,
+  DEFAULT_API_URL_NAME,
 } from 'twenty-shared/application';
 
 import { FileStorageExceptionCode } from 'src/engine/core-modules/file-storage/interfaces/file-storage-exception';
@@ -36,11 +35,11 @@ import {
   WorkflowVersionStepException,
   WorkflowVersionStepExceptionCode,
 } from 'src/modules/workflow/common/exceptions/workflow-version-step.exception';
-import { SERVERLESS_FUNCTION_LOGS_TRIGGER } from 'src/engine/metadata-modules/serverless-function/constants/serverless-function-logs-trigger';
 import { ApplicationTokenService } from 'src/engine/core-modules/auth/token/services/application-token.service';
 import { buildEnvVar } from 'src/engine/core-modules/serverless/drivers/utils/build-env-var';
-import { AccessTokenService } from 'src/engine/core-modules/auth/token/services/access-token.service';
 import { cleanServerUrl } from 'src/utils/clean-server-url';
+import { SubscriptionService } from 'src/engine/subscriptions/subscription.service';
+import { SubscriptionChannel } from 'src/engine/subscriptions/enums/subscription-channel.enum';
 
 const MIN_TOKEN_EXPIRATION_IN_SECONDS = 5;
 
@@ -55,10 +54,8 @@ export class ServerlessFunctionService {
     private readonly throttlerService: ThrottlerService,
     private readonly twentyConfigService: TwentyConfigService,
     private readonly auditService: AuditService,
-    private readonly accessTokenService: AccessTokenService,
     private readonly applicationTokenService: ApplicationTokenService,
-    @Inject('PUB_SUB')
-    private readonly pubSub: RedisPubSub,
+    private readonly subscriptionService: SubscriptionService,
   ) {}
 
   async hasServerlessFunctionPublishedVersion(serverlessFunctionId: string) {
@@ -166,15 +163,19 @@ export class ServerlessFunctionService {
       console.log(resultServerlessFunction.logs);
     }
 
-    await this.pubSub.publish(SERVERLESS_FUNCTION_LOGS_TRIGGER, {
-      serverlessFunctionLogs: {
-        logs: resultServerlessFunction.logs,
-        id: functionToExecute.id,
-        name: functionToExecute.name,
-        universalIdentifier: functionToExecute.universalIdentifier,
-        applicationId: functionToExecute.applicationId,
-        applicationUniversalIdentifier:
-          functionToExecute.application?.universalIdentifier,
+    await this.subscriptionService.publish({
+      channel: SubscriptionChannel.SERVERLESS_FUNCTION_LOGS_CHANNEL,
+      workspaceId,
+      payload: {
+        serverlessFunctionLogs: {
+          logs: resultServerlessFunction.logs,
+          id: functionToExecute.id,
+          name: functionToExecute.name,
+          universalIdentifier: functionToExecute.universalIdentifier,
+          applicationId: functionToExecute.applicationId,
+          applicationUniversalIdentifier:
+            functionToExecute.application?.universalIdentifier,
+        },
       },
     });
 

packages/twenty-server/src/engine/subscriptions/constants/on-db-event-trigger.ts

@@ -1 +0,0 @@
-export const ON_DB_EVENT_TRIGGER = 'onDbEvent';

packages/twenty-server/src/engine/subscriptions/enums/subscription-channel.enum.ts

@@ -0,0 +1,4 @@
+export enum SubscriptionChannel {
+  DATABASE_EVENT_CHANNEL = 'DATABASE_EVENT_CHANNEL',
+  SERVERLESS_FUNCTION_LOGS_CHANNEL = 'SERVERLESS_FUNCTION_LOGS_CHANNEL',
+}

packages/twenty-server/src/engine/subscriptions/subscription.service.ts

@@ -0,0 +1,50 @@
+import { Injectable } from '@nestjs/common';
+
+import { SubscriptionChannel } from 'src/engine/subscriptions/enums/subscription-channel.enum';
+import { RedisClientService } from 'src/engine/core-modules/redis-client/redis-client.service';
+
+@Injectable()
+export class SubscriptionService {
+  constructor(private readonly redisClient: RedisClientService) {}
+
+  private getSubscriptionChannel({
+    channel,
+    workspaceId,
+  }: {
+    channel: SubscriptionChannel;
+    workspaceId: string;
+  }) {
+    return `${channel}:${workspaceId}`;
+  }
+
+  async subscribe({
+    channel,
+    workspaceId,
+  }: {
+    channel: SubscriptionChannel;
+    workspaceId: string;
+  }) {
+    const client = this.redisClient.getPubSubClient();
+
+    return client.asyncIterator(
+      this.getSubscriptionChannel({ channel, workspaceId }),
+    );
+  }
+
+  async publish<T>({
+    channel,
+    payload,
+    workspaceId,
+  }: {
+    channel: SubscriptionChannel;
+    payload: T;
+    workspaceId: string;
+  }): Promise<void> {
+    const client = this.redisClient.getPubSubClient();
+
+    await client.publish(
+      this.getSubscriptionChannel({ channel, workspaceId }),
+      payload,
+    );
+  }
+}

packages/twenty-server/src/engine/subscriptions/subscriptions.module.ts

@@ -1,34 +1,11 @@
-import { Inject, Module, type OnModuleDestroy } from '@nestjs/common';
+import { Module } from '@nestjs/common';
 
-import { RedisPubSub } from 'graphql-redis-subscriptions';
-
-import { RedisClientService } from 'src/engine/core-modules/redis-client/redis-client.service';
-import { SubscriptionsResolver } from 'src/engine/subscriptions/subscriptions.resolver';
-import { SubscriptionsService } from 'src/engine/subscriptions/subscriptions.service';
+import { SubscriptionService } from 'src/engine/subscriptions/subscription.service';
+import { RedisClientModule } from 'src/engine/core-modules/redis-client/redis-client.module';
 
 @Module({
-  providers: [
-    {
-      provide: 'PUB_SUB',
-      inject: [RedisClientService],
-
-      useFactory: (redisClientService: RedisClientService) =>
-        new RedisPubSub({
-          publisher: redisClientService.getClient().duplicate(),
-          subscriber: redisClientService.getClient().duplicate(),
-        }),
-    },
-    SubscriptionsResolver,
-    SubscriptionsService,
-  ],
-  exports: ['PUB_SUB', SubscriptionsService],
+  imports: [RedisClientModule],
+  providers: [SubscriptionService],
+  exports: [SubscriptionService],
 })
-export class SubscriptionsModule implements OnModuleDestroy {
-  constructor(@Inject('PUB_SUB') private readonly pubSub: RedisPubSub) {}
-
-  async onModuleDestroy() {
-    if (this.pubSub) {
-      await this.pubSub.close();
-    }
-  }
-}
+export class SubscriptionsModule {}

packages/twenty-server/src/engine/subscriptions/subscriptions.service.ts

@@ -1,36 +0,0 @@
-import { Inject, Injectable } from '@nestjs/common';
-
-import { RedisPubSub } from 'graphql-redis-subscriptions';
-import { type ObjectRecordEvent } from 'twenty-shared/database-events';
-
-import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event-batch.type';
-import { transformEventToWebhookEvent } from 'src/engine/core-modules/webhook/utils/transform-event-to-webhook-event';
-import { ON_DB_EVENT_TRIGGER } from 'src/engine/subscriptions/constants/on-db-event-trigger';
-
-@Injectable()
-export class SubscriptionsService {
-  constructor(@Inject('PUB_SUB') private readonly pubSub: RedisPubSub) {}
-
-  async publish(
-    workspaceEventBatch: WorkspaceEventBatch<ObjectRecordEvent>,
-  ): Promise<void> {
-    const [nameSingular, operation] = workspaceEventBatch.name.split('.');
-
-    for (const eventData of workspaceEventBatch.events) {
-      const { record, updatedFields } = transformEventToWebhookEvent({
-        eventName: workspaceEventBatch.name,
-        event: eventData,
-      });
-
-      await this.pubSub.publish(ON_DB_EVENT_TRIGGER, {
-        onDbEvent: {
-          action: operation,
-          objectNameSingular: nameSingular,
-          eventDate: new Date(),
-          record,
-          ...(updatedFields && { updatedFields }),
-        },
-      });
-    }
-  }
-}

packages/twenty-server/src/engine/workspace-event-emitter/workspace-event-emitter.module.ts

@@ -1,11 +1,18 @@
 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 { WorkspaceEventEmitterResolver } from 'src/engine/workspace-event-emitter/workspace-event-emitter.resolver';
 
 @Global()
 @Module({
-  imports: [],
-  providers: [WorkspaceEventEmitter],
-  exports: [WorkspaceEventEmitter],
+  imports: [SubscriptionsModule],
+  providers: [
+    WorkspaceEventEmitter,
+    WorkspaceEventEmitterService,
+    WorkspaceEventEmitterResolver,
+  ],
+  exports: [WorkspaceEventEmitter, WorkspaceEventEmitterService],
 })
 export class WorkspaceEventEmitterModule {}

packages/twenty-server/src/engine/workspace-event-emitter/workspace-event-emitter.resolver.ts

@@ -1,7 +1,6 @@
-import { Inject, UseFilters, UseGuards, UsePipes } from '@nestjs/common';
+import { UseFilters, UseGuards, UsePipes } from '@nestjs/common';
 import { Args, Resolver, Subscription } from '@nestjs/graphql';
 
-import { RedisPubSub } from 'graphql-redis-subscriptions';
 import { isDefined } from 'twenty-shared/utils';
 
 import { PreventNestToAutoLogGraphqlErrorsFilter } from 'src/engine/core-modules/graphql/filters/prevent-nest-to-auto-log-graphql-errors.filter';
@@ -11,14 +10,17 @@ import { UserAuthGuard } from 'src/engine/guards/user-auth.guard';
 import { WorkspaceAuthGuard } from 'src/engine/guards/workspace-auth.guard';
 import { OnDbEventDTO } from 'src/engine/subscriptions/dtos/on-db-event.dto';
 import { OnDbEventInput } from 'src/engine/subscriptions/dtos/on-db-event.input';
-import { ON_DB_EVENT_TRIGGER } from 'src/engine/subscriptions/constants/on-db-event-trigger';
+import { SubscriptionService } from 'src/engine/subscriptions/subscription.service';
+import { AuthWorkspace } from 'src/engine/decorators/auth/auth-workspace.decorator';
+import { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity';
+import { SubscriptionChannel } from 'src/engine/subscriptions/enums/subscription-channel.enum';
 
 @Resolver()
 @UseGuards(WorkspaceAuthGuard, UserAuthGuard, NoPermissionGuard)
 @UsePipes(ResolverValidationPipe)
 @UseFilters(PreventNestToAutoLogGraphqlErrorsFilter)
-export class SubscriptionsResolver {
-  constructor(@Inject('PUB_SUB') private readonly pubSub: RedisPubSub) {}
+export class WorkspaceEventEmitterResolver {
+  constructor(private readonly subscriptionService: SubscriptionService) {}
 
   @Subscription(() => OnDbEventDTO, {
     filter: (
@@ -43,7 +45,13 @@ export class SubscriptionsResolver {
       );
     },
   })
-  onDbEvent(@Args('input') _: OnDbEventInput) {
-    return this.pubSub.asyncIterator(ON_DB_EVENT_TRIGGER);
+  onDbEvent(
+    @Args('input') _: OnDbEventInput,
+    @AuthWorkspace() workspace: WorkspaceEntity,
+  ) {
+    return this.subscriptionService.subscribe({
+      channel: SubscriptionChannel.DATABASE_EVENT_CHANNEL,
+      workspaceId: workspace.id,
+    });
   }
 }

packages/twenty-server/src/engine/workspace-event-emitter/workspace-event-emitter.service.ts

@@ -0,0 +1,40 @@
+import { Injectable } from '@nestjs/common';
+
+import { type ObjectRecordEvent } from 'twenty-shared/database-events';
+
+import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event-batch.type';
+import { transformEventToWebhookEvent } from 'src/engine/core-modules/webhook/utils/transform-event-to-webhook-event';
+import { SubscriptionService } from 'src/engine/subscriptions/subscription.service';
+import { SubscriptionChannel } from 'src/engine/subscriptions/enums/subscription-channel.enum';
+
+@Injectable()
+export class WorkspaceEventEmitterService {
+  constructor(private readonly subscriptionService: SubscriptionService) {}
+
+  async publish(
+    workspaceEventBatch: WorkspaceEventBatch<ObjectRecordEvent>,
+  ): Promise<void> {
+    const [nameSingular, operation] = workspaceEventBatch.name.split('.');
+
+    for (const eventData of workspaceEventBatch.events) {
+      const { record, updatedFields } = transformEventToWebhookEvent({
+        eventName: workspaceEventBatch.name,
+        event: eventData,
+      });
+
+      await this.subscriptionService.publish({
+        channel: SubscriptionChannel.DATABASE_EVENT_CHANNEL,
+        workspaceId: workspaceEventBatch.workspaceId,
+        payload: {
+          onDbEvent: {
+            action: operation,
+            objectNameSingular: nameSingular,
+            eventDate: new Date(),
+            record,
+            ...(updatedFields && { updatedFields }),
+          },
+        },
+      });
+    }
+  }
+}