From 8e5bdcc7817fdc320e0d1b1759e24047b15e3057 Mon Sep 17 00:00:00 2001 From: neo773 <62795688+neo773@users.noreply.github.com> Date: Tue, 4 Aug 2026 15:53:52 +0530 Subject: [PATCH] webhook subscriptions error handling (#23707) A `subscriptionRemoved` lifecycle notification was routed to `renewSubscription`, which PATCHes a subscription Microsoft has already deleted and always 404s ([TWENTY-SERVER-J1N](https://twenty-v7.sentry.io/issues/7604034376/), 884 events). Every sampled webhook event on that issue was `subscriptionRemoved`. Each lifecycle event now gets its own path: `subscriptionRemoved` recreates and resyncs the gap, `reauthorizationRequired` renews in place, `missed` resyncs, unrecognised events are logged and ignored. Provider errors are parsed into driver exception codes following the message-import drivers. Max retry for the renewal cron is deliberately left out and will follow separately. Review in cubic --------- Co-authored-by: neo773 --- ...microsoft-calendar-notification.handler.ts | 69 +++++-- ...icrosoft-messaging-notification.handler.ts | 69 +++++-- .../microsoft-graph-notification.type.ts | 12 +- .../webhook-subscription-driver.exception.ts | 24 ++- ...le-webhook-subscription.driver.dev.spec.ts | 107 +++++++++++ .../google-webhook-subscription.driver.ts | 64 ++++--- ...le-webhook-subscription-error.util.spec.ts | 172 ++++++++++++++++++ ...-google-webhook-subscription-error.util.ts | 110 +++++++++++ ...ft-webhook-subscription.driver.dev.spec.ts | 89 +++++++++ .../microsoft-webhook-subscription.driver.ts | 17 +- ...ft-webhook-subscription-error.util.spec.ts | 106 +++++++++++ ...crosoft-webhook-subscription-error.util.ts | 85 +++++++++ .../calendar-webhook-subscription.service.ts | 104 ++++++++--- .../messaging-webhook-subscription.service.ts | 102 ++++++++--- ...-subscription-exception-handler.service.ts | 144 +++++++++++++++ .../webhook-subscription-status.service.ts | 120 ++++++++++++ .../types/webhook-subscription-driver.type.ts | 11 ++ .../webhook-subscription.module.ts | 4 + ...soft-webhook-subscription-handlers.util.ts | 84 +++++++++ .../mocks/setup-microsoft-mock.util.ts | 17 ++ ...lifecycle-notification.integration-spec.ts | 163 +++++++++++++++++ ...lifecycle-notification.integration-spec.ts | 138 ++++++++++++++ 22 files changed, 1693 insertions(+), 118 deletions(-) create mode 100644 packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/drivers/google/google-webhook-subscription.driver.dev.spec.ts create mode 100644 packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/drivers/google/utils/__tests__/parse-google-webhook-subscription-error.util.spec.ts create mode 100644 packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/drivers/google/utils/parse-google-webhook-subscription-error.util.ts create mode 100644 packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/drivers/microsoft/microsoft-webhook-subscription.driver.dev.spec.ts create mode 100644 packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/drivers/microsoft/utils/__tests__/parse-microsoft-webhook-subscription-error.util.spec.ts create mode 100644 packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/drivers/microsoft/utils/parse-microsoft-webhook-subscription-error.util.ts create mode 100644 packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/services/webhook-subscription-exception-handler.service.ts create mode 100644 packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/services/webhook-subscription-status.service.ts create mode 100644 packages/twenty-server/test/integration/microsoft/mocks/microsoft-webhook-subscription-handlers.util.ts create mode 100644 packages/twenty-server/test/integration/microsoft/webhook/lifecycle-notification.integration-spec.ts create mode 100644 packages/twenty-server/test/integration/microsoft/webhook/messaging-lifecycle-notification.integration-spec.ts diff --git a/packages/twenty-server/src/modules/connected-account-sync-webhooks/drivers/microsoft/microsoft-calendar-notification.handler.ts b/packages/twenty-server/src/modules/connected-account-sync-webhooks/drivers/microsoft/microsoft-calendar-notification.handler.ts index a3577f2da7..95ead2ec60 100644 --- a/packages/twenty-server/src/modules/connected-account-sync-webhooks/drivers/microsoft/microsoft-calendar-notification.handler.ts +++ b/packages/twenty-server/src/modules/connected-account-sync-webhooks/drivers/microsoft/microsoft-calendar-notification.handler.ts @@ -3,7 +3,9 @@ import { timingSafeEqual } from 'crypto'; import { Injectable, Logger } from '@nestjs/common'; import { InjectRepository } from '@nestjs/typeorm'; +import { type LifecycleEventType } from '@microsoft/microsoft-graph-types'; import { isNonEmptyString } from '@sniptt/guards'; +import { type AssertUnreachable } from 'twenty-shared/types'; import { isDefined } from 'twenty-shared/utils'; import { In, Repository } from 'typeorm'; @@ -92,21 +94,17 @@ export class MicrosoftCalendarNotificationHandler implements WebhookNotification continue; } - if ( - isNonEmptyString(notification.lifecycleEvent) && - notification.lifecycleEvent !== 'missed' - ) { - try { - await this.calendarWebhookSubscriptionService.renewSubscription({ - calendarChannelId: calendarChannel.id, - workspaceId: calendarChannel.workspaceId, - }); - } catch (error) { + if (isNonEmptyString(notification.lifecycleEvent)) { + await this.handleLifecycleEvent({ + lifecycleEvent: notification.lifecycleEvent, + removedSubscriptionId: notification.subscriptionId, + calendarChannel, + }).catch((error) => this.logger.error( - `Failed to renew calendar subscription for channel ${calendarChannel.id}`, + `Failed to handle ${notification.lifecycleEvent} lifecycle event for calendar channel ${calendarChannel.id}`, error, - ); - } + ), + ); continue; } @@ -120,4 +118,49 @@ export class MicrosoftCalendarNotificationHandler implements WebhookNotification ); } } + + private async handleLifecycleEvent({ + lifecycleEvent, + removedSubscriptionId, + calendarChannel, + }: { + lifecycleEvent: LifecycleEventType; + removedSubscriptionId: string; + calendarChannel: CalendarChannelEntity; + }): Promise { + switch (lifecycleEvent) { + case 'subscriptionRemoved': + await this.calendarWebhookSubscriptionService.recreateSubscription({ + calendarChannelId: calendarChannel.id, + workspaceId: calendarChannel.workspaceId, + removedSubscriptionId, + }); + await this.webhookSyncTriggerService.triggerCalendarSync( + calendarChannel.id, + calendarChannel.workspaceId, + ); + break; + case 'reauthorizationRequired': + await this.calendarWebhookSubscriptionService.renewSubscription({ + calendarChannelId: calendarChannel.id, + workspaceId: calendarChannel.workspaceId, + }); + break; + case 'missed': + await this.webhookSyncTriggerService.triggerCalendarSync( + calendarChannel.id, + calendarChannel.workspaceId, + ); + break; + default: { + const unhandledLifecycleEvent: AssertUnreachable< + typeof lifecycleEvent + > = lifecycleEvent; + + this.logger.warn( + `Ignored unrecognized lifecycle event ${unhandledLifecycleEvent} for calendar channel ${calendarChannel.id}`, + ); + } + } + } } diff --git a/packages/twenty-server/src/modules/connected-account-sync-webhooks/drivers/microsoft/microsoft-messaging-notification.handler.ts b/packages/twenty-server/src/modules/connected-account-sync-webhooks/drivers/microsoft/microsoft-messaging-notification.handler.ts index 83111b0d13..13f0c92a90 100644 --- a/packages/twenty-server/src/modules/connected-account-sync-webhooks/drivers/microsoft/microsoft-messaging-notification.handler.ts +++ b/packages/twenty-server/src/modules/connected-account-sync-webhooks/drivers/microsoft/microsoft-messaging-notification.handler.ts @@ -3,7 +3,9 @@ import { timingSafeEqual } from 'crypto'; import { Injectable, Logger } from '@nestjs/common'; import { InjectRepository } from '@nestjs/typeorm'; +import { type LifecycleEventType } from '@microsoft/microsoft-graph-types'; import { isNonEmptyString } from '@sniptt/guards'; +import { type AssertUnreachable } from 'twenty-shared/types'; import { isDefined } from 'twenty-shared/utils'; import { In, Repository } from 'typeorm'; @@ -91,21 +93,17 @@ export class MicrosoftMessagingNotificationHandler implements WebhookNotificatio continue; } - if ( - isNonEmptyString(notification.lifecycleEvent) && - notification.lifecycleEvent !== 'missed' - ) { - try { - await this.messagingWebhookSubscriptionService.renewSubscription({ - messageChannelId: messageChannel.id, - workspaceId: messageChannel.workspaceId, - }); - } catch (error) { + if (isNonEmptyString(notification.lifecycleEvent)) { + await this.handleLifecycleEvent({ + lifecycleEvent: notification.lifecycleEvent, + removedSubscriptionId: notification.subscriptionId, + messageChannel, + }).catch((error) => this.logger.error( - `Failed to renew messaging subscription for channel ${messageChannel.id}`, + `Failed to handle ${notification.lifecycleEvent} lifecycle event for message channel ${messageChannel.id}`, error, - ); - } + ), + ); continue; } @@ -119,4 +117,49 @@ export class MicrosoftMessagingNotificationHandler implements WebhookNotificatio ); } } + + private async handleLifecycleEvent({ + lifecycleEvent, + removedSubscriptionId, + messageChannel, + }: { + lifecycleEvent: LifecycleEventType; + removedSubscriptionId: string; + messageChannel: MessageChannelEntity; + }): Promise { + switch (lifecycleEvent) { + case 'subscriptionRemoved': + await this.messagingWebhookSubscriptionService.recreateSubscription({ + messageChannelId: messageChannel.id, + workspaceId: messageChannel.workspaceId, + removedSubscriptionId, + }); + await this.webhookSyncTriggerService.triggerMessagingSync( + messageChannel.id, + messageChannel.workspaceId, + ); + break; + case 'reauthorizationRequired': + await this.messagingWebhookSubscriptionService.renewSubscription({ + messageChannelId: messageChannel.id, + workspaceId: messageChannel.workspaceId, + }); + break; + case 'missed': + await this.webhookSyncTriggerService.triggerMessagingSync( + messageChannel.id, + messageChannel.workspaceId, + ); + break; + default: { + const unhandledLifecycleEvent: AssertUnreachable< + typeof lifecycleEvent + > = lifecycleEvent; + + this.logger.warn( + `Ignored unrecognized lifecycle event ${unhandledLifecycleEvent} for message channel ${messageChannel.id}`, + ); + } + } + } } diff --git a/packages/twenty-server/src/modules/connected-account-sync-webhooks/types/microsoft-graph-notification.type.ts b/packages/twenty-server/src/modules/connected-account-sync-webhooks/types/microsoft-graph-notification.type.ts index 4d022d3863..c98b980e52 100644 --- a/packages/twenty-server/src/modules/connected-account-sync-webhooks/types/microsoft-graph-notification.type.ts +++ b/packages/twenty-server/src/modules/connected-account-sync-webhooks/types/microsoft-graph-notification.type.ts @@ -1,10 +1,10 @@ -export type MicrosoftGraphNotification = { +import { type ChangeNotification } from '@microsoft/microsoft-graph-types'; + +export type MicrosoftGraphNotification = Omit< + ChangeNotification, + 'subscriptionId' +> & { subscriptionId: string; - clientState?: string; - changeType?: string; - resource?: string; - lifecycleEvent?: string; - resourceData?: { id?: string } | null; }; export type MicrosoftGraphNotificationPayload = { diff --git a/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/drivers/exceptions/webhook-subscription-driver.exception.ts b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/drivers/exceptions/webhook-subscription-driver.exception.ts index 92fbb7ef54..7b88c19882 100644 --- a/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/drivers/exceptions/webhook-subscription-driver.exception.ts +++ b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/drivers/exceptions/webhook-subscription-driver.exception.ts @@ -1,6 +1,6 @@ import { type MessageDescriptor } from '@lingui/core'; import { msg } from '@lingui/core/macro'; -import { assertUnreachable } from 'twenty-shared/utils'; +import { assertUnreachable, isDefined } from 'twenty-shared/utils'; import { CustomException } from 'src/utils/custom-exception'; @@ -8,15 +8,26 @@ export enum WebhookSubscriptionDriverExceptionCode { PROVIDER_NOT_CONFIGURED = 'PROVIDER_NOT_CONFIGURED', PROVIDER_RESPONSE_INVALID = 'PROVIDER_RESPONSE_INVALID', UNSUPPORTED_PROVIDER = 'UNSUPPORTED_PROVIDER', + NOT_FOUND = 'NOT_FOUND', + INSUFFICIENT_PERMISSIONS = 'INSUFFICIENT_PERMISSIONS', + TEMPORARY_ERROR = 'TEMPORARY_ERROR', + UNKNOWN = 'UNKNOWN', } const getWebhookSubscriptionDriverExceptionUserFriendlyMessage = ( code: WebhookSubscriptionDriverExceptionCode, ) => { switch (code) { + case WebhookSubscriptionDriverExceptionCode.NOT_FOUND: + return msg`The subscription is no longer available on the provider.`; + case WebhookSubscriptionDriverExceptionCode.INSUFFICIENT_PERMISSIONS: + return msg`The provider denied access to this account. Please reconnect it.`; + case WebhookSubscriptionDriverExceptionCode.TEMPORARY_ERROR: + return msg`The provider is temporarily unavailable. Please try again later.`; case WebhookSubscriptionDriverExceptionCode.PROVIDER_NOT_CONFIGURED: case WebhookSubscriptionDriverExceptionCode.PROVIDER_RESPONSE_INVALID: case WebhookSubscriptionDriverExceptionCode.UNSUPPORTED_PROVIDER: + case WebhookSubscriptionDriverExceptionCode.UNKNOWN: return msg`The webhook subscription could not be managed for this account.`; default: assertUnreachable(code); @@ -24,15 +35,24 @@ const getWebhookSubscriptionDriverExceptionUserFriendlyMessage = ( }; export class WebhookSubscriptionDriverException extends CustomException { + cause?: unknown; + constructor( message: string, code: WebhookSubscriptionDriverExceptionCode, - { userFriendlyMessage }: { userFriendlyMessage?: MessageDescriptor } = {}, + { + userFriendlyMessage, + cause, + }: { userFriendlyMessage?: MessageDescriptor; cause?: unknown } = {}, ) { super(message, code, { userFriendlyMessage: userFriendlyMessage ?? getWebhookSubscriptionDriverExceptionUserFriendlyMessage(code), }); + + if (isDefined(cause)) { + this.cause = cause; + } } } diff --git a/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/drivers/google/google-webhook-subscription.driver.dev.spec.ts b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/drivers/google/google-webhook-subscription.driver.dev.spec.ts new file mode 100644 index 0000000000..a369c48742 --- /dev/null +++ b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/drivers/google/google-webhook-subscription.driver.dev.spec.ts @@ -0,0 +1,107 @@ +import { google } from 'googleapis'; +import { WebhookSubscriptionChannelType } from 'twenty-shared/types'; + +import { type TwentyConfigService } from 'src/engine/core-modules/twenty-config/twenty-config.service'; +import { type GoogleOAuth2ClientProvider } from 'src/modules/connected-account/oauth2-client-manager/drivers/google/google-oauth2-client.provider'; +import { + WebhookSubscriptionDriverException, + WebhookSubscriptionDriverExceptionCode, +} from 'src/modules/connected-account/webhook-subscription-manager/drivers/exceptions/webhook-subscription-driver.exception'; +import { GoogleWebhookSubscriptionDriver } from 'src/modules/connected-account/webhook-subscription-manager/drivers/google/google-webhook-subscription.driver'; + +const accessToken = process.env.GOOGLE_DEV_ACCESS_TOKEN ?? ''; +const connectedAccountId = 'dev-connected-account'; + +const buildDriver = ({ + serverUrl = 'https://twenty-probe.invalid', + pubSubTopic = 'projects/twenty-auth-dev/topics/gmail', + token = accessToken, +}: { + serverUrl?: string; + pubSubTopic?: string; + token?: string; +} = {}) => { + const oAuth2Client = new google.auth.OAuth2(); + + oAuth2Client.setCredentials({ access_token: token }); + + const oAuth2ClientProvider = { + getClient: async () => oAuth2Client, + } as unknown as GoogleOAuth2ClientProvider; + + const twentyConfigService = { + get: (key: string) => + key === 'MESSAGING_GMAIL_PUBSUB_TOPIC' ? pubSubTopic : serverUrl, + } as unknown as TwentyConfigService; + + return new GoogleWebhookSubscriptionDriver( + oAuth2ClientProvider, + twentyConfigService, + ); +}; + +const expectDriverExceptionCode = async ( + operation: Promise, + expected: WebhookSubscriptionDriverExceptionCode, +) => { + await expect(operation).rejects.toBeInstanceOf( + WebhookSubscriptionDriverException, + ); + await expect(operation).rejects.toMatchObject({ code: expected }); +}; + +xdescribe('Google dev tests : webhook subscription driver', () => { + beforeAll(() => { + jest.useRealTimers(); + }); + + it('should map a stopped channel that no longer exists to NOT_FOUND', async () => { + await expectDriverExceptionCode( + buildDriver().deleteSubscription({ + connectedAccountId, + channelType: WebhookSubscriptionChannelType.CALENDAR, + externalSubscriptionId: '00000000-0000-4000-8000-000000000000', + externalResourceId: 'does-not-exist', + clientState: 'dev', + }), + WebhookSubscriptionDriverExceptionCode.NOT_FOUND, + ); + }); + + it('should map a non-https notification url to UNKNOWN', async () => { + await expectDriverExceptionCode( + buildDriver({ + serverUrl: 'http://insecure.example.com', + }).createSubscription( + connectedAccountId, + WebhookSubscriptionChannelType.CALENDAR, + 'dev', + ), + WebhookSubscriptionDriverExceptionCode.UNKNOWN, + ); + }); + + it('should map a pubsub topic owned by another project to UNKNOWN', async () => { + await expectDriverExceptionCode( + buildDriver({ + pubSubTopic: 'projects/twenty-probe-nonexistent/topics/nope', + }).createSubscription( + connectedAccountId, + WebhookSubscriptionChannelType.MESSAGING, + 'dev', + ), + WebhookSubscriptionDriverExceptionCode.UNKNOWN, + ); + }); + + it('should map an expired access token to INSUFFICIENT_PERMISSIONS', async () => { + await expectDriverExceptionCode( + buildDriver({ token: 'not-a-real-token' }).createSubscription( + connectedAccountId, + WebhookSubscriptionChannelType.CALENDAR, + 'dev', + ), + WebhookSubscriptionDriverExceptionCode.INSUFFICIENT_PERMISSIONS, + ); + }); +}); diff --git a/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/drivers/google/google-webhook-subscription.driver.ts b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/drivers/google/google-webhook-subscription.driver.ts index a27953b684..bcfda43370 100644 --- a/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/drivers/google/google-webhook-subscription.driver.ts +++ b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/drivers/google/google-webhook-subscription.driver.ts @@ -1,6 +1,7 @@ import { Injectable } from '@nestjs/common'; import { isNonEmptyString } from '@sniptt/guards'; +import { type GaxiosError } from 'gaxios'; import { type calendar_v3, type gmail_v1, google } from 'googleapis'; import { isDefined } from 'twenty-shared/utils'; import { v4 } from 'uuid'; @@ -13,6 +14,7 @@ import { WebhookSubscriptionDriverException, WebhookSubscriptionDriverExceptionCode, } from 'src/modules/connected-account/webhook-subscription-manager/drivers/exceptions/webhook-subscription-driver.exception'; +import { parseGoogleWebhookSubscriptionError } from 'src/modules/connected-account/webhook-subscription-manager/drivers/google/utils/parse-google-webhook-subscription-error.util'; import { type WebhookSubscriptionContext, type WebhookSubscriptionDriver, @@ -83,12 +85,16 @@ export class GoogleWebhookSubscriptionDriver implements WebhookSubscriptionDrive const gmailClient = await this.getGmailClient(connectedAccountId); - const { data } = await gmailClient.users.watch({ - userId: 'me', - requestBody: { - topicName: pubSubTopicName, - }, - }); + const { data } = await gmailClient.users + .watch({ + userId: 'me', + requestBody: { + topicName: pubSubTopicName, + }, + }) + .catch((error: GaxiosError) => { + throw parseGoogleWebhookSubscriptionError(error, { cause: error }); + }); if (!isDefined(data.expiration)) { throw new WebhookSubscriptionDriverException( @@ -109,7 +115,11 @@ export class GoogleWebhookSubscriptionDriver implements WebhookSubscriptionDrive ): Promise { const gmailClient = await this.getGmailClient(connectedAccountId); - await gmailClient.users.stop({ userId: 'me' }); + await gmailClient.users + .stop({ userId: 'me' }) + .catch((error: GaxiosError) => { + throw parseGoogleWebhookSubscriptionError(error, { cause: error }); + }); } private async watchPrimaryCalendar( @@ -121,16 +131,20 @@ export class GoogleWebhookSubscriptionDriver implements WebhookSubscriptionDrive const notificationAddress = `${this.twentyConfigService.get('SERVER_URL')}/webhooks/google/calendar`; const watchChannelId = v4(); - const { data } = await calendarClient.events.watch({ - calendarId: 'primary', - requestBody: { - id: watchChannelId, - type: 'web_hook', - address: notificationAddress, - token: clientState, - params: { ttl: String(GOOGLE_CALENDAR_WATCH_TTL_MS / 1000) }, - }, - }); + const { data } = await calendarClient.events + .watch({ + calendarId: 'primary', + requestBody: { + id: watchChannelId, + type: 'web_hook', + address: notificationAddress, + token: clientState, + params: { ttl: String(GOOGLE_CALENDAR_WATCH_TTL_MS / 1000) }, + }, + }) + .catch((error: GaxiosError) => { + throw parseGoogleWebhookSubscriptionError(error, { cause: error }); + }); if (!isDefined(data.resourceId) || !isDefined(data.expiration)) { throw new WebhookSubscriptionDriverException( @@ -160,12 +174,16 @@ export class GoogleWebhookSubscriptionDriver implements WebhookSubscriptionDrive context.connectedAccountId, ); - await calendarClient.channels.stop({ - requestBody: { - id: context.externalSubscriptionId, - resourceId: context.externalResourceId, - }, - }); + await calendarClient.channels + .stop({ + requestBody: { + id: context.externalSubscriptionId, + resourceId: context.externalResourceId, + }, + }) + .catch((error: GaxiosError) => { + throw parseGoogleWebhookSubscriptionError(error, { cause: error }); + }); } private async getGmailClient( diff --git a/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/drivers/google/utils/__tests__/parse-google-webhook-subscription-error.util.spec.ts b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/drivers/google/utils/__tests__/parse-google-webhook-subscription-error.util.spec.ts new file mode 100644 index 0000000000..231cef4893 --- /dev/null +++ b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/drivers/google/utils/__tests__/parse-google-webhook-subscription-error.util.spec.ts @@ -0,0 +1,172 @@ +import { type GaxiosError } from 'gaxios'; + +import { WebhookSubscriptionDriverExceptionCode } from 'src/modules/connected-account/webhook-subscription-manager/drivers/exceptions/webhook-subscription-driver.exception'; +import { parseGoogleWebhookSubscriptionError } from 'src/modules/connected-account/webhook-subscription-manager/drivers/google/utils/parse-google-webhook-subscription-error.util'; + +const buildGaxiosError = ({ + status, + reason, + message = 'error', +}: { + status: number; + reason?: string; + message?: string; +}) => + ({ + response: { status, data: { error: { errors: [{ reason, message }] } } }, + }) as GaxiosError; + +describe('parseGoogleWebhookSubscriptionError', () => { + it('should return NOT_FOUND when stopping a channel that no longer exists', () => { + const exception = parseGoogleWebhookSubscriptionError( + buildGaxiosError({ + status: 404, + reason: 'notFound', + message: "Channel '01cb40eb' not found for project '904792538077'", + }), + ); + + expect(exception.code).toBe( + WebhookSubscriptionDriverExceptionCode.NOT_FOUND, + ); + }); + + it('should return INSUFFICIENT_PERMISSIONS on 401', () => { + const exception = parseGoogleWebhookSubscriptionError( + buildGaxiosError({ + status: 401, + reason: 'authError', + message: 'Invalid Credentials', + }), + ); + + expect(exception.code).toBe( + WebhookSubscriptionDriverExceptionCode.INSUFFICIENT_PERMISSIONS, + ); + }); + + it('should return TEMPORARY_ERROR when concurrent requests are throttled', () => { + const exception = parseGoogleWebhookSubscriptionError( + buildGaxiosError({ + status: 429, + reason: 'rateLimitExceeded', + message: 'Too many concurrent requests for user.', + }), + ); + + expect(exception.code).toBe( + WebhookSubscriptionDriverExceptionCode.TEMPORARY_ERROR, + ); + }); + + it('should return TEMPORARY_ERROR when the per-minute quota is exhausted', () => { + const exception = parseGoogleWebhookSubscriptionError( + buildGaxiosError({ + status: 403, + reason: 'rateLimitExceeded', + message: + "Quota exceeded for quota metric 'Queries' and limit 'Units per minute per user'", + }), + ); + + expect(exception.code).toBe( + WebhookSubscriptionDriverExceptionCode.TEMPORARY_ERROR, + ); + }); + + it('should return INSUFFICIENT_PERMISSIONS when the scope was not granted', () => { + const exception = parseGoogleWebhookSubscriptionError( + buildGaxiosError({ + status: 403, + reason: 'insufficientPermissions', + message: 'Request had insufficient authentication scopes.', + }), + ); + + expect(exception.code).toBe( + WebhookSubscriptionDriverExceptionCode.INSUFFICIENT_PERMISSIONS, + ); + }); + + it.each([429, 500, 502, 503, 504])( + 'should return TEMPORARY_ERROR on %i', + (status) => { + const exception = parseGoogleWebhookSubscriptionError( + buildGaxiosError({ status }), + ); + + expect(exception.code).toBe( + WebhookSubscriptionDriverExceptionCode.TEMPORARY_ERROR, + ); + }, + ); + + it.each([ + ['push.webhookUrlNotHttps', 'WebHook callback must be HTTPS'], + ['invalidArgument', 'Invalid topicName does not match projects/x/topics/*'], + ])('should return UNKNOWN for the 400 reason %s', (reason, message) => { + const exception = parseGoogleWebhookSubscriptionError( + buildGaxiosError({ status: 400, reason, message }), + ); + + expect(exception.code).toBe(WebhookSubscriptionDriverExceptionCode.UNKNOWN); + }); + + it('should return TEMPORARY_ERROR when the request never got a response', () => { + const exception = parseGoogleWebhookSubscriptionError({ + message: 'connect ECONNRESET', + } as GaxiosError); + + expect(exception.code).toBe( + WebhookSubscriptionDriverExceptionCode.TEMPORARY_ERROR, + ); + }); + + it('should return TEMPORARY_ERROR for a transient failed precondition', () => { + const exception = parseGoogleWebhookSubscriptionError( + buildGaxiosError({ + status: 400, + reason: 'failedPrecondition', + message: 'Precondition check failed.', + }), + ); + + expect(exception.code).toBe( + WebhookSubscriptionDriverExceptionCode.TEMPORARY_ERROR, + ); + }); + + it('should return INSUFFICIENT_PERMISSIONS when mail service is not enabled', () => { + const exception = parseGoogleWebhookSubscriptionError( + buildGaxiosError({ + status: 400, + reason: 'failedPrecondition', + message: 'Mail service not enabled for this account.', + }), + ); + + expect(exception.code).toBe( + WebhookSubscriptionDriverExceptionCode.INSUFFICIENT_PERMISSIONS, + ); + }); + + it('should keep the provider error as the exception cause', () => { + const providerError = buildGaxiosError({ status: 404, reason: 'notFound' }); + + const exception = parseGoogleWebhookSubscriptionError(providerError, { + cause: providerError, + }); + + expect(exception.cause).toBe(providerError); + }); + + it('should return INSUFFICIENT_PERMISSIONS when the refresh token was revoked', () => { + const exception = { + response: { status: 400, data: { error: 'invalid_grant' } }, + } as GaxiosError; + + expect(parseGoogleWebhookSubscriptionError(exception).code).toBe( + WebhookSubscriptionDriverExceptionCode.INSUFFICIENT_PERMISSIONS, + ); + }); +}); diff --git a/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/drivers/google/utils/parse-google-webhook-subscription-error.util.ts b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/drivers/google/utils/parse-google-webhook-subscription-error.util.ts new file mode 100644 index 0000000000..6719280976 --- /dev/null +++ b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/drivers/google/utils/parse-google-webhook-subscription-error.util.ts @@ -0,0 +1,110 @@ +import { type GaxiosError } from 'gaxios'; +import { isDefined } from 'twenty-shared/utils'; + +import { + WebhookSubscriptionDriverException, + WebhookSubscriptionDriverExceptionCode, +} from 'src/modules/connected-account/webhook-subscription-manager/drivers/exceptions/webhook-subscription-driver.exception'; + +export const parseGoogleWebhookSubscriptionError = ( + error: GaxiosError, + options?: { cause?: unknown }, +): WebhookSubscriptionDriverException => { + if (!isDefined(error.response)) { + return new WebhookSubscriptionDriverException( + `Google API transport error: ${error.message}`, + WebhookSubscriptionDriverExceptionCode.TEMPORARY_ERROR, + options, + ); + } + + const googleApiError = { + code: error.response?.status, + reason: + error.response?.data?.error?.errors?.[0].reason || + error.response?.data?.error || + 'Unknown reason', + message: + error.response?.data?.error?.errors?.[0].message || + error.response?.data?.error_description || + 'Unknown error', + }; + + switch (googleApiError.code) { + case 400: + if (googleApiError.reason === 'invalid_grant') { + return new WebhookSubscriptionDriverException( + googleApiError.message, + WebhookSubscriptionDriverExceptionCode.INSUFFICIENT_PERMISSIONS, + options, + ); + } + + if (googleApiError.reason === 'failedPrecondition') { + return new WebhookSubscriptionDriverException( + googleApiError.message, + googleApiError.message.includes('Mail service not enabled') + ? WebhookSubscriptionDriverExceptionCode.INSUFFICIENT_PERMISSIONS + : WebhookSubscriptionDriverExceptionCode.TEMPORARY_ERROR, + options, + ); + } + + return new WebhookSubscriptionDriverException( + googleApiError.message, + WebhookSubscriptionDriverExceptionCode.UNKNOWN, + options, + ); + + case 401: + return new WebhookSubscriptionDriverException( + googleApiError.message, + WebhookSubscriptionDriverExceptionCode.INSUFFICIENT_PERMISSIONS, + options, + ); + + case 403: + if ( + googleApiError.reason === 'rateLimitExceeded' || + googleApiError.reason === 'userRateLimitExceeded' || + googleApiError.reason === 'dailyLimitExceeded' + ) { + return new WebhookSubscriptionDriverException( + googleApiError.message, + WebhookSubscriptionDriverExceptionCode.TEMPORARY_ERROR, + options, + ); + } + + return new WebhookSubscriptionDriverException( + googleApiError.message, + WebhookSubscriptionDriverExceptionCode.INSUFFICIENT_PERMISSIONS, + options, + ); + + case 404: + return new WebhookSubscriptionDriverException( + googleApiError.message, + WebhookSubscriptionDriverExceptionCode.NOT_FOUND, + options, + ); + + case 429: + case 500: + case 502: + case 503: + case 504: + return new WebhookSubscriptionDriverException( + googleApiError.message, + WebhookSubscriptionDriverExceptionCode.TEMPORARY_ERROR, + options, + ); + + default: + return new WebhookSubscriptionDriverException( + googleApiError.message, + WebhookSubscriptionDriverExceptionCode.UNKNOWN, + options, + ); + } +}; diff --git a/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/drivers/microsoft/microsoft-webhook-subscription.driver.dev.spec.ts b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/drivers/microsoft/microsoft-webhook-subscription.driver.dev.spec.ts new file mode 100644 index 0000000000..08dd5452d7 --- /dev/null +++ b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/drivers/microsoft/microsoft-webhook-subscription.driver.dev.spec.ts @@ -0,0 +1,89 @@ +import { Client } from '@microsoft/microsoft-graph-client'; +import { WebhookSubscriptionChannelType } from 'twenty-shared/types'; + +import { type TwentyConfigService } from 'src/engine/core-modules/twenty-config/twenty-config.service'; +import { type MicrosoftOAuth2ClientProvider } from 'src/modules/connected-account/oauth2-client-manager/drivers/microsoft/microsoft-oauth2-client.provider'; +import { + WebhookSubscriptionDriverException, + WebhookSubscriptionDriverExceptionCode, +} from 'src/modules/connected-account/webhook-subscription-manager/drivers/exceptions/webhook-subscription-driver.exception'; +import { MicrosoftWebhookSubscriptionDriver } from 'src/modules/connected-account/webhook-subscription-manager/drivers/microsoft/microsoft-webhook-subscription.driver'; + +const accessToken = process.env.MICROSOFT_DEV_ACCESS_TOKEN ?? ''; +const connectedAccountId = 'dev-connected-account'; +const goneSubscriptionId = '00000000-0000-4000-8000-000000000000'; + +const buildDriver = ({ + serverUrl = 'https://twenty-probe-does-not-exist.invalid', + token = accessToken, +}: { serverUrl?: string; token?: string } = {}) => { + const oAuth2ClientProvider = { + getClient: async () => + Client.init({ authProvider: (done) => done(null, token) }), + } as unknown as MicrosoftOAuth2ClientProvider; + + const twentyConfigService = { + get: () => serverUrl, + } as unknown as TwentyConfigService; + + return new MicrosoftWebhookSubscriptionDriver( + oAuth2ClientProvider, + twentyConfigService, + ); +}; + +const context = { + connectedAccountId, + channelType: WebhookSubscriptionChannelType.CALENDAR, + externalSubscriptionId: goneSubscriptionId, + externalResourceId: null, + clientState: 'dev', +}; + +const expectDriverExceptionCode = async ( + operation: Promise, + expected: WebhookSubscriptionDriverExceptionCode, +) => { + await expect(operation).rejects.toBeInstanceOf( + WebhookSubscriptionDriverException, + ); + await expect(operation).rejects.toMatchObject({ code: expected }); +}; + +xdescribe('Microsoft dev tests : webhook subscription driver', () => { + beforeAll(() => { + jest.useRealTimers(); + }); + + it('should map renewing a removed subscription to NOT_FOUND', async () => { + await expectDriverExceptionCode( + buildDriver().renewSubscription(context), + WebhookSubscriptionDriverExceptionCode.NOT_FOUND, + ); + }); + + it('should map deleting a removed subscription to NOT_FOUND', async () => { + await expectDriverExceptionCode( + buildDriver().deleteSubscription(context), + WebhookSubscriptionDriverExceptionCode.NOT_FOUND, + ); + }); + + it('should map an unreachable notification url to UNKNOWN', async () => { + await expectDriverExceptionCode( + buildDriver().createSubscription( + connectedAccountId, + WebhookSubscriptionChannelType.CALENDAR, + 'dev', + ), + WebhookSubscriptionDriverExceptionCode.UNKNOWN, + ); + }); + + it('should map an expired access token to TEMPORARY_ERROR', async () => { + await expectDriverExceptionCode( + buildDriver({ token: 'not-a-real-token' }).renewSubscription(context), + WebhookSubscriptionDriverExceptionCode.TEMPORARY_ERROR, + ); + }); +}); diff --git a/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/drivers/microsoft/microsoft-webhook-subscription.driver.ts b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/drivers/microsoft/microsoft-webhook-subscription.driver.ts index 2dbdc10e72..8d7af1a70c 100644 --- a/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/drivers/microsoft/microsoft-webhook-subscription.driver.ts +++ b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/drivers/microsoft/microsoft-webhook-subscription.driver.ts @@ -1,5 +1,6 @@ import { Injectable } from '@nestjs/common'; +import { type GraphError } from '@microsoft/microsoft-graph-client'; import { type Subscription } from '@microsoft/microsoft-graph-types'; import { isDefined } from 'twenty-shared/utils'; @@ -17,6 +18,7 @@ import { type WebhookSubscriptionResult, } from 'src/modules/connected-account/webhook-subscription-manager/types/webhook-subscription-driver.type'; import { MicrosoftOAuth2ClientProvider } from 'src/modules/connected-account/oauth2-client-manager/drivers/microsoft/microsoft-oauth2-client.provider'; +import { parseMicrosoftWebhookSubscriptionError } from 'src/modules/connected-account/webhook-subscription-manager/drivers/microsoft/utils/parse-microsoft-webhook-subscription-error.util'; import { MICROSOFT_SUBSCRIPTION_TTL_BUFFER_MS } from './constants/microsoft-subscription-ttl-ms-buffer.constant'; type MicrosoftGraphResourceConfig = Pick< @@ -77,7 +79,10 @@ export class MicrosoftWebhookSubscriptionDriver implements WebhookSubscriptionDr const subscription: Subscription = await graphClient .api('/subscriptions') - .post(subscriptionPayload); + .post(subscriptionPayload) + .catch((error: GraphError) => { + throw parseMicrosoftWebhookSubscriptionError(error, { cause: error }); + }); return this.toResult(subscription); } @@ -100,7 +105,10 @@ export class MicrosoftWebhookSubscriptionDriver implements WebhookSubscriptionDr const renewedSubscription: Subscription = await graphClient .api(`/subscriptions/${context.externalSubscriptionId}`) - .patch(subscriptionPatch); + .patch(subscriptionPatch) + .catch((error: GraphError) => { + throw parseMicrosoftWebhookSubscriptionError(error, { cause: error }); + }); return this.toResult(renewedSubscription); } @@ -116,7 +124,10 @@ export class MicrosoftWebhookSubscriptionDriver implements WebhookSubscriptionDr await graphClient .api(`/subscriptions/${context.externalSubscriptionId}`) - .delete(); + .delete() + .catch((error: GraphError) => { + throw parseMicrosoftWebhookSubscriptionError(error, { cause: error }); + }); } private toResult(subscription: Subscription): WebhookSubscriptionResult { diff --git a/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/drivers/microsoft/utils/__tests__/parse-microsoft-webhook-subscription-error.util.spec.ts b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/drivers/microsoft/utils/__tests__/parse-microsoft-webhook-subscription-error.util.spec.ts new file mode 100644 index 0000000000..b30b11ae46 --- /dev/null +++ b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/drivers/microsoft/utils/__tests__/parse-microsoft-webhook-subscription-error.util.spec.ts @@ -0,0 +1,106 @@ +import { WebhookSubscriptionDriverExceptionCode } from 'src/modules/connected-account/webhook-subscription-manager/drivers/exceptions/webhook-subscription-driver.exception'; +import { parseMicrosoftWebhookSubscriptionError } from 'src/modules/connected-account/webhook-subscription-manager/drivers/microsoft/utils/parse-microsoft-webhook-subscription-error.util'; + +describe('parseMicrosoftWebhookSubscriptionError', () => { + it('should return INSUFFICIENT_PERMISSIONS when the mailbox is not enabled for the REST API', () => { + const exception = parseMicrosoftWebhookSubscriptionError({ + statusCode: 404, + code: 'MailboxNotEnabledForRESTAPI', + message: + 'The mailbox is either inactive, soft-deleted, or is hosted on-premise.', + }); + + expect(exception.code).toBe( + WebhookSubscriptionDriverExceptionCode.INSUFFICIENT_PERMISSIONS, + ); + }); + + it('should return NOT_FOUND when the subscription no longer exists', () => { + const exception = parseMicrosoftWebhookSubscriptionError({ + statusCode: 404, + code: 'ResourceNotFound', + message: 'The object was not found.', + }); + + expect(exception.code).toBe( + WebhookSubscriptionDriverExceptionCode.NOT_FOUND, + ); + }); + + it.each([ + ['InvalidRequest', 'Failed to resolve domain example.invalid'], + ['ValidationError', 'Subscription validation request failed.'], + ])('should return UNKNOWN for the 400 code %s', (code, message) => { + const exception = parseMicrosoftWebhookSubscriptionError({ + statusCode: 400, + code, + message, + }); + + expect(exception.code).toBe(WebhookSubscriptionDriverExceptionCode.UNKNOWN); + }); + + it('should return TEMPORARY_ERROR when a 400 carries no error body', () => { + const exception = parseMicrosoftWebhookSubscriptionError({ + statusCode: 400, + }); + + expect(exception.code).toBe( + WebhookSubscriptionDriverExceptionCode.TEMPORARY_ERROR, + ); + }); + + it('should keep the provider error as the exception cause', () => { + const providerError = new Error('graph exploded'); + + const exception = parseMicrosoftWebhookSubscriptionError( + { statusCode: 404, code: 'ResourceNotFound' }, + { cause: providerError }, + ); + + expect(exception.cause).toBe(providerError); + }); + + it('should return INSUFFICIENT_PERMISSIONS on 403', () => { + const exception = parseMicrosoftWebhookSubscriptionError({ + statusCode: 403, + code: 'ErrorAccessDenied', + message: 'Access is denied.', + }); + + expect(exception.code).toBe( + WebhookSubscriptionDriverExceptionCode.INSUFFICIENT_PERMISSIONS, + ); + }); + + it.each([401, 429, 500, 502, 503, 504, 509])( + 'should return TEMPORARY_ERROR on %i', + (statusCode) => { + const exception = parseMicrosoftWebhookSubscriptionError({ statusCode }); + + expect(exception.code).toBe( + WebhookSubscriptionDriverExceptionCode.TEMPORARY_ERROR, + ); + }, + ); + + it('should return TEMPORARY_ERROR when the application is throttled', () => { + const exception = parseMicrosoftWebhookSubscriptionError({ + statusCode: 429, + code: 'ApplicationThrottled', + message: 'Application is over its MailboxConcurrency limit.', + }); + + expect(exception.code).toBe( + WebhookSubscriptionDriverExceptionCode.TEMPORARY_ERROR, + ); + }); + + it('should return UNKNOWN on an unhandled status code', () => { + const exception = parseMicrosoftWebhookSubscriptionError({ + statusCode: 418, + }); + + expect(exception.code).toBe(WebhookSubscriptionDriverExceptionCode.UNKNOWN); + }); +}); diff --git a/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/drivers/microsoft/utils/parse-microsoft-webhook-subscription-error.util.ts b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/drivers/microsoft/utils/parse-microsoft-webhook-subscription-error.util.ts new file mode 100644 index 0000000000..5dea53b6ec --- /dev/null +++ b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/drivers/microsoft/utils/parse-microsoft-webhook-subscription-error.util.ts @@ -0,0 +1,85 @@ +import { isDefined } from 'twenty-shared/utils'; + +import { + WebhookSubscriptionDriverException, + WebhookSubscriptionDriverExceptionCode, +} from 'src/modules/connected-account/webhook-subscription-manager/drivers/exceptions/webhook-subscription-driver.exception'; + +const MICROSOFT_MAILBOX_NOT_ENABLED_FOR_REST_API_ERROR_CODE = + 'MailboxNotEnabledForRESTAPI'; + +export const parseMicrosoftWebhookSubscriptionError = ( + error: { + statusCode: number; + message?: string; + code?: string | null; + }, + options?: { cause?: unknown }, +): WebhookSubscriptionDriverException => { + switch (error.statusCode) { + case 400: + if (!isDefined(error.message)) { + return new WebhookSubscriptionDriverException( + 'Microsoft Graph API returned 400 with empty error body', + WebhookSubscriptionDriverExceptionCode.TEMPORARY_ERROR, + options, + ); + } + + return new WebhookSubscriptionDriverException( + `Invalid request to Microsoft Graph API: ${error.message}`, + WebhookSubscriptionDriverExceptionCode.UNKNOWN, + options, + ); + + case 401: + return new WebhookSubscriptionDriverException( + `Unauthorized access to Microsoft Graph API - code:${error.code} ${error.message}`, + WebhookSubscriptionDriverExceptionCode.TEMPORARY_ERROR, + options, + ); + + case 403: + return new WebhookSubscriptionDriverException( + `Forbidden access to Microsoft Graph API - code:${error.code} ${error.message}`, + WebhookSubscriptionDriverExceptionCode.INSUFFICIENT_PERMISSIONS, + options, + ); + + case 404: + if ( + error.code === MICROSOFT_MAILBOX_NOT_ENABLED_FOR_REST_API_ERROR_CODE + ) { + return new WebhookSubscriptionDriverException( + `Disabled, deleted, inactive or no licence Microsoft account - code:${error.code}`, + WebhookSubscriptionDriverExceptionCode.INSUFFICIENT_PERMISSIONS, + options, + ); + } + + return new WebhookSubscriptionDriverException( + `Not found - code:${error.code}`, + WebhookSubscriptionDriverExceptionCode.NOT_FOUND, + options, + ); + + case 429: + case 500: + case 502: + case 503: + case 504: + case 509: + return new WebhookSubscriptionDriverException( + `Microsoft Graph API ${error.code} ${error.statusCode} error: ${error.message}`, + WebhookSubscriptionDriverExceptionCode.TEMPORARY_ERROR, + options, + ); + + default: + return new WebhookSubscriptionDriverException( + `Microsoft Graph API unknown error: ${error.message} with status code ${error.statusCode}`, + WebhookSubscriptionDriverExceptionCode.UNKNOWN, + options, + ); + } +}; 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 8b48e56771..b954654cdf 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 @@ -14,7 +14,13 @@ import { MetricsService } from 'src/engine/core-modules/metrics/metrics.service' import { MetricsKeys } from 'src/engine/core-modules/metrics/types/metrics-keys.type'; import { CalendarChannelEntity } from 'src/engine/metadata-modules/calendar-channel/entities/calendar-channel.entity'; import { ConnectedAccountEntity } from 'src/engine/metadata-modules/connected-account/entities/connected-account.entity'; +import { + WebhookSubscriptionDriverException, + WebhookSubscriptionDriverExceptionCode, +} from 'src/modules/connected-account/webhook-subscription-manager/drivers/exceptions/webhook-subscription-driver.exception'; import { WebhookSubscriptionDriverFactory } from 'src/modules/connected-account/webhook-subscription-manager/services/webhook-subscription-driver-factory.service'; +import { WebhookSubscriptionExceptionHandlerService } from 'src/modules/connected-account/webhook-subscription-manager/services/webhook-subscription-exception-handler.service'; +import { WebhookSubscriptionStatusService } from 'src/modules/connected-account/webhook-subscription-manager/services/webhook-subscription-status.service'; import { type WebhookSubscriptionContext } from 'src/modules/connected-account/webhook-subscription-manager/types/webhook-subscription-driver.type'; @Injectable() @@ -27,6 +33,8 @@ export class CalendarWebhookSubscriptionService { private readonly webhookSubscriptionDriverFactory: WebhookSubscriptionDriverFactory, private readonly exceptionHandlerService: ExceptionHandlerService, private readonly metricsService: MetricsService, + private readonly webhookSubscriptionStatusService: WebhookSubscriptionStatusService, + private readonly webhookSubscriptionExceptionHandlerService: WebhookSubscriptionExceptionHandlerService, ) {} async createSubscription( @@ -78,13 +86,12 @@ export class CalendarWebhookSubscriptionService { clientState, ); - await this.calendarChannelRepository.update(calendarChannel.id, { - webhookSubscriptionExternalId: result.externalSubscriptionId, - webhookSubscriptionExternalResourceId: result.externalResourceId, - webhookSubscriptionClientState: clientState, - webhookSubscriptionStatus: WebhookSubscriptionStatus.ACTIVE, - webhookSubscriptionExpiresAt: result.expiresAt, - }); + await this.webhookSubscriptionStatusService.markAsActive( + WebhookSubscriptionChannelType.CALENDAR, + calendarChannel.id, + result, + clientState, + ); this.metricsService.incrementCounterBy({ key: MetricsKeys.ConnectedAccountWebhookSubscriptionCreated, @@ -92,11 +99,11 @@ export class CalendarWebhookSubscriptionService { attributes: this.buildMetricAttributes(connectedAccount.provider), }); } catch (error) { - await this.calendarChannelRepository.update(calendarChannel.id, { - webhookSubscriptionClientState: clientState, - webhookSubscriptionStatus: WebhookSubscriptionStatus.FAILED, - webhookSubscriptionExpiresAt: null, - }); + await this.webhookSubscriptionStatusService.resetPendingSubscription( + WebhookSubscriptionChannelType.CALENDAR, + calendarChannel.id, + clientState, + ); this.metricsService.incrementCounterBy({ key: MetricsKeys.ConnectedAccountWebhookSubscriptionCreationFailed, @@ -104,11 +111,15 @@ export class CalendarWebhookSubscriptionService { attributes: this.buildMetricAttributes(connectedAccount.provider), }); - this.exceptionHandlerService.captureExceptions([error], { - workspace: { id: workspaceId }, - }); + await this.webhookSubscriptionExceptionHandlerService.handleDriverException( + error, + 'CREATE', + WebhookSubscriptionChannelType.CALENDAR, + calendarChannel, + workspaceId, + ); - throw error; + return; } if (isDefined(previousSubscription)) { @@ -118,6 +129,30 @@ export class CalendarWebhookSubscriptionService { } } + async recreateSubscription({ + calendarChannelId, + workspaceId, + removedSubscriptionId, + }: { + calendarChannelId: string; + workspaceId: string; + removedSubscriptionId: string | null; + }): Promise { + const cleared = + await this.webhookSubscriptionStatusService.clearRemovedSubscription( + WebhookSubscriptionChannelType.CALENDAR, + calendarChannelId, + workspaceId, + removedSubscriptionId, + ); + + if (!cleared) { + return; + } + + await this.createSubscription(calendarChannelId, workspaceId); + } + async renewSubscription({ calendarChannelId, workspaceId, @@ -158,12 +193,11 @@ export class CalendarWebhookSubscriptionService { this.toContext(calendarChannel), ); - await this.calendarChannelRepository.update(calendarChannel.id, { - webhookSubscriptionExternalId: result.externalSubscriptionId, - webhookSubscriptionExternalResourceId: result.externalResourceId, - webhookSubscriptionStatus: WebhookSubscriptionStatus.ACTIVE, - webhookSubscriptionExpiresAt: result.expiresAt, - }); + await this.webhookSubscriptionStatusService.markAsActive( + WebhookSubscriptionChannelType.CALENDAR, + calendarChannel.id, + result, + ); this.metricsService.incrementCounterBy({ key: MetricsKeys.ConnectedAccountWebhookSubscriptionRenewed, @@ -171,21 +205,24 @@ export class CalendarWebhookSubscriptionService { attributes: this.buildMetricAttributes(connectedAccount.provider), }); } catch (error) { - await this.calendarChannelRepository.update(calendarChannel.id, { - webhookSubscriptionStatus: WebhookSubscriptionStatus.FAILED, - }); - this.metricsService.incrementCounterBy({ key: MetricsKeys.ConnectedAccountWebhookSubscriptionRenewalFailed, amount: 1, attributes: this.buildMetricAttributes(connectedAccount.provider), }); - this.exceptionHandlerService.captureExceptions([error], { - workspace: { id: calendarChannel.workspaceId }, - }); + const recoveryAction = + await this.webhookSubscriptionExceptionHandlerService.handleDriverException( + error, + 'RENEW', + WebhookSubscriptionChannelType.CALENDAR, + calendarChannel, + calendarChannel.workspaceId, + ); - throw error; + if (recoveryAction === 'RECREATE') { + await this.createSubscription(calendarChannelId, workspaceId); + } } } @@ -225,6 +262,13 @@ export class CalendarWebhookSubscriptionService { attributes: this.buildMetricAttributes(connectedAccount.provider), }); } catch (error) { + if ( + error instanceof WebhookSubscriptionDriverException && + error.code === WebhookSubscriptionDriverExceptionCode.NOT_FOUND + ) { + return; + } + this.metricsService.incrementCounterBy({ key: MetricsKeys.ConnectedAccountWebhookSubscriptionDeletionFailed, amount: 1, 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 e11b6f07fb..33912d4bbc 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 @@ -14,7 +14,13 @@ import { MetricsService } from 'src/engine/core-modules/metrics/metrics.service' import { MetricsKeys } from 'src/engine/core-modules/metrics/types/metrics-keys.type'; import { ConnectedAccountEntity } from 'src/engine/metadata-modules/connected-account/entities/connected-account.entity'; import { MessageChannelEntity } from 'src/engine/metadata-modules/message-channel/entities/message-channel.entity'; +import { + WebhookSubscriptionDriverException, + WebhookSubscriptionDriverExceptionCode, +} from 'src/modules/connected-account/webhook-subscription-manager/drivers/exceptions/webhook-subscription-driver.exception'; import { WebhookSubscriptionDriverFactory } from 'src/modules/connected-account/webhook-subscription-manager/services/webhook-subscription-driver-factory.service'; +import { WebhookSubscriptionExceptionHandlerService } from 'src/modules/connected-account/webhook-subscription-manager/services/webhook-subscription-exception-handler.service'; +import { WebhookSubscriptionStatusService } from 'src/modules/connected-account/webhook-subscription-manager/services/webhook-subscription-status.service'; import { type WebhookSubscriptionContext } from 'src/modules/connected-account/webhook-subscription-manager/types/webhook-subscription-driver.type'; @Injectable() @@ -27,6 +33,8 @@ export class MessagingWebhookSubscriptionService { private readonly webhookSubscriptionDriverFactory: WebhookSubscriptionDriverFactory, private readonly exceptionHandlerService: ExceptionHandlerService, private readonly metricsService: MetricsService, + private readonly webhookSubscriptionStatusService: WebhookSubscriptionStatusService, + private readonly webhookSubscriptionExceptionHandlerService: WebhookSubscriptionExceptionHandlerService, ) {} async createSubscription( @@ -78,12 +86,12 @@ export class MessagingWebhookSubscriptionService { clientState, ); - await this.messageChannelRepository.update(messageChannel.id, { - webhookSubscriptionExternalId: result.externalSubscriptionId, - webhookSubscriptionClientState: clientState, - webhookSubscriptionStatus: WebhookSubscriptionStatus.ACTIVE, - webhookSubscriptionExpiresAt: result.expiresAt, - }); + await this.webhookSubscriptionStatusService.markAsActive( + WebhookSubscriptionChannelType.MESSAGING, + messageChannel.id, + result, + clientState, + ); this.metricsService.incrementCounterBy({ key: MetricsKeys.ConnectedAccountWebhookSubscriptionCreated, @@ -91,11 +99,11 @@ export class MessagingWebhookSubscriptionService { attributes: this.buildMetricAttributes(connectedAccount.provider), }); } catch (error) { - await this.messageChannelRepository.update(messageChannel.id, { - webhookSubscriptionClientState: clientState, - webhookSubscriptionStatus: WebhookSubscriptionStatus.FAILED, - webhookSubscriptionExpiresAt: null, - }); + await this.webhookSubscriptionStatusService.resetPendingSubscription( + WebhookSubscriptionChannelType.MESSAGING, + messageChannel.id, + clientState, + ); this.metricsService.incrementCounterBy({ key: MetricsKeys.ConnectedAccountWebhookSubscriptionCreationFailed, @@ -103,11 +111,15 @@ export class MessagingWebhookSubscriptionService { attributes: this.buildMetricAttributes(connectedAccount.provider), }); - this.exceptionHandlerService.captureExceptions([error], { - workspace: { id: workspaceId }, - }); + await this.webhookSubscriptionExceptionHandlerService.handleDriverException( + error, + 'CREATE', + WebhookSubscriptionChannelType.MESSAGING, + messageChannel, + workspaceId, + ); - throw error; + return; } if (isDefined(previousSubscription)) { @@ -117,6 +129,30 @@ export class MessagingWebhookSubscriptionService { } } + async recreateSubscription({ + messageChannelId, + workspaceId, + removedSubscriptionId, + }: { + messageChannelId: string; + workspaceId: string; + removedSubscriptionId: string | null; + }): Promise { + const cleared = + await this.webhookSubscriptionStatusService.clearRemovedSubscription( + WebhookSubscriptionChannelType.MESSAGING, + messageChannelId, + workspaceId, + removedSubscriptionId, + ); + + if (!cleared) { + return; + } + + await this.createSubscription(messageChannelId, workspaceId); + } + async renewSubscription({ messageChannelId, workspaceId, @@ -157,11 +193,11 @@ export class MessagingWebhookSubscriptionService { this.toContext(messageChannel), ); - await this.messageChannelRepository.update(messageChannel.id, { - webhookSubscriptionExternalId: result.externalSubscriptionId, - webhookSubscriptionStatus: WebhookSubscriptionStatus.ACTIVE, - webhookSubscriptionExpiresAt: result.expiresAt, - }); + await this.webhookSubscriptionStatusService.markAsActive( + WebhookSubscriptionChannelType.MESSAGING, + messageChannel.id, + result, + ); this.metricsService.incrementCounterBy({ key: MetricsKeys.ConnectedAccountWebhookSubscriptionRenewed, @@ -169,21 +205,24 @@ export class MessagingWebhookSubscriptionService { attributes: this.buildMetricAttributes(connectedAccount.provider), }); } catch (error) { - await this.messageChannelRepository.update(messageChannel.id, { - webhookSubscriptionStatus: WebhookSubscriptionStatus.FAILED, - }); - this.metricsService.incrementCounterBy({ key: MetricsKeys.ConnectedAccountWebhookSubscriptionRenewalFailed, amount: 1, attributes: this.buildMetricAttributes(connectedAccount.provider), }); - this.exceptionHandlerService.captureExceptions([error], { - workspace: { id: messageChannel.workspaceId }, - }); + const recoveryAction = + await this.webhookSubscriptionExceptionHandlerService.handleDriverException( + error, + 'RENEW', + WebhookSubscriptionChannelType.MESSAGING, + messageChannel, + messageChannel.workspaceId, + ); - throw error; + if (recoveryAction === 'RECREATE') { + await this.createSubscription(messageChannelId, workspaceId); + } } } @@ -223,6 +262,13 @@ export class MessagingWebhookSubscriptionService { attributes: this.buildMetricAttributes(connectedAccount.provider), }); } catch (error) { + if ( + error instanceof WebhookSubscriptionDriverException && + error.code === WebhookSubscriptionDriverExceptionCode.NOT_FOUND + ) { + return; + } + this.metricsService.incrementCounterBy({ key: MetricsKeys.ConnectedAccountWebhookSubscriptionDeletionFailed, amount: 1, diff --git a/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/services/webhook-subscription-exception-handler.service.ts b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/services/webhook-subscription-exception-handler.service.ts new file mode 100644 index 0000000000..4b8047407c --- /dev/null +++ b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/services/webhook-subscription-exception-handler.service.ts @@ -0,0 +1,144 @@ +import { Injectable } from '@nestjs/common'; + +import { type WebhookSubscriptionChannelType } from 'twenty-shared/types'; + +import { ExceptionHandlerService } from 'src/engine/core-modules/exception-handler/exception-handler.service'; +import { + WebhookSubscriptionDriverException, + WebhookSubscriptionDriverExceptionCode, +} from 'src/modules/connected-account/webhook-subscription-manager/drivers/exceptions/webhook-subscription-driver.exception'; +import { WebhookSubscriptionStatusService } from 'src/modules/connected-account/webhook-subscription-manager/services/webhook-subscription-status.service'; +import { + type WebhookSubscribableChannel, + type WebhookSubscriptionOperation, + type WebhookSubscriptionRecoveryAction, +} from 'src/modules/connected-account/webhook-subscription-manager/types/webhook-subscription-driver.type'; + +type WebhookSubscribableChannelReference = Pick< + WebhookSubscribableChannel, + 'id' | 'webhookSubscriptionExternalId' +>; + +@Injectable() +export class WebhookSubscriptionExceptionHandlerService { + constructor( + private readonly webhookSubscriptionStatusService: WebhookSubscriptionStatusService, + private readonly exceptionHandlerService: ExceptionHandlerService, + ) {} + + public async handleDriverException( + exception: unknown, + operation: WebhookSubscriptionOperation, + channelType: WebhookSubscriptionChannelType, + channel: WebhookSubscribableChannelReference, + workspaceId: string, + ): Promise { + if (exception instanceof WebhookSubscriptionDriverException) { + switch (exception.code) { + case WebhookSubscriptionDriverExceptionCode.NOT_FOUND: + return await this.handleNotFoundException( + operation, + channelType, + channel, + workspaceId, + ); + case WebhookSubscriptionDriverExceptionCode.INSUFFICIENT_PERMISSIONS: + return await this.handleInsufficientPermissionsException( + channelType, + channel, + ); + case WebhookSubscriptionDriverExceptionCode.TEMPORARY_ERROR: + return await this.handleTemporaryException( + exception, + channelType, + channel, + ); + case WebhookSubscriptionDriverExceptionCode.PROVIDER_NOT_CONFIGURED: + case WebhookSubscriptionDriverExceptionCode.PROVIDER_RESPONSE_INVALID: + case WebhookSubscriptionDriverExceptionCode.UNSUPPORTED_PROVIDER: + case WebhookSubscriptionDriverExceptionCode.UNKNOWN: + default: + return await this.handleUnknownException( + exception, + channelType, + channel, + workspaceId, + ); + } + } + + return await this.handleUnknownException( + exception, + channelType, + channel, + workspaceId, + ); + } + + private async handleNotFoundException( + operation: WebhookSubscriptionOperation, + channelType: WebhookSubscriptionChannelType, + channel: WebhookSubscribableChannelReference, + workspaceId: string, + ): Promise { + if (operation === 'CREATE') { + return await this.handleInsufficientPermissionsException( + channelType, + channel, + ); + } + + const cleared = + await this.webhookSubscriptionStatusService.clearRemovedSubscription( + channelType, + channel.id, + workspaceId, + channel.webhookSubscriptionExternalId, + ); + + return cleared ? 'RECREATE' : 'NONE'; + } + + private async handleInsufficientPermissionsException( + channelType: WebhookSubscriptionChannelType, + channel: WebhookSubscribableChannelReference, + ): Promise { + await this.webhookSubscriptionStatusService.markAsExpired( + channelType, + channel.id, + ); + + return 'NONE'; + } + + private async handleTemporaryException( + exception: WebhookSubscriptionDriverException, + channelType: WebhookSubscriptionChannelType, + channel: WebhookSubscribableChannelReference, + ): Promise { + await this.webhookSubscriptionStatusService.markAsFailed( + channelType, + channel.id, + ); + + throw exception; + } + + private async handleUnknownException( + exception: unknown, + channelType: WebhookSubscriptionChannelType, + channel: WebhookSubscribableChannelReference, + workspaceId: string, + ): Promise { + await this.webhookSubscriptionStatusService.markAsFailed( + channelType, + channel.id, + ); + + this.exceptionHandlerService.captureExceptions([exception], { + workspace: { id: workspaceId }, + }); + + throw exception; + } +} diff --git a/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/services/webhook-subscription-status.service.ts b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/services/webhook-subscription-status.service.ts new file mode 100644 index 0000000000..de059b350d --- /dev/null +++ b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/services/webhook-subscription-status.service.ts @@ -0,0 +1,120 @@ +import { Injectable } from '@nestjs/common'; +import { InjectRepository } from '@nestjs/typeorm'; + +import { + WebhookSubscriptionChannelType, + WebhookSubscriptionStatus, +} from 'twenty-shared/types'; +import { isDefined } from 'twenty-shared/utils'; +import { Repository } from 'typeorm'; +import { type QueryDeepPartialEntity } from 'typeorm/query-builder/QueryPartialEntity'; + +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 { + type WebhookSubscribableChannel, + type WebhookSubscriptionResult, +} from 'src/modules/connected-account/webhook-subscription-manager/types/webhook-subscription-driver.type'; + +@Injectable() +export class WebhookSubscriptionStatusService { + constructor( + @InjectRepository(MessageChannelEntity) + private readonly messageChannelRepository: Repository, + @InjectRepository(CalendarChannelEntity) + private readonly calendarChannelRepository: Repository, + ) {} + + public async markAsActive( + channelType: WebhookSubscriptionChannelType, + channelId: string, + result: WebhookSubscriptionResult, + clientState?: string, + ) { + await this.update(channelType, channelId, { + webhookSubscriptionExternalId: result.externalSubscriptionId, + webhookSubscriptionStatus: WebhookSubscriptionStatus.ACTIVE, + webhookSubscriptionExpiresAt: result.expiresAt, + ...(isDefined(clientState) + ? { webhookSubscriptionClientState: clientState } + : {}), + ...(channelType === WebhookSubscriptionChannelType.CALENDAR + ? { webhookSubscriptionExternalResourceId: result.externalResourceId } + : {}), + }); + } + + public async markAsFailed( + channelType: WebhookSubscriptionChannelType, + channelId: string, + ) { + await this.update(channelType, channelId, { + webhookSubscriptionStatus: WebhookSubscriptionStatus.FAILED, + }); + } + + public async markAsExpired( + channelType: WebhookSubscriptionChannelType, + channelId: string, + ) { + await this.update(channelType, channelId, { + webhookSubscriptionStatus: WebhookSubscriptionStatus.EXPIRED, + }); + } + + public async resetPendingSubscription( + channelType: WebhookSubscriptionChannelType, + channelId: string, + clientState: string, + ) { + await this.update(channelType, channelId, { + webhookSubscriptionClientState: clientState, + webhookSubscriptionExpiresAt: null, + }); + } + + public async clearRemovedSubscription( + channelType: WebhookSubscriptionChannelType, + channelId: string, + workspaceId: string, + removedSubscriptionId: string | null, + ): Promise { + const { affected } = await this.getRepository(channelType).update( + { + id: channelId, + workspaceId, + ...(isDefined(removedSubscriptionId) + ? { webhookSubscriptionExternalId: removedSubscriptionId } + : {}), + }, + { + webhookSubscriptionExternalId: null, + webhookSubscriptionStatus: WebhookSubscriptionStatus.FAILED, + webhookSubscriptionExpiresAt: null, + ...(channelType === WebhookSubscriptionChannelType.CALENDAR + ? { webhookSubscriptionExternalResourceId: null } + : {}), + }, + ); + + return isDefined(affected) && affected > 0; + } + + private update( + channelType: WebhookSubscriptionChannelType, + channelId: string, + payload: QueryDeepPartialEntity, + ) { + return this.getRepository(channelType).update(channelId, payload); + } + + private getRepository( + channelType: WebhookSubscriptionChannelType, + ): Repository { + return ( + channelType === WebhookSubscriptionChannelType.CALENDAR + ? this.calendarChannelRepository + : this.messageChannelRepository + ) as Repository; + } +} diff --git a/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/types/webhook-subscription-driver.type.ts b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/types/webhook-subscription-driver.type.ts index 2f05b0628b..d518ec554d 100644 --- a/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/types/webhook-subscription-driver.type.ts +++ b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/types/webhook-subscription-driver.type.ts @@ -1,5 +1,16 @@ import { type WebhookSubscriptionChannelType } from 'twenty-shared/types'; +import { type CalendarChannelEntity } from 'src/engine/metadata-modules/calendar-channel/entities/calendar-channel.entity'; +import { type MessageChannelEntity } from 'src/engine/metadata-modules/message-channel/entities/message-channel.entity'; + +export type WebhookSubscribableChannel = + | MessageChannelEntity + | CalendarChannelEntity; + +export type WebhookSubscriptionOperation = 'CREATE' | 'RENEW'; + +export type WebhookSubscriptionRecoveryAction = 'NONE' | 'RECREATE'; + export type WebhookSubscriptionResult = { externalSubscriptionId: string | null; externalResourceId: string | null; 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 fbde141ba2..6b3a858541 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 @@ -16,6 +16,8 @@ import { WebhookSubscriptionRenewalCronJob } from 'src/modules/connected-account 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'; import { MessagingWebhookSubscriptionService } from 'src/modules/connected-account/webhook-subscription-manager/services/messaging-webhook-subscription.service'; +import { WebhookSubscriptionExceptionHandlerService } from 'src/modules/connected-account/webhook-subscription-manager/services/webhook-subscription-exception-handler.service'; +import { WebhookSubscriptionStatusService } from 'src/modules/connected-account/webhook-subscription-manager/services/webhook-subscription-status.service'; import { WebhookSubscriptionManagerModule } from 'src/modules/connected-account/webhook-subscription-manager/webhook-subscription-manager.module'; @Module({ @@ -32,6 +34,8 @@ import { WebhookSubscriptionManagerModule } from 'src/modules/connected-account/ ]), ], providers: [ + WebhookSubscriptionStatusService, + WebhookSubscriptionExceptionHandlerService, MessagingWebhookSubscriptionService, CalendarWebhookSubscriptionService, WebhookSubscriptionChannelDeletedListener, diff --git a/packages/twenty-server/test/integration/microsoft/mocks/microsoft-webhook-subscription-handlers.util.ts b/packages/twenty-server/test/integration/microsoft/mocks/microsoft-webhook-subscription-handlers.util.ts new file mode 100644 index 0000000000..b12278b3d1 --- /dev/null +++ b/packages/twenty-server/test/integration/microsoft/mocks/microsoft-webhook-subscription-handlers.util.ts @@ -0,0 +1,84 @@ +import { type Subscription } from '@microsoft/microsoft-graph-types'; +import { http, HttpResponse } from 'msw'; + +import { type MswHandler } from 'test/integration/utils/http-mock.util'; + +export type MicrosoftSubscriptionStore = { + created: Subscription[]; + renewed: string[]; + deleted: string[]; + reset: () => void; +}; + +export const createMicrosoftSubscriptionStore = + (): MicrosoftSubscriptionStore => { + const store: MicrosoftSubscriptionStore = { + created: [], + renewed: [], + deleted: [], + reset: () => { + store.created = []; + store.renewed = []; + store.deleted = []; + }, + }; + + return store; + }; + +const subscriptionId = (request: Request) => + new URL(request.url).pathname.split('/').pop() ?? ''; + +const resourceNotFound = () => + HttpResponse.json( + { + error: { + code: 'ResourceNotFound', + message: 'The object was not found.', + }, + }, + { status: 404 }, + ); + +export const microsoftWebhookSubscriptionHandlers = ( + store: MicrosoftSubscriptionStore, + { renewalFails = false }: { renewalFails?: boolean } = {}, +): MswHandler[] => [ + http.post('*/subscriptions', async ({ request }) => { + const payload = (await request.json()) as Subscription; + const subscription: Subscription = { + ...payload, + id: `subscription-${store.created.length + 1}`, + expirationDateTime: + payload.expirationDateTime ?? + new Date(Date.now() + 3600 * 1000).toISOString(), + }; + + store.created.push(subscription); + + return HttpResponse.json(subscription); + }), + http.patch('*/subscriptions/*', async ({ request }) => { + const id = subscriptionId(request); + + if (renewalFails) { + return resourceNotFound(); + } + + store.renewed.push(id); + + const payload = (await request.json()) as Subscription; + + return HttpResponse.json({ + id, + expirationDateTime: + payload.expirationDateTime ?? + new Date(Date.now() + 3600 * 1000).toISOString(), + }); + }), + http.delete('*/subscriptions/*', ({ request }) => { + store.deleted.push(subscriptionId(request)); + + return new HttpResponse(null, { status: 204 }); + }), +]; diff --git a/packages/twenty-server/test/integration/microsoft/mocks/setup-microsoft-mock.util.ts b/packages/twenty-server/test/integration/microsoft/mocks/setup-microsoft-mock.util.ts index 8ec39b140d..0a91c9ea04 100644 --- a/packages/twenty-server/test/integration/microsoft/mocks/setup-microsoft-mock.util.ts +++ b/packages/twenty-server/test/integration/microsoft/mocks/setup-microsoft-mock.util.ts @@ -4,6 +4,11 @@ import { setupHttpMock } from 'test/integration/utils/http-mock.util'; import { microsoftAuthHandlers } from 'test/integration/microsoft/mocks/microsoft-auth-handlers.util'; import { microsoftCalendarEventsHandlers } from 'test/integration/microsoft/mocks/microsoft-calendar-events-handlers.util'; import { microsoftMailboxHandlers } from 'test/integration/microsoft/mocks/microsoft-mailbox-handlers.util'; +import { + createMicrosoftSubscriptionStore, + microsoftWebhookSubscriptionHandlers, + type MicrosoftSubscriptionStore, +} from 'test/integration/microsoft/mocks/microsoft-webhook-subscription-handlers.util'; import { createMockEntityStore, type MockEntityStore, @@ -16,10 +21,12 @@ const DEFAULT_FOLDERS: MailFolder[] = [ export type MicrosoftMock = { folders: MockEntityStore; + subscriptions: MicrosoftSubscriptionStore; serveCalendarEvents: ( events: Event[], options?: { deltaToken?: string }, ) => void; + failSubscriptionRenewal: () => void; }; export const setupMicrosoftMock = ({ @@ -34,16 +41,26 @@ export const setupMicrosoftMock = ({ (folder) => folder.id ?? '', ); + const subscriptionStore = createMicrosoftSubscriptionStore(); + const httpMock = setupHttpMock( ...microsoftAuthHandlers(handle), ...microsoftMailboxHandlers(folderStore), + ...microsoftWebhookSubscriptionHandlers(subscriptionStore), ); return { folders: folderStore, + subscriptions: subscriptionStore, serveCalendarEvents: ( events, { deltaToken = 'mock-calendar-delta-token' } = {}, ) => httpMock.use(...microsoftCalendarEventsHandlers(events, deltaToken)), + failSubscriptionRenewal: () => + httpMock.use( + ...microsoftWebhookSubscriptionHandlers(subscriptionStore, { + renewalFails: true, + }), + ), }; }; diff --git a/packages/twenty-server/test/integration/microsoft/webhook/lifecycle-notification.integration-spec.ts b/packages/twenty-server/test/integration/microsoft/webhook/lifecycle-notification.integration-spec.ts new file mode 100644 index 0000000000..5a41c7a5d8 --- /dev/null +++ b/packages/twenty-server/test/integration/microsoft/webhook/lifecycle-notification.integration-spec.ts @@ -0,0 +1,163 @@ +import request from 'supertest'; +import { + ConnectedAccountProvider, + WebhookSubscriptionStatus, +} from 'twenty-shared/types'; + +import { CalendarChannelEntity } from 'src/engine/metadata-modules/calendar-channel/entities/calendar-channel.entity'; + +import { setupMicrosoftMock } from 'test/integration/microsoft/mocks/setup-microsoft-mock.util'; +import { connectMessagingAccount } from 'test/integration/utils/connect-messaging-account.util'; +import { getCoreRepository } from 'test/integration/utils/get-core-repository.util'; +import { waitForAllJobsToFinish } from 'test/integration/utils/wait-for-all-jobs-to-finish.util'; + +const HANDLE = 'microsoft-webhook-lifecycle@apple.dev'; +const CLIENT_STATE = 'lifecycle-client-state'; +const REMOVED_SUBSCRIPTION_ID = 'subscription-removed-by-microsoft'; + +const postLifecycleNotification = (lifecycleEvent: string, overrides = {}) => + request(`http://localhost:${APP_PORT}`) + .post('/webhooks/microsoft/calendar') + .send({ + value: [ + { + subscriptionId: REMOVED_SUBSCRIPTION_ID, + clientState: CLIENT_STATE, + lifecycleEvent, + tenantId: 'mock-tenant', + ...overrides, + }, + ], + }); + +describe('Microsoft webhook lifecycle notifications (integration)', () => { + const microsoft = setupMicrosoftMock({ handle: HANDLE }); + + let account: Awaited>; + + const calendarChannelRepository = () => + getCoreRepository(CalendarChannelEntity); + + const readChannel = async () => + await calendarChannelRepository().findOneOrFail({ + where: { id: account.calendarChannelId }, + }); + + const giveChannelAnActiveSubscription = async () => { + await calendarChannelRepository().update(account.calendarChannelId, { + webhookSubscriptionExternalId: REMOVED_SUBSCRIPTION_ID, + webhookSubscriptionClientState: CLIENT_STATE, + webhookSubscriptionStatus: WebhookSubscriptionStatus.ACTIVE, + webhookSubscriptionExpiresAt: new Date(Date.now() + 3600 * 1000), + }); + }; + + beforeAll(async () => { + account = await connectMessagingAccount({ + provider: ConnectedAccountProvider.MICROSOFT, + handle: HANDLE, + }); + }, 120000); + + beforeEach(async () => { + microsoft.subscriptions.reset(); + await giveChannelAnActiveSubscription(); + }); + + afterAll(async () => { + await account?.cleanup().catch(() => undefined); + }); + + it('creates a replacement subscription when Microsoft reports it was removed', async () => { + await postLifecycleNotification('subscriptionRemoved').expect(200); + + await waitForAllJobsToFinish(); + + expect(microsoft.subscriptions.created).toHaveLength(1); + + const channel = await readChannel(); + + expect(channel.webhookSubscriptionExternalId).not.toBe( + REMOVED_SUBSCRIPTION_ID, + ); + expect(channel.webhookSubscriptionStatus).toBe( + WebhookSubscriptionStatus.ACTIVE, + ); + }, 60000); + + it('never patches the removed subscription', async () => { + await postLifecycleNotification('subscriptionRemoved').expect(200); + + await waitForAllJobsToFinish(); + + expect(microsoft.subscriptions.renewed).not.toContain( + REMOVED_SUBSCRIPTION_ID, + ); + }, 60000); + + it('renews in place when Microsoft asks for reauthorization', async () => { + await postLifecycleNotification('reauthorizationRequired').expect(200); + + await waitForAllJobsToFinish(); + + expect(microsoft.subscriptions.renewed).toContain(REMOVED_SUBSCRIPTION_ID); + expect(microsoft.subscriptions.created).toHaveLength(0); + + const channel = await readChannel(); + + expect(channel.webhookSubscriptionExternalId).toBe(REMOVED_SUBSCRIPTION_ID); + }, 60000); + + it('leaves the subscription untouched on a missed notification', async () => { + await postLifecycleNotification('missed').expect(200); + + await waitForAllJobsToFinish(); + + expect(microsoft.subscriptions.created).toHaveLength(0); + expect(microsoft.subscriptions.renewed).toHaveLength(0); + + const channel = await readChannel(); + + expect(channel.webhookSubscriptionExternalId).toBe(REMOVED_SUBSCRIPTION_ID); + }, 60000); + + it('ignores a lifecycle event it does not recognize', async () => { + await postLifecycleNotification('someFutureLifecycleEvent').expect(200); + + await waitForAllJobsToFinish(); + + expect(microsoft.subscriptions.created).toHaveLength(0); + expect(microsoft.subscriptions.renewed).toHaveLength(0); + }, 60000); + + it('ignores a notification whose client state does not match', async () => { + await postLifecycleNotification('subscriptionRemoved', { + clientState: 'forged-client-state', + }).expect(200); + + await waitForAllJobsToFinish(); + + expect(microsoft.subscriptions.created).toHaveLength(0); + + const channel = await readChannel(); + + expect(channel.webhookSubscriptionExternalId).toBe(REMOVED_SUBSCRIPTION_ID); + }, 60000); + + it('recreates the subscription when renewal reports the resource is gone', async () => { + microsoft.failSubscriptionRenewal(); + + await postLifecycleNotification('reauthorizationRequired').expect(200); + + await waitForAllJobsToFinish(); + + const channel = await readChannel(); + + expect(channel.webhookSubscriptionStatus).toBe( + WebhookSubscriptionStatus.ACTIVE, + ); + expect(channel.webhookSubscriptionExternalId).not.toBe( + REMOVED_SUBSCRIPTION_ID, + ); + }, 60000); +}); diff --git a/packages/twenty-server/test/integration/microsoft/webhook/messaging-lifecycle-notification.integration-spec.ts b/packages/twenty-server/test/integration/microsoft/webhook/messaging-lifecycle-notification.integration-spec.ts new file mode 100644 index 0000000000..7b795e201e --- /dev/null +++ b/packages/twenty-server/test/integration/microsoft/webhook/messaging-lifecycle-notification.integration-spec.ts @@ -0,0 +1,138 @@ +import request from 'supertest'; +import { + ConnectedAccountProvider, + WebhookSubscriptionStatus, +} from 'twenty-shared/types'; + +import { MessageChannelEntity } from 'src/engine/metadata-modules/message-channel/entities/message-channel.entity'; + +import { setupMicrosoftMock } from 'test/integration/microsoft/mocks/setup-microsoft-mock.util'; +import { connectMessagingAccount } from 'test/integration/utils/connect-messaging-account.util'; +import { getCoreRepository } from 'test/integration/utils/get-core-repository.util'; +import { waitForAllJobsToFinish } from 'test/integration/utils/wait-for-all-jobs-to-finish.util'; + +const HANDLE = 'microsoft-messaging-lifecycle@apple.dev'; +const CLIENT_STATE = 'messaging-lifecycle-client-state'; +const REMOVED_SUBSCRIPTION_ID = 'messaging-subscription-removed'; + +const postLifecycleNotification = (lifecycleEvent: string, overrides = {}) => + request(`http://localhost:${APP_PORT}`) + .post('/webhooks/microsoft/messaging') + .send({ + value: [ + { + subscriptionId: REMOVED_SUBSCRIPTION_ID, + clientState: CLIENT_STATE, + lifecycleEvent, + tenantId: 'mock-tenant', + ...overrides, + }, + ], + }); + +describe('Microsoft messaging webhook lifecycle notifications (integration)', () => { + const microsoft = setupMicrosoftMock({ handle: HANDLE }); + + let account: Awaited>; + + const messageChannelRepository = () => + getCoreRepository(MessageChannelEntity); + + const readChannel = async () => + await messageChannelRepository().findOneOrFail({ + where: { id: account.channelId }, + }); + + beforeAll(async () => { + account = await connectMessagingAccount({ + provider: ConnectedAccountProvider.MICROSOFT, + handle: HANDLE, + }); + }, 120000); + + beforeEach(async () => { + microsoft.subscriptions.reset(); + await messageChannelRepository().update(account.channelId, { + webhookSubscriptionExternalId: REMOVED_SUBSCRIPTION_ID, + webhookSubscriptionClientState: CLIENT_STATE, + webhookSubscriptionStatus: WebhookSubscriptionStatus.ACTIVE, + webhookSubscriptionExpiresAt: new Date(Date.now() + 3600 * 1000), + }); + }); + + afterAll(async () => { + await account?.cleanup().catch(() => undefined); + }); + + it('creates a replacement subscription when Microsoft reports it was removed', async () => { + await postLifecycleNotification('subscriptionRemoved').expect(200); + + await waitForAllJobsToFinish(); + + expect(microsoft.subscriptions.created).toHaveLength(1); + expect(microsoft.subscriptions.renewed).not.toContain( + REMOVED_SUBSCRIPTION_ID, + ); + + const channel = await readChannel(); + + expect(channel.webhookSubscriptionExternalId).not.toBe( + REMOVED_SUBSCRIPTION_ID, + ); + expect(channel.webhookSubscriptionStatus).toBe( + WebhookSubscriptionStatus.ACTIVE, + ); + }, 60000); + + it('creates only one replacement when the same removal is delivered twice', async () => { + await Promise.all([ + postLifecycleNotification('subscriptionRemoved'), + postLifecycleNotification('subscriptionRemoved'), + ]); + + await waitForAllJobsToFinish(); + + expect(microsoft.subscriptions.created).toHaveLength(1); + }, 60000); + + it('renews in place when Microsoft asks for reauthorization', async () => { + await postLifecycleNotification('reauthorizationRequired').expect(200); + + await waitForAllJobsToFinish(); + + expect(microsoft.subscriptions.renewed).toContain(REMOVED_SUBSCRIPTION_ID); + expect(microsoft.subscriptions.created).toHaveLength(0); + }, 60000); + + it('leaves the subscription untouched on a missed notification', async () => { + await postLifecycleNotification('missed').expect(200); + + await waitForAllJobsToFinish(); + + expect(microsoft.subscriptions.created).toHaveLength(0); + expect(microsoft.subscriptions.renewed).toHaveLength(0); + }, 60000); + + it('ignores a lifecycle event it does not recognize', async () => { + await postLifecycleNotification('someFutureLifecycleEvent').expect(200); + + await waitForAllJobsToFinish(); + + expect(microsoft.subscriptions.created).toHaveLength(0); + expect(microsoft.subscriptions.renewed).toHaveLength(0); + }, 60000); + + it('ignores a notification whose client state does not match', async () => { + await postLifecycleNotification('subscriptionRemoved', { + clientState: 'forged-client-state', + }).expect(200); + + await waitForAllJobsToFinish(); + + expect(microsoft.subscriptions.created).toHaveLength(0); + + const channel = await readChannel(); + + expect(channel.webhookSubscriptionExternalId).toBe(REMOVED_SUBSCRIPTION_ID); + }, 60000); +});