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 220674febf..00f846efcf 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 @@ -4,7 +4,10 @@ import { InjectRepository } from '@nestjs/typeorm'; import { WorkspaceActivationStatus } from 'twenty-shared/workspace'; import { In, Repository } from 'typeorm'; -import { CalendarChannelSyncStage } from 'twenty-shared/types'; +import { + CalendarChannelSyncStage, + WebhookSubscriptionStatus, +} from 'twenty-shared/types'; 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'; @@ -18,6 +21,7 @@ import { type CalendarEventListFetchJobData, } from 'src/modules/calendar/calendar-event-import-manager/jobs/calendar-event-list-fetch.job'; import { CalendarChannelEntity } from 'src/engine/metadata-modules/calendar-channel/entities/calendar-channel.entity'; +import { isLastSuccessfulSyncStale } from 'src/modules/connected-account/utils/is-last-successful-sync-stale.util'; import { isThrottled } from 'src/modules/connected-account/utils/is-throttled'; import { toIsoStringOrNull } from 'src/utils/date/toIsoStringOrNull'; @@ -68,7 +72,12 @@ export class CalendarEventListFetchCronJob { !isThrottled( toIsoStringOrNull(calendarChannel.syncStageStartedAt), calendarChannel.throttleFailureCount, - ), + ) && + (calendarChannel.webhookSubscriptionStatus !== + WebhookSubscriptionStatus.ACTIVE || + isLastSuccessfulSyncStale( + toIsoStringOrNull(calendarChannel.syncedAt), + )), ); const throttledCount = diff --git a/packages/twenty-server/src/modules/connected-account/utils/__tests__/is-last-successful-sync-stale.util.spec.ts b/packages/twenty-server/src/modules/connected-account/utils/__tests__/is-last-successful-sync-stale.util.spec.ts new file mode 100644 index 0000000000..d0883999ec --- /dev/null +++ b/packages/twenty-server/src/modules/connected-account/utils/__tests__/is-last-successful-sync-stale.util.spec.ts @@ -0,0 +1,33 @@ +import { isLastSuccessfulSyncStale } from 'src/modules/connected-account/utils/is-last-successful-sync-stale.util'; +import { WEBHOOK_SYNC_STALENESS_THRESHOLD_MS } from 'src/modules/connected-account/webhook-subscription-manager/constants/webhook-sync-staleness-threshold-ms.constant'; + +jest.useFakeTimers().setSystemTime(new Date('2024-01-01')); + +describe('isLastSuccessfulSyncStale', () => { + it('should return true when the last sync is older than the staleness threshold', () => { + const syncedAt = new Date( + Date.now() - WEBHOOK_SYNC_STALENESS_THRESHOLD_MS - 1, + ).toISOString(); + + expect(isLastSuccessfulSyncStale(syncedAt)).toBe(true); + }); + + it('should return false when the last sync is within the staleness threshold', () => { + const syncedAt = new Date( + Date.now() - WEBHOOK_SYNC_STALENESS_THRESHOLD_MS + 1, + ).toISOString(); + + expect(isLastSuccessfulSyncStale(syncedAt)).toBe(false); + }); + + it('should return true when the channel has never been synced', () => { + expect(isLastSuccessfulSyncStale(null)).toBe(true); + expect(isLastSuccessfulSyncStale(undefined)).toBe(true); + }); + + it('should throw an error when the timestamp is invalid', () => { + expect(() => { + isLastSuccessfulSyncStale('invalid-date'); + }).toThrow('Invalid date format'); + }); +}); diff --git a/packages/twenty-server/src/modules/connected-account/utils/is-last-successful-sync-stale.util.ts b/packages/twenty-server/src/modules/connected-account/utils/is-last-successful-sync-stale.util.ts new file mode 100644 index 0000000000..7c456edfbf --- /dev/null +++ b/packages/twenty-server/src/modules/connected-account/utils/is-last-successful-sync-stale.util.ts @@ -0,0 +1,19 @@ +import { isDefined } from 'twenty-shared/utils'; + +import { WEBHOOK_SYNC_STALENESS_THRESHOLD_MS } from 'src/modules/connected-account/webhook-subscription-manager/constants/webhook-sync-staleness-threshold-ms.constant'; + +export const isLastSuccessfulSyncStale = ( + syncedAt?: string | null, +): boolean => { + if (!isDefined(syncedAt)) { + return true; + } + + const syncedTime = new Date(syncedAt).getTime(); + + if (isNaN(syncedTime)) { + throw new Error('Invalid date format'); + } + + return Date.now() - syncedTime > WEBHOOK_SYNC_STALENESS_THRESHOLD_MS; +}; diff --git a/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/constants/webhook-sync-staleness-threshold-ms.constant.ts b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/constants/webhook-sync-staleness-threshold-ms.constant.ts new file mode 100644 index 0000000000..459a63ef6d --- /dev/null +++ b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/constants/webhook-sync-staleness-threshold-ms.constant.ts @@ -0,0 +1 @@ +export const WEBHOOK_SYNC_STALENESS_THRESHOLD_MS = 60 * 60 * 1000; 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 415bc228e2..1ab94f76e3 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 @@ -7,6 +7,7 @@ import { In, Not, Repository } from 'typeorm'; import { MessageChannelSyncStage, MessageChannelType, + WebhookSubscriptionStatus, } from 'twenty-shared/types'; import { SentryCronMonitor } from 'src/engine/core-modules/cron/sentry-cron-monitor.decorator'; import { ExceptionHandlerService } from 'src/engine/core-modules/exception-handler/exception-handler.service'; @@ -20,6 +21,7 @@ import { MessagingMessageListFetchJob, type MessagingMessageListFetchJobData, } from 'src/modules/messaging/message-import-manager/jobs/messaging-message-list-fetch.job'; +import { isLastSuccessfulSyncStale } from 'src/modules/connected-account/utils/is-last-successful-sync-stale.util'; import { isThrottled } from 'src/modules/connected-account/utils/is-throttled'; import { MessageChannelEntity } from 'src/engine/metadata-modules/message-channel/entities/message-channel.entity'; import { toIsoStringOrNull } from 'src/utils/date/toIsoStringOrNull'; @@ -71,7 +73,12 @@ export class MessagingMessageListFetchCronJob { toIsoStringOrNull(messageChannel.syncStageStartedAt), messageChannel.throttleFailureCount, toIsoStringOrNull(messageChannel.throttleRetryAfter), - ), + ) && + (messageChannel.webhookSubscriptionStatus !== + WebhookSubscriptionStatus.ACTIVE || + isLastSuccessfulSyncStale( + toIsoStringOrNull(messageChannel.syncedAt), + )), ); const throttledCount =