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

Add scripts to relaunch message and calendar channels (#14579)

Browse Source 
## Context

1. We are debugging the experience on messaging and calendar sync. I'm
adding two scripts to relaunch messaging and calendar channels
2. We are seeing errors in Sentry regarding microsoft drivers errors. I
was about to fix them but I'm refactoring / simplifying a bit the logic
first. It will be a betters starting point
This commit is contained in:
Charles Bochet
2025-09-19 12:31:19 +02:00
committed by GitHub
parent 4a6ad48a62
commit 349b15e81f
10 changed files with 204 additions and 117 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/twenty-server/src/engine/workspace-event-emitter/workspace-event-emitter.ts
+1 -1
View File
@@ -15,8 +15,8 @@ import { ObjectRecordUpdateEvent } from 'src/engine/core-modules/event-emitter/t
import { objectRecordChangedValues } from 'src/engine/core-modules/event-emitter/utils/object-record-changed-values';
import { type ObjectMetadataItemWithFieldMaps } from 'src/engine/metadata-modules/types/object-metadata-item-with-field-maps';
import { type CustomEventName } from 'src/engine/workspace-event-emitter/types/custom-event-name.type';
import { STANDARD_OBJECT_IDS } from 'src/engine/workspace-manager/workspace-sync-metadata/constants/standard-object-ids';
import { computeEventName } from 'src/engine/workspace-event-emitter/utils/compute-event-name';
import { STANDARD_OBJECT_IDS } from 'src/engine/workspace-manager/workspace-sync-metadata/constants/standard-object-ids';
type ActionEventMap<T> = {
[DatabaseEventAction.CREATED]: ObjectRecordCreateEvent<T>;
packages/twenty-server/src/modules/calendar/calendar-event-import-manager/calendar-event-import-manager.module.ts
+2
View File
@@ -10,6 +10,7 @@ import { ObjectMetadataRepositoryModule } from 'src/engine/object-metadata-repos
import { WorkspaceDataSourceModule } from 'src/engine/workspace-datasource/workspace-datasource.module';
import { BlocklistWorkspaceEntity } from 'src/modules/blocklist/standard-objects/blocklist.workspace-entity';
import { CalendarEventCleanerModule } from 'src/modules/calendar/calendar-event-cleaner/calendar-event-cleaner.module';
import { CalendarRelaunchFailedCalendarChannelsCommand } from 'src/modules/calendar/calendar-event-import-manager/commands/calendar-relaunch-failed-calendar-channels.command';
import { CalendarEventListFetchCronCommand } from 'src/modules/calendar/calendar-event-import-manager/crons/commands/calendar-event-list-fetch.cron.command';
import { CalendarEventsImportCronCommand } from 'src/modules/calendar/calendar-event-import-manager/crons/commands/calendar-import.cron.command';
import { CalendarOngoingStaleCronCommand } from 'src/modules/calendar/calendar-event-import-manager/crons/commands/calendar-ongoing-stale.cron.command';
@@ -64,6 +65,7 @@ import { RefreshTokensManagerModule } from 'src/modules/connected-account/refres
CalendarEventsImportJob,
CalendarOngoingStaleCronJob,
CalendarOngoingStaleCronCommand,
CalendarRelaunchFailedCalendarChannelsCommand,
CalendarOngoingStaleJob,
],
exports: [
packages/twenty-server/src/modules/calendar/calendar-event-import-manager/commands/calendar-relaunch-failed-calendar-channels.command.ts
+85
View File
@@ -0,0 +1,85 @@
import { InjectRepository } from '@nestjs/typeorm';
import { Command } from 'nest-commander';
import { Repository } from 'typeorm';
import {
ActiveOrSuspendedWorkspacesMigrationCommandRunner,
type RunOnWorkspaceArgs,
} from 'src/database/commands/command-runners/active-or-suspended-workspaces-migration.command-runner';
import { Workspace } from 'src/engine/core-modules/workspace/workspace.entity';
import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager';
import {
CalendarChannelSyncStage,
CalendarChannelSyncStatus,
CalendarChannelWorkspaceEntity,
} from 'src/modules/calendar/common/standard-objects/calendar-channel.workspace-entity';
import { AccountsToReconnectService } from 'src/modules/connected-account/services/accounts-to-reconnect.service';
@Command({
name: 'calendar:relaunch-failed-calendar-channels',
description: 'Relaunch failed message channels',
})
export class CalendarRelaunchFailedCalendarChannelsCommand extends ActiveOrSuspendedWorkspacesMigrationCommandRunner {
constructor(
@InjectRepository(Workspace)
protected readonly workspaceRepository: Repository<Workspace>,
protected readonly twentyORMGlobalManager: TwentyORMGlobalManager,
protected readonly accountsToReconnectService: AccountsToReconnectService,
) {
super(workspaceRepository, twentyORMGlobalManager);
}
override async runOnWorkspace({
workspaceId,
options,
}: RunOnWorkspaceArgs): Promise<void> {
try {
const calendarChannelRepository =
await this.twentyORMGlobalManager.getRepositoryForWorkspace<CalendarChannelWorkspaceEntity>(
workspaceId,
'calendarChannel',
{ shouldBypassPermissionChecks: true },
);
const failedCalendarChannels = await calendarChannelRepository.find({
where: {
syncStage: CalendarChannelSyncStage.FAILED,
},
relations: {
connectedAccount: {
accountOwner: true,
},
},
});
if (!options.dryRun && failedCalendarChannels.length > 0) {
await calendarChannelRepository.update(
failedCalendarChannels.map(({ id }) => id),
{
syncStage:
CalendarChannelSyncStage.CALENDAR_EVENT_LIST_FETCH_PENDING,
syncStatus: CalendarChannelSyncStatus.ACTIVE,
},
);
for (const failedCalendarChannel of failedCalendarChannels) {
await this.accountsToReconnectService.removeAccountToReconnect(
failedCalendarChannel.connectedAccount.accountOwner.userId,
failedCalendarChannel.connectedAccountId,
workspaceId,
);
}
}
this.logger.log(
`${options.dryRun ? ' (DRY RUN): ' : ''}Relaunched ${failedCalendarChannels.length} failed calendar channels`,
);
} catch (error) {
this.logger.error(
'Error while relaunching failed message channels',
error,
);
}
}
}
packages/twenty-server/src/modules/calendar/calendar-event-import-manager/crons/jobs/calendar-event-list-fetch.cron.job.ts
+1 -1
View File
@@ -51,7 +51,7 @@ export class CalendarEventListFetchCronJob {
const schemaName = getWorkspaceSchemaName(activeWorkspace.id);
const calendarChannels = await this.coreDataSource.query(
`SELECT * FROM ${schemaName}."calendarChannel" WHERE "isSyncEnabled" = true AND "syncStage" IN ('${CalendarChannelSyncStage.FULL_CALENDAR_EVENT_LIST_FETCH_PENDING}', '${CalendarChannelSyncStage.PARTIAL_CALENDAR_EVENT_LIST_FETCH_PENDING}')`,
`SELECT * FROM ${schemaName}."calendarChannel" WHERE "isSyncEnabled" = true AND "syncStage" IN ('${CalendarChannelSyncStage.FULL_CALENDAR_EVENT_LIST_FETCH_PENDING}', '${CalendarChannelSyncStage.PARTIAL_CALENDAR_EVENT_LIST_FETCH_PENDING}', '${CalendarChannelSyncStage.CALENDAR_EVENT_LIST_FETCH_PENDING}')`,
);
for (const calendarChannel of calendarChannels) {
packages/twenty-server/src/modules/messaging/message-import-manager/commands/messaging-relaunch-failed-message-channels.command.ts
+84
View File
@@ -0,0 +1,84 @@
import { InjectRepository } from '@nestjs/typeorm';
import { Command } from 'nest-commander';
import { Repository } from 'typeorm';
import {
ActiveOrSuspendedWorkspacesMigrationCommandRunner,
type RunOnWorkspaceArgs,
} from 'src/database/commands/command-runners/active-or-suspended-workspaces-migration.command-runner';
import { Workspace } from 'src/engine/core-modules/workspace/workspace.entity';
import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager';
import { AccountsToReconnectService } from 'src/modules/connected-account/services/accounts-to-reconnect.service';
import {
MessageChannelSyncStage,
MessageChannelSyncStatus,
MessageChannelWorkspaceEntity,
} from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity';
@Command({
name: 'messaging:relaunch-failed-message-channels',
description: 'Relaunch failed message channels',
})
export class MessagingRelaunchFailedMessageChannelsCommand extends ActiveOrSuspendedWorkspacesMigrationCommandRunner {
constructor(
@InjectRepository(Workspace)
protected readonly workspaceRepository: Repository<Workspace>,
protected readonly twentyORMGlobalManager: TwentyORMGlobalManager,
protected readonly accountsToReconnectService: AccountsToReconnectService,
) {
super(workspaceRepository, twentyORMGlobalManager);
}
override async runOnWorkspace({
workspaceId,
options,
}: RunOnWorkspaceArgs): Promise<void> {
try {
const messageChannelRepository =
await this.twentyORMGlobalManager.getRepositoryForWorkspace<MessageChannelWorkspaceEntity>(
workspaceId,
'messageChannel',
{ shouldBypassPermissionChecks: true },
);
const failedMessageChannels = await messageChannelRepository.find({
where: {
syncStage: MessageChannelSyncStage.FAILED,
},
relations: {
connectedAccount: {
accountOwner: true,
},
},
});
if (!options.dryRun && failedMessageChannels.length > 0) {
await messageChannelRepository.update(
failedMessageChannels.map(({ id }) => id),
{
syncStage: MessageChannelSyncStage.MESSAGE_LIST_FETCH_PENDING,
syncStatus: MessageChannelSyncStatus.ACTIVE,
},
);
for (const failedMessageChannel of failedMessageChannels) {
await this.accountsToReconnectService.removeAccountToReconnect(
failedMessageChannel.connectedAccount.accountOwner.userId,
failedMessageChannel.connectedAccountId,
workspaceId,
);
}
}
this.logger.log(
`${options.dryRun ? ' (DRY RUN): ' : ''}Relaunched ${failedMessageChannels.length} failed message channels`,
);
} catch (error) {
this.logger.error(
'Error while relaunching failed message channels',
error,
);
}
}
}
packages/twenty-server/src/modules/messaging/message-import-manager/crons/jobs/messaging-message-list-fetch.cron.job.ts
+1 -1
View File
@@ -50,7 +50,7 @@ export class MessagingMessageListFetchCronJob {
// TODO: deprecate looking for FULL_MESSAGE_LIST_FETCH_PENDING as we introduce MESSAGE_LIST_FETCH_PENDING
const messageChannels = await this.coreDataSource.query(
`SELECT * FROM ${schemaName}."messageChannel" WHERE "isSyncEnabled" = true AND "syncStage" IN ('${MessageChannelSyncStage.PARTIAL_MESSAGE_LIST_FETCH_PENDING}', '${MessageChannelSyncStage.FULL_MESSAGE_LIST_FETCH_PENDING}')`,
`SELECT * FROM ${schemaName}."messageChannel" WHERE "isSyncEnabled" = true AND "syncStage" IN ('${MessageChannelSyncStage.PARTIAL_MESSAGE_LIST_FETCH_PENDING}', '${MessageChannelSyncStage.FULL_MESSAGE_LIST_FETCH_PENDING}', '${MessageChannelSyncStage.MESSAGE_LIST_FETCH_PENDING}')`,
);
for (const messageChannel of messageChannels) {
packages/twenty-server/src/modules/messaging/message-import-manager/drivers/microsoft/services/microsoft-fetch-by-batch.service.ts
+4 -41
View File
@@ -1,20 +1,13 @@
import { Injectable } from '@nestjs/common';
import { type ConnectedAccountWorkspaceEntity } from 'src/modules/connected-account/standard-objects/connected-account.workspace-entity';
import {
MessageImportDriverException,
MessageImportDriverExceptionCode,
} from 'src/modules/messaging/message-import-manager/drivers/exceptions/message-import-driver.exception';
import { MicrosoftClientProvider } from 'src/modules/messaging/message-import-manager/drivers/microsoft/providers/microsoft-client.provider';
import { type MicrosoftGraphBatchResponse } from 'src/modules/messaging/message-import-manager/drivers/microsoft/services/microsoft-get-messages.interface';
import { MicrosoftHandleErrorService } from 'src/modules/messaging/message-import-manager/drivers/microsoft/services/microsoft-handle-error.service';
import { isAccessTokenRefreshingError } from 'src/modules/messaging/message-import-manager/drivers/microsoft/utils/is-access-token-refreshing-error.utils';
@Injectable()
export class MicrosoftFetchByBatchService {
constructor(
private readonly microsoftClientProvider: MicrosoftClientProvider,
private readonly microsoftHandleErrorService: MicrosoftHandleErrorService,
) {}
async fetchAllByBatches(
@@ -49,23 +42,11 @@ export class MicrosoftFetchByBatchService {
},
}));
try {
const batchResponse = await client
.api('/$batch')
.post({ requests: batchRequests });
const batchResponse = await client
.api('/$batch')
.post({ requests: batchRequests });
batchResponses.push(batchResponse);
} catch (error) {
if (isAccessTokenRefreshingError(error?.body)) {
throw new MessageImportDriverException(
error.message,
MessageImportDriverExceptionCode.CLIENT_NOT_AVAILABLE,
);
}
this.microsoftHandleErrorService.handleMicrosoftMessageFetchByBatchError(
error,
);
}
batchResponses.push(batchResponse);
}
return {
@@ -73,22 +54,4 @@ export class MicrosoftFetchByBatchService {
batchResponses,
};
}
/**
* Microsoft client.api.post sometimes throws (hard to catch) temporary errors like this one:
*
* {
* statusCode: 200,
* code: "SyntaxError",
* requestId: null,
* date: "2025-05-14T11:43:02.024Z",
* body: "SyntaxError: Unexpected token < in JSON at position 19341",
* headers: {
* },
* }
*/
// eslint-disable-next-line @typescript-eslint/no-explicit-any
private _isTemporaryError(error: any): boolean {
return error?.body?.includes('Unexpected token < in JSON at position');
}
}
packages/twenty-server/src/modules/messaging/message-import-manager/drivers/microsoft/services/microsoft-handle-error.service.ts
+16 -70
View File
@@ -4,6 +4,7 @@ import {
MessageImportDriverException,
MessageImportDriverExceptionCode,
} from 'src/modules/messaging/message-import-manager/drivers/exceptions/message-import-driver.exception';
import { isAccessTokenRefreshingError } from 'src/modules/messaging/message-import-manager/drivers/microsoft/utils/is-access-token-refreshing-error.utils';
import { isMicrosoftClientTemporaryError } from 'src/modules/messaging/message-import-manager/drivers/microsoft/utils/is-temporary-error.utils';
import { parseMicrosoftMessagesImportError } from 'src/modules/messaging/message-import-manager/drivers/microsoft/utils/parse-microsoft-messages-import.util';
@@ -12,10 +13,20 @@ export class MicrosoftHandleErrorService {
private readonly logger = new Logger(MicrosoftHandleErrorService.name);
// eslint-disable-next-line @typescript-eslint/no-explicit-any
public handleMicrosoftMessageFetchByBatchError(error: any): void {
// TODO: remove this log once we catch better the error codes
this.logger.error(`Error temporary (${error.code}) fetching messages`);
this.logger.log(error);
public handleMicrosoftGetMessageListError(error: any): void {
this.logger.log(`Error fetching message list`, error);
throw parseMicrosoftMessagesImportError(error);
}
// eslint-disable-next-line @typescript-eslint/no-explicit-any
public handleMicrosoftGetMessagesError(error: any): void {
this.logger.log(`Error fetching messages`, error);
if (isAccessTokenRefreshingError(error?.body)) {
throw new MessageImportDriverException(
error.message,
MessageImportDriverExceptionCode.CLIENT_NOT_AVAILABLE,
);
}
const isBodyString = error.body && typeof error.body === 'string';
const isTemporaryError =
@@ -28,71 +39,6 @@ export class MicrosoftHandleErrorService {
);
}
if (!error.statusCode) {
throw new MessageImportDriverException(
`Microsoft Graph API unknown error: ${error}`,
MessageImportDriverExceptionCode.UNKNOWN,
);
}
const exception = parseMicrosoftMessagesImportError(error);
if (exception) {
throw exception;
}
throw new MessageImportDriverException(
`Microsoft driver error: ${error.message}`,
MessageImportDriverExceptionCode.UNKNOWN,
);
}
// eslint-disable-next-line @typescript-eslint/no-explicit-any
public handleMicrosoftGetMessageListError(error: any): void {
if (!error.statusCode) {
throw new MessageImportDriverException(
`Microsoft Graph API unknown error: ${error}`,
MessageImportDriverExceptionCode.UNKNOWN,
);
}
const exception = parseMicrosoftMessagesImportError(error);
if (exception) {
throw exception;
}
throw new MessageImportDriverException(
`Microsoft driver error: ${error.message}`,
MessageImportDriverExceptionCode.UNKNOWN,
);
}
// eslint-disable-next-line @typescript-eslint/no-explicit-any
public handleMicrosoftGetMessagesError(error: any): void {
if (
error instanceof MessageImportDriverException &&
error.code === MessageImportDriverExceptionCode.CLIENT_NOT_AVAILABLE
) {
throw error;
}
if (!error.statusCode) {
throw new MessageImportDriverException(
`Microsoft Graph API unknown error: ${error}`,
MessageImportDriverExceptionCode.UNKNOWN,
);
}
const exception = parseMicrosoftMessagesImportError(error);
if (exception) {
throw exception;
}
throw new MessageImportDriverException(
`Microsoft driver error: ${error.message}`,
MessageImportDriverExceptionCode.UNKNOWN,
);
throw parseMicrosoftMessagesImportError(error);
}
}
packages/twenty-server/src/modules/messaging/message-import-manager/drivers/microsoft/utils/parse-microsoft-messages-import.util.ts
+5 -2
View File
@@ -7,7 +7,7 @@ export const parseMicrosoftMessagesImportError = (error: {
statusCode: number;
message?: string;
code?: string;
}): MessageImportDriverException | undefined => {
}): MessageImportDriverException => {
if (error.statusCode === 401) {
return new MessageImportDriverException(
'Unauthorized access to Microsoft Graph API',
@@ -51,5 +51,8 @@ export const parseMicrosoftMessagesImportError = (error: {
);
}
return undefined;
return new MessageImportDriverException(
`Microsoft Graph API unknown error: ${error} with status code ${error.statusCode}`,
MessageImportDriverExceptionCode.UNKNOWN,
);
};
packages/twenty-server/src/modules/messaging/message-import-manager/messaging-import-manager.module.ts
+5 -1
View File
@@ -7,10 +7,13 @@ import { DataSourceEntity } from 'src/engine/metadata-modules/data-source/data-s
import { ObjectMetadataEntity } from 'src/engine/metadata-modules/object-metadata/object-metadata.entity';
import { WorkspaceDataSourceModule } from 'src/engine/workspace-datasource/workspace-datasource.module';
import { WorkspaceEventEmitterModule } from 'src/engine/workspace-event-emitter/workspace-event-emitter.module';
import { ConnectedAccountModule } from 'src/modules/connected-account/connected-account.module';
import { EmailAliasManagerModule } from 'src/modules/connected-account/email-alias-manager/email-alias-manager.module';
import { RefreshTokensManagerModule } from 'src/modules/connected-account/refresh-tokens-manager/connected-account-refresh-tokens-manager.module';
import { MessagingCommonModule } from 'src/modules/messaging/common/messaging-common.module';
import { MessagingMessageCleanerModule } from 'src/modules/messaging/message-cleaner/messaging-message-cleaner.module';
import { MessagingFolderSyncManagerModule } from 'src/modules/messaging/message-folder-manager/messaging-folder-sync-manager.module';
import { MessagingRelaunchFailedMessageChannelsCommand } from 'src/modules/messaging/message-import-manager/commands/messaging-relaunch-failed-message-channels.command';
import { MessagingSingleMessageImportCommand } from 'src/modules/messaging/message-import-manager/commands/messaging-single-message-import.command';
import { MessagingMessageListFetchCronCommand } from 'src/modules/messaging/message-import-manager/crons/commands/messaging-message-list-fetch.cron.command';
import { MessagingMessagesImportCronCommand } from 'src/modules/messaging/message-import-manager/crons/commands/messaging-messages-import.cron.command';
@@ -39,7 +42,6 @@ import { MessagingMessagesImportService } from 'src/modules/messaging/message-im
import { MessagingSaveMessagesAndEnqueueContactCreationService } from 'src/modules/messaging/message-import-manager/services/messaging-save-messages-and-enqueue-contact-creation.service';
import { MessagingSendMessageService } from 'src/modules/messaging/message-import-manager/services/messaging-send-message.service';
import { MessageParticipantManagerModule } from 'src/modules/messaging/message-participant-manager/message-participant-manager.module';
import { MessagingFolderSyncManagerModule } from 'src/modules/messaging/message-folder-manager/messaging-folder-sync-manager.module';
import { MessagingMonitoringModule } from 'src/modules/messaging/monitoring/messaging-monitoring.module';
@Module({
imports: [
@@ -62,12 +64,14 @@ import { MessagingMonitoringModule } from 'src/modules/messaging/monitoring/mess
MessagingMonitoringModule,
MessagingMessageCleanerModule,
WorkspaceEventEmitterModule,
ConnectedAccountModule,
],
providers: [
MessagingMessageListFetchCronCommand,
MessagingMessagesImportCronCommand,
MessagingOngoingStaleCronCommand,
MessagingSingleMessageImportCommand,
MessagingRelaunchFailedMessageChannelsCommand,
MessagingMessageListFetchJob,
MessagingMessagesImportJob,
MessagingOngoingStaleJob,