feat(messaging): skip webhook-active channels in list-fetch crons until sync is stale (#22183)

<!-- This is an auto-generated description by cubic. -->
<a
href="https://cubic.dev/pr/twentyhq/twenty/pull/22183?utm_source=github"
target="_blank" rel="noopener noreferrer"
data-no-image-dialog="true"><picture><source
media="(prefers-color-scheme: dark)"
srcset="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"><source
media="(prefers-color-scheme: light)"
srcset="https://www.cubic.dev/buttons/review-in-cubic-light.svg"><img
alt="Review in cubic"
src="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"></picture></a>
<!-- End of auto-generated description by cubic. -->

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 =

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');
+  });
+});

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;
+};

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;

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 =