6430 Part 1: remove all raw queries from the messaging and calendar modules (#6572)

Part 1 of #6430 - Remove all repositories which contained raw queries in `messaging` module - Replace them using `twentyORMManager`
2024-08-13 19:40:50 +02:00
parent 40bbee8d9f
commit d1c278d6b2
27 changed files with 379 additions and 1115 deletions
--- a/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-error-handling.service.ts
+++ b/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-error-handling.service.ts
@ -3,9 +3,9 @@ import { Injectable } from '@nestjs/common';
 import snakeCase from 'lodash.snakecase';

 import { InjectObjectMetadataRepository } from 'src/engine/object-metadata-repository/object-metadata-repository.decorator';
+import { TwentyORMManager } from 'src/engine/twenty-orm/twenty-orm.manager';
 import { ConnectedAccountRepository } from 'src/modules/connected-account/repositories/connected-account.repository';
 import { ConnectedAccountWorkspaceEntity } from 'src/modules/connected-account/standard-objects/connected-account.workspace-entity';
-import { MessageChannelRepository } from 'src/modules/messaging/common/repositories/message-channel.repository';
 import { MessagingChannelSyncStatusService } from 'src/modules/messaging/common/services/messaging-channel-sync-status.service';
 import { MessageChannelWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity';
 import { MESSAGING_THROTTLE_MAX_ATTEMPTS } from 'src/modules/messaging/message-import-manager/constants/messaging-throttle-max-attempts';
@ -28,8 +28,7 @@ export class MessagingErrorHandlingService {
    private readonly connectedAccountRepository: ConnectedAccountRepository,
    private readonly messagingChannelSyncStatusService: MessagingChannelSyncStatusService,
    private readonly messagingTelemetryService: MessagingTelemetryService,
-    @InjectObjectMetadataRepository(MessageChannelWorkspaceEntity)
-    private readonly messageChannelRepository: MessageChannelRepository,
+    private readonly twentyORMManager: TwentyORMManager,
  ) {}

  public async handleGmailError(
@ -263,21 +262,18 @@ export class MessagingErrorHandlingService {
      case 'full-message-list-fetch':
        await this.messagingChannelSyncStatusService.scheduleFullMessageListFetch(
          messageChannel.id,
-          workspaceId,
        );
        break;

      case 'partial-message-list-fetch':
        await this.messagingChannelSyncStatusService.schedulePartialMessageListFetch(
          messageChannel.id,
-          workspaceId,
        );
        break;

      case 'messages-import':
        await this.messagingChannelSyncStatusService.scheduleMessagesImport(
          messageChannel.id,
-          workspaceId,
        );
        break;

@ -290,9 +286,17 @@ export class MessagingErrorHandlingService {
    messageChannel: MessageChannelWorkspaceEntity,
    workspaceId: string,
  ): Promise<void> {
-    await this.messageChannelRepository.incrementThrottleFailureCount(
-      messageChannel.id,
-      workspaceId,
+    const messageChannelRepository =
+      await this.twentyORMManager.getRepository<MessageChannelWorkspaceEntity>(
+        'messageChannel',
+      );
+
+    await messageChannelRepository.increment(
+      {
+        id: messageChannel.id,
+      },
+      'throttleFailureCount',
+      1,
    );

    await this.messagingTelemetryService.track({
--- a/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-full-message-list-fetch.service.ts
+++ b/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-full-message-list-fetch.service.ts
@ -7,10 +7,8 @@ import { Any, EntityManager } from 'typeorm';
 import { CacheStorageService } from 'src/engine/integrations/cache-storage/cache-storage.service';
 import { InjectCacheStorage } from 'src/engine/integrations/cache-storage/decorators/cache-storage.decorator';
 import { CacheStorageNamespace } from 'src/engine/integrations/cache-storage/types/cache-storage-namespace.enum';
-import { InjectObjectMetadataRepository } from 'src/engine/object-metadata-repository/object-metadata-repository.decorator';
 import { TwentyORMManager } from 'src/engine/twenty-orm/twenty-orm.manager';
 import { ConnectedAccountWorkspaceEntity } from 'src/modules/connected-account/standard-objects/connected-account.workspace-entity';
-import { MessageChannelRepository } from 'src/modules/messaging/common/repositories/message-channel.repository';
 import { MessagingChannelSyncStatusService } from 'src/modules/messaging/common/services/messaging-channel-sync-status.service';
 import { MessageChannelMessageAssociationWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-channel-message-association.workspace-entity';
 import { MessageChannelWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity';
@ -31,8 +29,6 @@ export class MessagingFullMessageListFetchService {

  constructor(
    private readonly gmailClientProvider: MessagingGmailClientProvider,
-    @InjectObjectMetadataRepository(MessageChannelWorkspaceEntity)
-    private readonly messageChannelRepository: MessageChannelRepository,
    @InjectCacheStorage(CacheStorageNamespace.Messaging)
    private readonly cacheStorage: CacheStorageService,
    private readonly messagingChannelSyncStatusService: MessagingChannelSyncStatusService,
@ -47,7 +43,6 @@ export class MessagingFullMessageListFetchService {
  ) {
    await this.messagingChannelSyncStatusService.markAsMessagesListFetchOngoing(
      messageChannel.id,
-      workspaceId,
    );

    const gmailClient: gmail_v1.Gmail =
@ -70,19 +65,23 @@ export class MessagingFullMessageListFetchService {
      return;
    }

-    await this.messageChannelRepository.resetThrottleFailureCount(
-      messageChannel.id,
-      workspaceId,
-    );
+    const messageChannelRepository =
+      await this.twentyORMManager.getRepository<MessageChannelWorkspaceEntity>(
+        'messageChannel',
+      );

-    await this.messageChannelRepository.resetSyncStageStartedAt(
-      messageChannel.id,
-      workspaceId,
+    await messageChannelRepository.update(
+      {
+        id: messageChannel.id,
+      },
+      {
+        throttleFailureCount: 0,
+        syncStageStartedAt: null,
+      },
    );

    await this.messagingChannelSyncStatusService.scheduleMessagesImport(
      messageChannel.id,
-      workspaceId,
    );
  }

@ -216,11 +215,32 @@ export class MessagingFullMessageListFetchService {
      );
    }

-    await this.messageChannelRepository.updateLastSyncCursorIfHigher(
-      messageChannelId,
-      historyId,
-      workspaceId,
+    const messageChannelRepository =
+      await this.twentyORMManager.getRepository<MessageChannelWorkspaceEntity>(
+        'messageChannel',
+      );
+
+    const messageChannel = await messageChannelRepository.findOneOrFail(
+      {
+        where: {
+          id: messageChannelId,
+        },
+      },
      transactionManager,
    );
+
+    const currentSyncCursor = messageChannel.syncCursor;
+
+    if (!currentSyncCursor || historyId > currentSyncCursor) {
+      await messageChannelRepository.update(
+        {
+          id: messageChannel.id,
+        },
+        {
+          syncCursor: historyId,
+        },
+        transactionManager,
+      );
+    }
  }
 }
--- a/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-messages-import.service.ts
+++ b/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-messages-import.service.ts
@ -6,13 +6,13 @@ import { CacheStorageService } from 'src/engine/integrations/cache-storage/cache
 import { InjectCacheStorage } from 'src/engine/integrations/cache-storage/decorators/cache-storage.decorator';
 import { CacheStorageNamespace } from 'src/engine/integrations/cache-storage/types/cache-storage-namespace.enum';
 import { InjectObjectMetadataRepository } from 'src/engine/object-metadata-repository/object-metadata-repository.decorator';
+import { TwentyORMManager } from 'src/engine/twenty-orm/twenty-orm.manager';
 import { BlocklistRepository } from 'src/modules/blocklist/repositories/blocklist.repository';
 import { BlocklistWorkspaceEntity } from 'src/modules/blocklist/standard-objects/blocklist.workspace-entity';
 import { EmailAliasManagerService } from 'src/modules/connected-account/email-alias-manager/services/email-alias-manager.service';
 import { RefreshAccessTokenService } from 'src/modules/connected-account/refresh-access-token-manager/services/refresh-access-token.service';
 import { ConnectedAccountRepository } from 'src/modules/connected-account/repositories/connected-account.repository';
 import { ConnectedAccountWorkspaceEntity } from 'src/modules/connected-account/standard-objects/connected-account.workspace-entity';
-import { MessageChannelRepository } from 'src/modules/messaging/common/repositories/message-channel.repository';
 import { MessagingChannelSyncStatusService } from 'src/modules/messaging/common/services/messaging-channel-sync-status.service';
 import {
  MessageChannelSyncStage,
@ -40,12 +40,11 @@ export class MessagingMessagesImportService {
    private readonly messagingTelemetryService: MessagingTelemetryService,
    @InjectObjectMetadataRepository(BlocklistWorkspaceEntity)
    private readonly blocklistRepository: BlocklistRepository,
-    @InjectObjectMetadataRepository(MessageChannelWorkspaceEntity)
-    private readonly messageChannelRepository: MessageChannelRepository,
    private readonly emailAliasManagerService: EmailAliasManagerService,
    private readonly isFeatureEnabledService: IsFeatureEnabledService,
    @InjectObjectMetadataRepository(ConnectedAccountWorkspaceEntity)
    private readonly connectedAccountRepository: ConnectedAccountRepository,
+    private readonly twentyORMManager: TwentyORMManager,
  ) {}

  async processMessageBatchImport(
@ -73,7 +72,6 @@ export class MessagingMessagesImportService {

    await this.messagingChannelSyncStatusService.markAsMessagesImportOngoing(
      messageChannel.id,
-      workspaceId,
    );

    let accessToken: string;
@ -139,7 +137,6 @@ export class MessagingMessagesImportService {
    if (!messageIdsToFetch?.length) {
      await this.messagingChannelSyncStatusService.markAsCompletedAndSchedulePartialMessageListFetch(
        messageChannel.id,
-        workspaceId,
      );

      return await this.trackMessageImportCompleted(
@ -180,23 +177,26 @@ export class MessagingMessagesImportService {
      ) {
        await this.messagingChannelSyncStatusService.markAsCompletedAndSchedulePartialMessageListFetch(
          messageChannel.id,
-          workspaceId,
        );
      } else {
        await this.messagingChannelSyncStatusService.scheduleMessagesImport(
          messageChannel.id,
-          workspaceId,
        );
      }

-      await this.messageChannelRepository.resetThrottleFailureCount(
-        messageChannel.id,
-        workspaceId,
-      );
+      const messageChannelRepository =
+        await this.twentyORMManager.getRepository<MessageChannelWorkspaceEntity>(
+          'messageChannel',
+        );

-      await this.messageChannelRepository.resetSyncStageStartedAt(
-        messageChannel.id,
-        workspaceId,
+      await messageChannelRepository.update(
+        {
+          id: messageChannel.id,
+        },
+        {
+          throttleFailureCount: 0,
+          syncStageStartedAt: null,
+        },
      );

      return await this.trackMessageImportCompleted(
--- a/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-partial-message-list-fetch.service.ts
+++ b/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-partial-message-list-fetch.service.ts
@ -6,10 +6,8 @@ import { Any } from 'typeorm';
 import { CacheStorageService } from 'src/engine/integrations/cache-storage/cache-storage.service';
 import { InjectCacheStorage } from 'src/engine/integrations/cache-storage/decorators/cache-storage.decorator';
 import { CacheStorageNamespace } from 'src/engine/integrations/cache-storage/types/cache-storage-namespace.enum';
-import { InjectObjectMetadataRepository } from 'src/engine/object-metadata-repository/object-metadata-repository.decorator';
 import { TwentyORMManager } from 'src/engine/twenty-orm/twenty-orm.manager';
 import { ConnectedAccountWorkspaceEntity } from 'src/modules/connected-account/standard-objects/connected-account.workspace-entity';
-import { MessageChannelRepository } from 'src/modules/messaging/common/repositories/message-channel.repository';
 import { MessagingChannelSyncStatusService } from 'src/modules/messaging/common/services/messaging-channel-sync-status.service';
 import { MessageChannelMessageAssociationWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-channel-message-association.workspace-entity';
 import { MessageChannelWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity';
@ -26,8 +24,6 @@ export class MessagingPartialMessageListFetchService {

  constructor(
    private readonly gmailClientProvider: MessagingGmailClientProvider,
-    @InjectObjectMetadataRepository(MessageChannelWorkspaceEntity)
-    private readonly messageChannelRepository: MessageChannelRepository,
    @InjectCacheStorage(CacheStorageNamespace.Messaging)
    private readonly cacheStorage: CacheStorageService,
    private readonly gmailErrorHandlingService: MessagingErrorHandlingService,
@ -44,7 +40,6 @@ export class MessagingPartialMessageListFetchService {
  ): Promise<void> {
    await this.messagingChannelSyncStatusService.markAsMessagesListFetchOngoing(
      messageChannel.id,
-      workspaceId,
    );

    const lastSyncHistoryId = messageChannel.syncCursor;
@ -69,14 +64,19 @@ export class MessagingPartialMessageListFetchService {
      return;
    }

-    await this.messageChannelRepository.resetThrottleFailureCount(
-      messageChannel.id,
-      workspaceId,
-    );
+    const messageChannelRepository =
+      await this.twentyORMManager.getRepository<MessageChannelWorkspaceEntity>(
+        'messageChannel',
+      );

-    await this.messageChannelRepository.resetSyncStageStartedAt(
-      messageChannel.id,
-      workspaceId,
+    await messageChannelRepository.update(
+      {
+        id: messageChannel.id,
+      },
+      {
+        throttleFailureCount: 0,
+        syncStageStartedAt: null,
+      },
    );

    if (!historyId) {
@ -92,7 +92,6 @@ export class MessagingPartialMessageListFetchService {

      await this.messagingChannelSyncStatusService.markAsCompletedAndSchedulePartialMessageListFetch(
        messageChannel.id,
-        workspaceId,
      );

      return;
@ -147,15 +146,21 @@ export class MessagingPartialMessageListFetchService {
      `Deleted ${messagesDeleted.length} messages for workspace ${workspaceId} and account ${connectedAccount.id}`,
    );

-    await this.messageChannelRepository.updateLastSyncCursorIfHigher(
-      messageChannel.id,
-      historyId,
-      workspaceId,
-    );
+    const currentSyncCursor = messageChannel.syncCursor;
+
+    if (!currentSyncCursor || historyId > currentSyncCursor) {
+      await messageChannelRepository.update(
+        {
+          id: messageChannel.id,
+        },
+        {
+          syncCursor: historyId,
+        },
+      );
+    }

    await this.messagingChannelSyncStatusService.scheduleMessagesImport(
      messageChannel.id,
-      workspaceId,
    );
  }
 }