From 52cf3775b3a8dfcbb3aafcb1cfa88a917339ce1b Mon Sep 17 00:00:00 2001 From: martmull Date: Thu, 18 Dec 2025 15:22:17 +0100 Subject: [PATCH] Fix subscription cross tenant issue (#16670) As title --- .../listeners/entity-events-to-db.listener.ts | 6 +-- .../redis-client/redis-client.service.ts | 19 +++++++ .../serverless-function-logs-trigger.ts | 1 - .../serverless-function.resolver.ts | 19 ++++--- .../serverless-function.service.ts | 35 ++++++------- .../constants/on-db-event-trigger.ts | 1 - .../enums/subscription-channel.enum.ts | 4 ++ .../subscriptions/subscription.service.ts | 50 +++++++++++++++++++ .../subscriptions/subscriptions.module.ts | 37 +++----------- .../subscriptions/subscriptions.service.ts | 36 ------------- .../workspace-event-emitter.module.ts | 13 +++-- .../workspace-event-emitter.resolver.ts} | 22 +++++--- .../workspace-event-emitter.service.ts | 40 +++++++++++++++ 13 files changed, 178 insertions(+), 105 deletions(-) delete mode 100644 packages/twenty-server/src/engine/metadata-modules/serverless-function/constants/serverless-function-logs-trigger.ts delete mode 100644 packages/twenty-server/src/engine/subscriptions/constants/on-db-event-trigger.ts create mode 100644 packages/twenty-server/src/engine/subscriptions/enums/subscription-channel.enum.ts create mode 100644 packages/twenty-server/src/engine/subscriptions/subscription.service.ts delete mode 100644 packages/twenty-server/src/engine/subscriptions/subscriptions.service.ts rename packages/twenty-server/src/engine/{subscriptions/subscriptions.resolver.ts => workspace-event-emitter/workspace-event-emitter.resolver.ts} (67%) create mode 100644 packages/twenty-server/src/engine/workspace-event-emitter/workspace-event-emitter.service.ts diff --git a/packages/twenty-server/src/engine/api/graphql/workspace-query-runner/listeners/entity-events-to-db.listener.ts b/packages/twenty-server/src/engine/api/graphql/workspace-query-runner/listeners/entity-events-to-db.listener.ts index 3d1a34bb53e..3b9974ecefd 100644 --- a/packages/twenty-server/src/engine/api/graphql/workspace-query-runner/listeners/entity-events-to-db.listener.ts +++ b/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>( CallWebhookJobsJob.name, batchEventForWebhook, diff --git a/packages/twenty-server/src/engine/core-modules/redis-client/redis-client.service.ts b/packages/twenty-server/src/engine/core-modules/redis-client/redis-client.service.ts index af4e89a09ad..c974374fbd6 100644 --- a/packages/twenty-server/src/engine/core-modules/redis-client/redis-client.service.ts +++ b/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; + } } } diff --git a/packages/twenty-server/src/engine/metadata-modules/serverless-function/constants/serverless-function-logs-trigger.ts b/packages/twenty-server/src/engine/metadata-modules/serverless-function/constants/serverless-function-logs-trigger.ts deleted file mode 100644 index be8db977124..00000000000 --- a/packages/twenty-server/src/engine/metadata-modules/serverless-function/constants/serverless-function-logs-trigger.ts +++ /dev/null @@ -1 +0,0 @@ -export const SERVERLESS_FUNCTION_LOGS_TRIGGER = 'serverlessFunctionLogs'; diff --git a/packages/twenty-server/src/engine/metadata-modules/serverless-function/serverless-function.resolver.ts b/packages/twenty-server/src/engine/metadata-modules/serverless-function/serverless-function.resolver.ts index 44e8329ad7a..2f9d4772a73 100644 --- a/packages/twenty-server/src/engine/metadata-modules/serverless-function/serverless-function.resolver.ts +++ b/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, - @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, + }); } } diff --git a/packages/twenty-server/src/engine/metadata-modules/serverless-function/serverless-function.service.ts b/packages/twenty-server/src/engine/metadata-modules/serverless-function/serverless-function.service.ts index f1d3734926a..b90050e36de 100644 --- a/packages/twenty-server/src/engine/metadata-modules/serverless-function/serverless-function.service.ts +++ b/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, + }, }, }); diff --git a/packages/twenty-server/src/engine/subscriptions/constants/on-db-event-trigger.ts b/packages/twenty-server/src/engine/subscriptions/constants/on-db-event-trigger.ts deleted file mode 100644 index 3be4f4d989e..00000000000 --- a/packages/twenty-server/src/engine/subscriptions/constants/on-db-event-trigger.ts +++ /dev/null @@ -1 +0,0 @@ -export const ON_DB_EVENT_TRIGGER = 'onDbEvent'; 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 new file mode 100644 index 00000000000..26ff6acc169 --- /dev/null +++ b/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', +} diff --git a/packages/twenty-server/src/engine/subscriptions/subscription.service.ts b/packages/twenty-server/src/engine/subscriptions/subscription.service.ts new file mode 100644 index 00000000000..f9f5828b2eb --- /dev/null +++ b/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({ + channel, + payload, + workspaceId, + }: { + channel: SubscriptionChannel; + payload: T; + workspaceId: string; + }): Promise { + const client = this.redisClient.getPubSubClient(); + + await client.publish( + this.getSubscriptionChannel({ channel, workspaceId }), + payload, + ); + } +} diff --git a/packages/twenty-server/src/engine/subscriptions/subscriptions.module.ts b/packages/twenty-server/src/engine/subscriptions/subscriptions.module.ts index cf4afae38e4..72d334b13ab 100644 --- a/packages/twenty-server/src/engine/subscriptions/subscriptions.module.ts +++ b/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 {} diff --git a/packages/twenty-server/src/engine/subscriptions/subscriptions.service.ts b/packages/twenty-server/src/engine/subscriptions/subscriptions.service.ts deleted file mode 100644 index d8ef4da31ed..00000000000 --- a/packages/twenty-server/src/engine/subscriptions/subscriptions.service.ts +++ /dev/null @@ -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, - ): Promise { - 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 }), - }, - }); - } - } -} 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 f81fe0fee4a..f56f2a4a93c 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,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 {} diff --git a/packages/twenty-server/src/engine/subscriptions/subscriptions.resolver.ts b/packages/twenty-server/src/engine/workspace-event-emitter/workspace-event-emitter.resolver.ts similarity index 67% rename from packages/twenty-server/src/engine/subscriptions/subscriptions.resolver.ts rename to packages/twenty-server/src/engine/workspace-event-emitter/workspace-event-emitter.resolver.ts index 91628e1085b..4e94571da6c 100644 --- a/packages/twenty-server/src/engine/subscriptions/subscriptions.resolver.ts +++ b/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, + }); } } 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 new file mode 100644 index 00000000000..4083bf0f93d --- /dev/null +++ b/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, + ): Promise { + 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 }), + }, + }, + }); + } + } +}