From 68a9ef57f089d3f0a76d8734de6d29280c2a6986 Mon Sep 17 00:00:00 2001 From: Charles Bochet Date: Thu, 4 Dec 2025 16:01:11 +0100 Subject: [PATCH] Fix Message/Calendar channel stuck in SCHEDULED syncStage (#16326) As per title --- .../calendar-event-list-fetch.cron.job.ts | 5 ++- .../jobs/calendar-events-import.cron.job.ts | 5 ++- .../message-channel-sync-status.service.ts | 41 ++++++++++++++++++- .../messaging-message-list-fetch.cron.job.ts | 5 ++- .../messaging-messages-import.cron.job.ts | 5 ++- ...ssaging-message-list-fetch.service.spec.ts | 8 ++-- .../messaging-messages-import.service.spec.ts | 2 +- 7 files changed, 61 insertions(+), 10 deletions(-) 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 3700e337e5..aef27c0b06 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 @@ -50,8 +50,11 @@ export class CalendarEventListFetchCronJob { try { const schemaName = getWorkspaceSchemaName(activeWorkspace.id); + const now = new Date().toISOString(); + const [calendarChannels] = await this.coreDataSource.query( - `UPDATE ${schemaName}."calendarChannel" SET "syncStage" = '${CalendarChannelSyncStage.CALENDAR_EVENT_LIST_FETCH_SCHEDULED}' WHERE "isSyncEnabled" = true AND "syncStage" = '${CalendarChannelSyncStage.CALENDAR_EVENT_LIST_FETCH_PENDING}' RETURNING *`, + `UPDATE ${schemaName}."calendarChannel" SET "syncStage" = '${CalendarChannelSyncStage.CALENDAR_EVENT_LIST_FETCH_SCHEDULED}', "syncStageStartedAt" = '${now}' + WHERE "isSyncEnabled" = true AND "syncStage" = '${CalendarChannelSyncStage.CALENDAR_EVENT_LIST_FETCH_PENDING}' RETURNING *`, ); for (const calendarChannel of calendarChannels) { 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 e75e462985..c5e3318585 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 @@ -48,8 +48,11 @@ export class CalendarEventsImportCronJob { try { const schemaName = getWorkspaceSchemaName(activeWorkspace.id); + const now = new Date().toISOString(); + const [calendarChannels] = await this.coreDataSource.query( - `UPDATE ${schemaName}."calendarChannel" SET "syncStage" = '${CalendarChannelSyncStage.CALENDAR_EVENTS_IMPORT_SCHEDULED}' WHERE "isSyncEnabled" = true AND "syncStage" = '${CalendarChannelSyncStage.CALENDAR_EVENTS_IMPORT_PENDING}' RETURNING *`, + `UPDATE ${schemaName}."calendarChannel" SET "syncStage" = '${CalendarChannelSyncStage.CALENDAR_EVENTS_IMPORT_SCHEDULED}', "syncStageStartedAt" = '${now}' + WHERE "isSyncEnabled" = true AND "syncStage" = '${CalendarChannelSyncStage.CALENDAR_EVENTS_IMPORT_PENDING}' RETURNING *`, ); for (const calendarChannel of calendarChannels) { diff --git a/packages/twenty-server/src/modules/messaging/common/services/message-channel-sync-status.service.ts b/packages/twenty-server/src/modules/messaging/common/services/message-channel-sync-status.service.ts index 9484e8ddc0..fbfa95484d 100644 --- a/packages/twenty-server/src/modules/messaging/common/services/message-channel-sync-status.service.ts +++ b/packages/twenty-server/src/modules/messaging/common/services/message-channel-sync-status.service.ts @@ -133,6 +133,27 @@ export class MessageChannelSyncStatusService { }); } + public async markAsMessagesListFetchScheduled( + messageChannelIds: string[], + workspaceId: string, + ) { + if (!messageChannelIds.length) { + return; + } + + const messageChannelRepository = + await this.twentyORMGlobalManager.getRepositoryForWorkspace( + workspaceId, + 'messageChannel', + ); + + await messageChannelRepository.update(messageChannelIds, { + syncStage: MessageChannelSyncStage.MESSAGE_LIST_FETCH_SCHEDULED, + syncStatus: MessageChannelSyncStatus.ONGOING, + syncStageStartedAt: new Date().toISOString(), + }); + } + public async markAsMessagesListFetchOngoing( messageChannelIds: string[], workspaceId: string, @@ -150,7 +171,6 @@ export class MessageChannelSyncStatusService { await messageChannelRepository.update(messageChannelIds, { syncStage: MessageChannelSyncStage.MESSAGE_LIST_FETCH_ONGOING, syncStatus: MessageChannelSyncStatus.ONGOING, - syncStageStartedAt: new Date().toISOString(), }); } @@ -182,6 +202,25 @@ export class MessageChannelSyncStatusService { }); } + public async markAsMessagesImportScheduled( + messageChannelIds: string[], + workspaceId: string, + ) { + if (!messageChannelIds.length) { + return; + } + + const messageChannelRepository = + await this.twentyORMGlobalManager.getRepositoryForWorkspace( + workspaceId, + 'messageChannel', + ); + + await messageChannelRepository.update(messageChannelIds, { + syncStage: MessageChannelSyncStage.MESSAGES_IMPORT_SCHEDULED, + }); + } + public async markAsMessagesImportOngoing( messageChannelIds: string[], workspaceId: string, 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 e51101b553..7e9508919e 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 @@ -48,8 +48,11 @@ export class MessagingMessageListFetchCronJob { try { const schemaName = getWorkspaceSchemaName(activeWorkspace.id); + const now = new Date().toISOString(); + const [messageChannels] = await this.coreDataSource.query( - `UPDATE ${schemaName}."messageChannel" SET "syncStage" = '${MessageChannelSyncStage.MESSAGE_LIST_FETCH_SCHEDULED}' WHERE "isSyncEnabled" = true AND "syncStage" = '${MessageChannelSyncStage.MESSAGE_LIST_FETCH_PENDING}' RETURNING *`, + `UPDATE ${schemaName}."messageChannel" SET "syncStage" = '${MessageChannelSyncStage.MESSAGE_LIST_FETCH_SCHEDULED}', "syncStageStartedAt" = '${now}' + WHERE "isSyncEnabled" = true AND "syncStage" = '${MessageChannelSyncStage.MESSAGE_LIST_FETCH_PENDING}' RETURNING *`, ); for (const messageChannel of messageChannels) { 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 091197e4ff..9397b37e47 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 @@ -53,8 +53,11 @@ export class MessagingMessagesImportCronJob { try { const schemaName = getWorkspaceSchemaName(activeWorkspace.id); + const now = new Date().toISOString(); + const [messageChannels] = await this.coreDataSource.query( - `UPDATE ${schemaName}."messageChannel" SET "syncStage" = '${MessageChannelSyncStage.MESSAGES_IMPORT_SCHEDULED}' WHERE "isSyncEnabled" = true AND "syncStage" = '${MessageChannelSyncStage.MESSAGES_IMPORT_PENDING}' RETURNING *`, + `UPDATE ${schemaName}."messageChannel" SET "syncStage" = '${MessageChannelSyncStage.MESSAGES_IMPORT_SCHEDULED}', "syncStageStartedAt" = '${now}' + WHERE "isSyncEnabled" = true AND "syncStage" = '${MessageChannelSyncStage.MESSAGES_IMPORT_PENDING}' RETURNING *`, ); for (const messageChannel of messageChannels) { diff --git a/packages/twenty-server/src/modules/messaging/message-import-manager/services/__tests__/messaging-message-list-fetch.service.spec.ts b/packages/twenty-server/src/modules/messaging/message-import-manager/services/__tests__/messaging-message-list-fetch.service.spec.ts index 1f24fbd578..ec045ba16c 100644 --- a/packages/twenty-server/src/modules/messaging/message-import-manager/services/__tests__/messaging-message-list-fetch.service.spec.ts +++ b/packages/twenty-server/src/modules/messaging/message-import-manager/services/__tests__/messaging-message-list-fetch.service.spec.ts @@ -291,7 +291,7 @@ describe('MessagingMessageListFetchService', () => { }); expect( messageChannelSyncStatusService.markAsMessagesListFetchOngoing, - ).toHaveBeenCalledWith([mockMicrosoftMessageChannel.id]); + ).toHaveBeenCalledWith([mockMicrosoftMessageChannel.id], workspaceId); expect(messagingGetMessageListService.getMessageLists).toHaveBeenCalledWith( { @@ -332,7 +332,7 @@ describe('MessagingMessageListFetchService', () => { expect( messageChannelSyncStatusService.scheduleMessagesImport, - ).toHaveBeenCalledWith([mockMicrosoftMessageChannel.id]); + ).toHaveBeenCalledWith([mockMicrosoftMessageChannel.id], workspaceId); }); it('should process Google message list fetch correctly', async () => { @@ -350,7 +350,7 @@ describe('MessagingMessageListFetchService', () => { }); expect( messageChannelSyncStatusService.markAsMessagesListFetchOngoing, - ).toHaveBeenCalledWith([mockGoogleMessageChannel.id]); + ).toHaveBeenCalledWith([mockGoogleMessageChannel.id], workspaceId); expect(messagingGetMessageListService.getMessageLists).toHaveBeenCalledWith( { @@ -391,6 +391,6 @@ describe('MessagingMessageListFetchService', () => { expect( messageChannelSyncStatusService.scheduleMessagesImport, - ).toHaveBeenCalledWith([mockGoogleMessageChannel.id]); + ).toHaveBeenCalledWith([mockGoogleMessageChannel.id], workspaceId); }); }); diff --git a/packages/twenty-server/src/modules/messaging/message-import-manager/services/__tests__/messaging-messages-import.service.spec.ts b/packages/twenty-server/src/modules/messaging/message-import-manager/services/__tests__/messaging-messages-import.service.spec.ts index f11d03923f..f4e7cba4f7 100644 --- a/packages/twenty-server/src/modules/messaging/message-import-manager/services/__tests__/messaging-messages-import.service.spec.ts +++ b/packages/twenty-server/src/modules/messaging/message-import-manager/services/__tests__/messaging-messages-import.service.spec.ts @@ -213,7 +213,7 @@ describe('MessagingMessagesImportService', () => { ); expect( messageChannelSyncStatusService.markAsMessagesImportOngoing, - ).toHaveBeenCalledWith([mockMessageChannel.id]); + ).toHaveBeenCalledWith([mockMessageChannel.id], workspaceId); expect( connectedAccountRefreshTokensService.refreshAndSaveTokens,