From 4fef02394fd845fafe47913b40799d0298017bce Mon Sep 17 00:00:00 2001 From: neo773 <62795688+neo773@users.noreply.github.com> Date: Wed, 1 Jul 2026 15:02:18 +0530 Subject: [PATCH] Backfill webhook subscriptions for existing connected accounts (#22314) Add command iterating workspaces, enqueuing staggered per-channel jobs for Google/Microsoft channels still on polling Review in cubic --- .../channel-sync/channel-sync.module.ts | 2 - .../services/channel-sync.service.ts | 35 +++-- ...scription-for-connected-account.command.ts | 139 ++++++++++++++++++ .../jobs/create-webhook-subscription.job.ts | 41 ++++++ .../calendar-webhook-subscription.service.ts | 2 + .../messaging-webhook-subscription.service.ts | 2 + .../webhook-subscription.module.ts | 6 + 7 files changed, 213 insertions(+), 14 deletions(-) create mode 100644 packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/commands/create-webhook-subscription-for-connected-account.command.ts create mode 100644 packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/jobs/create-webhook-subscription.job.ts diff --git a/packages/twenty-server/src/modules/connected-account/channel-sync/channel-sync.module.ts b/packages/twenty-server/src/modules/connected-account/channel-sync/channel-sync.module.ts index 37fd5c621d..4ba36f589c 100644 --- a/packages/twenty-server/src/modules/connected-account/channel-sync/channel-sync.module.ts +++ b/packages/twenty-server/src/modules/connected-account/channel-sync/channel-sync.module.ts @@ -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], diff --git a/packages/twenty-server/src/modules/connected-account/channel-sync/services/channel-sync.service.ts b/packages/twenty-server/src/modules/connected-account/channel-sync/services/channel-sync.service.ts index 8f84746ba0..8ce21351a5 100644 --- a/packages/twenty-server/src/modules/connected-account/channel-sync/services/channel-sync.service.ts +++ b/packages/twenty-server/src/modules/connected-account/channel-sync/services/channel-sync.service.ts @@ -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, private readonly messageChannelSyncStatusService: MessageChannelSyncStatusService, @InjectRepository(CalendarChannelEntity) private readonly calendarChannelRepository: Repository, - private readonly messagingWebhookSubscriptionService: MessagingWebhookSubscriptionService, - private readonly calendarWebhookSubscriptionService: CalendarWebhookSubscriptionService, ) {} async startChannelSync(input: StartChannelSyncInput): Promise { @@ -90,13 +93,17 @@ export class ChannelSyncService { ); try { - await this.messagingWebhookSubscriptionService.createSubscription( - messageChannel.id, - workspaceId, + await this.webhookQueueService.add( + 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( + 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, ); } diff --git a/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/commands/create-webhook-subscription-for-connected-account.command.ts b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/commands/create-webhook-subscription-for-connected-account.command.ts new file mode 100644 index 0000000000..93cc3ede49 --- /dev/null +++ b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/commands/create-webhook-subscription-for-connected-account.command.ts @@ -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, + @InjectRepository(CalendarChannelEntity) + private readonly calendarChannelRepository: Repository, + @InjectMessageQueue(MessageQueue.webhookQueue) + private readonly webhookQueueService: MessageQueueService, + protected readonly workspaceIteratorService: WorkspaceIteratorService, + ) { + super(workspaceIteratorService); + } + + override async runOnWorkspace({ + workspaceId, + options, + }: RunOnWorkspaceArgs): Promise { + 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, workspaceId: string): Promise { + 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 { + 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( + 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`, + ); + } +} diff --git a/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/jobs/create-webhook-subscription.job.ts b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/jobs/create-webhook-subscription.job.ts new file mode 100644 index 0000000000..4f4405075f --- /dev/null +++ b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/jobs/create-webhook-subscription.job.ts @@ -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 { + 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; + } + } +} diff --git a/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/services/calendar-webhook-subscription.service.ts b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/services/calendar-webhook-subscription.service.ts index 8135e188d0..b273b91c84 100644 --- a/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/services/calendar-webhook-subscription.service.ts +++ b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/services/calendar-webhook-subscription.service.ts @@ -103,6 +103,8 @@ export class CalendarWebhookSubscriptionService { this.exceptionHandlerService.captureExceptions([error], { workspace: { id: workspaceId }, }); + + throw error; } } diff --git a/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/services/messaging-webhook-subscription.service.ts b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/services/messaging-webhook-subscription.service.ts index e1fc1719b7..bb1b9e418b 100644 --- a/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/services/messaging-webhook-subscription.service.ts +++ b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/services/messaging-webhook-subscription.service.ts @@ -102,6 +102,8 @@ export class MessagingWebhookSubscriptionService { this.exceptionHandlerService.captureExceptions([error], { workspace: { id: workspaceId }, }); + + throw error; } } diff --git a/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/webhook-subscription.module.ts b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/webhook-subscription.module.ts index 62a1c1eb8a..f90ca4d72a 100644 --- a/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/webhook-subscription.module.ts +++ b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/webhook-subscription.module.ts @@ -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,