refactor messaging jobs (#19626)

Cleans up the code quality by migrating from Raw SQL to TypeORM entities. The previous implementation was necessary to do cross‑schema table joins but since we've migrated to the core schema we don't need it anymore. - Also extracted `toIsoStringOrNull` to a utility it was duplicated several times - Moved `isThrottled` logic from job handler to cron enqueuer
2026-04-13 20:09:52 +05:30
parent 9f6855e7dd
commit 7dfc556250
14 changed files with 336 additions and 169 deletions
@@ -1,7 +1,8 @@
-import { InjectDataSource, InjectRepository } from '@nestjs/typeorm';
+import { Logger } from '@nestjs/common';
+import { InjectRepository } from '@nestjs/typeorm';

 import { WorkspaceActivationStatus } from 'twenty-shared/workspace';
-import { DataSource, Repository } from 'typeorm';
+import { In, Repository } from 'typeorm';

 import { MessageChannelSyncStage } from 'twenty-shared/types';
 import { SentryCronMonitor } from 'src/engine/core-modules/cron/sentry-cron-monitor.decorator';
@@ -16,18 +17,23 @@ import {
  MessagingMessageListFetchJob,
  type MessagingMessageListFetchJobData,
 } from 'src/modules/messaging/message-import-manager/jobs/messaging-message-list-fetch.job';
+import { isThrottled } from 'src/modules/connected-account/utils/is-throttled';
+import { MessageChannelEntity } from 'src/engine/metadata-modules/message-channel/entities/message-channel.entity';
+import { toIsoStringOrNull } from 'src/utils/date/toIsoStringOrNull';

 export const MESSAGING_MESSAGE_LIST_FETCH_CRON_PATTERN = '2-59/5 * * * *';

@Processor(MessageQueue.cronQueue)
 export class MessagingMessageListFetchCronJob {
+  private readonly logger = new Logger(MessagingMessageListFetchCronJob.name);
+
  constructor(
    @InjectRepository(WorkspaceEntity)
    private readonly workspaceRepository: Repository<WorkspaceEntity>,
    @InjectMessageQueue(MessageQueue.messagingQueue)
    private readonly messageQueueService: MessageQueueService,
-    @InjectDataSource()
-    private readonly coreDataSource: DataSource,
+    @InjectRepository(MessageChannelEntity)
+    private readonly messageChannelRepository: Repository<MessageChannelEntity>,
    private readonly exceptionHandlerService: ExceptionHandlerService,
  ) {}

@@ -45,19 +51,68 @@ export class MessagingMessageListFetchCronJob {

    for (const activeWorkspace of activeWorkspaces) {
      try {
-        const now = new Date().toISOString();
-
-        const [messageChannels] = await this.coreDataSource.query(
-          `UPDATE core."messageChannel" SET "syncStage" = '${MessageChannelSyncStage.MESSAGE_LIST_FETCH_SCHEDULED}', "syncStageStartedAt" = COALESCE("syncStageStartedAt", '${now}')
-           WHERE "workspaceId" = '${activeWorkspace.id}' AND "isSyncEnabled" = true AND "syncStage" = '${MessageChannelSyncStage.MESSAGE_LIST_FETCH_PENDING}' RETURNING *`,
+        const pendingMessageChannels = await this.messageChannelRepository.find(
+          {
+            where: {
+              workspaceId: activeWorkspace.id,
+              isSyncEnabled: true,
+              syncStage: MessageChannelSyncStage.MESSAGE_LIST_FETCH_PENDING,
+            },
+          },
        );

-        for (const messageChannel of messageChannels) {
+        const messageChannelsToSchedule = pendingMessageChannels.filter(
+          (messageChannel) =>
+            !isThrottled(
+              toIsoStringOrNull(messageChannel.syncStageStartedAt),
+              messageChannel.throttleFailureCount,
+              toIsoStringOrNull(messageChannel.throttleRetryAfter),
+            ),
+        );
+
+        const throttledCount =
+          pendingMessageChannels.length - messageChannelsToSchedule.length;
+
+        if (throttledCount > 0) {
+          this.logger.log(
+            `Skipped ${throttledCount} throttled message channels for workspace ${activeWorkspace.id}`,
+          );
+        }
+
+        if (messageChannelsToSchedule.length === 0) {
+          continue;
+        }
+
+        const messageChannelIdsToSchedule = messageChannelsToSchedule.map(
+          (messageChannel) => messageChannel.id,
+        );
+
+        const updateResult = await this.messageChannelRepository
+          .createQueryBuilder()
+          .update()
+          .set({
+            syncStage: MessageChannelSyncStage.MESSAGE_LIST_FETCH_SCHEDULED,
+            syncStageStartedAt: new Date(),
+          })
+          .where({
+            id: In(messageChannelIdsToSchedule),
+            workspaceId: activeWorkspace.id,
+            isSyncEnabled: true,
+            syncStage: MessageChannelSyncStage.MESSAGE_LIST_FETCH_PENDING,
+          })
+          .returning('id')
+          .execute();
+
+        const updatedIds = updateResult.raw.map(
+          (row: { id: string }) => row.id,
+        );
+
+        for (const messageChannelId of updatedIds) {
          await this.messageQueueService.add<MessagingMessageListFetchJobData>(
            MessagingMessageListFetchJob.name,
            {
              workspaceId: activeWorkspace.id,
-              messageChannelId: messageChannel.id,
+              messageChannelId,
            },
          );
        }
@@ -1,8 +1,9 @@
-import { InjectDataSource, InjectRepository } from '@nestjs/typeorm';
+import { Logger } from '@nestjs/common';
+import { InjectRepository } from '@nestjs/typeorm';

 import { isDefined } from 'twenty-shared/utils';
 import { WorkspaceActivationStatus } from 'twenty-shared/workspace';
-import { DataSource, Repository } from 'typeorm';
+import { In, Repository } from 'typeorm';

 import { MessageChannelSyncStage } from 'twenty-shared/types';
 import { SentryCronMonitor } from 'src/engine/core-modules/cron/sentry-cron-monitor.decorator';
@@ -17,19 +18,24 @@ import {
  MessagingMessagesImportJob,
  type MessagingMessagesImportJobData,
 } from 'src/modules/messaging/message-import-manager/jobs/messaging-messages-import.job';
+import { isThrottled } from 'src/modules/connected-account/utils/is-throttled';
+import { MessageChannelEntity } from 'src/engine/metadata-modules/message-channel/entities/message-channel.entity';
+import { toIsoStringOrNull } from 'src/utils/date/toIsoStringOrNull';

 export const MESSAGING_MESSAGES_IMPORT_CRON_PATTERN = '*/1 * * * *';

@Processor(MessageQueue.cronQueue)
 export class MessagingMessagesImportCronJob {
+  private readonly logger = new Logger(MessagingMessagesImportCronJob.name);
+
  constructor(
    @InjectRepository(WorkspaceEntity)
    private readonly workspaceRepository: Repository<WorkspaceEntity>,
    @InjectMessageQueue(MessageQueue.messagingQueue)
    private readonly messageQueueService: MessageQueueService,
    private readonly exceptionHandlerService: ExceptionHandlerService,
-    @InjectDataSource()
-    private readonly coreDataSource: DataSource,
+    @InjectRepository(MessageChannelEntity)
+    private readonly messageChannelRepository: Repository<MessageChannelEntity>,
  ) {}

  @Process(MessagingMessagesImportCronJob.name)
@@ -46,19 +52,68 @@ export class MessagingMessagesImportCronJob {

    for (const activeWorkspace of activeWorkspaces) {
      try {
-        const now = new Date().toISOString();
-
-        const [messageChannels] = await this.coreDataSource.query(
-          `UPDATE core."messageChannel" SET "syncStage" = '${MessageChannelSyncStage.MESSAGES_IMPORT_SCHEDULED}', "syncStageStartedAt" = COALESCE("syncStageStartedAt", '${now}')
-           WHERE "workspaceId" = '${activeWorkspace.id}' AND "isSyncEnabled" = true AND "syncStage" = '${MessageChannelSyncStage.MESSAGES_IMPORT_PENDING}' RETURNING *`,
+        const pendingMessageChannels = await this.messageChannelRepository.find(
+          {
+            where: {
+              workspaceId: activeWorkspace.id,
+              isSyncEnabled: true,
+              syncStage: MessageChannelSyncStage.MESSAGES_IMPORT_PENDING,
+            },
+          },
        );

-        for (const messageChannel of messageChannels) {
+        const messageChannelsToSchedule = pendingMessageChannels.filter(
+          (messageChannel) =>
+            !isThrottled(
+              toIsoStringOrNull(messageChannel.syncStageStartedAt),
+              messageChannel.throttleFailureCount,
+              toIsoStringOrNull(messageChannel.throttleRetryAfter),
+            ),
+        );
+
+        const throttledCount =
+          pendingMessageChannels.length - messageChannelsToSchedule.length;
+
+        if (throttledCount > 0) {
+          this.logger.log(
+            `Skipped ${throttledCount} throttled message channels for workspace ${activeWorkspace.id}`,
+          );
+        }
+
+        if (messageChannelsToSchedule.length === 0) {
+          continue;
+        }
+
+        const messageChannelIdsToSchedule = messageChannelsToSchedule.map(
+          (messageChannel) => messageChannel.id,
+        );
+
+        const updateResult = await this.messageChannelRepository
+          .createQueryBuilder()
+          .update()
+          .set({
+            syncStage: MessageChannelSyncStage.MESSAGES_IMPORT_SCHEDULED,
+            syncStageStartedAt: new Date(),
+          })
+          .where({
+            id: In(messageChannelIdsToSchedule),
+            workspaceId: activeWorkspace.id,
+            isSyncEnabled: true,
+            syncStage: MessageChannelSyncStage.MESSAGES_IMPORT_PENDING,
+          })
+          .returning('id')
+          .execute();
+
+        const updatedIds = updateResult.raw.map(
+          (row: { id: string }) => row.id,
+        );
+
+        for (const messageChannelId of updatedIds) {
          await this.messageQueueService.add<MessagingMessagesImportJobData>(
            MessagingMessagesImportJob.name,
            {
              workspaceId: activeWorkspace.id,
-              messageChannelId: messageChannel.id,
+              messageChannelId,
            },
          );
        }
@@ -9,8 +9,6 @@ import { Processor } from 'src/engine/core-modules/message-queue/decorators/proc
 import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
 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 { isThrottled } from 'src/modules/connected-account/utils/is-throttled';
-import { MessageChannelSyncStatusService } from 'src/modules/messaging/common/services/message-channel-sync-status.service';
 import {
  MessageImportExceptionHandlerService,
  MessageImportSyncStep,
@@ -19,16 +17,6 @@ import { MessagingMessageListFetchService } from 'src/modules/messaging/message-
 import { MessagingMonitoringService } from 'src/modules/messaging/monitoring/services/messaging-monitoring.service';
 import { MessageChannelEntity } from 'src/engine/metadata-modules/message-channel/entities/message-channel.entity';

-const toIsoStringOrNull = (
-  value: string | Date | null | undefined,
-): string | null => {
-  if (value == null) {
-    return null;
-  }
-
-  return value instanceof Date ? value.toISOString() : value;
-};
-
 export type MessagingMessageListFetchJobData = {
  messageChannelId: string;
  workspaceId: string;
@@ -46,7 +34,6 @@ export class MessagingMessageListFetchJob {
    @InjectRepository(MessageChannelEntity)
    private readonly messageChannelRepository: Repository<MessageChannelEntity>,
    private readonly messageImportErrorHandlerService: MessageImportExceptionHandlerService,
-    private readonly messageChannelSyncStatusService: MessageChannelSyncStatusService,
  ) {}

  @Process(MessagingMessageListFetchJob.name)
@@ -88,22 +75,6 @@ export class MessagingMessageListFetchJob {
      }

      try {
-        if (
-          isThrottled(
-            toIsoStringOrNull(messageChannel.syncStageStartedAt),
-            messageChannel.throttleFailureCount,
-            toIsoStringOrNull(messageChannel.throttleRetryAfter),
-          )
-        ) {
-          await this.messageChannelSyncStatusService.markAsMessagesListFetchPending(
-            [messageChannel.id],
-            workspaceId,
-            true,
-          );
-
-          return;
-        }
-
        await this.messagingMonitoringService.track({
          eventName: 'message_list_fetch.started',
          workspaceId,
@@ -9,22 +9,10 @@ import { Processor } from 'src/engine/core-modules/message-queue/decorators/proc
 import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
 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 { isThrottled } from 'src/modules/connected-account/utils/is-throttled';
-import { MessageChannelSyncStatusService } from 'src/modules/messaging/common/services/message-channel-sync-status.service';
 import { MessagingMessagesImportService } from 'src/modules/messaging/message-import-manager/services/messaging-messages-import.service';
 import { MessagingMonitoringService } from 'src/modules/messaging/monitoring/services/messaging-monitoring.service';
 import { MessageChannelEntity } from 'src/engine/metadata-modules/message-channel/entities/message-channel.entity';

-const toIsoStringOrNull = (
-  value: string | Date | null | undefined,
-): string | null => {
-  if (value == null) {
-    return null;
-  }
-
-  return value instanceof Date ? value.toISOString() : value;
-};
-
 export type MessagingMessagesImportJobData = {
  messageChannelId: string;
  workspaceId: string;
@@ -38,7 +26,6 @@ export class MessagingMessagesImportJob {
  constructor(
    private readonly messagingMessagesImportService: MessagingMessagesImportService,
    private readonly messagingMonitoringService: MessagingMonitoringService,
-    private readonly messageChannelSyncStatusService: MessageChannelSyncStatusService,
    private readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager,
    @InjectRepository(MessageChannelEntity)
    private readonly messageChannelRepository: Repository<MessageChannelEntity>,
@@ -86,22 +73,6 @@ export class MessagingMessagesImportJob {
        return;
      }

-      if (
-        isThrottled(
-          toIsoStringOrNull(messageChannel.syncStageStartedAt),
-          messageChannel.throttleFailureCount,
-          toIsoStringOrNull(messageChannel.throttleRetryAfter),
-        )
-      ) {
-        await this.messageChannelSyncStatusService.markAsMessagesImportPending(
-          [messageChannel.id],
-          workspaceId,
-          true,
-        );
-
-        return;
-      }
-
      await this.messagingMessagesImportService.processMessageBatchImport(
        messageChannel,
        messageChannel.connectedAccount,
@@ -12,16 +12,7 @@ import { GlobalWorkspaceOrmManager } from 'src/engine/twenty-orm/global-workspac
 import { buildSystemAuthContext } from 'src/engine/twenty-orm/utils/build-system-auth-context.util';
 import { MessageChannelSyncStatusService } from 'src/modules/messaging/common/services/message-channel-sync-status.service';
 import { isSyncStale } from 'src/modules/messaging/message-import-manager/utils/is-sync-stale.util';
-
-const toIsoStringOrNull = (
-  value: string | Date | null | undefined,
-): string | null => {
-  if (value == null) {
-    return null;
-  }
-
-  return value instanceof Date ? value.toISOString() : value;
-};
+import { toIsoStringOrNull } from 'src/utils/date/toIsoStringOrNull';

 export type MessagingOngoingStaleJobData = {
  workspaceId: string;