fix(messaging): emit channel and account deletion events from core metadata services (#21491)

/closes #21425

<!-- This is an auto-generated description by cubic. -->
<a
href="https://cubic.dev/pr/twentyhq/twenty/pull/21491?utm_source=github"
target="_blank" rel="noopener noreferrer"
data-no-image-dialog="true"><picture><source
media="(prefers-color-scheme: dark)"
srcset="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"><source
media="(prefers-color-scheme: light)"
srcset="https://www.cubic.dev/buttons/review-in-cubic-light.svg"><img
alt="Review in cubic"
src="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"></picture></a>
<!-- End of auto-generated description by cubic. -->

packages/twenty-server/src/modules/calendar/calendar-event-cleaner/listeners/calendar-event-cleaner-calendar-channel.listener.ts

@@ -1,18 +1,18 @@
 import { Injectable } from '@nestjs/common';
 
-import { type ObjectRecordDeleteEvent } from 'twenty-shared/database-events';
+import { isDefined } from 'twenty-shared/utils';
 
+import { OnCustomBatchEvent } from 'src/engine/api/graphql/graphql-query-runner/decorators/on-custom-batch-event.decorator';
 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-batch.type';
-import { type CalendarChannelEntity } from 'src/engine/metadata-modules/calendar-channel/entities/calendar-channel.entity';
+import { CALENDAR_CHANNEL_DELETED_EVENT } from 'src/engine/metadata-modules/calendar-channel/constants/calendar-channel-deleted.constant';
+import { type CalendarChannelDeletedEvent } from 'src/engine/metadata-modules/calendar-channel/types/calendar-channel-deleted.type';
+import { CustomWorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/custom-workspace-batch-event.type';
 import {
   CalendarChannelDeletionCleanupJob,
   type CalendarChannelDeletionCleanupJobData,
 } from 'src/modules/calendar/calendar-event-cleaner/jobs/calendar-channel-deletion-cleanup.job';
-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 CalendarEventCleanerCalendarChannelListener {
@@ -21,19 +21,23 @@ export class CalendarEventCleanerCalendarChannelListener {
     private readonly calendarQueueService: MessageQueueService,
   ) {}
 
-  @OnDatabaseBatchEvent('calendarChannel', DatabaseEventAction.DESTROYED)
-  async handleDestroyedEvent(
-    payload: WorkspaceEventBatch<
-      ObjectRecordDeleteEvent<CalendarChannelEntity>
-    >,
+  @OnCustomBatchEvent(CALENDAR_CHANNEL_DELETED_EVENT)
+  async handleDeletedEvent(
+    batchEvent: CustomWorkspaceEventBatch<CalendarChannelDeletedEvent>,
   ) {
+    const { workspaceId } = batchEvent;
+
+    if (!isDefined(workspaceId)) {
+      return;
+    }
+
     await Promise.all(
-      payload.events.map((eventPayload) =>
+      batchEvent.events.map((event) =>
         this.calendarQueueService.add<CalendarChannelDeletionCleanupJobData>(
           CalendarChannelDeletionCleanupJob.name,
           {
-            workspaceId: payload.workspaceId,
-            calendarChannelId: eventPayload.recordId,
+            workspaceId,
+            calendarChannelId: event.calendarChannelId,
           },
         ),
       ),

packages/twenty-server/src/modules/calendar/calendar-event-cleaner/listeners/calendar-event-cleaner-connected-account.listener.ts

@@ -1,18 +1,18 @@
 import { Injectable } from '@nestjs/common';
 
-import { type ObjectRecordDeleteEvent } from 'twenty-shared/database-events';
+import { isDefined } from 'twenty-shared/utils';
 
+import { OnCustomBatchEvent } from 'src/engine/api/graphql/graphql-query-runner/decorators/on-custom-batch-event.decorator';
 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-batch.type';
+import { CONNECTED_ACCOUNT_DELETED_EVENT } from 'src/engine/metadata-modules/connected-account/constants/connected-account-deleted.constant';
+import { type ConnectedAccountDeletedEvent } from 'src/engine/metadata-modules/connected-account/types/connected-account-deleted.type';
+import { CustomWorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/custom-workspace-batch-event.type';
 import {
   DeleteConnectedAccountAssociatedCalendarDataJob,
   type DeleteConnectedAccountAssociatedCalendarDataJobData,
 } from 'src/modules/calendar/calendar-event-cleaner/jobs/delete-connected-account-associated-calendar-data.job';
-import { type ConnectedAccountEntity } from 'src/engine/metadata-modules/connected-account/entities/connected-account.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 CalendarEventCleanerConnectedAccountListener {
@@ -21,19 +21,23 @@ export class CalendarEventCleanerConnectedAccountListener {
     private readonly calendarQueueService: MessageQueueService,
   ) {}
 
-  @OnDatabaseBatchEvent('connectedAccount', DatabaseEventAction.DESTROYED)
-  async handleDestroyedEvent(
-    payload: WorkspaceEventBatch<
-      ObjectRecordDeleteEvent<ConnectedAccountEntity>
-    >,
+  @OnCustomBatchEvent(CONNECTED_ACCOUNT_DELETED_EVENT)
+  async handleDeletedEvent(
+    batchEvent: CustomWorkspaceEventBatch<ConnectedAccountDeletedEvent>,
   ) {
+    const { workspaceId } = batchEvent;
+
+    if (!isDefined(workspaceId)) {
+      return;
+    }
+
     await Promise.all(
-      payload.events.map((eventPayload) =>
+      batchEvent.events.map((event) =>
         this.calendarQueueService.add<DeleteConnectedAccountAssociatedCalendarDataJobData>(
           DeleteConnectedAccountAssociatedCalendarDataJob.name,
           {
-            workspaceId: payload.workspaceId,
-            connectedAccountId: eventPayload.recordId,
+            workspaceId,
+            connectedAccountId: event.connectedAccountId,
           },
         ),
       ),

packages/twenty-server/src/modules/connected-account/listeners/connected-account.listener.ts

@@ -1,17 +1,17 @@
 import { Injectable } from '@nestjs/common';
 import { InjectRepository } from '@nestjs/typeorm';
 
-import { type ObjectRecordDeleteEvent } from 'twenty-shared/database-events';
+import { isDefined } from 'twenty-shared/utils';
 import { type Repository } from 'typeorm';
 
-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 { OnCustomBatchEvent } from 'src/engine/api/graphql/graphql-query-runner/decorators/on-custom-batch-event.decorator';
 import { UserWorkspaceEntity } from 'src/engine/core-modules/user-workspace/user-workspace.entity';
+import { CONNECTED_ACCOUNT_DELETED_EVENT } from 'src/engine/metadata-modules/connected-account/constants/connected-account-deleted.constant';
+import { type ConnectedAccountDeletedEvent } from 'src/engine/metadata-modules/connected-account/types/connected-account-deleted.type';
 import { GlobalWorkspaceOrmManager } from 'src/engine/twenty-orm/global-workspace-datasource/global-workspace-orm.manager';
 import { buildSystemAuthContext } from 'src/engine/twenty-orm/utils/build-system-auth-context.util';
-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';
 import { AccountsToReconnectService } from 'src/modules/connected-account/services/accounts-to-reconnect.service';
-import { type ConnectedAccountEntity } from 'src/engine/metadata-modules/connected-account/entities/connected-account.entity';
 
 @Injectable()
 export class ConnectedAccountListener {
@@ -22,35 +22,32 @@ export class ConnectedAccountListener {
     private readonly userWorkspaceRepository: Repository<UserWorkspaceEntity>,
   ) {}
 
-  @OnDatabaseBatchEvent('connectedAccount', DatabaseEventAction.DESTROYED)
-  async handleDestroyedEvent(
-    payload: WorkspaceEventBatch<
-      ObjectRecordDeleteEvent<ConnectedAccountEntity>
-    >,
+  @OnCustomBatchEvent(CONNECTED_ACCOUNT_DELETED_EVENT)
+  async handleDeletedEvent(
+    batchEvent: CustomWorkspaceEventBatch<ConnectedAccountDeletedEvent>,
   ) {
-    const workspaceId = payload.workspaceId;
+    const { workspaceId } = batchEvent;
+
+    if (!isDefined(workspaceId)) {
+      return;
+    }
+
     const authContext = buildSystemAuthContext(workspaceId);
 
     await this.globalWorkspaceOrmManager.executeInWorkspaceContext(async () => {
-      for (const eventPayload of payload.events) {
-        const userWorkspaceId = eventPayload.properties.before.userWorkspaceId;
-
+      for (const event of batchEvent.events) {
         const userWorkspace = await this.userWorkspaceRepository.findOne({
-          where: { id: userWorkspaceId },
+          where: { id: event.userWorkspaceId },
         });
 
         if (!userWorkspace) {
           continue;
         }
 
-        const userId = userWorkspace.userId;
-
-        const connectedAccountId = eventPayload.properties.before.id;
-
         await this.accountsToReconnectService.removeAccountToReconnect(
-          userId,
+          userWorkspace.userId,
           workspaceId,
-          connectedAccountId,
+          event.connectedAccountId,
         );
       }
     }, authContext);

packages/twenty-server/src/modules/messaging/message-cleaner/listeners/messaging-message-cleaner-connected-account.listener.ts

@@ -1,18 +1,18 @@
 import { Injectable } from '@nestjs/common';
 
-import { type ObjectRecordDeleteEvent } from 'twenty-shared/database-events';
+import { isDefined } from 'twenty-shared/utils';
 
+import { OnCustomBatchEvent } from 'src/engine/api/graphql/graphql-query-runner/decorators/on-custom-batch-event.decorator';
 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-batch.type';
-import { type ConnectedAccountEntity } from 'src/engine/metadata-modules/connected-account/entities/connected-account.entity';
+import { CONNECTED_ACCOUNT_DELETED_EVENT } from 'src/engine/metadata-modules/connected-account/constants/connected-account-deleted.constant';
+import { type ConnectedAccountDeletedEvent } from 'src/engine/metadata-modules/connected-account/types/connected-account-deleted.type';
+import { CustomWorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/custom-workspace-batch-event.type';
 import {
   MessagingConnectedAccountDeletionCleanupJob,
   type MessagingConnectedAccountDeletionCleanupJobData,
 } from 'src/modules/messaging/message-cleaner/jobs/messaging-connected-account-deletion-cleanup.job';
-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 MessagingMessageCleanerConnectedAccountListener {
@@ -21,19 +21,23 @@ export class MessagingMessageCleanerConnectedAccountListener {
     private readonly messageQueueService: MessageQueueService,
   ) {}
 
-  @OnDatabaseBatchEvent('connectedAccount', DatabaseEventAction.DESTROYED)
-  async handleDestroyedEvent(
-    payload: WorkspaceEventBatch<
-      ObjectRecordDeleteEvent<ConnectedAccountEntity>
-    >,
+  @OnCustomBatchEvent(CONNECTED_ACCOUNT_DELETED_EVENT)
+  async handleDeletedEvent(
+    batchEvent: CustomWorkspaceEventBatch<ConnectedAccountDeletedEvent>,
   ) {
+    const { workspaceId } = batchEvent;
+
+    if (!isDefined(workspaceId)) {
+      return;
+    }
+
     await Promise.all(
-      payload.events.map((eventPayload) =>
+      batchEvent.events.map((event) =>
         this.messageQueueService.add<MessagingConnectedAccountDeletionCleanupJobData>(
           MessagingConnectedAccountDeletionCleanupJob.name,
           {
-            workspaceId: payload.workspaceId,
-            connectedAccountId: eventPayload.recordId,
+            workspaceId,
+            connectedAccountId: event.connectedAccountId,
           },
         ),
       ),

packages/twenty-server/src/modules/messaging/message-cleaner/listeners/messaging-message-cleaner-message-channel.listener.ts

@@ -1,18 +1,18 @@
 import { Injectable } from '@nestjs/common';
 
-import { type ObjectRecordDeleteEvent } from 'twenty-shared/database-events';
+import { isDefined } from 'twenty-shared/utils';
 
+import { OnCustomBatchEvent } from 'src/engine/api/graphql/graphql-query-runner/decorators/on-custom-batch-event.decorator';
 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-batch.type';
-import { type MessageChannelEntity } from 'src/engine/metadata-modules/message-channel/entities/message-channel.entity';
+import { MESSAGE_CHANNEL_DELETED_EVENT } from 'src/engine/metadata-modules/message-channel/constants/message-channel-deleted.constant';
+import { type MessageChannelDeletedEvent } from 'src/engine/metadata-modules/message-channel/types/message-channel-deleted.type';
+import { CustomWorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/custom-workspace-batch-event.type';
 import {
   MessagingMessageChannelDeletionCleanupJob,
   type MessagingMessageChannelDeletionCleanupJobData,
 } from 'src/modules/messaging/message-cleaner/jobs/messaging-message-channel-deletion-cleanup.job';
-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 MessagingMessageCleanerMessageChannelListener {
@@ -21,17 +21,23 @@ export class MessagingMessageCleanerMessageChannelListener {
     private readonly messageQueueService: MessageQueueService,
   ) {}
 
-  @OnDatabaseBatchEvent('messageChannel', DatabaseEventAction.DESTROYED)
-  async handleDestroyedEvent(
-    payload: WorkspaceEventBatch<ObjectRecordDeleteEvent<MessageChannelEntity>>,
+  @OnCustomBatchEvent(MESSAGE_CHANNEL_DELETED_EVENT)
+  async handleDeletedEvent(
+    batchEvent: CustomWorkspaceEventBatch<MessageChannelDeletedEvent>,
   ) {
+    const { workspaceId } = batchEvent;
+
+    if (!isDefined(workspaceId)) {
+      return;
+    }
+
     await Promise.all(
-      payload.events.map((eventPayload) =>
+      batchEvent.events.map((event) =>
         this.messageQueueService.add<MessagingMessageChannelDeletionCleanupJobData>(
           MessagingMessageChannelDeletionCleanupJob.name,
           {
-            workspaceId: payload.workspaceId,
-            messageChannelId: eventPayload.recordId,
+            workspaceId,
+            messageChannelId: event.messageChannelId,
           },
         ),
       ),