6619 modify event emitter to emit an array of events (#6625)

Closes #6619 --------- Co-authored-by: Charles Bochet <charles@twenty.com>
2024-08-20 19:44:29 +02:00
parent 17a1760afd
commit 091c0f83be
41 changed files with 1005 additions and 722 deletions
--- 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
@ -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,40 +20,46 @@ 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);
  }

  @OnEvent('*.deleted')
-  async handleDelete(payload: ObjectRecordUpdateEvent<any>) {
+  async handleDelete(
+    payload: WorkspaceEventBatch<ObjectRecordUpdateEvent<any>>,
+  ) {
    return this.handle(payload);
  }

-  private async handle(payload: ObjectRecordBaseEvent) {
-    if (!payload.objectMetadata?.isAuditLogged) {
-      return;
-    }
-
-    this.messageQueueService.add<ObjectRecordBaseEvent>(
-      CreateAuditLogFromInternalEvent.name,
-      payload,
+  private async handle(payload: WorkspaceEventBatch<ObjectRecordBaseEvent>) {
+    payload.events = payload.events.filter(
+      (event) => event.objectMetadata?.isAuditLogged,
    );

-    this.messageQueueService.add<ObjectRecordBaseEvent>(
-      UpsertTimelineActivityFromInternalEvent.name,
-      payload,
-    );
+    await this.messageQueueService.add<
+      WorkspaceEventBatch<ObjectRecordBaseEvent>
+    >(CreateAuditLogFromInternalEvent.name, payload);
+
+    await this.messageQueueService.add<
+      WorkspaceEventBatch<ObjectRecordBaseEvent>
+    >(UpsertTimelineActivityFromInternalEvent.name, payload);
  }
 }
--- 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
@ -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((eventPayload) =>
+        this.analyticsService.create(
+          {
+            type: 'track',
+            data: {
+              eventName: 'user.signup',
+            },
+          },
+          eventPayload.userId,
+          payload.workspaceId,
+          '',
+          '',
+          this.environmentService.get('SERVER_URL'),
+        ),
+      ),
    );
  }
 }
--- a/packages/twenty-server/src/engine/api/graphql/workspace-query-runner/workspace-query-runner.service.ts
+++ b/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';
@ -55,6 +54,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,
@ -78,7 +79,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,
@ -304,18 +305,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;
  }
@ -440,18 +444,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];
  }
@ -513,30 +521,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;
  }
@ -602,18 +616,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;
  }
@ -744,18 +761,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: this.removeNestedProperties(record),
-        },
-      } satisfies ObjectRecordCreateEvent<any>);
-    });
+    this.workspaceEventEmitter.emit(
+      `${objectMetadataItem.nameSingular}.created`,
+      parsedResults.map(
+        (record) =>
+          ({
+            userId: authContext.user?.id,
+            recordId: record.id,
+            objectMetadata: objectMetadataItem,
+            properties: {
+              after: this.removeNestedProperties(record),
+            },
+          }) satisfies ObjectRecordCreateEvent<any>,
+      ),
+      authContext.workspace.id,
+    );

    return parsedResults;
  }
@ -821,19 +841,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];
  }