diff --git a/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/constants/calendar-ongoing-stale-sync-stages.constant.ts b/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/constants/calendar-ongoing-stale-sync-stages.constant.ts new file mode 100644 index 0000000000..adee796a7f --- /dev/null +++ b/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/constants/calendar-ongoing-stale-sync-stages.constant.ts @@ -0,0 +1,8 @@ +import { CalendarChannelSyncStage } from 'twenty-shared/types'; + +export const CALENDAR_ONGOING_STALE_SYNC_STAGES: CalendarChannelSyncStage[] = [ + CalendarChannelSyncStage.CALENDAR_EVENTS_IMPORT_ONGOING, + CalendarChannelSyncStage.CALENDAR_EVENT_LIST_FETCH_ONGOING, + CalendarChannelSyncStage.CALENDAR_EVENTS_IMPORT_SCHEDULED, + CalendarChannelSyncStage.CALENDAR_EVENT_LIST_FETCH_SCHEDULED, +]; diff --git a/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/crons/jobs/__tests__/calendar-ongoing-stale.cron.job.spec.ts b/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/crons/jobs/__tests__/calendar-ongoing-stale.cron.job.spec.ts new file mode 100644 index 0000000000..359948eba0 --- /dev/null +++ b/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/crons/jobs/__tests__/calendar-ongoing-stale.cron.job.spec.ts @@ -0,0 +1,61 @@ +import { type Repository } from 'typeorm'; + +import { type ExceptionHandlerService } from 'src/engine/core-modules/exception-handler/exception-handler.service'; +import { type MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service'; +import { CalendarChannelEntity } from 'src/engine/metadata-modules/calendar-channel/entities/calendar-channel.entity'; +import { CalendarOngoingStaleCronJob } from 'src/modules/calendar/calendar-event-import-manager/crons/jobs/calendar-ongoing-stale.cron.job'; +import { CalendarOngoingStaleJob } from 'src/modules/calendar/calendar-event-import-manager/jobs/calendar-ongoing-stale.job'; + +describe('CalendarOngoingStaleCronJob', () => { + let calendarChannelRepository: { find: jest.Mock }; + let messageQueueService: { add: jest.Mock }; + let job: CalendarOngoingStaleCronJob; + + beforeEach(() => { + calendarChannelRepository = { + find: jest + .fn() + .mockResolvedValue([ + { workspaceId: 'workspace-1' }, + { workspaceId: 'workspace-1' }, + { workspaceId: 'workspace-2' }, + ]), + }; + messageQueueService = { add: jest.fn() }; + job = new CalendarOngoingStaleCronJob( + calendarChannelRepository as unknown as Repository, + messageQueueService as unknown as MessageQueueService, + { + captureExceptions: jest.fn(), + } as unknown as ExceptionHandlerService, + ); + }); + + it('enqueues recovery once per workspace with a stale channel', async () => { + await job.handle(); + + expect(messageQueueService.add).toHaveBeenCalledTimes(2); + expect(messageQueueService.add).toHaveBeenNthCalledWith( + 1, + CalendarOngoingStaleJob.name, + { + workspaceId: 'workspace-1', + }, + ); + expect(messageQueueService.add).toHaveBeenNthCalledWith( + 2, + CalendarOngoingStaleJob.name, + { + workspaceId: 'workspace-2', + }, + ); + }); + + it('does not enqueue recovery when no workspace has a stale channel', async () => { + calendarChannelRepository.find.mockResolvedValue([]); + + await job.handle(); + + expect(messageQueueService.add).not.toHaveBeenCalled(); + }); +}); diff --git a/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/crons/jobs/calendar-ongoing-stale.cron.job.ts b/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/crons/jobs/calendar-ongoing-stale.cron.job.ts index da3b151537..dbb9fc8d4c 100644 --- a/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/crons/jobs/calendar-ongoing-stale.cron.job.ts +++ b/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/crons/jobs/calendar-ongoing-stale.cron.job.ts @@ -1,7 +1,7 @@ import { InjectRepository } from '@nestjs/typeorm'; import { WorkspaceActivationStatus } from 'twenty-shared/workspace'; -import { Repository } from 'typeorm'; +import { In, IsNull, LessThan, Or, Repository } from 'typeorm'; import { SentryCronMonitor } from 'src/engine/core-modules/cron/sentry-cron-monitor.decorator'; import { ExceptionHandlerService } from 'src/engine/core-modules/exception-handler/exception-handler.service'; @@ -10,7 +10,9 @@ import { Process } from 'src/engine/core-modules/message-queue/decorators/proces import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.decorator'; 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 { CalendarChannelEntity } from 'src/engine/metadata-modules/calendar-channel/entities/calendar-channel.entity'; +import { CALENDAR_IMPORT_ONGOING_SYNC_TIMEOUT } from 'src/modules/calendar/calendar-event-import-manager/constants/calendar-import-ongoing-sync-timeout.constant'; +import { CALENDAR_ONGOING_STALE_SYNC_STAGES } from 'src/modules/calendar/calendar-event-import-manager/constants/calendar-ongoing-stale-sync-stages.constant'; import { CalendarOngoingStaleJob, type CalendarOngoingStaleJobData, @@ -21,8 +23,8 @@ export const CALENDAR_ONGOING_STALE_CRON_PATTERN = '0 * * * *'; @Processor(MessageQueue.cronQueue) export class CalendarOngoingStaleCronJob { constructor( - @InjectRepository(WorkspaceEntity) - private readonly workspaceRepository: Repository, + @InjectRepository(CalendarChannelEntity) + private readonly calendarChannelRepository: Repository, @InjectMessageQueue(MessageQueue.calendarQueue) private readonly messageQueueService: MessageQueueService, private readonly exceptionHandlerService: ExceptionHandlerService, @@ -34,27 +36,44 @@ export class CalendarOngoingStaleCronJob { CALENDAR_ONGOING_STALE_CRON_PATTERN, ) async handle(): Promise { - const activeWorkspaces = await this.workspaceRepository.find({ - where: { - activationStatus: WorkspaceActivationStatus.ACTIVE, - }, - }); + const staleWorkspaceIds = await this.findStaleWorkspaceIds(); - for (const activeWorkspace of activeWorkspaces) { + for (const workspaceId of staleWorkspaceIds) { try { await this.messageQueueService.add( CalendarOngoingStaleJob.name, { - workspaceId: activeWorkspace.id, + workspaceId, }, ); } catch (error) { this.exceptionHandlerService.captureExceptions([error], { workspace: { - id: activeWorkspace.id, + id: workspaceId, }, }); } } } + + private async findStaleWorkspaceIds(): Promise { + const staleBefore = new Date( + Date.now() - CALENDAR_IMPORT_ONGOING_SYNC_TIMEOUT, + ); + const staleChannels = await this.calendarChannelRepository.find({ + select: { + workspaceId: true, + }, + where: { + syncStage: In(CALENDAR_ONGOING_STALE_SYNC_STAGES), + syncStageStartedAt: Or(IsNull(), LessThan(staleBefore)), + workspace: { + deletedAt: IsNull(), + activationStatus: WorkspaceActivationStatus.ACTIVE, + }, + }, + }); + + return [...new Set(staleChannels.map(({ workspaceId }) => workspaceId))]; + } } diff --git a/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/jobs/calendar-ongoing-stale.job.ts b/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/jobs/calendar-ongoing-stale.job.ts index 62b6180ad1..3e41739343 100644 --- a/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/jobs/calendar-ongoing-stale.job.ts +++ b/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/jobs/calendar-ongoing-stale.job.ts @@ -10,6 +10,7 @@ import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queu import { CalendarChannelEntity } from 'src/engine/metadata-modules/calendar-channel/entities/calendar-channel.entity'; 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 { CALENDAR_ONGOING_STALE_SYNC_STAGES } from 'src/modules/calendar/calendar-event-import-manager/constants/calendar-ongoing-stale-sync-stages.constant'; import { isSyncStale } from 'src/modules/calendar/calendar-event-import-manager/utils/is-sync-stale.util'; import { CalendarChannelSyncStatusService } from 'src/modules/calendar/common/services/calendar-channel-sync-status.service'; @@ -40,12 +41,7 @@ export class CalendarOngoingStaleJob { async () => { const calendarChannels = await this.calendarChannelRepository.find({ where: { - syncStage: In([ - CalendarChannelSyncStage.CALENDAR_EVENTS_IMPORT_ONGOING, - CalendarChannelSyncStage.CALENDAR_EVENT_LIST_FETCH_ONGOING, - CalendarChannelSyncStage.CALENDAR_EVENTS_IMPORT_SCHEDULED, - CalendarChannelSyncStage.CALENDAR_EVENT_LIST_FETCH_SCHEDULED, - ]), + syncStage: In(CALENDAR_ONGOING_STALE_SYNC_STAGES), workspaceId, }, }); diff --git a/packages/twenty-server/src/modules/messaging/message-import-manager/constants/messaging-ongoing-stale-sync-stages.constant.ts b/packages/twenty-server/src/modules/messaging/message-import-manager/constants/messaging-ongoing-stale-sync-stages.constant.ts new file mode 100644 index 0000000000..1d578edafe --- /dev/null +++ b/packages/twenty-server/src/modules/messaging/message-import-manager/constants/messaging-ongoing-stale-sync-stages.constant.ts @@ -0,0 +1,8 @@ +import { MessageChannelSyncStage } from 'twenty-shared/types'; + +export const MESSAGING_ONGOING_STALE_SYNC_STAGES: MessageChannelSyncStage[] = [ + MessageChannelSyncStage.MESSAGES_IMPORT_ONGOING, + MessageChannelSyncStage.MESSAGE_LIST_FETCH_ONGOING, + MessageChannelSyncStage.MESSAGES_IMPORT_SCHEDULED, + MessageChannelSyncStage.MESSAGE_LIST_FETCH_SCHEDULED, +]; diff --git a/packages/twenty-server/src/modules/messaging/message-import-manager/crons/jobs/__tests__/messaging-ongoing-stale.cron.job.spec.ts b/packages/twenty-server/src/modules/messaging/message-import-manager/crons/jobs/__tests__/messaging-ongoing-stale.cron.job.spec.ts new file mode 100644 index 0000000000..6f8728d3f1 --- /dev/null +++ b/packages/twenty-server/src/modules/messaging/message-import-manager/crons/jobs/__tests__/messaging-ongoing-stale.cron.job.spec.ts @@ -0,0 +1,61 @@ +import { type Repository } from 'typeorm'; + +import { type ExceptionHandlerService } from 'src/engine/core-modules/exception-handler/exception-handler.service'; +import { type MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service'; +import { MessageChannelEntity } from 'src/engine/metadata-modules/message-channel/entities/message-channel.entity'; +import { MessagingOngoingStaleCronJob } from 'src/modules/messaging/message-import-manager/crons/jobs/messaging-ongoing-stale.cron.job'; +import { MessagingOngoingStaleJob } from 'src/modules/messaging/message-import-manager/jobs/messaging-ongoing-stale.job'; + +describe('MessagingOngoingStaleCronJob', () => { + let messageChannelRepository: { find: jest.Mock }; + let messageQueueService: { add: jest.Mock }; + let job: MessagingOngoingStaleCronJob; + + beforeEach(() => { + messageChannelRepository = { + find: jest + .fn() + .mockResolvedValue([ + { workspaceId: 'workspace-1' }, + { workspaceId: 'workspace-1' }, + { workspaceId: 'workspace-2' }, + ]), + }; + messageQueueService = { add: jest.fn() }; + job = new MessagingOngoingStaleCronJob( + messageChannelRepository as unknown as Repository, + messageQueueService as unknown as MessageQueueService, + { + captureExceptions: jest.fn(), + } as unknown as ExceptionHandlerService, + ); + }); + + it('enqueues recovery once per workspace with a stale channel', async () => { + await job.handle(); + + expect(messageQueueService.add).toHaveBeenCalledTimes(2); + expect(messageQueueService.add).toHaveBeenNthCalledWith( + 1, + MessagingOngoingStaleJob.name, + { + workspaceId: 'workspace-1', + }, + ); + expect(messageQueueService.add).toHaveBeenNthCalledWith( + 2, + MessagingOngoingStaleJob.name, + { + workspaceId: 'workspace-2', + }, + ); + }); + + it('does not enqueue recovery when no workspace has a stale channel', async () => { + messageChannelRepository.find.mockResolvedValue([]); + + await job.handle(); + + expect(messageQueueService.add).not.toHaveBeenCalled(); + }); +}); diff --git a/packages/twenty-server/src/modules/messaging/message-import-manager/crons/jobs/messaging-ongoing-stale.cron.job.ts b/packages/twenty-server/src/modules/messaging/message-import-manager/crons/jobs/messaging-ongoing-stale.cron.job.ts index afad88c4dc..02203147c2 100644 --- a/packages/twenty-server/src/modules/messaging/message-import-manager/crons/jobs/messaging-ongoing-stale.cron.job.ts +++ b/packages/twenty-server/src/modules/messaging/message-import-manager/crons/jobs/messaging-ongoing-stale.cron.job.ts @@ -1,7 +1,7 @@ import { InjectRepository } from '@nestjs/typeorm'; -import { Repository } from 'typeorm'; import { WorkspaceActivationStatus } from 'twenty-shared/workspace'; +import { In, IsNull, LessThan, Or, Repository } from 'typeorm'; import { SentryCronMonitor } from 'src/engine/core-modules/cron/sentry-cron-monitor.decorator'; import { ExceptionHandlerService } from 'src/engine/core-modules/exception-handler/exception-handler.service'; @@ -10,7 +10,9 @@ import { Process } from 'src/engine/core-modules/message-queue/decorators/proces import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.decorator'; 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 { MessageChannelEntity } from 'src/engine/metadata-modules/message-channel/entities/message-channel.entity'; +import { MESSAGING_IMPORT_ONGOING_SYNC_TIMEOUT } from 'src/modules/messaging/message-import-manager/constants/messaging-import-ongoing-sync-timeout.constant'; +import { MESSAGING_ONGOING_STALE_SYNC_STAGES } from 'src/modules/messaging/message-import-manager/constants/messaging-ongoing-stale-sync-stages.constant'; import { MessagingOngoingStaleJob, type MessagingOngoingStaleJobData, @@ -21,8 +23,8 @@ export const MESSAGING_ONGOING_STALE_CRON_PATTERN = '0 * * * *'; @Processor(MessageQueue.cronQueue) export class MessagingOngoingStaleCronJob { constructor( - @InjectRepository(WorkspaceEntity) - private readonly workspaceRepository: Repository, + @InjectRepository(MessageChannelEntity) + private readonly messageChannelRepository: Repository, @InjectMessageQueue(MessageQueue.messagingQueue) private readonly messageQueueService: MessageQueueService, private readonly exceptionHandlerService: ExceptionHandlerService, @@ -34,27 +36,44 @@ export class MessagingOngoingStaleCronJob { MESSAGING_ONGOING_STALE_CRON_PATTERN, ) async handle(): Promise { - const activeWorkspaces = await this.workspaceRepository.find({ - where: { - activationStatus: WorkspaceActivationStatus.ACTIVE, - }, - }); + const staleWorkspaceIds = await this.findStaleWorkspaceIds(); - for (const activeWorkspace of activeWorkspaces) { + for (const workspaceId of staleWorkspaceIds) { try { await this.messageQueueService.add( MessagingOngoingStaleJob.name, { - workspaceId: activeWorkspace.id, + workspaceId, }, ); } catch (error) { this.exceptionHandlerService.captureExceptions([error], { workspace: { - id: activeWorkspace.id, + id: workspaceId, }, }); } } } + + private async findStaleWorkspaceIds(): Promise { + const staleBefore = new Date( + Date.now() - MESSAGING_IMPORT_ONGOING_SYNC_TIMEOUT, + ); + const staleChannels = await this.messageChannelRepository.find({ + select: { + workspaceId: true, + }, + where: { + syncStage: In(MESSAGING_ONGOING_STALE_SYNC_STAGES), + syncStageStartedAt: Or(IsNull(), LessThan(staleBefore)), + workspace: { + deletedAt: IsNull(), + activationStatus: WorkspaceActivationStatus.ACTIVE, + }, + }, + }); + + return [...new Set(staleChannels.map(({ workspaceId }) => workspaceId))]; + } } 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 0a906bfe90..c147c9c720 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 @@ -11,6 +11,7 @@ import { MessageChannelEntity } from 'src/engine/metadata-modules/message-channe 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 { MessageChannelSyncStatusService } from 'src/modules/messaging/common/services/message-channel-sync-status.service'; +import { MESSAGING_ONGOING_STALE_SYNC_STAGES } from 'src/modules/messaging/message-import-manager/constants/messaging-ongoing-stale-sync-stages.constant'; import { isSyncStale } from 'src/modules/messaging/message-import-manager/utils/is-sync-stale.util'; import { toIsoStringOrNull } from 'src/utils/date/toIsoStringOrNull'; @@ -41,12 +42,7 @@ export class MessagingOngoingStaleJob { async () => { const messageChannels = await this.messageChannelRepository.find({ where: { - syncStage: In([ - MessageChannelSyncStage.MESSAGES_IMPORT_ONGOING, - MessageChannelSyncStage.MESSAGE_LIST_FETCH_ONGOING, - MessageChannelSyncStage.MESSAGES_IMPORT_SCHEDULED, - MessageChannelSyncStage.MESSAGE_LIST_FETCH_SCHEDULED, - ]), + syncStage: In(MESSAGING_ONGOING_STALE_SYNC_STAGES), workspaceId, }, });