vynte-infra/twenty
2
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

Message channel change 3 fix (#15986)

Browse Source
This commit is contained in:
neo773
2025-11-23 20:56:02 +01:00
committed by GitHub
parent 061cc897af
commit 44c0dcde5b
10 changed files with 40 additions and 199 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/twenty-server/src/engine/core-modules/auth/services/create-message-channel.service.ts
+2
View File
@@ -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,
packages/twenty-server/src/modules/connected-account/services/imap-smtp-caldav-apis.service.spec.ts
+4
View File
@@ -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: '[email protected]',
};
@@ -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,
packages/twenty-server/src/modules/connected-account/services/imap-smtp-caldav-apis.service.ts
+1
View File
@@ -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,
packages/twenty-server/src/modules/messaging/message-channel-manager/query-hooks/message-channel-update-one.pre-query.hook.ts
+18
View File
@@ -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;
packages/twenty-server/src/modules/messaging/message-import-manager/crons/commands/messaging-process-group-email-actions.cron.command.ts
-35
View File
@@ -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<void> {
await this.messageQueueService.addCron<undefined>({
jobName: MessagingProcessGroupEmailActionsCronJob.name,
data: undefined,
options: {
repeat: {
pattern: MESSAGING_PROCESS_GROUP_EMAIL_ACTIONS_CRON_PATTERN,
},
},
});
}
}
packages/twenty-server/src/modules/messaging/message-import-manager/crons/jobs/messaging-process-group-email-actions.cron.job.ts
-73
View File
@@ -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<WorkspaceEntity>,
@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<void> {
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<MessagingProcessGroupEmailActionsJobData>(
MessagingProcessGroupEmailActionsJob.name,
{
workspaceId: activeWorkspace.id,
messageChannelId: messageChannel.id,
},
);
}
} catch (error) {
this.exceptionHandlerService.captureExceptions([error], {
workspace: {
id: activeWorkspace.id,
},
});
}
}
}
}
packages/twenty-server/src/modules/messaging/message-import-manager/jobs/messaging-process-group-email-actions.job.ts
-84
View File
@@ -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<void> {
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<MessageChannelWorkspaceEntity>(
'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;
}
}
}
packages/twenty-server/src/modules/messaging/message-import-manager/messaging-import-manager.module.ts
-6
View File
@@ -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,
packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-message-list-fetch.service.spec.ts
+7
View File
@@ -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();
packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-message-list-fetch.service.ts
+8 -1
View File
@@ -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;