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; +};