messaging cleanup (#19124)

2026-03-30 19:51:00 +05:30
parent 30bdc24bf8
commit 2fccd29ec6
10 changed files with 444 additions and 5 deletions
@@ -0,0 +1,58 @@
+import { Logger, Scope } from '@nestjs/common';
+
+import { FeatureFlagKey } from 'twenty-shared/types';
+
+import { FeatureFlagService } from 'src/engine/core-modules/feature-flag/services/feature-flag.service';
+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 { MessagingMessageCleanerService } from 'src/modules/messaging/message-cleaner/services/messaging-message-cleaner.service';
+
+export type MessagingMessageChannelDeletionCleanupJobData = {
+  workspaceId: string;
+  messageChannelId: string;
+};
+
+@Processor({
+  queueName: MessageQueue.messagingQueue,
+  scope: Scope.REQUEST,
+})
+export class MessagingMessageChannelDeletionCleanupJob {
+  private readonly logger = new Logger(
+    MessagingMessageChannelDeletionCleanupJob.name,
+  );
+
+  constructor(
+    private readonly messageCleanerService: MessagingMessageCleanerService,
+    private readonly featureFlagService: FeatureFlagService,
+  ) {}
+
+  @Process(MessagingMessageChannelDeletionCleanupJob.name)
+  async handle(
+    data: MessagingMessageChannelDeletionCleanupJobData,
+  ): Promise<void> {
+    const isMigrated = await this.featureFlagService.isFeatureEnabled(
+      FeatureFlagKey.IS_CONNECTED_ACCOUNT_MIGRATED,
+      data.workspaceId,
+    );
+
+    if (!isMigrated) {
+      return;
+    }
+
+    this.logger.debug(
+      `WorkspaceId: ${data.workspaceId} Cleaning up message channel message associations for channel ${data.messageChannelId}`,
+    );
+
+    await this.messageCleanerService.deleteMessageChannelMessageAssociationsByChannelId(
+      {
+        workspaceId: data.workspaceId,
+        messageChannelId: data.messageChannelId,
+      },
+    );
+
+    await this.messageCleanerService.cleanOrphanMessagesAndThreads(
+      data.workspaceId,
+    );
+  }
+}
@@ -0,0 +1,42 @@
+import { Injectable } from '@nestjs/common';
+
+import { type ObjectRecordDeleteEvent } from 'twenty-shared/database-events';
+
+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 MessageChannelWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity';
+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 {
+  constructor(
+    @InjectMessageQueue(MessageQueue.messagingQueue)
+    private readonly messageQueueService: MessageQueueService,
+  ) {}
+
+  @OnDatabaseBatchEvent('messageChannel', DatabaseEventAction.DESTROYED)
+  async handleDestroyedEvent(
+    payload: WorkspaceEventBatch<
+      ObjectRecordDeleteEvent<MessageChannelWorkspaceEntity>
+    >,
+  ) {
+    await Promise.all(
+      payload.events.map((eventPayload) =>
+        this.messageQueueService.add<MessagingMessageChannelDeletionCleanupJobData>(
+          MessagingMessageChannelDeletionCleanupJob.name,
+          {
+            workspaceId: payload.workspaceId,
+            messageChannelId: eventPayload.recordId,
+          },
+        ),
+      ),
+    );
+  }
+}
@@ -1,6 +1,7 @@
 import { Module } from '@nestjs/common';
 import { TypeOrmModule } from '@nestjs/typeorm';

+import { FeatureFlagModule } from 'src/engine/core-modules/feature-flag/feature-flag.module';
 import { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity';
 import { DataSourceModule } from 'src/engine/metadata-modules/data-source/data-source.module';
 import { MessageChannelDataAccessModule } from 'src/engine/metadata-modules/message-channel/data-access/message-channel-data-access.module';
@@ -8,19 +9,24 @@ import { MessagingCommonModule } from 'src/modules/messaging/common/messaging-co
 import { MessagingMessageCleanerRemoveOrphansCommand } from 'src/modules/messaging/message-cleaner/commands/messaging-message-clearner-remove-orphans.command';
 import { MessagingResetChannelCommand } from 'src/modules/messaging/message-cleaner/commands/messaging-reset-channel.command';
 import { MessagingConnectedAccountDeletionCleanupJob } from 'src/modules/messaging/message-cleaner/jobs/messaging-connected-account-deletion-cleanup.job';
+import { MessagingMessageChannelDeletionCleanupJob } from 'src/modules/messaging/message-cleaner/jobs/messaging-message-channel-deletion-cleanup.job';
 import { MessagingMessageCleanerConnectedAccountListener } from 'src/modules/messaging/message-cleaner/listeners/messaging-message-cleaner-connected-account.listener';
+import { MessagingMessageCleanerMessageChannelListener } from 'src/modules/messaging/message-cleaner/listeners/messaging-message-cleaner-message-channel.listener';
 import { MessagingMessageCleanerService } from 'src/modules/messaging/message-cleaner/services/messaging-message-cleaner.service';

@Module({
  imports: [
    TypeOrmModule.forFeature([WorkspaceEntity]),
    DataSourceModule,
+    FeatureFlagModule,
    MessagingCommonModule,
    MessageChannelDataAccessModule,
  ],
  providers: [
    MessagingConnectedAccountDeletionCleanupJob,
+    MessagingMessageChannelDeletionCleanupJob,
    MessagingMessageCleanerConnectedAccountListener,
+    MessagingMessageCleanerMessageChannelListener,
    MessagingMessageCleanerRemoveOrphansCommand,
    MessagingResetChannelCommand,
    MessagingMessageCleanerService,
@@ -120,6 +120,68 @@ export class MessagingMessageCleanerService {
    }, authContext);
  }

+  async deleteMessageChannelMessageAssociationsByChannelId({
+    workspaceId,
+    messageChannelId,
+  }: {
+    workspaceId: string;
+    messageChannelId: string;
+  }) {
+    const authContext = buildSystemAuthContext(workspaceId);
+
+    await this.globalWorkspaceOrmManager.executeInWorkspaceContext(async () => {
+      const messageChannelMessageAssociationRepository =
+        await this.globalWorkspaceOrmManager.getRepository<MessageChannelMessageAssociationWorkspaceEntity>(
+          workspaceId,
+          'messageChannelMessageAssociation',
+        );
+
+      const workspaceDataSource =
+        await this.globalWorkspaceOrmManager.getGlobalWorkspaceDataSource();
+
+      await workspaceDataSource.transaction(async (manager) => {
+        const transactionManager = manager as WorkspaceEntityManager;
+
+        await deleteUsingPagination(
+          workspaceId,
+          500,
+          async (
+            limit: number,
+            offset: number,
+            _workspaceId: string,
+            transactionManager?: WorkspaceEntityManager,
+          ) => {
+            const associations =
+              await messageChannelMessageAssociationRepository.find(
+                {
+                  where: { messageChannelId },
+                  take: limit,
+                  skip: offset,
+                },
+                transactionManager,
+              );
+
+            return associations.map(({ id }) => id);
+          },
+          async (
+            ids: string[],
+            workspaceId: string,
+            transactionManager?: WorkspaceEntityManager,
+          ) => {
+            this.logger.log(
+              `WorkspaceId: ${workspaceId} Deleting ${ids.length} message channel message associations for channel ${messageChannelId}`,
+            );
+            await messageChannelMessageAssociationRepository.delete(
+              ids,
+              transactionManager,
+            );
+          },
+          transactionManager,
+        );
+      });
+    }, authContext);
+  }
+
  public async cleanOrphanMessagesAndThreads(workspaceId: string) {
    const authContext = buildSystemAuthContext(workspaceId);