6619 modify event emitter to emit an array of events

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

@@ -8,6 +8,7 @@ import { objectRecordChangedValues } from 'src/engine/integrations/event-emitter
 import { InjectMessageQueue } from 'src/engine/integrations/message-queue/decorators/message-queue.decorator';
 import { MessageQueue } from 'src/engine/integrations/message-queue/message-queue.constants';
 import { MessageQueueService } from 'src/engine/integrations/message-queue/services/message-queue.service';
+import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/workspace-event.type';
 import { CreateAuditLogFromInternalEvent } from 'src/modules/timeline/jobs/create-audit-log-from-internal-event';
 import { UpsertTimelineActivityFromInternalEvent } from 'src/modules/timeline/jobs/upsert-timeline-activity-from-internal-event.job';
 
@@ -19,18 +20,24 @@ export class EntityEventsToDbListener {
   ) {}
 
   @OnEvent('*.created')
-  async handleCreate(payload: ObjectRecordCreateEvent<any>) {
+  async handleCreate(
+    payload: WorkspaceEventBatch<ObjectRecordCreateEvent<any>>,
+  ) {
     return this.handle(payload);
   }
 
   @OnEvent('*.updated')
-  async handleUpdate(payload: ObjectRecordUpdateEvent<any>) {
-    payload.properties.diff = objectRecordChangedValues(
-      payload.properties.before,
-      payload.properties.after,
-      payload.properties.updatedFields,
-      payload.objectMetadata,
-    );
+  async handleUpdate(
+    payload: WorkspaceEventBatch<ObjectRecordUpdateEvent<any>>,
+  ) {
+    for (const eventPayload of payload.events) {
+      eventPayload.properties.diff = objectRecordChangedValues(
+        eventPayload.properties.before,
+        eventPayload.properties.after,
+        eventPayload.properties.updatedFields,
+        eventPayload.objectMetadata,
+      );
+    }
 
     return this.handle(payload);
   }
@@ -41,19 +48,21 @@ export class EntityEventsToDbListener {
   // @OnEvent('*.restored') - TODO: implement when we soft delete has been implemented
   // ....
 
-  private async handle(payload: ObjectRecordBaseEvent) {
-    if (!payload.objectMetadata?.isAuditLogged) {
-      return;
-    }
+  private async handle(payload: WorkspaceEventBatch<ObjectRecordBaseEvent>) {
+    for (const eventPayload of payload.events) {
+      if (!eventPayload.objectMetadata?.isAuditLogged) {
+        return;
+      }
 
-    this.messageQueueService.add<ObjectRecordBaseEvent>(
-      CreateAuditLogFromInternalEvent.name,
-      payload,
-    );
+      await this.messageQueueService.add<ObjectRecordBaseEvent>(
+        CreateAuditLogFromInternalEvent.name,
+        eventPayload,
+      );
 
-    this.messageQueueService.add<ObjectRecordBaseEvent>(
-      UpsertTimelineActivityFromInternalEvent.name,
-      payload,
-    );
+      await this.messageQueueService.add<ObjectRecordBaseEvent>(
+        UpsertTimelineActivityFromInternalEvent.name,
+        eventPayload,
+      );
+    }
   }
 }

packages/twenty-server/src/engine/api/graphql/workspace-query-runner/listeners/telemetry.listener.ts

@@ -4,6 +4,7 @@ import { OnEvent } from '@nestjs/event-emitter';
 import { AnalyticsService } from 'src/engine/core-modules/analytics/analytics.service';
 import { EnvironmentService } from 'src/engine/integrations/environment/environment.service';
 import { ObjectRecordCreateEvent } from 'src/engine/integrations/event-emitter/types/object-record-create.event';
+import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/workspace-event.type';
 
 @Injectable()
 export class TelemetryListener {
@@ -13,36 +14,48 @@ export class TelemetryListener {
   ) {}
 
   @OnEvent('*.created')
-  async handleAllCreate(payload: ObjectRecordCreateEvent<any>) {
-    await this.analyticsService.create(
-      {
-        type: 'track',
-        data: {
-          eventName: payload.name,
-        },
-      },
-      payload.userId,
-      payload.workspaceId,
-      '', // voluntarely not retrieving this
-      '', // to avoid slowing down
-      this.environmentService.get('SERVER_URL'),
+  async handleAllCreate(
+    payload: WorkspaceEventBatch<ObjectRecordCreateEvent<any>>,
+  ) {
+    await Promise.all(
+      payload.events.map((eventPayload) =>
+        this.analyticsService.create(
+          {
+            type: 'track',
+            data: {
+              eventName: payload.name,
+            },
+          },
+          eventPayload.userId,
+          payload.workspaceId,
+          '', // voluntarily not retrieving this
+          '', // to avoid slowing down
+          this.environmentService.get('SERVER_URL'),
+        ),
+      ),
     );
   }
 
   @OnEvent('user.signup')
-  async handleUserSignup(payload: ObjectRecordCreateEvent<any>) {
-    await this.analyticsService.create(
-      {
-        type: 'track',
-        data: {
-          eventName: 'user.signup',
-        },
-      },
-      payload.userId,
-      payload.workspaceId,
-      '',
-      '',
-      this.environmentService.get('SERVER_URL'),
+  async handleUserSignup(
+    payload: WorkspaceEventBatch<ObjectRecordCreateEvent<any>>,
+  ) {
+    await Promise.all(
+      payload.events.map(async (eventPayload) =>
+        this.analyticsService.create(
+          {
+            type: 'track',
+            data: {
+              eventName: 'user.signup',
+            },
+          },
+          eventPayload.userId,
+          payload.workspaceId,
+          '',
+          '',
+          this.environmentService.get('SERVER_URL'),
+        ),
+      ),
     );
   }
 }

packages/twenty-server/src/engine/api/graphql/workspace-query-runner/workspace-query-runner.service.ts

@@ -1,5 +1,4 @@
 import { Injectable, Logger } from '@nestjs/common';
-import { EventEmitter2 } from '@nestjs/event-emitter';
 
 import isEmpty from 'lodash.isempty';
 import { DataSource } from 'typeorm';
@@ -52,6 +51,8 @@ import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.
 import { computeObjectTargetTable } from 'src/engine/utils/compute-object-target-table.util';
 import { isQueryTimeoutError } from 'src/engine/utils/query-timeout.util';
 import { WorkspaceDataSourceService } from 'src/engine/workspace-datasource/workspace-datasource.service';
+import { WorkspaceEventEmitter } from 'src/engine/workspace-event-emitter/workspace-event-emitter';
+import { isDefined } from 'src/utils/is-defined';
 
 import {
   PGGraphQLMutation,
@@ -75,7 +76,7 @@ export class WorkspaceQueryRunnerService {
     private readonly queryResultGettersFactory: QueryResultGettersFactory,
     @InjectMessageQueue(MessageQueue.webhookQueue)
     private readonly messageQueueService: MessageQueueService,
-    private readonly eventEmitter: EventEmitter2,
+    private readonly workspaceEventEmitter: WorkspaceEventEmitter,
     private readonly workspaceQueryHookService: WorkspaceQueryHookService,
     private readonly environmentService: EnvironmentService,
     private readonly duplicateService: DuplicateService,
@@ -295,18 +296,21 @@ export class WorkspaceQueryRunnerService {
       options,
     );
 
-    parsedResults.forEach((record) => {
-      this.eventEmitter.emit(`${objectMetadataItem.nameSingular}.created`, {
-        name: `${objectMetadataItem.nameSingular}.created`,
-        workspaceId: authContext.workspace.id,
-        userId: authContext.user?.id,
-        recordId: record.id,
-        objectMetadata: objectMetadataItem,
-        properties: {
-          after: record,
-        },
-      } satisfies ObjectRecordCreateEvent<any>);
-    });
+    this.workspaceEventEmitter.emit(
+      `${objectMetadataItem.nameSingular}.created`,
+      parsedResults.map(
+        (record) =>
+          ({
+            userId: authContext.user?.id,
+            recordId: record.id,
+            objectMetadata: objectMetadataItem,
+            properties: {
+              after: record,
+            },
+          }) satisfies ObjectRecordCreateEvent<any>,
+      ),
+      authContext.workspace.id,
+    );
 
     return parsedResults;
   }
@@ -431,18 +435,22 @@ export class WorkspaceQueryRunnerService {
       options,
     );
 
-    this.eventEmitter.emit(`${objectMetadataItem.nameSingular}.updated`, {
-      name: `${objectMetadataItem.nameSingular}.updated`,
-      workspaceId: authContext.workspace.id,
-      userId: authContext.user?.id,
-      recordId: existingRecord.id,
-      objectMetadata: objectMetadataItem,
-      properties: {
-        updatedFields: Object.keys(args.data),
-        before: this.removeNestedProperties(existingRecord as Record),
-        after: this.removeNestedProperties(parsedResults?.[0]),
-      },
-    } satisfies ObjectRecordUpdateEvent<any>);
+    this.workspaceEventEmitter.emit(
+      `${objectMetadataItem.nameSingular}.updated`,
+      [
+        {
+          userId: authContext.user?.id,
+          recordId: existingRecord.id,
+          objectMetadata: objectMetadataItem,
+          properties: {
+            updatedFields: Object.keys(args.data),
+            before: this.removeNestedProperties(existingRecord as Record),
+            after: this.removeNestedProperties(parsedResults?.[0]),
+          },
+        } satisfies ObjectRecordUpdateEvent<any>,
+      ],
+      authContext.workspace.id,
+    );
 
     return parsedResults?.[0];
   }
@@ -504,30 +512,36 @@ export class WorkspaceQueryRunnerService {
       options,
     );
 
-    parsedResults.forEach((record) => {
-      const existingRecord = mappedRecords.get(record.id);
+    const eventsToEmit: ObjectRecordUpdateEvent<any>[] = parsedResults
+      .map((record) => {
+        const existingRecord = mappedRecords.get(record.id);
 
-      if (!existingRecord) {
-        this.logger.warn(
-          `Record with id ${record.id} not found in the database`,
-        );
+        if (!existingRecord) {
+          this.logger.warn(
+            `Record with id ${record.id} not found in the database`,
+          );
 
-        return;
-      }
+          return;
+        }
 
-      this.eventEmitter.emit(`${objectMetadataItem.nameSingular}.updated`, {
-        name: `${objectMetadataItem.nameSingular}.updated`,
-        workspaceId: authContext.workspace.id,
-        userId: authContext.user?.id,
-        recordId: existingRecord.id,
-        objectMetadata: objectMetadataItem,
-        properties: {
-          updatedFields: Object.keys(args.data),
-          before: this.removeNestedProperties(existingRecord as Record),
-          after: this.removeNestedProperties(record),
-        },
-      } satisfies ObjectRecordUpdateEvent<any>);
-    });
+        return {
+          userId: authContext.user?.id,
+          recordId: existingRecord.id,
+          objectMetadata: objectMetadataItem,
+          properties: {
+            updatedFields: Object.keys(args.data),
+            before: this.removeNestedProperties(existingRecord as Record),
+            after: this.removeNestedProperties(record),
+          },
+        };
+      })
+      .filter(isDefined);
+
+    this.workspaceEventEmitter.emit(
+      `${objectMetadataItem.nameSingular}.updated`,
+      eventsToEmit,
+      authContext.workspace.id,
+    );
 
     return parsedResults;
   }
@@ -580,18 +594,21 @@ export class WorkspaceQueryRunnerService {
       options,
     );
 
-    parsedResults.forEach((record) => {
-      this.eventEmitter.emit(`${objectMetadataItem.nameSingular}.deleted`, {
-        name: `${objectMetadataItem.nameSingular}.deleted`,
-        workspaceId: authContext.workspace.id,
-        userId: authContext.user?.id,
-        recordId: record.id,
-        objectMetadata: objectMetadataItem,
-        properties: {
-          before: this.removeNestedProperties(record),
-        },
-      } satisfies ObjectRecordDeleteEvent<any>);
-    });
+    this.workspaceEventEmitter.emit(
+      `${objectMetadataItem.nameSingular}.deleted`,
+      parsedResults.map(
+        (record) =>
+          ({
+            userId: authContext.user?.id,
+            recordId: record.id,
+            objectMetadata: objectMetadataItem,
+            properties: {
+              before: this.removeNestedProperties(record),
+            },
+          }) satisfies ObjectRecordDeleteEvent<any>,
+      ),
+      authContext.workspace.id,
+    );
 
     return parsedResults;
   }
@@ -644,19 +661,23 @@ export class WorkspaceQueryRunnerService {
       options,
     );
 
-    this.eventEmitter.emit(`${objectMetadataItem.nameSingular}.deleted`, {
-      name: `${objectMetadataItem.nameSingular}.deleted`,
-      workspaceId: authContext.workspace.id,
-      userId: authContext.user?.id,
-      recordId: args.id,
-      objectMetadata: objectMetadataItem,
-      properties: {
-        before: {
-          ...(existingRecord ?? {}),
-          ...this.removeNestedProperties(parsedResults?.[0]),
-        },
-      },
-    } satisfies ObjectRecordDeleteEvent<any>);
+    this.workspaceEventEmitter.emit(
+      `${objectMetadataItem.nameSingular}.deleted`,
+      [
+        {
+          userId: authContext.user?.id,
+          recordId: args.id,
+          objectMetadata: objectMetadataItem,
+          properties: {
+            before: {
+              ...(existingRecord ?? {}),
+              ...this.removeNestedProperties(parsedResults?.[0]),
+            },
+          },
+        } satisfies ObjectRecordDeleteEvent<any>,
+      ],
+      authContext.workspace.id,
+    );
 
     return parsedResults?.[0];
   }

packages/twenty-server/src/engine/core-modules/billing/listeners/billing-workspace-member.listener.ts

@@ -10,6 +10,7 @@ import { ObjectRecordCreateEvent } from 'src/engine/integrations/event-emitter/t
 import { InjectMessageQueue } from 'src/engine/integrations/message-queue/decorators/message-queue.decorator';
 import { MessageQueue } from 'src/engine/integrations/message-queue/message-queue.constants';
 import { MessageQueueService } from 'src/engine/integrations/message-queue/services/message-queue.service';
+import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/workspace-event.type';
 import { WorkspaceMemberWorkspaceEntity } from 'src/modules/workspace-member/standard-objects/workspace-member.workspace-entity';
 
 @Injectable()
@@ -23,7 +24,9 @@ export class BillingWorkspaceMemberListener {
   @OnEvent('workspaceMember.created')
   @OnEvent('workspaceMember.deleted')
   async handleCreateOrDeleteEvent(
-    payload: ObjectRecordCreateEvent<WorkspaceMemberWorkspaceEntity>,
+    payload: WorkspaceEventBatch<
+      ObjectRecordCreateEvent<WorkspaceMemberWorkspaceEntity>
+    >,
   ) {
     if (!this.environmentService.get('IS_BILLING_ENABLED')) {
       return;