fix folder actions (#16344)

Co-authored-by: Charles Bochet <charles@twenty.com>

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/constants/calendar-import-ongoing-sync-timeout.constant.ts

@@ -1 +1 @@
-export const CALENDAR_IMPORT_ONGOING_SYNC_TIMEOUT = 1000 * 60 * 60; // 1 hour
+export const CALENDAR_IMPORT_ONGOING_SYNC_TIMEOUT = 1000 * 60 * 30; // 30 minutes

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

@@ -53,7 +53,7 @@ export class CalendarEventListFetchCronJob {
         const now = new Date().toISOString();
 
         const [calendarChannels] = await this.coreDataSource.query(
-          `UPDATE ${schemaName}."calendarChannel" SET "syncStage" = '${CalendarChannelSyncStage.CALENDAR_EVENT_LIST_FETCH_SCHEDULED}', "syncStageStartedAt" = '${now}'
+          `UPDATE ${schemaName}."calendarChannel" SET "syncStage" = '${CalendarChannelSyncStage.CALENDAR_EVENT_LIST_FETCH_SCHEDULED}', "syncStageStartedAt" = COALESCE("syncStageStartedAt", '${now}')
           WHERE "isSyncEnabled" = true AND "syncStage" = '${CalendarChannelSyncStage.CALENDAR_EVENT_LIST_FETCH_PENDING}' RETURNING *`,
         );
 

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

@@ -51,7 +51,7 @@ export class CalendarEventsImportCronJob {
         const now = new Date().toISOString();
 
         const [calendarChannels] = await this.coreDataSource.query(
-          `UPDATE ${schemaName}."calendarChannel" SET "syncStage" = '${CalendarChannelSyncStage.CALENDAR_EVENTS_IMPORT_SCHEDULED}', "syncStageStartedAt" = '${now}'
+          `UPDATE ${schemaName}."calendarChannel" SET "syncStage" = '${CalendarChannelSyncStage.CALENDAR_EVENTS_IMPORT_SCHEDULED}', "syncStageStartedAt" = COALESCE("syncStageStartedAt", '${now}')
            WHERE "isSyncEnabled" = true AND "syncStage" = '${CalendarChannelSyncStage.CALENDAR_EVENTS_IMPORT_PENDING}' RETURNING *`,
         );
 

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

@@ -5,6 +5,7 @@ import { Processor } from 'src/engine/core-modules/message-queue/decorators/proc
 import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
 import { TwentyORMManager } from 'src/engine/twenty-orm/twenty-orm.manager';
 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 {
   CalendarChannelSyncStage,
   type CalendarChannelWorkspaceEntity,
@@ -23,6 +24,7 @@ export type CalendarEventListFetchJobData = {
 export class CalendarEventListFetchJob {
   constructor(
     private readonly twentyORMManager: TwentyORMManager,
+    private readonly calendarChannelSyncStatusService: CalendarChannelSyncStatusService,
     private readonly calendarFetchEventsService: CalendarFetchEventsService,
   ) {}
 
@@ -47,26 +49,32 @@ export class CalendarEventListFetchJob {
       return;
     }
 
+    if (
+      calendarChannel.syncStage !==
+      CalendarChannelSyncStage.CALENDAR_EVENT_LIST_FETCH_SCHEDULED
+    ) {
+      return;
+    }
+
     if (
       isThrottled(
         calendarChannel.syncStageStartedAt,
         calendarChannel.throttleFailureCount,
       )
     ) {
+      await this.calendarChannelSyncStatusService.markAsCalendarEventListFetchPending(
+        [calendarChannel.id],
+        workspaceId,
+        true,
+      );
+
       return;
     }
 
-    switch (calendarChannel.syncStage) {
-      case CalendarChannelSyncStage.CALENDAR_EVENT_LIST_FETCH_SCHEDULED:
-        await this.calendarFetchEventsService.fetchCalendarEvents(
-          calendarChannel,
-          calendarChannel.connectedAccount,
-          workspaceId,
-        );
-        break;
-
-      default:
-        break;
-    }
+    await this.calendarFetchEventsService.fetchCalendarEvents(
+      calendarChannel,
+      calendarChannel.connectedAccount,
+      workspaceId,
+    );
   }
 }

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

@@ -5,6 +5,7 @@ import { Processor } from 'src/engine/core-modules/message-queue/decorators/proc
 import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
 import { TwentyORMManager } from 'src/engine/twenty-orm/twenty-orm.manager';
 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 {
   CalendarChannelSyncStage,
   type CalendarChannelWorkspaceEntity,
@@ -23,6 +24,7 @@ export type CalendarEventsImportJobData = {
 export class CalendarEventsImportJob {
   constructor(
     private readonly calendarEventsImportService: CalendarEventsImportService,
+    private readonly calendarChannelSyncStatusService: CalendarChannelSyncStatusService,
     private readonly twentyORMManager: TwentyORMManager,
   ) {}
 
@@ -47,18 +49,24 @@ export class CalendarEventsImportJob {
     }
 
     if (
-      isThrottled(
-        calendarChannel.syncStageStartedAt,
-        calendarChannel.throttleFailureCount,
-      )
+      calendarChannel.syncStage !==
+      CalendarChannelSyncStage.CALENDAR_EVENTS_IMPORT_SCHEDULED
     ) {
       return;
     }
 
     if (
-      calendarChannel.syncStage !==
-      CalendarChannelSyncStage.CALENDAR_EVENTS_IMPORT_SCHEDULED
+      isThrottled(
+        calendarChannel.syncStageStartedAt,
+        calendarChannel.throttleFailureCount,
+      )
     ) {
+      await this.calendarChannelSyncStatusService.markAsCalendarEventsImportPending(
+        [calendarChannel.id],
+        workspaceId,
+        true,
+      );
+
       return;
     }
 

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

@@ -1,4 +1,4 @@
-import { Scope } from '@nestjs/common';
+import { Logger, Scope } from '@nestjs/common';
 
 import { In } from 'typeorm';
 
@@ -22,13 +22,16 @@ export type CalendarOngoingStaleJobData = {
   scope: Scope.REQUEST,
 })
 export class CalendarOngoingStaleJob {
+  private readonly logger = new Logger(CalendarOngoingStaleJob.name);
   constructor(
     private readonly twentyORMManager: TwentyORMManager,
     private readonly calendarChannelSyncStatusService: CalendarChannelSyncStatusService,
   ) {}
 
   @Process(CalendarOngoingStaleJob.name)
-  async handle(): Promise<void> {
+  async handle(data: CalendarOngoingStaleJobData): Promise<void> {
+    const { workspaceId } = data;
+
     const calendarChannelRepository =
       await this.twentyORMManager.getRepository<CalendarChannelWorkspaceEntity>(
         'calendarChannel',
@@ -50,21 +53,30 @@ export class CalendarOngoingStaleJob {
         calendarChannel.syncStageStartedAt &&
         isSyncStale(calendarChannel.syncStageStartedAt)
       ) {
-        await this.calendarChannelSyncStatusService.resetSyncStageStartedAt([
-          calendarChannel.id,
-        ]);
+        await this.calendarChannelSyncStatusService.resetSyncStageStartedAt(
+          [calendarChannel.id],
+          workspaceId,
+        );
 
         switch (calendarChannel.syncStage) {
           case CalendarChannelSyncStage.CALENDAR_EVENT_LIST_FETCH_ONGOING:
           case CalendarChannelSyncStage.CALENDAR_EVENT_LIST_FETCH_SCHEDULED:
-            await this.calendarChannelSyncStatusService.scheduleCalendarEventListFetch(
+            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.markAsCalendarEventListFetchPending(
               [calendarChannel.id],
+              workspaceId,
             );
             break;
           case CalendarChannelSyncStage.CALENDAR_EVENTS_IMPORT_ONGOING:
           case CalendarChannelSyncStage.CALENDAR_EVENTS_IMPORT_SCHEDULED:
-            await this.calendarChannelSyncStatusService.scheduleCalendarEventsImport(
+            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.markAsCalendarEventsImportPending(
               [calendarChannel.id],
+              workspaceId,
             );
             break;
           default:

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,14 +147,18 @@ export class CalendarEventImportErrorHandlerService {
 
     switch (syncStep) {
       case CalendarEventImportSyncStep.CALENDAR_EVENT_LIST_FETCH:
-        await this.calendarChannelSyncStatusService.scheduleCalendarEventListFetch(
+        await this.calendarChannelSyncStatusService.markAsCalendarEventListFetchPending(
           [calendarChannel.id],
+          workspaceId,
+          true,
         );
         break;
 
       case CalendarEventImportSyncStep.CALENDAR_EVENTS_IMPORT:
-        await this.calendarChannelSyncStatusService.scheduleCalendarEventsImport(
+        await this.calendarChannelSyncStatusService.markAsCalendarEventsImportPending(
           [calendarChannel.id],
+          workspaceId,
+          true,
         );
         break;
 
@@ -214,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

@@ -52,6 +52,7 @@ export class CalendarEventsImportService {
   ): Promise<void> {
     await this.calendarChannelSyncStatusService.markAsCalendarEventsImportOngoing(
       [calendarChannel.id],
+      workspaceId,
     );
 
     let calendarEvents: FetchedCalendarEvent[] = [];
@@ -66,8 +67,9 @@ export class CalendarEventsImportService {
         );
 
         if (!eventIdsToFetch || eventIdsToFetch.length === 0) {
-          await this.calendarChannelSyncStatusService.markAsCompletedAndScheduleCalendarEventListFetch(
+          await this.calendarChannelSyncStatusService.markAsCompletedAndMarkAsCalendarEventListFetchPending(
             [calendarChannel.id],
+            workspaceId,
           );
 
           return;
@@ -87,8 +89,9 @@ export class CalendarEventsImportService {
       }
 
       if (!calendarEvents || calendarEvents?.length === 0) {
-        await this.calendarChannelSyncStatusService.scheduleCalendarEventListFetch(
+        await this.calendarChannelSyncStatusService.markAsCalendarEventListFetchPending(
           [calendarChannel.id],
+          workspaceId,
         );
       }
 
@@ -149,8 +152,9 @@ export class CalendarEventsImportService {
         workspaceId,
       );
 
-      await this.calendarChannelSyncStatusService.markAsCompletedAndScheduleCalendarEventListFetch(
+      await this.calendarChannelSyncStatusService.markAsCompletedAndMarkAsCalendarEventListFetchPending(
         [calendarChannel.id],
+        workspaceId,
       );
     } catch (error) {
       await this.calendarEventImportErrorHandlerService.handleDriverException(

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

@@ -42,6 +42,7 @@ export class CalendarFetchEventsService {
   ): Promise<void> {
     await this.calendarChannelSyncStatusService.markAsCalendarEventListFetchOngoing(
       [calendarChannel.id],
+      workspaceId,
     );
 
     try {
@@ -96,8 +97,9 @@ export class CalendarFetchEventsService {
           },
         );
 
-        await this.calendarChannelSyncStatusService.scheduleCalendarEventListFetch(
+        await this.calendarChannelSyncStatusService.markAsCalendarEventListFetchPending(
           [calendarChannel.id],
+          workspaceId,
         );
       }
 
@@ -126,8 +128,9 @@ export class CalendarFetchEventsService {
           calendarEventIds,
         );
 
-        await this.calendarChannelSyncStatusService.scheduleCalendarEventsImport(
+        await this.calendarChannelSyncStatusService.markAsCalendarEventsImportPending(
           [calendarChannel.id],
+          workspaceId,
         );
       } else {
         throw new CalendarEventImportDriverException(

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

@@ -7,7 +7,7 @@ import { CacheStorageService } from 'src/engine/core-modules/cache-storage/servi
 import { CacheStorageNamespace } from 'src/engine/core-modules/cache-storage/types/cache-storage-namespace.enum';
 import { MetricsService } from 'src/engine/core-modules/metrics/metrics.service';
 import { MetricsKeys } from 'src/engine/core-modules/metrics/types/metrics-keys.type';
-import { TwentyORMManager } from 'src/engine/twenty-orm/twenty-orm.manager';
+import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager';
 import {
   CalendarChannelSyncStage,
   CalendarChannelSyncStatus,
@@ -20,37 +20,45 @@ import { AccountsToReconnectKeys } from 'src/modules/connected-account/types/acc
 @Injectable()
 export class CalendarChannelSyncStatusService {
   constructor(
-    private readonly twentyORMManager: TwentyORMManager,
+    private readonly twentyORMGlobalManager: TwentyORMGlobalManager,
     @InjectCacheStorage(CacheStorageNamespace.ModuleCalendar)
     private readonly cacheStorage: CacheStorageService,
     private readonly accountsToReconnectService: AccountsToReconnectService,
     private readonly metricsService: MetricsService,
   ) {}
 
-  public async scheduleCalendarEventListFetch(calendarChannelIds: string[]) {
-    if (!calendarChannelIds.length) {
-      return;
-    }
-
-    const calendarChannelRepository =
-      await this.twentyORMManager.getRepository<CalendarChannelWorkspaceEntity>(
-        'calendarChannel',
-      );
-
-    await calendarChannelRepository.update(calendarChannelIds, {
-      syncStage: CalendarChannelSyncStage.CALENDAR_EVENT_LIST_FETCH_PENDING,
-    });
-  }
-
-  public async markAsCalendarEventListFetchOngoing(
+  public async markAsCalendarEventListFetchPending(
     calendarChannelIds: string[],
+    workspaceId: string,
+    preserveSyncStageStartedAt: boolean = false,
   ) {
     if (!calendarChannelIds.length) {
       return;
     }
 
     const calendarChannelRepository =
-      await this.twentyORMManager.getRepository<CalendarChannelWorkspaceEntity>(
+      await this.twentyORMGlobalManager.getRepositoryForWorkspace<CalendarChannelWorkspaceEntity>(
+        workspaceId,
+        'calendarChannel',
+      );
+
+    await calendarChannelRepository.update(calendarChannelIds, {
+      syncStage: CalendarChannelSyncStage.CALENDAR_EVENT_LIST_FETCH_PENDING,
+      ...(!preserveSyncStageStartedAt ? { syncStageStartedAt: null } : {}),
+    });
+  }
+
+  public async markAsCalendarEventListFetchOngoing(
+    calendarChannelIds: string[],
+    workspaceId: string,
+  ) {
+    if (!calendarChannelIds.length) {
+      return;
+    }
+
+    const calendarChannelRepository =
+      await this.twentyORMGlobalManager.getRepositoryForWorkspace<CalendarChannelWorkspaceEntity>(
+        workspaceId,
         'calendarChannel',
       );
 
@@ -61,7 +69,7 @@ export class CalendarChannelSyncStatusService {
     });
   }
 
-  public async resetAndScheduleCalendarEventListFetch(
+  public async resetAndMarkAsCalendarEventListFetchPending(
     calendarChannelIds: string[],
     workspaceId: string,
   ) {
@@ -76,7 +84,8 @@ export class CalendarChannelSyncStatusService {
     }
 
     const calendarChannelRepository =
-      await this.twentyORMManager.getRepository<CalendarChannelWorkspaceEntity>(
+      await this.twentyORMGlobalManager.getRepositoryForWorkspace<CalendarChannelWorkspaceEntity>(
+        workspaceId,
         'calendarChannel',
       );
 
@@ -86,16 +95,23 @@ export class CalendarChannelSyncStatusService {
       throttleFailureCount: 0,
     });
 
-    await this.scheduleCalendarEventListFetch(calendarChannelIds);
+    await this.markAsCalendarEventListFetchPending(
+      calendarChannelIds,
+      workspaceId,
+    );
   }
 
-  public async resetSyncStageStartedAt(calendarChannelIds: string[]) {
+  public async resetSyncStageStartedAt(
+    calendarChannelIds: string[],
+    workspaceId: string,
+  ) {
     if (!calendarChannelIds.length) {
       return;
     }
 
     const calendarChannelRepository =
-      await this.twentyORMManager.getRepository<CalendarChannelWorkspaceEntity>(
+      await this.twentyORMGlobalManager.getRepositoryForWorkspace<CalendarChannelWorkspaceEntity>(
+        workspaceId,
         'calendarChannel',
       );
 
@@ -104,28 +120,38 @@ export class CalendarChannelSyncStatusService {
     });
   }
 
-  public async scheduleCalendarEventsImport(calendarChannelIds: string[]) {
+  public async markAsCalendarEventsImportPending(
+    calendarChannelIds: string[],
+    workspaceId: string,
+    preserveSyncStageStartedAt: boolean = false,
+  ) {
     if (!calendarChannelIds.length) {
       return;
     }
 
     const calendarChannelRepository =
-      await this.twentyORMManager.getRepository<CalendarChannelWorkspaceEntity>(
+      await this.twentyORMGlobalManager.getRepositoryForWorkspace<CalendarChannelWorkspaceEntity>(
+        workspaceId,
         'calendarChannel',
       );
 
     await calendarChannelRepository.update(calendarChannelIds, {
       syncStage: CalendarChannelSyncStage.CALENDAR_EVENTS_IMPORT_PENDING,
+      ...(!preserveSyncStageStartedAt ? { syncStageStartedAt: null } : {}),
     });
   }
 
-  public async markAsCalendarEventsImportOngoing(calendarChannelIds: string[]) {
+  public async markAsCalendarEventsImportOngoing(
+    calendarChannelIds: string[],
+    workspaceId: string,
+  ) {
     if (!calendarChannelIds.length) {
       return;
     }
 
     const calendarChannelRepository =
-      await this.twentyORMManager.getRepository<CalendarChannelWorkspaceEntity>(
+      await this.twentyORMGlobalManager.getRepositoryForWorkspace<CalendarChannelWorkspaceEntity>(
+        workspaceId,
         'calendarChannel',
       );
 
@@ -135,15 +161,17 @@ export class CalendarChannelSyncStatusService {
     });
   }
 
-  public async markAsCompletedAndScheduleCalendarEventListFetch(
+  public async markAsCompletedAndMarkAsCalendarEventListFetchPending(
     calendarChannelIds: string[],
+    workspaceId: string,
   ) {
     if (!calendarChannelIds.length) {
       return;
     }
 
     const calendarChannelRepository =
-      await this.twentyORMManager.getRepository<CalendarChannelWorkspaceEntity>(
+      await this.twentyORMGlobalManager.getRepositoryForWorkspace<CalendarChannelWorkspaceEntity>(
+        workspaceId,
         'calendarChannel',
       );
 
@@ -155,7 +183,10 @@ export class CalendarChannelSyncStatusService {
       syncedAt: new Date().toISOString(),
     });
 
-    await this.scheduleCalendarEventListFetch(calendarChannelIds);
+    await this.markAsCalendarEventListFetchPending(
+      calendarChannelIds,
+      workspaceId,
+    );
 
     await this.metricsService.batchIncrementCounter({
       key: MetricsKeys.CalendarEventSyncJobActive,
@@ -172,7 +203,8 @@ export class CalendarChannelSyncStatusService {
     }
 
     const calendarChannelRepository =
-      await this.twentyORMManager.getRepository<CalendarChannelWorkspaceEntity>(
+      await this.twentyORMGlobalManager.getRepositoryForWorkspace<CalendarChannelWorkspaceEntity>(
+        workspaceId,
         'calendarChannel',
       );
 
@@ -202,7 +234,8 @@ export class CalendarChannelSyncStatusService {
     }
 
     const calendarChannelRepository =
-      await this.twentyORMManager.getRepository<CalendarChannelWorkspaceEntity>(
+      await this.twentyORMGlobalManager.getRepositoryForWorkspace<CalendarChannelWorkspaceEntity>(
+        workspaceId,
         'calendarChannel',
       );
 
@@ -217,7 +250,8 @@ export class CalendarChannelSyncStatusService {
     });
 
     const connectedAccountRepository =
-      await this.twentyORMManager.getRepository<ConnectedAccountWorkspaceEntity>(
+      await this.twentyORMGlobalManager.getRepositoryForWorkspace<ConnectedAccountWorkspaceEntity>(
+        workspaceId,
         'connectedAccount',
       );
 
@@ -257,7 +291,8 @@ export class CalendarChannelSyncStatusService {
     }
 
     const calendarChannelRepository =
-      await this.twentyORMManager.getRepository<CalendarChannelWorkspaceEntity>(
+      await this.twentyORMGlobalManager.getRepositoryForWorkspace<CalendarChannelWorkspaceEntity>(
+        workspaceId,
         'calendarChannel',
       );
 

packages/twenty-server/src/modules/connected-account/utils/is-throttled.ts

@@ -8,6 +8,10 @@ export const isThrottled = (
     return false;
   }
 
+  if (throttleFailureCount === 0) {
+    return false;
+  }
+
   return (
     computeThrottlePauseUntil(syncStageStartedAt, throttleFailureCount) >
     new Date()

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,9 +32,10 @@ export class MessageChannelSyncStatusService {
     private readonly metricsService: MetricsService,
   ) {}
 
-  public async scheduleMessageListFetch(
+  public async markAsMessagesListFetchPending(
     messageChannelIds: string[],
     workspaceId: string,
+    preserveSyncStageStartedAt: boolean = false,
   ) {
     if (!messageChannelIds.length) {
       return;
@@ -48,12 +49,14 @@ export class MessageChannelSyncStatusService {
 
     await messageChannelRepository.update(messageChannelIds, {
       syncStage: MessageChannelSyncStage.MESSAGE_LIST_FETCH_PENDING,
+      ...(!preserveSyncStageStartedAt ? { syncStageStartedAt: null } : {}),
     });
   }
 
-  public async scheduleMessagesImport(
+  public async markAsMessagesImportPending(
     messageChannelIds: string[],
     workspaceId: string,
+    preserveSyncStageStartedAt: boolean = false,
   ) {
     if (!messageChannelIds.length) {
       return;
@@ -67,10 +70,11 @@ export class MessageChannelSyncStatusService {
 
     await messageChannelRepository.update(messageChannelIds, {
       syncStage: MessageChannelSyncStage.MESSAGES_IMPORT_PENDING,
+      ...(!preserveSyncStageStartedAt ? { syncStageStartedAt: null } : {}),
     });
   }
 
-  public async resetAndScheduleMessageListFetch(
+  public async resetAndMarkAsMessagesListFetchPending(
     messageChannelIds: string[],
     workspaceId: string,
   ) {
@@ -111,7 +115,7 @@ export class MessageChannelSyncStatusService {
       },
     );
 
-    await this.scheduleMessageListFetch(messageChannelIds, workspaceId);
+    await this.markAsMessagesListFetchPending(messageChannelIds, workspaceId);
   }
 
   public async resetSyncStageStartedAt(
@@ -174,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/constants/messaging-import-ongoing-sync-timeout.constant.ts

@@ -1 +1 @@
-export const MESSAGING_IMPORT_ONGOING_SYNC_TIMEOUT = 1000 * 60 * 60; // 1 hour
+export const MESSAGING_IMPORT_ONGOING_SYNC_TIMEOUT = 1000 * 60 * 30; // 30 minutes

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

@@ -51,7 +51,7 @@ export class MessagingMessageListFetchCronJob {
         const now = new Date().toISOString();
 
         const [messageChannels] = await this.coreDataSource.query(
-          `UPDATE ${schemaName}."messageChannel" SET "syncStage" = '${MessageChannelSyncStage.MESSAGE_LIST_FETCH_SCHEDULED}', "syncStageStartedAt" = '${now}'
+          `UPDATE ${schemaName}."messageChannel" SET "syncStage" = '${MessageChannelSyncStage.MESSAGE_LIST_FETCH_SCHEDULED}', "syncStageStartedAt" = COALESCE("syncStageStartedAt", '${now}')
            WHERE "isSyncEnabled" = true AND "syncStage" = '${MessageChannelSyncStage.MESSAGE_LIST_FETCH_PENDING}' RETURNING *`,
         );
 

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

@@ -56,7 +56,7 @@ export class MessagingMessagesImportCronJob {
         const now = new Date().toISOString();
 
         const [messageChannels] = await this.coreDataSource.query(
-          `UPDATE ${schemaName}."messageChannel" SET "syncStage" = '${MessageChannelSyncStage.MESSAGES_IMPORT_SCHEDULED}', "syncStageStartedAt" = '${now}'
+          `UPDATE ${schemaName}."messageChannel" SET "syncStage" = '${MessageChannelSyncStage.MESSAGES_IMPORT_SCHEDULED}', "syncStageStartedAt" = COALESCE("syncStageStartedAt", '${now}')
           WHERE "isSyncEnabled" = true AND "syncStage" = '${MessageChannelSyncStage.MESSAGES_IMPORT_PENDING}' RETURNING *`,
         );
 

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

@@ -5,6 +5,7 @@ import { Processor } from 'src/engine/core-modules/message-queue/decorators/proc
 import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
 import { TwentyORMManager } from 'src/engine/twenty-orm/twenty-orm.manager';
 import { isThrottled } from 'src/modules/connected-account/utils/is-throttled';
+import { MessageChannelSyncStatusService } from 'src/modules/messaging/common/services/message-channel-sync-status.service';
 import {
   MessageChannelSyncStage,
   type MessageChannelWorkspaceEntity,
@@ -31,6 +32,7 @@ export class MessagingMessageListFetchJob {
     private readonly messagingMonitoringService: MessagingMonitoringService,
     private readonly twentyORMManager: TwentyORMManager,
     private readonly messageImportErrorHandlerService: MessageImportExceptionHandlerService,
+    private readonly messageChannelSyncStatusService: MessageChannelSyncStatusService,
   ) {}
 
   @Process(MessagingMessageListFetchJob.name)
@@ -65,6 +67,13 @@ export class MessagingMessageListFetchJob {
       return;
     }
 
+    if (
+      messageChannel.syncStage !==
+      MessageChannelSyncStage.MESSAGE_LIST_FETCH_SCHEDULED
+    ) {
+      return;
+    }
+
     try {
       if (
         isThrottled(
@@ -72,35 +81,33 @@ export class MessagingMessageListFetchJob {
           messageChannel.throttleFailureCount,
         )
       ) {
+        await this.messageChannelSyncStatusService.markAsMessagesListFetchPending(
+          [messageChannel.id],
+          workspaceId,
+          true,
+        );
+
         return;
       }
 
-      switch (messageChannel.syncStage) {
-        case MessageChannelSyncStage.MESSAGE_LIST_FETCH_SCHEDULED:
-          await this.messagingMonitoringService.track({
-            eventName: 'message_list_fetch.started',
-            workspaceId,
-            connectedAccountId: messageChannel.connectedAccount.id,
-            messageChannelId: messageChannel.id,
-          });
+      await this.messagingMonitoringService.track({
+        eventName: 'message_list_fetch.started',
+        workspaceId,
+        connectedAccountId: messageChannel.connectedAccount.id,
+        messageChannelId: messageChannel.id,
+      });
 
-          await this.messagingMessageListFetchService.processMessageListFetch(
-            messageChannel,
-            workspaceId,
-          );
+      await this.messagingMessageListFetchService.processMessageListFetch(
+        messageChannel,
+        workspaceId,
+      );
 
-          await this.messagingMonitoringService.track({
-            eventName: 'message_list_fetch.completed',
-            workspaceId,
-            connectedAccountId: messageChannel.connectedAccount.id,
-            messageChannelId: messageChannel.id,
-          });
-
-          break;
-
-        default:
-          break;
-      }
+      await this.messagingMonitoringService.track({
+        eventName: 'message_list_fetch.completed',
+        workspaceId,
+        connectedAccountId: messageChannel.connectedAccount.id,
+        messageChannelId: messageChannel.id,
+      });
     } catch (error) {
       await this.messageImportErrorHandlerService.handleDriverException(
         error,

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

@@ -5,6 +5,7 @@ import { Processor } from 'src/engine/core-modules/message-queue/decorators/proc
 import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
 import { TwentyORMManager } from 'src/engine/twenty-orm/twenty-orm.manager';
 import { isThrottled } from 'src/modules/connected-account/utils/is-throttled';
+import { MessageChannelSyncStatusService } from 'src/modules/messaging/common/services/message-channel-sync-status.service';
 import {
   MessageChannelSyncStage,
   type MessageChannelWorkspaceEntity,
@@ -24,6 +25,7 @@ export class MessagingMessagesImportJob {
   constructor(
     private readonly messagingMessagesImportService: MessagingMessagesImportService,
     private readonly messagingMonitoringService: MessagingMonitoringService,
+    private readonly messageChannelSyncStatusService: MessageChannelSyncStatusService,
     private readonly twentyORMManager: TwentyORMManager,
   ) {}
 
@@ -64,18 +66,24 @@ export class MessagingMessagesImportJob {
     }
 
     if (
-      isThrottled(
-        messageChannel.syncStageStartedAt,
-        messageChannel.throttleFailureCount,
-      )
+      messageChannel.syncStage !==
+      MessageChannelSyncStage.MESSAGES_IMPORT_SCHEDULED
     ) {
       return;
     }
 
     if (
-      messageChannel.syncStage !==
-      MessageChannelSyncStage.MESSAGES_IMPORT_SCHEDULED
+      isThrottled(
+        messageChannel.syncStageStartedAt,
+        messageChannel.throttleFailureCount,
+      )
     ) {
+      await this.messageChannelSyncStatusService.markAsMessagesImportPending(
+        [messageChannel.id],
+        workspaceId,
+        true,
+      );
+
       return;
     }
 

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

@@ -53,10 +53,6 @@ export class MessagingOngoingStaleJob {
         messageChannel.syncStageStartedAt &&
         isSyncStale(messageChannel.syncStageStartedAt)
       ) {
-        this.logger.log(
-          `Sync for message channel ${messageChannel.id} and workspace ${workspaceId} is stale. Setting sync stage to MESSAGES_IMPORT_PENDING`,
-        );
-
         await this.messageChannelSyncStatusService.resetSyncStageStartedAt(
           [messageChannel.id],
           workspaceId,
@@ -65,14 +61,20 @@ export class MessagingOngoingStaleJob {
         switch (messageChannel.syncStage) {
           case MessageChannelSyncStage.MESSAGE_LIST_FETCH_ONGOING:
           case MessageChannelSyncStage.MESSAGE_LIST_FETCH_SCHEDULED:
-            await this.messageChannelSyncStatusService.scheduleMessageListFetch(
+            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.markAsMessagesListFetchPending(
               [messageChannel.id],
               workspaceId,
             );
             break;
           case MessageChannelSyncStage.MESSAGES_IMPORT_ONGOING:
           case MessageChannelSyncStage.MESSAGES_IMPORT_SCHEDULED:
-            await this.messageChannelSyncStatusService.scheduleMessagesImport(
+            this.logger.log(
+              `Sync for message channel ${messageChannel.id} and workspace ${workspaceId} is stale. Setting sync stage to MESSAGES_IMPORT_PENDING`,
+            );
+            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-get-message-list.service.ts

@@ -2,10 +2,7 @@ import { Injectable } from '@nestjs/common';
 
 import { ConnectedAccountProvider } from 'twenty-shared/types';
 
-import {
-  MessageFolderImportPolicy,
-  type MessageChannelWorkspaceEntity,
-} from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity';
+import { type MessageChannelWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity';
 import { MessageFolderWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-folder.workspace-entity';
 import {
   MessageImportDriverException,
@@ -33,30 +30,24 @@ export class MessagingGetMessageListService {
     messageChannel: MessageChannelWorkspaceEntity,
     messageFolders: MessageFolder[],
   ): Promise<GetMessageListsResponse> {
-    const messageFoldersToSync: MessageFolder[] =
-      messageChannel.messageFolderImportPolicy ===
-      MessageFolderImportPolicy.ALL_FOLDERS
-        ? messageFolders
-        : messageFolders.filter((folder) => folder.isSynced);
-
     switch (messageChannel.connectedAccount.provider) {
       case ConnectedAccountProvider.GOOGLE:
         return await this.gmailGetMessageListService.getMessageLists({
           messageChannel,
           connectedAccount: messageChannel.connectedAccount,
-          messageFolders: messageFoldersToSync,
+          messageFolders,
         });
       case ConnectedAccountProvider.MICROSOFT:
         return this.microsoftGetMessageListService.getMessageLists({
           messageChannel,
           connectedAccount: messageChannel.connectedAccount,
-          messageFolders: messageFoldersToSync,
+          messageFolders,
         });
       case ConnectedAccountProvider.IMAP_SMTP_CALDAV: {
         return await this.imapGetMessageListService.getMessageLists({
           messageChannel,
           connectedAccount: messageChannel.connectedAccount,
-          messageFolders: messageFoldersToSync,
+          messageFolders,
         });
       }
       default:

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

@@ -163,17 +163,19 @@ export class MessageImportExceptionHandlerService {
 
     switch (syncStep) {
       case MessageImportSyncStep.MESSAGE_LIST_FETCH:
-        await this.messageChannelSyncStatusService.scheduleMessageListFetch(
+        await this.messageChannelSyncStatusService.markAsMessagesListFetchPending(
           [messageChannel.id],
           workspaceId,
+          true,
         );
         break;
 
       case MessageImportSyncStep.MESSAGES_IMPORT_PENDING:
       case MessageImportSyncStep.MESSAGES_IMPORT_ONGOING:
-        await this.messageChannelSyncStatusService.scheduleMessagesImport(
+        await this.messageChannelSyncStatusService.markAsMessagesImportPending(
           [messageChannel.id],
           workspaceId,
+          true,
         );
         break;
 
@@ -248,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

@@ -15,6 +15,7 @@ import {
   MessageChannelPendingGroupEmailsAction,
   MessageChannelSyncStage,
   MessageChannelWorkspaceEntity,
+  MessageFolderImportPolicy,
 } from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity';
 import {
   MessageFolderPendingSyncAction,
@@ -135,10 +136,16 @@ export class MessagingMessageListFetchService {
         },
       });
 
+      const messageFoldersToSync =
+        messageChannelWithFreshTokens.messageFolderImportPolicy ===
+        MessageFolderImportPolicy.ALL_FOLDERS
+          ? messageFolders
+          : messageFolders.filter((folder) => folder.isSynced);
+
       const messageLists =
         await this.messagingGetMessageListService.getMessageLists(
           messageChannelWithFreshTokens,
-          messageFolders,
+          messageFoldersToSync,
         );
 
       await this.cacheStorage.del(
@@ -269,7 +276,7 @@ export class MessagingMessageListFetchService {
       );
 
       if (totalMessagesToImportCount === 0) {
-        await this.messageChannelSyncStatusService.markAsCompletedAndScheduleMessageListFetch(
+        await this.messageChannelSyncStatusService.markAsCompletedAndMarkAsMessagesListFetchPending(
           [messageChannelWithFreshTokens.id],
           workspaceId,
         );
@@ -281,7 +288,7 @@ export class MessagingMessageListFetchService {
         `messageChannelId: ${freshMessageChannel.id} Scheduling direct messages import`,
       );
 
-      await this.messageChannelSyncStatusService.scheduleMessagesImport(
+      await this.messageChannelSyncStatusService.markAsMessagesImportScheduled(
         [messageChannelWithFreshTokens.id],
         workspaceId,
       );
@@ -289,7 +296,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,
         );

packages/twenty-server/src/modules/workflow/workflow-builder/workflow-schema/utils/__tests__/should-generate-field-fake-value.spec.ts

@@ -56,7 +56,7 @@ describe('shouldGenerateFieldFakeValue', () => {
     expect(shouldGenerateFieldFakeValue(field)).toBe(true);
   });
 
-  it('should return false for many-to-one relation fields', () => {
+  it('should return true for many-to-one relation fields', () => {
     const field = getFlatFieldMetadataMock({
       objectMetadataId: '20202020-0000-0000-0000-000000000001',
       isSystem: false,
@@ -69,6 +69,6 @@ describe('shouldGenerateFieldFakeValue', () => {
       },
     });
 
-    expect(shouldGenerateFieldFakeValue(field)).toBe(false);
+    expect(shouldGenerateFieldFakeValue(field)).toBe(true);
   });
 });

packages/twenty-server/src/modules/workflow/workflow-builder/workflow-schema/utils/generate-object-record-fields.ts

@@ -2,6 +2,7 @@ import { FieldMetadataType } from 'twenty-shared/types';
 import { isDefined } from 'twenty-shared/utils';
 
 import { findFlatEntityByIdInFlatEntityMapsOrThrow } from 'src/engine/metadata-modules/flat-entity/utils/find-flat-entity-by-id-in-flat-entity-maps-or-throw.util';
+import { isMorphOrRelationFlatFieldMetadata } from 'src/engine/metadata-modules/flat-field-metadata/utils/is-morph-or-relation-flat-field-metadata.util';
 import { type ObjectMetadataInfo } from 'src/modules/workflow/common/workspace-services/workflow-common.workspace-service';
 import { type FieldOutputSchema } from 'src/modules/workflow/workflow-builder/workflow-schema/types/output-schema.type';
 import { generateFakeObjectRecord } from 'src/modules/workflow/workflow-builder/workflow-schema/utils/generate-fake-object-record';
@@ -33,7 +34,7 @@ export const generateObjectRecordFields = ({
       continue;
     }
 
-    if (field.type !== FieldMetadataType.RELATION) {
+    if (!isMorphOrRelationFlatFieldMetadata(field)) {
       result[field.name] = generateFakeRecordField({
         type: field.type,
         label: field.label,
@@ -44,10 +45,6 @@ export const generateObjectRecordFields = ({
       continue;
     }
 
-    if (!isDefined(field.relationTargetObjectMetadataId)) {
-      continue;
-    }
-
     if (depth < maxDepth) {
       const relationTargetObjectMetadata =
         flatObjectMetadataMaps.byId[field.relationTargetObjectMetadataId];

packages/twenty-server/src/modules/workflow/workflow-builder/workflow-schema/utils/should-generate-field-fake-value.ts

@@ -2,6 +2,17 @@ import { FieldMetadataType, RelationType } from 'twenty-shared/types';
 
 import { type FlatFieldMetadata } from 'src/engine/metadata-modules/flat-field-metadata/types/flat-field-metadata.type';
 import { isFlatFieldMetadataOfType } from 'src/engine/metadata-modules/flat-field-metadata/utils/is-flat-field-metadata-of-type.util';
+import { isMorphOrRelationFlatFieldMetadata } from 'src/engine/metadata-modules/flat-field-metadata/utils/is-morph-or-relation-flat-field-metadata.util';
+
+const EXCLUDED_SYSTEM_FIELDS = ['searchVector', 'position'];
+
+const isActiveField = (field: FlatFieldMetadata) => {
+  return field.isActive;
+};
+
+const isExcludedSystemField = (field: FlatFieldMetadata) => {
+  return field.isSystem && EXCLUDED_SYSTEM_FIELDS.includes(field.name);
+};
 
 const isManyToOneRelationField = (field: FlatFieldMetadata) => {
   if (isFlatFieldMetadataOfType(field, FieldMetadataType.RELATION)) {
@@ -11,12 +22,17 @@ const isManyToOneRelationField = (field: FlatFieldMetadata) => {
   return false;
 };
 
-const EXCLUDED_SYSTEM_FIELDS = ['searchVector', 'position'];
-
-export const shouldGenerateFieldFakeValue = (field: FlatFieldMetadata) => {
+const isExcludedRelationField = (field: FlatFieldMetadata) => {
   return (
-    field.isActive &&
-    !(EXCLUDED_SYSTEM_FIELDS.includes(field.name) && field.isSystem) &&
+    isMorphOrRelationFlatFieldMetadata(field) &&
     !isManyToOneRelationField(field)
   );
 };
+
+export const shouldGenerateFieldFakeValue = (field: FlatFieldMetadata) => {
+  return (
+    isActiveField(field) &&
+    !isExcludedSystemField(field) &&
+    !isExcludedRelationField(field)
+  );
+};