bugfix: escape destroyed objects on workers (#9719)

# This PR - Fixes #9358 @FelixMalfait please check this workaround --------- Co-authored-by: Félix Malfait <[email protected]>
twentyhq · Jan 23, 2025 · bbd3af1 · bbd3af1
1 parent bbb0c9a
commit bbd3af1
Show file tree

Hide file tree

Showing 11 changed files with 134 additions and 80 deletions.
diff --git a/...r/src/engine/api/graphql/workspace-query-runner/listeners/entity-events-to-db.listener.ts b/...r/src/engine/api/graphql/workspace-query-runner/listeners/entity-events-to-db.listener.ts
@@ -1,17 +1,21 @@
 import { Injectable } from '@nestjs/common';
 
+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';
 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 { ObjectRecordEvent } from 'src/engine/core-modules/event-emitter/types/object-record-event.event';
+import { ObjectRecordNonDestructiveEvent } from 'src/engine/core-modules/event-emitter/types/object-record-non-destructive-event';
+import { 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 { ObjectRecordBaseEvent } from 'src/engine/core-modules/event-emitter/types/object-record.base.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 { 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';
-import { OnDatabaseBatchEvent } from 'src/engine/api/graphql/graphql-query-runner/decorators/on-database-batch-event.decorator';
 import { CallWebhookJobsJob } from 'src/modules/webhook/jobs/call-webhook-jobs.job';
-import { DatabaseEventAction } from 'src/engine/api/graphql/graphql-query-runner/enums/database-event-action';
 
 @Injectable()
 export class EntityEventsToDbListener {
@@ -24,47 +28,66 @@ export class EntityEventsToDbListener {
 
   @OnDatabaseBatchEvent('*', DatabaseEventAction.CREATED)
   async handleCreate(batchEvent: WorkspaceEventBatch<ObjectRecordCreateEvent>) {
-    return this.handle(batchEvent);
+    return this.handleEvent(batchEvent, DatabaseEventAction.CREATED);
   }
 
   @OnDatabaseBatchEvent('*', DatabaseEventAction.UPDATED)
   async handleUpdate(batchEvent: WorkspaceEventBatch<ObjectRecordUpdateEvent>) {
-    return this.handle(batchEvent);
+    return this.handleEvent(batchEvent, DatabaseEventAction.UPDATED);
   }
 
   @OnDatabaseBatchEvent('*', DatabaseEventAction.DELETED)
-  async handleDelete(batchEvent: WorkspaceEventBatch<ObjectRecordUpdateEvent>) {
-    return this.handle(batchEvent);
+  async handleDelete(batchEvent: WorkspaceEventBatch<ObjectRecordDeleteEvent>) {
+    return this.handleEvent(batchEvent, DatabaseEventAction.DELETED);
+  }
+
+  @OnDatabaseBatchEvent('*', DatabaseEventAction.RESTORED)
+  async handleRestore(
+    batchEvent: WorkspaceEventBatch<ObjectRecordRestoreEvent>,
+  ) {
+    return this.handleEvent(batchEvent, DatabaseEventAction.RESTORED);
   }
 
   @OnDatabaseBatchEvent('*', DatabaseEventAction.DESTROYED)
   async handleDestroy(
-    batchEvent: WorkspaceEventBatch<ObjectRecordUpdateEvent>,
+    batchEvent: WorkspaceEventBatch<ObjectRecordDestroyEvent>,
   ) {
-    return this.handle(batchEvent);
+    return this.handleEvent(batchEvent, DatabaseEventAction.DESTROYED);
   }
 
-  private async handle(batchEvent: WorkspaceEventBatch<ObjectRecordBaseEvent>) {
+  private async handleEvent<T extends ObjectRecordEvent>(
+    batchEvent: WorkspaceEventBatch<T>,
+    action: DatabaseEventAction,
+  ) {
     const filteredEvents = batchEvent.events.filter(
       (event) => event.objectMetadata?.isAuditLogged,
     );
 
-    await this.entityEventsToDbQueueService.add<
-      WorkspaceEventBatch<ObjectRecordBaseEvent>
-    >(CreateAuditLogFromInternalEvent.name, {
-      ...batchEvent,
-      events: filteredEvents,
-    });
-
-    await this.entityEventsToDbQueueService.add<
-      WorkspaceEventBatch<ObjectRecordBaseEvent>
-    >(UpsertTimelineActivityFromInternalEvent.name, {
-      ...batchEvent,
-      events: filteredEvents,
-    });
-
-    await this.webhookQueueService.add<
-      WorkspaceEventBatch<ObjectRecordBaseEvent>
-    >(CallWebhookJobsJob.name, batchEvent, { retryLimit: 3 });
+    await Promise.all([
+      this.webhookQueueService.add<WorkspaceEventBatch<T>>(
+        CallWebhookJobsJob.name,
+        batchEvent,
+        {
+          retryLimit: 3,
+        },
+      ),
+      this.entityEventsToDbQueueService.add<WorkspaceEventBatch<T>>(
+        CreateAuditLogFromInternalEvent.name,
+        {
+          ...batchEvent,
+          events: filteredEvents,
+        },
+      ),
+      ...(action !== DatabaseEventAction.DESTROYED
+        ? [
+            this.entityEventsToDbQueueService.add<
+              WorkspaceEventBatch<ObjectRecordNonDestructiveEvent>
+            >(UpsertTimelineActivityFromInternalEvent.name, {
+              ...batchEvent,
+              events: filteredEvents,
+            }),
+          ]
+        : []),
+    ]);
   }
 }
diff --git a/packages/twenty-server/src/engine/core-modules/auth/auth.module.ts b/packages/twenty-server/src/engine/core-modules/auth/auth.module.ts
@@ -17,6 +17,7 @@ import { MicrosoftAPIsService } from 'src/engine/core-modules/auth/services/micr
 // import { OAuthService } from 'src/engine/core-modules/auth/services/oauth.service';
 import { ResetPasswordService } from 'src/engine/core-modules/auth/services/reset-password.service';
 import { SignInUpService } from 'src/engine/core-modules/auth/services/sign-in-up.service';
+import { SocialSsoService } from 'src/engine/core-modules/auth/services/social-sso.service';
 import { SamlAuthStrategy } from 'src/engine/core-modules/auth/strategies/saml.auth.strategy';
 import { AccessTokenService } from 'src/engine/core-modules/auth/token/services/access-token.service';
 import { LoginTokenService } from 'src/engine/core-modules/auth/token/services/login-token.service';
@@ -43,7 +44,6 @@ import { DataSourceModule } from 'src/engine/metadata-modules/data-source/data-s
 import { WorkspaceDataSourceModule } from 'src/engine/workspace-datasource/workspace-datasource.module';
 import { WorkspaceManagerModule } from 'src/engine/workspace-manager/workspace-manager.module';
 import { ConnectedAccountModule } from 'src/modules/connected-account/connected-account.module';
-import { SocialSsoService } from 'src/engine/core-modules/auth/services/social-sso.service';
 
 import { AuthResolver } from './auth.resolver';
 

diff --git a/...es/twenty-server/src/engine/core-modules/event-emitter/types/object-record-event.event.ts b/...es/twenty-server/src/engine/core-modules/event-emitter/types/object-record-event.event.ts
@@ -0,0 +1,12 @@
+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 { 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';
+
+export type ObjectRecordEvent<T = object> =
+  | ObjectRecordUpdateEvent<T>
+  | ObjectRecordDeleteEvent<T>
+  | ObjectRecordCreateEvent<T>
+  | ObjectRecordDestroyEvent<T>
+  | ObjectRecordRestoreEvent<T>;
diff --git a/...server/src/engine/core-modules/event-emitter/types/object-record-non-destructive-event.ts b/...server/src/engine/core-modules/event-emitter/types/object-record-non-destructive-event.ts
@@ -0,0 +1,10 @@
+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 { 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';
+
+export type ObjectRecordNonDestructiveEvent =
+  | ObjectRecordCreateEvent
+  | ObjectRecordUpdateEvent
+  | ObjectRecordDeleteEvent
+  | ObjectRecordRestoreEvent;
diff --git a/.../twenty-server/src/engine/core-modules/event-emitter/types/object-record-restore.event.ts b/.../twenty-server/src/engine/core-modules/event-emitter/types/object-record-restore.event.ts
@@ -1,9 +1,9 @@
-import { ObjectRecordBaseEvent } from 'src/engine/core-modules/event-emitter/types/object-record.base.event';
+import { ObjectRecordCreateEvent } from 'src/engine/core-modules/event-emitter/types/object-record-create.event';
 
 export class ObjectRecordRestoreEvent<
   T = object,
-> extends ObjectRecordBaseEvent<T> {
+> extends ObjectRecordCreateEvent<T> {
   properties: {
-    before: T;
+    after: T;
   };
 }
diff --git a/...s/twenty-server/src/engine/core-modules/event-emitter/types/object-record-update.event.ts b/...s/twenty-server/src/engine/core-modules/event-emitter/types/object-record-update.event.ts
@@ -1,13 +1,13 @@
-import { ObjectRecordBaseEvent } from 'src/engine/core-modules/event-emitter/types/object-record.base.event';
 import { ObjectRecordDiff } from 'src/engine/core-modules/event-emitter/types/object-record-diff';
+import { ObjectRecordBaseEvent } from 'src/engine/core-modules/event-emitter/types/object-record.base.event';
 
 export class ObjectRecordUpdateEvent<
   T = object,
 > extends ObjectRecordBaseEvent<T> {
   properties: {
     updatedFields?: string[];
+    diff?: Partial<ObjectRecordDiff<T>>;
     before: T;
     after: T;
-    diff?: Partial<ObjectRecordDiff<T>>;
   };
 }
diff --git a/...es/messaging/message-participant-manager/listeners/message-participant-person.listener.ts b/...es/messaging/message-participant-manager/listeners/message-participant-person.listener.ts
@@ -1,5 +1,7 @@
 import { Injectable } from '@nestjs/common';
 
+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';
 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 { objectRecordChangedProperties as objectRecordUpdateEventChangedProperties } from 'src/engine/core-modules/event-emitter/utils/object-record-changed-properties.util';
@@ -16,8 +18,6 @@ import {
   MessageParticipantUnmatchParticipantJobData,
 } from 'src/modules/messaging/message-participant-manager/jobs/message-participant-unmatch-participant.job';
 import { PersonWorkspaceEntity } from 'src/modules/person/standard-objects/person.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';
 
 @Injectable()
 export class MessageParticipantPersonListener {

diff --git a/packages/twenty-server/src/modules/timeline/jobs/create-audit-log-from-internal-event.ts b/packages/twenty-server/src/modules/timeline/jobs/create-audit-log-from-internal-event.ts
@@ -1,4 +1,4 @@
-import { ObjectRecordBaseEvent } from 'src/engine/core-modules/event-emitter/types/object-record.base.event';
+import { ObjectRecordEvent } from 'src/engine/core-modules/event-emitter/types/object-record-event.event';
 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';
@@ -20,7 +20,7 @@ export class CreateAuditLogFromInternalEvent {
 
   @Process(CreateAuditLogFromInternalEvent.name)
   async handle(
-    workspaceEventBatch: WorkspaceEventBatch<ObjectRecordBaseEvent>,
+    workspaceEventBatch: WorkspaceEventBatch<ObjectRecordEvent>,
   ): Promise<void> {
     for (const eventData of workspaceEventBatch.events) {
       let workspaceMemberId: string | null = null;
@@ -34,16 +34,14 @@ export class CreateAuditLogFromInternalEvent {
         workspaceMemberId = workspaceMember.id;
       }
 
-      if (eventData.properties.diff) {
-        // we remove "before" and "after" property for a cleaner/slimmer event payload
-        eventData.properties = {
-          diff: eventData.properties.diff,
-        };
-      }
-
       await this.auditLogRepository.insert(
         workspaceEventBatch.name,
-        eventData.properties,
+        'diff' in eventData.properties
+          ? {
+              // we remove "before" and "after" property for a cleaner/slimmer event payload
+              diff: eventData.properties.diff,
+            }
+          : eventData.properties,
         workspaceMemberId,
         workspaceEventBatch.name.split('.')[0],
         eventData.objectMetadata.id,

diff --git a/...enty-server/src/modules/timeline/jobs/upsert-timeline-activity-from-internal-event.job.ts b/...enty-server/src/modules/timeline/jobs/upsert-timeline-activity-from-internal-event.job.ts
@@ -1,4 +1,4 @@
-import { ObjectRecordBaseEvent } from 'src/engine/core-modules/event-emitter/types/object-record.base.event';
+import { ObjectRecordNonDestructiveEvent } from 'src/engine/core-modules/event-emitter/types/object-record-non-destructive-event';
 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';
@@ -7,7 +7,6 @@ import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/wo
 import { TimelineActivityService } from 'src/modules/timeline/services/timeline-activity.service';
 import { WorkspaceMemberRepository } from 'src/modules/workspace-member/repositories/workspace-member.repository';
 import { WorkspaceMemberWorkspaceEntity } from 'src/modules/workspace-member/standard-objects/workspace-member.workspace-entity';
-import { TimelineActivityWorkspaceEntity } from 'src/modules/timeline/standard-objects/timeline-activity.workspace-entity';
 
 @Processor(MessageQueue.entityEventsToDbQueue)
 export class UpsertTimelineActivityFromInternalEvent {
@@ -19,9 +18,7 @@ export class UpsertTimelineActivityFromInternalEvent {
 
   @Process(UpsertTimelineActivityFromInternalEvent.name)
   async handle(
-    workspaceEventBatch: WorkspaceEventBatch<
-      ObjectRecordBaseEvent<TimelineActivityWorkspaceEntity>
-    >,
+    workspaceEventBatch: WorkspaceEventBatch<ObjectRecordNonDestructiveEvent>,
   ): Promise<void> {
     for (const eventData of workspaceEventBatch.events) {
       if (eventData.userId) {
@@ -33,13 +30,6 @@ export class UpsertTimelineActivityFromInternalEvent {
         eventData.workspaceMemberId = workspaceMember.id;
       }
 
-      if (eventData.properties.diff) {
-        // we remove "before" and "after" property for a cleaner/slimmer event payload
-        eventData.properties = {
-          diff: eventData.properties.diff,
-        };
-      }
-
       // Temporary
       // We ignore every that is not a LinkedObject or a Business Object
       if (
@@ -51,7 +41,16 @@ export class UpsertTimelineActivityFromInternalEvent {
       }
 
       await this.timelineActivityService.upsertEvent({
-        event: eventData,
+        event:
+          // we remove "before" and "after" property for a cleaner/slimmer event payload
+          'diff' in eventData.properties && eventData.properties.diff
+            ? {
+                ...eventData,
+                properties: {
+                  diff: eventData.properties.diff,
+                },
+              }
+            : eventData,
         eventName: workspaceEventBatch.name,
         workspaceId: workspaceEventBatch.workspaceId,
       });