Backfill webhook subscriptions for existing connected accounts (#22314)

Add command iterating workspaces, enqueuing staggered per-channel jobs for Google/Microsoft channels still on polling  <a href="https://cubic.dev/pr/twentyhq/twenty/pull/22314?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>
2026-07-01 15:02:18 +05:30
parent 59d708e324
commit 4fef02394f
7 changed files with 213 additions and 14 deletions
@@ -8,7 +8,6 @@ import { PermissionsModule } from 'src/engine/metadata-modules/permissions/permi
 import { WorkspaceDataSourceModule } from 'src/engine/workspace-datasource/workspace-datasource.module';
 import { ChannelSyncResolver } from 'src/modules/connected-account/channel-sync/channel-sync.resolver';
 import { ChannelSyncService } from 'src/modules/connected-account/channel-sync/services/channel-sync.service';
-import { WebhookSubscriptionModule } from 'src/modules/connected-account/webhook-subscription-manager/webhook-subscription.module';
 import { MessagingCommonModule } from 'src/modules/messaging/common/messaging-common.module';

@Module({
@@ -18,7 +17,6 @@ import { MessagingCommonModule } from 'src/modules/messaging/common/messaging-co
    PermissionsModule,
    WorkspaceDataSourceModule,
    MessagingCommonModule,
-    WebhookSubscriptionModule,
  ],
  providers: [ChannelSyncResolver, ChannelSyncService],
  exports: [ChannelSyncService],
@@ -6,6 +6,7 @@ import {
  CalendarChannelSyncStatus,
  MessageChannelSyncStage,
  MessageChannelType,
+  WebhookSubscriptionChannelType,
 } from 'twenty-shared/types';
 import { Not, Repository } from 'typeorm';

@@ -20,8 +21,10 @@ import {
  CalendarEventListFetchJob,
  type CalendarEventListFetchJobData,
 } from 'src/modules/calendar/calendar-event-import-manager/jobs/calendar-event-list-fetch.job';
-import { CalendarWebhookSubscriptionService } from 'src/modules/connected-account/webhook-subscription-manager/services/calendar-webhook-subscription.service';
-import { MessagingWebhookSubscriptionService } from 'src/modules/connected-account/webhook-subscription-manager/services/messaging-webhook-subscription.service';
+import {
+  CreateWebhookSubscriptionJob,
+  type CreateWebhookSubscriptionJobData,
+} from 'src/modules/connected-account/webhook-subscription-manager/jobs/create-webhook-subscription.job';
 import { MessageChannelSyncStatusService } from 'src/modules/messaging/common/services/message-channel-sync-status.service';
 import {
  MessagingMessageListFetchJob,
@@ -43,13 +46,13 @@ export class ChannelSyncService {
    private readonly messageQueueService: MessageQueueService,
    @InjectMessageQueue(MessageQueue.calendarQueue)
    private readonly calendarQueueService: MessageQueueService,
+    @InjectMessageQueue(MessageQueue.webhookQueue)
+    private readonly webhookQueueService: MessageQueueService,
    @InjectRepository(MessageChannelEntity)
    private readonly messageChannelRepository: Repository<MessageChannelEntity>,
    private readonly messageChannelSyncStatusService: MessageChannelSyncStatusService,
    @InjectRepository(CalendarChannelEntity)
    private readonly calendarChannelRepository: Repository<CalendarChannelEntity>,
-    private readonly messagingWebhookSubscriptionService: MessagingWebhookSubscriptionService,
-    private readonly calendarWebhookSubscriptionService: CalendarWebhookSubscriptionService,
  ) {}

  async startChannelSync(input: StartChannelSyncInput): Promise<void> {
@@ -90,13 +93,17 @@ export class ChannelSyncService {
        );

        try {
-          await this.messagingWebhookSubscriptionService.createSubscription(
-            messageChannel.id,
-            workspaceId,
+          await this.webhookQueueService.add<CreateWebhookSubscriptionJobData>(
+            CreateWebhookSubscriptionJob.name,
+            {
+              channelType: WebhookSubscriptionChannelType.MESSAGING,
+              channelId: messageChannel.id,
+              workspaceId,
+            },
          );
        } catch (error) {
          this.logger.warn(
-            `Failed to create messaging webhook subscription for message channel ${messageChannel.id}`,
+            `Failed to enqueue webhook subscription job for message channel ${messageChannel.id}`,
            error,
          );
        }
@@ -138,13 +145,17 @@ export class ChannelSyncService {
        );

        try {
-          await this.calendarWebhookSubscriptionService.createSubscription(
-            calendarChannel.id,
-            workspaceId,
+          await this.webhookQueueService.add<CreateWebhookSubscriptionJobData>(
+            CreateWebhookSubscriptionJob.name,
+            {
+              channelType: WebhookSubscriptionChannelType.CALENDAR,
+              channelId: calendarChannel.id,
+              workspaceId,
+            },
          );
        } catch (error) {
          this.logger.warn(
-            `Failed to create calendar webhook subscription for calendar channel ${calendarChannel.id}`,
+            `Failed to enqueue webhook subscription job for calendar channel ${calendarChannel.id}`,
            error,
          );
        }
@@ -0,0 +1,139 @@
+import { InjectRepository } from '@nestjs/typeorm';
+
+import { Command } from 'nest-commander';
+import {
+  ConnectedAccountProvider,
+  WebhookSubscriptionChannelType,
+  WebhookSubscriptionStatus,
+} from 'twenty-shared/types';
+import { Repository } from 'typeorm';
+
+import { ActiveOrSuspendedWorkspaceCommandRunner } from 'src/database/commands/command-runners/active-or-suspended-workspace.command-runner';
+import { type RunOnWorkspaceArgs } from 'src/database/commands/command-runners/workspace.command-runner';
+import { WorkspaceIteratorService } from 'src/database/commands/command-runners/workspace-iterator.service';
+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 { CalendarChannelEntity } from 'src/engine/metadata-modules/calendar-channel/entities/calendar-channel.entity';
+import { MessageChannelEntity } from 'src/engine/metadata-modules/message-channel/entities/message-channel.entity';
+import {
+  CreateWebhookSubscriptionJob,
+  type CreateWebhookSubscriptionJobData,
+} from 'src/modules/connected-account/webhook-subscription-manager/jobs/create-webhook-subscription.job';
+
+const WEBHOOK_BACKFILL_SPACING_MS = 2000;
+const WEBHOOK_BACKFILL_RETRY_LIMIT = 3;
+
+const WEBHOOK_CAPABLE_PROVIDERS = [
+  ConnectedAccountProvider.GOOGLE,
+  ConnectedAccountProvider.MICROSOFT,
+];
+
+@Command({
+  name: 'connected-account:create-webhook-subscription',
+  description:
+    'Enqueue webhook subscription creation for existing Google/Microsoft channels still on polling, staggered to avoid provider rate limiting',
+})
+export class CreateWebhookSubscriptionForConnectedAccountCommand extends ActiveOrSuspendedWorkspaceCommandRunner {
+  private enqueueCursorMs = 0;
+
+  constructor(
+    @InjectRepository(MessageChannelEntity)
+    private readonly messageChannelRepository: Repository<MessageChannelEntity>,
+    @InjectRepository(CalendarChannelEntity)
+    private readonly calendarChannelRepository: Repository<CalendarChannelEntity>,
+    @InjectMessageQueue(MessageQueue.webhookQueue)
+    private readonly webhookQueueService: MessageQueueService,
+    protected readonly workspaceIteratorService: WorkspaceIteratorService,
+  ) {
+    super(workspaceIteratorService);
+  }
+
+  override async runOnWorkspace({
+    workspaceId,
+    options,
+  }: RunOnWorkspaceArgs): Promise<void> {
+    const isDryRun = options.dryRun ?? false;
+
+    const messageChannelIds = await this.findEligibleChannelIds(
+      this.messageChannelRepository,
+      workspaceId,
+    );
+
+    await this.enqueueChannels(
+      WebhookSubscriptionChannelType.MESSAGING,
+      messageChannelIds,
+      workspaceId,
+      isDryRun,
+    );
+
+    const calendarChannelIds = await this.findEligibleChannelIds(
+      this.calendarChannelRepository,
+      workspaceId,
+    );
+
+    await this.enqueueChannels(
+      WebhookSubscriptionChannelType.CALENDAR,
+      calendarChannelIds,
+      workspaceId,
+      isDryRun,
+    );
+  }
+
+  private async findEligibleChannelIds<
+    TChannel extends MessageChannelEntity | CalendarChannelEntity,
+  >(repository: Repository<TChannel>, workspaceId: string): Promise<string[]> {
+    const rows = await repository
+      .createQueryBuilder('core')
+      .select('core.id', 'id')
+      .innerJoin('core.connectedAccount', 'connectedAccount')
+      .where('core.workspaceId = :workspaceId', { workspaceId })
+      .andWhere('core.isSyncEnabled = true')
+      .andWhere('connectedAccount.provider IN (:...providers)', {
+        providers: WEBHOOK_CAPABLE_PROVIDERS,
+      })
+      .andWhere(
+        '(core.webhookSubscriptionStatus IS NULL OR core.webhookSubscriptionStatus != :activeStatus)',
+        { activeStatus: WebhookSubscriptionStatus.ACTIVE },
+      )
+      .getRawMany<{ id: string }>();
+
+    return rows.map((row) => row.id);
+  }
+
+  private async enqueueChannels(
+    channelType: WebhookSubscriptionChannelType,
+    channelIds: string[],
+    workspaceId: string,
+    isDryRun: boolean,
+  ): Promise<void> {
+    if (channelIds.length === 0) {
+      return;
+    }
+
+    if (isDryRun) {
+      this.logger.log(
+        `[DRY RUN] Workspace ${workspaceId}: ${channelIds.length} ${channelType} channels eligible for webhook backfill`,
+      );
+
+      return;
+    }
+
+    for (const channelId of channelIds) {
+      await this.webhookQueueService.add<CreateWebhookSubscriptionJobData>(
+        CreateWebhookSubscriptionJob.name,
+        { channelType, channelId, workspaceId },
+        {
+          delay: this.enqueueCursorMs,
+          retryLimit: WEBHOOK_BACKFILL_RETRY_LIMIT,
+        },
+      );
+
+      this.enqueueCursorMs += WEBHOOK_BACKFILL_SPACING_MS;
+    }
+
+    this.logger.log(
+      `Workspace ${workspaceId}: enqueued ${channelIds.length} ${channelType} webhook backfill jobs`,
+    );
+  }
+}
@@ -0,0 +1,41 @@
+import { WebhookSubscriptionChannelType } from 'twenty-shared/types';
+
+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 { CalendarWebhookSubscriptionService } from 'src/modules/connected-account/webhook-subscription-manager/services/calendar-webhook-subscription.service';
+import { MessagingWebhookSubscriptionService } from 'src/modules/connected-account/webhook-subscription-manager/services/messaging-webhook-subscription.service';
+
+export type CreateWebhookSubscriptionJobData = {
+  channelType: WebhookSubscriptionChannelType;
+  channelId: string;
+  workspaceId: string;
+};
+
+@Processor(MessageQueue.webhookQueue)
+export class CreateWebhookSubscriptionJob {
+  constructor(
+    private readonly messagingWebhookSubscriptionService: MessagingWebhookSubscriptionService,
+    private readonly calendarWebhookSubscriptionService: CalendarWebhookSubscriptionService,
+  ) {}
+
+  @Process(CreateWebhookSubscriptionJob.name)
+  async handle(data: CreateWebhookSubscriptionJobData): Promise<void> {
+    const { channelType, channelId, workspaceId } = data;
+
+    switch (channelType) {
+      case WebhookSubscriptionChannelType.MESSAGING:
+        await this.messagingWebhookSubscriptionService.createSubscription(
+          channelId,
+          workspaceId,
+        );
+        break;
+      case WebhookSubscriptionChannelType.CALENDAR:
+        await this.calendarWebhookSubscriptionService.createSubscription(
+          channelId,
+          workspaceId,
+        );
+        break;
+    }
+  }
+}
@@ -103,6 +103,8 @@ export class CalendarWebhookSubscriptionService {
      this.exceptionHandlerService.captureExceptions([error], {
        workspace: { id: workspaceId },
      });
+
+      throw error;
    }
  }

@@ -102,6 +102,8 @@ export class MessagingWebhookSubscriptionService {
      this.exceptionHandlerService.captureExceptions([error], {
        workspace: { id: workspaceId },
      });
+
+      throw error;
    }
  }

@@ -1,11 +1,14 @@
 import { Module } from '@nestjs/common';
 import { TypeOrmModule } from '@nestjs/typeorm';

+import { WorkspaceIteratorModule } from 'src/database/commands/command-runners/workspace-iterator.module';
 import { FeatureFlagModule } from 'src/engine/core-modules/feature-flag/feature-flag.module';
 import { ConnectedAccountEntity } from 'src/engine/metadata-modules/connected-account/entities/connected-account.entity';
 import { CalendarChannelEntity } from 'src/engine/metadata-modules/calendar-channel/entities/calendar-channel.entity';
 import { MessageChannelEntity } from 'src/engine/metadata-modules/message-channel/entities/message-channel.entity';
+import { CreateWebhookSubscriptionForConnectedAccountCommand } from 'src/modules/connected-account/webhook-subscription-manager/commands/create-webhook-subscription-for-connected-account.command';
 import { WebhookSubscriptionRenewalCronCommand } from 'src/modules/connected-account/webhook-subscription-manager/crons/commands/webhook-subscription-renewal.cron.command';
+import { CreateWebhookSubscriptionJob } from 'src/modules/connected-account/webhook-subscription-manager/jobs/create-webhook-subscription.job';
 import { WebhookSubscriptionRenewalCronJob } from 'src/modules/connected-account/webhook-subscription-manager/crons/jobs/webhook-subscription-renewal.cron.job';
 import { WebhookSubscriptionChannelDeletedListener } from 'src/modules/connected-account/webhook-subscription-manager/listeners/webhook-subscription-channel-deleted.listener';
 import { CalendarWebhookSubscriptionService } from 'src/modules/connected-account/webhook-subscription-manager/services/calendar-webhook-subscription.service';
@@ -16,6 +19,7 @@ import { WebhookSubscriptionManagerModule } from 'src/modules/connected-account/
  imports: [
    WebhookSubscriptionManagerModule,
    FeatureFlagModule,
+    WorkspaceIteratorModule,
    TypeOrmModule.forFeature([
      ConnectedAccountEntity,
      MessageChannelEntity,
@@ -28,6 +32,8 @@ import { WebhookSubscriptionManagerModule } from 'src/modules/connected-account/
    WebhookSubscriptionChannelDeletedListener,
    WebhookSubscriptionRenewalCronJob,
    WebhookSubscriptionRenewalCronCommand,
+    CreateWebhookSubscriptionJob,
+    CreateWebhookSubscriptionForConnectedAccountCommand,
  ],
  exports: [
    MessagingWebhookSubscriptionService,