connected accounts follow up (#18998)

- Fixed group emails actions
- Fixed delta updates for folder manager
- Updated crons

packages/twenty-server/src/modules/calendar/calendar-event-import-manager/calendar-event-import-manager.module.ts

@@ -3,6 +3,7 @@ import { TypeOrmModule } from '@nestjs/typeorm';
 
 import { BillingModule } from 'src/engine/core-modules/billing/billing.module';
 import { FeatureFlagEntity } from 'src/engine/core-modules/feature-flag/feature-flag.entity';
+import { FeatureFlagModule } from 'src/engine/core-modules/feature-flag/feature-flag.module';
 import { MetricsModule } from 'src/engine/core-modules/metrics/metrics.module';
 import { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity';
 import { DataSourceEntity } from 'src/engine/metadata-modules/data-source/data-source.entity';
@@ -61,6 +62,7 @@ import { RefreshTokensManagerModule } from 'src/modules/connected-account/refres
     MetricsModule,
     CalendarChannelDataAccessModule,
     ConnectedAccountDataAccessModule,
+    FeatureFlagModule,
   ],
   providers: [
     CalendarAccountAuthenticationService,

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

@@ -1,10 +1,12 @@
 import { InjectDataSource, InjectRepository } from '@nestjs/typeorm';
 
+import { FeatureFlagKey } from 'twenty-shared/types';
 import { WorkspaceActivationStatus } from 'twenty-shared/workspace';
 import { DataSource, 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';
+import { FeatureFlagService } from 'src/engine/core-modules/feature-flag/services/feature-flag.service';
 import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator';
 import { Process } from 'src/engine/core-modules/message-queue/decorators/process.decorator';
 import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.decorator';
@@ -32,6 +34,7 @@ export class CalendarEventListFetchCronJob {
     private readonly exceptionHandlerService: ExceptionHandlerService,
     @InjectDataSource()
     private readonly coreDataSource: DataSource,
+    private readonly featureFlagService: FeatureFlagService,
   ) {}
 
   @Process(CalendarEventListFetchCronJob.name)
@@ -48,15 +51,24 @@ export class CalendarEventListFetchCronJob {
 
     for (const activeWorkspace of activeWorkspaces) {
       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}', "syncStageStartedAt" = COALESCE("syncStageStartedAt", '${now}')
-          WHERE "isSyncEnabled" = true AND "syncStage" = '${CalendarChannelSyncStage.CALENDAR_EVENT_LIST_FETCH_PENDING}' RETURNING *`,
+        // TODO: remove workspace schema branch once IS_CONNECTED_ACCOUNT_MIGRATED feature flag is removed
+        const isMigrated = await this.featureFlagService.isFeatureEnabled(
+          FeatureFlagKey.IS_CONNECTED_ACCOUNT_MIGRATED,
+          activeWorkspace.id,
         );
 
+        const [calendarChannels] = isMigrated
+          ? await this.coreDataSource.query(
+              `UPDATE core."calendarChannel" SET "syncStage" = '${CalendarChannelSyncStage.CALENDAR_EVENT_LIST_FETCH_SCHEDULED}', "syncStageStartedAt" = COALESCE("syncStageStartedAt", '${now}')
+               WHERE "workspaceId" = '${activeWorkspace.id}' AND "isSyncEnabled" = true AND "syncStage" = '${CalendarChannelSyncStage.CALENDAR_EVENT_LIST_FETCH_PENDING}' RETURNING *`,
+            )
+          : await this.coreDataSource.query(
+              `UPDATE ${getWorkspaceSchemaName(activeWorkspace.id)}."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 *`,
+            );
+
         for (const calendarChannel of calendarChannels) {
           await this.messageQueueService.add<CalendarEventListFetchJobData>(
             CalendarEventListFetchJob.name,

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

@@ -1,10 +1,12 @@
 import { InjectDataSource, InjectRepository } from '@nestjs/typeorm';
 
+import { FeatureFlagKey } from 'twenty-shared/types';
 import { WorkspaceActivationStatus } from 'twenty-shared/workspace';
 import { DataSource, 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';
+import { FeatureFlagService } from 'src/engine/core-modules/feature-flag/services/feature-flag.service';
 import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator';
 import { Process } from 'src/engine/core-modules/message-queue/decorators/process.decorator';
 import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.decorator';
@@ -30,6 +32,7 @@ export class CalendarEventsImportCronJob {
     @InjectDataSource()
     private readonly coreDataSource: DataSource,
     private readonly exceptionHandlerService: ExceptionHandlerService,
+    private readonly featureFlagService: FeatureFlagService,
   ) {}
 
   @Process(CalendarEventsImportCronJob.name)
@@ -46,15 +49,24 @@ export class CalendarEventsImportCronJob {
 
     for (const activeWorkspace of activeWorkspaces) {
       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}', "syncStageStartedAt" = COALESCE("syncStageStartedAt", '${now}')
-           WHERE "isSyncEnabled" = true AND "syncStage" = '${CalendarChannelSyncStage.CALENDAR_EVENTS_IMPORT_PENDING}' RETURNING *`,
+        // TODO: remove workspace schema branch once IS_CONNECTED_ACCOUNT_MIGRATED feature flag is removed
+        const isMigrated = await this.featureFlagService.isFeatureEnabled(
+          FeatureFlagKey.IS_CONNECTED_ACCOUNT_MIGRATED,
+          activeWorkspace.id,
         );
 
+        const [calendarChannels] = isMigrated
+          ? await this.coreDataSource.query(
+              `UPDATE core."calendarChannel" SET "syncStage" = '${CalendarChannelSyncStage.CALENDAR_EVENTS_IMPORT_SCHEDULED}', "syncStageStartedAt" = COALESCE("syncStageStartedAt", '${now}')
+               WHERE "workspaceId" = '${activeWorkspace.id}' AND "isSyncEnabled" = true AND "syncStage" = '${CalendarChannelSyncStage.CALENDAR_EVENTS_IMPORT_PENDING}' RETURNING *`,
+            )
+          : await this.coreDataSource.query(
+              `UPDATE ${getWorkspaceSchemaName(activeWorkspace.id)}."calendarChannel" SET "syncStage" = '${CalendarChannelSyncStage.CALENDAR_EVENTS_IMPORT_SCHEDULED}', "syncStageStartedAt" = COALESCE("syncStageStartedAt", '${now}')
+               WHERE "isSyncEnabled" = true AND "syncStage" = '${CalendarChannelSyncStage.CALENDAR_EVENTS_IMPORT_PENDING}' RETURNING *`,
+            );
+
         for (const calendarChannel of calendarChannels) {
           await this.messageQueueService.add<CalendarEventListFetchJobData>(
             CalendarEventsImportJob.name,

packages/twenty-server/src/modules/messaging/message-folder-manager/services/sync-message-folders.service.spec.ts

@@ -313,7 +313,7 @@ describe('SyncMessageFoldersService', () => {
 
         expect(mockMessageFolderDataAccessService.update).toHaveBeenCalledWith(
           workspaceId,
-          { id: 'folder-1' },
+          { id: 'folder-1', messageChannelId: 'channel-123' },
           expect.objectContaining({ name: 'Primary Inbox' }),
         );
         expect(result).toContainEqual(
@@ -353,7 +353,7 @@ describe('SyncMessageFoldersService', () => {
 
         expect(mockMessageFolderDataAccessService.update).toHaveBeenCalledWith(
           workspaceId,
-          { id: 'folder-1' },
+          { id: 'folder-1', messageChannelId: 'channel-123' },
           expect.objectContaining({
             parentFolderId: 'new-parent-id',
           }),
@@ -500,7 +500,7 @@ describe('SyncMessageFoldersService', () => {
         );
         expect(mockMessageFolderDataAccessService.update).toHaveBeenCalledWith(
           workspaceId,
-          { id: 'folder-to-update' },
+          { id: 'folder-to-update', messageChannelId: 'channel-123' },
           expect.objectContaining({ name: 'New Name' }),
         );
         expect(mockMessageFolderDataAccessService.save).toHaveBeenCalledWith(

packages/twenty-server/src/modules/messaging/message-folder-manager/services/sync-message-folders.service.ts

@@ -152,7 +152,7 @@ export class SyncMessageFoldersService {
           for (const [id, data] of foldersToUpdate.entries()) {
             await this.messageFolderDataAccessService.update(
               workspaceId,
-              { id },
+              { id, messageChannelId },
               data,
             );
           }

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

@@ -1,10 +1,12 @@
 import { InjectDataSource, InjectRepository } from '@nestjs/typeorm';
 
+import { FeatureFlagKey } from 'twenty-shared/types';
 import { WorkspaceActivationStatus } from 'twenty-shared/workspace';
 import { DataSource, 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';
+import { FeatureFlagService } from 'src/engine/core-modules/feature-flag/services/feature-flag.service';
 import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator';
 import { Process } from 'src/engine/core-modules/message-queue/decorators/process.decorator';
 import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.decorator';
@@ -30,6 +32,7 @@ export class MessagingMessageListFetchCronJob {
     @InjectDataSource()
     private readonly coreDataSource: DataSource,
     private readonly exceptionHandlerService: ExceptionHandlerService,
+    private readonly featureFlagService: FeatureFlagService,
   ) {}
 
   @Process(MessagingMessageListFetchCronJob.name)
@@ -46,15 +49,24 @@ export class MessagingMessageListFetchCronJob {
 
     for (const activeWorkspace of activeWorkspaces) {
       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}', "syncStageStartedAt" = COALESCE("syncStageStartedAt", '${now}')
-           WHERE "isSyncEnabled" = true AND "syncStage" = '${MessageChannelSyncStage.MESSAGE_LIST_FETCH_PENDING}' RETURNING *`,
+        // TODO: remove workspace schema branch once IS_CONNECTED_ACCOUNT_MIGRATED feature flag is removed
+        const isMigrated = await this.featureFlagService.isFeatureEnabled(
+          FeatureFlagKey.IS_CONNECTED_ACCOUNT_MIGRATED,
+          activeWorkspace.id,
         );
 
+        const [messageChannels] = isMigrated
+          ? await this.coreDataSource.query(
+              `UPDATE core."messageChannel" SET "syncStage" = '${MessageChannelSyncStage.MESSAGE_LIST_FETCH_SCHEDULED}', "syncStageStartedAt" = COALESCE("syncStageStartedAt", '${now}')
+               WHERE "workspaceId" = '${activeWorkspace.id}' AND "isSyncEnabled" = true AND "syncStage" = '${MessageChannelSyncStage.MESSAGE_LIST_FETCH_PENDING}' RETURNING *`,
+            )
+          : await this.coreDataSource.query(
+              `UPDATE ${getWorkspaceSchemaName(activeWorkspace.id)}."messageChannel" SET "syncStage" = '${MessageChannelSyncStage.MESSAGE_LIST_FETCH_SCHEDULED}', "syncStageStartedAt" = COALESCE("syncStageStartedAt", '${now}')
+               WHERE "isSyncEnabled" = true AND "syncStage" = '${MessageChannelSyncStage.MESSAGE_LIST_FETCH_PENDING}' RETURNING *`,
+            );
+
         for (const messageChannel of messageChannels) {
           await this.messageQueueService.add<MessagingMessageListFetchJobData>(
             MessagingMessageListFetchJob.name,

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

@@ -1,11 +1,13 @@
 import { InjectDataSource, InjectRepository } from '@nestjs/typeorm';
 
+import { FeatureFlagKey } from 'twenty-shared/types';
 import { isDefined } from 'twenty-shared/utils';
 import { WorkspaceActivationStatus } from 'twenty-shared/workspace';
 import { DataSource, 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';
+import { FeatureFlagService } from 'src/engine/core-modules/feature-flag/services/feature-flag.service';
 import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator';
 import { Process } from 'src/engine/core-modules/message-queue/decorators/process.decorator';
 import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.decorator';
@@ -35,6 +37,7 @@ export class MessagingMessagesImportCronJob {
     private readonly exceptionHandlerService: ExceptionHandlerService,
     @InjectDataSource()
     private readonly coreDataSource: DataSource,
+    private readonly featureFlagService: FeatureFlagService,
   ) {}
 
   @Process(MessagingMessagesImportCronJob.name)
@@ -51,15 +54,24 @@ export class MessagingMessagesImportCronJob {
 
     for (const activeWorkspace of activeWorkspaces) {
       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}', "syncStageStartedAt" = COALESCE("syncStageStartedAt", '${now}')
-          WHERE "isSyncEnabled" = true AND "syncStage" = '${MessageChannelSyncStage.MESSAGES_IMPORT_PENDING}' RETURNING *`,
+        // TODO: remove workspace schema branch once IS_CONNECTED_ACCOUNT_MIGRATED feature flag is removed
+        const isMigrated = await this.featureFlagService.isFeatureEnabled(
+          FeatureFlagKey.IS_CONNECTED_ACCOUNT_MIGRATED,
+          activeWorkspace.id,
         );
 
+        const [messageChannels] = isMigrated
+          ? await this.coreDataSource.query(
+              `UPDATE core."messageChannel" SET "syncStage" = '${MessageChannelSyncStage.MESSAGES_IMPORT_SCHEDULED}', "syncStageStartedAt" = COALESCE("syncStageStartedAt", '${now}')
+               WHERE "workspaceId" = '${activeWorkspace.id}' AND "isSyncEnabled" = true AND "syncStage" = '${MessageChannelSyncStage.MESSAGES_IMPORT_PENDING}' RETURNING *`,
+            )
+          : await this.coreDataSource.query(
+              `UPDATE ${getWorkspaceSchemaName(activeWorkspace.id)}."messageChannel" SET "syncStage" = '${MessageChannelSyncStage.MESSAGES_IMPORT_SCHEDULED}', "syncStageStartedAt" = COALESCE("syncStageStartedAt", '${now}')
+               WHERE "isSyncEnabled" = true AND "syncStage" = '${MessageChannelSyncStage.MESSAGES_IMPORT_PENDING}' RETURNING *`,
+            );
+
         for (const messageChannel of messageChannels) {
           await this.messageQueueService.add<MessagingMessagesImportJobData>(
             MessagingMessagesImportJob.name,

packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-process-group-email-actions.service.ts

@@ -31,11 +31,15 @@ export class MessagingProcessGroupEmailActionsService {
     workspaceId: string,
     pendingGroupEmailsAction: MessageChannelPendingGroupEmailsAction,
   ): Promise<void> {
-    await this.messageChannelDataAccessService.update(
-      workspaceId,
-      { id: messageChannel.id },
-      { pendingGroupEmailsAction },
-    );
+    const authContext = buildSystemAuthContext(workspaceId);
+
+    await this.globalWorkspaceOrmManager.executeInWorkspaceContext(async () => {
+      await this.messageChannelDataAccessService.update(
+        workspaceId,
+        { id: messageChannel.id },
+        { pendingGroupEmailsAction },
+      );
+    }, authContext);
 
     this.logger.debug(
       `WorkspaceId: ${workspaceId}, MessageChannelId: ${messageChannel.id} - Marked message channel as pending group emails action: ${pendingGroupEmailsAction}`,