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

fix(messaging): register and harden webhook subscription renewal cron (#23006)

Browse Source 
The renewal cron was never wired into cron:register:all, so
Gmail/Calendar/Graph watches were never renewed and went dark ~7 days
after connect (the max watch lifetime all three providers allow).
Register it, fan renewals out as per-channel queue jobs scoped to active
workspaces, retry FAILED channels, and recreate Google Calendar watches
before stopping the old one.

<!-- This is an auto-generated description by cubic. -->
<a
href="https://cubic.dev/pr/twentyhq/twenty/pull/23006?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. -->
This commit is contained in:
neo773
2026-07-20 17:26:47 +05:30
committed by GitHub
parent fa720358d9
commit c04714b9e0
10 changed files with 266 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/twenty-server/src/database/commands/cron-register-all.command.ts
+7
View File
@@ -15,6 +15,7 @@ import { CronTriggerCronCommand } from 'src/engine/core-modules/logic-function/l
import { CheckPublicDomainsValidRecordsCronCommand } from 'src/engine/core-modules/public-domain/crons/commands/check-public-domains-valid-records.cron.command';
import { TwentyConfigService } from 'src/engine/core-modules/twenty-config/twenty-config.service';
import { CheckCustomDomainValidRecordsCronCommand } from 'src/engine/core-modules/workspace/crons/commands/check-custom-domain-valid-records.cron.command';
import { WebhookSubscriptionRenewalCronCommand } from 'src/modules/connected-account/webhook-subscription-manager/crons/commands/webhook-subscription-renewal.cron.command';
import { TrashCleanupCronCommand } from 'src/engine/trash-cleanup/commands/trash-cleanup.cron.command';
import { CleanOnboardingWorkspacesCronCommand } from 'src/engine/workspace-manager/workspace-cleaner/commands/clean-onboarding-workspaces.cron.command';
import { CleanSuspendedWorkspacesCronCommand } from 'src/engine/workspace-manager/workspace-cleaner/commands/clean-suspended-workspaces.cron.command';
@@ -49,6 +50,8 @@ export class CronRegisterAllCommand extends CommandRunner {
private readonly calendarOngoingStaleCronCommand: CalendarOngoingStaleCronCommand,
private readonly calendarRelaunchFailedCalendarChannelsCronCommand: CalendarRelaunchFailedCalendarChannelsCronCommand,
private readonly webhookSubscriptionRenewalCronCommand: WebhookSubscriptionRenewalCronCommand,
private readonly workflowCronTriggerCronCommand: WorkflowCronTriggerCronCommand,
private readonly workflowRunEnqueueCronCommand: WorkflowRunEnqueueCronCommand,
private readonly workflowHandleStaledRunsCronCommand: WorkflowHandleStaledRunsCronCommand,
@@ -119,6 +122,10 @@ export class CronRegisterAllCommand extends CommandRunner {
name: 'CalendarRelaunchFailedCalendarChannels',
command: this.calendarRelaunchFailedCalendarChannelsCronCommand,
},
{
name: 'WebhookSubscriptionRenewal',
command: this.webhookSubscriptionRenewalCronCommand,
},
{
name: 'CheckCustomDomainValidRecords',
command: this.checkCustomDomainValidRecordsCronCommand,
packages/twenty-server/src/database/commands/database-command.module.ts
+2
View File
@@ -48,6 +48,7 @@ import { WorkspaceCleanerModule } from 'src/engine/workspace-manager/workspace-c
import { WorkspaceManagerModule } from 'src/engine/workspace-manager/workspace-manager.module';
import { WorkspaceMigrationModule } from 'src/engine/workspace-manager/workspace-migration/workspace-migration.module';
import { WorkspaceVersionModule } from 'src/engine/workspace-manager/workspace-version/workspace-version.module';
import { WebhookSubscriptionModule } from 'src/modules/connected-account/webhook-subscription-manager/webhook-subscription.module';
import { CalendarEventImportManagerModule } from 'src/modules/calendar/calendar-event-import-manager/calendar-event-import-manager.module';
import { MessagingImportManagerModule } from 'src/modules/messaging/message-import-manager/messaging-import-manager.module';
import { WorkflowRunQueueModule } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workflow-run-queue.module';
@@ -61,6 +62,7 @@ import { AutomatedTriggerModule } from 'src/modules/workflow/workflow-trigger/au
// Cron command dependencies
MessagingImportManagerModule,
CalendarEventImportManagerModule,
WebhookSubscriptionModule,
AutomatedTriggerModule,
FileModule,
WorkspaceModule,
packages/twenty-server/src/modules/connected-account-sync-webhooks/drivers/microsoft/microsoft-calendar-notification.handler.ts
+11 -3
View File
@@ -96,9 +96,17 @@ export class MicrosoftCalendarNotificationHandler implements WebhookNotification
isNonEmptyString(notification.lifecycleEvent) &&
notification.lifecycleEvent !== 'missed'
) {
await this.calendarWebhookSubscriptionService.renewSubscription(
calendarChannel,
);
try {
await this.calendarWebhookSubscriptionService.renewSubscription({
calendarChannelId: calendarChannel.id,
workspaceId: calendarChannel.workspaceId,
});
} catch (error) {
this.logger.error(
`Failed to renew calendar subscription for channel ${calendarChannel.id}`,
error,
);
}
continue;
}
packages/twenty-server/src/modules/connected-account-sync-webhooks/drivers/microsoft/microsoft-messaging-notification.handler.ts
+11 -3
View File
@@ -95,9 +95,17 @@ export class MicrosoftMessagingNotificationHandler implements WebhookNotificatio
isNonEmptyString(notification.lifecycleEvent) &&
notification.lifecycleEvent !== 'missed'
) {
await this.messagingWebhookSubscriptionService.renewSubscription(
messageChannel,
);
try {
await this.messagingWebhookSubscriptionService.renewSubscription({
messageChannelId: messageChannel.id,
workspaceId: messageChannel.workspaceId,
});
} catch (error) {
this.logger.error(
`Failed to renew messaging subscription for channel ${messageChannel.id}`,
error,
);
}
continue;
}
packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/crons/jobs/webhook-subscription-renewal.cron.job.ts
+100 -53
View File
@@ -1,31 +1,46 @@
import { Logger } from '@nestjs/common';
import { InjectRepository } from '@nestjs/typeorm';
import { WebhookSubscriptionStatus } from 'twenty-shared/types';
import { LessThanOrEqual, Repository } from 'typeorm';
import {
WebhookSubscriptionChannelType,
WebhookSubscriptionStatus,
} from 'twenty-shared/types';
import { WorkspaceActivationStatus } from 'twenty-shared/workspace';
import { type FindManyOptions, In, LessThanOrEqual, Repository } from 'typeorm';
import { SentryCronMonitor } from 'src/engine/core-modules/cron/sentry-cron-monitor.decorator';
import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator';
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 { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service';
import { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.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 { WEBHOOK_SUBSCRIPTION_RENEWAL_BUFFER_MS } from 'src/modules/connected-account/webhook-subscription-manager/constants/webhook-subscription-renewal-buffer-ms.constant';
import { WEBHOOK_SUBSCRIPTION_RENEWAL_CRON_PATTERN } from 'src/modules/connected-account/webhook-subscription-manager/constants/webhook-subscription-renewal-cron-pattern.constant';
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 {
RenewWebhookSubscriptionJob,
type RenewWebhookSubscriptionJobData,
} from 'src/modules/connected-account/webhook-subscription-manager/jobs/renew-webhook-subscription.job';
type WebhookSubscribableChannel = MessageChannelEntity | CalendarChannelEntity;
type StaleChannel = Pick<WebhookSubscribableChannel, 'id' | 'workspaceId'>;
@Processor(MessageQueue.cronQueue)
export class WebhookSubscriptionRenewalCronJob {
private readonly logger = new Logger(WebhookSubscriptionRenewalCronJob.name);
constructor(
@InjectRepository(WorkspaceEntity)
private readonly workspaceRepository: Repository<WorkspaceEntity>,
@InjectRepository(MessageChannelEntity)
private readonly messageChannelRepository: Repository<MessageChannelEntity>,
@InjectRepository(CalendarChannelEntity)
private readonly calendarChannelRepository: Repository<CalendarChannelEntity>,
private readonly messagingWebhookSubscriptionService: MessagingWebhookSubscriptionService,
private readonly calendarWebhookSubscriptionService: CalendarWebhookSubscriptionService,
@InjectMessageQueue(MessageQueue.webhookQueue)
private readonly webhookQueueService: MessageQueueService,
) {}
@Process(WebhookSubscriptionRenewalCronJob.name)
@@ -34,59 +49,91 @@ export class WebhookSubscriptionRenewalCronJob {
WEBHOOK_SUBSCRIPTION_RENEWAL_CRON_PATTERN,
)
async handle(): Promise<void> {
const activeWorkspaces = await this.workspaceRepository.find({
where: { activationStatus: WorkspaceActivationStatus.ACTIVE },
select: { id: true },
});
const activeWorkspaceIds = activeWorkspaces.map(
(workspace) => workspace.id,
);
if (activeWorkspaceIds.length === 0) {
return;
}
const [messageChannels, calendarChannels] = await Promise.all([
this.findStaleChannels(this.messageChannelRepository, activeWorkspaceIds),
this.findStaleChannels(
this.calendarChannelRepository,
activeWorkspaceIds,
),
]);
if (messageChannels.length === 0 && calendarChannels.length === 0) {
return;
}
await Promise.all([
this.enqueueRenewals(
WebhookSubscriptionChannelType.MESSAGING,
messageChannels,
),
this.enqueueRenewals(
WebhookSubscriptionChannelType.CALENDAR,
calendarChannels,
),
]);
this.logger.log(
`Enqueued webhook subscription renewals: ${messageChannels.length} messaging, ${calendarChannels.length} calendar`,
);
}
private findStaleChannels<TChannel extends WebhookSubscribableChannel>(
repository: Repository<TChannel>,
activeWorkspaceIds: string[],
): Promise<StaleChannel[]> {
const workspaceScope = { workspaceId: In(activeWorkspaceIds) };
const renewalThreshold = new Date(
Date.now() + WEBHOOK_SUBSCRIPTION_RENEWAL_BUFFER_MS,
);
const expiringMessageChannels = await this.messageChannelRepository.find({
where: {
webhookSubscriptionStatus: WebhookSubscriptionStatus.ACTIVE,
webhookSubscriptionExpiresAt: LessThanOrEqual(renewalThreshold),
},
});
const options: FindManyOptions<WebhookSubscribableChannel> = {
where: [
{
...workspaceScope,
webhookSubscriptionStatus: WebhookSubscriptionStatus.FAILED,
},
{
...workspaceScope,
webhookSubscriptionStatus: WebhookSubscriptionStatus.ACTIVE,
webhookSubscriptionExpiresAt: LessThanOrEqual(renewalThreshold),
},
],
select: { id: true, workspaceId: true },
};
const expiringCalendarChannels = await this.calendarChannelRepository.find({
where: {
webhookSubscriptionStatus: WebhookSubscriptionStatus.ACTIVE,
webhookSubscriptionExpiresAt: LessThanOrEqual(renewalThreshold),
},
});
return repository.find(options as FindManyOptions<TChannel>);
}
const expiringSubscriptionCount =
expiringMessageChannels.length + expiringCalendarChannels.length;
if (expiringSubscriptionCount === 0) {
return;
}
this.logger.log(
`Renewing ${expiringSubscriptionCount} webhook subscriptions`,
);
for (const messageChannel of expiringMessageChannels) {
try {
await this.messagingWebhookSubscriptionService.renewSubscription(
messageChannel,
);
} catch (error) {
this.logger.warn(
`Failed to renew messaging webhook subscription for channel ${messageChannel.id}`,
error,
);
}
}
for (const calendarChannel of expiringCalendarChannels) {
try {
await this.calendarWebhookSubscriptionService.renewSubscription(
calendarChannel,
);
} catch (error) {
this.logger.warn(
`Failed to renew calendar webhook subscription for channel ${calendarChannel.id}`,
error,
);
}
private async enqueueRenewals(
channelType: WebhookSubscriptionChannelType,
channels: StaleChannel[],
): Promise<void> {
for (const channel of channels) {
await this.webhookQueueService.add<RenewWebhookSubscriptionJobData>(
RenewWebhookSubscriptionJob.name,
{
channelType,
channelId: channel.id,
workspaceId: channel.workspaceId,
},
{
id: `${RenewWebhookSubscriptionJob.name}:${channelType}:${channel.id}`,
retryLimit: 3,
},
);
}
}
}
packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/drivers/google/google-webhook-subscription.driver.ts
+11 -1
View File
@@ -41,7 +41,17 @@ export class GoogleWebhookSubscriptionDriver implements WebhookSubscriptionDrive
context: WebhookSubscriptionContext,
): Promise<WebhookSubscriptionResult> {
if (context.channelType === WebhookSubscriptionChannelType.CALENDAR) {
await this.deleteSubscription(context);
// Google Calendar watches can't be extended, only replaced by a fresh channel.
// Recreate before stopping the old watch so a failed recreate never leaves the account with no live watch.
const result = await this.createSubscription(
context.connectedAccountId,
context.channelType,
context.clientState,
);
await this.deleteSubscription(context).catch(() => undefined);
return result;
}
return this.createSubscription(
packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/jobs/renew-webhook-subscription.job.ts
+41
View File
@@ -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 RenewWebhookSubscriptionJobData = {
channelType: WebhookSubscriptionChannelType;
channelId: string;
workspaceId: string;
};
@Processor(MessageQueue.webhookQueue)
export class RenewWebhookSubscriptionJob {
constructor(
private readonly messagingWebhookSubscriptionService: MessagingWebhookSubscriptionService,
private readonly calendarWebhookSubscriptionService: CalendarWebhookSubscriptionService,
) {}
@Process(RenewWebhookSubscriptionJob.name)
async handle(data: RenewWebhookSubscriptionJobData): Promise<void> {
const { channelType, channelId, workspaceId } = data;
switch (channelType) {
case WebhookSubscriptionChannelType.MESSAGING:
await this.messagingWebhookSubscriptionService.renewSubscription({
messageChannelId: channelId,
workspaceId,
});
break;
case WebhookSubscriptionChannelType.CALENDAR:
await this.calendarWebhookSubscriptionService.renewSubscription({
calendarChannelId: channelId,
workspaceId,
});
break;
}
}
}
packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/services/calendar-webhook-subscription.service.ts
+39 -13
View File
@@ -61,11 +61,12 @@ export class CalendarWebhookSubscriptionService {
connectedAccount.provider,
);
if (isDefined(calendarChannel.webhookSubscriptionExternalId)) {
await driver
.deleteSubscription(this.toContext(calendarChannel))
.catch(() => undefined);
}
// Keep any existing watch live until the replacement is created, then stop it.
const previousSubscription = isDefined(
calendarChannel.webhookSubscriptionExternalId,
)
? this.toContext(calendarChannel)
: null;
try {
const result = await driver.createSubscription(
@@ -94,18 +95,41 @@ export class CalendarWebhookSubscriptionService {
throw error;
}
if (isDefined(previousSubscription)) {
await driver
.deleteSubscription(previousSubscription)
.catch(() => undefined);
}
}
async renewSubscription(
calendarChannel: CalendarChannelEntity,
): Promise<void> {
const connectedAccount = await this.connectedAccountRepository.findOne({
where: {
id: calendarChannel.connectedAccountId,
workspaceId: calendarChannel.workspaceId,
},
async renewSubscription({
calendarChannelId,
workspaceId,
}: {
calendarChannelId: string;
workspaceId: string;
}): Promise<void> {
const calendarChannel = await this.calendarChannelRepository.findOne({
where: { id: calendarChannelId, workspaceId },
relations: ['connectedAccount'],
});
if (!isDefined(calendarChannel)) {
return;
}
if (
calendarChannel.webhookSubscriptionStatus !==
WebhookSubscriptionStatus.ACTIVE
) {
await this.createSubscription(calendarChannelId, workspaceId);
return;
}
const { connectedAccount } = calendarChannel;
if (!isDefined(connectedAccount)) {
return;
}
@@ -133,6 +157,8 @@ export class CalendarWebhookSubscriptionService {
this.exceptionHandlerService.captureExceptions([error], {
workspace: { id: calendarChannel.workspaceId },
});
throw error;
}
}
packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/services/messaging-webhook-subscription.service.ts
+39 -11
View File
@@ -61,11 +61,12 @@ export class MessagingWebhookSubscriptionService {
connectedAccount.provider,
);
if (isDefined(messageChannel.webhookSubscriptionExternalId)) {
await driver
.deleteSubscription(this.toContext(messageChannel))
.catch(() => undefined);
}
// Keep any existing watch live until the replacement is created, then stop it.
const previousSubscription = isDefined(
messageChannel.webhookSubscriptionExternalId,
)
? this.toContext(messageChannel)
: null;
try {
const result = await driver.createSubscription(
@@ -93,16 +94,41 @@ export class MessagingWebhookSubscriptionService {
throw error;
}
if (isDefined(previousSubscription)) {
await driver
.deleteSubscription(previousSubscription)
.catch(() => undefined);
}
}
async renewSubscription(messageChannel: MessageChannelEntity): Promise<void> {
const connectedAccount = await this.connectedAccountRepository.findOne({
where: {
id: messageChannel.connectedAccountId,
workspaceId: messageChannel.workspaceId,
},
async renewSubscription({
messageChannelId,
workspaceId,
}: {
messageChannelId: string;
workspaceId: string;
}): Promise<void> {
const messageChannel = await this.messageChannelRepository.findOne({
where: { id: messageChannelId, workspaceId },
relations: ['connectedAccount'],
});
if (!isDefined(messageChannel)) {
return;
}
if (
messageChannel.webhookSubscriptionStatus !==
WebhookSubscriptionStatus.ACTIVE
) {
await this.createSubscription(messageChannelId, workspaceId);
return;
}
const { connectedAccount } = messageChannel;
if (!isDefined(connectedAccount)) {
return;
}
@@ -129,6 +155,8 @@ export class MessagingWebhookSubscriptionService {
this.exceptionHandlerService.captureExceptions([error], {
workspace: { id: messageChannel.workspaceId },
});
throw error;
}
}
packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/webhook-subscription.module.ts
+5
View File
@@ -3,12 +3,14 @@ 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 { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity';
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 { RenewWebhookSubscriptionJob } from 'src/modules/connected-account/webhook-subscription-manager/jobs/renew-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';
@@ -21,6 +23,7 @@ import { WebhookSubscriptionManagerModule } from 'src/modules/connected-account/
FeatureFlagModule,
WorkspaceIteratorModule,
TypeOrmModule.forFeature([
WorkspaceEntity,
ConnectedAccountEntity,
MessageChannelEntity,
CalendarChannelEntity,
@@ -33,11 +36,13 @@ import { WebhookSubscriptionManagerModule } from 'src/modules/connected-account/
WebhookSubscriptionRenewalCronJob,
WebhookSubscriptionRenewalCronCommand,
CreateWebhookSubscriptionJob,
RenewWebhookSubscriptionJob,
CreateWebhookSubscriptionForConnectedAccountCommand,
],
exports: [
MessagingWebhookSubscriptionService,
CalendarWebhookSubscriptionService,
WebhookSubscriptionRenewalCronCommand,
],
})
export class WebhookSubscriptionModule {}