From 82611de9b670656e1faa2bf72b2b5c41ac0f69fd Mon Sep 17 00:00:00 2001 From: neo773 <62795688+neo773@users.noreply.github.com> Date: Thu, 26 Mar 2026 18:44:49 +0530 Subject: [PATCH] connected accounts follow up (#18998) - Fixed group emails actions - Fixed delta updates for folder manager - Updated crons --- .../workspaces-migration.command-runner.ts | 5 ++++- .../message-channel-data-access.module.ts | 2 ++ .../message-channel-data-access.service.ts | 18 ++++++++------- .../message-folder-data-access.service.ts | 14 +++++++++--- .../calendar-event-import-manager.module.ts | 2 ++ .../calendar-event-list-fetch.cron.job.ts | 22 ++++++++++++++----- .../jobs/calendar-events-import.cron.job.ts | 22 ++++++++++++++----- .../sync-message-folders.service.spec.ts | 6 ++--- .../services/sync-message-folders.service.ts | 2 +- .../messaging-message-list-fetch.cron.job.ts | 22 ++++++++++++++----- .../messaging-messages-import.cron.job.ts | 22 ++++++++++++++----- ...ing-process-group-email-actions.service.ts | 14 +++++++----- 12 files changed, 110 insertions(+), 41 deletions(-) diff --git a/packages/twenty-server/src/database/commands/command-runners/workspaces-migration.command-runner.ts b/packages/twenty-server/src/database/commands/command-runners/workspaces-migration.command-runner.ts index abb59c428e..8aa911a7a6 100644 --- a/packages/twenty-server/src/database/commands/command-runners/workspaces-migration.command-runner.ts +++ b/packages/twenty-server/src/database/commands/command-runners/workspaces-migration.command-runner.ts @@ -178,7 +178,10 @@ export abstract class WorkspacesMigrationCommandRunner< } this.migrationReport.fail.forEach(({ error, workspaceId }) => - this.logger.error(`Error in workspace ${workspaceId}: ${error.message}`), + this.logger.error( + `Error in workspace ${workspaceId}: ${error.message}`, + error.stack, + ), ); } diff --git a/packages/twenty-server/src/engine/metadata-modules/message-channel/data-access/message-channel-data-access.module.ts b/packages/twenty-server/src/engine/metadata-modules/message-channel/data-access/message-channel-data-access.module.ts index 73a1ab61c4..75b7e3dcd6 100644 --- a/packages/twenty-server/src/engine/metadata-modules/message-channel/data-access/message-channel-data-access.module.ts +++ b/packages/twenty-server/src/engine/metadata-modules/message-channel/data-access/message-channel-data-access.module.ts @@ -5,12 +5,14 @@ import { FeatureFlagModule } from 'src/engine/core-modules/feature-flag/feature- import { ConnectedAccountDataAccessModule } from 'src/engine/metadata-modules/connected-account/data-access/connected-account-data-access.module'; import { MessageChannelDataAccessService } from 'src/engine/metadata-modules/message-channel/data-access/services/message-channel-data-access.service'; import { MessageChannelEntity } from 'src/engine/metadata-modules/message-channel/entities/message-channel.entity'; +import { MessageFolderDataAccessModule } from 'src/engine/metadata-modules/message-folder/data-access/message-folder-data-access.module'; @Module({ imports: [ TypeOrmModule.forFeature([MessageChannelEntity]), FeatureFlagModule, ConnectedAccountDataAccessModule, + MessageFolderDataAccessModule, ], providers: [MessageChannelDataAccessService], exports: [MessageChannelDataAccessService], diff --git a/packages/twenty-server/src/engine/metadata-modules/message-channel/data-access/services/message-channel-data-access.service.ts b/packages/twenty-server/src/engine/metadata-modules/message-channel/data-access/services/message-channel-data-access.service.ts index 53481ac4f4..1dd99c1f1c 100644 --- a/packages/twenty-server/src/engine/metadata-modules/message-channel/data-access/services/message-channel-data-access.service.ts +++ b/packages/twenty-server/src/engine/metadata-modules/message-channel/data-access/services/message-channel-data-access.service.ts @@ -12,6 +12,7 @@ import { import { FeatureFlagService } from 'src/engine/core-modules/feature-flag/services/feature-flag.service'; import { ConnectedAccountDataAccessService } from 'src/engine/metadata-modules/connected-account/data-access/services/connected-account-data-access.service'; import { MessageChannelEntity } from 'src/engine/metadata-modules/message-channel/entities/message-channel.entity'; +import { MessageFolderDataAccessService } from 'src/engine/metadata-modules/message-folder/data-access/services/message-folder-data-access.service'; import { type WorkspaceEntityManager } from 'src/engine/twenty-orm/entity-manager/workspace-entity-manager'; import { GlobalWorkspaceOrmManager } from 'src/engine/twenty-orm/global-workspace-datasource/global-workspace-orm.manager'; import { type MessageChannelWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity'; @@ -26,6 +27,7 @@ export class MessageChannelDataAccessService { private readonly featureFlagService: FeatureFlagService, private readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager, private readonly connectedAccountDataAccessService: ConnectedAccountDataAccessService, + private readonly messageFolderDataAccessService: MessageFolderDataAccessService, ) {} private async isMigrated(workspaceId: string): Promise { @@ -133,15 +135,15 @@ export class MessageChannelDataAccessService { } if (needsMessageFolders) { - const workspaceRepository = - await this.getWorkspaceRepository(workspaceId); + const messageFolders = await this.messageFolderDataAccessService.find( + workspaceId, + { + messageChannelId: result.id, + }, + ); - const workspaceChannel = await workspaceRepository.findOne({ - where: { id: result.id }, - relations: ['messageFolders'], - }); - - workspaceResult.messageFolders = workspaceChannel?.messageFolders ?? []; + workspaceResult.messageFolders = + messageFolders as unknown as MessageChannelWorkspaceEntity['messageFolders']; } return workspaceResult; diff --git a/packages/twenty-server/src/engine/metadata-modules/message-folder/data-access/services/message-folder-data-access.service.ts b/packages/twenty-server/src/engine/metadata-modules/message-folder/data-access/services/message-folder-data-access.service.ts index b2d64563e7..1a1439f683 100644 --- a/packages/twenty-server/src/engine/metadata-modules/message-folder/data-access/services/message-folder-data-access.service.ts +++ b/packages/twenty-server/src/engine/metadata-modules/message-folder/data-access/services/message-folder-data-access.service.ts @@ -38,15 +38,17 @@ export class MessageFolderDataAccessService { private async toCore( workspaceId: string, data: Partial, + messageChannelId?: string, ): Promise> { const coreData: Record = { ...data, workspaceId }; const parentFolderId = coreData.parentFolderId as string | null; + const channelId = (coreData.messageChannelId as string) ?? messageChannelId; - if (parentFolderId && !uuidValidate(parentFolderId)) { + if (parentFolderId && !uuidValidate(parentFolderId) && channelId) { const parentFolder = await this.coreRepository.findOne({ where: { workspaceId, - messageChannelId: coreData.messageChannelId as string, + messageChannelId: channelId, externalId: parentFolderId, }, select: ['id'], @@ -146,9 +148,15 @@ export class MessageFolderDataAccessService { if (await this.isMigrated(workspaceId)) { try { + const coreData = await this.toCore( + workspaceId, + data, + (where as Record).messageChannelId as string, + ); + await this.coreRepository.update( { ...where, workspaceId } as FindOptionsWhere, - data as Record, + coreData, ); } catch (error) { this.logger.error( diff --git a/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/calendar-event-import-manager.module.ts b/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/calendar-event-import-manager.module.ts index 8632e43770..973e228b4b 100644 --- a/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/calendar-event-import-manager.module.ts +++ b/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, diff --git a/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/crons/jobs/calendar-event-list-fetch.cron.job.ts b/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/crons/jobs/calendar-event-list-fetch.cron.job.ts index b56bbe59d8..1627f3e4ad 100644 --- a/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/crons/jobs/calendar-event-list-fetch.cron.job.ts +++ b/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/crons/jobs/calendar-event-list-fetch.cron.job.ts @@ -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( CalendarEventListFetchJob.name, diff --git a/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/crons/jobs/calendar-events-import.cron.job.ts b/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/crons/jobs/calendar-events-import.cron.job.ts index d4a803560e..db39482706 100644 --- a/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/crons/jobs/calendar-events-import.cron.job.ts +++ b/packages/twenty-server/src/modules/calendar/calendar-event-import-manager/crons/jobs/calendar-events-import.cron.job.ts @@ -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( CalendarEventsImportJob.name, diff --git a/packages/twenty-server/src/modules/messaging/message-folder-manager/services/sync-message-folders.service.spec.ts b/packages/twenty-server/src/modules/messaging/message-folder-manager/services/sync-message-folders.service.spec.ts index 8da989fd20..f3145b5456 100644 --- a/packages/twenty-server/src/modules/messaging/message-folder-manager/services/sync-message-folders.service.spec.ts +++ b/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( diff --git a/packages/twenty-server/src/modules/messaging/message-folder-manager/services/sync-message-folders.service.ts b/packages/twenty-server/src/modules/messaging/message-folder-manager/services/sync-message-folders.service.ts index 59c319dfa1..6d7589351f 100644 --- a/packages/twenty-server/src/modules/messaging/message-folder-manager/services/sync-message-folders.service.ts +++ b/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, ); } diff --git a/packages/twenty-server/src/modules/messaging/message-import-manager/crons/jobs/messaging-message-list-fetch.cron.job.ts b/packages/twenty-server/src/modules/messaging/message-import-manager/crons/jobs/messaging-message-list-fetch.cron.job.ts index 02933aa551..06ed4b6fd2 100644 --- a/packages/twenty-server/src/modules/messaging/message-import-manager/crons/jobs/messaging-message-list-fetch.cron.job.ts +++ b/packages/twenty-server/src/modules/messaging/message-import-manager/crons/jobs/messaging-message-list-fetch.cron.job.ts @@ -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( MessagingMessageListFetchJob.name, diff --git a/packages/twenty-server/src/modules/messaging/message-import-manager/crons/jobs/messaging-messages-import.cron.job.ts b/packages/twenty-server/src/modules/messaging/message-import-manager/crons/jobs/messaging-messages-import.cron.job.ts index 0eda768259..44917ba1b9 100644 --- a/packages/twenty-server/src/modules/messaging/message-import-manager/crons/jobs/messaging-messages-import.cron.job.ts +++ b/packages/twenty-server/src/modules/messaging/message-import-manager/crons/jobs/messaging-messages-import.cron.job.ts @@ -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( MessagingMessagesImportJob.name, diff --git a/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-process-group-email-actions.service.ts b/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-process-group-email-actions.service.ts index 3ea315938e..9c3a68c817 100644 --- a/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-process-group-email-actions.service.ts +++ b/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 { - 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}`,