feat(messaging): webhook push sync for Gmail, Calendar and Microsoft (#21970)

<!-- This is an auto-generated description by cubic. -->
<a
href="https://cubic.dev/pr/twentyhq/twenty/pull/21970?utm_source=github"
target="_blank" rel="noopener noreferrer"
data-no-image-dialog="true"><picture><source
media="(prefers-color-scheme: dark)"
srcset="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"><source
media="(prefers-color-scheme: light)"
srcset="https://www.cubic.dev/buttons/review-in-cubic-light.svg"><img
alt="Review in cubic"
src="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"></picture></a>
<!-- End of auto-generated description by cubic. -->

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 {

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 = {

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',

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',

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<void> {
+    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<void> {
+    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"');
+  }
+}

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,
 ];

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,

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
     ? [
         // {

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/<project>/topics/<topic>). 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',

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;
 

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;
 

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: {

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<ConnectedAccountSyncWebhookExceptionCode> {
+  constructor(
+    message: string,
+    code: ConnectedAccountSyncWebhookExceptionCode,
+    { userFriendlyMessage }: { userFriendlyMessage?: MessageDescriptor } = {},
+  ) {
+    super(message, code, {
+      userFriendlyMessage:
+        userFriendlyMessage ??
+        getConnectedAccountSyncWebhookExceptionUserFriendlyMessage(code),
+    });
+  }
+}

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<void> {
+    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<void> {
+    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<string> {
+    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<string> {
+    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);
+  }
+}

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 {}

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<GoogleCalendarChannelNotification> {
+  constructor(
+    @InjectRepository(CalendarChannelEntity)
+    private readonly calendarChannelRepository: Repository<CalendarChannelEntity>,
+    private readonly webhookSyncTriggerService: WebhookSyncTriggerService,
+  ) {}
+
+  async handle(request: GoogleCalendarChannelNotification): Promise<void> {
+    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,
+    );
+  }
+}

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<GoogleMessagingNotificationRequest> {
+  constructor(
+    private readonly twentyConfigService: TwentyConfigService,
+    @InjectRepository(ConnectedAccountEntity)
+    private readonly connectedAccountRepository: Repository<ConnectedAccountEntity>,
+    @InjectRepository(MessageChannelEntity)
+    private readonly messageChannelRepository: Repository<MessageChannelEntity>,
+    private readonly webhookSyncTriggerService: WebhookSyncTriggerService,
+  ) {}
+
+  async handle(request: GoogleMessagingNotificationRequest): Promise<void> {
+    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<void> {
+    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;
+  }
+}

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 {}

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<CalendarChannelEntity>,
+    private readonly calendarWebhookSubscriptionService: CalendarWebhookSubscriptionService,
+    private readonly webhookSyncTriggerService: WebhookSyncTriggerService,
+  ) {}
+
+  async handle(notifications: MicrosoftGraphNotification[]): Promise<void> {
+    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,
+      );
+    }
+  }
+}

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<MessageChannelEntity>,
+    private readonly messagingWebhookSubscriptionService: MessagingWebhookSubscriptionService,
+    private readonly webhookSyncTriggerService: WebhookSyncTriggerService,
+  ) {}
+
+  async handle(notifications: MicrosoftGraphNotification[]): Promise<void> {
+    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,
+      );
+    }
+  }
+}

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 {}

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<Response>();
+
+    return this.httpExceptionHandlerService.handleError(
+      exception,
+      response,
+      getConnectedAccountSyncWebhookExceptionStatusCode(exception),
+    );
+  }
+}

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;
+};

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;
+};

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[];
+};

packages/twenty-server/src/modules/connected-account-sync-webhooks/types/webhook-notification-handler.type.ts

@@ -0,0 +1,3 @@
+export type WebhookNotificationHandler<TRequest> = {
+  handle(request: TRequest): Promise<void>;
+};

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);
+    }
+  }
+};

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],

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<CalendarChannelEntity>,
+    private readonly messagingWebhookSubscriptionService: MessagingWebhookSubscriptionService,
+    private readonly calendarWebhookSubscriptionService: CalendarWebhookSubscriptionService,
   ) {}
 
   async startChannelSync(input: StartChannelSyncInput): Promise<void> {
@@ -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);
   }

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;

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 * * * *';

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<void> {
+    await this.messageQueueService.addCron<undefined>({
+      jobName: WebhookSubscriptionRenewalCronJob.name,
+      data: undefined,
+      options: {
+        repeat: {
+          pattern: WEBHOOK_SUBSCRIPTION_RENEWAL_CRON_PATTERN,
+        },
+      },
+    });
+  }
+}

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<MessageChannelEntity>,
+    @InjectRepository(CalendarChannelEntity)
+    private readonly calendarChannelRepository: Repository<CalendarChannelEntity>,
+    private readonly messagingWebhookSubscriptionService: MessagingWebhookSubscriptionService,
+    private readonly calendarWebhookSubscriptionService: CalendarWebhookSubscriptionService,
+  ) {}
+
+  @Process(WebhookSubscriptionRenewalCronJob.name)
+  @SentryCronMonitor(
+    WebhookSubscriptionRenewalCronJob.name,
+    WEBHOOK_SUBSCRIPTION_RENEWAL_CRON_PATTERN,
+  )
+  async handle(): Promise<void> {
+    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,
+        );
+      }
+    }
+  }
+}

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<WebhookSubscriptionDriverExceptionCode> {
+  constructor(
+    message: string,
+    code: WebhookSubscriptionDriverExceptionCode,
+    { userFriendlyMessage }: { userFriendlyMessage?: MessageDescriptor } = {},
+  ) {
+    super(message, code, {
+      userFriendlyMessage:
+        userFriendlyMessage ??
+        getWebhookSubscriptionDriverExceptionUserFriendlyMessage(code),
+    });
+  }
+}

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;

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<WebhookSubscriptionResult> {
+    return channelType === WebhookSubscriptionChannelType.MESSAGING
+      ? this.watchGmailMailbox(connectedAccountId)
+      : this.watchPrimaryCalendar(connectedAccountId, clientState);
+  }
+
+  async renewSubscription(
+    context: WebhookSubscriptionContext,
+  ): Promise<WebhookSubscriptionResult> {
+    if (context.channelType === WebhookSubscriptionChannelType.CALENDAR) {
+      await this.deleteSubscription(context);
+    }
+
+    return this.createSubscription(
+      context.connectedAccountId,
+      context.channelType,
+      context.clientState,
+    );
+  }
+
+  async deleteSubscription(context: WebhookSubscriptionContext): Promise<void> {
+    return context.channelType === WebhookSubscriptionChannelType.MESSAGING
+      ? this.stopGmailMailboxWatch(context.connectedAccountId)
+      : this.stopCalendarWatch(context);
+  }
+
+  private async watchGmailMailbox(
+    connectedAccountId: string,
+  ): Promise<WebhookSubscriptionResult> {
+    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<void> {
+    const gmailClient = await this.getGmailClient(connectedAccountId);
+
+    await gmailClient.users.stop({ userId: 'me' });
+  }
+
+  private async watchPrimaryCalendar(
+    connectedAccountId: string,
+    clientState: string,
+  ): Promise<WebhookSubscriptionResult> {
+    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<void> {
+    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<gmail_v1.Gmail> {
+    const oAuth2Client =
+      await this.googleOAuth2ClientProvider.getClient(connectedAccountId);
+
+    return google.gmail({ version: 'v1', auth: oAuth2Client });
+  }
+
+  private async getCalendarClient(
+    connectedAccountId: string,
+  ): Promise<calendar_v3.Calendar> {
+    const oAuth2Client =
+      await this.googleOAuth2ClientProvider.getClient(connectedAccountId);
+
+    return google.calendar({ version: 'v3', auth: oAuth2Client });
+  }
+}

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;

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<WebhookSubscriptionResult> {
+    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<WebhookSubscriptionResult> {
+    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<void> {
+    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),
+    };
+  }
+}

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<MessageChannelDeletedEvent>,
+  ): Promise<void> {
+    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<CalendarChannelDeletedEvent>,
+  ): Promise<void> {
+    const { workspaceId } = batchEvent;
+
+    if (!isDefined(workspaceId)) {
+      return;
+    }
+
+    for (const event of batchEvent.events) {
+      await this.calendarWebhookSubscriptionService.deleteSubscription(
+        event.calendarChannelId,
+        workspaceId,
+      );
+    }
+  }
+}

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<ConnectedAccountEntity>,
+    @InjectRepository(CalendarChannelEntity)
+    private readonly calendarChannelRepository: Repository<CalendarChannelEntity>,
+    private readonly webhookSubscriptionDriverFactory: WebhookSubscriptionDriverFactory,
+    private readonly featureFlagService: FeatureFlagService,
+    private readonly exceptionHandlerService: ExceptionHandlerService,
+  ) {}
+
+  async createSubscription(
+    calendarChannelId: string,
+    workspaceId: string,
+  ): Promise<void> {
+    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<void> {
+    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<void> {
+    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 ?? '',
+    };
+  }
+}

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<ConnectedAccountEntity>,
+    @InjectRepository(MessageChannelEntity)
+    private readonly messageChannelRepository: Repository<MessageChannelEntity>,
+    private readonly webhookSubscriptionDriverFactory: WebhookSubscriptionDriverFactory,
+    private readonly featureFlagService: FeatureFlagService,
+    private readonly exceptionHandlerService: ExceptionHandlerService,
+  ) {}
+
+  async createSubscription(
+    messageChannelId: string,
+    workspaceId: string,
+  ): Promise<void> {
+    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<void> {
+    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<void> {
+    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 ?? '',
+    };
+  }
+}

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<ConnectedAccountProvider, WebhookSubscriptionDriver>
+  >;
+
+  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;
+  }
+}

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<MessageChannelEntity>,
+    @InjectRepository(CalendarChannelEntity)
+    private readonly calendarChannelRepository: Repository<CalendarChannelEntity>,
+  ) {}
+
+  async triggerMessagingSync(
+    messageChannelId: string,
+    workspaceId: string,
+  ): Promise<void> {
+    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<MessagingMessageListFetchJobData>(
+        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<void> {
+    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<CalendarEventListFetchJobData>(
+        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;
+    }
+  }
+}

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<WebhookSubscriptionResult>;
+
+  renewSubscription(
+    context: WebhookSubscriptionContext,
+  ): Promise<WebhookSubscriptionResult>;
+
+  deleteSubscription(context: WebhookSubscriptionContext): Promise<void>;
+};

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 {}

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 {}

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),
+          ),
+      );
   }
 }

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',
 }

packages/twenty-shared/src/types/WebhookSubscriptionChannelType.ts

@@ -0,0 +1,4 @@
+export enum WebhookSubscriptionChannelType {
+  MESSAGING = 'messaging',
+  CALENDAR = 'calendar',
+}

packages/twenty-shared/src/types/WebhookSubscriptionStatus.ts

@@ -0,0 +1,6 @@
+export enum WebhookSubscriptionStatus {
+  PENDING = 'PENDING',
+  ACTIVE = 'ACTIVE',
+  FAILED = 'FAILED',
+  EXPIRED = 'EXPIRED',
+}

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';