[messaging] clean orphan threads and messages after connected account deletion (#4195)

* [messaging] add connected account associated data delete

* add threadCleanerService

* fix

* fix import

* add thread cleaner import

* remove log

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

@@ -77,11 +77,13 @@ export class CreateCompanyService {
     const { name, city } = await this.getCompanyInfoFromDomainName(domainName);
 
     this.companyService.createCompany(
-      companyId,
-      name,
-      domainName,
-      city,
       workspaceId,
+      {
+        id: companyId,
+        domainName,
+        name,
+        city,
+      },
       transactionManager,
     );
 

packages/twenty-server/src/workspace/messaging/services/thread-cleaner/thread-cleaner.module.ts

@@ -0,0 +1,19 @@
+import { Module } from '@nestjs/common';
+
+import { TypeORMModule } from 'src/database/typeorm/typeorm.module';
+import { DataSourceModule } from 'src/metadata/data-source/data-source.module';
+import { MessageThreadModule } from 'src/workspace/messaging/repositories/message-thread/message-thread.module';
+import { MessageModule } from 'src/workspace/messaging/repositories/message/message.module';
+import { ThreadCleanerService } from 'src/workspace/messaging/services/thread-cleaner/thread-cleaner.service';
+
+@Module({
+  imports: [
+    DataSourceModule,
+    TypeORMModule,
+    MessageThreadModule,
+    MessageModule,
+  ],
+  providers: [ThreadCleanerService],
+  exports: [ThreadCleanerService],
+})
+export class ThreadCleanerModule {}

packages/twenty-server/src/workspace/messaging/services/thread-cleaner/thread-cleaner.service.ts

@@ -0,0 +1,62 @@
+import { Injectable } from '@nestjs/common';
+
+import { TypeORMService } from 'src/database/typeorm/typeorm.service';
+import { DataSourceService } from 'src/metadata/data-source/data-source.service';
+import { MessageThreadService } from 'src/workspace/messaging/repositories/message-thread/message-thread.service';
+import { MessageService } from 'src/workspace/messaging/repositories/message/message.service';
+
+@Injectable()
+export class ThreadCleanerService {
+  constructor(
+    private readonly dataSourceService: DataSourceService,
+    private readonly typeORMService: TypeORMService,
+    private readonly messageService: MessageService,
+    private readonly messageThreadService: MessageThreadService,
+  ) {}
+
+  public async cleanWorkspaceThreads(workspaceId: string) {
+    const dataSourceMetadata =
+      await this.dataSourceService.getLastDataSourceMetadataFromWorkspaceIdOrFail(
+        workspaceId,
+      );
+
+    const workspaceDataSource =
+      await this.typeORMService.connectToDataSource(dataSourceMetadata);
+
+    await workspaceDataSource?.transaction(async (transactionManager) => {
+      const messagesToDelete =
+        await this.messageService.getNonAssociatedMessages(
+          workspaceId,
+          transactionManager,
+        );
+
+      const messageIdsToDelete = messagesToDelete.map(({ id }) => id);
+
+      if (messageIdsToDelete.length > 0) {
+        await this.messageService.deleteByIds(
+          messageIdsToDelete,
+          workspaceId,
+          transactionManager,
+        );
+      }
+
+      const messageThreadsToDelete =
+        await this.messageThreadService.getOrphanThreads(
+          workspaceId,
+          transactionManager,
+        );
+
+      const messageThreadToDeleteIds = messageThreadsToDelete.map(
+        ({ id }) => id,
+      );
+
+      if (messageThreadToDeleteIds.length > 0) {
+        await this.messageThreadService.deleteByIds(
+          messageThreadToDeleteIds,
+          workspaceId,
+          transactionManager,
+        );
+      }
+    });
+  }
+}