diff --git a/packages/twenty-server/src/engine/core-modules/auth/services/create-message-channel.service.ts b/packages/twenty-server/src/engine/core-modules/auth/services/create-message-channel.service.ts index 3c7e893250d..05b6d880ba8 100644 --- a/packages/twenty-server/src/engine/core-modules/auth/services/create-message-channel.service.ts +++ b/packages/twenty-server/src/engine/core-modules/auth/services/create-message-channel.service.ts @@ -7,6 +7,7 @@ import { type WorkspaceEntityManager } from 'src/engine/twenty-orm/entity-manage import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager'; import { type ConnectedAccountWorkspaceEntity } from 'src/modules/connected-account/standard-objects/connected-account.workspace-entity'; import { + MessageChannelPendingGroupEmailsAction, MessageChannelSyncStage, MessageChannelSyncStatus, MessageChannelType, @@ -67,6 +68,7 @@ export class CreateMessageChannelService { messageVisibility || MessageChannelVisibility.SHARE_EVERYTHING, syncStatus: MessageChannelSyncStatus.NOT_SYNCED, syncStage: MessageChannelSyncStage.PENDING_CONFIGURATION, + pendingGroupEmailsAction: MessageChannelPendingGroupEmailsAction.NONE, }, {}, manager, diff --git a/packages/twenty-server/src/modules/connected-account/services/imap-smtp-caldav-apis.service.spec.ts b/packages/twenty-server/src/modules/connected-account/services/imap-smtp-caldav-apis.service.spec.ts index a7172d9dced..d8e5190579c 100644 --- a/packages/twenty-server/src/modules/connected-account/services/imap-smtp-caldav-apis.service.spec.ts +++ b/packages/twenty-server/src/modules/connected-account/services/imap-smtp-caldav-apis.service.spec.ts @@ -126,6 +126,7 @@ describe('ImapSmtpCalDavAPIService', () => { isSyncEnabled: true, syncStatus: MessageChannelSyncStatus.NOT_SYNCED, syncStage: MessageChannelSyncStage.PENDING_CONFIGURATION, + pendingGroupEmailsAction: MessageChannelPendingGroupEmailsAction.NONE, syncCursor: '', syncStageStartedAt: null, }; @@ -156,6 +157,7 @@ describe('ImapSmtpCalDavAPIService', () => { isSyncEnabled: true, syncStatus: MessageChannelSyncStatus.NOT_SYNCED, syncStage: MessageChannelSyncStage.PENDING_CONFIGURATION, + pendingGroupEmailsAction: MessageChannelPendingGroupEmailsAction.NONE, syncCursor: '', syncStageStartedAt: null, pendingGroupEmailsAction: MessageChannelPendingGroupEmailsAction.NONE, @@ -281,6 +283,7 @@ describe('ImapSmtpCalDavAPIService', () => { const expectedCalendarChannel = { id: 'mocked-uuid', connectedAccountId: 'mocked-uuid', + handle: 'test@example.com', }; @@ -299,6 +302,7 @@ describe('ImapSmtpCalDavAPIService', () => { isSyncEnabled: false, syncStatus: MessageChannelSyncStatus.NOT_SYNCED, syncStage: MessageChannelSyncStage.PENDING_CONFIGURATION, + pendingGroupEmailsAction: MessageChannelPendingGroupEmailsAction.NONE, syncCursor: '', syncStageStartedAt: null, pendingGroupEmailsAction: MessageChannelPendingGroupEmailsAction.NONE, diff --git a/packages/twenty-server/src/modules/connected-account/services/imap-smtp-caldav-apis.service.ts b/packages/twenty-server/src/modules/connected-account/services/imap-smtp-caldav-apis.service.ts index 5723f9374b2..dfe703719f4 100644 --- a/packages/twenty-server/src/modules/connected-account/services/imap-smtp-caldav-apis.service.ts +++ b/packages/twenty-server/src/modules/connected-account/services/imap-smtp-caldav-apis.service.ts @@ -149,6 +149,7 @@ export class ImapSmtpCalDavAPIService { isSyncEnabled: shouldEnableSync, syncStatus: MessageChannelSyncStatus.NOT_SYNCED, syncStage: MessageChannelSyncStage.PENDING_CONFIGURATION, + pendingGroupEmailsAction: MessageChannelPendingGroupEmailsAction.NONE, syncCursor: '', syncStageStartedAt: null, pendingGroupEmailsAction: MessageChannelPendingGroupEmailsAction.NONE, diff --git a/packages/twenty-server/src/modules/messaging/message-channel-manager/query-hooks/message-channel-update-one.pre-query.hook.ts b/packages/twenty-server/src/modules/messaging/message-channel-manager/query-hooks/message-channel-update-one.pre-query.hook.ts index 4a788e1a47f..99aa1fdefe4 100644 --- a/packages/twenty-server/src/modules/messaging/message-channel-manager/query-hooks/message-channel-update-one.pre-query.hook.ts +++ b/packages/twenty-server/src/modules/messaging/message-channel-manager/query-hooks/message-channel-update-one.pre-query.hook.ts @@ -1,3 +1,5 @@ +import { Logger } from '@nestjs/common'; + import { msg } from '@lingui/core/macro'; import { assertIsDefinedOrThrow } from 'twenty-shared/utils'; import { Not } from 'typeorm'; @@ -33,6 +35,10 @@ const ONGOING_SYNC_STAGES = [ export class MessageChannelUpdateOnePreQueryHook implements WorkspacePreQueryHookInstance { + private readonly logger = new Logger( + MessageChannelUpdateOnePreQueryHook.name, + ); + constructor( private readonly twentyORMGlobalManager: TwentyORMGlobalManager, private readonly messagingProcessGroupEmailActionsService: MessagingProcessGroupEmailActionsService, @@ -103,6 +109,18 @@ export class MessageChannelUpdateOnePreQueryHook ); } + const hasCompletedConfiguration = + messageChannel.syncStage !== + MessageChannelSyncStage.PENDING_CONFIGURATION; + + if (!hasCompletedConfiguration) { + this.logger.log( + `MessageChannelId: ${messageChannel.id} - Skipping pending action for message channel in PENDING_CONFIGURATION state`, + ); + + return payload; + } + const excludeGroupEmailsChanged = payload.data.excludeGroupEmails !== messageChannel.excludeGroupEmails; diff --git a/packages/twenty-server/src/modules/messaging/message-import-manager/crons/commands/messaging-process-group-email-actions.cron.command.ts b/packages/twenty-server/src/modules/messaging/message-import-manager/crons/commands/messaging-process-group-email-actions.cron.command.ts deleted file mode 100644 index 53015e3f1d5..00000000000 --- a/packages/twenty-server/src/modules/messaging/message-import-manager/crons/commands/messaging-process-group-email-actions.cron.command.ts +++ /dev/null @@ -1,35 +0,0 @@ -import { Command, CommandRunner } from 'nest-commander'; - -import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator'; -import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants'; -import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service'; -import { - MESSAGING_PROCESS_GROUP_EMAIL_ACTIONS_CRON_PATTERN, - MessagingProcessGroupEmailActionsCronJob, -} from 'src/modules/messaging/message-import-manager/crons/jobs/messaging-process-group-email-actions.cron.job'; - -@Command({ - name: 'cron:messaging:process-group-email-actions', - description: - 'Starts a cron job to process pending group email actions (deletion or import) for message channels', -}) -export class MessagingProcessGroupEmailActionsCronCommand extends CommandRunner { - constructor( - @InjectMessageQueue(MessageQueue.cronQueue) - private readonly messageQueueService: MessageQueueService, - ) { - super(); - } - - async run(): Promise { - await this.messageQueueService.addCron({ - jobName: MessagingProcessGroupEmailActionsCronJob.name, - data: undefined, - options: { - repeat: { - pattern: MESSAGING_PROCESS_GROUP_EMAIL_ACTIONS_CRON_PATTERN, - }, - }, - }); - } -} diff --git a/packages/twenty-server/src/modules/messaging/message-import-manager/crons/jobs/messaging-process-group-email-actions.cron.job.ts b/packages/twenty-server/src/modules/messaging/message-import-manager/crons/jobs/messaging-process-group-email-actions.cron.job.ts deleted file mode 100644 index 2e636f3a38f..00000000000 --- a/packages/twenty-server/src/modules/messaging/message-import-manager/crons/jobs/messaging-process-group-email-actions.cron.job.ts +++ /dev/null @@ -1,73 +0,0 @@ -import { InjectDataSource, InjectRepository } from '@nestjs/typeorm'; - -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 { 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'; -import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants'; -import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service'; -import { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity'; -import { getWorkspaceSchemaName } from 'src/engine/workspace-datasource/utils/get-workspace-schema-name.util'; -import { MessageChannelPendingGroupEmailsAction } from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity'; -import { - MessagingProcessGroupEmailActionsJob, - type MessagingProcessGroupEmailActionsJobData, -} from 'src/modules/messaging/message-import-manager/jobs/messaging-process-group-email-actions.job'; - -export const MESSAGING_PROCESS_GROUP_EMAIL_ACTIONS_CRON_PATTERN = '0 */2 * * *'; - -@Processor(MessageQueue.cronQueue) -export class MessagingProcessGroupEmailActionsCronJob { - constructor( - @InjectRepository(WorkspaceEntity) - private readonly workspaceRepository: Repository, - @InjectMessageQueue(MessageQueue.messagingQueue) - private readonly messageQueueService: MessageQueueService, - @InjectDataSource() - private readonly coreDataSource: DataSource, - private readonly exceptionHandlerService: ExceptionHandlerService, - ) {} - - @Process(MessagingProcessGroupEmailActionsCronJob.name) - @SentryCronMonitor( - MessagingProcessGroupEmailActionsCronJob.name, - MESSAGING_PROCESS_GROUP_EMAIL_ACTIONS_CRON_PATTERN, - ) - async handle(): Promise { - const activeWorkspaces = await this.workspaceRepository.find({ - where: { - activationStatus: WorkspaceActivationStatus.ACTIVE, - }, - }); - - for (const activeWorkspace of activeWorkspaces) { - try { - const schemaName = getWorkspaceSchemaName(activeWorkspace.id); - - const messageChannels = await this.coreDataSource.query( - `SELECT id FROM ${schemaName}."messageChannel" WHERE "pendingGroupEmailsAction" IN ('${MessageChannelPendingGroupEmailsAction.GROUP_EMAILS_DELETION}', '${MessageChannelPendingGroupEmailsAction.GROUP_EMAILS_IMPORT}')`, - ); - - for (const messageChannel of messageChannels) { - await this.messageQueueService.add( - MessagingProcessGroupEmailActionsJob.name, - { - workspaceId: activeWorkspace.id, - messageChannelId: messageChannel.id, - }, - ); - } - } catch (error) { - this.exceptionHandlerService.captureExceptions([error], { - workspace: { - id: activeWorkspace.id, - }, - }); - } - } - } -} diff --git a/packages/twenty-server/src/modules/messaging/message-import-manager/jobs/messaging-process-group-email-actions.job.ts b/packages/twenty-server/src/modules/messaging/message-import-manager/jobs/messaging-process-group-email-actions.job.ts deleted file mode 100644 index 0c6250afda7..00000000000 --- a/packages/twenty-server/src/modules/messaging/message-import-manager/jobs/messaging-process-group-email-actions.job.ts +++ /dev/null @@ -1,84 +0,0 @@ -import { Logger, Scope } from '@nestjs/common'; - -import { Process } from 'src/engine/core-modules/message-queue/decorators/process.decorator'; -import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.decorator'; -import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants'; -import { TwentyORMManager } from 'src/engine/twenty-orm/twenty-orm.manager'; -import { - MessageChannelPendingGroupEmailsAction, - type MessageChannelWorkspaceEntity, -} from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity'; -import { MessagingProcessGroupEmailActionsService } from 'src/modules/messaging/message-import-manager/services/messaging-process-group-email-actions.service'; - -export type MessagingProcessGroupEmailActionsJobData = { - workspaceId: string; - messageChannelId: string; -}; - -@Processor({ - queueName: MessageQueue.messagingQueue, - scope: Scope.REQUEST, -}) -export class MessagingProcessGroupEmailActionsJob { - private readonly logger = new Logger( - MessagingProcessGroupEmailActionsJob.name, - ); - - constructor( - private readonly twentyORMManager: TwentyORMManager, - private readonly messagingProcessGroupEmailActionsService: MessagingProcessGroupEmailActionsService, - ) {} - - @Process(MessagingProcessGroupEmailActionsJob.name) - async handle(data: MessagingProcessGroupEmailActionsJobData): Promise { - const { workspaceId, messageChannelId } = data; - - this.logger.log( - `Processing pending group email action for message channel ${messageChannelId} in workspace ${workspaceId}`, - ); - - const messageChannelRepository = - await this.twentyORMManager.getRepository( - 'messageChannel', - ); - - const messageChannel = await messageChannelRepository.findOne({ - where: { - id: messageChannelId, - }, - }); - - if (!messageChannel) { - this.logger.warn( - `Message channel ${messageChannelId} not found in workspace ${workspaceId}`, - ); - - return; - } - - if ( - messageChannel.pendingGroupEmailsAction === - MessageChannelPendingGroupEmailsAction.NONE || - !messageChannel.pendingGroupEmailsAction - ) { - this.logger.log( - `Message channel ${messageChannelId} no longer has a pending action, skipping`, - ); - - return; - } - - try { - await this.messagingProcessGroupEmailActionsService.processGroupEmailActions( - messageChannel, - workspaceId, - ); - } catch (error) { - this.logger.error( - `Error processing group email actions for message channel ${messageChannelId} in workspace ${workspaceId}: ${error.message}`, - error.stack, - ); - throw error; - } - } -} diff --git a/packages/twenty-server/src/modules/messaging/message-import-manager/messaging-import-manager.module.ts b/packages/twenty-server/src/modules/messaging/message-import-manager/messaging-import-manager.module.ts index 344e8f3acd9..aec2e14b860 100644 --- a/packages/twenty-server/src/modules/messaging/message-import-manager/messaging-import-manager.module.ts +++ b/packages/twenty-server/src/modules/messaging/message-import-manager/messaging-import-manager.module.ts @@ -19,13 +19,11 @@ import { MessagingMessageListFetchCronCommand } from 'src/modules/messaging/mess import { MessagingMessagesImportCronCommand } from 'src/modules/messaging/message-import-manager/crons/commands/messaging-messages-import.cron.command'; import { MessagingOngoingStaleCronCommand } from 'src/modules/messaging/message-import-manager/crons/commands/messaging-ongoing-stale.cron.command'; import { MessagingProcessFolderActionsCronCommand } from 'src/modules/messaging/message-import-manager/crons/commands/messaging-process-folder-actions.cron.command'; -import { MessagingProcessGroupEmailActionsCronCommand } from 'src/modules/messaging/message-import-manager/crons/commands/messaging-process-group-email-actions.cron.command'; import { MessagingRelaunchFailedMessageChannelsCronCommand } from 'src/modules/messaging/message-import-manager/crons/commands/messaging-relaunch-failed-message-channels.cron.command'; import { MessagingMessageListFetchCronJob } from 'src/modules/messaging/message-import-manager/crons/jobs/messaging-message-list-fetch.cron.job'; import { MessagingMessagesImportCronJob } from 'src/modules/messaging/message-import-manager/crons/jobs/messaging-messages-import.cron.job'; import { MessagingOngoingStaleCronJob } from 'src/modules/messaging/message-import-manager/crons/jobs/messaging-ongoing-stale.cron.job'; import { MessagingProcessFolderActionsCronJob } from 'src/modules/messaging/message-import-manager/crons/jobs/messaging-process-folder-actions.cron.job'; -import { MessagingProcessGroupEmailActionsCronJob } from 'src/modules/messaging/message-import-manager/crons/jobs/messaging-process-group-email-actions.cron.job'; import { MessagingRelaunchFailedMessageChannelsCronJob } from 'src/modules/messaging/message-import-manager/crons/jobs/messaging-relaunch-failed-message-channels.cron.job'; import { MessagingGmailDriverModule } from 'src/modules/messaging/message-import-manager/drivers/gmail/messaging-gmail-driver.module'; import { MessagingIMAPDriverModule } from 'src/modules/messaging/message-import-manager/drivers/imap/messaging-imap-driver.module'; @@ -37,7 +35,6 @@ import { MessagingMessageListFetchJob } from 'src/modules/messaging/message-impo import { MessagingMessagesImportJob } from 'src/modules/messaging/message-import-manager/jobs/messaging-messages-import.job'; import { MessagingOngoingStaleJob } from 'src/modules/messaging/message-import-manager/jobs/messaging-ongoing-stale.job'; import { MessagingProcessFolderActionsJob } from 'src/modules/messaging/message-import-manager/jobs/messaging-process-folder-actions.job'; -import { MessagingProcessGroupEmailActionsJob } from 'src/modules/messaging/message-import-manager/jobs/messaging-process-group-email-actions.job'; import { MessagingRelaunchFailedMessageChannelJob } from 'src/modules/messaging/message-import-manager/jobs/messaging-relaunch-failed-message-channel.job'; import { MessagingMessageImportManagerMessageChannelListener } from 'src/modules/messaging/message-import-manager/listeners/messaging-import-manager-message-channel.listener'; import { MessagingAccountAuthenticationService } from 'src/modules/messaging/message-import-manager/services/messaging-account-authentication.service'; @@ -87,20 +84,17 @@ import { MessagingMonitoringModule } from 'src/modules/messaging/monitoring/mess MessagingMessagesImportCronCommand, MessagingOngoingStaleCronCommand, MessagingProcessFolderActionsCronCommand, - MessagingProcessGroupEmailActionsCronCommand, MessagingRelaunchFailedMessageChannelsCronCommand, MessagingSingleMessageImportCommand, MessagingMessageListFetchJob, MessagingMessagesImportJob, MessagingOngoingStaleJob, MessagingProcessFolderActionsJob, - MessagingProcessGroupEmailActionsJob, MessagingRelaunchFailedMessageChannelJob, MessagingMessageListFetchCronJob, MessagingMessagesImportCronJob, MessagingOngoingStaleCronJob, MessagingProcessFolderActionsCronJob, - MessagingProcessGroupEmailActionsCronJob, MessagingRelaunchFailedMessageChannelsCronJob, MessagingAddSingleMessageToCacheForImportJob, MessagingMessageImportManagerMessageChannelListener, diff --git a/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-message-list-fetch.service.spec.ts b/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-message-list-fetch.service.spec.ts index 1899591efdc..6c6c14a77f4 100644 --- a/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-message-list-fetch.service.spec.ts +++ b/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-message-list-fetch.service.spec.ts @@ -16,6 +16,7 @@ import { MessagingGetMessageListService } from 'src/modules/messaging/message-im import { MessageImportExceptionHandlerService } from 'src/modules/messaging/message-import-manager/services/messaging-import-exception-handler.service'; import { MessagingMessageListFetchService } from 'src/modules/messaging/message-import-manager/services/messaging-message-list-fetch.service'; import { MessagingMessagesImportService } from 'src/modules/messaging/message-import-manager/services/messaging-messages-import.service'; +import { MessagingProcessGroupEmailActionsService } from 'src/modules/messaging/message-import-manager/services/messaging-process-group-email-actions.service'; describe('MessagingMessageListFetchService', () => { let messagingMessageListFetchService: MessagingMessageListFetchService; @@ -231,6 +232,12 @@ describe('MessagingMessageListFetchService', () => { syncMessageFolders: jest.fn().mockResolvedValue(undefined), }, }, + { + provide: MessagingProcessGroupEmailActionsService, + useValue: { + processGroupEmailActions: jest.fn().mockResolvedValue(undefined), + }, + }, ], }).compile(); diff --git a/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-message-list-fetch.service.ts b/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-message-list-fetch.service.ts index bd3e32e3b05..067742bfe16 100644 --- a/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-message-list-fetch.service.ts +++ b/packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-message-list-fetch.service.ts @@ -30,6 +30,7 @@ import { MessageImportSyncStep, } from 'src/modules/messaging/message-import-manager/services/messaging-import-exception-handler.service'; import { MessagingMessagesImportService } from 'src/modules/messaging/message-import-manager/services/messaging-messages-import.service'; +import { MessagingProcessGroupEmailActionsService } from 'src/modules/messaging/message-import-manager/services/messaging-process-group-email-actions.service'; const ONE_WEEK_IN_MILLISECONDS = 7 * 24 * 60 * 60 * 1000; @@ -48,6 +49,7 @@ export class MessagingMessageListFetchService { private readonly messagingMessagesImportService: MessagingMessagesImportService, private readonly messagingAccountAuthenticationService: MessagingAccountAuthenticationService, private readonly syncMessageFoldersService: SyncMessageFoldersService, + private readonly messagingProcessGroupEmailActionsService: MessagingProcessGroupEmailActionsService, ) {} public async processMessageListFetch( @@ -62,7 +64,12 @@ export class MessagingMessageListFetchService { MessageChannelPendingGroupEmailsAction.GROUP_EMAILS_IMPORT ) { this.logger.log( - `messageChannelId: ${messageChannel.id} Skipping message list fetch due to pending group emails action: ${messageChannel.pendingGroupEmailsAction}`, + `messageChannelId: ${messageChannel.id} Processing pending group emails action before message list fetch: ${messageChannel.pendingGroupEmailsAction}`, + ); + + await this.messagingProcessGroupEmailActionsService.processGroupEmailActions( + messageChannel, + workspaceId, ); return;