From 9e31ffdf68592c27893fe5ea4bd8ee544f627f6c Mon Sep 17 00:00:00 2001 From: neo773 <62795688+neo773@users.noreply.github.com> Date: Tue, 23 Jun 2026 14:04:16 +0530 Subject: [PATCH] feat(messaging): webhook push sync for Gmail, Calendar and Microsoft (#21970) Review in cubic --- .../src/metadata/generated/schema.graphql | 1 + .../src/metadata/generated/schema.ts | 5 +- .../src/generated-admin/graphql.ts | 1 + .../src/generated-metadata/graphql.ts | 1 + ...add-channel-webhook-subscription-fields.ts | 35 ++++ .../instance-commands.constant.ts | 2 + .../engine/core-modules/core-engine.module.ts | 2 + .../constants/public-feature-flag.const.ts | 8 + .../twenty-config/config-variables.ts | 23 ++ .../entities/calendar-channel.entity.ts | 20 ++ .../entities/message-channel.entity.ts | 17 ++ .../workspace-entity-manager.spec.ts | 1 + ...onnected-account-sync-webhook.exception.ts | 39 ++++ ...nected-account-sync-webhooks.controller.ts | 106 ++++++++++ .../connected-account-sync-webhooks.module.ts | 11 + .../google-calendar-notification.handler.ts | 76 +++++++ .../google-messaging-notification.handler.ts | 154 ++++++++++++++ .../google/google-webhook-driver.module.ts | 33 +++ ...microsoft-calendar-notification.handler.ts | 101 +++++++++ ...icrosoft-messaging-notification.handler.ts | 100 +++++++++ .../microsoft-webhook-driver.module.ts | 28 +++ ...count-sync-webhook-api-exception.filter.ts | 29 +++ .../google-calendar-notification.type.ts | 5 + .../types/google-pubsub-push.type.ts | 13 ++ .../microsoft-graph-notification.type.ts | 12 ++ .../webhook-notification-handler.type.ts | 3 + ...sync-webhook-exception-status-code.util.ts | 21 ++ .../channel-sync/channel-sync.module.ts | 2 + .../services/channel-sync.service.ts | 36 +++- ...subscription-renewal-buffer-ms.constant.ts | 1 + ...scription-renewal-cron-pattern.constant.ts | 1 + ...bhook-subscription-renewal.cron.command.ts | 33 +++ .../webhook-subscription-renewal.cron.job.ts | 92 ++++++++ .../webhook-subscription-driver.exception.ts | 38 ++++ .../google-calendar-watch-ttl-ms.constant.ts | 1 + .../google-webhook-subscription.driver.ts | 178 ++++++++++++++++ .../microsoft-subscription-ttl-ms.constant.ts | 1 + .../microsoft-webhook-subscription.driver.ts | 132 ++++++++++++ ...k-subscription-channel-deleted.listener.ts | 56 +++++ .../calendar-webhook-subscription.service.ts | 196 ++++++++++++++++++ .../messaging-webhook-subscription.service.ts | 192 +++++++++++++++++ ...ook-subscription-driver-factory.service.ts | 46 ++++ .../services/webhook-sync-trigger.service.ts | 128 ++++++++++++ .../types/webhook-subscription-driver.type.ts | 29 +++ .../webhook-subscription-manager.module.ts | 17 ++ .../webhook-subscription.module.ts | 37 ++++ .../services/gmail-get-messages.service.ts | 9 +- .../twenty-shared/src/types/FeatureFlagKey.ts | 1 + .../types/WebhookSubscriptionChannelType.ts | 4 + .../src/types/WebhookSubscriptionStatus.ts | 6 + packages/twenty-shared/src/types/index.ts | 2 + 51 files changed, 2079 insertions(+), 6 deletions(-) create mode 100644 packages/twenty-server/src/database/commands/upgrade-version-command/2-16/2-16-instance-command-fast-1782152096938-add-channel-webhook-subscription-fields.ts create mode 100644 packages/twenty-server/src/modules/connected-account-sync-webhooks/connected-account-sync-webhook.exception.ts create mode 100644 packages/twenty-server/src/modules/connected-account-sync-webhooks/connected-account-sync-webhooks.controller.ts create mode 100644 packages/twenty-server/src/modules/connected-account-sync-webhooks/connected-account-sync-webhooks.module.ts create mode 100644 packages/twenty-server/src/modules/connected-account-sync-webhooks/drivers/google/google-calendar-notification.handler.ts create mode 100644 packages/twenty-server/src/modules/connected-account-sync-webhooks/drivers/google/google-messaging-notification.handler.ts create mode 100644 packages/twenty-server/src/modules/connected-account-sync-webhooks/drivers/google/google-webhook-driver.module.ts create mode 100644 packages/twenty-server/src/modules/connected-account-sync-webhooks/drivers/microsoft/microsoft-calendar-notification.handler.ts create mode 100644 packages/twenty-server/src/modules/connected-account-sync-webhooks/drivers/microsoft/microsoft-messaging-notification.handler.ts create mode 100644 packages/twenty-server/src/modules/connected-account-sync-webhooks/drivers/microsoft/microsoft-webhook-driver.module.ts create mode 100644 packages/twenty-server/src/modules/connected-account-sync-webhooks/filters/connected-account-sync-webhook-api-exception.filter.ts create mode 100644 packages/twenty-server/src/modules/connected-account-sync-webhooks/types/google-calendar-notification.type.ts create mode 100644 packages/twenty-server/src/modules/connected-account-sync-webhooks/types/google-pubsub-push.type.ts create mode 100644 packages/twenty-server/src/modules/connected-account-sync-webhooks/types/microsoft-graph-notification.type.ts create mode 100644 packages/twenty-server/src/modules/connected-account-sync-webhooks/types/webhook-notification-handler.type.ts create mode 100644 packages/twenty-server/src/modules/connected-account-sync-webhooks/utils/get-connected-account-sync-webhook-exception-status-code.util.ts create mode 100644 packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/constants/webhook-subscription-renewal-buffer-ms.constant.ts create mode 100644 packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/constants/webhook-subscription-renewal-cron-pattern.constant.ts create mode 100644 packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/crons/commands/webhook-subscription-renewal.cron.command.ts create mode 100644 packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/crons/jobs/webhook-subscription-renewal.cron.job.ts create mode 100644 packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/drivers/exceptions/webhook-subscription-driver.exception.ts create mode 100644 packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/drivers/google/google-calendar-watch-ttl-ms.constant.ts create mode 100644 packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/drivers/google/google-webhook-subscription.driver.ts create mode 100644 packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/drivers/microsoft/microsoft-subscription-ttl-ms.constant.ts create mode 100644 packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/drivers/microsoft/microsoft-webhook-subscription.driver.ts create mode 100644 packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/listeners/webhook-subscription-channel-deleted.listener.ts create mode 100644 packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/services/calendar-webhook-subscription.service.ts create mode 100644 packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/services/messaging-webhook-subscription.service.ts create mode 100644 packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/services/webhook-subscription-driver-factory.service.ts create mode 100644 packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/services/webhook-sync-trigger.service.ts create mode 100644 packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/types/webhook-subscription-driver.type.ts create mode 100644 packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/webhook-subscription-manager.module.ts create mode 100644 packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/webhook-subscription.module.ts create mode 100644 packages/twenty-shared/src/types/WebhookSubscriptionChannelType.ts create mode 100644 packages/twenty-shared/src/types/WebhookSubscriptionStatus.ts diff --git a/packages/twenty-client-sdk/src/metadata/generated/schema.graphql b/packages/twenty-client-sdk/src/metadata/generated/schema.graphql index 5ef751705e..cd1c872519 100644 --- a/packages/twenty-client-sdk/src/metadata/generated/schema.graphql +++ b/packages/twenty-client-sdk/src/metadata/generated/schema.graphql @@ -1779,6 +1779,7 @@ enum FeatureFlagKey { IS_LOGIC_FUNCTION_PREBUILT_MODE_ENABLED IS_SETTINGS_DISCOVERY_HERO_ENABLED IS_CALL_RECORDING_ENABLED + IS_MESSAGING_CALENDAR_WEBHOOK_ENABLED } type WorkspaceUrls { diff --git a/packages/twenty-client-sdk/src/metadata/generated/schema.ts b/packages/twenty-client-sdk/src/metadata/generated/schema.ts index b975490997..13856b8747 100644 --- a/packages/twenty-client-sdk/src/metadata/generated/schema.ts +++ b/packages/twenty-client-sdk/src/metadata/generated/schema.ts @@ -1408,7 +1408,7 @@ export interface FeatureFlag { __typename: 'FeatureFlag' } -export type FeatureFlagKey = 'IS_UNIQUE_INDEXES_ENABLED' | 'IS_JSON_FILTER_ENABLED' | 'IS_MARKETPLACE_SETTING_TAB_VISIBLE' | 'IS_PUBLIC_DOMAIN_ENABLED' | 'IS_EMAIL_GROUP_ENABLED' | 'IS_JUNCTION_RELATIONS_ENABLED' | 'IS_REST_METADATA_API_NEW_FORMAT_DIRECT' | 'IS_LOGIC_FUNCTION_PREBUILT_MODE_ENABLED' | 'IS_SETTINGS_DISCOVERY_HERO_ENABLED' | 'IS_CALL_RECORDING_ENABLED' +export type FeatureFlagKey = 'IS_UNIQUE_INDEXES_ENABLED' | 'IS_JSON_FILTER_ENABLED' | 'IS_MARKETPLACE_SETTING_TAB_VISIBLE' | 'IS_PUBLIC_DOMAIN_ENABLED' | 'IS_EMAIL_GROUP_ENABLED' | 'IS_JUNCTION_RELATIONS_ENABLED' | 'IS_REST_METADATA_API_NEW_FORMAT_DIRECT' | 'IS_LOGIC_FUNCTION_PREBUILT_MODE_ENABLED' | 'IS_SETTINGS_DISCOVERY_HERO_ENABLED' | 'IS_CALL_RECORDING_ENABLED' | 'IS_MESSAGING_CALENDAR_WEBHOOK_ENABLED' export interface WorkspaceUrls { customUrl?: Scalars['String'] @@ -9081,7 +9081,8 @@ export const enumFeatureFlagKey = { IS_REST_METADATA_API_NEW_FORMAT_DIRECT: 'IS_REST_METADATA_API_NEW_FORMAT_DIRECT' as const, IS_LOGIC_FUNCTION_PREBUILT_MODE_ENABLED: 'IS_LOGIC_FUNCTION_PREBUILT_MODE_ENABLED' as const, IS_SETTINGS_DISCOVERY_HERO_ENABLED: 'IS_SETTINGS_DISCOVERY_HERO_ENABLED' as const, - IS_CALL_RECORDING_ENABLED: 'IS_CALL_RECORDING_ENABLED' as const + IS_CALL_RECORDING_ENABLED: 'IS_CALL_RECORDING_ENABLED' as const, + IS_MESSAGING_CALENDAR_WEBHOOK_ENABLED: 'IS_MESSAGING_CALENDAR_WEBHOOK_ENABLED' as const } export const enumIdentityProviderType = { diff --git a/packages/twenty-front/src/generated-admin/graphql.ts b/packages/twenty-front/src/generated-admin/graphql.ts index 76656a890d..111e1fbaea 100644 --- a/packages/twenty-front/src/generated-admin/graphql.ts +++ b/packages/twenty-front/src/generated-admin/graphql.ts @@ -303,6 +303,7 @@ export enum FeatureFlagKey { IS_JUNCTION_RELATIONS_ENABLED = 'IS_JUNCTION_RELATIONS_ENABLED', IS_LOGIC_FUNCTION_PREBUILT_MODE_ENABLED = 'IS_LOGIC_FUNCTION_PREBUILT_MODE_ENABLED', IS_MARKETPLACE_SETTING_TAB_VISIBLE = 'IS_MARKETPLACE_SETTING_TAB_VISIBLE', + IS_MESSAGING_CALENDAR_WEBHOOK_ENABLED = 'IS_MESSAGING_CALENDAR_WEBHOOK_ENABLED', IS_PUBLIC_DOMAIN_ENABLED = 'IS_PUBLIC_DOMAIN_ENABLED', IS_REST_METADATA_API_NEW_FORMAT_DIRECT = 'IS_REST_METADATA_API_NEW_FORMAT_DIRECT', IS_SETTINGS_DISCOVERY_HERO_ENABLED = 'IS_SETTINGS_DISCOVERY_HERO_ENABLED', diff --git a/packages/twenty-front/src/generated-metadata/graphql.ts b/packages/twenty-front/src/generated-metadata/graphql.ts index 059a03ee72..bb42a02985 100644 --- a/packages/twenty-front/src/generated-metadata/graphql.ts +++ b/packages/twenty-front/src/generated-metadata/graphql.ts @@ -1709,6 +1709,7 @@ export enum FeatureFlagKey { IS_JUNCTION_RELATIONS_ENABLED = 'IS_JUNCTION_RELATIONS_ENABLED', IS_LOGIC_FUNCTION_PREBUILT_MODE_ENABLED = 'IS_LOGIC_FUNCTION_PREBUILT_MODE_ENABLED', IS_MARKETPLACE_SETTING_TAB_VISIBLE = 'IS_MARKETPLACE_SETTING_TAB_VISIBLE', + IS_MESSAGING_CALENDAR_WEBHOOK_ENABLED = 'IS_MESSAGING_CALENDAR_WEBHOOK_ENABLED', IS_PUBLIC_DOMAIN_ENABLED = 'IS_PUBLIC_DOMAIN_ENABLED', IS_REST_METADATA_API_NEW_FORMAT_DIRECT = 'IS_REST_METADATA_API_NEW_FORMAT_DIRECT', IS_SETTINGS_DISCOVERY_HERO_ENABLED = 'IS_SETTINGS_DISCOVERY_HERO_ENABLED', diff --git a/packages/twenty-server/src/database/commands/upgrade-version-command/2-16/2-16-instance-command-fast-1782152096938-add-channel-webhook-subscription-fields.ts b/packages/twenty-server/src/database/commands/upgrade-version-command/2-16/2-16-instance-command-fast-1782152096938-add-channel-webhook-subscription-fields.ts new file mode 100644 index 0000000000..e034388d1f --- /dev/null +++ b/packages/twenty-server/src/database/commands/upgrade-version-command/2-16/2-16-instance-command-fast-1782152096938-add-channel-webhook-subscription-fields.ts @@ -0,0 +1,35 @@ +import { QueryRunner } from 'typeorm'; + +import { RegisteredInstanceCommand } from 'src/engine/core-modules/upgrade/decorators/registered-instance-command.decorator'; +import { FastInstanceCommand } from 'src/engine/core-modules/upgrade/interfaces/fast-instance-command.interface'; + +@RegisteredInstanceCommand('2.16.0', 1782152096938) +export class AddChannelWebhookSubscriptionFieldsFastInstanceCommand implements FastInstanceCommand { + public async up(queryRunner: QueryRunner): Promise { + await queryRunner.query('ALTER TABLE "core"."calendarChannel" ADD "webhookSubscriptionExternalId" character varying'); + await queryRunner.query('ALTER TABLE "core"."calendarChannel" ADD "webhookSubscriptionExternalResourceId" character varying'); + await queryRunner.query('ALTER TABLE "core"."calendarChannel" ADD "webhookSubscriptionClientState" character varying'); + await queryRunner.query('CREATE TYPE "core"."calendarChannel_webhooksubscriptionstatus_enum" AS ENUM(\'PENDING\', \'ACTIVE\', \'FAILED\', \'EXPIRED\')'); + await queryRunner.query('ALTER TABLE "core"."calendarChannel" ADD "webhookSubscriptionStatus" "core"."calendarChannel_webhooksubscriptionstatus_enum"'); + await queryRunner.query('ALTER TABLE "core"."calendarChannel" ADD "webhookSubscriptionExpiresAt" TIMESTAMP WITH TIME ZONE'); + await queryRunner.query('ALTER TABLE "core"."messageChannel" ADD "webhookSubscriptionExternalId" character varying'); + await queryRunner.query('ALTER TABLE "core"."messageChannel" ADD "webhookSubscriptionClientState" character varying'); + await queryRunner.query('CREATE TYPE "core"."messageChannel_webhooksubscriptionstatus_enum" AS ENUM(\'PENDING\', \'ACTIVE\', \'FAILED\', \'EXPIRED\')'); + await queryRunner.query('ALTER TABLE "core"."messageChannel" ADD "webhookSubscriptionStatus" "core"."messageChannel_webhooksubscriptionstatus_enum"'); + await queryRunner.query('ALTER TABLE "core"."messageChannel" ADD "webhookSubscriptionExpiresAt" TIMESTAMP WITH TIME ZONE'); + } + + public async down(queryRunner: QueryRunner): Promise { + await queryRunner.query('ALTER TABLE "core"."messageChannel" DROP COLUMN "webhookSubscriptionExpiresAt"'); + await queryRunner.query('ALTER TABLE "core"."messageChannel" DROP COLUMN "webhookSubscriptionStatus"'); + await queryRunner.query('DROP TYPE "core"."messageChannel_webhooksubscriptionstatus_enum"'); + await queryRunner.query('ALTER TABLE "core"."messageChannel" DROP COLUMN "webhookSubscriptionClientState"'); + await queryRunner.query('ALTER TABLE "core"."messageChannel" DROP COLUMN "webhookSubscriptionExternalId"'); + await queryRunner.query('ALTER TABLE "core"."calendarChannel" DROP COLUMN "webhookSubscriptionExpiresAt"'); + await queryRunner.query('ALTER TABLE "core"."calendarChannel" DROP COLUMN "webhookSubscriptionStatus"'); + await queryRunner.query('DROP TYPE "core"."calendarChannel_webhooksubscriptionstatus_enum"'); + await queryRunner.query('ALTER TABLE "core"."calendarChannel" DROP COLUMN "webhookSubscriptionClientState"'); + await queryRunner.query('ALTER TABLE "core"."calendarChannel" DROP COLUMN "webhookSubscriptionExternalResourceId"'); + await queryRunner.query('ALTER TABLE "core"."calendarChannel" DROP COLUMN "webhookSubscriptionExternalId"'); + } +} diff --git a/packages/twenty-server/src/database/commands/upgrade-version-command/instance-commands.constant.ts b/packages/twenty-server/src/database/commands/upgrade-version-command/instance-commands.constant.ts index 5f9328c5ac..b947fce001 100644 --- a/packages/twenty-server/src/database/commands/upgrade-version-command/instance-commands.constant.ts +++ b/packages/twenty-server/src/database/commands/upgrade-version-command/instance-commands.constant.ts @@ -75,6 +75,7 @@ import { MigrateAiModelPreferencesSlowInstanceCommand } from 'src/database/comma import { AddHasPaymentMethodToBillingCustomerFastInstanceCommand } from 'src/database/commands/upgrade-version-command/2-15/2-15-instance-command-fast-1781280240009-add-has-payment-method-to-billing-customer'; import { AddFolderImportToMessageFolderPendingSyncActionFastInstanceCommand } from './2-15/2-15-instance-command-fast-1781714499016-add-folder-import-to-message-folder-pending-sync-action'; import { AddViewKanbanColumnWidthFastInstanceCommand } from './2-15/2-15-instance-command-fast-1781900000000-add-view-kanban-column-width'; +import { AddChannelWebhookSubscriptionFieldsFastInstanceCommand } from 'src/database/commands/upgrade-version-command/2-16/2-16-instance-command-fast-1782152096938-add-channel-webhook-subscription-fields'; export const INSTANCE_COMMANDS = [ AddViewFieldGroupIdIndexOnViewFieldFastInstanceCommand, @@ -152,4 +153,5 @@ export const INSTANCE_COMMANDS = [ BackfillConnectionSecuritySlowInstanceCommand, AddFolderImportToMessageFolderPendingSyncActionFastInstanceCommand, AddViewKanbanColumnWidthFastInstanceCommand, + AddChannelWebhookSubscriptionFieldsFastInstanceCommand, ]; diff --git a/packages/twenty-server/src/engine/core-modules/core-engine.module.ts b/packages/twenty-server/src/engine/core-modules/core-engine.module.ts index cf8208006a..786aaec3c3 100644 --- a/packages/twenty-server/src/engine/core-modules/core-engine.module.ts +++ b/packages/twenty-server/src/engine/core-modules/core-engine.module.ts @@ -47,6 +47,7 @@ import { MessageQueueModule } from 'src/engine/core-modules/message-queue/messag import { messageQueueModuleFactory } from 'src/engine/core-modules/message-queue/message-queue.module-factory'; import { TimelineMessagingModule } from 'src/engine/core-modules/messaging/timeline-messaging.module'; import { MessagingWebhooksModule } from 'src/modules/messaging-webhooks/messaging-webhooks.module'; +import { ConnectedAccountSyncWebhooksModule } from 'src/modules/connected-account-sync-webhooks/connected-account-sync-webhooks.module'; import { MetricsModule } from 'src/engine/core-modules/metrics/metrics.module'; import { MetricsService } from 'src/engine/core-modules/metrics/metrics.service'; import { OpenApiModule } from 'src/engine/core-modules/open-api/open-api.module'; @@ -88,6 +89,7 @@ import { FileModule } from './file/file.module'; BillingModule, BillingWebhookModule, MessagingWebhooksModule, + ConnectedAccountSyncWebhooksModule, UsageModule, ClientConfigModule, FeatureFlagModule, diff --git a/packages/twenty-server/src/engine/core-modules/feature-flag/constants/public-feature-flag.const.ts b/packages/twenty-server/src/engine/core-modules/feature-flag/constants/public-feature-flag.const.ts index 72a1f4380f..a6497aac15 100644 --- a/packages/twenty-server/src/engine/core-modules/feature-flag/constants/public-feature-flag.const.ts +++ b/packages/twenty-server/src/engine/core-modules/feature-flag/constants/public-feature-flag.const.ts @@ -28,6 +28,14 @@ export const PUBLIC_FEATURE_FLAGS: PublicFeatureFlag[] = [ 'Show the per-page hero illustration + video walkthrough modal on settings pages', }, }, + { + key: FeatureFlagKey.IS_MESSAGING_CALENDAR_WEBHOOK_ENABLED, + metadata: { + label: 'Messaging & Calendar Webhooks', + description: + 'Sync Gmail, Google Calendar, and Microsoft 365 mail/calendar via provider push notifications instead of cron polling', + }, + }, ...(process.env.CLOUDFLARE_API_KEY ? [ // { diff --git a/packages/twenty-server/src/engine/core-modules/twenty-config/config-variables.ts b/packages/twenty-server/src/engine/core-modules/twenty-config/config-variables.ts index dfe1b200bd..8f8c92a8cb 100644 --- a/packages/twenty-server/src/engine/core-modules/twenty-config/config-variables.ts +++ b/packages/twenty-server/src/engine/core-modules/twenty-config/config-variables.ts @@ -177,6 +177,29 @@ export class ConfigVariables { }) MESSAGING_PROVIDER_GMAIL_ENABLED = false; + @ConfigVariablesMetadata({ + group: ConfigVariablesGroup.GOOGLE_AUTH, + isSensitive: false, + description: + 'Google Cloud Pub/Sub topic that Gmail push notifications publish to ' + + '(format: projects//topics/). Required for webhook-based Gmail sync.', + type: ConfigVariableType.STRING, + }) + @IsOptional() + MESSAGING_GMAIL_PUBSUB_TOPIC: string; + + @ConfigVariablesMetadata({ + group: ConfigVariablesGroup.GOOGLE_AUTH, + isSensitive: false, + description: + 'Service account email authorized to deliver Gmail Pub/Sub push ' + + 'notifications. The signed OIDC token on each push is verified against ' + + 'this email. Required for webhook-based Gmail sync.', + type: ConfigVariableType.STRING, + }) + @IsOptional() + MESSAGING_GMAIL_PUBSUB_VERIFICATION_EMAIL: string; + @ConfigVariablesMetadata({ group: ConfigVariablesGroup.ADVANCED_SETTINGS, description: 'Enable or disable the IMAP messaging integration', diff --git a/packages/twenty-server/src/engine/metadata-modules/calendar-channel/entities/calendar-channel.entity.ts b/packages/twenty-server/src/engine/metadata-modules/calendar-channel/entities/calendar-channel.entity.ts index 6f1dd65189..08be7d5cd5 100644 --- a/packages/twenty-server/src/engine/metadata-modules/calendar-channel/entities/calendar-channel.entity.ts +++ b/packages/twenty-server/src/engine/metadata-modules/calendar-channel/entities/calendar-channel.entity.ts @@ -17,6 +17,7 @@ import { CalendarChannelSyncStage, CalendarChannelSyncStatus, CalendarChannelVisibility, + WebhookSubscriptionStatus, } from 'twenty-shared/types'; import { ConnectedAccountEntity } from 'src/engine/metadata-modules/connected-account/entities/connected-account.entity'; @@ -97,6 +98,25 @@ export class CalendarChannelEntity extends WorkspaceRelatedEntity { @Column({ type: 'integer', nullable: false, default: 0 }) throttleFailureCount: number; + @Column({ type: 'varchar', nullable: true }) + webhookSubscriptionExternalId: string | null; + + @Column({ type: 'varchar', nullable: true }) + webhookSubscriptionExternalResourceId: string | null; + + @Column({ type: 'varchar', nullable: true }) + webhookSubscriptionClientState: string | null; + + @Column({ + type: 'enum', + enum: WebhookSubscriptionStatus, + nullable: true, + }) + webhookSubscriptionStatus: WebhookSubscriptionStatus | null; + + @Column({ type: 'timestamptz', nullable: true }) + webhookSubscriptionExpiresAt: Date | null; + @Column({ type: 'uuid', nullable: false }) connectedAccountId: string; diff --git a/packages/twenty-server/src/engine/metadata-modules/message-channel/entities/message-channel.entity.ts b/packages/twenty-server/src/engine/metadata-modules/message-channel/entities/message-channel.entity.ts index 36f55714fb..bfa47835bb 100644 --- a/packages/twenty-server/src/engine/metadata-modules/message-channel/entities/message-channel.entity.ts +++ b/packages/twenty-server/src/engine/metadata-modules/message-channel/entities/message-channel.entity.ts @@ -21,6 +21,7 @@ import { MessageChannelType, MessageChannelVisibility, MessageFolderImportPolicy, + WebhookSubscriptionStatus, } from 'twenty-shared/types'; import { ConnectedAccountEntity } from 'src/engine/metadata-modules/connected-account/entities/connected-account.entity'; @@ -139,6 +140,22 @@ export class MessageChannelEntity extends WorkspaceRelatedEntity { @Column({ type: 'timestamptz', nullable: true }) throttleRetryAfter: Date | null; + @Column({ type: 'varchar', nullable: true }) + webhookSubscriptionExternalId: string | null; + + @Column({ type: 'varchar', nullable: true }) + webhookSubscriptionClientState: string | null; + + @Column({ + type: 'enum', + enum: WebhookSubscriptionStatus, + nullable: true, + }) + webhookSubscriptionStatus: WebhookSubscriptionStatus | null; + + @Column({ type: 'timestamptz', nullable: true }) + webhookSubscriptionExpiresAt: Date | null; + @Column({ type: 'uuid', nullable: false }) connectedAccountId: string; diff --git a/packages/twenty-server/src/engine/twenty-orm/entity-manager/workspace-entity-manager.spec.ts b/packages/twenty-server/src/engine/twenty-orm/entity-manager/workspace-entity-manager.spec.ts index 8bdd0d75c1..2b3a39af63 100644 --- a/packages/twenty-server/src/engine/twenty-orm/entity-manager/workspace-entity-manager.spec.ts +++ b/packages/twenty-server/src/engine/twenty-orm/entity-manager/workspace-entity-manager.spec.ts @@ -240,6 +240,7 @@ describe('WorkspaceEntityManager', () => { IS_LOGIC_FUNCTION_PREBUILT_MODE_ENABLED: false, IS_SETTINGS_DISCOVERY_HERO_ENABLED: false, IS_CALL_RECORDING_ENABLED: false, + IS_MESSAGING_CALENDAR_WEBHOOK_ENABLED: false, }, userWorkspaceRoleMap: {}, eventEmitterService: { diff --git a/packages/twenty-server/src/modules/connected-account-sync-webhooks/connected-account-sync-webhook.exception.ts b/packages/twenty-server/src/modules/connected-account-sync-webhooks/connected-account-sync-webhook.exception.ts new file mode 100644 index 0000000000..f5db93da3b --- /dev/null +++ b/packages/twenty-server/src/modules/connected-account-sync-webhooks/connected-account-sync-webhook.exception.ts @@ -0,0 +1,39 @@ +import { type MessageDescriptor } from '@lingui/core'; +import { msg } from '@lingui/core/macro'; +import { assertUnreachable } from 'twenty-shared/utils'; + +import { CustomException } from 'src/utils/custom-exception'; + +export enum ConnectedAccountSyncWebhookExceptionCode { + MISSING_REQUEST_BODY = 'MISSING_REQUEST_BODY', + INVALID_PAYLOAD = 'INVALID_PAYLOAD', + INVALID_SIGNATURE = 'INVALID_SIGNATURE', +} + +const getConnectedAccountSyncWebhookExceptionUserFriendlyMessage = ( + code: ConnectedAccountSyncWebhookExceptionCode, +) => { + switch (code) { + case ConnectedAccountSyncWebhookExceptionCode.MISSING_REQUEST_BODY: + case ConnectedAccountSyncWebhookExceptionCode.INVALID_PAYLOAD: + return msg`The webhook request could not be processed.`; + case ConnectedAccountSyncWebhookExceptionCode.INVALID_SIGNATURE: + return msg`The webhook request could not be authenticated.`; + default: + assertUnreachable(code); + } +}; + +export class ConnectedAccountSyncWebhookException extends CustomException { + constructor( + message: string, + code: ConnectedAccountSyncWebhookExceptionCode, + { userFriendlyMessage }: { userFriendlyMessage?: MessageDescriptor } = {}, + ) { + super(message, code, { + userFriendlyMessage: + userFriendlyMessage ?? + getConnectedAccountSyncWebhookExceptionUserFriendlyMessage(code), + }); + } +} diff --git a/packages/twenty-server/src/modules/connected-account-sync-webhooks/connected-account-sync-webhooks.controller.ts b/packages/twenty-server/src/modules/connected-account-sync-webhooks/connected-account-sync-webhooks.controller.ts new file mode 100644 index 0000000000..d175a244b7 --- /dev/null +++ b/packages/twenty-server/src/modules/connected-account-sync-webhooks/connected-account-sync-webhooks.controller.ts @@ -0,0 +1,106 @@ +import { + Body, + Controller, + Headers, + HttpCode, + HttpStatus, + Post, + Query, + Res, + UseFilters, + UseGuards, +} from '@nestjs/common'; + +import { type Response } from 'express'; +import { isDefined } from 'twenty-shared/utils'; + +import { escapeHtml } from 'src/engine/core-modules/emailing-domain/utils/escape-html.util'; + +import { NoPermissionGuard } from 'src/engine/guards/no-permission.guard'; +import { PublicEndpointGuard } from 'src/engine/guards/public-endpoint.guard'; +import { GoogleCalendarNotificationHandler } from 'src/modules/connected-account-sync-webhooks/drivers/google/google-calendar-notification.handler'; +import { GoogleMessagingNotificationHandler } from 'src/modules/connected-account-sync-webhooks/drivers/google/google-messaging-notification.handler'; +import { MicrosoftCalendarNotificationHandler } from 'src/modules/connected-account-sync-webhooks/drivers/microsoft/microsoft-calendar-notification.handler'; +import { MicrosoftMessagingNotificationHandler } from 'src/modules/connected-account-sync-webhooks/drivers/microsoft/microsoft-messaging-notification.handler'; +import { ConnectedAccountSyncWebhookApiExceptionFilter } from 'src/modules/connected-account-sync-webhooks/filters/connected-account-sync-webhook-api-exception.filter'; +import { type GooglePubSubPushMessage } from 'src/modules/connected-account-sync-webhooks/types/google-pubsub-push.type'; +import { type MicrosoftGraphNotificationPayload } from 'src/modules/connected-account-sync-webhooks/types/microsoft-graph-notification.type'; + +@Controller() +@UseFilters(ConnectedAccountSyncWebhookApiExceptionFilter) +@UseGuards(PublicEndpointGuard, NoPermissionGuard) +export class ConnectedAccountSyncWebhooksController { + constructor( + private readonly googleMessagingNotificationHandler: GoogleMessagingNotificationHandler, + private readonly googleCalendarNotificationHandler: GoogleCalendarNotificationHandler, + private readonly microsoftMessagingNotificationHandler: MicrosoftMessagingNotificationHandler, + private readonly microsoftCalendarNotificationHandler: MicrosoftCalendarNotificationHandler, + ) {} + + @Post('webhooks/google/messaging') + @HttpCode(HttpStatus.OK) + async handleGoogleMessaging( + @Body() body: GooglePubSubPushMessage, + @Headers('authorization') authorizationHeader: string | undefined, + ): Promise { + await this.googleMessagingNotificationHandler.handle({ + body, + authorizationHeader, + }); + } + + @Post('webhooks/google/calendar') + @HttpCode(HttpStatus.OK) + async handleGoogleCalendar( + @Headers('x-goog-channel-id') channelId: string | undefined, + @Headers('x-goog-resource-state') resourceState: string | undefined, + @Headers('x-goog-channel-token') channelToken: string | undefined, + ): Promise { + await this.googleCalendarNotificationHandler.handle({ + channelId, + resourceState, + channelToken, + }); + } + + @Post('webhooks/microsoft/messaging') + @HttpCode(HttpStatus.OK) + async handleMicrosoftMessaging( + @Body() body: MicrosoftGraphNotificationPayload, + @Query('validationToken') validationToken: string | undefined, + @Res({ passthrough: true }) response: Response, + ): Promise { + if (isDefined(validationToken)) { + return this.respondToValidationHandshake(validationToken, response); + } + + await this.microsoftMessagingNotificationHandler.handle(body.value ?? []); + + return ''; + } + + @Post('webhooks/microsoft/calendar') + @HttpCode(HttpStatus.OK) + async handleMicrosoftCalendar( + @Body() body: MicrosoftGraphNotificationPayload, + @Query('validationToken') validationToken: string | undefined, + @Res({ passthrough: true }) response: Response, + ): Promise { + if (isDefined(validationToken)) { + return this.respondToValidationHandshake(validationToken, response); + } + + await this.microsoftCalendarNotificationHandler.handle(body.value ?? []); + + return ''; + } + + private respondToValidationHandshake( + validationToken: string, + response: Response, + ): string { + response.type('text/plain'); + + return escapeHtml(validationToken); + } +} diff --git a/packages/twenty-server/src/modules/connected-account-sync-webhooks/connected-account-sync-webhooks.module.ts b/packages/twenty-server/src/modules/connected-account-sync-webhooks/connected-account-sync-webhooks.module.ts new file mode 100644 index 0000000000..9e5c1f54dc --- /dev/null +++ b/packages/twenty-server/src/modules/connected-account-sync-webhooks/connected-account-sync-webhooks.module.ts @@ -0,0 +1,11 @@ +import { Module } from '@nestjs/common'; + +import { ConnectedAccountSyncWebhooksController } from 'src/modules/connected-account-sync-webhooks/connected-account-sync-webhooks.controller'; +import { GoogleWebhookDriverModule } from 'src/modules/connected-account-sync-webhooks/drivers/google/google-webhook-driver.module'; +import { MicrosoftWebhookDriverModule } from 'src/modules/connected-account-sync-webhooks/drivers/microsoft/microsoft-webhook-driver.module'; + +@Module({ + imports: [GoogleWebhookDriverModule, MicrosoftWebhookDriverModule], + controllers: [ConnectedAccountSyncWebhooksController], +}) +export class ConnectedAccountSyncWebhooksModule {} diff --git a/packages/twenty-server/src/modules/connected-account-sync-webhooks/drivers/google/google-calendar-notification.handler.ts b/packages/twenty-server/src/modules/connected-account-sync-webhooks/drivers/google/google-calendar-notification.handler.ts new file mode 100644 index 0000000000..bfa67b2a8e --- /dev/null +++ b/packages/twenty-server/src/modules/connected-account-sync-webhooks/drivers/google/google-calendar-notification.handler.ts @@ -0,0 +1,76 @@ +import { timingSafeEqual } from 'crypto'; + +import { Injectable } from '@nestjs/common'; +import { InjectRepository } from '@nestjs/typeorm'; + +import { isNonEmptyString } from '@sniptt/guards'; +import { WebhookSubscriptionStatus } from 'twenty-shared/types'; +import { isDefined } from 'twenty-shared/utils'; +import { Repository } from 'typeorm'; + +import { CalendarChannelEntity } from 'src/engine/metadata-modules/calendar-channel/entities/calendar-channel.entity'; +import { WebhookSyncTriggerService } from 'src/modules/connected-account/webhook-subscription-manager/services/webhook-sync-trigger.service'; +import { + ConnectedAccountSyncWebhookException, + ConnectedAccountSyncWebhookExceptionCode, +} from 'src/modules/connected-account-sync-webhooks/connected-account-sync-webhook.exception'; +import { type GoogleCalendarChannelNotification } from 'src/modules/connected-account-sync-webhooks/types/google-calendar-notification.type'; +import { type WebhookNotificationHandler } from 'src/modules/connected-account-sync-webhooks/types/webhook-notification-handler.type'; + +const GOOGLE_CALENDAR_SYNC_RESOURCE_STATE = 'sync'; + +@Injectable() +export class GoogleCalendarNotificationHandler implements WebhookNotificationHandler { + constructor( + @InjectRepository(CalendarChannelEntity) + private readonly calendarChannelRepository: Repository, + private readonly webhookSyncTriggerService: WebhookSyncTriggerService, + ) {} + + async handle(request: GoogleCalendarChannelNotification): Promise { + if (request.resourceState === GOOGLE_CALENDAR_SYNC_RESOURCE_STATE) { + return; + } + + if (!isNonEmptyString(request.channelId)) { + return; + } + + const calendarChannel = await this.calendarChannelRepository.findOne({ + where: { + webhookSubscriptionExternalId: request.channelId, + webhookSubscriptionStatus: WebhookSubscriptionStatus.ACTIVE, + }, + }); + + if (!isDefined(calendarChannel)) { + return; + } + + const channelToken = request.channelToken; + const clientState = calendarChannel.webhookSubscriptionClientState; + const channelTokenBuffer = isNonEmptyString(channelToken) + ? Buffer.from(channelToken) + : null; + const clientStateBuffer = isNonEmptyString(clientState) + ? Buffer.from(clientState) + : null; + + if ( + !isDefined(channelTokenBuffer) || + !isDefined(clientStateBuffer) || + channelTokenBuffer.length !== clientStateBuffer.length || + !timingSafeEqual(channelTokenBuffer, clientStateBuffer) + ) { + throw new ConnectedAccountSyncWebhookException( + 'Google calendar channel token mismatch', + ConnectedAccountSyncWebhookExceptionCode.INVALID_SIGNATURE, + ); + } + + await this.webhookSyncTriggerService.triggerCalendarSync( + calendarChannel.id, + calendarChannel.workspaceId, + ); + } +} diff --git a/packages/twenty-server/src/modules/connected-account-sync-webhooks/drivers/google/google-messaging-notification.handler.ts b/packages/twenty-server/src/modules/connected-account-sync-webhooks/drivers/google/google-messaging-notification.handler.ts new file mode 100644 index 0000000000..fd144b20ed --- /dev/null +++ b/packages/twenty-server/src/modules/connected-account-sync-webhooks/drivers/google/google-messaging-notification.handler.ts @@ -0,0 +1,154 @@ +import { Injectable } from '@nestjs/common'; +import { InjectRepository } from '@nestjs/typeorm'; + +import { isNonEmptyString } from '@sniptt/guards'; +import { OAuth2Client } from 'google-auth-library'; +import { + ConnectedAccountProvider, + WebhookSubscriptionStatus, +} from 'twenty-shared/types'; +import { isDefined } from 'twenty-shared/utils'; +import { In, Repository } from 'typeorm'; + +import { TwentyConfigService } from 'src/engine/core-modules/twenty-config/twenty-config.service'; +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 { WebhookSyncTriggerService } from 'src/modules/connected-account/webhook-subscription-manager/services/webhook-sync-trigger.service'; +import { + ConnectedAccountSyncWebhookException, + ConnectedAccountSyncWebhookExceptionCode, +} from 'src/modules/connected-account-sync-webhooks/connected-account-sync-webhook.exception'; +import { + type GmailPushDecodedData, + type GooglePubSubPushMessage, +} from 'src/modules/connected-account-sync-webhooks/types/google-pubsub-push.type'; +import { type WebhookNotificationHandler } from 'src/modules/connected-account-sync-webhooks/types/webhook-notification-handler.type'; + +export type GoogleMessagingNotificationRequest = { + body: GooglePubSubPushMessage; + authorizationHeader: string | undefined; +}; + +@Injectable() +export class GoogleMessagingNotificationHandler implements WebhookNotificationHandler { + constructor( + private readonly twentyConfigService: TwentyConfigService, + @InjectRepository(ConnectedAccountEntity) + private readonly connectedAccountRepository: Repository, + @InjectRepository(MessageChannelEntity) + private readonly messageChannelRepository: Repository, + private readonly webhookSyncTriggerService: WebhookSyncTriggerService, + ) {} + + async handle(request: GoogleMessagingNotificationRequest): Promise { + await this.verify(request.authorizationHeader); + + const decodedData = this.decodeMessageData(request.body); + + if (!isDefined(decodedData)) { + return; + } + + const connectedAccounts = await this.connectedAccountRepository.find({ + where: { + handle: decodedData.emailAddress, + provider: ConnectedAccountProvider.GOOGLE, + }, + }); + + const connectedAccountIds = connectedAccounts.map( + (connectedAccount) => connectedAccount.id, + ); + + if (connectedAccountIds.length === 0) { + return; + } + + const messageChannels = await this.messageChannelRepository.find({ + where: { + connectedAccountId: In(connectedAccountIds), + webhookSubscriptionStatus: WebhookSubscriptionStatus.ACTIVE, + }, + }); + + for (const messageChannel of messageChannels) { + await this.webhookSyncTriggerService.triggerMessagingSync( + messageChannel.id, + messageChannel.workspaceId, + ); + } + } + + private async verify(authorizationHeader: string | undefined): Promise { + const expectedEmail = this.twentyConfigService.get( + 'MESSAGING_GMAIL_PUBSUB_VERIFICATION_EMAIL', + ); + + if (!isNonEmptyString(expectedEmail)) { + throw new ConnectedAccountSyncWebhookException( + 'MESSAGING_GMAIL_PUBSUB_VERIFICATION_EMAIL is not configured', + ConnectedAccountSyncWebhookExceptionCode.INVALID_SIGNATURE, + ); + } + + const idToken = authorizationHeader?.replace(/^Bearer\s+/i, ''); + + if (!isNonEmptyString(idToken)) { + throw new ConnectedAccountSyncWebhookException( + 'Missing Pub/Sub OIDC token', + ConnectedAccountSyncWebhookExceptionCode.INVALID_SIGNATURE, + ); + } + + const expectedAudience = `${this.twentyConfigService.get('SERVER_URL')}/webhooks/google/messaging`; + + try { + const ticket = await new OAuth2Client().verifyIdToken({ + idToken, + audience: expectedAudience, + }); + + const payload = ticket.getPayload(); + + if ( + !isDefined(payload) || + payload.email !== expectedEmail || + payload.email_verified !== true + ) { + throw new ConnectedAccountSyncWebhookException( + 'Pub/Sub OIDC token failed verification', + ConnectedAccountSyncWebhookExceptionCode.INVALID_SIGNATURE, + ); + } + } catch (error) { + if (error instanceof ConnectedAccountSyncWebhookException) { + throw error; + } + + throw new ConnectedAccountSyncWebhookException( + 'Pub/Sub OIDC token verification failed', + ConnectedAccountSyncWebhookExceptionCode.INVALID_SIGNATURE, + ); + } + } + + private decodeMessageData( + body: GooglePubSubPushMessage, + ): GmailPushDecodedData | undefined { + const encodedData = body.message?.data; + + if (!isNonEmptyString(encodedData)) { + return; + } + + const decoded = JSON.parse( + Buffer.from(encodedData, 'base64').toString('utf-8'), + ) as GmailPushDecodedData; + + if (!isNonEmptyString(decoded.emailAddress)) { + return; + } + + return decoded; + } +} diff --git a/packages/twenty-server/src/modules/connected-account-sync-webhooks/drivers/google/google-webhook-driver.module.ts b/packages/twenty-server/src/modules/connected-account-sync-webhooks/drivers/google/google-webhook-driver.module.ts new file mode 100644 index 0000000000..cfc2423362 --- /dev/null +++ b/packages/twenty-server/src/modules/connected-account-sync-webhooks/drivers/google/google-webhook-driver.module.ts @@ -0,0 +1,33 @@ +import { Module } from '@nestjs/common'; +import { TypeOrmModule } from '@nestjs/typeorm'; + +import { TwentyConfigModule } from 'src/engine/core-modules/twenty-config/twenty-config.module'; +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 { MessageChannelEntity } from 'src/engine/metadata-modules/message-channel/entities/message-channel.entity'; +import { WebhookSyncTriggerService } from 'src/modules/connected-account/webhook-subscription-manager/services/webhook-sync-trigger.service'; +import { WebhookSubscriptionModule } from 'src/modules/connected-account/webhook-subscription-manager/webhook-subscription.module'; +import { GoogleCalendarNotificationHandler } from 'src/modules/connected-account-sync-webhooks/drivers/google/google-calendar-notification.handler'; +import { GoogleMessagingNotificationHandler } from 'src/modules/connected-account-sync-webhooks/drivers/google/google-messaging-notification.handler'; + +@Module({ + imports: [ + TwentyConfigModule, + WebhookSubscriptionModule, + TypeOrmModule.forFeature([ + ConnectedAccountEntity, + MessageChannelEntity, + CalendarChannelEntity, + ]), + ], + providers: [ + GoogleMessagingNotificationHandler, + GoogleCalendarNotificationHandler, + WebhookSyncTriggerService, + ], + exports: [ + GoogleMessagingNotificationHandler, + GoogleCalendarNotificationHandler, + ], +}) +export class GoogleWebhookDriverModule {} 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 new file mode 100644 index 0000000000..b8fbaf84a6 --- /dev/null +++ b/packages/twenty-server/src/modules/connected-account-sync-webhooks/drivers/microsoft/microsoft-calendar-notification.handler.ts @@ -0,0 +1,101 @@ +import { timingSafeEqual } from 'crypto'; + +import { Injectable, Logger } from '@nestjs/common'; +import { InjectRepository } from '@nestjs/typeorm'; + +import { isNonEmptyString } from '@sniptt/guards'; +import { isDefined } from 'twenty-shared/utils'; +import { In, Repository } from 'typeorm'; + +import { CalendarChannelEntity } from 'src/engine/metadata-modules/calendar-channel/entities/calendar-channel.entity'; +import { CalendarWebhookSubscriptionService } from 'src/modules/connected-account/webhook-subscription-manager/services/calendar-webhook-subscription.service'; +import { WebhookSyncTriggerService } from 'src/modules/connected-account/webhook-subscription-manager/services/webhook-sync-trigger.service'; +import { type MicrosoftGraphNotification } from 'src/modules/connected-account-sync-webhooks/types/microsoft-graph-notification.type'; +import { type WebhookNotificationHandler } from 'src/modules/connected-account-sync-webhooks/types/webhook-notification-handler.type'; + +@Injectable() +export class MicrosoftCalendarNotificationHandler implements WebhookNotificationHandler< + MicrosoftGraphNotification[] +> { + private readonly logger = new Logger( + MicrosoftCalendarNotificationHandler.name, + ); + + constructor( + @InjectRepository(CalendarChannelEntity) + private readonly calendarChannelRepository: Repository, + private readonly calendarWebhookSubscriptionService: CalendarWebhookSubscriptionService, + private readonly webhookSyncTriggerService: WebhookSyncTriggerService, + ) {} + + async handle(notifications: MicrosoftGraphNotification[]): Promise { + const subscriptionIds = notifications + .map((notification) => notification.subscriptionId) + .filter(isNonEmptyString); + + if (subscriptionIds.length === 0) { + return; + } + + const calendarChannels = await this.calendarChannelRepository.find({ + where: { webhookSubscriptionExternalId: In(subscriptionIds) }, + }); + + const calendarChannelByExternalId = new Map( + calendarChannels.map((calendarChannel) => [ + calendarChannel.webhookSubscriptionExternalId, + calendarChannel, + ]), + ); + + for (const notification of notifications) { + const calendarChannel = calendarChannelByExternalId.get( + notification.subscriptionId, + ); + + if (!isDefined(calendarChannel)) { + this.logger.warn( + `No calendar subscription found for ${notification.subscriptionId}`, + ); + continue; + } + + const clientState = notification.clientState; + const expectedClientState = + calendarChannel.webhookSubscriptionClientState; + const clientStateBuffer = isNonEmptyString(clientState) + ? Buffer.from(clientState) + : null; + const expectedClientStateBuffer = isNonEmptyString(expectedClientState) + ? Buffer.from(expectedClientState) + : null; + + if ( + !isDefined(clientStateBuffer) || + !isDefined(expectedClientStateBuffer) || + clientStateBuffer.length !== expectedClientStateBuffer.length || + !timingSafeEqual(clientStateBuffer, expectedClientStateBuffer) + ) { + this.logger.warn( + `Client state mismatch for subscription ${notification.subscriptionId}`, + ); + continue; + } + + if ( + isNonEmptyString(notification.lifecycleEvent) && + notification.lifecycleEvent !== 'missed' + ) { + await this.calendarWebhookSubscriptionService.renewSubscription( + calendarChannel, + ); + continue; + } + + await this.webhookSyncTriggerService.triggerCalendarSync( + calendarChannel.id, + calendarChannel.workspaceId, + ); + } + } +} 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 new file mode 100644 index 0000000000..c33ebf0904 --- /dev/null +++ b/packages/twenty-server/src/modules/connected-account-sync-webhooks/drivers/microsoft/microsoft-messaging-notification.handler.ts @@ -0,0 +1,100 @@ +import { timingSafeEqual } from 'crypto'; + +import { Injectable, Logger } from '@nestjs/common'; +import { InjectRepository } from '@nestjs/typeorm'; + +import { isNonEmptyString } from '@sniptt/guards'; +import { isDefined } from 'twenty-shared/utils'; +import { In, Repository } from 'typeorm'; + +import { MessageChannelEntity } from 'src/engine/metadata-modules/message-channel/entities/message-channel.entity'; +import { MessagingWebhookSubscriptionService } from 'src/modules/connected-account/webhook-subscription-manager/services/messaging-webhook-subscription.service'; +import { WebhookSyncTriggerService } from 'src/modules/connected-account/webhook-subscription-manager/services/webhook-sync-trigger.service'; +import { type MicrosoftGraphNotification } from 'src/modules/connected-account-sync-webhooks/types/microsoft-graph-notification.type'; +import { type WebhookNotificationHandler } from 'src/modules/connected-account-sync-webhooks/types/webhook-notification-handler.type'; + +@Injectable() +export class MicrosoftMessagingNotificationHandler implements WebhookNotificationHandler< + MicrosoftGraphNotification[] +> { + private readonly logger = new Logger( + MicrosoftMessagingNotificationHandler.name, + ); + + constructor( + @InjectRepository(MessageChannelEntity) + private readonly messageChannelRepository: Repository, + private readonly messagingWebhookSubscriptionService: MessagingWebhookSubscriptionService, + private readonly webhookSyncTriggerService: WebhookSyncTriggerService, + ) {} + + async handle(notifications: MicrosoftGraphNotification[]): Promise { + const subscriptionIds = notifications + .map((notification) => notification.subscriptionId) + .filter(isNonEmptyString); + + if (subscriptionIds.length === 0) { + return; + } + + const messageChannels = await this.messageChannelRepository.find({ + where: { webhookSubscriptionExternalId: In(subscriptionIds) }, + }); + + const messageChannelByExternalId = new Map( + messageChannels.map((messageChannel) => [ + messageChannel.webhookSubscriptionExternalId, + messageChannel, + ]), + ); + + for (const notification of notifications) { + const messageChannel = messageChannelByExternalId.get( + notification.subscriptionId, + ); + + if (!isDefined(messageChannel)) { + this.logger.warn( + `No messaging subscription found for ${notification.subscriptionId}`, + ); + continue; + } + + const clientState = notification.clientState; + const expectedClientState = messageChannel.webhookSubscriptionClientState; + const clientStateBuffer = isNonEmptyString(clientState) + ? Buffer.from(clientState) + : null; + const expectedClientStateBuffer = isNonEmptyString(expectedClientState) + ? Buffer.from(expectedClientState) + : null; + + if ( + !isDefined(clientStateBuffer) || + !isDefined(expectedClientStateBuffer) || + clientStateBuffer.length !== expectedClientStateBuffer.length || + !timingSafeEqual(clientStateBuffer, expectedClientStateBuffer) + ) { + this.logger.warn( + `Client state mismatch for subscription ${notification.subscriptionId}`, + ); + continue; + } + + if ( + isNonEmptyString(notification.lifecycleEvent) && + notification.lifecycleEvent !== 'missed' + ) { + await this.messagingWebhookSubscriptionService.renewSubscription( + messageChannel, + ); + continue; + } + + await this.webhookSyncTriggerService.triggerMessagingSync( + messageChannel.id, + messageChannel.workspaceId, + ); + } + } +} diff --git a/packages/twenty-server/src/modules/connected-account-sync-webhooks/drivers/microsoft/microsoft-webhook-driver.module.ts b/packages/twenty-server/src/modules/connected-account-sync-webhooks/drivers/microsoft/microsoft-webhook-driver.module.ts new file mode 100644 index 0000000000..8c9f4bab85 --- /dev/null +++ b/packages/twenty-server/src/modules/connected-account-sync-webhooks/drivers/microsoft/microsoft-webhook-driver.module.ts @@ -0,0 +1,28 @@ +import { Module } from '@nestjs/common'; +import { TypeOrmModule } from '@nestjs/typeorm'; + +import { TwentyConfigModule } from 'src/engine/core-modules/twenty-config/twenty-config.module'; +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 { WebhookSyncTriggerService } from 'src/modules/connected-account/webhook-subscription-manager/services/webhook-sync-trigger.service'; +import { WebhookSubscriptionModule } from 'src/modules/connected-account/webhook-subscription-manager/webhook-subscription.module'; +import { MicrosoftCalendarNotificationHandler } from 'src/modules/connected-account-sync-webhooks/drivers/microsoft/microsoft-calendar-notification.handler'; +import { MicrosoftMessagingNotificationHandler } from 'src/modules/connected-account-sync-webhooks/drivers/microsoft/microsoft-messaging-notification.handler'; + +@Module({ + imports: [ + TwentyConfigModule, + WebhookSubscriptionModule, + TypeOrmModule.forFeature([MessageChannelEntity, CalendarChannelEntity]), + ], + providers: [ + MicrosoftMessagingNotificationHandler, + MicrosoftCalendarNotificationHandler, + WebhookSyncTriggerService, + ], + exports: [ + MicrosoftMessagingNotificationHandler, + MicrosoftCalendarNotificationHandler, + ], +}) +export class MicrosoftWebhookDriverModule {} diff --git a/packages/twenty-server/src/modules/connected-account-sync-webhooks/filters/connected-account-sync-webhook-api-exception.filter.ts b/packages/twenty-server/src/modules/connected-account-sync-webhooks/filters/connected-account-sync-webhook-api-exception.filter.ts new file mode 100644 index 0000000000..ae2ae50ace --- /dev/null +++ b/packages/twenty-server/src/modules/connected-account-sync-webhooks/filters/connected-account-sync-webhook-api-exception.filter.ts @@ -0,0 +1,29 @@ +import { + type ArgumentsHost, + Catch, + type ExceptionFilter, +} from '@nestjs/common'; + +import { type Response } from 'express'; + +import { HttpExceptionHandlerService } from 'src/engine/core-modules/exception-handler/http-exception-handler.service'; +import { ConnectedAccountSyncWebhookException } from 'src/modules/connected-account-sync-webhooks/connected-account-sync-webhook.exception'; +import { getConnectedAccountSyncWebhookExceptionStatusCode } from 'src/modules/connected-account-sync-webhooks/utils/get-connected-account-sync-webhook-exception-status-code.util'; + +@Catch(ConnectedAccountSyncWebhookException) +export class ConnectedAccountSyncWebhookApiExceptionFilter implements ExceptionFilter { + constructor( + private readonly httpExceptionHandlerService: HttpExceptionHandlerService, + ) {} + + catch(exception: ConnectedAccountSyncWebhookException, host: ArgumentsHost) { + const ctx = host.switchToHttp(); + const response = ctx.getResponse(); + + return this.httpExceptionHandlerService.handleError( + exception, + response, + getConnectedAccountSyncWebhookExceptionStatusCode(exception), + ); + } +} diff --git a/packages/twenty-server/src/modules/connected-account-sync-webhooks/types/google-calendar-notification.type.ts b/packages/twenty-server/src/modules/connected-account-sync-webhooks/types/google-calendar-notification.type.ts new file mode 100644 index 0000000000..518f5691a8 --- /dev/null +++ b/packages/twenty-server/src/modules/connected-account-sync-webhooks/types/google-calendar-notification.type.ts @@ -0,0 +1,5 @@ +export type GoogleCalendarChannelNotification = { + channelId: string | undefined; + resourceState: string | undefined; + channelToken: string | undefined; +}; diff --git a/packages/twenty-server/src/modules/connected-account-sync-webhooks/types/google-pubsub-push.type.ts b/packages/twenty-server/src/modules/connected-account-sync-webhooks/types/google-pubsub-push.type.ts new file mode 100644 index 0000000000..68d88ee2c3 --- /dev/null +++ b/packages/twenty-server/src/modules/connected-account-sync-webhooks/types/google-pubsub-push.type.ts @@ -0,0 +1,13 @@ +export type GooglePubSubPushMessage = { + message?: { + data?: string; + messageId?: string; + publishTime?: string; + }; + subscription?: string; +}; + +export type GmailPushDecodedData = { + emailAddress: string; + historyId: string | number; +}; 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 new file mode 100644 index 0000000000..4d022d3863 --- /dev/null +++ b/packages/twenty-server/src/modules/connected-account-sync-webhooks/types/microsoft-graph-notification.type.ts @@ -0,0 +1,12 @@ +export type MicrosoftGraphNotification = { + subscriptionId: string; + clientState?: string; + changeType?: string; + resource?: string; + lifecycleEvent?: string; + resourceData?: { id?: string } | null; +}; + +export type MicrosoftGraphNotificationPayload = { + value?: MicrosoftGraphNotification[]; +}; diff --git a/packages/twenty-server/src/modules/connected-account-sync-webhooks/types/webhook-notification-handler.type.ts b/packages/twenty-server/src/modules/connected-account-sync-webhooks/types/webhook-notification-handler.type.ts new file mode 100644 index 0000000000..00b3f4f114 --- /dev/null +++ b/packages/twenty-server/src/modules/connected-account-sync-webhooks/types/webhook-notification-handler.type.ts @@ -0,0 +1,3 @@ +export type WebhookNotificationHandler = { + handle(request: TRequest): Promise; +}; diff --git a/packages/twenty-server/src/modules/connected-account-sync-webhooks/utils/get-connected-account-sync-webhook-exception-status-code.util.ts b/packages/twenty-server/src/modules/connected-account-sync-webhooks/utils/get-connected-account-sync-webhook-exception-status-code.util.ts new file mode 100644 index 0000000000..8e27d66cdd --- /dev/null +++ b/packages/twenty-server/src/modules/connected-account-sync-webhooks/utils/get-connected-account-sync-webhook-exception-status-code.util.ts @@ -0,0 +1,21 @@ +import { assertUnreachable } from 'twenty-shared/utils'; + +import { + ConnectedAccountSyncWebhookException, + ConnectedAccountSyncWebhookExceptionCode, +} from 'src/modules/connected-account-sync-webhooks/connected-account-sync-webhook.exception'; + +export const getConnectedAccountSyncWebhookExceptionStatusCode = ( + exception: ConnectedAccountSyncWebhookException, +): 400 | 403 => { + switch (exception.code) { + case ConnectedAccountSyncWebhookExceptionCode.MISSING_REQUEST_BODY: + case ConnectedAccountSyncWebhookExceptionCode.INVALID_PAYLOAD: + return 400; + case ConnectedAccountSyncWebhookExceptionCode.INVALID_SIGNATURE: + return 403; + default: { + return assertUnreachable(exception.code); + } + } +}; diff --git a/packages/twenty-server/src/modules/connected-account/channel-sync/channel-sync.module.ts b/packages/twenty-server/src/modules/connected-account/channel-sync/channel-sync.module.ts index 4ba36f589c..37fd5c621d 100644 --- a/packages/twenty-server/src/modules/connected-account/channel-sync/channel-sync.module.ts +++ b/packages/twenty-server/src/modules/connected-account/channel-sync/channel-sync.module.ts @@ -8,6 +8,7 @@ import { PermissionsModule } from 'src/engine/metadata-modules/permissions/permi import { WorkspaceDataSourceModule } from 'src/engine/workspace-datasource/workspace-datasource.module'; import { ChannelSyncResolver } from 'src/modules/connected-account/channel-sync/channel-sync.resolver'; import { ChannelSyncService } from 'src/modules/connected-account/channel-sync/services/channel-sync.service'; +import { WebhookSubscriptionModule } from 'src/modules/connected-account/webhook-subscription-manager/webhook-subscription.module'; import { MessagingCommonModule } from 'src/modules/messaging/common/messaging-common.module'; @Module({ @@ -17,6 +18,7 @@ import { MessagingCommonModule } from 'src/modules/messaging/common/messaging-co PermissionsModule, WorkspaceDataSourceModule, MessagingCommonModule, + WebhookSubscriptionModule, ], providers: [ChannelSyncResolver, ChannelSyncService], exports: [ChannelSyncService], diff --git a/packages/twenty-server/src/modules/connected-account/channel-sync/services/channel-sync.service.ts b/packages/twenty-server/src/modules/connected-account/channel-sync/services/channel-sync.service.ts index d007fe49a6..8f84746ba0 100644 --- a/packages/twenty-server/src/modules/connected-account/channel-sync/services/channel-sync.service.ts +++ b/packages/twenty-server/src/modules/connected-account/channel-sync/services/channel-sync.service.ts @@ -1,14 +1,14 @@ -import { Injectable } from '@nestjs/common'; +import { Injectable, Logger } from '@nestjs/common'; import { InjectRepository } from '@nestjs/typeorm'; -import { Not, Repository } from 'typeorm'; - import { CalendarChannelSyncStage, CalendarChannelSyncStatus, MessageChannelSyncStage, MessageChannelType, } from 'twenty-shared/types'; +import { Not, Repository } from 'typeorm'; + import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator'; import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants'; import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service'; @@ -20,6 +20,8 @@ import { CalendarEventListFetchJob, type CalendarEventListFetchJobData, } from 'src/modules/calendar/calendar-event-import-manager/jobs/calendar-event-list-fetch.job'; +import { CalendarWebhookSubscriptionService } from 'src/modules/connected-account/webhook-subscription-manager/services/calendar-webhook-subscription.service'; +import { MessagingWebhookSubscriptionService } from 'src/modules/connected-account/webhook-subscription-manager/services/messaging-webhook-subscription.service'; import { MessageChannelSyncStatusService } from 'src/modules/messaging/common/services/message-channel-sync-status.service'; import { MessagingMessageListFetchJob, @@ -33,6 +35,8 @@ export type StartChannelSyncInput = { @Injectable() export class ChannelSyncService { + private readonly logger = new Logger(ChannelSyncService.name); + constructor( private readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager, @InjectMessageQueue(MessageQueue.messagingQueue) @@ -44,6 +48,8 @@ export class ChannelSyncService { private readonly messageChannelSyncStatusService: MessageChannelSyncStatusService, @InjectRepository(CalendarChannelEntity) private readonly calendarChannelRepository: Repository, + private readonly messagingWebhookSubscriptionService: MessagingWebhookSubscriptionService, + private readonly calendarWebhookSubscriptionService: CalendarWebhookSubscriptionService, ) {} async startChannelSync(input: StartChannelSyncInput): Promise { @@ -82,6 +88,18 @@ export class ChannelSyncService { messageChannelId: messageChannel.id, }, ); + + try { + await this.messagingWebhookSubscriptionService.createSubscription( + messageChannel.id, + workspaceId, + ); + } catch (error) { + this.logger.warn( + `Failed to create messaging webhook subscription for message channel ${messageChannel.id}`, + error, + ); + } } }, authContext); } @@ -118,6 +136,18 @@ export class ChannelSyncService { calendarChannelId: calendarChannel.id, }, ); + + try { + await this.calendarWebhookSubscriptionService.createSubscription( + calendarChannel.id, + workspaceId, + ); + } catch (error) { + this.logger.warn( + `Failed to create calendar webhook subscription for calendar channel ${calendarChannel.id}`, + error, + ); + } } }, authContext); } diff --git a/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/constants/webhook-subscription-renewal-buffer-ms.constant.ts b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/constants/webhook-subscription-renewal-buffer-ms.constant.ts new file mode 100644 index 0000000000..75f4c5b4e4 --- /dev/null +++ b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/constants/webhook-subscription-renewal-buffer-ms.constant.ts @@ -0,0 +1 @@ +export const WEBHOOK_SUBSCRIPTION_RENEWAL_BUFFER_MS = 24 * 60 * 60 * 1000; diff --git a/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/constants/webhook-subscription-renewal-cron-pattern.constant.ts b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/constants/webhook-subscription-renewal-cron-pattern.constant.ts new file mode 100644 index 0000000000..dc3bdee0ac --- /dev/null +++ b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/constants/webhook-subscription-renewal-cron-pattern.constant.ts @@ -0,0 +1 @@ +export const WEBHOOK_SUBSCRIPTION_RENEWAL_CRON_PATTERN = '0 * * * *'; diff --git a/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/crons/commands/webhook-subscription-renewal.cron.command.ts b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/crons/commands/webhook-subscription-renewal.cron.command.ts new file mode 100644 index 0000000000..3604353858 --- /dev/null +++ b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/crons/commands/webhook-subscription-renewal.cron.command.ts @@ -0,0 +1,33 @@ +import { Command, CommandRunner } from 'nest-commander'; + +import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator'; +import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants'; +import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service'; +import { WEBHOOK_SUBSCRIPTION_RENEWAL_CRON_PATTERN } from 'src/modules/connected-account/webhook-subscription-manager/constants/webhook-subscription-renewal-cron-pattern.constant'; +import { WebhookSubscriptionRenewalCronJob } from 'src/modules/connected-account/webhook-subscription-manager/crons/jobs/webhook-subscription-renewal.cron.job'; + +@Command({ + name: 'cron:messaging-calendar:webhook-subscription-renewal', + description: + 'Starts a cron job to renew messaging/calendar webhook subscriptions before they expire', +}) +export class WebhookSubscriptionRenewalCronCommand extends CommandRunner { + constructor( + @InjectMessageQueue(MessageQueue.cronQueue) + private readonly messageQueueService: MessageQueueService, + ) { + super(); + } + + async run(): Promise { + await this.messageQueueService.addCron({ + jobName: WebhookSubscriptionRenewalCronJob.name, + data: undefined, + options: { + repeat: { + pattern: WEBHOOK_SUBSCRIPTION_RENEWAL_CRON_PATTERN, + }, + }, + }); + } +} diff --git a/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/crons/jobs/webhook-subscription-renewal.cron.job.ts b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/crons/jobs/webhook-subscription-renewal.cron.job.ts new file mode 100644 index 0000000000..7572982f3a --- /dev/null +++ b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/crons/jobs/webhook-subscription-renewal.cron.job.ts @@ -0,0 +1,92 @@ +import { Logger } from '@nestjs/common'; +import { InjectRepository } from '@nestjs/typeorm'; + +import { WebhookSubscriptionStatus } from 'twenty-shared/types'; +import { LessThanOrEqual, Repository } from 'typeorm'; + +import { SentryCronMonitor } from 'src/engine/core-modules/cron/sentry-cron-monitor.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 { 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'; + +@Processor(MessageQueue.cronQueue) +export class WebhookSubscriptionRenewalCronJob { + private readonly logger = new Logger(WebhookSubscriptionRenewalCronJob.name); + + constructor( + @InjectRepository(MessageChannelEntity) + private readonly messageChannelRepository: Repository, + @InjectRepository(CalendarChannelEntity) + private readonly calendarChannelRepository: Repository, + private readonly messagingWebhookSubscriptionService: MessagingWebhookSubscriptionService, + private readonly calendarWebhookSubscriptionService: CalendarWebhookSubscriptionService, + ) {} + + @Process(WebhookSubscriptionRenewalCronJob.name) + @SentryCronMonitor( + WebhookSubscriptionRenewalCronJob.name, + WEBHOOK_SUBSCRIPTION_RENEWAL_CRON_PATTERN, + ) + async handle(): Promise { + 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 expiringCalendarChannels = await this.calendarChannelRepository.find({ + where: { + webhookSubscriptionStatus: WebhookSubscriptionStatus.ACTIVE, + webhookSubscriptionExpiresAt: LessThanOrEqual(renewalThreshold), + }, + }); + + 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, + ); + } + } + } +} 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 new file mode 100644 index 0000000000..92fbb7ef54 --- /dev/null +++ b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/drivers/exceptions/webhook-subscription-driver.exception.ts @@ -0,0 +1,38 @@ +import { type MessageDescriptor } from '@lingui/core'; +import { msg } from '@lingui/core/macro'; +import { assertUnreachable } from 'twenty-shared/utils'; + +import { CustomException } from 'src/utils/custom-exception'; + +export enum WebhookSubscriptionDriverExceptionCode { + PROVIDER_NOT_CONFIGURED = 'PROVIDER_NOT_CONFIGURED', + PROVIDER_RESPONSE_INVALID = 'PROVIDER_RESPONSE_INVALID', + UNSUPPORTED_PROVIDER = 'UNSUPPORTED_PROVIDER', +} + +const getWebhookSubscriptionDriverExceptionUserFriendlyMessage = ( + code: WebhookSubscriptionDriverExceptionCode, +) => { + switch (code) { + case WebhookSubscriptionDriverExceptionCode.PROVIDER_NOT_CONFIGURED: + case WebhookSubscriptionDriverExceptionCode.PROVIDER_RESPONSE_INVALID: + case WebhookSubscriptionDriverExceptionCode.UNSUPPORTED_PROVIDER: + return msg`The webhook subscription could not be managed for this account.`; + default: + assertUnreachable(code); + } +}; + +export class WebhookSubscriptionDriverException extends CustomException { + constructor( + message: string, + code: WebhookSubscriptionDriverExceptionCode, + { userFriendlyMessage }: { userFriendlyMessage?: MessageDescriptor } = {}, + ) { + super(message, code, { + userFriendlyMessage: + userFriendlyMessage ?? + getWebhookSubscriptionDriverExceptionUserFriendlyMessage(code), + }); + } +} diff --git a/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/drivers/google/google-calendar-watch-ttl-ms.constant.ts b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/drivers/google/google-calendar-watch-ttl-ms.constant.ts new file mode 100644 index 0000000000..37803fc114 --- /dev/null +++ b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/drivers/google/google-calendar-watch-ttl-ms.constant.ts @@ -0,0 +1 @@ +export const GOOGLE_CALENDAR_WATCH_TTL_MS = 7 * 24 * 60 * 60 * 1000; 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 new file mode 100644 index 0000000000..f333a45ee0 --- /dev/null +++ b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/drivers/google/google-webhook-subscription.driver.ts @@ -0,0 +1,178 @@ +import { Injectable } from '@nestjs/common'; + +import { isNonEmptyString } from '@sniptt/guards'; +import { type calendar_v3, type gmail_v1, google } from 'googleapis'; +import { isDefined } from 'twenty-shared/utils'; +import { v4 } from 'uuid'; + +import { WebhookSubscriptionChannelType } from 'twenty-shared/types'; + +import { TwentyConfigService } from 'src/engine/core-modules/twenty-config/twenty-config.service'; +import { GOOGLE_CALENDAR_WATCH_TTL_MS } from 'src/modules/connected-account/webhook-subscription-manager/drivers/google/google-calendar-watch-ttl-ms.constant'; +import { + WebhookSubscriptionDriverException, + WebhookSubscriptionDriverExceptionCode, +} from 'src/modules/connected-account/webhook-subscription-manager/drivers/exceptions/webhook-subscription-driver.exception'; +import { + type WebhookSubscriptionContext, + type WebhookSubscriptionDriver, + type WebhookSubscriptionResult, +} from 'src/modules/connected-account/webhook-subscription-manager/types/webhook-subscription-driver.type'; +import { GoogleOAuth2ClientProvider } from 'src/modules/connected-account/oauth2-client-manager/drivers/google/google-oauth2-client.provider'; + +@Injectable() +export class GoogleWebhookSubscriptionDriver implements WebhookSubscriptionDriver { + constructor( + private readonly googleOAuth2ClientProvider: GoogleOAuth2ClientProvider, + private readonly twentyConfigService: TwentyConfigService, + ) {} + + async createSubscription( + connectedAccountId: string, + channelType: WebhookSubscriptionChannelType, + clientState: string, + ): Promise { + return channelType === WebhookSubscriptionChannelType.MESSAGING + ? this.watchGmailMailbox(connectedAccountId) + : this.watchPrimaryCalendar(connectedAccountId, clientState); + } + + async renewSubscription( + context: WebhookSubscriptionContext, + ): Promise { + if (context.channelType === WebhookSubscriptionChannelType.CALENDAR) { + await this.deleteSubscription(context); + } + + return this.createSubscription( + context.connectedAccountId, + context.channelType, + context.clientState, + ); + } + + async deleteSubscription(context: WebhookSubscriptionContext): Promise { + return context.channelType === WebhookSubscriptionChannelType.MESSAGING + ? this.stopGmailMailboxWatch(context.connectedAccountId) + : this.stopCalendarWatch(context); + } + + private async watchGmailMailbox( + connectedAccountId: string, + ): Promise { + const pubSubTopicName = this.twentyConfigService.get( + 'MESSAGING_GMAIL_PUBSUB_TOPIC', + ); + + if (!isNonEmptyString(pubSubTopicName)) { + throw new WebhookSubscriptionDriverException( + 'MESSAGING_GMAIL_PUBSUB_TOPIC is not configured', + WebhookSubscriptionDriverExceptionCode.PROVIDER_NOT_CONFIGURED, + ); + } + + const gmailClient = await this.getGmailClient(connectedAccountId); + + const { data } = await gmailClient.users.watch({ + userId: 'me', + requestBody: { + topicName: pubSubTopicName, + }, + }); + + if (!isDefined(data.expiration)) { + throw new WebhookSubscriptionDriverException( + 'Gmail watch response did not include an expiration', + WebhookSubscriptionDriverExceptionCode.PROVIDER_RESPONSE_INVALID, + ); + } + + return { + externalSubscriptionId: null, + externalResourceId: null, + expiresAt: new Date(Number(data.expiration)), + }; + } + + private async stopGmailMailboxWatch( + connectedAccountId: string, + ): Promise { + const gmailClient = await this.getGmailClient(connectedAccountId); + + await gmailClient.users.stop({ userId: 'me' }); + } + + private async watchPrimaryCalendar( + connectedAccountId: string, + clientState: string, + ): Promise { + const calendarClient = await this.getCalendarClient(connectedAccountId); + + 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) }, + }, + }); + + if (!isDefined(data.resourceId) || !isDefined(data.expiration)) { + throw new WebhookSubscriptionDriverException( + 'Google Calendar watch response did not include a resourceId or expiration', + WebhookSubscriptionDriverExceptionCode.PROVIDER_RESPONSE_INVALID, + ); + } + + return { + externalSubscriptionId: watchChannelId, + externalResourceId: data.resourceId, + expiresAt: new Date(Number(data.expiration)), + }; + } + + private async stopCalendarWatch( + context: WebhookSubscriptionContext, + ): Promise { + if ( + !isDefined(context.externalSubscriptionId) || + !isDefined(context.externalResourceId) + ) { + return; + } + + const calendarClient = await this.getCalendarClient( + context.connectedAccountId, + ); + + await calendarClient.channels.stop({ + requestBody: { + id: context.externalSubscriptionId, + resourceId: context.externalResourceId, + }, + }); + } + + private async getGmailClient( + connectedAccountId: string, + ): Promise { + const oAuth2Client = + await this.googleOAuth2ClientProvider.getClient(connectedAccountId); + + return google.gmail({ version: 'v1', auth: oAuth2Client }); + } + + private async getCalendarClient( + connectedAccountId: string, + ): Promise { + const oAuth2Client = + await this.googleOAuth2ClientProvider.getClient(connectedAccountId); + + return google.calendar({ version: 'v3', auth: oAuth2Client }); + } +} diff --git a/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/drivers/microsoft/microsoft-subscription-ttl-ms.constant.ts b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/drivers/microsoft/microsoft-subscription-ttl-ms.constant.ts new file mode 100644 index 0000000000..6ca6914da1 --- /dev/null +++ b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/drivers/microsoft/microsoft-subscription-ttl-ms.constant.ts @@ -0,0 +1 @@ +export const MICROSOFT_SUBSCRIPTION_TTL_MS = 7 * 24 * 60 * 60 * 1000; 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 new file mode 100644 index 0000000000..fb2595202f --- /dev/null +++ b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/drivers/microsoft/microsoft-webhook-subscription.driver.ts @@ -0,0 +1,132 @@ +import { Injectable } from '@nestjs/common'; + +import { type Subscription } from '@microsoft/microsoft-graph-types'; +import { isDefined } from 'twenty-shared/utils'; + +import { WebhookSubscriptionChannelType } from 'twenty-shared/types'; + +import { TwentyConfigService } from 'src/engine/core-modules/twenty-config/twenty-config.service'; +import { MICROSOFT_SUBSCRIPTION_TTL_MS } from 'src/modules/connected-account/webhook-subscription-manager/drivers/microsoft/microsoft-subscription-ttl-ms.constant'; +import { + WebhookSubscriptionDriverException, + WebhookSubscriptionDriverExceptionCode, +} from 'src/modules/connected-account/webhook-subscription-manager/drivers/exceptions/webhook-subscription-driver.exception'; +import { + type WebhookSubscriptionContext, + type WebhookSubscriptionDriver, + 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'; + +type MicrosoftGraphResourceConfig = Pick< + Subscription, + 'resource' | 'changeType' +> & { + notificationPath: string; +}; + +const MICROSOFT_GRAPH_RESOURCE_CONFIG_BY_CHANNEL_TYPE: Record< + WebhookSubscriptionChannelType, + MicrosoftGraphResourceConfig +> = { + [WebhookSubscriptionChannelType.MESSAGING]: { + resource: '/me/messages', + changeType: 'created,updated', + notificationPath: 'webhooks/microsoft/messaging', + }, + [WebhookSubscriptionChannelType.CALENDAR]: { + resource: '/me/events', + changeType: 'created,updated,deleted', + notificationPath: 'webhooks/microsoft/calendar', + }, +}; + +@Injectable() +export class MicrosoftWebhookSubscriptionDriver implements WebhookSubscriptionDriver { + constructor( + private readonly microsoftOAuth2ClientProvider: MicrosoftOAuth2ClientProvider, + private readonly twentyConfigService: TwentyConfigService, + ) {} + + async createSubscription( + connectedAccountId: string, + channelType: WebhookSubscriptionChannelType, + clientState: string, + ): Promise { + const resourceConfig = + MICROSOFT_GRAPH_RESOURCE_CONFIG_BY_CHANNEL_TYPE[channelType]; + const graphClient = + await this.microsoftOAuth2ClientProvider.getClient(connectedAccountId); + + const notificationUrl = `${this.twentyConfigService.get('SERVER_URL')}/${resourceConfig.notificationPath}`; + + const subscriptionPayload: Subscription = { + changeType: resourceConfig.changeType, + notificationUrl, + lifecycleNotificationUrl: notificationUrl, + resource: resourceConfig.resource, + expirationDateTime: new Date( + Date.now() + MICROSOFT_SUBSCRIPTION_TTL_MS, + ).toISOString(), + clientState, + }; + + const subscription: Subscription = await graphClient + .api('/subscriptions') + .post(subscriptionPayload); + + return this.toResult(subscription); + } + + async renewSubscription( + context: WebhookSubscriptionContext, + ): Promise { + const graphClient = await this.microsoftOAuth2ClientProvider.getClient( + context.connectedAccountId, + ); + + const subscriptionPatch: Subscription = { + expirationDateTime: new Date( + Date.now() + MICROSOFT_SUBSCRIPTION_TTL_MS, + ).toISOString(), + }; + + const renewedSubscription: Subscription = await graphClient + .api(`/subscriptions/${context.externalSubscriptionId}`) + .patch(subscriptionPatch); + + return this.toResult(renewedSubscription); + } + + async deleteSubscription(context: WebhookSubscriptionContext): Promise { + if (!isDefined(context.externalSubscriptionId)) { + return; + } + + const graphClient = await this.microsoftOAuth2ClientProvider.getClient( + context.connectedAccountId, + ); + + await graphClient + .api(`/subscriptions/${context.externalSubscriptionId}`) + .delete(); + } + + private toResult(subscription: Subscription): WebhookSubscriptionResult { + if ( + !isDefined(subscription.id) || + !isDefined(subscription.expirationDateTime) + ) { + throw new WebhookSubscriptionDriverException( + 'Microsoft Graph subscription response did not include an id or expiration', + WebhookSubscriptionDriverExceptionCode.PROVIDER_RESPONSE_INVALID, + ); + } + + return { + externalSubscriptionId: subscription.id, + externalResourceId: null, + expiresAt: new Date(subscription.expirationDateTime), + }; + } +} diff --git a/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/listeners/webhook-subscription-channel-deleted.listener.ts b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/listeners/webhook-subscription-channel-deleted.listener.ts new file mode 100644 index 0000000000..580a8d2505 --- /dev/null +++ b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/listeners/webhook-subscription-channel-deleted.listener.ts @@ -0,0 +1,56 @@ +import { Injectable } from '@nestjs/common'; + +import { isDefined } from 'twenty-shared/utils'; + +import { OnCustomBatchEvent } from 'src/engine/api/graphql/graphql-query-runner/decorators/on-custom-batch-event.decorator'; +import { CALENDAR_CHANNEL_DELETED_EVENT } from 'src/engine/metadata-modules/calendar-channel/constants/calendar-channel-deleted.constant'; +import { type CalendarChannelDeletedEvent } from 'src/engine/metadata-modules/calendar-channel/types/calendar-channel-deleted.type'; +import { MESSAGE_CHANNEL_DELETED_EVENT } from 'src/engine/metadata-modules/message-channel/constants/message-channel-deleted.constant'; +import { type MessageChannelDeletedEvent } from 'src/engine/metadata-modules/message-channel/types/message-channel-deleted.type'; +import { CustomWorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/custom-workspace-batch-event.type'; +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'; + +@Injectable() +export class WebhookSubscriptionChannelDeletedListener { + constructor( + private readonly messagingWebhookSubscriptionService: MessagingWebhookSubscriptionService, + private readonly calendarWebhookSubscriptionService: CalendarWebhookSubscriptionService, + ) {} + + @OnCustomBatchEvent(MESSAGE_CHANNEL_DELETED_EVENT) + async handleMessageChannelDeleted( + batchEvent: CustomWorkspaceEventBatch, + ): Promise { + const { workspaceId } = batchEvent; + + if (!isDefined(workspaceId)) { + return; + } + + for (const event of batchEvent.events) { + await this.messagingWebhookSubscriptionService.deleteSubscription( + event.messageChannelId, + workspaceId, + ); + } + } + + @OnCustomBatchEvent(CALENDAR_CHANNEL_DELETED_EVENT) + async handleCalendarChannelDeleted( + batchEvent: CustomWorkspaceEventBatch, + ): Promise { + const { workspaceId } = batchEvent; + + if (!isDefined(workspaceId)) { + return; + } + + for (const event of batchEvent.events) { + await this.calendarWebhookSubscriptionService.deleteSubscription( + event.calendarChannelId, + workspaceId, + ); + } + } +} 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 new file mode 100644 index 0000000000..8135e188d0 --- /dev/null +++ b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/services/calendar-webhook-subscription.service.ts @@ -0,0 +1,196 @@ +import { Injectable } from '@nestjs/common'; +import { InjectRepository } from '@nestjs/typeorm'; + +import { + FeatureFlagKey, + WebhookSubscriptionChannelType, + WebhookSubscriptionStatus, +} from 'twenty-shared/types'; +import { isDefined } from 'twenty-shared/utils'; +import { Repository } from 'typeorm'; +import { v4 } from 'uuid'; + +import { ExceptionHandlerService } from 'src/engine/core-modules/exception-handler/exception-handler.service'; +import { FeatureFlagService } from 'src/engine/core-modules/feature-flag/services/feature-flag.service'; +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 { WebhookSubscriptionDriverFactory } from 'src/modules/connected-account/webhook-subscription-manager/services/webhook-subscription-driver-factory.service'; +import { type WebhookSubscriptionContext } from 'src/modules/connected-account/webhook-subscription-manager/types/webhook-subscription-driver.type'; + +@Injectable() +export class CalendarWebhookSubscriptionService { + constructor( + @InjectRepository(ConnectedAccountEntity) + private readonly connectedAccountRepository: Repository, + @InjectRepository(CalendarChannelEntity) + private readonly calendarChannelRepository: Repository, + private readonly webhookSubscriptionDriverFactory: WebhookSubscriptionDriverFactory, + private readonly featureFlagService: FeatureFlagService, + private readonly exceptionHandlerService: ExceptionHandlerService, + ) {} + + async createSubscription( + calendarChannelId: string, + workspaceId: string, + ): Promise { + const isWebhookEnabled = await this.featureFlagService.isFeatureEnabled( + FeatureFlagKey.IS_MESSAGING_CALENDAR_WEBHOOK_ENABLED, + workspaceId, + ); + + if (!isWebhookEnabled) { + return; + } + + const calendarChannel = await this.calendarChannelRepository.findOne({ + where: { id: calendarChannelId, workspaceId }, + relations: ['connectedAccount'], + }); + + if (!isDefined(calendarChannel?.connectedAccount)) { + return; + } + + const { connectedAccount } = calendarChannel; + + if ( + !this.webhookSubscriptionDriverFactory.isProviderSupported( + connectedAccount.provider, + ) + ) { + return; + } + + if ( + calendarChannel.webhookSubscriptionStatus === + WebhookSubscriptionStatus.ACTIVE + ) { + return; + } + + const clientState = calendarChannel.webhookSubscriptionClientState ?? v4(); + const driver = this.webhookSubscriptionDriverFactory.getDriver( + connectedAccount.provider, + ); + + if (isDefined(calendarChannel.webhookSubscriptionExternalId)) { + await driver + .deleteSubscription(this.toContext(calendarChannel)) + .catch(() => undefined); + } + + try { + const result = await driver.createSubscription( + calendarChannel.connectedAccountId, + WebhookSubscriptionChannelType.CALENDAR, + clientState, + ); + + await this.calendarChannelRepository.update(calendarChannel.id, { + webhookSubscriptionExternalId: result.externalSubscriptionId, + webhookSubscriptionExternalResourceId: result.externalResourceId, + webhookSubscriptionClientState: clientState, + webhookSubscriptionStatus: WebhookSubscriptionStatus.ACTIVE, + webhookSubscriptionExpiresAt: result.expiresAt, + }); + } catch (error) { + await this.calendarChannelRepository.update(calendarChannel.id, { + webhookSubscriptionClientState: clientState, + webhookSubscriptionStatus: WebhookSubscriptionStatus.FAILED, + webhookSubscriptionExpiresAt: null, + }); + + this.exceptionHandlerService.captureExceptions([error], { + workspace: { id: workspaceId }, + }); + } + } + + async renewSubscription( + calendarChannel: CalendarChannelEntity, + ): Promise { + const connectedAccount = await this.connectedAccountRepository.findOne({ + where: { + id: calendarChannel.connectedAccountId, + workspaceId: calendarChannel.workspaceId, + }, + }); + + if (!isDefined(connectedAccount)) { + return; + } + + const driver = this.webhookSubscriptionDriverFactory.getDriver( + connectedAccount.provider, + ); + + try { + const result = await driver.renewSubscription( + this.toContext(calendarChannel), + ); + + await this.calendarChannelRepository.update(calendarChannel.id, { + webhookSubscriptionExternalId: result.externalSubscriptionId, + webhookSubscriptionExternalResourceId: result.externalResourceId, + webhookSubscriptionStatus: WebhookSubscriptionStatus.ACTIVE, + webhookSubscriptionExpiresAt: result.expiresAt, + }); + } catch (error) { + await this.calendarChannelRepository.update(calendarChannel.id, { + webhookSubscriptionStatus: WebhookSubscriptionStatus.FAILED, + }); + + this.exceptionHandlerService.captureExceptions([error], { + workspace: { id: calendarChannel.workspaceId }, + }); + } + } + + async deleteSubscription( + calendarChannelId: string, + workspaceId: string, + ): Promise { + const calendarChannel = await this.calendarChannelRepository.findOne({ + where: { id: calendarChannelId, workspaceId }, + }); + + if (!isDefined(calendarChannel)) { + return; + } + + const connectedAccount = await this.connectedAccountRepository.findOne({ + where: { + id: calendarChannel.connectedAccountId, + workspaceId: calendarChannel.workspaceId, + }, + }); + + if (!isDefined(connectedAccount)) { + return; + } + + const driver = this.webhookSubscriptionDriverFactory.getDriver( + connectedAccount.provider, + ); + + try { + await driver.deleteSubscription(this.toContext(calendarChannel)); + } catch (error) { + this.exceptionHandlerService.captureExceptions([error], { + workspace: { id: calendarChannel.workspaceId }, + }); + } + } + + private toContext( + calendarChannel: CalendarChannelEntity, + ): WebhookSubscriptionContext { + return { + connectedAccountId: calendarChannel.connectedAccountId, + channelType: WebhookSubscriptionChannelType.CALENDAR, + externalSubscriptionId: calendarChannel.webhookSubscriptionExternalId, + externalResourceId: calendarChannel.webhookSubscriptionExternalResourceId, + clientState: calendarChannel.webhookSubscriptionClientState ?? '', + }; + } +} 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 new file mode 100644 index 0000000000..e1fc1719b7 --- /dev/null +++ b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/services/messaging-webhook-subscription.service.ts @@ -0,0 +1,192 @@ +import { Injectable } from '@nestjs/common'; +import { InjectRepository } from '@nestjs/typeorm'; + +import { + FeatureFlagKey, + WebhookSubscriptionChannelType, + WebhookSubscriptionStatus, +} from 'twenty-shared/types'; +import { isDefined } from 'twenty-shared/utils'; +import { Repository } from 'typeorm'; +import { v4 } from 'uuid'; + +import { ExceptionHandlerService } from 'src/engine/core-modules/exception-handler/exception-handler.service'; +import { FeatureFlagService } from 'src/engine/core-modules/feature-flag/services/feature-flag.service'; +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 { WebhookSubscriptionDriverFactory } from 'src/modules/connected-account/webhook-subscription-manager/services/webhook-subscription-driver-factory.service'; +import { type WebhookSubscriptionContext } from 'src/modules/connected-account/webhook-subscription-manager/types/webhook-subscription-driver.type'; + +@Injectable() +export class MessagingWebhookSubscriptionService { + constructor( + @InjectRepository(ConnectedAccountEntity) + private readonly connectedAccountRepository: Repository, + @InjectRepository(MessageChannelEntity) + private readonly messageChannelRepository: Repository, + private readonly webhookSubscriptionDriverFactory: WebhookSubscriptionDriverFactory, + private readonly featureFlagService: FeatureFlagService, + private readonly exceptionHandlerService: ExceptionHandlerService, + ) {} + + async createSubscription( + messageChannelId: string, + workspaceId: string, + ): Promise { + const isWebhookEnabled = await this.featureFlagService.isFeatureEnabled( + FeatureFlagKey.IS_MESSAGING_CALENDAR_WEBHOOK_ENABLED, + workspaceId, + ); + + if (!isWebhookEnabled) { + return; + } + + const messageChannel = await this.messageChannelRepository.findOne({ + where: { id: messageChannelId, workspaceId }, + relations: ['connectedAccount'], + }); + + if (!isDefined(messageChannel?.connectedAccount)) { + return; + } + + const { connectedAccount } = messageChannel; + + if ( + !this.webhookSubscriptionDriverFactory.isProviderSupported( + connectedAccount.provider, + ) + ) { + return; + } + + if ( + messageChannel.webhookSubscriptionStatus === + WebhookSubscriptionStatus.ACTIVE + ) { + return; + } + + const clientState = messageChannel.webhookSubscriptionClientState ?? v4(); + const driver = this.webhookSubscriptionDriverFactory.getDriver( + connectedAccount.provider, + ); + + if (isDefined(messageChannel.webhookSubscriptionExternalId)) { + await driver + .deleteSubscription(this.toContext(messageChannel)) + .catch(() => undefined); + } + + try { + const result = await driver.createSubscription( + messageChannel.connectedAccountId, + WebhookSubscriptionChannelType.MESSAGING, + clientState, + ); + + await this.messageChannelRepository.update(messageChannel.id, { + webhookSubscriptionExternalId: result.externalSubscriptionId, + webhookSubscriptionClientState: clientState, + webhookSubscriptionStatus: WebhookSubscriptionStatus.ACTIVE, + webhookSubscriptionExpiresAt: result.expiresAt, + }); + } catch (error) { + await this.messageChannelRepository.update(messageChannel.id, { + webhookSubscriptionClientState: clientState, + webhookSubscriptionStatus: WebhookSubscriptionStatus.FAILED, + webhookSubscriptionExpiresAt: null, + }); + + this.exceptionHandlerService.captureExceptions([error], { + workspace: { id: workspaceId }, + }); + } + } + + async renewSubscription(messageChannel: MessageChannelEntity): Promise { + const connectedAccount = await this.connectedAccountRepository.findOne({ + where: { + id: messageChannel.connectedAccountId, + workspaceId: messageChannel.workspaceId, + }, + }); + + if (!isDefined(connectedAccount)) { + return; + } + + const driver = this.webhookSubscriptionDriverFactory.getDriver( + connectedAccount.provider, + ); + + try { + const result = await driver.renewSubscription( + this.toContext(messageChannel), + ); + + await this.messageChannelRepository.update(messageChannel.id, { + webhookSubscriptionExternalId: result.externalSubscriptionId, + webhookSubscriptionStatus: WebhookSubscriptionStatus.ACTIVE, + webhookSubscriptionExpiresAt: result.expiresAt, + }); + } catch (error) { + await this.messageChannelRepository.update(messageChannel.id, { + webhookSubscriptionStatus: WebhookSubscriptionStatus.FAILED, + }); + + this.exceptionHandlerService.captureExceptions([error], { + workspace: { id: messageChannel.workspaceId }, + }); + } + } + + async deleteSubscription( + messageChannelId: string, + workspaceId: string, + ): Promise { + const messageChannel = await this.messageChannelRepository.findOne({ + where: { id: messageChannelId, workspaceId }, + }); + + if (!isDefined(messageChannel)) { + return; + } + + const connectedAccount = await this.connectedAccountRepository.findOne({ + where: { + id: messageChannel.connectedAccountId, + workspaceId: messageChannel.workspaceId, + }, + }); + + if (!isDefined(connectedAccount)) { + return; + } + + const driver = this.webhookSubscriptionDriverFactory.getDriver( + connectedAccount.provider, + ); + + try { + await driver.deleteSubscription(this.toContext(messageChannel)); + } catch (error) { + this.exceptionHandlerService.captureExceptions([error], { + workspace: { id: messageChannel.workspaceId }, + }); + } + } + + private toContext( + messageChannel: MessageChannelEntity, + ): WebhookSubscriptionContext { + return { + connectedAccountId: messageChannel.connectedAccountId, + channelType: WebhookSubscriptionChannelType.MESSAGING, + externalSubscriptionId: messageChannel.webhookSubscriptionExternalId, + externalResourceId: null, + clientState: messageChannel.webhookSubscriptionClientState ?? '', + }; + } +} diff --git a/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/services/webhook-subscription-driver-factory.service.ts b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/services/webhook-subscription-driver-factory.service.ts new file mode 100644 index 0000000000..87e933e91d --- /dev/null +++ b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/services/webhook-subscription-driver-factory.service.ts @@ -0,0 +1,46 @@ +import { Injectable } from '@nestjs/common'; + +import { ConnectedAccountProvider } from 'twenty-shared/types'; + +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'; +import { MicrosoftWebhookSubscriptionDriver } from 'src/modules/connected-account/webhook-subscription-manager/drivers/microsoft/microsoft-webhook-subscription.driver'; +import { type WebhookSubscriptionDriver } from 'src/modules/connected-account/webhook-subscription-manager/types/webhook-subscription-driver.type'; + +@Injectable() +export class WebhookSubscriptionDriverFactory { + private readonly driversByProvider: Partial< + Record + >; + + constructor( + private readonly googleWebhookSubscriptionDriver: GoogleWebhookSubscriptionDriver, + private readonly microsoftWebhookSubscriptionDriver: MicrosoftWebhookSubscriptionDriver, + ) { + this.driversByProvider = { + [ConnectedAccountProvider.GOOGLE]: this.googleWebhookSubscriptionDriver, + [ConnectedAccountProvider.MICROSOFT]: + this.microsoftWebhookSubscriptionDriver, + }; + } + + isProviderSupported(provider: ConnectedAccountProvider): boolean { + return provider in this.driversByProvider; + } + + getDriver(provider: ConnectedAccountProvider): WebhookSubscriptionDriver { + const driver = this.driversByProvider[provider]; + + if (!driver) { + throw new WebhookSubscriptionDriverException( + `Webhook subscriptions are not supported for provider ${provider}`, + WebhookSubscriptionDriverExceptionCode.UNSUPPORTED_PROVIDER, + ); + } + + return driver; + } +} diff --git a/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/services/webhook-sync-trigger.service.ts b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/services/webhook-sync-trigger.service.ts new file mode 100644 index 0000000000..625a1063e5 --- /dev/null +++ b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/services/webhook-sync-trigger.service.ts @@ -0,0 +1,128 @@ +import { Injectable } from '@nestjs/common'; +import { InjectRepository } from '@nestjs/typeorm'; + +import { + CalendarChannelSyncStage, + MessageChannelSyncStage, +} from 'twenty-shared/types'; +import { Repository } from 'typeorm'; + +import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator'; +import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants'; +import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service'; +import { CalendarChannelEntity } from 'src/engine/metadata-modules/calendar-channel/entities/calendar-channel.entity'; +import { MessageChannelEntity } from 'src/engine/metadata-modules/message-channel/entities/message-channel.entity'; +import { + CalendarEventListFetchJob, + type CalendarEventListFetchJobData, +} from 'src/modules/calendar/calendar-event-import-manager/jobs/calendar-event-list-fetch.job'; +import { + MessagingMessageListFetchJob, + type MessagingMessageListFetchJobData, +} from 'src/modules/messaging/message-import-manager/jobs/messaging-message-list-fetch.job'; + +@Injectable() +export class WebhookSyncTriggerService { + constructor( + @InjectMessageQueue(MessageQueue.messagingQueue) + private readonly messagingQueueService: MessageQueueService, + @InjectMessageQueue(MessageQueue.calendarQueue) + private readonly calendarQueueService: MessageQueueService, + @InjectRepository(MessageChannelEntity) + private readonly messageChannelRepository: Repository, + @InjectRepository(CalendarChannelEntity) + private readonly calendarChannelRepository: Repository, + ) {} + + async triggerMessagingSync( + messageChannelId: string, + workspaceId: string, + ): Promise { + const updateResult = await this.messageChannelRepository + .createQueryBuilder() + .update() + .set({ + syncStage: MessageChannelSyncStage.MESSAGE_LIST_FETCH_SCHEDULED, + syncStageStartedAt: new Date(), + }) + .where({ + id: messageChannelId, + workspaceId, + isSyncEnabled: true, + syncStage: MessageChannelSyncStage.MESSAGE_LIST_FETCH_PENDING, + }) + .returning('id') + .execute(); + + if (updateResult.raw.length === 0) { + return; + } + + try { + await this.messagingQueueService.add( + MessagingMessageListFetchJob.name, + { workspaceId, messageChannelId }, + ); + } catch (error) { + await this.messageChannelRepository + .createQueryBuilder() + .update() + .set({ + syncStage: MessageChannelSyncStage.MESSAGE_LIST_FETCH_PENDING, + }) + .where({ + id: messageChannelId, + workspaceId, + }) + .execute(); + + throw error; + } + } + + async triggerCalendarSync( + calendarChannelId: string, + workspaceId: string, + ): Promise { + const updateResult = await this.calendarChannelRepository + .createQueryBuilder() + .update() + .set({ + syncStage: CalendarChannelSyncStage.CALENDAR_EVENT_LIST_FETCH_SCHEDULED, + syncStageStartedAt: new Date(), + }) + .where({ + id: calendarChannelId, + workspaceId, + isSyncEnabled: true, + syncStage: CalendarChannelSyncStage.CALENDAR_EVENT_LIST_FETCH_PENDING, + }) + .returning('id') + .execute(); + + if (updateResult.raw.length === 0) { + return; + } + + try { + await this.calendarQueueService.add( + CalendarEventListFetchJob.name, + { workspaceId, calendarChannelId }, + ); + } catch (error) { + await this.calendarChannelRepository + .createQueryBuilder() + .update() + .set({ + syncStage: CalendarChannelSyncStage.CALENDAR_EVENT_LIST_FETCH_PENDING, + }) + .where({ + id: calendarChannelId, + workspaceId, + }) + .execute(); + + throw error; + } + } +} 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 new file mode 100644 index 0000000000..2f05b0628b --- /dev/null +++ b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/types/webhook-subscription-driver.type.ts @@ -0,0 +1,29 @@ +import { type WebhookSubscriptionChannelType } from 'twenty-shared/types'; + +export type WebhookSubscriptionResult = { + externalSubscriptionId: string | null; + externalResourceId: string | null; + expiresAt: Date; +}; + +export type WebhookSubscriptionContext = { + connectedAccountId: string; + channelType: WebhookSubscriptionChannelType; + externalSubscriptionId: string | null; + externalResourceId: string | null; + clientState: string; +}; + +export type WebhookSubscriptionDriver = { + createSubscription( + connectedAccountId: string, + channelType: WebhookSubscriptionChannelType, + clientState: string, + ): Promise; + + renewSubscription( + context: WebhookSubscriptionContext, + ): Promise; + + deleteSubscription(context: WebhookSubscriptionContext): Promise; +}; diff --git a/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/webhook-subscription-manager.module.ts b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/webhook-subscription-manager.module.ts new file mode 100644 index 0000000000..c8a9102e46 --- /dev/null +++ b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/webhook-subscription-manager.module.ts @@ -0,0 +1,17 @@ +import { Module } from '@nestjs/common'; + +import { GoogleWebhookSubscriptionDriver } from 'src/modules/connected-account/webhook-subscription-manager/drivers/google/google-webhook-subscription.driver'; +import { MicrosoftWebhookSubscriptionDriver } from 'src/modules/connected-account/webhook-subscription-manager/drivers/microsoft/microsoft-webhook-subscription.driver'; +import { WebhookSubscriptionDriverFactory } from 'src/modules/connected-account/webhook-subscription-manager/services/webhook-subscription-driver-factory.service'; +import { OAuth2ClientManagerModule } from 'src/modules/connected-account/oauth2-client-manager/oauth2-client-manager.module'; + +@Module({ + imports: [OAuth2ClientManagerModule], + providers: [ + GoogleWebhookSubscriptionDriver, + MicrosoftWebhookSubscriptionDriver, + WebhookSubscriptionDriverFactory, + ], + exports: [WebhookSubscriptionDriverFactory], +}) +export class WebhookSubscriptionManagerModule {} 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 new file mode 100644 index 0000000000..62a1c1eb8a --- /dev/null +++ b/packages/twenty-server/src/modules/connected-account/webhook-subscription-manager/webhook-subscription.module.ts @@ -0,0 +1,37 @@ +import { Module } from '@nestjs/common'; +import { TypeOrmModule } from '@nestjs/typeorm'; + +import { FeatureFlagModule } from 'src/engine/core-modules/feature-flag/feature-flag.module'; +import { ConnectedAccountEntity } from 'src/engine/metadata-modules/connected-account/entities/connected-account.entity'; +import { CalendarChannelEntity } from 'src/engine/metadata-modules/calendar-channel/entities/calendar-channel.entity'; +import { MessageChannelEntity } from 'src/engine/metadata-modules/message-channel/entities/message-channel.entity'; +import { WebhookSubscriptionRenewalCronCommand } from 'src/modules/connected-account/webhook-subscription-manager/crons/commands/webhook-subscription-renewal.cron.command'; +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'; +import { MessagingWebhookSubscriptionService } from 'src/modules/connected-account/webhook-subscription-manager/services/messaging-webhook-subscription.service'; +import { WebhookSubscriptionManagerModule } from 'src/modules/connected-account/webhook-subscription-manager/webhook-subscription-manager.module'; + +@Module({ + imports: [ + WebhookSubscriptionManagerModule, + FeatureFlagModule, + TypeOrmModule.forFeature([ + ConnectedAccountEntity, + MessageChannelEntity, + CalendarChannelEntity, + ]), + ], + providers: [ + MessagingWebhookSubscriptionService, + CalendarWebhookSubscriptionService, + WebhookSubscriptionChannelDeletedListener, + WebhookSubscriptionRenewalCronJob, + WebhookSubscriptionRenewalCronCommand, + ], + exports: [ + MessagingWebhookSubscriptionService, + CalendarWebhookSubscriptionService, + ], +}) +export class WebhookSubscriptionModule {} diff --git a/packages/twenty-server/src/modules/messaging/message-import-manager/drivers/gmail/services/gmail-get-messages.service.ts b/packages/twenty-server/src/modules/messaging/message-import-manager/drivers/gmail/services/gmail-get-messages.service.ts index ad3f8b3b42..ac79950f73 100644 --- a/packages/twenty-server/src/modules/messaging/message-import-manager/drivers/gmail/services/gmail-get-messages.service.ts +++ b/packages/twenty-server/src/modules/messaging/message-import-manager/drivers/gmail/services/gmail-get-messages.service.ts @@ -9,6 +9,7 @@ import { MessageFolderImportPolicy } from 'twenty-shared/types'; import { type MessageChannelEntity } from 'src/engine/metadata-modules/message-channel/entities/message-channel.entity'; import { GoogleOAuth2ClientProvider } from 'src/modules/connected-account/oauth2-client-manager/drivers/google/google-oauth2-client.provider'; import { type ConnectedAccountEntity } from 'src/engine/metadata-modules/connected-account/entities/connected-account.entity'; +import { MESSAGING_GMAIL_EXCLUDED_SYSTEM_LABELS } from 'src/modules/messaging/message-import-manager/drivers/gmail/constants/messaging-gmail-excluded-system-labels.constant'; import { GmailMessagesImportErrorHandler } from 'src/modules/messaging/message-import-manager/drivers/gmail/services/gmail-messages-import-error-handler.service'; import { filterGmailMessagesByFolderPolicy } from 'src/modules/messaging/message-import-manager/drivers/gmail/utils/filter-gmail-messages-by-folder-policy.util'; import { parseAndFormatGmailMessage } from 'src/modules/messaging/message-import-manager/drivers/gmail/utils/parse-and-format-gmail-message.util'; @@ -185,6 +186,12 @@ export class GmailGetMessagesService { connectedAccount, ); }) - .filter(isDefined); + .filter(isDefined) + .filter( + (message) => + !(message.labelIds ?? []).some((labelId) => + MESSAGING_GMAIL_EXCLUDED_SYSTEM_LABELS.includes(labelId), + ), + ); } } diff --git a/packages/twenty-shared/src/types/FeatureFlagKey.ts b/packages/twenty-shared/src/types/FeatureFlagKey.ts index 81c8bdbbb2..54d55a73bf 100644 --- a/packages/twenty-shared/src/types/FeatureFlagKey.ts +++ b/packages/twenty-shared/src/types/FeatureFlagKey.ts @@ -9,4 +9,5 @@ export enum FeatureFlagKey { IS_LOGIC_FUNCTION_PREBUILT_MODE_ENABLED = 'IS_LOGIC_FUNCTION_PREBUILT_MODE_ENABLED', IS_SETTINGS_DISCOVERY_HERO_ENABLED = 'IS_SETTINGS_DISCOVERY_HERO_ENABLED', IS_CALL_RECORDING_ENABLED = 'IS_CALL_RECORDING_ENABLED', + IS_MESSAGING_CALENDAR_WEBHOOK_ENABLED = 'IS_MESSAGING_CALENDAR_WEBHOOK_ENABLED', } diff --git a/packages/twenty-shared/src/types/WebhookSubscriptionChannelType.ts b/packages/twenty-shared/src/types/WebhookSubscriptionChannelType.ts new file mode 100644 index 0000000000..110f751cc4 --- /dev/null +++ b/packages/twenty-shared/src/types/WebhookSubscriptionChannelType.ts @@ -0,0 +1,4 @@ +export enum WebhookSubscriptionChannelType { + MESSAGING = 'messaging', + CALENDAR = 'calendar', +} diff --git a/packages/twenty-shared/src/types/WebhookSubscriptionStatus.ts b/packages/twenty-shared/src/types/WebhookSubscriptionStatus.ts new file mode 100644 index 0000000000..cb29965967 --- /dev/null +++ b/packages/twenty-shared/src/types/WebhookSubscriptionStatus.ts @@ -0,0 +1,6 @@ +export enum WebhookSubscriptionStatus { + PENDING = 'PENDING', + ACTIVE = 'ACTIVE', + FAILED = 'FAILED', + EXPIRED = 'EXPIRED', +} diff --git a/packages/twenty-shared/src/types/index.ts b/packages/twenty-shared/src/types/index.ts index 004d6e82da..cd1e3ab8a9 100644 --- a/packages/twenty-shared/src/types/index.ts +++ b/packages/twenty-shared/src/types/index.ts @@ -296,3 +296,5 @@ export { ViewOpenRecordIn } from './ViewOpenRecordIn'; export { ViewSortDirection } from './ViewSortDirection'; export { ViewType } from './ViewType'; export { ViewVisibility } from './ViewVisibility'; +export { WebhookSubscriptionChannelType } from './WebhookSubscriptionChannelType'; +export { WebhookSubscriptionStatus } from './WebhookSubscriptionStatus';