[messaging] clean orphan threads and messages after connected account deletion

packages/twenty-server/src/integrations/message-queue/jobs.module.ts

@@ -24,8 +24,10 @@ import { CreateCompaniesAndContactsAfterSyncJob } from 'src/workspace/messaging/
 import { CreateCompaniesAndContactsModule } from 'src/workspace/messaging/services/create-companies-and-contacts/create-companies-and-contacts.module';
 import { MessageChannelModule } from 'src/workspace/messaging/repositories/message-channel/message-channel.module';
 import { MessageParticipantModule } from 'src/workspace/messaging/repositories/message-participant/message-participant.module';
-import { DataSeedDemoWorkspaceJob } from 'src/database/commands/data-seed-demo-workspace/jobs/data-seed-demo-workspace.job';
 import { DataSeedDemoWorkspaceModule } from 'src/database/commands/data-seed-demo-workspace/data-seed-demo-workspace.module';
+import { DataSeedDemoWorkspaceJob } from 'src/database/commands/data-seed-demo-workspace/jobs/data-seed-demo-workspace.job';
+import { DeleteConnectedAccountAssociatedDataJob } from 'src/workspace/messaging/jobs/delete-connected-acount-associated-data.job';
+import { ThreadCleanerModule } from 'src/workspace/messaging/services/thread-cleaner/thread-cleaner.module';
 
 @Module({
   imports: [
@@ -44,6 +46,7 @@ import { DataSeedDemoWorkspaceModule } from 'src/database/commands/data-seed-dem
     CreateCompaniesAndContactsModule,
     MessageChannelModule,
     DataSeedDemoWorkspaceModule,
+    ThreadCleanerModule,
   ],
   providers: [
     {
@@ -83,6 +86,10 @@ import { DataSeedDemoWorkspaceModule } from 'src/database/commands/data-seed-dem
       provide: DataSeedDemoWorkspaceJob.name,
       useClass: DataSeedDemoWorkspaceJob,
     },
+    {
+      provide: DeleteConnectedAccountAssociatedDataJob.name,
+      useClass: DeleteConnectedAccountAssociatedDataJob,
+    },
   ],
 })
 export class JobsModule {

packages/twenty-server/src/metadata/relation-metadata/relation-metadata.entity.ts

@@ -24,6 +24,7 @@ export enum RelationOnDeleteAction {
   CASCADE = 'CASCADE',
   RESTRICT = 'RESTRICT',
   SET_NULL = 'SET_NULL',
+  NO_ACTION = 'NO_ACTION',
 }
 
 @Entity('relationMetadata')

packages/twenty-server/src/workspace/messaging/jobs/delete-connected-acount-associated-data.job.ts

@@ -0,0 +1,35 @@
+import { Injectable, Logger } from '@nestjs/common';
+
+import { MessageQueueJob } from 'src/integrations/message-queue/interfaces/message-queue-job.interface';
+
+import { ThreadCleanerService } from 'src/workspace/messaging/services/thread-cleaner/thread-cleaner.service';
+
+export type DeleteConnectedAccountAssociatedDataJobData = {
+  workspaceId: string;
+  connectedAccountId: string;
+};
+
+@Injectable()
+export class DeleteConnectedAccountAssociatedDataJob
+  implements MessageQueueJob<DeleteConnectedAccountAssociatedDataJobData>
+{
+  private readonly logger = new Logger(
+    DeleteConnectedAccountAssociatedDataJob.name,
+  );
+
+  constructor(private readonly threadCleanerService: ThreadCleanerService) {}
+
+  async handle(
+    data: DeleteConnectedAccountAssociatedDataJobData,
+  ): Promise<void> {
+    this.logger.log(
+      `Deleting connected account ${data.connectedAccountId} associated data in workspace ${data.workspaceId}`,
+    );
+
+    await this.threadCleanerService.cleanWorkspaceThreads(data.workspaceId);
+
+    this.logger.log(
+      `Deleted connected account ${data.connectedAccountId} associated data in workspace ${data.workspaceId}`,
+    );
+  }
+}

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

@@ -0,0 +1,32 @@
+import { Injectable, Inject } from '@nestjs/common';
+import { OnEvent } from '@nestjs/event-emitter';
+
+import { ObjectRecordDeleteEvent } from 'src/integrations/event-emitter/types/object-record-delete.event';
+import { MessageQueue } from 'src/integrations/message-queue/message-queue.constants';
+import { MessageQueueService } from 'src/integrations/message-queue/services/message-queue.service';
+import {
+  DeleteConnectedAccountAssociatedDataJobData,
+  DeleteConnectedAccountAssociatedDataJob,
+} from 'src/workspace/messaging/jobs/delete-connected-acount-associated-data.job';
+import { ConnectedAccountObjectMetadata } from 'src/workspace/workspace-sync-metadata/standard-objects/connected-account.object-metadata';
+
+@Injectable()
+export class MessagingConnectedAccountListener {
+  constructor(
+    @Inject(MessageQueue.messagingQueue)
+    private readonly messageQueueService: MessageQueueService,
+  ) {}
+
+  @OnEvent('connectedAccount.deleted')
+  handleDeletedEvent(
+    payload: ObjectRecordDeleteEvent<ConnectedAccountObjectMetadata>,
+  ) {
+    this.messageQueueService.add<DeleteConnectedAccountAssociatedDataJobData>(
+      DeleteConnectedAccountAssociatedDataJob.name,
+      {
+        workspaceId: payload.workspaceId,
+        connectedAccountId: payload.deletedRecord.id,
+      },
+    );
+  }
+}

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

@@ -1,13 +1,14 @@
 import { Inject, Injectable } from '@nestjs/common';
 import { OnEvent } from '@nestjs/event-emitter';
 
-import { ObjectRecordDeleteEvent } from 'src/integrations/event-emitter/types/object-record-delete.event';
+import { ObjectRecordUpdateEvent } from 'src/integrations/event-emitter/types/object-record-update.event';
+import { objectRecordChangedProperties } from 'src/integrations/event-emitter/utils/object-record-changed-properties.util';
 import { MessageQueue } from 'src/integrations/message-queue/message-queue.constants';
 import { MessageQueueService } from 'src/integrations/message-queue/services/message-queue.service';
 import {
-  DeleteMessageChannelMessageAssociationJob,
-  DeleteMessageChannelMessageAssociationJobData,
-} from 'src/workspace/messaging/jobs/delete-message-channel-message-association.job';
+  CreateCompaniesAndContactsAfterSyncJobData,
+  CreateCompaniesAndContactsAfterSyncJob,
+} from 'src/workspace/messaging/jobs/create-companies-and-contacts-after-sync.job';
 import { MessageChannelObjectMetadata } from 'src/workspace/workspace-sync-metadata/standard-objects/message-channel.object-metadata';
 
 @Injectable()
@@ -17,16 +18,24 @@ export class MessagingMessageChannelListener {
     private readonly messageQueueService: MessageQueueService,
   ) {}
 
-  @OnEvent('messageChannel.deleted')
-  handleDeletedEvent(
-    payload: ObjectRecordDeleteEvent<MessageChannelObjectMetadata>,
+  @OnEvent('messageChannel.updated')
+  handleUpdatedEvent(
+    payload: ObjectRecordUpdateEvent<MessageChannelObjectMetadata>,
   ) {
-    this.messageQueueService.add<DeleteMessageChannelMessageAssociationJobData>(
-      DeleteMessageChannelMessageAssociationJob.name,
-      {
-        workspaceId: payload.workspaceId,
-        messageChannelId: payload.deletedRecord.id,
-      },
-    );
+    if (
+      objectRecordChangedProperties(
+        payload.previousRecord,
+        payload.updatedRecord,
+      ).includes('isContactAutoCreationEnabled') &&
+      payload.updatedRecord.isContactAutoCreationEnabled
+    ) {
+      this.messageQueueService.add<CreateCompaniesAndContactsAfterSyncJobData>(
+        CreateCompaniesAndContactsAfterSyncJob.name,
+        {
+          workspaceId: payload.workspaceId,
+          messageChannelId: payload.updatedRecord.id,
+        },
+      );
+    }
   }
 }

packages/twenty-server/src/workspace/messaging/messaging.module.ts

@@ -18,14 +18,14 @@ import { GmailRefreshAccessTokenService } from 'src/workspace/messaging/services
 import { WorkspaceDataSourceModule } from 'src/workspace/workspace-datasource/workspace-datasource.module';
 import { MessageParticipantModule } from 'src/workspace/messaging/repositories/message-participant/message-participant.module';
 import { MessagingWorkspaceMemberListener } from 'src/workspace/messaging/listeners/messaging-workspace-member.listener';
-import { IsContactAutoCreationEnabledListener } from 'src/workspace/messaging/listeners/is-contact-auto-creation-enabled-listener';
 import { MessagingMessageChannelListener } from 'src/workspace/messaging/listeners/messaging-message-channel.listener';
 import { MessageService } from 'src/workspace/messaging/repositories/message/message.service';
 import { WorkspaceMemberModule } from 'src/workspace/messaging/repositories/workspace-member/workspace-member.module';
 import { FeatureFlagEntity } from 'src/core/feature-flag/feature-flag.entity';
 import { CreateCompaniesAndContactsModule } from 'src/workspace/messaging/services/create-companies-and-contacts/create-companies-and-contacts.module';
 import { CompanyModule } from 'src/workspace/messaging/repositories/company/company.module';
 import { PersonModule } from 'src/workspace/messaging/repositories/person/person.module';
+import { MessagingConnectedAccountListener } from 'src/workspace/messaging/listeners/messaging-connected-account.listener';
 @Module({
   imports: [
     EnvironmentModule,
@@ -52,9 +52,9 @@ import { PersonModule } from 'src/workspace/messaging/repositories/person/person
     CreateCompanyService,
     MessagingPersonListener,
     MessagingWorkspaceMemberListener,
-    IsContactAutoCreationEnabledListener,
     MessagingMessageChannelListener,
     MessageService,
+    MessagingConnectedAccountListener,
   ],
   exports: [
     GmailPartialSyncService,

packages/twenty-server/src/workspace/messaging/repositories/company/company.service.ts

@@ -4,6 +4,13 @@ import { EntityManager } from 'typeorm';
 
 import { WorkspaceDataSourceService } from 'src/workspace/workspace-datasource/workspace-datasource.service';
 
+export type CompanyToCreate = {
+  id: string;
+  domainName: string;
+  name?: string;
+  city?: string;
+};
+
 // TODO: Move outside of the messaging module
 @Injectable()
 export class CompanyService {
@@ -31,20 +38,22 @@ export class CompanyService {
   }
 
   public async createCompany(
-    id: string,
-    name: string,
-    domainName: string,
-    city: string,
     workspaceId: string,
+    companyToCreate: CompanyToCreate,
     transactionManager?: EntityManager,
   ): Promise<void> {
     const dataSourceSchema =
       this.workspaceDataSourceService.getSchemaName(workspaceId);
 
     await this.workspaceDataSourceService.executeRawQuery(
-      `INSERT INTO ${dataSourceSchema}.company (id, name, "domainName", address)
+      `INSERT INTO ${dataSourceSchema}.company (id, "domainName", name, address)
       VALUES ($1, $2, $3, $4)`,
-      [id, name, domainName, city],
+      [
+        companyToCreate.id,
+        companyToCreate.domainName,
+        companyToCreate.name ?? '',
+        companyToCreate.city ?? '',
+      ],
       workspaceId,
       transactionManager,
     );

packages/twenty-server/src/workspace/messaging/repositories/message-channel-message-association/message-channel-message-association.service.ts

@@ -67,17 +67,50 @@ export class MessageChannelMessageAssociationService {
     );
   }
 
+  public async getByMessageChannelIds(
+    messageChannelIds: string[],
+    workspaceId: string,
+    transactionManager?: EntityManager,
+  ): Promise<ObjectRecord<MessageChannelMessageAssociationObjectMetadata>[]> {
+    const dataSourceSchema =
+      this.workspaceDataSourceService.getSchemaName(workspaceId);
+
+    return await this.workspaceDataSourceService.executeRawQuery(
+      `SELECT * FROM ${dataSourceSchema}."messageChannelMessageAssociation"
+    WHERE "messageChannelId" = ANY($1)`,
+      [messageChannelIds],
+      workspaceId,
+      transactionManager,
+    );
+  }
+
   public async deleteByMessageChannelId(
     messageChannelId: string,
     workspaceId: string,
     transactionManager?: EntityManager,
   ) {
+    this.deleteByMessageChannelIds(
+      [messageChannelId],
+      workspaceId,
+      transactionManager,
+    );
+  }
+
+  public async deleteByMessageChannelIds(
+    messageChannelIds: string[],
+    workspaceId: string,
+    transactionManager?: EntityManager,
+  ) {
+    if (messageChannelIds.length === 0) {
+      return;
+    }
+
     const dataSourceSchema =
       this.workspaceDataSourceService.getSchemaName(workspaceId);
 
     await this.workspaceDataSourceService.executeRawQuery(
-      `DELETE FROM ${dataSourceSchema}."messageChannelMessageAssociation" WHERE "messageChannelId" = $1`,
-      [messageChannelId],
+      `DELETE FROM ${dataSourceSchema}."messageChannelMessageAssociation" WHERE "messageChannelId" = ANY($1)`,
+      [messageChannelIds],
       workspaceId,
       transactionManager,
     );