From 2fccd29ec6d9ba4a8aff2a4de99b3343329f20bd Mon Sep 17 00:00:00 2001 From: neo773 <62795688+neo773@users.noreply.github.com> Date: Mon, 30 Mar 2026 19:51:00 +0530 Subject: [PATCH] messaging cleanup (#19124) --- .../connected-account-metadata.module.ts | 10 +- .../connected-account-metadata.service.ts | 101 +++++++++++++++++- .../calendar-event-cleaner.module.ts | 7 +- .../calendar-channel-deletion-cleanup.job.ts | 54 ++++++++++ ...event-cleaner-calendar-channel.listener.ts | 42 ++++++++ .../calendar-event-cleaner.service.ts | 67 +++++++++++- ...ng-message-channel-deletion-cleanup.job.ts | 58 ++++++++++ ...essage-cleaner-message-channel.listener.ts | 42 ++++++++ .../messaging-message-cleaner.module.ts | 6 ++ .../messaging-message-cleaner.service.ts | 62 +++++++++++ 10 files changed, 444 insertions(+), 5 deletions(-) create mode 100644 packages/twenty-server/src/modules/calendar/calendar-event-cleaner/jobs/calendar-channel-deletion-cleanup.job.ts create mode 100644 packages/twenty-server/src/modules/calendar/calendar-event-cleaner/listeners/calendar-event-cleaner-calendar-channel.listener.ts create mode 100644 packages/twenty-server/src/modules/messaging/message-cleaner/jobs/messaging-message-channel-deletion-cleanup.job.ts create mode 100644 packages/twenty-server/src/modules/messaging/message-cleaner/listeners/messaging-message-cleaner-message-channel.listener.ts diff --git a/packages/twenty-server/src/engine/metadata-modules/connected-account/connected-account-metadata.module.ts b/packages/twenty-server/src/engine/metadata-modules/connected-account/connected-account-metadata.module.ts index 626d80173d..c2c9c6389b 100644 --- a/packages/twenty-server/src/engine/metadata-modules/connected-account/connected-account-metadata.module.ts +++ b/packages/twenty-server/src/engine/metadata-modules/connected-account/connected-account-metadata.module.ts @@ -2,17 +2,25 @@ import { Module } from '@nestjs/common'; import { TypeOrmModule } from '@nestjs/typeorm'; import { FeatureFlagModule } from 'src/engine/core-modules/feature-flag/feature-flag.module'; +import { CalendarChannelDataAccessModule } from 'src/engine/metadata-modules/calendar-channel/data-access/calendar-channel-data-access.module'; +import { ConnectedAccountDataAccessModule } from 'src/engine/metadata-modules/connected-account/data-access/connected-account-data-access.module'; import { ConnectedAccountMetadataService } from 'src/engine/metadata-modules/connected-account/connected-account-metadata.service'; import { ConnectedAccountEntity } from 'src/engine/metadata-modules/connected-account/entities/connected-account.entity'; import { ConnectedAccountGraphqlApiExceptionInterceptor } from 'src/engine/metadata-modules/connected-account/interceptors/connected-account-graphql-api-exception.interceptor'; import { ConnectedAccountResolver } from 'src/engine/metadata-modules/connected-account/resolvers/connected-account.resolver'; +import { WorkspaceManyOrAllFlatEntityMapsCacheModule } from 'src/engine/metadata-modules/flat-entity/services/workspace-many-or-all-flat-entity-maps-cache.module'; +import { MessageChannelDataAccessModule } from 'src/engine/metadata-modules/message-channel/data-access/message-channel-data-access.module'; import { PermissionsModule } from 'src/engine/metadata-modules/permissions/permissions.module'; @Module({ imports: [ TypeOrmModule.forFeature([ConnectedAccountEntity]), - PermissionsModule, + CalendarChannelDataAccessModule, + ConnectedAccountDataAccessModule, FeatureFlagModule, + WorkspaceManyOrAllFlatEntityMapsCacheModule, + MessageChannelDataAccessModule, + PermissionsModule, ], providers: [ ConnectedAccountMetadataService, diff --git a/packages/twenty-server/src/engine/metadata-modules/connected-account/connected-account-metadata.service.ts b/packages/twenty-server/src/engine/metadata-modules/connected-account/connected-account-metadata.service.ts index e9f2e4845d..db048aff17 100644 --- a/packages/twenty-server/src/engine/metadata-modules/connected-account/connected-account-metadata.service.ts +++ b/packages/twenty-server/src/engine/metadata-modules/connected-account/connected-account-metadata.service.ts @@ -1,20 +1,38 @@ -import { Injectable } from '@nestjs/common'; +import { Injectable, Logger } from '@nestjs/common'; import { InjectRepository } from '@nestjs/typeorm'; +import { STANDARD_OBJECTS } from 'twenty-shared/metadata'; import { In, Repository } from 'typeorm'; +import { DatabaseEventAction } from 'src/engine/api/graphql/graphql-query-runner/enums/database-event-action'; import { ConnectedAccountException, ConnectedAccountExceptionCode, } from 'src/engine/metadata-modules/connected-account/connected-account.exception'; import { ConnectedAccountDTO } from 'src/engine/metadata-modules/connected-account/dtos/connected-account.dto'; import { ConnectedAccountEntity } from 'src/engine/metadata-modules/connected-account/entities/connected-account.entity'; +import { CalendarChannelDataAccessService } from 'src/engine/metadata-modules/calendar-channel/data-access/services/calendar-channel-data-access.service'; +import { MessageChannelDataAccessService } from 'src/engine/metadata-modules/message-channel/data-access/services/message-channel-data-access.service'; +import { WorkspaceManyOrAllFlatEntityMapsCacheService } from 'src/engine/metadata-modules/flat-entity/services/workspace-many-or-all-flat-entity-maps-cache.service'; +import { findFlatEntityByUniversalIdentifierOrThrow } from 'src/engine/metadata-modules/flat-entity/utils/find-flat-entity-by-universal-identifier-or-throw.util'; +import { ConnectedAccountDataAccessService } from 'src/engine/metadata-modules/connected-account/data-access/services/connected-account-data-access.service'; +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 { WorkspaceEventEmitter } from 'src/engine/workspace-event-emitter/workspace-event-emitter'; @Injectable() export class ConnectedAccountMetadataService { + private readonly logger = new Logger(ConnectedAccountMetadataService.name); + constructor( @InjectRepository(ConnectedAccountEntity) private readonly repository: Repository, + private readonly connectedAccountDataAccessService: ConnectedAccountDataAccessService, + private readonly calendarChannelDataAccessService: CalendarChannelDataAccessService, + private readonly messageChannelDataAccessService: MessageChannelDataAccessService, + private readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager, + private readonly workspaceEventEmitter: WorkspaceEventEmitter, + private readonly workspaceManyOrAllFlatEntityMapsCacheService: WorkspaceManyOrAllFlatEntityMapsCacheService, ) {} async findAll(workspaceId: string): Promise { @@ -141,7 +159,86 @@ export class ConnectedAccountMetadataService { where: { id, workspaceId }, }); - await this.repository.delete({ id, workspaceId }); + const authContext = buildSystemAuthContext(workspaceId); + + const [messageChannels, calendarChannels] = await Promise.all([ + this.globalWorkspaceOrmManager.executeInWorkspaceContext( + async () => + this.messageChannelDataAccessService.find(workspaceId, { + connectedAccountId: id, + }), + authContext, + ), + this.globalWorkspaceOrmManager.executeInWorkspaceContext( + async () => + this.calendarChannelDataAccessService.find(workspaceId, { + connectedAccountId: id, + }), + authContext, + ), + ]); + + this.logger.log( + `WorkspaceId: ${workspaceId} Deleting connected account ${id} with ${messageChannels.length} message channel(s) and ${calendarChannels.length} calendar channel(s)`, + ); + + const { flatObjectMetadataMaps } = + await this.workspaceManyOrAllFlatEntityMapsCacheService.getOrRecomputeManyOrAllFlatEntityMaps( + { + workspaceId, + flatMapsKeys: ['flatObjectMetadataMaps'], + }, + ); + + if (messageChannels.length > 0) { + const flatMessageChannelMetadata = + findFlatEntityByUniversalIdentifierOrThrow({ + flatEntityMaps: flatObjectMetadataMaps, + universalIdentifier: + STANDARD_OBJECTS.messageChannel.universalIdentifier, + }); + + this.workspaceEventEmitter.emitDatabaseBatchEvent({ + objectMetadataNameSingular: 'messageChannel', + action: DatabaseEventAction.DESTROYED, + objectMetadata: flatMessageChannelMetadata, + events: messageChannels.map((messageChannel) => ({ + recordId: messageChannel.id, + properties: { + before: messageChannel, + }, + })), + workspaceId, + }); + } + + if (calendarChannels.length > 0) { + const flatCalendarChannelMetadata = + findFlatEntityByUniversalIdentifierOrThrow({ + flatEntityMaps: flatObjectMetadataMaps, + universalIdentifier: + STANDARD_OBJECTS.calendarChannel.universalIdentifier, + }); + + this.workspaceEventEmitter.emitDatabaseBatchEvent({ + objectMetadataNameSingular: 'calendarChannel', + action: DatabaseEventAction.DESTROYED, + objectMetadata: flatCalendarChannelMetadata, + events: calendarChannels.map((calendarChannel) => ({ + recordId: calendarChannel.id, + properties: { + before: calendarChannel, + }, + })), + workspaceId, + }); + } + + await this.globalWorkspaceOrmManager.executeInWorkspaceContext(async () => { + await this.connectedAccountDataAccessService.delete(workspaceId, { + id, + }); + }, authContext); return connectedAccount; } diff --git a/packages/twenty-server/src/modules/calendar/calendar-event-cleaner/calendar-event-cleaner.module.ts b/packages/twenty-server/src/modules/calendar/calendar-event-cleaner/calendar-event-cleaner.module.ts index bb9af960eb..bcdd886a3f 100644 --- a/packages/twenty-server/src/modules/calendar/calendar-event-cleaner/calendar-event-cleaner.module.ts +++ b/packages/twenty-server/src/modules/calendar/calendar-event-cleaner/calendar-event-cleaner.module.ts @@ -1,14 +1,19 @@ import { Module } from '@nestjs/common'; +import { FeatureFlagModule } from 'src/engine/core-modules/feature-flag/feature-flag.module'; +import { CalendarChannelDeletionCleanupJob } from 'src/modules/calendar/calendar-event-cleaner/jobs/calendar-channel-deletion-cleanup.job'; import { DeleteConnectedAccountAssociatedCalendarDataJob } from 'src/modules/calendar/calendar-event-cleaner/jobs/delete-connected-account-associated-calendar-data.job'; +import { CalendarEventCleanerCalendarChannelListener } from 'src/modules/calendar/calendar-event-cleaner/listeners/calendar-event-cleaner-calendar-channel.listener'; import { CalendarEventCleanerConnectedAccountListener } from 'src/modules/calendar/calendar-event-cleaner/listeners/calendar-event-cleaner-connected-account.listener'; import { CalendarEventCleanerService } from 'src/modules/calendar/calendar-event-cleaner/services/calendar-event-cleaner.service'; @Module({ - imports: [], + imports: [FeatureFlagModule], providers: [ CalendarEventCleanerService, + CalendarChannelDeletionCleanupJob, DeleteConnectedAccountAssociatedCalendarDataJob, + CalendarEventCleanerCalendarChannelListener, CalendarEventCleanerConnectedAccountListener, ], exports: [CalendarEventCleanerService], diff --git a/packages/twenty-server/src/modules/calendar/calendar-event-cleaner/jobs/calendar-channel-deletion-cleanup.job.ts b/packages/twenty-server/src/modules/calendar/calendar-event-cleaner/jobs/calendar-channel-deletion-cleanup.job.ts new file mode 100644 index 0000000000..d4253802d4 --- /dev/null +++ b/packages/twenty-server/src/modules/calendar/calendar-event-cleaner/jobs/calendar-channel-deletion-cleanup.job.ts @@ -0,0 +1,54 @@ +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 { CalendarEventCleanerService } from 'src/modules/calendar/calendar-event-cleaner/services/calendar-event-cleaner.service'; + +export type CalendarChannelDeletionCleanupJobData = { + workspaceId: string; + calendarChannelId: string; +}; + +@Processor({ + queueName: MessageQueue.calendarQueue, + scope: Scope.REQUEST, +}) +export class CalendarChannelDeletionCleanupJob { + private readonly logger = new Logger(CalendarChannelDeletionCleanupJob.name); + + constructor( + private readonly calendarEventCleanerService: CalendarEventCleanerService, + private readonly featureFlagService: FeatureFlagService, + ) {} + + @Process(CalendarChannelDeletionCleanupJob.name) + async handle(data: CalendarChannelDeletionCleanupJobData): Promise { + const isMigrated = await this.featureFlagService.isFeatureEnabled( + FeatureFlagKey.IS_CONNECTED_ACCOUNT_MIGRATED, + data.workspaceId, + ); + + if (!isMigrated) { + return; + } + + this.logger.debug( + `WorkspaceId: ${data.workspaceId} Cleaning up calendar channel event associations for channel ${data.calendarChannelId}`, + ); + + await this.calendarEventCleanerService.deleteCalendarChannelEventAssociationsByChannelId( + { + workspaceId: data.workspaceId, + calendarChannelId: data.calendarChannelId, + }, + ); + + await this.calendarEventCleanerService.cleanWorkspaceCalendarEvents( + data.workspaceId, + ); + } +} diff --git a/packages/twenty-server/src/modules/calendar/calendar-event-cleaner/listeners/calendar-event-cleaner-calendar-channel.listener.ts b/packages/twenty-server/src/modules/calendar/calendar-event-cleaner/listeners/calendar-event-cleaner-calendar-channel.listener.ts new file mode 100644 index 0000000000..5050c1b5ea --- /dev/null +++ b/packages/twenty-server/src/modules/calendar/calendar-event-cleaner/listeners/calendar-event-cleaner-calendar-channel.listener.ts @@ -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 CalendarChannelWorkspaceEntity } from 'src/modules/calendar/common/standard-objects/calendar-channel.workspace-entity'; +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 { + constructor( + @InjectMessageQueue(MessageQueue.calendarQueue) + private readonly calendarQueueService: MessageQueueService, + ) {} + + @OnDatabaseBatchEvent('calendarChannel', DatabaseEventAction.DESTROYED) + async handleDestroyedEvent( + payload: WorkspaceEventBatch< + ObjectRecordDeleteEvent + >, + ) { + await Promise.all( + payload.events.map((eventPayload) => + this.calendarQueueService.add( + CalendarChannelDeletionCleanupJob.name, + { + workspaceId: payload.workspaceId, + calendarChannelId: eventPayload.recordId, + }, + ), + ), + ); + } +} diff --git a/packages/twenty-server/src/modules/calendar/calendar-event-cleaner/services/calendar-event-cleaner.service.ts b/packages/twenty-server/src/modules/calendar/calendar-event-cleaner/services/calendar-event-cleaner.service.ts index ac689389e0..169fa10527 100644 --- a/packages/twenty-server/src/modules/calendar/calendar-event-cleaner/services/calendar-event-cleaner.service.ts +++ b/packages/twenty-server/src/modules/calendar/calendar-event-cleaner/services/calendar-event-cleaner.service.ts @@ -1,17 +1,82 @@ -import { Injectable } from '@nestjs/common'; +import { Injectable, Logger } from '@nestjs/common'; import { Any, IsNull } from 'typeorm'; +import { type WorkspaceEntityManager } from 'src/engine/twenty-orm/entity-manager/workspace-entity-manager'; 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 { deleteUsingPagination } from 'src/modules/messaging/message-cleaner/utils/delete-using-pagination.util'; @Injectable() export class CalendarEventCleanerService { + private readonly logger = new Logger(CalendarEventCleanerService.name); + constructor( private readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager, ) {} + async deleteCalendarChannelEventAssociationsByChannelId({ + workspaceId, + calendarChannelId, + }: { + workspaceId: string; + calendarChannelId: string; + }) { + const authContext = buildSystemAuthContext(workspaceId); + + await this.globalWorkspaceOrmManager.executeInWorkspaceContext(async () => { + const calendarChannelEventAssociationRepository = + await this.globalWorkspaceOrmManager.getRepository( + workspaceId, + 'calendarChannelEventAssociation', + ); + + 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 calendarChannelEventAssociationRepository.find( + { + where: { calendarChannelId }, + 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} calendar channel event associations for channel ${calendarChannelId}`, + ); + await calendarChannelEventAssociationRepository.delete( + ids, + transactionManager, + ); + }, + transactionManager, + ); + }); + }, authContext); + } + public async cleanWorkspaceCalendarEvents(workspaceId: string) { const authContext = buildSystemAuthContext(workspaceId); diff --git a/packages/twenty-server/src/modules/messaging/message-cleaner/jobs/messaging-message-channel-deletion-cleanup.job.ts b/packages/twenty-server/src/modules/messaging/message-cleaner/jobs/messaging-message-channel-deletion-cleanup.job.ts new file mode 100644 index 0000000000..dde8c1bd14 --- /dev/null +++ b/packages/twenty-server/src/modules/messaging/message-cleaner/jobs/messaging-message-channel-deletion-cleanup.job.ts @@ -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 { + 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, + ); + } +} diff --git a/packages/twenty-server/src/modules/messaging/message-cleaner/listeners/messaging-message-cleaner-message-channel.listener.ts b/packages/twenty-server/src/modules/messaging/message-cleaner/listeners/messaging-message-cleaner-message-channel.listener.ts new file mode 100644 index 0000000000..77261e71ae --- /dev/null +++ b/packages/twenty-server/src/modules/messaging/message-cleaner/listeners/messaging-message-cleaner-message-channel.listener.ts @@ -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 + >, + ) { + await Promise.all( + payload.events.map((eventPayload) => + this.messageQueueService.add( + MessagingMessageChannelDeletionCleanupJob.name, + { + workspaceId: payload.workspaceId, + messageChannelId: eventPayload.recordId, + }, + ), + ), + ); + } +} diff --git a/packages/twenty-server/src/modules/messaging/message-cleaner/messaging-message-cleaner.module.ts b/packages/twenty-server/src/modules/messaging/message-cleaner/messaging-message-cleaner.module.ts index b0312a014e..7920b1c6e9 100644 --- a/packages/twenty-server/src/modules/messaging/message-cleaner/messaging-message-cleaner.module.ts +++ b/packages/twenty-server/src/modules/messaging/message-cleaner/messaging-message-cleaner.module.ts @@ -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, diff --git a/packages/twenty-server/src/modules/messaging/message-cleaner/services/messaging-message-cleaner.service.ts b/packages/twenty-server/src/modules/messaging/message-cleaner/services/messaging-message-cleaner.service.ts index 5601f1984f..7b2f6ce43c 100644 --- a/packages/twenty-server/src/modules/messaging/message-cleaner/services/messaging-message-cleaner.service.ts +++ b/packages/twenty-server/src/modules/messaging/message-cleaner/services/messaging-message-cleaner.service.ts @@ -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( + 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);