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 d46582c9f5f..47d88ad41b4 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 @@ -10,16 +10,15 @@ import { type ObjectRecordEvent } from 'src/engine/core-modules/event-emitter/ty import { type ObjectRecordNonDestructiveEvent } from 'src/engine/core-modules/event-emitter/types/object-record-non-destructive-event'; import { type ObjectRecordRestoreEvent } from 'src/engine/core-modules/event-emitter/types/object-record-restore.event'; import { type ObjectRecordUpdateEvent } from 'src/engine/core-modules/event-emitter/types/object-record-update.event'; -import { FeatureFlagService } from 'src/engine/core-modules/feature-flag/services/feature-flag.service'; import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator'; import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants'; 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 { type ObjectRecordEventForWebhook } from 'src/engine/core-modules/webhook/types/object-record-event-for-webhook.type'; 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.type'; +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'; @Injectable() export class EntityEventsToDbListener { @@ -31,7 +30,6 @@ export class EntityEventsToDbListener { @InjectMessageQueue(MessageQueue.triggerQueue) private readonly triggerQueueService: MessageQueueService, private readonly subscriptionsService: SubscriptionsService, - private readonly featureFlagService: FeatureFlagService, ) {} @OnDatabaseBatchEvent('*', DatabaseEventAction.CREATED) @@ -67,26 +65,21 @@ export class EntityEventsToDbListener { batchEvent: WorkspaceEventBatch, action: DatabaseEventAction, ) { - const auditLogsEvents = batchEvent.events.filter( - (event) => event.objectMetadata?.isAuditLogged, - ); + const isAuditLogBatchEvent = batchEvent.objectMetadata?.isAuditLogged; - const batchEventEventsForWebhook: ObjectRecordEventForWebhook[] = - batchEvent.events.map((event) => ({ - ...event, - objectMetadata: { - id: event.objectMetadata.id, - nameSingular: event.objectMetadata.nameSingular, - }, - })); + const batchEventForWebhook = { + ...batchEvent, + objectMetadata: { + id: batchEvent.objectMetadata.id, + nameSingular: batchEvent.objectMetadata.nameSingular, + }, + }; const promises = [ this.subscriptionsService.publish(batchEvent), - this.webhookQueueService.add< - WorkspaceEventBatch - >( + this.webhookQueueService.add>( CallWebhookJobsJob.name, - { ...batchEvent, events: batchEventEventsForWebhook }, + batchEventForWebhook, { retryLimit: 3, }, @@ -101,14 +94,11 @@ export class EntityEventsToDbListener { ), ); - if (auditLogsEvents.length > 0) { + if (isAuditLogBatchEvent) { promises.push( this.entityEventsToDbQueueService.add>( CreateAuditLogFromInternalEvent.name, - { - ...batchEvent, - events: auditLogsEvents, - }, + batchEvent, ), ); @@ -116,10 +106,7 @@ export class EntityEventsToDbListener { promises.push( this.entityEventsToDbQueueService.add< WorkspaceEventBatch - >(UpsertTimelineActivityFromInternalEvent.name, { - ...batchEvent, - events: auditLogsEvents, - }), + >(UpsertTimelineActivityFromInternalEvent.name, batchEvent), ); } } diff --git a/packages/twenty-server/src/engine/api/graphql/workspace-query-runner/listeners/telemetry.listener.ts b/packages/twenty-server/src/engine/api/graphql/workspace-query-runner/listeners/telemetry.listener.ts index 268bfb49b20..93546017513 100644 --- a/packages/twenty-server/src/engine/api/graphql/workspace-query-runner/listeners/telemetry.listener.ts +++ b/packages/twenty-server/src/engine/api/graphql/workspace-query-runner/listeners/telemetry.listener.ts @@ -6,7 +6,7 @@ import { AuditService } from 'src/engine/core-modules/audit/services/audit.servi import { USER_SIGNUP_EVENT } from 'src/engine/core-modules/audit/utils/events/workspace-event/user/user-signup'; import { type ObjectRecordCreateEvent } from 'src/engine/core-modules/event-emitter/types/object-record-create.event'; import { TelemetryService } from 'src/engine/core-modules/telemetry/telemetry.service'; -import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type'; +import { CustomWorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/custom-workspace-batch-event.type'; @Injectable() export class TelemetryListener { @@ -17,7 +17,7 @@ export class TelemetryListener { @OnCustomBatchEvent(USER_SIGNUP_EVENT_NAME) async handleUserSignup( - payload: WorkspaceEventBatch, + payload: CustomWorkspaceEventBatch, ) { await Promise.all( payload.events.map(async (eventPayload) => { diff --git a/packages/twenty-server/src/engine/api/graphql/workspace-query-runner/workspace-query-runner.module.ts b/packages/twenty-server/src/engine/api/graphql/workspace-query-runner/workspace-query-runner.module.ts index ede72c6faf6..921fbf01f0c 100644 --- a/packages/twenty-server/src/engine/api/graphql/workspace-query-runner/workspace-query-runner.module.ts +++ b/packages/twenty-server/src/engine/api/graphql/workspace-query-runner/workspace-query-runner.module.ts @@ -8,7 +8,6 @@ import { WorkspaceQueryHookModule } from 'src/engine/api/graphql/workspace-query import { AuditModule } from 'src/engine/core-modules/audit/audit.module'; import { AuthModule } from 'src/engine/core-modules/auth/auth.module'; import { FeatureFlag } from 'src/engine/core-modules/feature-flag/feature-flag.entity'; -import { FeatureFlagModule } from 'src/engine/core-modules/feature-flag/feature-flag.module'; import { FileModule } from 'src/engine/core-modules/file/file.module'; import { RecordPositionModule } from 'src/engine/core-modules/record-position/record-position.module'; import { RecordTransformerModule } from 'src/engine/core-modules/record-transformer/record-transformer.module'; @@ -28,7 +27,6 @@ import { EntityEventsToDbListener } from './listeners/entity-events-to-db.listen AuditModule, TelemetryModule, FileModule, - FeatureFlagModule, RecordTransformerModule, RecordPositionModule, SubscriptionsModule, diff --git a/packages/twenty-server/src/engine/core-modules/audit/jobs/create-audit-log-from-internal-event.ts b/packages/twenty-server/src/engine/core-modules/audit/jobs/create-audit-log-from-internal-event.ts index e8db5253377..a7ec5dde9c8 100644 --- a/packages/twenty-server/src/engine/core-modules/audit/jobs/create-audit-log-from-internal-event.ts +++ b/packages/twenty-server/src/engine/core-modules/audit/jobs/create-audit-log-from-internal-event.ts @@ -7,7 +7,7 @@ import { type ObjectRecordEvent } from 'src/engine/core-modules/event-emitter/ty import { Process } from 'src/engine/core-modules/message-queue/decorators/process.decorator'; import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.decorator'; import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants'; -import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type'; +import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event-batch.type'; @Processor(MessageQueue.entityEventsToDbQueue) export class CreateAuditLogFromInternalEvent { @@ -37,25 +37,25 @@ export class CreateAuditLogFromInternalEvent { auditService.createObjectEvent(OBJECT_RECORD_UPDATED_EVENT, { ...eventProperties, recordId: eventData.recordId, - objectMetadataId: eventData.objectMetadata.id, + objectMetadataId: workspaceEventBatch.objectMetadata.id, }); } else if (workspaceEventBatch.name.endsWith('.created')) { auditService.createObjectEvent(OBJECT_RECORD_CREATED_EVENT, { ...eventProperties, recordId: eventData.recordId, - objectMetadataId: eventData.objectMetadata.id, + objectMetadataId: workspaceEventBatch.objectMetadata.id, }); } else if (workspaceEventBatch.name.endsWith('.deleted')) { auditService.createObjectEvent(OBJECT_RECORD_DELETED_EVENT, { ...eventProperties, recordId: eventData.recordId, - objectMetadataId: eventData.objectMetadata.id, + objectMetadataId: workspaceEventBatch.objectMetadata.id, }); } else if (workspaceEventBatch.name.endsWith('.upserted')) { auditService.createObjectEvent(OBJECT_RECORD_UPSERTED_EVENT, { ...eventProperties, recordId: eventData.recordId, - objectMetadataId: eventData.objectMetadata.id, + objectMetadataId: workspaceEventBatch.objectMetadata.id, }); } } diff --git a/packages/twenty-server/src/engine/core-modules/billing/listeners/billing-feature-used.listener.ts b/packages/twenty-server/src/engine/core-modules/billing/listeners/billing-feature-used.listener.ts index 6cca7861fb4..9fe7d9afc70 100644 --- a/packages/twenty-server/src/engine/core-modules/billing/listeners/billing-feature-used.listener.ts +++ b/packages/twenty-server/src/engine/core-modules/billing/listeners/billing-feature-used.listener.ts @@ -2,12 +2,14 @@ import { Injectable } from '@nestjs/common'; +import { isDefined } from 'twenty-shared/utils'; + import { OnCustomBatchEvent } from 'src/engine/api/graphql/graphql-query-runner/decorators/on-custom-batch-event.decorator'; import { BILLING_FEATURE_USED } from 'src/engine/core-modules/billing/constants/billing-feature-used.constant'; import { BillingUsageService } from 'src/engine/core-modules/billing/services/billing-usage.service'; import { type BillingUsageEvent } from 'src/engine/core-modules/billing/types/billing-usage-event.type'; import { TwentyConfigService } from 'src/engine/core-modules/twenty-config/twenty-config.service'; -import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type'; +import { CustomWorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/custom-workspace-batch-event.type'; @Injectable() export class BillingFeatureUsedListener { @@ -18,8 +20,12 @@ export class BillingFeatureUsedListener { @OnCustomBatchEvent(BILLING_FEATURE_USED) async handleBillingFeatureUsedEvent( - payload: WorkspaceEventBatch, + payload: CustomWorkspaceEventBatch, ) { + if (!isDefined(payload.workspaceId)) { + return; + } + if (!this.twentyConfigService.get('IS_BILLING_ENABLED')) { return; } diff --git a/packages/twenty-server/src/engine/core-modules/billing/listeners/billing-workspace-member.listener.ts b/packages/twenty-server/src/engine/core-modules/billing/listeners/billing-workspace-member.listener.ts index 90382dabb03..aa0cb6da4dc 100644 --- a/packages/twenty-server/src/engine/core-modules/billing/listeners/billing-workspace-member.listener.ts +++ b/packages/twenty-server/src/engine/core-modules/billing/listeners/billing-workspace-member.listener.ts @@ -13,7 +13,7 @@ import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decora import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants'; import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service'; import { TwentyConfigService } from 'src/engine/core-modules/twenty-config/twenty-config.service'; -import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type'; +import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event-batch.type'; import { type WorkspaceMemberWorkspaceEntity } from 'src/modules/workspace-member/standard-objects/workspace-member.workspace-entity'; @Injectable() diff --git a/packages/twenty-server/src/engine/core-modules/event-emitter/types/object-record.base.event.ts b/packages/twenty-server/src/engine/core-modules/event-emitter/types/object-record.base.event.ts index 3f602b1fb32..414839bbb06 100644 --- a/packages/twenty-server/src/engine/core-modules/event-emitter/types/object-record.base.event.ts +++ b/packages/twenty-server/src/engine/core-modules/event-emitter/types/object-record.base.event.ts @@ -1,5 +1,4 @@ import { type ObjectRecordDiff } from 'src/engine/core-modules/event-emitter/types/object-record-diff'; -import { type ObjectMetadataEntity } from 'src/engine/metadata-modules/object-metadata/object-metadata.entity'; type Properties = { updatedFields?: string[]; @@ -12,6 +11,5 @@ export class ObjectRecordBaseEvent { recordId: string; userId?: string; workspaceMemberId?: string; - objectMetadata: Omit; properties: Properties; } diff --git a/packages/twenty-server/src/engine/core-modules/file/listeners/file-attachment.listener.ts b/packages/twenty-server/src/engine/core-modules/file/listeners/file-attachment.listener.ts index 4ebe4ad0c48..05e7a641d2d 100644 --- a/packages/twenty-server/src/engine/core-modules/file/listeners/file-attachment.listener.ts +++ b/packages/twenty-server/src/engine/core-modules/file/listeners/file-attachment.listener.ts @@ -10,7 +10,7 @@ import { import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator'; import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants'; import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service'; -import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type'; +import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event-batch.type'; import { type AttachmentWorkspaceEntity } from 'src/modules/attachment/standard-objects/attachment.workspace-entity'; @Injectable() diff --git a/packages/twenty-server/src/engine/core-modules/file/listeners/file-workspace-member.listener.ts b/packages/twenty-server/src/engine/core-modules/file/listeners/file-workspace-member.listener.ts index 085eca64156..e280edc86a6 100644 --- a/packages/twenty-server/src/engine/core-modules/file/listeners/file-workspace-member.listener.ts +++ b/packages/twenty-server/src/engine/core-modules/file/listeners/file-workspace-member.listener.ts @@ -10,7 +10,7 @@ import { import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator'; import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants'; import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service'; -import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type'; +import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event-batch.type'; import { type WorkspaceMemberWorkspaceEntity } from 'src/modules/workspace-member/standard-objects/workspace-member.workspace-entity'; @Injectable() diff --git a/packages/twenty-server/src/engine/core-modules/webhook/jobs/call-webhook-jobs.job.ts b/packages/twenty-server/src/engine/core-modules/webhook/jobs/call-webhook-jobs.job.ts index 9b4250fc465..b6c6c1b75b5 100644 --- a/packages/twenty-server/src/engine/core-modules/webhook/jobs/call-webhook-jobs.job.ts +++ b/packages/twenty-server/src/engine/core-modules/webhook/jobs/call-webhook-jobs.job.ts @@ -1,6 +1,6 @@ import { Logger } from '@nestjs/common'; -import { isDefined } from 'twenty-shared/utils'; +import chunk from 'lodash.chunk'; import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator'; import { Process } from 'src/engine/core-modules/message-queue/decorators/process.decorator'; @@ -11,11 +11,12 @@ import { CallWebhookJob, type CallWebhookJobData, } from 'src/engine/core-modules/webhook/jobs/call-webhook.job'; -import { type ObjectRecordEventForWebhook } from 'src/engine/core-modules/webhook/types/object-record-event-for-webhook.type'; import { WebhookService } from 'src/engine/core-modules/webhook/webhook.service'; -import { type ObjectMetadataEntity } from 'src/engine/metadata-modules/object-metadata/object-metadata.entity'; -import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type'; -import { removeSecretFromWebhookRecord } from 'src/utils/remove-secret-from-webhook-record'; +import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event-batch.type'; +import type { ObjectRecordEvent } from 'src/engine/core-modules/event-emitter/types/object-record-event.event'; +import { transformEventBatchToWebhookEvents } from 'src/engine/core-modules/webhook/utils/transform-event-batch-to-webhook-events'; + +const WEBHOOK_JOBS_CHUNK_SIZE = 20; @Processor(MessageQueue.webhookQueue) export class CallWebhookJobsJob { @@ -28,7 +29,7 @@ export class CallWebhookJobsJob { @Process(CallWebhookJobsJob.name) async handle( - workspaceEventBatch: WorkspaceEventBatch, + workspaceEventBatch: WorkspaceEventBatch, ): Promise { // If you change that function, double check it does not break Zapier // trigger in packages/twenty-zapier/src/triggers/trigger_record.ts @@ -47,51 +48,19 @@ export class CallWebhookJobsJob { ], ); - for (const eventData of workspaceEventBatch.events) { - const eventName = workspaceEventBatch.name; - const objectMetadata: Pick = - { - id: eventData.objectMetadata.id, - nameSingular: eventData.objectMetadata.nameSingular, - }; - const workspaceId = workspaceEventBatch.workspaceId; - const record = - 'after' in eventData.properties && isDefined(eventData.properties.after) - ? eventData.properties.after - : 'before' in eventData.properties && - isDefined(eventData.properties.before) - ? eventData.properties.before - : {}; - const updatedFields = - 'updatedFields' in eventData.properties - ? eventData.properties.updatedFields - : undefined; + const webhookEvents = transformEventBatchToWebhookEvents({ + workspaceEventBatch, + webhooks, + }); - const isWebhookEvent = nameSingular === 'webhook'; - const sanitizedRecord = removeSecretFromWebhookRecord( - record, - isWebhookEvent, + const webhookEventsChunks = chunk(webhookEvents, WEBHOOK_JOBS_CHUNK_SIZE); + + for (const webhookEventsChunk of webhookEventsChunks) { + await this.messageQueueService.add( + CallWebhookJob.name, + webhookEventsChunk, + { retryLimit: 3 }, ); - - webhooks.forEach((webhook) => { - const webhookData = { - targetUrl: webhook.targetUrl, - secret: webhook.secret, - eventName, - objectMetadata, - workspaceId, - webhookId: webhook.id, - eventDate: new Date(), - record: sanitizedRecord, - ...(updatedFields && { updatedFields }), - }; - - this.messageQueueService.add( - CallWebhookJob.name, - webhookData, - { retryLimit: 3 }, - ); - }); } } } diff --git a/packages/twenty-server/src/engine/core-modules/webhook/jobs/call-webhook.job.ts b/packages/twenty-server/src/engine/core-modules/webhook/jobs/call-webhook.job.ts index 7c04cc3080e..640380d6138 100644 --- a/packages/twenty-server/src/engine/core-modules/webhook/jobs/call-webhook.job.ts +++ b/packages/twenty-server/src/engine/core-modules/webhook/jobs/call-webhook.job.ts @@ -42,7 +42,15 @@ export class CallWebhookJob { } @Process(CallWebhookJob.name) - async handle(data: CallWebhookJobData): Promise { + async handle(webhookJobEvents: CallWebhookJobData[]): Promise { + await Promise.all( + webhookJobEvents.map( + async (webhookJobEvent) => await this.callWebhook(webhookJobEvent), + ), + ); + } + + private async callWebhook(data: CallWebhookJobData): Promise { const commonPayload = { url: data.targetUrl, webhookId: data.webhookId, diff --git a/packages/twenty-server/src/engine/core-modules/webhook/types/object-record-event-for-webhook.type.ts b/packages/twenty-server/src/engine/core-modules/webhook/types/object-record-event-for-webhook.type.ts deleted file mode 100644 index 82d687f1605..00000000000 --- a/packages/twenty-server/src/engine/core-modules/webhook/types/object-record-event-for-webhook.type.ts +++ /dev/null @@ -1,9 +0,0 @@ -import { type ObjectRecordEvent } from 'src/engine/core-modules/event-emitter/types/object-record-event.event'; -import { type ObjectMetadataEntity } from 'src/engine/metadata-modules/object-metadata/object-metadata.entity'; - -export type ObjectRecordEventForWebhook = Omit< - ObjectRecordEvent, - 'objectMetadata' -> & { - objectMetadata: Pick; -}; diff --git a/packages/twenty-server/src/engine/core-modules/webhook/types/workspace-event-batch-for-webhook.type.ts b/packages/twenty-server/src/engine/core-modules/webhook/types/workspace-event-batch-for-webhook.type.ts new file mode 100644 index 00000000000..49f1889f30d --- /dev/null +++ b/packages/twenty-server/src/engine/core-modules/webhook/types/workspace-event-batch-for-webhook.type.ts @@ -0,0 +1,9 @@ +import { type ObjectMetadataEntity } from 'src/engine/metadata-modules/object-metadata/object-metadata.entity'; +import { type WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event-batch.type'; + +export type WorkspaceEventBatchForWebhook = Omit< + WorkspaceEventBatch, + 'objectMetadata' +> & { + objectMetadata: Pick; +}; diff --git a/packages/twenty-server/src/engine/core-modules/webhook/utils/__tests__/transform-event-batch-to-webhook-events.spec.ts b/packages/twenty-server/src/engine/core-modules/webhook/utils/__tests__/transform-event-batch-to-webhook-events.spec.ts new file mode 100644 index 00000000000..1d117ee0b5a --- /dev/null +++ b/packages/twenty-server/src/engine/core-modules/webhook/utils/__tests__/transform-event-batch-to-webhook-events.spec.ts @@ -0,0 +1,235 @@ +import type { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event-batch.type'; +import type { ObjectRecordEvent } from 'src/engine/core-modules/event-emitter/types/object-record-event.event'; +import type { Webhook } from 'src/engine/core-modules/webhook/webhook.entity'; +import { transformEventBatchToWebhookEvents } from 'src/engine/core-modules/webhook/utils/transform-event-batch-to-webhook-events'; +import { getMockObjectMetadataEntity } from 'src/utils/__test__/get-object-metadata-entity.mock'; + +const mockObjectMetadata = getMockObjectMetadataEntity({ + id: 'id', + nameSingular: 'nameSingular', + namePlural: 'namePlural', + workspaceId: 'workspaceId', +}); + +describe('transformEventBatchToWebhookEvents', () => { + it('should transform properly', () => { + const workspaceEventBatch = { + workspaceId: 'workspaceId', + objectMetadata: mockObjectMetadata, + name: 'objectNameSingular.created', + events: [ + { + recordId: 'recordId-1', + properties: { + after: { + id: 'id-1', + nameSingular: 'nameSingular-1', + }, + }, + }, + { + recordId: 'recordId-2', + properties: { + before: { + id: 'id-2', + nameSingular: 'nameSingular-2', + }, + }, + }, + { + recordId: 'recordId-3', + properties: { + after: { + id: 'id-3', + nameSingular: 'nameSingular-3', + secret: 'secret-3', + }, + updatedFields: ['nameSingular'], + }, + }, + ], + } as WorkspaceEventBatch; + + const webhooks = [ + { + id: 'webhook-id', + targetUrl: 'targetUrl', + secret: 'secret', + }, + { + id: 'webhook-id-2', + targetUrl: 'targetUrl-2', + secret: 'secret-2', + }, + ] as Webhook[]; + + const result = transformEventBatchToWebhookEvents({ + workspaceEventBatch, + webhooks, + }); + const expectedResultWithoutEventDate = [ + { + targetUrl: 'targetUrl', + eventName: 'objectNameSingular.created', + objectMetadata: { + id: mockObjectMetadata.id, + nameSingular: mockObjectMetadata.nameSingular, + }, + workspaceId: 'workspaceId', + webhookId: 'webhook-id', + record: { + id: 'id-1', + nameSingular: 'nameSingular-1', + }, + secret: 'secret', + }, + { + targetUrl: 'targetUrl', + eventName: 'objectNameSingular.created', + objectMetadata: { + id: mockObjectMetadata.id, + nameSingular: mockObjectMetadata.nameSingular, + }, + workspaceId: 'workspaceId', + webhookId: 'webhook-id', + record: { + id: 'id-2', + nameSingular: 'nameSingular-2', + }, + secret: 'secret', + }, + { + targetUrl: 'targetUrl', + eventName: 'objectNameSingular.created', + objectMetadata: { + id: mockObjectMetadata.id, + nameSingular: mockObjectMetadata.nameSingular, + }, + workspaceId: 'workspaceId', + webhookId: 'webhook-id', + record: { + id: 'id-3', + nameSingular: 'nameSingular-3', + secret: 'secret-3', + }, + updatedFields: ['nameSingular'], + secret: 'secret', + }, + { + targetUrl: 'targetUrl-2', + eventName: 'objectNameSingular.created', + objectMetadata: { + id: mockObjectMetadata.id, + nameSingular: mockObjectMetadata.nameSingular, + }, + workspaceId: 'workspaceId', + webhookId: 'webhook-id-2', + record: { + id: 'id-1', + nameSingular: 'nameSingular-1', + }, + secret: 'secret-2', + }, + { + targetUrl: 'targetUrl-2', + eventName: 'objectNameSingular.created', + objectMetadata: { + id: mockObjectMetadata.id, + nameSingular: mockObjectMetadata.nameSingular, + }, + workspaceId: 'workspaceId', + webhookId: 'webhook-id-2', + record: { + id: 'id-2', + nameSingular: 'nameSingular-2', + }, + secret: 'secret-2', + }, + { + targetUrl: 'targetUrl-2', + eventName: 'objectNameSingular.created', + objectMetadata: { + id: mockObjectMetadata.id, + nameSingular: mockObjectMetadata.nameSingular, + }, + workspaceId: 'workspaceId', + webhookId: 'webhook-id-2', + record: { + id: 'id-3', + nameSingular: 'nameSingular-3', + secret: 'secret-3', + }, + updatedFields: ['nameSingular'], + secret: 'secret-2', + }, + ]; + + const resultWithoutEventDate = result.map((event) => { + const { eventDate: _, ...eventWithoutEventDate } = event; + + return eventWithoutEventDate; + }); + + expect(resultWithoutEventDate).toEqual(expectedResultWithoutEventDate); + }); + + it('should sanitize records properly', () => { + const workspaceEventBatch = { + workspaceId: 'workspaceId', + objectMetadata: mockObjectMetadata, + name: 'webhook.created', + events: [ + { + recordId: 'recordId-1', + properties: { + after: { + id: 'id-1', + targetUrl: 'targetUrl-1', + secret: 'secret-1', + }, + }, + }, + ], + } as WorkspaceEventBatch; + + const webhooks = [ + { + id: 'webhook-id', + targetUrl: 'targetUrl', + secret: 'secret', + }, + ] as Webhook[]; + + const result = transformEventBatchToWebhookEvents({ + workspaceEventBatch, + webhooks, + }); + + const expectedResultWithoutEventDate = [ + { + targetUrl: 'targetUrl', + eventName: 'webhook.created', + objectMetadata: { + id: mockObjectMetadata.id, + nameSingular: mockObjectMetadata.nameSingular, + }, + workspaceId: 'workspaceId', + webhookId: 'webhook-id', + record: { + id: 'id-1', + targetUrl: 'targetUrl-1', + // No secret + }, + secret: 'secret', + }, + ]; + + const resultWithoutEventDate = result.map((event) => { + const { eventDate: _, ...eventWithoutEventDate } = event; + + return eventWithoutEventDate; + }); + + expect(resultWithoutEventDate).toEqual(expectedResultWithoutEventDate); + }); +}); diff --git a/packages/twenty-server/src/engine/core-modules/webhook/utils/__tests__/transform-event-to-webhook-event.spec.ts b/packages/twenty-server/src/engine/core-modules/webhook/utils/__tests__/transform-event-to-webhook-event.spec.ts new file mode 100644 index 00000000000..b2b32bccf0d --- /dev/null +++ b/packages/twenty-server/src/engine/core-modules/webhook/utils/__tests__/transform-event-to-webhook-event.spec.ts @@ -0,0 +1,34 @@ +import { transformEventToWebhookEvent } from 'src/engine/core-modules/webhook/utils/transform-event-to-webhook-event'; +import type { ObjectRecordEvent } from 'src/engine/core-modules/event-emitter/types/object-record-event.event'; + +describe('transformEventToWebhookEvent', () => { + it('should transform event to webhook event', () => { + const record = { + recordId: 'recordId', + properties: { + after: { + id: 'id', + nameSingular: 'nameSingular', + secret: 'secret', + }, + updatedFields: ['nameSingular'], + }, + } as ObjectRecordEvent; + + const expectedResult = { + record: { + id: 'id', + nameSingular: 'nameSingular', + secret: 'secret', + }, + updatedFields: ['nameSingular'], + }; + + expect( + transformEventToWebhookEvent({ + eventName: 'nameSingular.created', + event: record, + }), + ).toEqual(expectedResult); + }); +}); diff --git a/packages/twenty-server/src/engine/core-modules/webhook/utils/transform-event-batch-to-webhook-events.ts b/packages/twenty-server/src/engine/core-modules/webhook/utils/transform-event-batch-to-webhook-events.ts new file mode 100644 index 00000000000..cd1dac2a69c --- /dev/null +++ b/packages/twenty-server/src/engine/core-modules/webhook/utils/transform-event-batch-to-webhook-events.ts @@ -0,0 +1,49 @@ +import { type WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event-batch.type'; +import { type CallWebhookJobData } from 'src/engine/core-modules/webhook/jobs/call-webhook.job'; +import { type Webhook } from 'src/engine/core-modules/webhook/webhook.entity'; +import type { ObjectRecordEvent } from 'src/engine/core-modules/event-emitter/types/object-record-event.event'; +import { transformEventToWebhookEvent } from 'src/engine/core-modules/webhook/utils/transform-event-to-webhook-event'; + +export const transformEventBatchToWebhookEvents = ({ + workspaceEventBatch, + webhooks, +}: { + workspaceEventBatch: WorkspaceEventBatch; + webhooks: Webhook[]; +}): CallWebhookJobData[] => { + const result: CallWebhookJobData[] = []; + + for (const webhook of webhooks) { + const targetUrl = webhook.targetUrl; + const eventName = workspaceEventBatch.name; + const objectMetadataForWebhook = { + id: workspaceEventBatch.objectMetadata.id, + nameSingular: workspaceEventBatch.objectMetadata.nameSingular, + }; + const workspaceId = workspaceEventBatch.workspaceId; + const webhookId = webhook.id; + const eventDate = new Date(); + const secret = webhook.secret; + + for (const eventData of workspaceEventBatch.events) { + const { record, updatedFields } = transformEventToWebhookEvent({ + eventName: workspaceEventBatch.name, + event: eventData, + }); + + result.push({ + targetUrl, + eventName, + objectMetadata: objectMetadataForWebhook, + workspaceId, + webhookId, + eventDate, + record, + ...(updatedFields && { updatedFields }), + secret, + }); + } + } + + return result; +}; diff --git a/packages/twenty-server/src/engine/core-modules/webhook/utils/transform-event-to-webhook-event.ts b/packages/twenty-server/src/engine/core-modules/webhook/utils/transform-event-to-webhook-event.ts new file mode 100644 index 00000000000..9d7eb45de52 --- /dev/null +++ b/packages/twenty-server/src/engine/core-modules/webhook/utils/transform-event-to-webhook-event.ts @@ -0,0 +1,34 @@ +import { isDefined } from 'twenty-shared/utils'; + +import type { ObjectRecordEvent } from 'src/engine/core-modules/event-emitter/types/object-record-event.event'; +import { removeSecretFromWebhookRecord } from 'src/utils/remove-secret-from-webhook-record'; + +export const transformEventToWebhookEvent = ({ + eventName, + event, +}: { + eventName: string; + event: ObjectRecordEvent; +}) => { + const [nameSingular, _] = eventName.split('.'); + + const record = + 'after' in event.properties && isDefined(event.properties.after) + ? event.properties.after + : 'before' in event.properties && isDefined(event.properties.before) + ? event.properties.before + : {}; + const updatedFields = + 'updatedFields' in event.properties + ? event.properties.updatedFields + : undefined; + + const isWebhookEvent = nameSingular === 'webhook'; + + const sanitizedRecord = removeSecretFromWebhookRecord(record, isWebhookEvent); + + return { + record: sanitizedRecord, + ...(updatedFields && { updatedFields }), + }; +}; diff --git a/packages/twenty-server/src/engine/core-modules/workspace/workspace-workspace-member.listener.ts b/packages/twenty-server/src/engine/core-modules/workspace/workspace-workspace-member.listener.ts index 0725c0d4ec2..df47d0e45c7 100644 --- a/packages/twenty-server/src/engine/core-modules/workspace/workspace-workspace-member.listener.ts +++ b/packages/twenty-server/src/engine/core-modules/workspace/workspace-workspace-member.listener.ts @@ -10,7 +10,7 @@ import { type ObjectRecordUpdateEvent } from 'src/engine/core-modules/event-emit import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator'; import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants'; import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service'; -import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type'; +import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event-batch.type'; import { type WorkspaceMemberWorkspaceEntity } from 'src/modules/workspace-member/standard-objects/workspace-member.workspace-entity'; import { OnDatabaseBatchEvent } from 'src/engine/api/graphql/graphql-query-runner/decorators/on-database-batch-event.decorator'; import { DatabaseEventAction } from 'src/engine/api/graphql/graphql-query-runner/enums/database-event-action'; diff --git a/packages/twenty-server/src/engine/metadata-modules/database-event-trigger/jobs/call-database-event-trigger-jobs.job.ts b/packages/twenty-server/src/engine/metadata-modules/database-event-trigger/jobs/call-database-event-trigger-jobs.job.ts index 3c90ccab7db..464b0ffe331 100644 --- a/packages/twenty-server/src/engine/metadata-modules/database-event-trigger/jobs/call-database-event-trigger-jobs.job.ts +++ b/packages/twenty-server/src/engine/metadata-modules/database-event-trigger/jobs/call-database-event-trigger-jobs.job.ts @@ -13,7 +13,7 @@ import { ServerlessFunctionTriggerJob, ServerlessFunctionTriggerJobData, } from 'src/engine/metadata-modules/serverless-function/jobs/serverless-function-trigger.job'; -import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type'; +import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event-batch.type'; @Processor(MessageQueue.triggerQueue) export class CallDatabaseEventTriggerJobsJob { diff --git a/packages/twenty-server/src/engine/subscriptions/subscriptions.service.ts b/packages/twenty-server/src/engine/subscriptions/subscriptions.service.ts index 14b16fad143..b709cb6ec62 100644 --- a/packages/twenty-server/src/engine/subscriptions/subscriptions.service.ts +++ b/packages/twenty-server/src/engine/subscriptions/subscriptions.service.ts @@ -1,11 +1,10 @@ import { Inject, Injectable } from '@nestjs/common'; import { RedisPubSub } from 'graphql-redis-subscriptions'; -import { isDefined } from 'twenty-shared/utils'; import { type ObjectRecordEvent } from 'src/engine/core-modules/event-emitter/types/object-record-event.event'; -import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type'; -import { removeSecretFromWebhookRecord } from 'src/utils/remove-secret-from-webhook-record'; +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'; @Injectable() export class SubscriptionsService { @@ -14,32 +13,20 @@ export class SubscriptionsService { async publish( workspaceEventBatch: WorkspaceEventBatch, ): Promise { - for (const eventData of workspaceEventBatch.events) { - const [nameSingular, operation] = workspaceEventBatch.name.split('.'); - const record = - 'after' in eventData.properties && isDefined(eventData.properties.after) - ? eventData.properties.after - : 'before' in eventData.properties && - isDefined(eventData.properties.before) - ? eventData.properties.before - : {}; - const updatedFields = - 'updatedFields' in eventData.properties - ? eventData.properties.updatedFields - : undefined; + const [nameSingular, operation] = workspaceEventBatch.name.split('.'); - const isWebhookEvent = nameSingular === 'webhook'; - const sanitizedRecord = removeSecretFromWebhookRecord( - record, - isWebhookEvent, - ); + for (const eventData of workspaceEventBatch.events) { + const { record, updatedFields } = transformEventToWebhookEvent({ + eventName: workspaceEventBatch.name, + event: eventData, + }); await this.pubSub.publish('onDbEvent', { onDbEvent: { action: operation, objectNameSingular: nameSingular, eventDate: new Date(), - record: sanitizedRecord, + record, ...(updatedFields && { updatedFields }), }, }); diff --git a/packages/twenty-server/src/engine/twenty-orm/entity-manager/workspace-entity-manager.ts b/packages/twenty-server/src/engine/twenty-orm/entity-manager/workspace-entity-manager.ts index a3e415016d6..5cbc8991620 100644 --- a/packages/twenty-server/src/engine/twenty-orm/entity-manager/workspace-entity-manager.ts +++ b/packages/twenty-server/src/engine/twenty-orm/entity-manager/workspace-entity-manager.ts @@ -56,6 +56,7 @@ import { WorkspaceRepository } from 'src/engine/twenty-orm/repository/workspace. import { formatData } from 'src/engine/twenty-orm/utils/format-data.util'; import { formatResult } from 'src/engine/twenty-orm/utils/format-result.util'; import { getObjectMetadataFromEntityTarget } from 'src/engine/twenty-orm/utils/get-object-metadata-from-entity-target.util'; +import { formatTwentyOrmEventToDatabaseBatchEvent } from 'src/engine/twenty-orm/utils/format-twenty-orm-event-to-database-batch-event.util'; type PermissionOptions = { shouldBypassPermissionChecks?: boolean; @@ -1195,22 +1196,26 @@ export class WorkspaceEntityManager extends EntityManager { (entity) => !beforeUpdateMapById[entity.id], ); - await this.internalContext.eventEmitterService.emitMutationEvent({ - action: DatabaseEventAction.UPDATED, - objectMetadataItem, - workspaceId: this.internalContext.workspaceId, - entities: updatedEntities, - beforeEntities: updatedEntities.map( - (entity) => beforeUpdateMapById[entity.id], - ), - }); + this.internalContext.eventEmitterService.emitDatabaseBatchEvent( + formatTwentyOrmEventToDatabaseBatchEvent({ + action: DatabaseEventAction.UPDATED, + objectMetadataItem, + workspaceId: this.internalContext.workspaceId, + entities: updatedEntities, + beforeEntities: updatedEntities.map( + (entity) => beforeUpdateMapById[entity.id], + ), + }), + ); - await this.internalContext.eventEmitterService.emitMutationEvent({ - action: DatabaseEventAction.CREATED, - objectMetadataItem, - workspaceId: this.internalContext.workspaceId, - entities: createdEntities, - }); + this.internalContext.eventEmitterService.emitDatabaseBatchEvent( + formatTwentyOrmEventToDatabaseBatchEvent({ + action: DatabaseEventAction.CREATED, + objectMetadataItem, + workspaceId: this.internalContext.workspaceId, + entities: createdEntities, + }), + ); const permissionCheckApplies = permissionOptionsFromArgs?.shouldBypassPermissionChecks !== true && @@ -1387,12 +1392,14 @@ export class WorkspaceEntityManager extends EntityManager { this.internalContext.objectMetadataMaps, ); - await this.internalContext.eventEmitterService.emitMutationEvent({ - action: DatabaseEventAction.DESTROYED, - objectMetadataItem, - workspaceId: this.internalContext.workspaceId, - entities: formattedResult, - }); + this.internalContext.eventEmitterService.emitDatabaseBatchEvent( + formatTwentyOrmEventToDatabaseBatchEvent({ + action: DatabaseEventAction.DESTROYED, + objectMetadataItem, + workspaceId: this.internalContext.workspaceId, + entities: formattedResult, + }), + ); return isEntityArray ? formattedResult : formattedResult[0]; } @@ -1500,12 +1507,14 @@ export class WorkspaceEntityManager extends EntityManager { this.internalContext.objectMetadataMaps, ); - await this.internalContext.eventEmitterService.emitMutationEvent({ - action: DatabaseEventAction.DELETED, - objectMetadataItem, - workspaceId: this.internalContext.workspaceId, - entities: formattedResult, - }); + this.internalContext.eventEmitterService.emitDatabaseBatchEvent( + formatTwentyOrmEventToDatabaseBatchEvent({ + action: DatabaseEventAction.DELETED, + objectMetadataItem, + workspaceId: this.internalContext.workspaceId, + entities: formattedResult, + }), + ); return isEntityArray ? formattedResult : formattedResult[0]; } @@ -1609,12 +1618,14 @@ export class WorkspaceEntityManager extends EntityManager { this.internalContext.objectMetadataMaps, ); - await this.internalContext.eventEmitterService.emitMutationEvent({ - action: DatabaseEventAction.RESTORED, - objectMetadataItem, - workspaceId: this.internalContext.workspaceId, - entities: formattedResult, - }); + this.internalContext.eventEmitterService.emitDatabaseBatchEvent( + formatTwentyOrmEventToDatabaseBatchEvent({ + action: DatabaseEventAction.RESTORED, + objectMetadataItem, + workspaceId: this.internalContext.workspaceId, + entities: formattedResult, + }), + ); return isEntityArray ? formattedResult : formattedResult[0]; } diff --git a/packages/twenty-server/src/engine/twenty-orm/repository/workspace-delete-query-builder.ts b/packages/twenty-server/src/engine/twenty-orm/repository/workspace-delete-query-builder.ts index cc77cbc8bf3..c9e3a4ec36b 100644 --- a/packages/twenty-server/src/engine/twenty-orm/repository/workspace-delete-query-builder.ts +++ b/packages/twenty-server/src/engine/twenty-orm/repository/workspace-delete-query-builder.ts @@ -27,6 +27,7 @@ import { applyTableAliasOnWhereCondition } from 'src/engine/twenty-orm/utils/app import { formatResult } from 'src/engine/twenty-orm/utils/format-result.util'; import { getObjectMetadataFromEntityTarget } from 'src/engine/twenty-orm/utils/get-object-metadata-from-entity-target.util'; import { computeTableName } from 'src/engine/utils/compute-table-name.util'; +import { formatTwentyOrmEventToDatabaseBatchEvent } from 'src/engine/twenty-orm/utils/format-twenty-orm-event-to-database-batch-event.util'; export class WorkspaceDeleteQueryBuilder< T extends ObjectLiteral, @@ -122,13 +123,15 @@ export class WorkspaceDeleteQueryBuilder< this.internalContext.objectMetadataMaps, ); - await this.internalContext.eventEmitterService.emitMutationEvent({ - action: DatabaseEventAction.DESTROYED, - objectMetadataItem: objectMetadata, - workspaceId: this.internalContext.workspaceId, - entities: formattedBefore, - authContext: this.authContext, - }); + this.internalContext.eventEmitterService.emitDatabaseBatchEvent( + formatTwentyOrmEventToDatabaseBatchEvent({ + action: DatabaseEventAction.DESTROYED, + objectMetadataItem: objectMetadata, + workspaceId: this.internalContext.workspaceId, + entities: formattedBefore, + authContext: this.authContext, + }), + ); return { raw: result.raw, diff --git a/packages/twenty-server/src/engine/twenty-orm/repository/workspace-insert-query-builder.ts b/packages/twenty-server/src/engine/twenty-orm/repository/workspace-insert-query-builder.ts index fb43a411c27..2692134705e 100644 --- a/packages/twenty-server/src/engine/twenty-orm/repository/workspace-insert-query-builder.ts +++ b/packages/twenty-server/src/engine/twenty-orm/repository/workspace-insert-query-builder.ts @@ -30,6 +30,7 @@ import { type WorkspaceUpdateQueryBuilder } from 'src/engine/twenty-orm/reposito import { formatData } from 'src/engine/twenty-orm/utils/format-data.util'; import { formatResult } from 'src/engine/twenty-orm/utils/format-result.util'; import { getObjectMetadataFromEntityTarget } from 'src/engine/twenty-orm/utils/get-object-metadata-from-entity-target.util'; +import { formatTwentyOrmEventToDatabaseBatchEvent } from 'src/engine/twenty-orm/utils/format-twenty-orm-event-to-database-batch-event.util'; export class WorkspaceInsertQueryBuilder< T extends ObjectLiteral, @@ -160,21 +161,25 @@ export class WorkspaceInsertQueryBuilder< this.internalContext.objectMetadataMaps, ); - await this.internalContext.eventEmitterService.emitMutationEvent({ - action: DatabaseEventAction.CREATED, - objectMetadataItem: objectMetadata, - workspaceId: this.internalContext.workspaceId, - entities: formattedResultForEvent, - authContext: this.authContext, - }); + this.internalContext.eventEmitterService.emitDatabaseBatchEvent( + formatTwentyOrmEventToDatabaseBatchEvent({ + action: DatabaseEventAction.CREATED, + objectMetadataItem: objectMetadata, + workspaceId: this.internalContext.workspaceId, + entities: formattedResultForEvent, + authContext: this.authContext, + }), + ); - await this.internalContext.eventEmitterService.emitMutationEvent({ - action: DatabaseEventAction.UPSERTED, - objectMetadataItem: objectMetadata, - workspaceId: this.internalContext.workspaceId, - entities: formattedResultForEvent, - authContext: this.authContext, - }); + this.internalContext.eventEmitterService.emitDatabaseBatchEvent( + formatTwentyOrmEventToDatabaseBatchEvent({ + action: DatabaseEventAction.UPSERTED, + objectMetadataItem: objectMetadata, + workspaceId: this.internalContext.workspaceId, + entities: formattedResultForEvent, + authContext: this.authContext, + }), + ); // TypeORM returns all entity columns for insertions const resultWithoutInsertionExtraColumns = !isDefined(result.raw) diff --git a/packages/twenty-server/src/engine/twenty-orm/repository/workspace-soft-delete-query-builder.ts b/packages/twenty-server/src/engine/twenty-orm/repository/workspace-soft-delete-query-builder.ts index 7a9b26aee47..698d722be95 100644 --- a/packages/twenty-server/src/engine/twenty-orm/repository/workspace-soft-delete-query-builder.ts +++ b/packages/twenty-server/src/engine/twenty-orm/repository/workspace-soft-delete-query-builder.ts @@ -23,6 +23,7 @@ import { type WorkspaceSelectQueryBuilder } from 'src/engine/twenty-orm/reposito import { type WorkspaceUpdateQueryBuilder } from 'src/engine/twenty-orm/repository/workspace-update-query-builder'; import { formatResult } from 'src/engine/twenty-orm/utils/format-result.util'; import { getObjectMetadataFromEntityTarget } from 'src/engine/twenty-orm/utils/get-object-metadata-from-entity-target.util'; +import { formatTwentyOrmEventToDatabaseBatchEvent } from 'src/engine/twenty-orm/utils/format-twenty-orm-event-to-database-batch-event.util'; export class WorkspaceSoftDeleteQueryBuilder< T extends ObjectLiteral, @@ -85,13 +86,15 @@ export class WorkspaceSoftDeleteQueryBuilder< this.internalContext.objectMetadataMaps, ); - await this.internalContext.eventEmitterService.emitMutationEvent({ - action: DatabaseEventAction.DELETED, - objectMetadataItem: objectMetadata, - workspaceId: this.internalContext.workspaceId, - entities: formattedAfter, - authContext: this.authContext, - }); + this.internalContext.eventEmitterService.emitDatabaseBatchEvent( + formatTwentyOrmEventToDatabaseBatchEvent({ + action: DatabaseEventAction.DELETED, + objectMetadataItem: objectMetadata, + workspaceId: this.internalContext.workspaceId, + entities: formattedAfter, + authContext: this.authContext, + }), + ); return { raw: after.raw, diff --git a/packages/twenty-server/src/engine/twenty-orm/repository/workspace-update-query-builder.ts b/packages/twenty-server/src/engine/twenty-orm/repository/workspace-update-query-builder.ts index 4fe9c411016..51d153a0d26 100644 --- a/packages/twenty-server/src/engine/twenty-orm/repository/workspace-update-query-builder.ts +++ b/packages/twenty-server/src/engine/twenty-orm/repository/workspace-update-query-builder.ts @@ -29,6 +29,7 @@ import { type WorkspaceSoftDeleteQueryBuilder } from 'src/engine/twenty-orm/repo import { formatData } from 'src/engine/twenty-orm/utils/format-data.util'; import { formatResult } from 'src/engine/twenty-orm/utils/format-result.util'; import { getObjectMetadataFromEntityTarget } from 'src/engine/twenty-orm/utils/get-object-metadata-from-entity-target.util'; +import { formatTwentyOrmEventToDatabaseBatchEvent } from 'src/engine/twenty-orm/utils/format-twenty-orm-event-to-database-batch-event.util'; export class WorkspaceUpdateQueryBuilder< T extends ObjectLiteral, @@ -152,23 +153,27 @@ export class WorkspaceUpdateQueryBuilder< this.internalContext.objectMetadataMaps, ); - await this.internalContext.eventEmitterService.emitMutationEvent({ - action: DatabaseEventAction.UPDATED, - objectMetadataItem: objectMetadata, - workspaceId: this.internalContext.workspaceId, - entities: formattedAfter, - beforeEntities: formattedBefore, - authContext: this.authContext, - }); + this.internalContext.eventEmitterService.emitDatabaseBatchEvent( + formatTwentyOrmEventToDatabaseBatchEvent({ + action: DatabaseEventAction.UPDATED, + objectMetadataItem: objectMetadata, + workspaceId: this.internalContext.workspaceId, + entities: formattedAfter, + beforeEntities: formattedBefore, + authContext: this.authContext, + }), + ); - await this.internalContext.eventEmitterService.emitMutationEvent({ - action: DatabaseEventAction.UPSERTED, - objectMetadataItem: objectMetadata, - workspaceId: this.internalContext.workspaceId, - entities: formattedAfter, - beforeEntities: formattedBefore, - authContext: this.authContext, - }); + this.internalContext.eventEmitterService.emitDatabaseBatchEvent( + formatTwentyOrmEventToDatabaseBatchEvent({ + action: DatabaseEventAction.UPSERTED, + objectMetadataItem: objectMetadata, + workspaceId: this.internalContext.workspaceId, + entities: formattedAfter, + beforeEntities: formattedBefore, + authContext: this.authContext, + }), + ); const formattedResult = formatResult( result.raw, @@ -293,23 +298,27 @@ export class WorkspaceUpdateQueryBuilder< this.internalContext.objectMetadataMaps, ); - await this.internalContext.eventEmitterService.emitMutationEvent({ - action: DatabaseEventAction.UPDATED, - objectMetadataItem: objectMetadata, - workspaceId: this.internalContext.workspaceId, - entities: formattedAfter, - beforeEntities: formattedBefore, - authContext: this.authContext, - }); + this.internalContext.eventEmitterService.emitDatabaseBatchEvent( + formatTwentyOrmEventToDatabaseBatchEvent({ + action: DatabaseEventAction.UPDATED, + objectMetadataItem: objectMetadata, + workspaceId: this.internalContext.workspaceId, + entities: formattedAfter, + beforeEntities: formattedBefore, + authContext: this.authContext, + }), + ); - await this.internalContext.eventEmitterService.emitMutationEvent({ - action: DatabaseEventAction.UPSERTED, - objectMetadataItem: objectMetadata, - workspaceId: this.internalContext.workspaceId, - entities: formattedAfter, - beforeEntities: formattedBefore, - authContext: this.authContext, - }); + this.internalContext.eventEmitterService.emitDatabaseBatchEvent( + formatTwentyOrmEventToDatabaseBatchEvent({ + action: DatabaseEventAction.UPSERTED, + objectMetadataItem: objectMetadata, + workspaceId: this.internalContext.workspaceId, + entities: formattedAfter, + beforeEntities: formattedBefore, + authContext: this.authContext, + }), + ); const formattedResults = formatResult( results.flatMap((result) => result.raw), diff --git a/packages/twenty-server/src/engine/twenty-orm/utils/format-twenty-orm-event-to-database-batch-event.util.ts b/packages/twenty-server/src/engine/twenty-orm/utils/format-twenty-orm-event-to-database-batch-event.util.ts new file mode 100644 index 00000000000..3a122882f9d --- /dev/null +++ b/packages/twenty-server/src/engine/twenty-orm/utils/format-twenty-orm-event-to-database-batch-event.util.ts @@ -0,0 +1,174 @@ +import { isDefined } from 'twenty-shared/utils'; + +import type { ObjectLiteral } from 'typeorm'; + +import { DatabaseEventAction } from 'src/engine/api/graphql/graphql-query-runner/enums/database-event-action'; +import type { ObjectMetadataItemWithFieldMaps } from 'src/engine/metadata-modules/types/object-metadata-item-with-field-maps'; +import type { AuthContext } from 'src/engine/core-modules/auth/types/auth-context.type'; +import { STANDARD_OBJECT_IDS } from 'src/engine/workspace-manager/workspace-sync-metadata/constants/standard-object-ids'; +import { ObjectRecordCreateEvent } from 'src/engine/core-modules/event-emitter/types/object-record-create.event'; +import { ObjectRecordUpdateEvent } from 'src/engine/core-modules/event-emitter/types/object-record-update.event'; +import { ObjectRecordDeleteEvent } from 'src/engine/core-modules/event-emitter/types/object-record-delete.event'; +import { ObjectRecordUpsertEvent } from 'src/engine/core-modules/event-emitter/types/object-record-upsert.event'; +import { objectRecordChangedValues } from 'src/engine/core-modules/event-emitter/utils/object-record-changed-values'; +import type { ObjectRecordDiff } from 'src/engine/core-modules/event-emitter/types/object-record-diff'; +import { ObjectRecordDestroyEvent } from 'src/engine/core-modules/event-emitter/types/object-record-destroy.event'; +import { type DatabaseBatchEventInput } from 'src/engine/workspace-event-emitter/workspace-event-emitter'; + +export const formatTwentyOrmEventToDatabaseBatchEvent = < + T extends ObjectLiteral, +>({ + action, + objectMetadataItem, + workspaceId, + authContext, + entities, + beforeEntities, +}: { + action: DatabaseEventAction; + objectMetadataItem: ObjectMetadataItemWithFieldMaps; + workspaceId: string; + authContext?: AuthContext; + entities: T | T[]; + beforeEntities?: T | T[]; +}): DatabaseBatchEventInput | undefined => { + if (objectMetadataItem.standardId === STANDARD_OBJECT_IDS.timelineActivity) { + return; + } + + const objectMetadataNameSingular = objectMetadataItem.nameSingular; + const fields = Object.values(objectMetadataItem.fieldsById ?? {}); + const entityArray = isDefined(entities) + ? Array.isArray(entities) + ? entities + : [entities] + : []; + let events: ( + | ObjectRecordCreateEvent + | ObjectRecordUpdateEvent + | ObjectRecordDeleteEvent + | ObjectRecordUpsertEvent + )[] = []; + + switch (action) { + case DatabaseEventAction.CREATED: + events = entityArray.map((after) => { + const event = new ObjectRecordCreateEvent(); + + event.userId = authContext?.user?.id; + event.recordId = after.id; + event.properties = { after }; + + return event; + }); + break; + case DatabaseEventAction.UPDATED: + events = entityArray + .map((after, idx) => { + if (!beforeEntities) { + throw new Error('beforeEntities is required for UPDATED action'); + } + + const before = Array.isArray(beforeEntities) + ? beforeEntities?.[idx] + : beforeEntities; + + const diff = objectRecordChangedValues( + before, + after, + objectMetadataItem, + ) as Partial>; + + const updatedFields = Object.keys(diff); + + if (updatedFields.length === 0) { + return; + } + + const event = new ObjectRecordUpdateEvent(); + + event.userId = authContext?.user?.id; + event.recordId = after.id; + event.properties = { + before, + after, + updatedFields, + diff, + }; + + return event; + }) + .filter(isDefined); + break; + case DatabaseEventAction.DELETED: + events = entityArray.map((before) => { + const event = new ObjectRecordDeleteEvent(); + + event.userId = authContext?.user?.id; + event.recordId = before.id; + event.properties = { before }; + + return event; + }); + break; + case DatabaseEventAction.DESTROYED: + events = entityArray.map((before) => { + const event = new ObjectRecordDestroyEvent(); + + event.userId = authContext?.user?.id; + event.recordId = before.id; + event.properties = { before }; + + return event; + }); + break; + case DatabaseEventAction.UPSERTED: + events = entityArray.map((after, index) => { + const event = new ObjectRecordUpsertEvent(); + + event.userId = authContext?.user?.id; + event.recordId = after.id; + + const before = beforeEntities + ? Array.isArray(beforeEntities) + ? beforeEntities[index] + : beforeEntities + : undefined; + + let updatedFields; + let diff; + + diff = objectRecordChangedValues( + before ?? {}, + after, + objectMetadataItem, + ) as Partial>; + + updatedFields = Object.keys(diff); + + event.properties = { + after, + ...(before && { before }), + ...(diff && { diff }), + ...(updatedFields && { updatedFields }), + }; + + return event; + }); + break; + default: + return; + } + + if (!events.length) { + return; + } + + return { + objectMetadataNameSingular, + action, + events, + objectMetadata: { ...objectMetadataItem, fields }, + workspaceId, + }; +}; diff --git a/packages/twenty-server/src/engine/workspace-event-emitter/types/custom-workspace-batch-event.type.ts b/packages/twenty-server/src/engine/workspace-event-emitter/types/custom-workspace-batch-event.type.ts new file mode 100644 index 00000000000..9b415fd5e0b --- /dev/null +++ b/packages/twenty-server/src/engine/workspace-event-emitter/types/custom-workspace-batch-event.type.ts @@ -0,0 +1,5 @@ +export type CustomWorkspaceEventBatch = { + name: string; + workspaceId?: string; + events: WorkspaceEvent[]; +}; diff --git a/packages/twenty-server/src/engine/workspace-event-emitter/types/workspace-event-batch.type.ts b/packages/twenty-server/src/engine/workspace-event-emitter/types/workspace-event-batch.type.ts new file mode 100644 index 00000000000..9d313629cf3 --- /dev/null +++ b/packages/twenty-server/src/engine/workspace-event-emitter/types/workspace-event-batch.type.ts @@ -0,0 +1,8 @@ +import type { ObjectMetadataEntity } from 'src/engine/metadata-modules/object-metadata/object-metadata.entity'; + +export type WorkspaceEventBatch = { + name: string; + workspaceId: string; + objectMetadata: Omit; + events: WorkspaceEvent[]; +}; diff --git a/packages/twenty-server/src/engine/workspace-event-emitter/types/workspace-event.type.ts b/packages/twenty-server/src/engine/workspace-event-emitter/types/workspace-event.type.ts deleted file mode 100644 index 623c523f280..00000000000 --- a/packages/twenty-server/src/engine/workspace-event-emitter/types/workspace-event.type.ts +++ /dev/null @@ -1,5 +0,0 @@ -export type WorkspaceEventBatch = { - name: string; - workspaceId: string; - events: WorkspaceEvent[]; -}; diff --git a/packages/twenty-server/src/engine/workspace-event-emitter/workspace-event-emitter.ts b/packages/twenty-server/src/engine/workspace-event-emitter/workspace-event-emitter.ts index 15d6bf522db..8d99d318af8 100644 --- a/packages/twenty-server/src/engine/workspace-event-emitter/workspace-event-emitter.ts +++ b/packages/twenty-server/src/engine/workspace-event-emitter/workspace-event-emitter.ts @@ -2,22 +2,19 @@ import { Injectable } from '@nestjs/common'; import { EventEmitter2 } from '@nestjs/event-emitter'; import { isDefined } from 'twenty-shared/utils'; -import { type ObjectLiteral } from 'typeorm'; import { DatabaseEventAction } from 'src/engine/api/graphql/graphql-query-runner/enums/database-event-action'; -import { type AuthContext } from 'src/engine/core-modules/auth/types/auth-context.type'; import { ObjectRecordCreateEvent } from 'src/engine/core-modules/event-emitter/types/object-record-create.event'; import { ObjectRecordDeleteEvent } from 'src/engine/core-modules/event-emitter/types/object-record-delete.event'; import { ObjectRecordDestroyEvent } from 'src/engine/core-modules/event-emitter/types/object-record-destroy.event'; -import { type ObjectRecordDiff } from 'src/engine/core-modules/event-emitter/types/object-record-diff'; import { type ObjectRecordRestoreEvent } from 'src/engine/core-modules/event-emitter/types/object-record-restore.event'; import { ObjectRecordUpdateEvent } from 'src/engine/core-modules/event-emitter/types/object-record-update.event'; import { ObjectRecordUpsertEvent } from 'src/engine/core-modules/event-emitter/types/object-record-upsert.event'; -import { objectRecordChangedValues } from 'src/engine/core-modules/event-emitter/utils/object-record-changed-values'; -import { type ObjectMetadataItemWithFieldMaps } from 'src/engine/metadata-modules/types/object-metadata-item-with-field-maps'; import { type CustomEventName } from 'src/engine/workspace-event-emitter/types/custom-event-name.type'; import { computeEventName } from 'src/engine/workspace-event-emitter/utils/compute-event-name'; -import { STANDARD_OBJECT_IDS } from 'src/engine/workspace-manager/workspace-sync-metadata/constants/standard-object-ids'; +import type { ObjectMetadataEntity } from 'src/engine/metadata-modules/object-metadata/object-metadata.entity'; +import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event-batch.type'; +import { CustomWorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/custom-workspace-batch-event.type'; type ActionEventMap = { [DatabaseEventAction.CREATED]: ObjectRecordCreateEvent; @@ -28,159 +25,32 @@ type ActionEventMap = { [DatabaseEventAction.UPSERTED]: ObjectRecordUpsertEvent; }; +export type DatabaseBatchEventInput> = { + objectMetadataNameSingular: string; + action: A; + events: ActionEventMap[A][]; + objectMetadata: Omit; + workspaceId: string; +}; + @Injectable() export class WorkspaceEventEmitter { constructor(private readonly eventEmitter: EventEmitter2) {} - async emitMutationEvent({ - action, - objectMetadataItem, - workspaceId, - authContext, - entities, - beforeEntities, - }: { - action: DatabaseEventAction; - objectMetadataItem: ObjectMetadataItemWithFieldMaps; - workspaceId: string; - authContext?: AuthContext; - entities: T | T[]; - beforeEntities?: T | T[]; - }) { - if ( - objectMetadataItem.standardId === STANDARD_OBJECT_IDS.timelineActivity - ) { + public emitDatabaseBatchEvent>( + databaseBatchEventInput: DatabaseBatchEventInput | undefined, + ) { + if (!isDefined(databaseBatchEventInput)) { return; } - const objectMetadataNameSingular = objectMetadataItem.nameSingular; - const fields = Object.values(objectMetadataItem.fieldsById ?? {}); - const entityArray = isDefined(entities) - ? Array.isArray(entities) - ? entities - : [entities] - : []; - let events: ( - | ObjectRecordCreateEvent - | ObjectRecordUpdateEvent - | ObjectRecordDeleteEvent - | ObjectRecordUpsertEvent - )[] = []; - - switch (action) { - case DatabaseEventAction.CREATED: - events = entityArray.map((after) => { - const event = new ObjectRecordCreateEvent(); - - event.userId = authContext?.user?.id; - event.recordId = after.id; - event.objectMetadata = { ...objectMetadataItem, fields }; - event.properties = { after }; - - return event; - }); - break; - case DatabaseEventAction.UPDATED: - events = entityArray - .map((after, idx) => { - if (!beforeEntities) { - throw new Error('beforeEntities is required for UPDATED action'); - } - - const before = Array.isArray(beforeEntities) - ? beforeEntities?.[idx] - : beforeEntities; - - const diff = objectRecordChangedValues( - before, - after, - objectMetadataItem, - ) as Partial>; - - const updatedFields = Object.keys(diff); - - if (updatedFields.length === 0) { - return; - } - - const event = new ObjectRecordUpdateEvent(); - - event.userId = authContext?.user?.id; - event.recordId = after.id; - event.objectMetadata = { ...objectMetadataItem, fields }; - event.properties = { - before, - after, - updatedFields, - diff, - }; - - return event; - }) - .filter(isDefined); - break; - case DatabaseEventAction.DELETED: - events = entityArray.map((before) => { - const event = new ObjectRecordDeleteEvent(); - - event.userId = authContext?.user?.id; - event.recordId = before.id; - event.objectMetadata = { ...objectMetadataItem, fields }; - event.properties = { before }; - - return event; - }); - break; - case DatabaseEventAction.DESTROYED: - events = entityArray.map((before) => { - const event = new ObjectRecordDestroyEvent(); - - event.userId = authContext?.user?.id; - event.recordId = before.id; - event.objectMetadata = { ...objectMetadataItem, fields }; - event.properties = { before }; - - return event; - }); - break; - case DatabaseEventAction.UPSERTED: - events = entityArray.map((after, index) => { - const event = new ObjectRecordUpsertEvent(); - - event.userId = authContext?.user?.id; - event.recordId = after.id; - event.objectMetadata = { ...objectMetadataItem, fields }; - - const before = beforeEntities - ? Array.isArray(beforeEntities) - ? beforeEntities[index] - : beforeEntities - : undefined; - - let updatedFields; - let diff; - - diff = objectRecordChangedValues( - before ?? {}, - after, - objectMetadataItem, - ) as Partial>; - - updatedFields = Object.keys(diff); - - event.properties = { - after, - ...(before && { before }), - ...(diff && { diff }), - ...(updatedFields && { updatedFields }), - }; - - return event; - }); - break; - default: - return; - } + const { + objectMetadataNameSingular, + action, + events, + objectMetadata, + workspaceId, + } = databaseBatchEventInput; if (!events.length) { return; @@ -188,35 +58,14 @@ export class WorkspaceEventEmitter { const eventName = computeEventName(objectMetadataNameSingular, action); - this.eventEmitter.emit(eventName, { + const workspaceEventBatch: WorkspaceEventBatch[A]> = { name: eventName, workspaceId, + objectMetadata, events, - }); - } + }; - public emitDatabaseBatchEvent>({ - objectMetadataNameSingular, - action, - events, - workspaceId, - }: { - objectMetadataNameSingular: string; - action: A; - events: ActionEventMap[A][]; - workspaceId: string | undefined; - }) { - if (!events.length) { - return; - } - - const eventName = `${objectMetadataNameSingular}.${action}`; - - this.eventEmitter.emit(eventName, { - name: eventName, - workspaceId, - events, - }); + this.eventEmitter.emit(eventName, workspaceEventBatch); } public emitCustomBatchEvent( @@ -228,10 +77,12 @@ export class WorkspaceEventEmitter { return; } - this.eventEmitter.emit(eventName, { + const customWorkspaceEventBatch: CustomWorkspaceEventBatch = { name: eventName, workspaceId, events, - }); + }; + + this.eventEmitter.emit(eventName, customWorkspaceEventBatch); } } diff --git a/packages/twenty-server/src/modules/calendar/blocklist-manager/jobs/blocklist-item-delete-calendar-events.job.ts b/packages/twenty-server/src/modules/calendar/blocklist-manager/jobs/blocklist-item-delete-calendar-events.job.ts index afd5e6c4718..ccf4c283dcd 100644 --- a/packages/twenty-server/src/modules/calendar/blocklist-manager/jobs/blocklist-item-delete-calendar-events.job.ts +++ b/packages/twenty-server/src/modules/calendar/blocklist-manager/jobs/blocklist-item-delete-calendar-events.job.ts @@ -7,7 +7,7 @@ import { Process } from 'src/engine/core-modules/message-queue/decorators/proces import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.decorator'; import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants'; import { TwentyORMManager } from 'src/engine/twenty-orm/twenty-orm.manager'; -import { type WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type'; +import { type WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event-batch.type'; import { type BlocklistWorkspaceEntity } from 'src/modules/blocklist/standard-objects/blocklist.workspace-entity'; import { CalendarEventCleanerService } from 'src/modules/calendar/calendar-event-cleaner/services/calendar-event-cleaner.service'; import { type CalendarChannelEventAssociationWorkspaceEntity } from 'src/modules/calendar/common/standard-objects/calendar-channel-event-association.workspace-entity'; diff --git a/packages/twenty-server/src/modules/calendar/blocklist-manager/jobs/blocklist-reimport-calendar-events.job.ts b/packages/twenty-server/src/modules/calendar/blocklist-manager/jobs/blocklist-reimport-calendar-events.job.ts index 6bb0506c1d4..256eeb6373f 100644 --- a/packages/twenty-server/src/modules/calendar/blocklist-manager/jobs/blocklist-reimport-calendar-events.job.ts +++ b/packages/twenty-server/src/modules/calendar/blocklist-manager/jobs/blocklist-reimport-calendar-events.job.ts @@ -7,7 +7,7 @@ import { Process } from 'src/engine/core-modules/message-queue/decorators/proces import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.decorator'; import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants'; import { TwentyORMManager } from 'src/engine/twenty-orm/twenty-orm.manager'; -import { type WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type'; +import { type WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event-batch.type'; import { type BlocklistWorkspaceEntity } from 'src/modules/blocklist/standard-objects/blocklist.workspace-entity'; import { CalendarChannelSyncStatusService } from 'src/modules/calendar/common/services/calendar-channel-sync-status.service'; import { diff --git a/packages/twenty-server/src/modules/calendar/blocklist-manager/listeners/calendar-blocklist.listener.ts b/packages/twenty-server/src/modules/calendar/blocklist-manager/listeners/calendar-blocklist.listener.ts index 44f4c954639..39a5fc7c500 100644 --- a/packages/twenty-server/src/modules/calendar/blocklist-manager/listeners/calendar-blocklist.listener.ts +++ b/packages/twenty-server/src/modules/calendar/blocklist-manager/listeners/calendar-blocklist.listener.ts @@ -6,7 +6,7 @@ import { type ObjectRecordUpdateEvent } from 'src/engine/core-modules/event-emit import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator'; import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants'; import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service'; -import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type'; +import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event-batch.type'; import { type BlocklistWorkspaceEntity } from 'src/modules/blocklist/standard-objects/blocklist.workspace-entity'; import { BlocklistItemDeleteCalendarEventsJob, diff --git a/packages/twenty-server/src/modules/calendar/calendar-event-cleaner/listeners/calendar-event-cleaner-connected-account.listener.ts b/packages/twenty-server/src/modules/calendar/calendar-event-cleaner/listeners/calendar-event-cleaner-connected-account.listener.ts index b961f41de6d..30966437884 100644 --- a/packages/twenty-server/src/modules/calendar/calendar-event-cleaner/listeners/calendar-event-cleaner-connected-account.listener.ts +++ b/packages/twenty-server/src/modules/calendar/calendar-event-cleaner/listeners/calendar-event-cleaner-connected-account.listener.ts @@ -4,7 +4,7 @@ import { type ObjectRecordDeleteEvent } from 'src/engine/core-modules/event-emit import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator'; import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants'; import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service'; -import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type'; +import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event-batch.type'; import { DeleteConnectedAccountAssociatedCalendarDataJob, type DeleteConnectedAccountAssociatedCalendarDataJobData, diff --git a/packages/twenty-server/src/modules/calendar/calendar-event-participant-manager/listeners/calendar-event-participant-person.listener.ts b/packages/twenty-server/src/modules/calendar/calendar-event-participant-manager/listeners/calendar-event-participant-person.listener.ts index 30e4f2e0355..4881e83aae2 100644 --- a/packages/twenty-server/src/modules/calendar/calendar-event-participant-manager/listeners/calendar-event-participant-person.listener.ts +++ b/packages/twenty-server/src/modules/calendar/calendar-event-participant-manager/listeners/calendar-event-participant-person.listener.ts @@ -11,7 +11,7 @@ import { objectRecordChangedProperties as objectRecordUpdateEventChangedProperti import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator'; import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants'; import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service'; -import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type'; +import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event-batch.type'; import { CalendarEventParticipantMatchParticipantJob, type CalendarEventParticipantMatchParticipantJobData, diff --git a/packages/twenty-server/src/modules/calendar/calendar-event-participant-manager/listeners/calendar-event-participant-workspace-member.listener.ts b/packages/twenty-server/src/modules/calendar/calendar-event-participant-manager/listeners/calendar-event-participant-workspace-member.listener.ts index 561263c2a18..3d53bea723a 100644 --- a/packages/twenty-server/src/modules/calendar/calendar-event-participant-manager/listeners/calendar-event-participant-workspace-member.listener.ts +++ b/packages/twenty-server/src/modules/calendar/calendar-event-participant-manager/listeners/calendar-event-participant-workspace-member.listener.ts @@ -8,7 +8,7 @@ import { objectRecordChangedProperties as objectRecordUpdateEventChangedProperti import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator'; import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants'; import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service'; -import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type'; +import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event-batch.type'; import { CalendarEventParticipantMatchParticipantJob, type CalendarEventParticipantMatchParticipantJobData, diff --git a/packages/twenty-server/src/modules/calendar/calendar-event-participant-manager/listeners/calendar-event-participant.listener.ts b/packages/twenty-server/src/modules/calendar/calendar-event-participant-manager/listeners/calendar-event-participant.listener.ts index e558e6b7fa5..ab0ff293643 100644 --- a/packages/twenty-server/src/modules/calendar/calendar-event-participant-manager/listeners/calendar-event-participant.listener.ts +++ b/packages/twenty-server/src/modules/calendar/calendar-event-participant-manager/listeners/calendar-event-participant.listener.ts @@ -7,10 +7,10 @@ import { Repository } from 'typeorm'; import { OnCustomBatchEvent } from 'src/engine/api/graphql/graphql-query-runner/decorators/on-custom-batch-event.decorator'; import { ObjectMetadataEntity } from 'src/engine/metadata-modules/object-metadata/object-metadata.entity'; import { InjectObjectMetadataRepository } from 'src/engine/object-metadata-repository/object-metadata-repository.decorator'; -import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type'; import { type CalendarEventParticipantWorkspaceEntity } from 'src/modules/calendar/common/standard-objects/calendar-event-participant.workspace-entity'; import { TimelineActivityRepository } from 'src/modules/timeline/repositories/timeline-activity.repository'; import { TimelineActivityWorkspaceEntity } from 'src/modules/timeline/standard-objects/timeline-activity.workspace-entity'; +import { CustomWorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/custom-workspace-batch-event.type'; @Injectable() export class CalendarEventParticipantListener { @@ -23,11 +23,15 @@ export class CalendarEventParticipantListener { @OnCustomBatchEvent('calendarEventParticipant_matched') public async handleCalendarEventParticipantMatchedEvent( - batchEvent: WorkspaceEventBatch<{ + batchEvent: CustomWorkspaceEventBatch<{ workspaceMemberId: string; participants: CalendarEventParticipantWorkspaceEntity[]; }>, ): Promise { + if (!isDefined(batchEvent.workspaceId)) { + return; + } + const calendarEventObjectMetadata = await this.objectMetadataRepository.findOneOrFail({ where: { diff --git a/packages/twenty-server/src/modules/connected-account/listeners/connected-account-workspace-member.listener.ts b/packages/twenty-server/src/modules/connected-account/listeners/connected-account-workspace-member.listener.ts index d8de2649fc9..15496cbb14a 100644 --- a/packages/twenty-server/src/modules/connected-account/listeners/connected-account-workspace-member.listener.ts +++ b/packages/twenty-server/src/modules/connected-account/listeners/connected-account-workspace-member.listener.ts @@ -7,7 +7,7 @@ import { type ObjectRecordDestroyEvent } from 'src/engine/core-modules/event-emi import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator'; import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants'; import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service'; -import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type'; +import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event-batch.type'; import { DeleteWorkspaceMemberConnectedAccountsCleanupJob, type DeleteWorkspaceMemberConnectedAccountsCleanupJobData, diff --git a/packages/twenty-server/src/modules/connected-account/listeners/connected-account.listener.ts b/packages/twenty-server/src/modules/connected-account/listeners/connected-account.listener.ts index 8ba24e93f72..fbfae0a87a5 100644 --- a/packages/twenty-server/src/modules/connected-account/listeners/connected-account.listener.ts +++ b/packages/twenty-server/src/modules/connected-account/listeners/connected-account.listener.ts @@ -4,7 +4,7 @@ import { OnDatabaseBatchEvent } from 'src/engine/api/graphql/graphql-query-runne import { DatabaseEventAction } from 'src/engine/api/graphql/graphql-query-runner/enums/database-event-action'; import { type ObjectRecordDeleteEvent } from 'src/engine/core-modules/event-emitter/types/object-record-delete.event'; import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager'; -import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type'; +import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event-batch.type'; import { AccountsToReconnectService } from 'src/modules/connected-account/services/accounts-to-reconnect.service'; import { type ConnectedAccountWorkspaceEntity } from 'src/modules/connected-account/standard-objects/connected-account.workspace-entity'; import { type WorkspaceMemberWorkspaceEntity } from 'src/modules/workspace-member/standard-objects/workspace-member.workspace-entity'; diff --git a/packages/twenty-server/src/modules/connected-account/query-hooks/connected-account-delete-one.pre-query.hook.ts b/packages/twenty-server/src/modules/connected-account/query-hooks/connected-account-delete-one.pre-query.hook.ts index 60396128631..e56b567246f 100644 --- a/packages/twenty-server/src/modules/connected-account/query-hooks/connected-account-delete-one.pre-query.hook.ts +++ b/packages/twenty-server/src/modules/connected-account/query-hooks/connected-account-delete-one.pre-query.hook.ts @@ -57,9 +57,9 @@ export class ConnectedAccountDeleteOnePreQueryHook this.workspaceEventEmitter.emitDatabaseBatchEvent({ objectMetadataNameSingular: 'messageChannel', action: DatabaseEventAction.DESTROYED, + objectMetadata, events: messageChannels.map((messageChannel) => ({ recordId: messageChannel.id, - objectMetadata, properties: { before: messageChannel, }, diff --git a/packages/twenty-server/src/modules/contact-creation-manager/listeners/auto-companies-and-contacts-creation-calendar-channel.listener.ts b/packages/twenty-server/src/modules/contact-creation-manager/listeners/auto-companies-and-contacts-creation-calendar-channel.listener.ts index b55301d3598..00a98522efd 100644 --- a/packages/twenty-server/src/modules/contact-creation-manager/listeners/auto-companies-and-contacts-creation-calendar-channel.listener.ts +++ b/packages/twenty-server/src/modules/contact-creation-manager/listeners/auto-companies-and-contacts-creation-calendar-channel.listener.ts @@ -5,7 +5,7 @@ import { objectRecordChangedProperties } from 'src/engine/core-modules/event-emi import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator'; import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants'; import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service'; -import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type'; +import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event-batch.type'; import { CalendarCreateCompanyAndContactAfterSyncJob, type CalendarCreateCompanyAndContactAfterSyncJobData, diff --git a/packages/twenty-server/src/modules/contact-creation-manager/listeners/auto-companies-and-contacts-creation-message-channel.listener.ts b/packages/twenty-server/src/modules/contact-creation-manager/listeners/auto-companies-and-contacts-creation-message-channel.listener.ts index c307d19614e..098d7f4f1c7 100644 --- a/packages/twenty-server/src/modules/contact-creation-manager/listeners/auto-companies-and-contacts-creation-message-channel.listener.ts +++ b/packages/twenty-server/src/modules/contact-creation-manager/listeners/auto-companies-and-contacts-creation-message-channel.listener.ts @@ -5,7 +5,7 @@ import { objectRecordChangedProperties } from 'src/engine/core-modules/event-emi import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator'; import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants'; import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service'; -import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type'; +import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event-batch.type'; import { type MessageChannelWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity'; import { MessagingCreateCompanyAndContactAfterSyncJob, diff --git a/packages/twenty-server/src/modules/favorite-folder/listeners/favorite-folder.listener.ts b/packages/twenty-server/src/modules/favorite-folder/listeners/favorite-folder.listener.ts index 18cefe1dfbd..378aa2494c6 100644 --- a/packages/twenty-server/src/modules/favorite-folder/listeners/favorite-folder.listener.ts +++ b/packages/twenty-server/src/modules/favorite-folder/listeners/favorite-folder.listener.ts @@ -4,7 +4,7 @@ import { OnDatabaseBatchEvent } from 'src/engine/api/graphql/graphql-query-runne import { DatabaseEventAction } from 'src/engine/api/graphql/graphql-query-runner/enums/database-event-action'; import { type ObjectRecordDeleteEvent } from 'src/engine/core-modules/event-emitter/types/object-record-delete.event'; import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager'; -import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type'; +import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event-batch.type'; import { type FavoriteFolderWorkspaceEntity } from 'src/modules/favorite-folder/standard-objects/favorite-folder.workspace-entity'; import { type FavoriteWorkspaceEntity } from 'src/modules/favorite/standard-objects/favorite.workspace-entity'; diff --git a/packages/twenty-server/src/modules/favorite/listeners/favorite-deletion.listener.ts b/packages/twenty-server/src/modules/favorite/listeners/favorite-deletion.listener.ts index 439d71043e9..66c80ce983e 100644 --- a/packages/twenty-server/src/modules/favorite/listeners/favorite-deletion.listener.ts +++ b/packages/twenty-server/src/modules/favorite/listeners/favorite-deletion.listener.ts @@ -6,7 +6,7 @@ import { type ObjectRecordDeleteEvent } from 'src/engine/core-modules/event-emit import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator'; import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants'; import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service'; -import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type'; +import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event-batch.type'; import { FavoriteDeletionJob, type FavoriteDeletionJobData, diff --git a/packages/twenty-server/src/modules/messaging/blocklist-manager/jobs/messaging-blocklist-item-delete-messages.job.ts b/packages/twenty-server/src/modules/messaging/blocklist-manager/jobs/messaging-blocklist-item-delete-messages.job.ts index c4cb444486a..21718b351bb 100644 --- a/packages/twenty-server/src/modules/messaging/blocklist-manager/jobs/messaging-blocklist-item-delete-messages.job.ts +++ b/packages/twenty-server/src/modules/messaging/blocklist-manager/jobs/messaging-blocklist-item-delete-messages.job.ts @@ -7,7 +7,7 @@ import { Process } from 'src/engine/core-modules/message-queue/decorators/proces import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.decorator'; import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants'; import { TwentyORMManager } from 'src/engine/twenty-orm/twenty-orm.manager'; -import { type WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type'; +import { type WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event-batch.type'; import { type BlocklistWorkspaceEntity } from 'src/modules/blocklist/standard-objects/blocklist.workspace-entity'; import { type MessageChannelMessageAssociationWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-channel-message-association.workspace-entity'; import { type MessageChannelWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity'; diff --git a/packages/twenty-server/src/modules/messaging/blocklist-manager/jobs/messaging-blocklist-reimport-messages.job.ts b/packages/twenty-server/src/modules/messaging/blocklist-manager/jobs/messaging-blocklist-reimport-messages.job.ts index 3e019e33a13..b51e1be42e9 100644 --- a/packages/twenty-server/src/modules/messaging/blocklist-manager/jobs/messaging-blocklist-reimport-messages.job.ts +++ b/packages/twenty-server/src/modules/messaging/blocklist-manager/jobs/messaging-blocklist-reimport-messages.job.ts @@ -7,7 +7,7 @@ import { Process } from 'src/engine/core-modules/message-queue/decorators/proces import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.decorator'; import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants'; import { TwentyORMManager } from 'src/engine/twenty-orm/twenty-orm.manager'; -import { type WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type'; +import { type WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event-batch.type'; import { type BlocklistWorkspaceEntity } from 'src/modules/blocklist/standard-objects/blocklist.workspace-entity'; import { MessageChannelSyncStatusService } from 'src/modules/messaging/common/services/message-channel-sync-status.service'; import { diff --git a/packages/twenty-server/src/modules/messaging/blocklist-manager/listeners/messaging-blocklist.listener.ts b/packages/twenty-server/src/modules/messaging/blocklist-manager/listeners/messaging-blocklist.listener.ts index af4805eb453..322aac97527 100644 --- a/packages/twenty-server/src/modules/messaging/blocklist-manager/listeners/messaging-blocklist.listener.ts +++ b/packages/twenty-server/src/modules/messaging/blocklist-manager/listeners/messaging-blocklist.listener.ts @@ -6,7 +6,7 @@ import { type ObjectRecordUpdateEvent } from 'src/engine/core-modules/event-emit import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator'; import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants'; import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service'; -import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type'; +import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event-batch.type'; import { type BlocklistWorkspaceEntity } from 'src/modules/blocklist/standard-objects/blocklist.workspace-entity'; import { BlocklistItemDeleteMessagesJob, diff --git a/packages/twenty-server/src/modules/messaging/message-cleaner/listeners/messaging-message-cleaner-connected-account.listener.ts b/packages/twenty-server/src/modules/messaging/message-cleaner/listeners/messaging-message-cleaner-connected-account.listener.ts index 888044884d9..a3a7139574f 100644 --- a/packages/twenty-server/src/modules/messaging/message-cleaner/listeners/messaging-message-cleaner-connected-account.listener.ts +++ b/packages/twenty-server/src/modules/messaging/message-cleaner/listeners/messaging-message-cleaner-connected-account.listener.ts @@ -4,7 +4,7 @@ import { type ObjectRecordDeleteEvent } from 'src/engine/core-modules/event-emit import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator'; import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants'; import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service'; -import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type'; +import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event-batch.type'; import { type ConnectedAccountWorkspaceEntity } from 'src/modules/connected-account/standard-objects/connected-account.workspace-entity'; import { MessagingConnectedAccountDeletionCleanupJob, diff --git a/packages/twenty-server/src/modules/messaging/message-import-manager/listeners/messaging-import-manager-message-channel.listener.ts b/packages/twenty-server/src/modules/messaging/message-import-manager/listeners/messaging-import-manager-message-channel.listener.ts index d1d3d04f46a..2dc1e52dc88 100644 --- a/packages/twenty-server/src/modules/messaging/message-import-manager/listeners/messaging-import-manager-message-channel.listener.ts +++ b/packages/twenty-server/src/modules/messaging/message-import-manager/listeners/messaging-import-manager-message-channel.listener.ts @@ -4,7 +4,7 @@ import { type ObjectRecordDeleteEvent } from 'src/engine/core-modules/event-emit import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator'; import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants'; import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service'; -import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type'; +import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event-batch.type'; import { type MessageChannelWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity'; import { MessagingCleanCacheJob, diff --git a/packages/twenty-server/src/modules/messaging/message-participant-manager/listeners/message-participant-person.listener.ts b/packages/twenty-server/src/modules/messaging/message-participant-manager/listeners/message-participant-person.listener.ts index 1cfbb504472..e465f110172 100644 --- a/packages/twenty-server/src/modules/messaging/message-participant-manager/listeners/message-participant-person.listener.ts +++ b/packages/twenty-server/src/modules/messaging/message-participant-manager/listeners/message-participant-person.listener.ts @@ -11,7 +11,7 @@ import { objectRecordChangedProperties as objectRecordUpdateEventChangedProperti import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator'; import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants'; import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service'; -import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type'; +import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event-batch.type'; import { MessageParticipantMatchParticipantJob, type MessageParticipantMatchParticipantJobData, diff --git a/packages/twenty-server/src/modules/messaging/message-participant-manager/listeners/message-participant-workspace-member.listener.ts b/packages/twenty-server/src/modules/messaging/message-participant-manager/listeners/message-participant-workspace-member.listener.ts index 5c27eff8998..30745860c4c 100644 --- a/packages/twenty-server/src/modules/messaging/message-participant-manager/listeners/message-participant-workspace-member.listener.ts +++ b/packages/twenty-server/src/modules/messaging/message-participant-manager/listeners/message-participant-workspace-member.listener.ts @@ -13,7 +13,7 @@ import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decora import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants'; import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service'; import { Workspace } from 'src/engine/core-modules/workspace/workspace.entity'; -import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type'; +import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event-batch.type'; import { MessageParticipantMatchParticipantJob, type MessageParticipantMatchParticipantJobData, diff --git a/packages/twenty-server/src/modules/messaging/message-participant-manager/listeners/message-participant.listener.ts b/packages/twenty-server/src/modules/messaging/message-participant-manager/listeners/message-participant.listener.ts index 4465c4898e1..85faea7a9ab 100644 --- a/packages/twenty-server/src/modules/messaging/message-participant-manager/listeners/message-participant.listener.ts +++ b/packages/twenty-server/src/modules/messaging/message-participant-manager/listeners/message-participant.listener.ts @@ -7,10 +7,10 @@ import { Repository } from 'typeorm'; import { OnCustomBatchEvent } from 'src/engine/api/graphql/graphql-query-runner/decorators/on-custom-batch-event.decorator'; import { ObjectMetadataEntity } from 'src/engine/metadata-modules/object-metadata/object-metadata.entity'; import { InjectObjectMetadataRepository } from 'src/engine/object-metadata-repository/object-metadata-repository.decorator'; -import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type'; import { type MessageParticipantWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-participant.workspace-entity'; import { TimelineActivityRepository } from 'src/modules/timeline/repositories/timeline-activity.repository'; import { TimelineActivityWorkspaceEntity } from 'src/modules/timeline/standard-objects/timeline-activity.workspace-entity'; +import { CustomWorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/custom-workspace-batch-event.type'; @Injectable() export class MessageParticipantListener { @@ -23,11 +23,15 @@ export class MessageParticipantListener { @OnCustomBatchEvent('messageParticipant_matched') public async handleMessageParticipantMatched( - batchEvent: WorkspaceEventBatch<{ + batchEvent: CustomWorkspaceEventBatch<{ workspaceMemberId: string; participants: MessageParticipantWorkspaceEntity[]; }>, ): Promise { + if (!isDefined(batchEvent.workspaceId)) { + return; + } + const messageObjectMetadata = await this.objectMetadataRepository.findOneOrFail({ where: { diff --git a/packages/twenty-server/src/modules/timeline/jobs/upsert-timeline-activity-from-internal-event.job.ts b/packages/twenty-server/src/modules/timeline/jobs/upsert-timeline-activity-from-internal-event.job.ts index ad83f7f8598..4ffee94847e 100644 --- a/packages/twenty-server/src/modules/timeline/jobs/upsert-timeline-activity-from-internal-event.job.ts +++ b/packages/twenty-server/src/modules/timeline/jobs/upsert-timeline-activity-from-internal-event.job.ts @@ -6,7 +6,7 @@ import { Process } from 'src/engine/core-modules/message-queue/decorators/proces import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.decorator'; import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants'; import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager'; -import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type'; +import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event-batch.type'; import { TimelineActivityService } from 'src/modules/timeline/services/timeline-activity.service'; import { WorkspaceMemberWorkspaceEntity } from 'src/modules/workspace-member/standard-objects/workspace-member.workspace-entity'; @@ -21,6 +21,18 @@ export class UpsertTimelineActivityFromInternalEvent { async handle( workspaceEventBatch: WorkspaceEventBatch, ): Promise { + if (workspaceEventBatch.events.length === 0) { + return; + } + + if ( + workspaceEventBatch.objectMetadata.isSystem && + workspaceEventBatch.objectMetadata.nameSingular !== 'noteTarget' && + workspaceEventBatch.objectMetadata.nameSingular !== 'taskTarget' + ) { + return; + } + const workspaceMemberRepository = await this.twentyORMGlobalManager.getRepositoryForWorkspace( workspaceEventBatch.workspaceId, @@ -48,35 +60,22 @@ export class UpsertTimelineActivityFromInternalEvent { } } - const filteredEvents = workspaceEventBatch.events - .filter((event) => { - return ( - !event.objectMetadata.isSystem || - event.objectMetadata.nameSingular === 'noteTarget' || - event.objectMetadata.nameSingular === 'taskTarget' - ); - }) - .map((event) => { - if ('diff' in event.properties && event.properties.diff) { - return { - ...event, - properties: { - diff: event.properties.diff, - }, - }; - } + const formattedEvents = workspaceEventBatch.events.map((event) => { + if ('diff' in event.properties && event.properties.diff) { + return { + ...event, + properties: { + diff: event.properties.diff, + }, + }; + } - return event; - }); - - if (filteredEvents.length === 0) { - return; - } + return event; + }); await this.timelineActivityService.upsertEvents({ - events: filteredEvents, - eventName: workspaceEventBatch.name, - workspaceId: workspaceEventBatch.workspaceId, + ...workspaceEventBatch, + events: formattedEvents, }); } } diff --git a/packages/twenty-server/src/modules/timeline/services/timeline-activity.service.ts b/packages/twenty-server/src/modules/timeline/services/timeline-activity.service.ts index 7f9e0228451..fb5ae69970a 100644 --- a/packages/twenty-server/src/modules/timeline/services/timeline-activity.service.ts +++ b/packages/twenty-server/src/modules/timeline/services/timeline-activity.service.ts @@ -12,15 +12,10 @@ import { parseEventNameOrThrow } from 'src/engine/workspace-event-emitter/utils/ import { TimelineActivityRepository } from 'src/modules/timeline/repositories/timeline-activity.repository'; import { TimelineActivityWorkspaceEntity } from 'src/modules/timeline/standard-objects/timeline-activity.workspace-entity'; import { type TimelineActivityPayload } from 'src/modules/timeline/types/timeline-activity-payload'; +import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event-batch.type'; type ActivityType = 'note' | 'task'; -type EventsWithNameAndWorkspaceId = { - events: ObjectRecordBaseEvent[]; - eventName: string; - workspaceId: string; -}; - @Injectable() export class TimelineActivityService { constructor( @@ -36,16 +31,22 @@ export class TimelineActivityService { async upsertEvents({ events, - eventName, + name, + objectMetadata, workspaceId, - }: EventsWithNameAndWorkspaceId) { - const { objectSingularName } = parseEventNameOrThrow(eventName); + }: WorkspaceEventBatch) { + if (!isDefined(workspaceId)) { + return; + } + + const { objectSingularName } = parseEventNameOrThrow(name); const timelineActivitiesPayloads = await this.transformEventsToTimelineActivityPayloads({ events, + objectMetadata, workspaceId, - eventName, + name, }); if ( @@ -82,11 +83,12 @@ export class TimelineActivityService { private async transformEventsToTimelineActivityPayloads({ events, workspaceId, - eventName, - }: EventsWithNameAndWorkspaceId): Promise< + objectMetadata, + name, + }: WorkspaceEventBatch): Promise< TimelineActivityPayload[] | undefined > { - const { objectSingularName } = parseEventNameOrThrow(eventName); + const { objectSingularName } = parseEventNameOrThrow(name); if (objectSingularName === 'note') { const noteEventsTimelineActivities = @@ -94,13 +96,14 @@ export class TimelineActivityService { events, activityType: 'note', workspaceId, - eventName, + objectMetadata, + name, }); return [ ...noteEventsTimelineActivities, ...(events.map((event) => ({ - name: eventName, + name, objectSingularName, recordId: event.recordId, workspaceMemberId: event.workspaceMemberId, @@ -115,13 +118,14 @@ export class TimelineActivityService { events, activityType: 'task', workspaceId, - eventName, + objectMetadata, + name, }); return [ ...taskEventsTimelineActivities, ...(events.map((event) => ({ - name: eventName, + name, objectSingularName, recordId: event.recordId, workspaceMemberId: event.workspaceMemberId, @@ -138,12 +142,13 @@ export class TimelineActivityService { events, activityType: objectSingularName === 'noteTarget' ? 'note' : 'task', workspaceId, - eventName, + objectMetadata, + name, }); } return events.map((event) => ({ - name: eventName, + name, objectSingularName, recordId: event.recordId, workspaceMemberId: event.workspaceMemberId, @@ -154,12 +159,17 @@ export class TimelineActivityService { private async computeTimelineActivityPayloadsForActivities({ events, activityType, - eventName, + name, workspaceId, - }: EventsWithNameAndWorkspaceId & { activityType: ActivityType }): Promise< - TimelineActivityPayload[] - > { - const { action } = parseEventNameOrThrow(eventName); + objectMetadata, + }: WorkspaceEventBatch & { + activityType: ActivityType; + }): Promise { + if (!isDefined(workspaceId)) { + return []; + } + + const { action } = parseEventNameOrThrow(name); const activityTargetRepository = await this.twentyORMGlobalManager.getRepositoryForWorkspace( @@ -218,9 +228,9 @@ export class TimelineActivityService { recordId: activityTarget[targetColumn.replace(/Id$/, '')], linkedRecordCachedName: activityTitle, linkedRecordId: activityId, - linkedObjectMetadataId: event.objectMetadata.id, + linkedObjectMetadataId: objectMetadata.id, properties: event.properties, - overrideObjectSingularName: event.objectMetadata.nameSingular, + overrideObjectSingularName: objectMetadata.nameSingular, } satisfies TimelineActivityPayload; }); }) @@ -230,12 +240,13 @@ export class TimelineActivityService { private async computeTimelineActivityPayloadsForActivityTargets({ events, activityType, - eventName, + name, + objectMetadata, workspaceId, - }: EventsWithNameAndWorkspaceId & { activityType: ActivityType }): Promise< - TimelineActivityPayload[] - > { - const { action } = parseEventNameOrThrow(eventName); + }: WorkspaceEventBatch & { + activityType: ActivityType; + }): Promise { + const { action } = parseEventNameOrThrow(name); const activityRepository = await this.twentyORMGlobalManager.getRepositoryForWorkspace( @@ -277,7 +288,7 @@ export class TimelineActivityService { return; } - const activityObjectMetadataId = event.objectMetadata.fields.find( + const activityObjectMetadataId = objectMetadata.fields.find( (field) => field.name === activityType, )?.relationTargetObjectMetadataId; diff --git a/packages/twenty-server/src/modules/workflow/workflow-status/listeners/workflow-version-status.listener.ts b/packages/twenty-server/src/modules/workflow/workflow-status/listeners/workflow-version-status.listener.ts index b72acf55e79..2f9ee5cf803 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-status/listeners/workflow-version-status.listener.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-status/listeners/workflow-version-status.listener.ts @@ -1,11 +1,12 @@ import { Injectable } from '@nestjs/common'; +import { isDefined } from 'twenty-shared/utils'; + import { type ObjectRecordCreateEvent } from 'src/engine/core-modules/event-emitter/types/object-record-create.event'; import { type ObjectRecordDeleteEvent } from 'src/engine/core-modules/event-emitter/types/object-record-delete.event'; import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator'; import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants'; import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service'; -import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type'; import { WorkflowVersionStatus, type WorkflowVersionWorkspaceEntity, @@ -20,6 +21,7 @@ import { OnDatabaseBatchEvent } from 'src/engine/api/graphql/graphql-query-runne import { DatabaseEventAction } from 'src/engine/api/graphql/graphql-query-runner/enums/database-event-action'; import { WORKFLOW_VERSION_STATUS_UPDATED } from 'src/modules/workflow/workflow-status/constants/workflow-version-status-updated.constants'; import { OnCustomBatchEvent } from 'src/engine/api/graphql/graphql-query-runner/decorators/on-custom-batch-event.decorator'; +import { CustomWorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/custom-workspace-batch-event.type'; @Injectable() export class WorkflowVersionStatusListener { @@ -30,10 +32,14 @@ export class WorkflowVersionStatusListener { @OnDatabaseBatchEvent('workflowVersion', DatabaseEventAction.CREATED) async handleWorkflowVersionCreated( - batchEvent: WorkspaceEventBatch< + batchEvent: CustomWorkspaceEventBatch< ObjectRecordCreateEvent >, ): Promise { + if (!isDefined(batchEvent.workspaceId)) { + return; + } + const workflowIds = batchEvent.events .filter( (event) => @@ -58,8 +64,12 @@ export class WorkflowVersionStatusListener { @OnCustomBatchEvent(WORKFLOW_VERSION_STATUS_UPDATED) async handleWorkflowVersionUpdated( - batchEvent: WorkspaceEventBatch, + batchEvent: CustomWorkspaceEventBatch, ): Promise { + if (!isDefined(batchEvent.workspaceId)) { + return; + } + await this.messageQueueService.add( WorkflowStatusesUpdateJob.name, { @@ -72,10 +82,14 @@ export class WorkflowVersionStatusListener { @OnDatabaseBatchEvent('workflowVersion', DatabaseEventAction.DELETED) async handleWorkflowVersionDeleted( - batchEvent: WorkspaceEventBatch< + batchEvent: CustomWorkspaceEventBatch< ObjectRecordDeleteEvent >, ): Promise { + if (!isDefined(batchEvent.workspaceId)) { + return; + } + const workflowIds = batchEvent.events .filter( (event) => diff --git a/packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/listeners/__tests__/workflow-database-event-trigger.listener.spec.ts b/packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/listeners/__tests__/workflow-database-event-trigger.listener.spec.ts index 28639244158..03084f9dc3a 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/listeners/__tests__/workflow-database-event-trigger.listener.spec.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/listeners/__tests__/workflow-database-event-trigger.listener.spec.ts @@ -96,30 +96,30 @@ describe('WorkflowDatabaseEventTriggerListener', () => { const mockPayload = { workspaceId, name: databaseEventName, + objectMetadata: getMockObjectMetadataEntity({ + id: 'test-object-metadata', + workspaceId, + nameSingular: 'testObject', + namePlural: 'testObjects', + labelSingular: 'Test Object', + labelPlural: 'Test Objects', + description: 'Test object for testing', + targetTableName: 'test_objects', + isSystem: false, + isCustom: false, + isActive: true, + isRemote: false, + isAuditLogged: true, + isSearchable: true, + createdAt: new Date(), + updatedAt: new Date(), + fields: [], + indexMetadatas: [], + icon: 'Icon123', + }), events: [ { recordId: 'test-record', - objectMetadata: getMockObjectMetadataEntity({ - id: 'test-object-metadata', - workspaceId, - nameSingular: 'testObject', - namePlural: 'testObjects', - labelSingular: 'Test Object', - labelPlural: 'Test Objects', - description: 'Test object for testing', - targetTableName: 'test_objects', - isSystem: false, - isCustom: false, - isActive: true, - isRemote: false, - isAuditLogged: true, - isSearchable: true, - createdAt: new Date(), - updatedAt: new Date(), - fields: [], - indexMetadatas: [], - icon: 'Icon123', - }), properties: { updatedFields: ['field1', 'field2'], before: { field1: 'old', field2: 'old' }, diff --git a/packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/listeners/workflow-database-event-trigger.listener.ts b/packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/listeners/workflow-database-event-trigger.listener.ts index 9c85d1b977a..05802930b52 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/listeners/workflow-database-event-trigger.listener.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/listeners/workflow-database-event-trigger.listener.ts @@ -19,7 +19,7 @@ import { MessageQueueService } from 'src/engine/core-modules/message-queue/servi import { ObjectMetadataItemWithFieldMaps } from 'src/engine/metadata-modules/types/object-metadata-item-with-field-maps'; import { ObjectMetadataMaps } from 'src/engine/metadata-modules/types/object-metadata-maps'; import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager'; -import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type'; +import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event-batch.type'; import { AutomatedTriggerType, type WorkflowAutomatedTriggerWorkspaceEntity, @@ -138,7 +138,7 @@ export class WorkflowDatabaseEventTriggerListener { const workspaceId = payload.workspaceId; const { objectMetadataMaps, objectMetadataItemWithFieldsMaps } = await this.workflowCommonWorkspaceService.getObjectMetadataItemWithFieldsMaps( - payload.events[0].objectMetadata.nameSingular, + payload.objectMetadata.nameSingular, workspaceId, ); @@ -156,7 +156,7 @@ export class WorkflowDatabaseEventTriggerListener { const workspaceId = payload.workspaceId; const { objectMetadataMaps, objectMetadataItemWithFieldsMaps } = await this.workflowCommonWorkspaceService.getObjectMetadataItemWithFieldsMaps( - payload.events[0].objectMetadata.nameSingular, + payload.objectMetadata.nameSingular, workspaceId, ); @@ -180,7 +180,7 @@ export class WorkflowDatabaseEventTriggerListener { const workspaceId = payload.workspaceId; const { objectMetadataMaps, objectMetadataItemWithFieldsMaps } = await this.workflowCommonWorkspaceService.getObjectMetadataItemWithFieldsMaps( - payload.events[0].objectMetadata.nameSingular, + payload.objectMetadata.nameSingular, workspaceId, ); @@ -198,7 +198,7 @@ export class WorkflowDatabaseEventTriggerListener { const workspaceId = payload.workspaceId; const { objectMetadataMaps, objectMetadataItemWithFieldsMaps } = await this.workflowCommonWorkspaceService.getObjectMetadataItemWithFieldsMaps( - payload.events[0].objectMetadata.nameSingular, + payload.objectMetadata.nameSingular, workspaceId, ); diff --git a/packages/twenty-zapier/src/utils/triggers/triggers.utils.ts b/packages/twenty-zapier/src/utils/triggers/triggers.utils.ts index 89f01f479c6..2383d4c38cb 100644 --- a/packages/twenty-zapier/src/utils/triggers/triggers.utils.ts +++ b/packages/twenty-zapier/src/utils/triggers/triggers.utils.ts @@ -1,4 +1,4 @@ -import { Bundle, ZObject } from 'zapier-platform-core'; +import { type Bundle, type ZObject } from 'zapier-platform-core'; import handleQueryParams from '../../utils/handleQueryParams'; import requestDb, { @@ -93,9 +93,7 @@ export const performList = async ( return results.map((result) => ({ record: result, ...(bundle.inputData.operation === DatabaseEventAction.UPDATED && { - updatedFields: [ - Object.keys(result).filter((key) => key !== 'id')?.[0], - ] || ['updatedField'], + updatedFields: [Object.keys(result).filter((key) => key !== 'id')?.[0]], }), })); };