From 7dfc556250a1d78e9a5ffa669802e68fb37090c6 Mon Sep 17 00:00:00 2001 From: neo773 <62795688+neo773@users.noreply.github.com> Date: Mon, 13 Apr 2026 20:09:52 +0530 Subject: [PATCH] refactor messaging jobs (#19626) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- packages/twenty-server/project.json | 4 +- .../calendar-event-list-fetch.cron.job.ts | 76 ++++++++++++++--- .../jobs/calendar-events-import.cron.job.ts | 83 ++++++++++++++++--- .../jobs/calendar-event-list-fetch.job.ts | 20 ----- .../jobs/calendar-events-import.job.ts | 20 ----- .../utils/__tests__/is-throttled.spec.ts | 18 ++++ .../connected-account/utils/is-throttled.ts | 28 +++---- .../messaging-message-list-fetch.cron.job.ts | 77 ++++++++++++++--- .../messaging-messages-import.cron.job.ts | 77 ++++++++++++++--- .../jobs/messaging-message-list-fetch.job.ts | 29 ------- .../jobs/messaging-messages-import.job.ts | 29 ------- .../jobs/messaging-ongoing-stale.job.ts | 11 +-- .../date/__tests__/toIsoStringOrNull.spec.ts | 24 ++++++ .../src/utils/date/toIsoStringOrNull.ts | 9 ++ 14 files changed, 336 insertions(+), 169 deletions(-) create mode 100644 packages/twenty-server/src/utils/date/__tests__/toIsoStringOrNull.spec.ts create mode 100644 packages/twenty-server/src/utils/date/toIsoStringOrNull.ts diff --git a/packages/twenty-server/project.json b/packages/twenty-server/project.json index c4254421be..54d809636e 100644 --- a/packages/twenty-server/project.json +++ b/packages/twenty-server/project.json @@ -117,10 +117,10 @@ }, "worker": { "executor": "nx:run-commands", - "dependsOn": ["build"], + "dependsOn": ["^build"], "options": { "cwd": "packages/twenty-server", - "command": "node dist/queue-worker/queue-worker.js" + "command": "NODE_ENV=development nest start --watch --entryFile queue-worker/queue-worker" } }, "ts-node": { diff --git a/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/crons/jobs/calendar-event-list-fetch.cron.job.ts b/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/crons/jobs/calendar-event-list-fetch.cron.job.ts index 890d33af0d..220674febf 100644 --- a/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/crons/jobs/calendar-event-list-fetch.cron.job.ts +++ b/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/crons/jobs/calendar-event-list-fetch.cron.job.ts @@ -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 { CalendarChannelSyncStage } from 'twenty-shared/types'; import { SentryCronMonitor } from 'src/engine/core-modules/cron/sentry-cron-monitor.decorator'; @@ -16,6 +17,9 @@ import { CalendarEventListFetchJob, type CalendarEventListFetchJobData, } from 'src/modules/calendar/calendar-event-import-manager/jobs/calendar-event-list-fetch.job'; +import { CalendarChannelEntity } from 'src/engine/metadata-modules/calendar-channel/entities/calendar-channel.entity'; +import { isThrottled } from 'src/modules/connected-account/utils/is-throttled'; +import { toIsoStringOrNull } from 'src/utils/date/toIsoStringOrNull'; export const CALENDAR_EVENT_LIST_FETCH_CRON_PATTERN = '*/5 * * * *'; @@ -23,14 +27,16 @@ 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, @InjectMessageQueue(MessageQueue.calendarQueue) private readonly messageQueueService: MessageQueueService, private readonly exceptionHandlerService: ExceptionHandlerService, - @InjectDataSource() - private readonly coreDataSource: DataSource, + @InjectRepository(CalendarChannelEntity) + private readonly calendarChannelRepository: Repository, ) {} @Process(CalendarEventListFetchCronJob.name) @@ -47,18 +53,68 @@ export class CalendarEventListFetchCronJob { for (const activeWorkspace of activeWorkspaces) { try { - const now = new Date().toISOString(); + const pendingCalendarChannels = + await this.calendarChannelRepository.find({ + where: { + workspaceId: activeWorkspace.id, + isSyncEnabled: true, + syncStage: + CalendarChannelSyncStage.CALENDAR_EVENT_LIST_FETCH_PENDING, + }, + }); - const [calendarChannels] = 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 *`, + const calendarChannelsToSchedule = pendingCalendarChannels.filter( + (calendarChannel) => + !isThrottled( + toIsoStringOrNull(calendarChannel.syncStageStartedAt), + calendarChannel.throttleFailureCount, + ), ); - for (const calendarChannel of calendarChannels) { + const throttledCount = + pendingCalendarChannels.length - calendarChannelsToSchedule.length; + + if (throttledCount > 0) { + this.logger.log( + `Skipped ${throttledCount} throttled calendar channels for workspace ${activeWorkspace.id}`, + ); + } + + if (calendarChannelsToSchedule.length === 0) { + continue; + } + + const calendarChannelIds = calendarChannelsToSchedule.map( + (calendarChannel) => calendarChannel.id, + ); + + const updateResult = await this.calendarChannelRepository + .createQueryBuilder() + .update() + .set({ + syncStage: + CalendarChannelSyncStage.CALENDAR_EVENT_LIST_FETCH_SCHEDULED, + syncStageStartedAt: new Date(), + }) + .where({ + id: In(calendarChannelIds), + workspaceId: activeWorkspace.id, + isSyncEnabled: true, + syncStage: + CalendarChannelSyncStage.CALENDAR_EVENT_LIST_FETCH_PENDING, + }) + .returning('id') + .execute(); + + const updatedIds = updateResult.raw.map( + (row: { id: string }) => row.id, + ); + + for (const calendarChannelId of updatedIds) { await this.messageQueueService.add( CalendarEventListFetchJob.name, { - calendarChannelId: calendarChannel.id, + calendarChannelId, workspaceId: activeWorkspace.id, }, ); diff --git a/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/crons/jobs/calendar-events-import.cron.job.ts b/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/crons/jobs/calendar-events-import.cron.job.ts index 060cfbc074..c97daa4075 100644 --- a/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/crons/jobs/calendar-events-import.cron.job.ts +++ b/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/crons/jobs/calendar-events-import.cron.job.ts @@ -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 { CalendarChannelSyncStage } from 'twenty-shared/types'; import { SentryCronMonitor } from 'src/engine/core-modules/cron/sentry-cron-monitor.decorator'; @@ -12,8 +13,13 @@ import { Processor } from 'src/engine/core-modules/message-queue/decorators/proc import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants'; import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service'; import { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity'; -import { type CalendarEventListFetchJobData } from 'src/modules/calendar/calendar-event-import-manager/jobs/calendar-event-list-fetch.job'; -import { CalendarEventsImportJob } from 'src/modules/calendar/calendar-event-import-manager/jobs/calendar-events-import.job'; +import { + CalendarEventsImportJob, + type CalendarEventsImportJobData, +} from 'src/modules/calendar/calendar-event-import-manager/jobs/calendar-events-import.job'; +import { CalendarChannelEntity } from 'src/engine/metadata-modules/calendar-channel/entities/calendar-channel.entity'; +import { isThrottled } from 'src/modules/connected-account/utils/is-throttled'; +import { toIsoStringOrNull } from 'src/utils/date/toIsoStringOrNull'; export const CALENDAR_EVENTS_IMPORT_CRON_PATTERN = '*/1 * * * *'; @@ -21,13 +27,15 @@ 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, @InjectMessageQueue(MessageQueue.calendarQueue) private readonly messageQueueService: MessageQueueService, - @InjectDataSource() - private readonly coreDataSource: DataSource, + @InjectRepository(CalendarChannelEntity) + private readonly calendarChannelRepository: Repository, private readonly exceptionHandlerService: ExceptionHandlerService, ) {} @@ -45,18 +53,67 @@ export class CalendarEventsImportCronJob { for (const activeWorkspace of activeWorkspaces) { try { - const now = new Date().toISOString(); + const pendingCalendarChannels = + await this.calendarChannelRepository.find({ + where: { + workspaceId: activeWorkspace.id, + isSyncEnabled: true, + syncStage: + CalendarChannelSyncStage.CALENDAR_EVENTS_IMPORT_PENDING, + }, + }); - const [calendarChannels] = 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 *`, + const calendarChannelsToSchedule = pendingCalendarChannels.filter( + (calendarChannel) => + !isThrottled( + toIsoStringOrNull(calendarChannel.syncStageStartedAt), + calendarChannel.throttleFailureCount, + ), ); - for (const calendarChannel of calendarChannels) { - await this.messageQueueService.add( + const throttledCount = + pendingCalendarChannels.length - calendarChannelsToSchedule.length; + + if (throttledCount > 0) { + this.logger.log( + `Skipped ${throttledCount} throttled calendar channels for workspace ${activeWorkspace.id}`, + ); + } + + if (calendarChannelsToSchedule.length === 0) { + continue; + } + + const calendarChannelIds = calendarChannelsToSchedule.map( + (calendarChannel) => calendarChannel.id, + ); + + const updateResult = await this.calendarChannelRepository + .createQueryBuilder() + .update() + .set({ + syncStage: + CalendarChannelSyncStage.CALENDAR_EVENTS_IMPORT_SCHEDULED, + syncStageStartedAt: new Date(), + }) + .where({ + id: In(calendarChannelIds), + workspaceId: activeWorkspace.id, + isSyncEnabled: true, + syncStage: CalendarChannelSyncStage.CALENDAR_EVENTS_IMPORT_PENDING, + }) + .returning('id') + .execute(); + + const updatedIds = updateResult.raw.map( + (row: { id: string }) => row.id, + ); + + for (const calendarChannelId of updatedIds) { + await this.messageQueueService.add( CalendarEventsImportJob.name, { - calendarChannelId: calendarChannel.id, + calendarChannelId, workspaceId: activeWorkspace.id, }, ); diff --git a/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/jobs/calendar-event-list-fetch.job.ts b/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/jobs/calendar-event-list-fetch.job.ts index 8d6f03e5a1..2271432210 100644 --- a/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/jobs/calendar-event-list-fetch.job.ts +++ b/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/jobs/calendar-event-list-fetch.job.ts @@ -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, - 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, diff --git a/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/jobs/calendar-events-import.job.ts b/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/jobs/calendar-events-import.job.ts index f37338388c..929aafa482 100644 --- a/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/jobs/calendar-events-import.job.ts +++ b/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/jobs/calendar-events-import.job.ts @@ -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, @@ -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, diff --git a/packages/twenty-server/src/modules/connected-account/utils/__tests__/is-throttled.spec.ts b/packages/twenty-server/src/modules/connected-account/utils/__tests__/is-throttled.spec.ts index 5ceb925ab7..80cb95fc8c 100644 --- a/packages/twenty-server/src/modules/connected-account/utils/__tests__/is-throttled.spec.ts +++ b/packages/twenty-server/src/modules/connected-account/utils/__tests__/is-throttled.spec.ts @@ -5,10 +5,28 @@ describe('isThrottled', () => { expect(isThrottled(null, 3)).toBe(false); }); + it('should throttle until retryAfter even when no sync stage is active', () => { + const fiveMinutesFromNow = new Date( + Date.now() + 5 * 60 * 1000, + ).toISOString(); + + expect(isThrottled(null, 3, fiveMinutesFromNow)).toBe(true); + }); + it('should not throttle when there have been no failures', () => { expect(isThrottled(new Date().toISOString(), 0)).toBe(false); }); + it('should throttle until retryAfter even when there have been no failures', () => { + const fiveMinutesFromNow = new Date( + Date.now() + 5 * 60 * 1000, + ).toISOString(); + + expect(isThrottled(new Date().toISOString(), 0, fiveMinutesFromNow)).toBe( + true, + ); + }); + it('should keep throttling when retryAfter is in the future even though exponential backoff has expired', () => { const tenMinutesAgo = new Date(Date.now() - 10 * 60 * 1000).toISOString(); const fiveMinutesFromNow = new Date( diff --git a/packages/twenty-server/src/modules/connected-account/utils/is-throttled.ts b/packages/twenty-server/src/modules/connected-account/utils/is-throttled.ts index 1eaee3b899..72be424050 100644 --- a/packages/twenty-server/src/modules/connected-account/utils/is-throttled.ts +++ b/packages/twenty-server/src/modules/connected-account/utils/is-throttled.ts @@ -8,6 +8,19 @@ export const isThrottled = ( throttleFailureCount: number, throttleRetryAfter?: string | null, ): boolean => { + const now = new Date(); + + const retryAfterCandidate = isDefined(throttleRetryAfter) + ? new Date(throttleRetryAfter) + : null; + const retryAfterDate = isValidDate(retryAfterCandidate) + ? retryAfterCandidate + : null; + + if (isDefined(retryAfterDate) && retryAfterDate > now) { + return true; + } + if (!syncStageStartedAt) { return false; } @@ -16,25 +29,12 @@ export const isThrottled = ( return false; } - const now = new Date(); - const exponentialBackoffUntil = computeThrottlePauseUntil( syncStageStartedAt, throttleFailureCount, ); - const retryAfterCandidate = isDefined(throttleRetryAfter) - ? new Date(throttleRetryAfter) - : null; - const retryAfterDate = isValidDate(retryAfterCandidate) - ? retryAfterCandidate - : null; - const effectiveUntil = - isDefined(retryAfterDate) && retryAfterDate > exponentialBackoffUntil - ? retryAfterDate - : exponentialBackoffUntil; - - return effectiveUntil > now; + return exponentialBackoffUntil > now; }; const computeThrottlePauseUntil = ( diff --git a/packages/twenty-server/src/modules/messaging/message-import-manager/crons/jobs/messaging-message-list-fetch.cron.job.ts b/packages/twenty-server/src/modules/messaging/message-import-manager/crons/jobs/messaging-message-list-fetch.cron.job.ts index ac945ccbc3..4e83cb4913 100644 --- a/packages/twenty-server/src/modules/messaging/message-import-manager/crons/jobs/messaging-message-list-fetch.cron.job.ts +++ b/packages/twenty-server/src/modules/messaging/message-import-manager/crons/jobs/messaging-message-list-fetch.cron.job.ts @@ -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, @InjectMessageQueue(MessageQueue.messagingQueue) private readonly messageQueueService: MessageQueueService, - @InjectDataSource() - private readonly coreDataSource: DataSource, + @InjectRepository(MessageChannelEntity) + private readonly messageChannelRepository: Repository, 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( MessagingMessageListFetchJob.name, { workspaceId: activeWorkspace.id, - messageChannelId: messageChannel.id, + messageChannelId, }, ); } diff --git a/packages/twenty-server/src/modules/messaging/message-import-manager/crons/jobs/messaging-messages-import.cron.job.ts b/packages/twenty-server/src/modules/messaging/message-import-manager/crons/jobs/messaging-messages-import.cron.job.ts index 6e884052b7..02db9699a6 100644 --- a/packages/twenty-server/src/modules/messaging/message-import-manager/crons/jobs/messaging-messages-import.cron.job.ts +++ b/packages/twenty-server/src/modules/messaging/message-import-manager/crons/jobs/messaging-messages-import.cron.job.ts @@ -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, @InjectMessageQueue(MessageQueue.messagingQueue) private readonly messageQueueService: MessageQueueService, private readonly exceptionHandlerService: ExceptionHandlerService, - @InjectDataSource() - private readonly coreDataSource: DataSource, + @InjectRepository(MessageChannelEntity) + private readonly messageChannelRepository: Repository, ) {} @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( MessagingMessagesImportJob.name, { workspaceId: activeWorkspace.id, - messageChannelId: messageChannel.id, + messageChannelId, }, ); } diff --git a/packages/twenty-server/src/modules/messaging/message-import-manager/jobs/messaging-message-list-fetch.job.ts b/packages/twenty-server/src/modules/messaging/message-import-manager/jobs/messaging-message-list-fetch.job.ts index 3a7152bc9d..0cc6f868d8 100644 --- a/packages/twenty-server/src/modules/messaging/message-import-manager/jobs/messaging-message-list-fetch.job.ts +++ b/packages/twenty-server/src/modules/messaging/message-import-manager/jobs/messaging-message-list-fetch.job.ts @@ -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, 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, diff --git a/packages/twenty-server/src/modules/messaging/message-import-manager/jobs/messaging-messages-import.job.ts b/packages/twenty-server/src/modules/messaging/message-import-manager/jobs/messaging-messages-import.job.ts index c0c98d4b80..a8a30d91cc 100644 --- a/packages/twenty-server/src/modules/messaging/message-import-manager/jobs/messaging-messages-import.job.ts +++ b/packages/twenty-server/src/modules/messaging/message-import-manager/jobs/messaging-messages-import.job.ts @@ -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, @@ -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, diff --git a/packages/twenty-server/src/modules/messaging/message-import-manager/jobs/messaging-ongoing-stale.job.ts b/packages/twenty-server/src/modules/messaging/message-import-manager/jobs/messaging-ongoing-stale.job.ts index 7912a0e688..25021a2d05 100644 --- a/packages/twenty-server/src/modules/messaging/message-import-manager/jobs/messaging-ongoing-stale.job.ts +++ b/packages/twenty-server/src/modules/messaging/message-import-manager/jobs/messaging-ongoing-stale.job.ts @@ -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; diff --git a/packages/twenty-server/src/utils/date/__tests__/toIsoStringOrNull.spec.ts b/packages/twenty-server/src/utils/date/__tests__/toIsoStringOrNull.spec.ts new file mode 100644 index 0000000000..59e26b720a --- /dev/null +++ b/packages/twenty-server/src/utils/date/__tests__/toIsoStringOrNull.spec.ts @@ -0,0 +1,24 @@ +import { toIsoStringOrNull } from 'src/utils/date/toIsoStringOrNull'; + +describe('toIsoStringOrNull', () => { + it('should return null for null or undefined', () => { + expect(toIsoStringOrNull(null)).toBeNull(); + expect(toIsoStringOrNull(undefined)).toBeNull(); + }); + + it('should convert Date to ISO string', () => { + const date = new Date('2024-01-15T10:30:00.000Z'); + + expect(toIsoStringOrNull(date)).toBe('2024-01-15T10:30:00.000Z'); + }); + + it('should pass through strings unchanged', () => { + expect(toIsoStringOrNull('2024-01-15T10:30:00.000Z')).toBe( + '2024-01-15T10:30:00.000Z', + ); + }); + + it('should throw on invalid Date', () => { + expect(() => toIsoStringOrNull(new Date('invalid'))).toThrow(RangeError); + }); +}); diff --git a/packages/twenty-server/src/utils/date/toIsoStringOrNull.ts b/packages/twenty-server/src/utils/date/toIsoStringOrNull.ts new file mode 100644 index 0000000000..ba92adaa07 --- /dev/null +++ b/packages/twenty-server/src/utils/date/toIsoStringOrNull.ts @@ -0,0 +1,9 @@ +export const toIsoStringOrNull = ( + value: string | Date | null | undefined, +): string | null => { + if (value == null) { + return null; + } + + return value instanceof Date ? value.toISOString() : value; +};