refactor messaging throttle logic

2026-04-12 18:48:29 +05:30
8 changed files with 86 additions and 110 deletions
@@ -1,3 +1,4 @@
+import { Logger } from '@nestjs/common';
 import { InjectDataSource, InjectRepository } from '@nestjs/typeorm';

 import { WorkspaceActivationStatus } from 'twenty-shared/workspace';
@@ -5,6 +6,8 @@ import { DataSource, Repository } from 'typeorm';

 import { CalendarChannelSyncStage } from 'twenty-shared/types';
 import { SentryCronMonitor } from 'src/engine/core-modules/cron/sentry-cron-monitor.decorator';
+import { CalendarChannelEntity } from 'src/engine/metadata-modules/calendar-channel/entities/calendar-channel.entity';
+import { isThrottled } from 'src/modules/connected-account/utils/is-throttled';
 import { ExceptionHandlerService } from 'src/engine/core-modules/exception-handler/exception-handler.service';
 import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator';
 import { Process } from 'src/engine/core-modules/message-queue/decorators/process.decorator';
@@ -23,6 +26,8 @@ export const CALENDAR_EVENT_LIST_FETCH_CRON_PATTERN = '*/5 * * * *';
  queueName: MessageQueue.cronQueue,
 })
 export class CalendarEventListFetchCronJob {
+  private readonly logger = new Logger(CalendarEventListFetchCronJob.name);
+
  constructor(
    @InjectRepository(WorkspaceEntity)
    private readonly workspaceRepository: Repository<WorkspaceEntity>,
@@ -49,12 +54,25 @@ export class CalendarEventListFetchCronJob {
      try {
        const now = new Date().toISOString();

-        const [calendarChannels] = await this.coreDataSource.query(
-          `UPDATE core."calendarChannel" SET "syncStage" = '${CalendarChannelSyncStage.CALENDAR_EVENT_LIST_FETCH_SCHEDULED}', "syncStageStartedAt" = COALESCE("syncStageStartedAt", '${now}')
+        const [calendarChannels]: [CalendarChannelEntity[]] =
+          await this.coreDataSource.query(
+            `UPDATE core."calendarChannel" SET "syncStage" = '${CalendarChannelSyncStage.CALENDAR_EVENT_LIST_FETCH_SCHEDULED}', "syncStageStartedAt" = COALESCE("syncStageStartedAt", '${now}')
           WHERE "workspaceId" = '${activeWorkspace.id}' AND "isSyncEnabled" = true AND "syncStage" = '${CalendarChannelSyncStage.CALENDAR_EVENT_LIST_FETCH_PENDING}' RETURNING *`,
-        );
+          );

        for (const calendarChannel of calendarChannels) {
+          if (
+            isThrottled(
+              calendarChannel.syncStageStartedAt?.toISOString() ?? null,
+              calendarChannel.throttleFailureCount,
+            )
+          ) {
+            this.logger.debug(
+              `Skipping throttled calendar channel ${calendarChannel.id} in workspace ${activeWorkspace.id}`,
+            );
+            continue;
+          }
+
          await this.messageQueueService.add<CalendarEventListFetchJobData>(
            CalendarEventListFetchJob.name,
            {
@@ -1,3 +1,4 @@
+import { Logger } from '@nestjs/common';
 import { InjectDataSource, InjectRepository } from '@nestjs/typeorm';

 import { WorkspaceActivationStatus } from 'twenty-shared/workspace';
@@ -5,6 +6,8 @@ import { DataSource, Repository } from 'typeorm';

 import { CalendarChannelSyncStage } from 'twenty-shared/types';
 import { SentryCronMonitor } from 'src/engine/core-modules/cron/sentry-cron-monitor.decorator';
+import { CalendarChannelEntity } from 'src/engine/metadata-modules/calendar-channel/entities/calendar-channel.entity';
+import { isThrottled } from 'src/modules/connected-account/utils/is-throttled';
 import { ExceptionHandlerService } from 'src/engine/core-modules/exception-handler/exception-handler.service';
 import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator';
 import { Process } from 'src/engine/core-modules/message-queue/decorators/process.decorator';
@@ -21,6 +24,8 @@ export const CALENDAR_EVENTS_IMPORT_CRON_PATTERN = '*/1 * * * *';
  queueName: MessageQueue.cronQueue,
 })
 export class CalendarEventsImportCronJob {
+  private readonly logger = new Logger(CalendarEventsImportCronJob.name);
+
  constructor(
    @InjectRepository(WorkspaceEntity)
    private readonly workspaceRepository: Repository<WorkspaceEntity>,
@@ -47,12 +52,25 @@ export class CalendarEventsImportCronJob {
      try {
        const now = new Date().toISOString();

-        const [calendarChannels] = await this.coreDataSource.query(
-          `UPDATE core."calendarChannel" SET "syncStage" = '${CalendarChannelSyncStage.CALENDAR_EVENTS_IMPORT_SCHEDULED}', "syncStageStartedAt" = COALESCE("syncStageStartedAt", '${now}')
+        const [calendarChannels]: [CalendarChannelEntity[]] =
+          await this.coreDataSource.query(
+            `UPDATE core."calendarChannel" SET "syncStage" = '${CalendarChannelSyncStage.CALENDAR_EVENTS_IMPORT_SCHEDULED}', "syncStageStartedAt" = COALESCE("syncStageStartedAt", '${now}')
           WHERE "workspaceId" = '${activeWorkspace.id}' AND "isSyncEnabled" = true AND "syncStage" = '${CalendarChannelSyncStage.CALENDAR_EVENTS_IMPORT_PENDING}' RETURNING *`,
-        );
+          );

        for (const calendarChannel of calendarChannels) {
+          if (
+            isThrottled(
+              calendarChannel.syncStageStartedAt?.toISOString() ?? null,
+              calendarChannel.throttleFailureCount,
+            )
+          ) {
+            this.logger.debug(
+              `Skipping throttled calendar channel ${calendarChannel.id} in workspace ${activeWorkspace.id}`,
+            );
+            continue;
+          }
+
          await this.messageQueueService.add<CalendarEventListFetchJobData>(
            CalendarEventsImportJob.name,
            {
@@ -10,8 +10,6 @@ import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queu
 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 { CalendarFetchEventsService } from 'src/modules/calendar/calendar-event-import-manager/services/calendar-fetch-events.service';
-import { CalendarChannelSyncStatusService } from 'src/modules/calendar/common/services/calendar-channel-sync-status.service';
-import { isThrottled } from 'src/modules/connected-account/utils/is-throttled';
 import { CalendarChannelEntity } from 'src/engine/metadata-modules/calendar-channel/entities/calendar-channel.entity';

 export type CalendarEventListFetchJobData = {
@@ -28,7 +26,6 @@ export class CalendarEventListFetchJob {
    private readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager,
    @InjectRepository(CalendarChannelEntity)
    private readonly calendarChannelRepository: Repository<CalendarChannelEntity>,
-    private readonly calendarChannelSyncStatusService: CalendarChannelSyncStatusService,
    private readonly calendarFetchEventsService: CalendarFetchEventsService,
  ) {}

@@ -59,23 +56,6 @@ export class CalendarEventListFetchJob {
        return;
      }

-      const syncStageStartedAt = calendarChannel.syncStageStartedAt;
-
-      if (
-        isThrottled(
-          syncStageStartedAt?.toISOString() ?? null,
-          calendarChannel.throttleFailureCount,
-        )
-      ) {
-        await this.calendarChannelSyncStatusService.markAsCalendarEventListFetchPending(
-          [calendarChannel.id],
-          workspaceId,
-          true,
-        );
-
-        return;
-      }
-
      await this.calendarFetchEventsService.fetchCalendarEvents(
        calendarChannel as unknown as CalendarChannelEntity,
        calendarChannel.connectedAccount,
@@ -10,8 +10,6 @@ import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queu
 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 { CalendarEventsImportService } from 'src/modules/calendar/calendar-event-import-manager/services/calendar-events-import.service';
-import { CalendarChannelSyncStatusService } from 'src/modules/calendar/common/services/calendar-channel-sync-status.service';
-import { isThrottled } from 'src/modules/connected-account/utils/is-throttled';
 import { CalendarChannelEntity } from 'src/engine/metadata-modules/calendar-channel/entities/calendar-channel.entity';

 export type CalendarEventsImportJobData = {
@@ -26,7 +24,6 @@ export type CalendarEventsImportJobData = {
 export class CalendarEventsImportJob {
  constructor(
    private readonly calendarEventsImportService: CalendarEventsImportService,
-    private readonly calendarChannelSyncStatusService: CalendarChannelSyncStatusService,
    private readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager,
    @InjectRepository(CalendarChannelEntity)
    private readonly calendarChannelRepository: Repository<CalendarChannelEntity>,
@@ -59,23 +56,6 @@ export class CalendarEventsImportJob {
        return;
      }

-      const syncStageStartedAt = calendarChannel.syncStageStartedAt;
-
-      if (
-        isThrottled(
-          syncStageStartedAt?.toISOString() ?? null,
-          calendarChannel.throttleFailureCount,
-        )
-      ) {
-        await this.calendarChannelSyncStatusService.markAsCalendarEventsImportPending(
-          [calendarChannel.id],
-          workspaceId,
-          true,
-        );
-
-        return;
-      }
-
      await this.calendarEventsImportService.processCalendarEventsImport(
        calendarChannel as unknown as CalendarChannelEntity,
        calendarChannel.connectedAccount,
@@ -1,3 +1,4 @@
+import { Logger } from '@nestjs/common';
 import { InjectDataSource, InjectRepository } from '@nestjs/typeorm';

 import { WorkspaceActivationStatus } from 'twenty-shared/workspace';
@@ -5,6 +6,8 @@ import { DataSource, Repository } from 'typeorm';

 import { MessageChannelSyncStage } from 'twenty-shared/types';
 import { SentryCronMonitor } from 'src/engine/core-modules/cron/sentry-cron-monitor.decorator';
+import { MessageChannelEntity } from 'src/engine/metadata-modules/message-channel/entities/message-channel.entity';
+import { isThrottled } from 'src/modules/connected-account/utils/is-throttled';
 import { ExceptionHandlerService } from 'src/engine/core-modules/exception-handler/exception-handler.service';
 import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator';
 import { Process } from 'src/engine/core-modules/message-queue/decorators/process.decorator';
@@ -21,6 +24,8 @@ 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>,
@@ -47,12 +52,26 @@ export class MessagingMessageListFetchCronJob {
      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}')
+        const [messageChannels]: [MessageChannelEntity[]] =
+          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 *`,
-        );
+          );

        for (const messageChannel of messageChannels) {
+          if (
+            isThrottled(
+              messageChannel.syncStageStartedAt?.toISOString() ?? null,
+              messageChannel.throttleFailureCount,
+              messageChannel.throttleRetryAfter?.toISOString() ?? null,
+            )
+          ) {
+            this.logger.debug(
+              `Skipping throttled message channel ${messageChannel.id} in workspace ${activeWorkspace.id}`,
+            );
+            continue;
+          }
+
          await this.messageQueueService.add<MessagingMessageListFetchJobData>(
            MessagingMessageListFetchJob.name,
            {
@@ -1,3 +1,4 @@
+import { Logger } from '@nestjs/common';
 import { InjectDataSource, InjectRepository } from '@nestjs/typeorm';

 import { isDefined } from 'twenty-shared/utils';
@@ -6,6 +7,8 @@ import { DataSource, Repository } from 'typeorm';

 import { MessageChannelSyncStage } from 'twenty-shared/types';
 import { SentryCronMonitor } from 'src/engine/core-modules/cron/sentry-cron-monitor.decorator';
+import { MessageChannelEntity } from 'src/engine/metadata-modules/message-channel/entities/message-channel.entity';
+import { isThrottled } from 'src/modules/connected-account/utils/is-throttled';
 import { ExceptionHandlerService } from 'src/engine/core-modules/exception-handler/exception-handler.service';
 import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator';
 import { Process } from 'src/engine/core-modules/message-queue/decorators/process.decorator';
@@ -22,6 +25,8 @@ 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>,
@@ -48,12 +53,26 @@ export class MessagingMessagesImportCronJob {
      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}')
+        const [messageChannels]: [MessageChannelEntity[]] =
+          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 *`,
-        );
+          );

        for (const messageChannel of messageChannels) {
+          if (
+            isThrottled(
+              messageChannel.syncStageStartedAt?.toISOString() ?? null,
+              messageChannel.throttleFailureCount,
+              messageChannel.throttleRetryAfter?.toISOString() ?? null,
+            )
+          ) {
+            this.logger.debug(
+              `Skipping throttled message channel ${messageChannel.id} in workspace ${activeWorkspace.id}`,
+            );
+            continue;
+          }
+
          await this.messageQueueService.add<MessagingMessagesImportJobData>(
            MessagingMessagesImportJob.name,
            {
@@ -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,