houseassassin/twenty
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix Message/Calendar channel stuck in SCHEDULED syncStage (#16326)

Browse Source 
As per title
This commit is contained in:
Charles Bochet
2025-12-04 16:01:11 +01:00
committed by GitHub
parent ce082cd97a
commit 68a9ef57f0
7 changed files with 61 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/twenty-server/src/modules/calendar/calendar-event-import-manager/crons/jobs/calendar-event-list-fetch.cron.job.ts
+4 -1
View File
@@ -50,8 +50,11 @@ export class CalendarEventListFetchCronJob {
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}' WHERE "isSyncEnabled" = true AND "syncStage" = '${CalendarChannelSyncStage.CALENDAR_EVENT_LIST_FETCH_PENDING}' RETURNING *`,
`UPDATE ${schemaName}."calendarChannel" SET "syncStage" = '${CalendarChannelSyncStage.CALENDAR_EVENT_LIST_FETCH_SCHEDULED}', "syncStageStartedAt" = '${now}'
WHERE "isSyncEnabled" = true AND "syncStage" = '${CalendarChannelSyncStage.CALENDAR_EVENT_LIST_FETCH_PENDING}' RETURNING *`,
);
for (const calendarChannel of calendarChannels) {
packages/twenty-server/src/modules/calendar/calendar-event-import-manager/crons/jobs/calendar-events-import.cron.job.ts
+4 -1
View File
@@ -48,8 +48,11 @@ export class CalendarEventsImportCronJob {
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}' WHERE "isSyncEnabled" = true AND "syncStage" = '${CalendarChannelSyncStage.CALENDAR_EVENTS_IMPORT_PENDING}' RETURNING *`,
`UPDATE ${schemaName}."calendarChannel" SET "syncStage" = '${CalendarChannelSyncStage.CALENDAR_EVENTS_IMPORT_SCHEDULED}', "syncStageStartedAt" = '${now}'
WHERE "isSyncEnabled" = true AND "syncStage" = '${CalendarChannelSyncStage.CALENDAR_EVENTS_IMPORT_PENDING}' RETURNING *`,
);
for (const calendarChannel of calendarChannels) {
packages/twenty-server/src/modules/messaging/common/services/message-channel-sync-status.service.ts
+40 -1
View File
@@ -133,6 +133,27 @@ export class MessageChannelSyncStatusService {
});
}
public async markAsMessagesListFetchScheduled(
messageChannelIds: string[],
workspaceId: string,
) {
if (!messageChannelIds.length) {
return;
}
const messageChannelRepository =
await this.twentyORMGlobalManager.getRepositoryForWorkspace<MessageChannelWorkspaceEntity>(
workspaceId,
'messageChannel',
);
await messageChannelRepository.update(messageChannelIds, {
syncStage: MessageChannelSyncStage.MESSAGE_LIST_FETCH_SCHEDULED,
syncStatus: MessageChannelSyncStatus.ONGOING,
syncStageStartedAt: new Date().toISOString(),
});
}
public async markAsMessagesListFetchOngoing(
messageChannelIds: string[],
workspaceId: string,
@@ -150,7 +171,6 @@ export class MessageChannelSyncStatusService {
await messageChannelRepository.update(messageChannelIds, {
syncStage: MessageChannelSyncStage.MESSAGE_LIST_FETCH_ONGOING,
syncStatus: MessageChannelSyncStatus.ONGOING,
syncStageStartedAt: new Date().toISOString(),
});
}
@@ -182,6 +202,25 @@ export class MessageChannelSyncStatusService {
});
}
public async markAsMessagesImportScheduled(
messageChannelIds: string[],
workspaceId: string,
) {
if (!messageChannelIds.length) {
return;
}
const messageChannelRepository =
await this.twentyORMGlobalManager.getRepositoryForWorkspace<MessageChannelWorkspaceEntity>(
workspaceId,
'messageChannel',
);
await messageChannelRepository.update(messageChannelIds, {
syncStage: MessageChannelSyncStage.MESSAGES_IMPORT_SCHEDULED,
});
}
public async markAsMessagesImportOngoing(
messageChannelIds: string[],
workspaceId: string,
packages/twenty-server/src/modules/messaging/message-import-manager/crons/jobs/messaging-message-list-fetch.cron.job.ts
+4 -1
View File
@@ -48,8 +48,11 @@ export class MessagingMessageListFetchCronJob {
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}' WHERE "isSyncEnabled" = true AND "syncStage" = '${MessageChannelSyncStage.MESSAGE_LIST_FETCH_PENDING}' RETURNING *`,
`UPDATE ${schemaName}."messageChannel" SET "syncStage" = '${MessageChannelSyncStage.MESSAGE_LIST_FETCH_SCHEDULED}', "syncStageStartedAt" = '${now}'
WHERE "isSyncEnabled" = true AND "syncStage" = '${MessageChannelSyncStage.MESSAGE_LIST_FETCH_PENDING}' RETURNING *`,
);
for (const messageChannel of messageChannels) {
packages/twenty-server/src/modules/messaging/message-import-manager/crons/jobs/messaging-messages-import.cron.job.ts
+4 -1
View File
@@ -53,8 +53,11 @@ export class MessagingMessagesImportCronJob {
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}' WHERE "isSyncEnabled" = true AND "syncStage" = '${MessageChannelSyncStage.MESSAGES_IMPORT_PENDING}' RETURNING *`,
`UPDATE ${schemaName}."messageChannel" SET "syncStage" = '${MessageChannelSyncStage.MESSAGES_IMPORT_SCHEDULED}', "syncStageStartedAt" = '${now}'
WHERE "isSyncEnabled" = true AND "syncStage" = '${MessageChannelSyncStage.MESSAGES_IMPORT_PENDING}' RETURNING *`,
);
for (const messageChannel of messageChannels) {
packages/twenty-server/src/modules/messaging/message-import-manager/services/__tests__/messaging-message-list-fetch.service.spec.ts
+4 -4
View File
@@ -291,7 +291,7 @@ describe('MessagingMessageListFetchService', () => {
});
expect(
messageChannelSyncStatusService.markAsMessagesListFetchOngoing,
).toHaveBeenCalledWith([mockMicrosoftMessageChannel.id]);
).toHaveBeenCalledWith([mockMicrosoftMessageChannel.id], workspaceId);
expect(messagingGetMessageListService.getMessageLists).toHaveBeenCalledWith(
{
@@ -332,7 +332,7 @@ describe('MessagingMessageListFetchService', () => {
expect(
messageChannelSyncStatusService.scheduleMessagesImport,
).toHaveBeenCalledWith([mockMicrosoftMessageChannel.id]);
).toHaveBeenCalledWith([mockMicrosoftMessageChannel.id], workspaceId);
});
it('should process Google message list fetch correctly', async () => {
@@ -350,7 +350,7 @@ describe('MessagingMessageListFetchService', () => {
});
expect(
messageChannelSyncStatusService.markAsMessagesListFetchOngoing,
).toHaveBeenCalledWith([mockGoogleMessageChannel.id]);
).toHaveBeenCalledWith([mockGoogleMessageChannel.id], workspaceId);
expect(messagingGetMessageListService.getMessageLists).toHaveBeenCalledWith(
{
@@ -391,6 +391,6 @@ describe('MessagingMessageListFetchService', () => {
expect(
messageChannelSyncStatusService.scheduleMessagesImport,
).toHaveBeenCalledWith([mockGoogleMessageChannel.id]);
).toHaveBeenCalledWith([mockGoogleMessageChannel.id], workspaceId);
});
});
packages/twenty-server/src/modules/messaging/message-import-manager/services/__tests__/messaging-messages-import.service.spec.ts
+1 -1
View File
@@ -213,7 +213,7 @@ describe('MessagingMessagesImportService', () => {
);
expect(
messageChannelSyncStatusService.markAsMessagesImportOngoing,
).toHaveBeenCalledWith([mockMessageChannel.id]);
).toHaveBeenCalledWith([mockMessageChannel.id], workspaceId);
expect(
connectedAccountRefreshTokensService.refreshAndSaveTokens,