Fix messaging direct import (#16354)

Mainly clarifying naming (scheduleMessageImport was actually tagging as
PENDING)
+ tagging as SCHEDULED in case of direct import (we skip PENDING state)

packages/twenty-server/src/engine/core-modules/auth/services/google-apis.service.spec.ts

@@ -111,7 +111,7 @@ describe('GoogleAPIsService', () => {
         {
           provide: CalendarChannelSyncStatusService,
           useValue: {
-            resetAndScheduleCalendarEventListFetch: jest.fn(),
+            resetAndMarkAsCalendarEventListFetchPending: jest.fn(),
           },
         },
         {
@@ -127,7 +127,7 @@ describe('GoogleAPIsService', () => {
         {
           provide: MessageChannelSyncStatusService,
           useValue: {
-            resetAndScheduleMessageListFetch: jest.fn(),
+            resetAndMarkAsMessagesListFetchPending: jest.fn(),
           },
         },
         {
@@ -234,11 +234,11 @@ describe('GoogleAPIsService', () => {
       });
 
       expect(
-        calendarChannelSyncStatusService.resetAndScheduleCalendarEventListFetch,
+        calendarChannelSyncStatusService.resetAndMarkAsCalendarEventListFetchPending,
       ).toHaveBeenCalledWith([existingConnectedAccount.id], 'workspace-id');
 
       expect(
-        messagingChannelSyncStatusService.resetAndScheduleMessageListFetch,
+        messagingChannelSyncStatusService.resetAndMarkAsMessagesListFetchPending,
       ).toHaveBeenCalledWith([existingConnectedAccount.id], 'workspace-id');
 
       expect(

packages/twenty-server/src/engine/core-modules/auth/services/google-apis.service.ts

@@ -190,12 +190,12 @@ export class GoogleAPIsService {
             newOrExistingConnectedAccountId,
           );
 
-          await this.messagingChannelSyncStatusService.resetAndScheduleMessageListFetch(
+          await this.messagingChannelSyncStatusService.resetAndMarkAsMessagesListFetchPending(
             [newOrExistingConnectedAccountId],
             workspaceId,
           );
 
-          await this.calendarChannelSyncStatusService.resetAndScheduleCalendarEventListFetch(
+          await this.calendarChannelSyncStatusService.resetAndMarkAsCalendarEventListFetchPending(
             [newOrExistingConnectedAccountId],
             workspaceId,
           );

packages/twenty-server/src/engine/core-modules/auth/services/microsoft-apis.service.spec.ts

@@ -110,13 +110,13 @@ describe('MicrosoftAPIsService', () => {
         {
           provide: CalendarChannelSyncStatusService,
           useValue: {
-            resetAndScheduleCalendarEventListFetch: jest.fn(),
+            resetAndMarkAsCalendarEventListFetchPending: jest.fn(),
           },
         },
         {
           provide: MessageChannelSyncStatusService,
           useValue: {
-            resetAndScheduleMessageListFetch: jest.fn(),
+            resetAndMarkAsMessagesListFetchPending: jest.fn(),
           },
         },
         {
@@ -230,11 +230,11 @@ describe('MicrosoftAPIsService', () => {
       });
 
       expect(
-        calendarChannelSyncStatusService.resetAndScheduleCalendarEventListFetch,
+        calendarChannelSyncStatusService.resetAndMarkAsCalendarEventListFetchPending,
       ).toHaveBeenCalledWith([existingConnectedAccount.id], 'workspace-id');
 
       expect(
-        messagingChannelSyncStatusService.resetAndScheduleMessageListFetch,
+        messagingChannelSyncStatusService.resetAndMarkAsMessagesListFetchPending,
       ).toHaveBeenCalledWith([existingConnectedAccount.id], 'workspace-id');
 
       expect(

packages/twenty-server/src/engine/core-modules/auth/services/microsoft-apis.service.ts

@@ -173,17 +173,17 @@ export class MicrosoftAPIsService {
             newOrExistingConnectedAccountId,
           );
 
-          await this.messagingChannelSyncStatusService.resetAndScheduleMessageListFetch(
+          await this.messagingChannelSyncStatusService.resetAndMarkAsMessagesListFetchPending(
             [newOrExistingConnectedAccountId],
             workspaceId,
           );
 
-          await this.calendarChannelSyncStatusService.resetAndScheduleCalendarEventListFetch(
+          await this.calendarChannelSyncStatusService.resetAndMarkAsCalendarEventListFetchPending(
             [newOrExistingConnectedAccountId],
             workspaceId,
           );
 
-          await this.calendarChannelSyncStatusService.resetAndScheduleCalendarEventListFetch(
+          await this.calendarChannelSyncStatusService.resetAndMarkAsCalendarEventListFetchPending(
             [newOrExistingConnectedAccountId],
             workspaceId,
           );

packages/twenty-server/src/modules/calendar/blocklist-manager/jobs/blocklist-reimport-calendar-events.job.ts

@@ -54,7 +54,7 @@ export class BlocklistReimportCalendarEventsJob {
         },
       });
 
-      await this.calendarChannelSyncStatusService.resetAndScheduleCalendarEventListFetch(
+      await this.calendarChannelSyncStatusService.resetAndMarkAsCalendarEventListFetchPending(
         calendarChannels.map((calendarChannel) => calendarChannel.id),
         workspaceId,
       );

packages/twenty-server/src/modules/calendar/calendar-event-import-manager/jobs/calendar-event-list-fetch.job.ts

@@ -62,7 +62,7 @@ export class CalendarEventListFetchJob {
         calendarChannel.throttleFailureCount,
       )
     ) {
-      await this.calendarChannelSyncStatusService.scheduleCalendarEventListFetch(
+      await this.calendarChannelSyncStatusService.markAsCalendarEventListFetchPending(
         [calendarChannel.id],
         workspaceId,
         true,

packages/twenty-server/src/modules/calendar/calendar-event-import-manager/jobs/calendar-events-import.job.ts

@@ -61,7 +61,7 @@ export class CalendarEventsImportJob {
         calendarChannel.throttleFailureCount,
       )
     ) {
-      await this.calendarChannelSyncStatusService.scheduleCalendarEventsImport(
+      await this.calendarChannelSyncStatusService.markAsCalendarEventsImportPending(
         [calendarChannel.id],
         workspaceId,
         true,

packages/twenty-server/src/modules/calendar/calendar-event-import-manager/jobs/calendar-ongoing-stale.job.ts

@@ -64,7 +64,7 @@ export class CalendarOngoingStaleJob {
             this.logger.log(
               `Sync for calendar channel ${calendarChannel.id} and workspace ${workspaceId} is stale. Setting sync stage to CALENDAR_EVENT_LIST_FETCH_PENDING`,
             );
-            await this.calendarChannelSyncStatusService.scheduleCalendarEventListFetch(
+            await this.calendarChannelSyncStatusService.markAsCalendarEventListFetchPending(
               [calendarChannel.id],
               workspaceId,
             );
@@ -74,7 +74,7 @@ export class CalendarOngoingStaleJob {
             this.logger.log(
               `Sync for calendar channel ${calendarChannel.id} and workspace ${workspaceId} is stale. Setting sync stage to CALENDAR_EVENTS_IMPORT_PENDING`,
             );
-            await this.calendarChannelSyncStatusService.scheduleCalendarEventsImport(
+            await this.calendarChannelSyncStatusService.markAsCalendarEventsImportPending(
               [calendarChannel.id],
               workspaceId,
             );

packages/twenty-server/src/modules/calendar/calendar-event-import-manager/services/calendar-event-import-exception-handler.service.ts

@@ -88,7 +88,7 @@ export class CalendarEventImportErrorHandlerService {
       `CalendarChannelId: ${calendarChannel.id} - Sync cursor error, resetting and rescheduling`,
     );
 
-    await this.calendarChannelSyncStatusService.resetAndScheduleCalendarEventListFetch(
+    await this.calendarChannelSyncStatusService.resetAndMarkAsCalendarEventListFetchPending(
       [calendarChannel.id],
       workspaceId,
     );
@@ -147,7 +147,7 @@ export class CalendarEventImportErrorHandlerService {
 
     switch (syncStep) {
       case CalendarEventImportSyncStep.CALENDAR_EVENT_LIST_FETCH:
-        await this.calendarChannelSyncStatusService.scheduleCalendarEventListFetch(
+        await this.calendarChannelSyncStatusService.markAsCalendarEventListFetchPending(
           [calendarChannel.id],
           workspaceId,
           true,
@@ -155,7 +155,7 @@ export class CalendarEventImportErrorHandlerService {
         break;
 
       case CalendarEventImportSyncStep.CALENDAR_EVENTS_IMPORT:
-        await this.calendarChannelSyncStatusService.scheduleCalendarEventsImport(
+        await this.calendarChannelSyncStatusService.markAsCalendarEventsImportPending(
           [calendarChannel.id],
           workspaceId,
           true,
@@ -218,7 +218,7 @@ export class CalendarEventImportErrorHandlerService {
       return;
     }
 
-    await this.calendarChannelSyncStatusService.resetAndScheduleCalendarEventListFetch(
+    await this.calendarChannelSyncStatusService.resetAndMarkAsCalendarEventListFetchPending(
       [calendarChannel.id],
       workspaceId,
     );

packages/twenty-server/src/modules/calendar/calendar-event-import-manager/services/calendar-events-import.service.ts

@@ -67,7 +67,7 @@ export class CalendarEventsImportService {
         );
 
         if (!eventIdsToFetch || eventIdsToFetch.length === 0) {
-          await this.calendarChannelSyncStatusService.markAsCompletedAndScheduleCalendarEventListFetch(
+          await this.calendarChannelSyncStatusService.markAsCompletedAndMarkAsCalendarEventListFetchPending(
             [calendarChannel.id],
             workspaceId,
           );
@@ -89,7 +89,7 @@ export class CalendarEventsImportService {
       }
 
       if (!calendarEvents || calendarEvents?.length === 0) {
-        await this.calendarChannelSyncStatusService.scheduleCalendarEventListFetch(
+        await this.calendarChannelSyncStatusService.markAsCalendarEventListFetchPending(
           [calendarChannel.id],
           workspaceId,
         );
@@ -152,7 +152,7 @@ export class CalendarEventsImportService {
         workspaceId,
       );
 
-      await this.calendarChannelSyncStatusService.markAsCompletedAndScheduleCalendarEventListFetch(
+      await this.calendarChannelSyncStatusService.markAsCompletedAndMarkAsCalendarEventListFetchPending(
         [calendarChannel.id],
         workspaceId,
       );

packages/twenty-server/src/modules/calendar/calendar-event-import-manager/services/calendar-fetch-events.service.ts

@@ -97,7 +97,7 @@ export class CalendarFetchEventsService {
           },
         );
 
-        await this.calendarChannelSyncStatusService.scheduleCalendarEventListFetch(
+        await this.calendarChannelSyncStatusService.markAsCalendarEventListFetchPending(
           [calendarChannel.id],
           workspaceId,
         );
@@ -128,7 +128,7 @@ export class CalendarFetchEventsService {
           calendarEventIds,
         );
 
-        await this.calendarChannelSyncStatusService.scheduleCalendarEventsImport(
+        await this.calendarChannelSyncStatusService.markAsCalendarEventsImportPending(
           [calendarChannel.id],
           workspaceId,
         );

packages/twenty-server/src/modules/calendar/common/services/calendar-channel-sync-status.service.ts

@@ -27,7 +27,7 @@ export class CalendarChannelSyncStatusService {
     private readonly metricsService: MetricsService,
   ) {}
 
-  public async scheduleCalendarEventListFetch(
+  public async markAsCalendarEventListFetchPending(
     calendarChannelIds: string[],
     workspaceId: string,
     preserveSyncStageStartedAt: boolean = false,
@@ -69,7 +69,7 @@ export class CalendarChannelSyncStatusService {
     });
   }
 
-  public async resetAndScheduleCalendarEventListFetch(
+  public async resetAndMarkAsCalendarEventListFetchPending(
     calendarChannelIds: string[],
     workspaceId: string,
   ) {
@@ -95,7 +95,10 @@ export class CalendarChannelSyncStatusService {
       throttleFailureCount: 0,
     });
 
-    await this.scheduleCalendarEventListFetch(calendarChannelIds, workspaceId);
+    await this.markAsCalendarEventListFetchPending(
+      calendarChannelIds,
+      workspaceId,
+    );
   }
 
   public async resetSyncStageStartedAt(
@@ -117,7 +120,7 @@ export class CalendarChannelSyncStatusService {
     });
   }
 
-  public async scheduleCalendarEventsImport(
+  public async markAsCalendarEventsImportPending(
     calendarChannelIds: string[],
     workspaceId: string,
     preserveSyncStageStartedAt: boolean = false,
@@ -158,7 +161,7 @@ export class CalendarChannelSyncStatusService {
     });
   }
 
-  public async markAsCompletedAndScheduleCalendarEventListFetch(
+  public async markAsCompletedAndMarkAsCalendarEventListFetchPending(
     calendarChannelIds: string[],
     workspaceId: string,
   ) {
@@ -180,7 +183,10 @@ export class CalendarChannelSyncStatusService {
       syncedAt: new Date().toISOString(),
     });
 
-    await this.scheduleCalendarEventListFetch(calendarChannelIds, workspaceId);
+    await this.markAsCalendarEventListFetchPending(
+      calendarChannelIds,
+      workspaceId,
+    );
 
     await this.metricsService.batchIncrementCounter({
       key: MetricsKeys.CalendarEventSyncJobActive,

packages/twenty-server/src/modules/messaging/blocklist-manager/jobs/messaging-blocklist-reimport-messages.job.ts

@@ -52,7 +52,7 @@ export class BlocklistReimportMessagesJob {
         },
       });
 
-      await this.messagingChannelSyncStatusService.resetAndScheduleMessageListFetch(
+      await this.messagingChannelSyncStatusService.resetAndMarkAsMessagesListFetchPending(
         messageChannels.map((messageChannel) => messageChannel.id),
         workspaceId,
       );

packages/twenty-server/src/modules/messaging/common/services/message-channel-sync-status.service.ts

@@ -32,7 +32,7 @@ export class MessageChannelSyncStatusService {
     private readonly metricsService: MetricsService,
   ) {}
 
-  public async scheduleMessageListFetch(
+  public async markAsMessagesListFetchPending(
     messageChannelIds: string[],
     workspaceId: string,
     preserveSyncStageStartedAt: boolean = false,
@@ -53,7 +53,7 @@ export class MessageChannelSyncStatusService {
     });
   }
 
-  public async scheduleMessagesImport(
+  public async markAsMessagesImportPending(
     messageChannelIds: string[],
     workspaceId: string,
     preserveSyncStageStartedAt: boolean = false,
@@ -74,7 +74,7 @@ export class MessageChannelSyncStatusService {
     });
   }
 
-  public async resetAndScheduleMessageListFetch(
+  public async resetAndMarkAsMessagesListFetchPending(
     messageChannelIds: string[],
     workspaceId: string,
   ) {
@@ -115,7 +115,7 @@ export class MessageChannelSyncStatusService {
       },
     );
 
-    await this.scheduleMessageListFetch(messageChannelIds, workspaceId);
+    await this.markAsMessagesListFetchPending(messageChannelIds, workspaceId);
   }
 
   public async resetSyncStageStartedAt(
@@ -178,7 +178,7 @@ export class MessageChannelSyncStatusService {
     });
   }
 
-  public async markAsCompletedAndScheduleMessageListFetch(
+  public async markAsCompletedAndMarkAsMessagesListFetchPending(
     messageChannelIds: string[],
     workspaceId: string,
   ) {

packages/twenty-server/src/modules/messaging/message-cleaner/commands/messaging-reset-channel.command.ts

@@ -62,7 +62,7 @@ export class MessagingResetChannelCommand extends CommandRunner {
     );
 
     for (const messageChannel of messageChannels) {
-      await this.messagingChannelSyncStatusService.resetAndScheduleMessageListFetch(
+      await this.messagingChannelSyncStatusService.resetAndMarkAsMessagesListFetchPending(
         [messageChannel.id],
         workspaceId,
       );

packages/twenty-server/src/modules/messaging/message-import-manager/jobs/messaging-message-list-fetch.job.ts

@@ -81,7 +81,7 @@ export class MessagingMessageListFetchJob {
           messageChannel.throttleFailureCount,
         )
       ) {
-        await this.messageChannelSyncStatusService.scheduleMessageListFetch(
+        await this.messageChannelSyncStatusService.markAsMessagesListFetchPending(
           [messageChannel.id],
           workspaceId,
           true,

packages/twenty-server/src/modules/messaging/message-import-manager/jobs/messaging-messages-import.job.ts

@@ -78,7 +78,7 @@ export class MessagingMessagesImportJob {
         messageChannel.throttleFailureCount,
       )
     ) {
-      await this.messageChannelSyncStatusService.scheduleMessagesImport(
+      await this.messageChannelSyncStatusService.markAsMessagesImportPending(
         [messageChannel.id],
         workspaceId,
         true,

packages/twenty-server/src/modules/messaging/message-import-manager/jobs/messaging-ongoing-stale.job.ts

@@ -64,7 +64,7 @@ export class MessagingOngoingStaleJob {
             this.logger.log(
               `Sync for message channel ${messageChannel.id} and workspace ${workspaceId} is stale. Setting sync stage to MESSAGE_LIST_FETCH_PENDING`,
             );
-            await this.messageChannelSyncStatusService.scheduleMessageListFetch(
+            await this.messageChannelSyncStatusService.markAsMessagesListFetchPending(
               [messageChannel.id],
               workspaceId,
             );
@@ -74,7 +74,7 @@ export class MessagingOngoingStaleJob {
             this.logger.log(
               `Sync for message channel ${messageChannel.id} and workspace ${workspaceId} is stale. Setting sync stage to MESSAGES_IMPORT_PENDING`,
             );
-            await this.messageChannelSyncStatusService.scheduleMessagesImport(
+            await this.messageChannelSyncStatusService.markAsMessagesImportPending(
               [messageChannel.id],
               workspaceId,
             );

packages/twenty-server/src/modules/messaging/message-import-manager/services/__tests__/messaging-message-list-fetch.service.spec.ts

@@ -189,7 +189,10 @@ describe('MessagingMessageListFetchService', () => {
             markAsMessagesListFetchOngoing: jest
               .fn()
               .mockResolvedValue(undefined),
-            scheduleMessagesImport: jest.fn().mockResolvedValue(undefined),
+            markAsMessagesImportPending: jest.fn().mockResolvedValue(undefined),
+            markAsMessagesImportScheduled: jest
+              .fn()
+              .mockResolvedValue(undefined),
           },
         },
         {
@@ -331,7 +334,7 @@ describe('MessagingMessageListFetchService', () => {
     );
 
     expect(
-      messageChannelSyncStatusService.scheduleMessagesImport,
+      messageChannelSyncStatusService.markAsMessagesImportScheduled,
     ).toHaveBeenCalledWith([mockMicrosoftMessageChannel.id], workspaceId);
   });
 
@@ -390,7 +393,7 @@ describe('MessagingMessageListFetchService', () => {
     );
 
     expect(
-      messageChannelSyncStatusService.scheduleMessagesImport,
+      messageChannelSyncStatusService.markAsMessagesImportScheduled,
     ).toHaveBeenCalledWith([mockGoogleMessageChannel.id], workspaceId);
   });
 });

packages/twenty-server/src/modules/messaging/message-import-manager/services/__tests__/messaging-messages-import.service.spec.ts

@@ -66,10 +66,10 @@ describe('MessagingMessagesImportService', () => {
         provide: MessageChannelSyncStatusService,
         useValue: {
           markAsMessagesImportOngoing: jest.fn().mockResolvedValue(undefined),
-          markAsCompletedAndScheduleMessageListFetch: jest
+          markAsCompletedAndMarkAsMessagesListFetchPending: jest
             .fn()
             .mockResolvedValue(undefined),
-          scheduleMessagesImport: jest.fn().mockResolvedValue(undefined),
+          markAsMessagesImportPending: jest.fn().mockResolvedValue(undefined),
         },
       },
       {
@@ -236,7 +236,7 @@ describe('MessagingMessagesImportService', () => {
       saveMessagesService.saveMessagesAndEnqueueContactCreation,
     ).toHaveBeenCalled();
     expect(
-      messageChannelSyncStatusService.scheduleMessagesImport,
+      messageChannelSyncStatusService.markAsMessagesImportPending,
     ).toHaveBeenCalledTimes(0);
   });
 
@@ -293,7 +293,7 @@ describe('MessagingMessagesImportService', () => {
     );
 
     expect(
-      messageChannelSyncStatusService.scheduleMessagesImport,
+      messageChannelSyncStatusService.markAsMessagesImportPending,
     ).toHaveBeenCalledTimes(1);
   });
 });

packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-import-exception-handler.service.ts

@@ -163,7 +163,7 @@ export class MessageImportExceptionHandlerService {
 
     switch (syncStep) {
       case MessageImportSyncStep.MESSAGE_LIST_FETCH:
-        await this.messageChannelSyncStatusService.scheduleMessageListFetch(
+        await this.messageChannelSyncStatusService.markAsMessagesListFetchPending(
           [messageChannel.id],
           workspaceId,
           true,
@@ -172,7 +172,7 @@ export class MessageImportExceptionHandlerService {
 
       case MessageImportSyncStep.MESSAGES_IMPORT_PENDING:
       case MessageImportSyncStep.MESSAGES_IMPORT_ONGOING:
-        await this.messageChannelSyncStatusService.scheduleMessagesImport(
+        await this.messageChannelSyncStatusService.markAsMessagesImportPending(
           [messageChannel.id],
           workspaceId,
           true,
@@ -250,7 +250,7 @@ export class MessageImportExceptionHandlerService {
       return;
     }
 
-    await this.messageChannelSyncStatusService.resetAndScheduleMessageListFetch(
+    await this.messageChannelSyncStatusService.resetAndMarkAsMessagesListFetchPending(
       [messageChannel.id],
       workspaceId,
     );

packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-message-list-fetch.service.ts

@@ -269,7 +269,7 @@ export class MessagingMessageListFetchService {
       );
 
       if (totalMessagesToImportCount === 0) {
-        await this.messageChannelSyncStatusService.markAsCompletedAndScheduleMessageListFetch(
+        await this.messageChannelSyncStatusService.markAsCompletedAndMarkAsMessagesListFetchPending(
           [messageChannelWithFreshTokens.id],
           workspaceId,
         );
@@ -281,7 +281,7 @@ export class MessagingMessageListFetchService {
         `messageChannelId: ${freshMessageChannel.id} Scheduling direct messages import`,
       );
 
-      await this.messageChannelSyncStatusService.scheduleMessagesImport(
+      await this.messageChannelSyncStatusService.markAsMessagesImportScheduled(
         [messageChannelWithFreshTokens.id],
         workspaceId,
       );
@@ -289,7 +289,7 @@ export class MessagingMessageListFetchService {
       await this.messagingMessagesImportService.processMessageBatchImport(
         {
           ...messageChannelWithFreshTokens,
-          syncStage: MessageChannelSyncStage.MESSAGES_IMPORT_PENDING,
+          syncStage: MessageChannelSyncStage.MESSAGES_IMPORT_SCHEDULED,
         },
         messageChannelWithFreshTokens.connectedAccount,
         workspaceId,

packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-messages-import.service.ts

@@ -101,7 +101,7 @@ export class MessagingMessagesImportService {
       );
 
       if (!messageIdsToFetch?.length) {
-        await this.messageChannelSyncStatusService.markAsCompletedAndScheduleMessageListFetch(
+        await this.messageChannelSyncStatusService.markAsCompletedAndMarkAsMessagesListFetchPending(
           [messageChannel.id],
           workspaceId,
         );
@@ -158,12 +158,12 @@ export class MessagingMessagesImportService {
       if (
         messageIdsToFetch.length < MESSAGING_GMAIL_USERS_MESSAGES_GET_BATCH_SIZE
       ) {
-        await this.messageChannelSyncStatusService.markAsCompletedAndScheduleMessageListFetch(
+        await this.messageChannelSyncStatusService.markAsCompletedAndMarkAsMessagesListFetchPending(
           [messageChannel.id],
           workspaceId,
         );
       } else {
-        await this.messageChannelSyncStatusService.scheduleMessagesImport(
+        await this.messageChannelSyncStatusService.markAsMessagesImportPending(
           [messageChannel.id],
           workspaceId,
         );