wip

packages/twenty-server/src/database/typeorm/core/migrations/common/1778500000000-addInboundWebhookSubscription.ts

@@ -0,0 +1,74 @@
+import { type MigrationInterface, type QueryRunner } from 'typeorm';
+
+export class AddInboundWebhookSubscription1778500000000
+  implements MigrationInterface
+{
+  name = 'AddInboundWebhookSubscription1778500000000';
+
+  public async up(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query(
+      `CREATE TABLE "core"."inboundWebhookSubscription" (
+        "id" uuid NOT NULL DEFAULT uuid_generate_v4(),
+        "workspaceId" uuid NOT NULL,
+        "source" varchar NOT NULL,
+        "connectedAccountId" uuid,
+        "externalSubscriptionId" varchar,
+        "externalResourceId" varchar,
+        "secret" varchar NOT NULL,
+        "expiresAt" TIMESTAMP WITH TIME ZONE,
+        "lastNotificationAt" TIMESTAMP WITH TIME ZONE,
+        "metadata" jsonb,
+        "createdAt" TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT now(),
+        "updatedAt" TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT now(),
+        CONSTRAINT "PK_inboundWebhookSubscription_id" PRIMARY KEY ("id")
+      )`,
+    );
+
+    await queryRunner.query(
+      `CREATE INDEX "IDX_INBOUND_WEBHOOK_SUBSCRIPTION_WORKSPACE_ID" ON "core"."inboundWebhookSubscription" ("workspaceId")`,
+    );
+
+    await queryRunner.query(
+      `CREATE INDEX "IDX_INBOUND_WEBHOOK_SUBSCRIPTION_SOURCE" ON "core"."inboundWebhookSubscription" ("source")`,
+    );
+
+    await queryRunner.query(
+      `CREATE INDEX "IDX_INBOUND_WEBHOOK_SUBSCRIPTION_EXTERNAL_ID" ON "core"."inboundWebhookSubscription" ("externalSubscriptionId")`,
+    );
+
+    await queryRunner.query(
+      `CREATE INDEX "IDX_INBOUND_WEBHOOK_SUBSCRIPTION_EXPIRES_AT" ON "core"."inboundWebhookSubscription" ("expiresAt")`,
+    );
+
+    await queryRunner.query(
+      `ALTER TABLE "core"."inboundWebhookSubscription"
+       ADD CONSTRAINT "FK_inboundWebhookSubscription_workspaceId"
+       FOREIGN KEY ("workspaceId") REFERENCES "core"."workspace"("id")
+       ON DELETE CASCADE ON UPDATE NO ACTION`,
+    );
+  }
+
+  public async down(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query(
+      `ALTER TABLE "core"."inboundWebhookSubscription" DROP CONSTRAINT "FK_inboundWebhookSubscription_workspaceId"`,
+    );
+
+    await queryRunner.query(
+      `DROP INDEX "core"."IDX_INBOUND_WEBHOOK_SUBSCRIPTION_EXPIRES_AT"`,
+    );
+
+    await queryRunner.query(
+      `DROP INDEX "core"."IDX_INBOUND_WEBHOOK_SUBSCRIPTION_EXTERNAL_ID"`,
+    );
+
+    await queryRunner.query(
+      `DROP INDEX "core"."IDX_INBOUND_WEBHOOK_SUBSCRIPTION_SOURCE"`,
+    );
+
+    await queryRunner.query(
+      `DROP INDEX "core"."IDX_INBOUND_WEBHOOK_SUBSCRIPTION_WORKSPACE_ID"`,
+    );
+
+    await queryRunner.query(`DROP TABLE "core"."inboundWebhookSubscription"`);
+  }
+}

packages/twenty-server/src/engine/core-modules/core-engine.module.ts

@@ -40,6 +40,7 @@ import { GeoMapModule } from 'src/engine/core-modules/geo-map/geo-map-module';
 import { HealthModule } from 'src/engine/core-modules/health/health.module';
 import { ImapSmtpCaldavModule } from 'src/engine/core-modules/imap-smtp-caldav-connection/imap-smtp-caldav-connection.module';
 import { ImpersonationModule } from 'src/engine/core-modules/impersonation/impersonation.module';
+import { InboundWebhookModule } from 'src/engine/core-modules/inbound-webhook/inbound-webhook.module';
 import { LabModule } from 'src/engine/core-modules/lab/lab.module';
 import { LoggerModule } from 'src/engine/core-modules/logger/logger.module';
 import { loggerModuleFactory } from 'src/engine/core-modules/logger/logger.module-factory';
@@ -89,6 +90,7 @@ import { FileModule } from './file/file.module';
     AuthModule,
     BillingModule,
     BillingWebhookModule,
+    InboundWebhookModule,
     UsageModule,
     ClientConfigModule,
     FeatureFlagModule,

packages/twenty-server/src/engine/core-modules/inbound-webhook/controllers/inbound-webhook.controller.ts

@@ -0,0 +1,89 @@
+import {
+  Controller,
+  Logger,
+  Param,
+  Post,
+  type RawBodyRequest,
+  Req,
+  Res,
+  UseGuards,
+} from '@nestjs/common';
+
+import { type Request, type Response } from 'express';
+
+import {
+  InboundWebhookException,
+  InboundWebhookExceptionCode,
+} from 'src/engine/core-modules/inbound-webhook/inbound-webhook.exception';
+import { InboundWebhookDispatcherService } from 'src/engine/core-modules/inbound-webhook/services/inbound-webhook-dispatcher.service';
+import { InboundWebhookIdempotencyService } from 'src/engine/core-modules/inbound-webhook/services/inbound-webhook-idempotency.service';
+import { ProcessInboundWebhookJob } from 'src/engine/core-modules/inbound-webhook/jobs/process-inbound-webhook.job';
+import { type InboundWebhookJobData } from 'src/engine/core-modules/inbound-webhook/dtos/inbound-webhook-job-data.type';
+import { type InboundWebhookSource } from 'src/engine/core-modules/inbound-webhook/types/inbound-webhook-source.type';
+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 { NoPermissionGuard } from 'src/engine/guards/no-permission.guard';
+import { PublicEndpointGuard } from 'src/engine/guards/public-endpoint.guard';
+
+@Controller()
+export class InboundWebhookController {
+  private readonly logger = new Logger(InboundWebhookController.name);
+
+  constructor(
+    private readonly dispatcher: InboundWebhookDispatcherService,
+    private readonly idempotencyService: InboundWebhookIdempotencyService,
+    @InjectMessageQueue(MessageQueue.inboundWebhookQueue)
+    private readonly inboundWebhookQueue: MessageQueueService,
+  ) {}
+
+  @Post(['inbound-webhooks/:source'])
+  @UseGuards(PublicEndpointGuard, NoPermissionGuard)
+  async receive(
+    @Param('source') sourceParam: string,
+    @Req() request: RawBodyRequest<Request>,
+    @Res() response: Response,
+  ): Promise<void> {
+    if (request.rawBody === undefined) {
+      throw new InboundWebhookException(
+        'Inbound webhook missing raw body',
+        InboundWebhookExceptionCode.MISSING_RAW_BODY,
+      );
+    }
+
+    const source: InboundWebhookSource = sourceParam;
+    const handler = this.dispatcher.resolve(source);
+
+    if (!(await handler.verify(request))) {
+      throw new InboundWebhookException(
+        `Inbound webhook signature verification failed for ${source}`,
+        InboundWebhookExceptionCode.INVALID_SIGNATURE,
+      );
+    }
+
+    const envelope = await handler.buildEnvelope(request);
+
+    const claimed = await this.idempotencyService.claim({
+      source,
+      externalEventId: envelope.externalEventId,
+    });
+
+    if (!claimed) {
+      this.logger.log(
+        `Dropping duplicate inbound webhook ${source}:${envelope.externalEventId}`,
+      );
+      response.status(200).json({ ok: true, duplicate: true });
+
+      return;
+    }
+
+    const jobData: InboundWebhookJobData = { envelope };
+
+    await this.inboundWebhookQueue.add<InboundWebhookJobData>(
+      ProcessInboundWebhookJob.name,
+      jobData,
+    );
+
+    response.status(200).json({ ok: true });
+  }
+}

packages/twenty-server/src/engine/core-modules/inbound-webhook/crons/commands/inbound-webhook-renewal.cron.command.ts

@@ -0,0 +1,32 @@
+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 { INBOUND_WEBHOOK_RENEWAL_CRON_PATTERN } from 'src/engine/core-modules/inbound-webhook/inbound-webhook.constants';
+import { InboundWebhookRenewalCronJob } from 'src/engine/core-modules/inbound-webhook/crons/jobs/inbound-webhook-renewal.cron.job';
+
+@Command({
+  name: 'cron:inbound-webhook:renewal',
+  description:
+    'Starts a cron job to renew inbound webhook subscriptions before they expire',
+})
+export class InboundWebhookRenewalCronCommand extends CommandRunner {
+  constructor(
+    @InjectMessageQueue(MessageQueue.cronQueue)
+    private readonly messageQueueService: MessageQueueService,
+  ) {
+    super();
+  }
+
+  async run(): Promise<void> {
+    await this.messageQueueService.addCron<undefined>({
+      jobName: InboundWebhookRenewalCronJob.name,
+      data: undefined,
+      options: {
+        repeat: { pattern: INBOUND_WEBHOOK_RENEWAL_CRON_PATTERN },
+      },
+    });
+  }
+}

packages/twenty-server/src/engine/core-modules/inbound-webhook/crons/jobs/inbound-webhook-renewal.cron.job.ts

@@ -0,0 +1,43 @@
+import { Logger } from '@nestjs/common';
+
+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 { INBOUND_WEBHOOK_RENEWAL_BUFFER_MS } from 'src/engine/core-modules/inbound-webhook/inbound-webhook.constants';
+import { InboundWebhookDispatcherService } from 'src/engine/core-modules/inbound-webhook/services/inbound-webhook-dispatcher.service';
+import { InboundWebhookSubscriptionService } from 'src/engine/core-modules/inbound-webhook/services/inbound-webhook-subscription.service';
+
+@Processor(MessageQueue.cronQueue)
+export class InboundWebhookRenewalCronJob {
+  private readonly logger = new Logger(InboundWebhookRenewalCronJob.name);
+
+  constructor(
+    private readonly subscriptionService: InboundWebhookSubscriptionService,
+    private readonly dispatcher: InboundWebhookDispatcherService,
+  ) {}
+
+  @Process(InboundWebhookRenewalCronJob.name)
+  async handle(): Promise<void> {
+    const threshold = new Date(Date.now() + INBOUND_WEBHOOK_RENEWAL_BUFFER_MS);
+    const expiring =
+      await this.subscriptionService.findExpiringBefore(threshold);
+
+    for (const subscription of expiring) {
+      const handler = this.dispatcher.resolveSubscribable(subscription.source);
+
+      if (handler === null) {
+        continue;
+      }
+
+      try {
+        await handler.renewSubscription(subscription);
+      } catch (error) {
+        this.logger.error(
+          `Failed to renew inbound webhook subscription ${subscription.id}`,
+          error instanceof Error ? error.stack : error,
+        );
+      }
+    }
+  }
+}

packages/twenty-server/src/engine/core-modules/inbound-webhook/dtos/inbound-webhook-job-data.type.ts

@@ -0,0 +1,5 @@
+import { type InboundWebhookEnvelope } from 'src/engine/core-modules/inbound-webhook/types/inbound-webhook-context.type';
+
+export type InboundWebhookJobData = {
+  envelope: InboundWebhookEnvelope;
+};

packages/twenty-server/src/engine/core-modules/inbound-webhook/entities/inbound-webhook-subscription.entity.ts

@@ -0,0 +1,55 @@
+import {
+  Column,
+  CreateDateColumn,
+  Entity,
+  Index,
+  PrimaryGeneratedColumn,
+  UpdateDateColumn,
+} from 'typeorm';
+
+import { type InboundWebhookSource } from 'src/engine/core-modules/inbound-webhook/types/inbound-webhook-source.type';
+
+@Index('IDX_INBOUND_WEBHOOK_SUBSCRIPTION_WORKSPACE_ID', ['workspaceId'])
+@Index('IDX_INBOUND_WEBHOOK_SUBSCRIPTION_SOURCE', ['source'])
+@Index('IDX_INBOUND_WEBHOOK_SUBSCRIPTION_EXTERNAL_ID', [
+  'externalSubscriptionId',
+])
+@Index('IDX_INBOUND_WEBHOOK_SUBSCRIPTION_EXPIRES_AT', ['expiresAt'])
+@Entity({ name: 'inboundWebhookSubscription', schema: 'core' })
+export class InboundWebhookSubscriptionEntity {
+  @PrimaryGeneratedColumn('uuid')
+  id: string;
+
+  @Column({ type: 'uuid' })
+  workspaceId: string;
+
+  @Column({ type: 'varchar' })
+  source: InboundWebhookSource;
+
+  @Column({ type: 'uuid', nullable: true })
+  connectedAccountId: string | null;
+
+  @Column({ type: 'varchar', nullable: true })
+  externalSubscriptionId: string | null;
+
+  @Column({ type: 'varchar', nullable: true })
+  externalResourceId: string | null;
+
+  @Column({ type: 'varchar' })
+  secret: string;
+
+  @Column({ type: 'timestamptz', nullable: true })
+  expiresAt: Date | null;
+
+  @Column({ type: 'timestamptz', nullable: true })
+  lastNotificationAt: Date | null;
+
+  @Column({ type: 'jsonb', nullable: true })
+  metadata: Record<string, unknown> | null;
+
+  @CreateDateColumn({ type: 'timestamptz' })
+  createdAt: Date;
+
+  @UpdateDateColumn({ type: 'timestamptz' })
+  updatedAt: Date;
+}

packages/twenty-server/src/engine/core-modules/inbound-webhook/inbound-webhook.constants.ts

@@ -0,0 +1,14 @@
+// Default replay window for HMAC-signed sources. Inbound timestamps older
+// than this are rejected to defeat replay attacks.
+export const INBOUND_WEBHOOK_REPLAY_WINDOW_MS = 5 * 60 * 1000;
+
+// Idempotency lock TTL — duplicate inbound events within this window are
+// dropped. 7d covers retry policies of all current providers.
+export const INBOUND_WEBHOOK_IDEMPOTENCY_TTL_MS = 7 * 24 * 60 * 60 * 1000;
+
+// Renewal cron pattern. Runs hourly — well below shortest provider expiry
+// (Microsoft Graph: 3 days; Gmail watch: 7 days).
+export const INBOUND_WEBHOOK_RENEWAL_CRON_PATTERN = '0 * * * *';
+
+// Renewal buffer — refresh subscriptions that expire within this window.
+export const INBOUND_WEBHOOK_RENEWAL_BUFFER_MS = 6 * 60 * 60 * 1000;

packages/twenty-server/src/engine/core-modules/inbound-webhook/inbound-webhook.exception.ts

@@ -0,0 +1,49 @@
+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 InboundWebhookExceptionCode {
+  SOURCE_NOT_SUPPORTED = 'SOURCE_NOT_SUPPORTED',
+  INVALID_SIGNATURE = 'INVALID_SIGNATURE',
+  MISSING_RAW_BODY = 'MISSING_RAW_BODY',
+  SUBSCRIPTION_NOT_FOUND = 'SUBSCRIPTION_NOT_FOUND',
+  REPLAY_WINDOW_EXCEEDED = 'REPLAY_WINDOW_EXCEEDED',
+  DUPLICATE_EVENT = 'DUPLICATE_EVENT',
+}
+
+const getInboundWebhookExceptionUserFriendlyMessage = (
+  code: InboundWebhookExceptionCode,
+): MessageDescriptor => {
+  switch (code) {
+    case InboundWebhookExceptionCode.SOURCE_NOT_SUPPORTED:
+      return msg`Inbound webhook source is not supported.`;
+    case InboundWebhookExceptionCode.INVALID_SIGNATURE:
+      return msg`Inbound webhook signature is invalid.`;
+    case InboundWebhookExceptionCode.MISSING_RAW_BODY:
+      return msg`Inbound webhook request is missing a raw body.`;
+    case InboundWebhookExceptionCode.SUBSCRIPTION_NOT_FOUND:
+      return msg`Inbound webhook subscription was not found.`;
+    case InboundWebhookExceptionCode.REPLAY_WINDOW_EXCEEDED:
+      return msg`Inbound webhook timestamp is outside the accepted replay window.`;
+    case InboundWebhookExceptionCode.DUPLICATE_EVENT:
+      return msg`Inbound webhook event was already processed.`;
+    default:
+      assertUnreachable(code);
+  }
+};
+
+export class InboundWebhookException extends CustomException<InboundWebhookExceptionCode> {
+  constructor(
+    message: string,
+    code: InboundWebhookExceptionCode,
+    { userFriendlyMessage }: { userFriendlyMessage?: MessageDescriptor } = {},
+  ) {
+    super(message, code, {
+      userFriendlyMessage:
+        userFriendlyMessage ??
+        getInboundWebhookExceptionUserFriendlyMessage(code),
+    });
+  }
+}

packages/twenty-server/src/engine/core-modules/inbound-webhook/inbound-webhook.module.ts

@@ -0,0 +1,34 @@
+import { Module } from '@nestjs/common';
+import { TypeOrmModule } from '@nestjs/typeorm';
+
+import { InboundWebhookRenewalCronCommand } from 'src/engine/core-modules/inbound-webhook/crons/commands/inbound-webhook-renewal.cron.command';
+import { InboundWebhookRenewalCronJob } from 'src/engine/core-modules/inbound-webhook/crons/jobs/inbound-webhook-renewal.cron.job';
+import { InboundWebhookController } from 'src/engine/core-modules/inbound-webhook/controllers/inbound-webhook.controller';
+import { InboundWebhookSubscriptionEntity } from 'src/engine/core-modules/inbound-webhook/entities/inbound-webhook-subscription.entity';
+import { ProcessInboundWebhookJob } from 'src/engine/core-modules/inbound-webhook/jobs/process-inbound-webhook.job';
+import { InboundWebhookDispatcherService } from 'src/engine/core-modules/inbound-webhook/services/inbound-webhook-dispatcher.service';
+import { InboundWebhookIdempotencyService } from 'src/engine/core-modules/inbound-webhook/services/inbound-webhook-idempotency.service';
+import { InboundWebhookSubscriptionService } from 'src/engine/core-modules/inbound-webhook/services/inbound-webhook-subscription.service';
+import { InboundWebhookVerifyService } from 'src/engine/core-modules/inbound-webhook/services/inbound-webhook-verify.service';
+
+@Module({
+  imports: [
+    TypeOrmModule.forFeature([InboundWebhookSubscriptionEntity], 'core'),
+  ],
+  controllers: [InboundWebhookController],
+  providers: [
+    InboundWebhookDispatcherService,
+    InboundWebhookIdempotencyService,
+    InboundWebhookSubscriptionService,
+    InboundWebhookVerifyService,
+    ProcessInboundWebhookJob,
+    InboundWebhookRenewalCronJob,
+    InboundWebhookRenewalCronCommand,
+  ],
+  exports: [
+    InboundWebhookDispatcherService,
+    InboundWebhookSubscriptionService,
+    InboundWebhookVerifyService,
+  ],
+})
+export class InboundWebhookModule {}

packages/twenty-server/src/engine/core-modules/inbound-webhook/jobs/process-inbound-webhook.job.ts

@@ -0,0 +1,32 @@
+import { Logger } from '@nestjs/common';
+
+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 { InboundWebhookDispatcherService } from 'src/engine/core-modules/inbound-webhook/services/inbound-webhook-dispatcher.service';
+import { type InboundWebhookJobData } from 'src/engine/core-modules/inbound-webhook/dtos/inbound-webhook-job-data.type';
+
+@Processor(MessageQueue.inboundWebhookQueue)
+export class ProcessInboundWebhookJob {
+  private readonly logger = new Logger(ProcessInboundWebhookJob.name);
+
+  constructor(private readonly dispatcher: InboundWebhookDispatcherService) {}
+
+  @Process(ProcessInboundWebhookJob.name)
+  async handle(data: InboundWebhookJobData): Promise<void> {
+    const { envelope } = data;
+    const handler = this.dispatcher.resolve(envelope.source);
+
+    try {
+      await handler.handle(envelope);
+    } catch (error) {
+      this.logger.error(
+        `Failed to process inbound webhook ${envelope.source}:${envelope.externalEventId}`,
+        error instanceof Error ? error.stack : error,
+      );
+
+      throw error;
+    }
+  }
+}

packages/twenty-server/src/engine/core-modules/inbound-webhook/services/inbound-webhook-dispatcher.service.ts

@@ -0,0 +1,38 @@
+import { Injectable } from '@nestjs/common';
+
+import {
+  InboundWebhookException,
+  InboundWebhookExceptionCode,
+} from 'src/engine/core-modules/inbound-webhook/inbound-webhook.exception';
+import {
+  type InboundWebhookHandler,
+  type SubscribableInboundWebhookHandler,
+  isSubscribableInboundWebhookHandler,
+} from 'src/engine/core-modules/inbound-webhook/types/inbound-webhook-handler.type';
+import { type InboundWebhookSource } from 'src/engine/core-modules/inbound-webhook/types/inbound-webhook-source.type';
+
+// Switch dispatcher (mirrors MessagingGetMessageListService and the Stripe
+// webhook handler in billing-webhook). Concrete sources are added by each
+// consumer PR — driver migration, inbound email, etc. — by injecting their
+// handler class and adding a case below.
+@Injectable()
+export class InboundWebhookDispatcherService {
+  resolve(source: InboundWebhookSource): InboundWebhookHandler {
+    switch (source) {
+      // Per-source handlers registered in follow-up PRs.
+      default:
+        throw new InboundWebhookException(
+          `Inbound webhook source ${source} is not supported`,
+          InboundWebhookExceptionCode.SOURCE_NOT_SUPPORTED,
+        );
+    }
+  }
+
+  resolveSubscribable(
+    source: InboundWebhookSource,
+  ): SubscribableInboundWebhookHandler | null {
+    const handler = this.resolve(source);
+
+    return isSubscribableInboundWebhookHandler(handler) ? handler : null;
+  }
+}

packages/twenty-server/src/engine/core-modules/inbound-webhook/services/inbound-webhook-idempotency.service.ts

@@ -0,0 +1,41 @@
+import { Injectable } from '@nestjs/common';
+
+import { CacheStorageNamespace } from 'src/engine/core-modules/cache-storage/types/cache-storage-namespace.enum';
+import { CacheStorageService } from 'src/engine/core-modules/cache-storage/services/cache-storage.service';
+import { InjectCacheStorage } from 'src/engine/core-modules/cache-storage/decorators/cache-storage.decorator';
+import { INBOUND_WEBHOOK_IDEMPOTENCY_TTL_MS } from 'src/engine/core-modules/inbound-webhook/inbound-webhook.constants';
+import { type InboundWebhookSource } from 'src/engine/core-modules/inbound-webhook/types/inbound-webhook-source.type';
+
+@Injectable()
+export class InboundWebhookIdempotencyService {
+  constructor(
+    @InjectCacheStorage(CacheStorageNamespace.EngineLock)
+    private readonly cacheStorageService: CacheStorageService,
+  ) {}
+
+  // Returns true if the event was claimed (first time seen) — caller should
+  // process. Returns false if a prior call already claimed it within the
+  // TTL — caller should drop as duplicate.
+  async claim({
+    source,
+    externalEventId,
+  }: {
+    source: InboundWebhookSource;
+    externalEventId: string;
+  }): Promise<boolean> {
+    return this.cacheStorageService.acquireLock(
+      this.buildKey({ source, externalEventId }),
+      INBOUND_WEBHOOK_IDEMPOTENCY_TTL_MS,
+    );
+  }
+
+  private buildKey({
+    source,
+    externalEventId,
+  }: {
+    source: InboundWebhookSource;
+    externalEventId: string;
+  }): string {
+    return `inbound-webhook:idempotency:${source}:${externalEventId}`;
+  }
+}

packages/twenty-server/src/engine/core-modules/inbound-webhook/services/inbound-webhook-subscription.service.ts

@@ -0,0 +1,63 @@
+import { Injectable } from '@nestjs/common';
+import { InjectRepository } from '@nestjs/typeorm';
+
+import { LessThan, Repository } from 'typeorm';
+
+import { InboundWebhookSubscriptionEntity } from 'src/engine/core-modules/inbound-webhook/entities/inbound-webhook-subscription.entity';
+import { type InboundWebhookSource } from 'src/engine/core-modules/inbound-webhook/types/inbound-webhook-source.type';
+
+@Injectable()
+export class InboundWebhookSubscriptionService {
+  constructor(
+    @InjectRepository(InboundWebhookSubscriptionEntity, 'core')
+    private readonly subscriptionRepository: Repository<InboundWebhookSubscriptionEntity>,
+  ) {}
+
+  async findById(id: string): Promise<InboundWebhookSubscriptionEntity | null> {
+    return this.subscriptionRepository.findOne({ where: { id } });
+  }
+
+  async findByExternalSubscriptionId({
+    source,
+    externalSubscriptionId,
+  }: {
+    source: InboundWebhookSource;
+    externalSubscriptionId: string;
+  }): Promise<InboundWebhookSubscriptionEntity | null> {
+    return this.subscriptionRepository.findOne({
+      where: { source, externalSubscriptionId },
+    });
+  }
+
+  async findExpiringBefore(
+    threshold: Date,
+  ): Promise<InboundWebhookSubscriptionEntity[]> {
+    return this.subscriptionRepository.find({
+      where: { expiresAt: LessThan(threshold) },
+    });
+  }
+
+  async markNotified(id: string): Promise<void> {
+    await this.subscriptionRepository.update(id, {
+      lastNotificationAt: new Date(),
+    });
+  }
+
+  async updateExpiry({
+    id,
+    expiresAt,
+    externalSubscriptionId,
+    externalResourceId,
+  }: {
+    id: string;
+    expiresAt: Date | null;
+    externalSubscriptionId?: string | null;
+    externalResourceId?: string | null;
+  }): Promise<void> {
+    await this.subscriptionRepository.update(id, {
+      expiresAt,
+      ...(externalSubscriptionId !== undefined && { externalSubscriptionId }),
+      ...(externalResourceId !== undefined && { externalResourceId }),
+    });
+  }
+}

packages/twenty-server/src/engine/core-modules/inbound-webhook/services/inbound-webhook-verify.service.ts

@@ -0,0 +1,44 @@
+import { Injectable } from '@nestjs/common';
+
+import {
+  InboundWebhookException,
+  InboundWebhookExceptionCode,
+} from 'src/engine/core-modules/inbound-webhook/inbound-webhook.exception';
+import { INBOUND_WEBHOOK_REPLAY_WINDOW_MS } from 'src/engine/core-modules/inbound-webhook/inbound-webhook.constants';
+import {
+  type HmacAlgorithm,
+  verifyHmacSignature,
+} from 'src/engine/core-modules/inbound-webhook/utils/hmac-signature.util';
+
+@Injectable()
+export class InboundWebhookVerifyService {
+  verifyHmac({
+    algorithm,
+    secret,
+    payload,
+    expectedSignature,
+  }: {
+    algorithm: HmacAlgorithm;
+    secret: string;
+    payload: string | Buffer;
+    expectedSignature: string;
+  }): boolean {
+    return verifyHmacSignature({
+      algorithm,
+      secret,
+      payload,
+      expectedSignature,
+    });
+  }
+
+  assertWithinReplayWindow(timestampMs: number): void {
+    const drift = Math.abs(Date.now() - timestampMs);
+
+    if (drift > INBOUND_WEBHOOK_REPLAY_WINDOW_MS) {
+      throw new InboundWebhookException(
+        `Inbound webhook timestamp drift ${drift}ms exceeds replay window`,
+        InboundWebhookExceptionCode.REPLAY_WINDOW_EXCEEDED,
+      );
+    }
+  }
+}

packages/twenty-server/src/engine/core-modules/inbound-webhook/types/inbound-webhook-context.type.ts

@@ -0,0 +1,16 @@
+import { type RawBodyRequest } from '@nestjs/common';
+
+import { type Request } from 'express';
+
+import { type InboundWebhookSource } from 'src/engine/core-modules/inbound-webhook/types/inbound-webhook-source.type';
+
+export type InboundWebhookRequest = RawBodyRequest<Request>;
+
+export type InboundWebhookEnvelope = {
+  source: InboundWebhookSource;
+  externalEventId: string;
+  workspaceId: string | null;
+  subscriptionId: string | null;
+  payload: unknown;
+  headers: Record<string, string | string[] | undefined>;
+};

packages/twenty-server/src/engine/core-modules/inbound-webhook/types/inbound-webhook-handler.type.ts

@@ -0,0 +1,31 @@
+import {
+  type InboundWebhookEnvelope,
+  type InboundWebhookRequest,
+} from 'src/engine/core-modules/inbound-webhook/types/inbound-webhook-context.type';
+import { type InboundWebhookSubscriptionEntity } from 'src/engine/core-modules/inbound-webhook/entities/inbound-webhook-subscription.entity';
+
+// Per-source contract. verify + buildEnvelope run on the API pod (must stay
+// fast — only HMAC + lookups). handle runs on the worker pod after the job
+// is dequeued and is where heavy work (EML parsing, fetch job dispatch)
+// belongs.
+export type InboundWebhookHandler = {
+  verify(request: InboundWebhookRequest): Promise<boolean>;
+  buildEnvelope(
+    request: InboundWebhookRequest,
+  ): Promise<InboundWebhookEnvelope>;
+  handle(envelope: InboundWebhookEnvelope): Promise<void>;
+};
+
+// Optional capability for sources backed by an external push subscription
+// (Google watch, Microsoft Graph subscriptions). Drives the renewal cron.
+export type SubscribableInboundWebhookHandler = InboundWebhookHandler & {
+  renewSubscription(
+    subscription: InboundWebhookSubscriptionEntity,
+  ): Promise<void>;
+};
+
+export const isSubscribableInboundWebhookHandler = (
+  handler: InboundWebhookHandler,
+): handler is SubscribableInboundWebhookHandler =>
+  typeof (handler as SubscribableInboundWebhookHandler).renewSubscription ===
+  'function';

packages/twenty-server/src/engine/core-modules/inbound-webhook/types/inbound-webhook-source.type.ts

@@ -0,0 +1,4 @@
+// Sources are identified by an opaque string supplied as a URL path segment.
+// Concrete values are added by each consumer PR (driver migration, inbound
+// email, etc.) when they register an InboundWebhookHandler.
+export type InboundWebhookSource = string;

packages/twenty-server/src/engine/core-modules/inbound-webhook/utils/hmac-signature.util.ts

@@ -0,0 +1,41 @@
+import * as crypto from 'node:crypto';
+
+export type HmacAlgorithm = 'sha1' | 'sha256';
+
+export const computeHmac = ({
+  algorithm,
+  secret,
+  payload,
+}: {
+  algorithm: HmacAlgorithm;
+  secret: string;
+  payload: string | Buffer;
+}): string =>
+  crypto.createHmac(algorithm, secret).update(payload).digest('hex');
+
+export const timingSafeStringEqual = (a: string, b: string): boolean => {
+  const bufferA = Buffer.from(a);
+  const bufferB = Buffer.from(b);
+
+  if (bufferA.length !== bufferB.length) {
+    return false;
+  }
+
+  return crypto.timingSafeEqual(bufferA, bufferB);
+};
+
+export const verifyHmacSignature = ({
+  algorithm,
+  secret,
+  payload,
+  expectedSignature,
+}: {
+  algorithm: HmacAlgorithm;
+  secret: string;
+  payload: string | Buffer;
+  expectedSignature: string;
+}): boolean => {
+  const computed = computeHmac({ algorithm, secret, payload });
+
+  return timingSafeStringEqual(computed, expectedSignature);
+};

packages/twenty-server/src/engine/core-modules/message-queue/message-queue-priority.constant.ts

@@ -18,4 +18,5 @@ export const MESSAGE_QUEUE_PRIORITY = {
   [MessageQueue.cronQueue]: 7,
   [MessageQueue.aiQueue]: 5,
   [MessageQueue.aiStreamQueue]: 2,
+  [MessageQueue.inboundWebhookQueue]: 2,
 };

packages/twenty-server/src/engine/core-modules/message-queue/message-queue.constants.ts

@@ -20,4 +20,5 @@ export enum MessageQueue {
   triggerQueue = 'trigger-queue',
   aiQueue = 'ai-queue',
   aiStreamQueue = 'ai-stream-queue',
+  inboundWebhookQueue = 'inbound-webhook-queue',
 }