[WIP] Feat/marketing emails (#21173)

Marketing/campaign emails on top of the emailing-domain (SES) feature: send a broadcast to a hand-picked list, with per-customer-domain unsubscribe links and opt-out-only **unsubscribe topics**. ## Model Standard objects (workspace schema, flat-metadata): - `messageCampaign` — a campaign send (subject, body template, from address, status, list, optional unsubscribe topic). - `messageList` + `messageListMember` — the hand-picked audience (person ↔ list join). A campaign's recipients are its list's members; everyone is sendable unless suppressed. Core entities (`core` schema, workspace-scoped — readable by the public unsubscribe flow without a workspace context): - `unsubscribeTopic` — an opt-out-only category (name, description, visibility). There is no opt-in subscription state. - `messageSuppression` — the single consent store: a row with `unsubscribeTopicId` NULL is a global block; a row with an `unsubscribeTopicId` and reason `UNSUBSCRIBE` is a per-topic opt-out. Two partial unique indexes dedupe global vs per-topic rows (Postgres treats NULLs as distinct). - `emailingDomain` — the workspace's SES sending domain, auto-provisioned when an email channel is added (and cleaned up when its last channel is removed), with verification status + DNS records. Campaign messages reuse the existing `message` / `messageThread` / `messageParticipant` model — one outbound `message` per recipient with a `deliveryStatus` state machine. ## Sending - `sendMessageCampaign` resolves the audience **under the caller's permissions**, creates the campaign, and enqueues a single fan-out job (the request never materializes per-recipient rows or jobs). - The fan-out job materializes one QUEUED message per recipient (deterministic ids → idempotent re-runs, reconciles crash-orphaned rows) and fans out per-recipient send jobs carrying **only ids**. - Each send job renders per-recipient `{{variable}}` merge fields and sends via `EmailingDomainSenderService`, which applies suppression (global + per-topic) and the unsubscribe footer/headers. Suppressed recipients are recorded `SKIPPED`. - The campaign finalizes `SENT`, or `SENT_WITH_ERRORS` if any recipient terminally failed. - `previewMessageCampaignAudience` returns a pre-send breakdown (total / without-email / duplicate / globally-unsubscribed / topic-unsubscribed / sendable), shown as a hint under the composer pickers. ## Unsubscribe - Encrypted (AES-256-GCM) token carrying workspaceId, address, optional `unsubscribeTopicId`, `issuedAt`, and a `preview` flag. - One-click POST (RFC 8058) + `mailto:` — topic-scoped when the token carries a topic, global otherwise. - Preferences page: a checkbox per visible topic (checked = still receiving); submitting creates per-topic opt-outs for unchecked topics and lifts re-checked ones (UNSUBSCRIBE only — never `BOUNCE`/`COMPLAINT`, never a global block). - A **Preview** action in settings opens the live page via a preview-claim token; opt-out POSTs are no-ops for preview tokens, so previewing never mutates state. - SES webhooks: inbound unsubscribe + outbound bounce/complaint → suppression (race-safe against at-least-once delivery, with reason escalation that never downgrades). - Per-customer unsubscribe hostname (Cloudflare DNS); sends are gated on it being active, except in LOG/demo mode. ## Architecture Campaign orchestration, suppression, the sender, the unsubscribe controller, and the SES webhook handlers live in `src/modules/emailing` + `src/modules/messaging-webhooks` (the workspace-feature layer). `core-modules/emailing-domain` keeps the SES driver, domain provisioning, the `unsubscribeTopic` / `messageSuppression` core entities, and the unsubscribe token/hostname plumbing. Domain creation is validated (`CreateEmailingDomainInput` — domain-format regex, lowercased) before any value reaches SES or the unsubscribe hostname. ## Frontend - Campaign composer side panel (from / list / unsubscribe topic / subject / body) with a live audience-preview hint. - Email settings: email channels each showing their auto-provisioned sending domain in a single section (status + DNS records + a "Check verification" action), plus an **Unsubscribe Topics** section to create/manage topics and preview the recipient page. A demo-mode banner is shown when the LOG driver is active. --------- Co-authored-by: Félix Malfait <felix@twenty.com> Co-authored-by: Félix Malfait <felix.malfait@gmail.com>
2026-06-13 22:07:39 +05:30
parent 2c5da39dc5
commit 5d892bdfd0
238 changed files with 10433 additions and 3517 deletions
@@ -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 { MessagingWebhookException } from 'src/modules/messaging-webhooks/messaging-webhook.exception';
+import { getMessagingWebhookExceptionStatusCode } from 'src/modules/messaging-webhooks/utils/get-messaging-webhook-exception-status-code.util';
+
+@Catch(MessagingWebhookException)
+export class MessagingWebhookApiExceptionFilter implements ExceptionFilter {
+  constructor(
+    private readonly httpExceptionHandlerService: HttpExceptionHandlerService,
+  ) {}
+
+  catch(exception: MessagingWebhookException, host: ArgumentsHost) {
+    const ctx = host.switchToHttp();
+    const response = ctx.getResponse<Response>();
+
+    return this.httpExceptionHandlerService.handleError(
+      exception,
+      response,
+      getMessagingWebhookExceptionStatusCode(exception),
+    );
+  }
+}
@@ -0,0 +1,9 @@
+export enum MessagingWebhookExceptionCode {
+  MESSAGING_WEBHOOK_MISSING_REQUEST_BODY = 'MESSAGING_WEBHOOK_MISSING_REQUEST_BODY',
+  MESSAGING_WEBHOOK_INVALID_PAYLOAD = 'MESSAGING_WEBHOOK_INVALID_PAYLOAD',
+  MESSAGING_WEBHOOK_FORBIDDEN_TOPIC = 'MESSAGING_WEBHOOK_FORBIDDEN_TOPIC',
+  MESSAGING_WEBHOOK_INVALID_SIGNATURE = 'MESSAGING_WEBHOOK_INVALID_SIGNATURE',
+  MESSAGING_WEBHOOK_INVALID_SUBSCRIBE_URL = 'MESSAGING_WEBHOOK_INVALID_SUBSCRIBE_URL',
+  MESSAGING_WEBHOOK_SUBSCRIPTION_CONFIRMATION_FAILED = 'MESSAGING_WEBHOOK_SUBSCRIPTION_CONFIRMATION_FAILED',
+  MESSAGING_WEBHOOK_UNHANDLED_ERROR = 'MESSAGING_WEBHOOK_UNHANDLED_ERROR',
+}
@@ -0,0 +1,39 @@
+import { type MessageDescriptor } from '@lingui/core';
+import { msg } from '@lingui/core/macro';
+import { assertUnreachable } from 'twenty-shared/utils';
+
+import { MessagingWebhookExceptionCode } from 'src/modules/messaging-webhooks/messaging-webhook-exception-code.enum';
+import { CustomException } from 'src/utils/custom-exception';
+
+const getMessagingWebhookExceptionUserFriendlyMessage = (
+  code: MessagingWebhookExceptionCode,
+) => {
+  switch (code) {
+    case MessagingWebhookExceptionCode.MESSAGING_WEBHOOK_MISSING_REQUEST_BODY:
+    case MessagingWebhookExceptionCode.MESSAGING_WEBHOOK_INVALID_PAYLOAD:
+    case MessagingWebhookExceptionCode.MESSAGING_WEBHOOK_INVALID_SUBSCRIBE_URL:
+      return msg`The webhook request could not be processed.`;
+    case MessagingWebhookExceptionCode.MESSAGING_WEBHOOK_FORBIDDEN_TOPIC:
+    case MessagingWebhookExceptionCode.MESSAGING_WEBHOOK_INVALID_SIGNATURE:
+      return msg`The webhook request could not be authenticated.`;
+    case MessagingWebhookExceptionCode.MESSAGING_WEBHOOK_SUBSCRIPTION_CONFIRMATION_FAILED:
+    case MessagingWebhookExceptionCode.MESSAGING_WEBHOOK_UNHANDLED_ERROR:
+      return msg`An error occurred while processing the webhook.`;
+    default:
+      assertUnreachable(code);
+  }
+};
+
+export class MessagingWebhookException extends CustomException<MessagingWebhookExceptionCode> {
+  constructor(
+    message: string,
+    code: MessagingWebhookExceptionCode,
+    { userFriendlyMessage }: { userFriendlyMessage?: MessageDescriptor } = {},
+  ) {
+    super(message, code, {
+      userFriendlyMessage:
+        userFriendlyMessage ??
+        getMessagingWebhookExceptionUserFriendlyMessage(code),
+    });
+  }
+}
@@ -0,0 +1,61 @@
+import {
+  Controller,
+  HttpCode,
+  Post,
+  type RawBodyRequest,
+  Req,
+  UseFilters,
+  UseGuards,
+} from '@nestjs/common';
+
+import { type Request } from 'express';
+
+import { MessagingWebhookApiExceptionFilter } from 'src/modules/messaging-webhooks/filters/messaging-webhook-api-exception.filter';
+import { MessagingWebhookExceptionCode } from 'src/modules/messaging-webhooks/messaging-webhook-exception-code.enum';
+import { MessagingWebhookException } from 'src/modules/messaging-webhooks/messaging-webhook.exception';
+import { SesInboundWebhookRouterService } from 'src/modules/messaging-webhooks/services/ses-inbound-webhook-router.service';
+import { SesOutboundWebhookRouterService } from 'src/modules/messaging-webhooks/services/ses-outbound-webhook-router.service';
+import { NoPermissionGuard } from 'src/engine/guards/no-permission.guard';
+import { PublicEndpointGuard } from 'src/engine/guards/public-endpoint.guard';
+import { isDefined } from 'twenty-shared/utils';
+
+@Controller()
+@UseFilters(MessagingWebhookApiExceptionFilter)
+export class MessagingWebhooksController {
+  constructor(
+    private readonly sesInboundWebhookRouterService: SesInboundWebhookRouterService,
+    private readonly sesOutboundWebhookRouterService: SesOutboundWebhookRouterService,
+  ) {}
+
+  @Post(['webhooks/messaging/ses/inbound'])
+  @UseGuards(PublicEndpointGuard, NoPermissionGuard)
+  @HttpCode(200)
+  async handleSesInboundWebhook(
+    @Req() request: RawBodyRequest<Request>,
+  ): Promise<void> {
+    if (!isDefined(request.rawBody)) {
+      throw new MessagingWebhookException(
+        'Missing SNS payload',
+        MessagingWebhookExceptionCode.MESSAGING_WEBHOOK_MISSING_REQUEST_BODY,
+      );
+    }
+
+    await this.sesInboundWebhookRouterService.route(request.rawBody);
+  }
+
+  @Post(['webhooks/messaging/ses/outbound'])
+  @UseGuards(PublicEndpointGuard, NoPermissionGuard)
+  @HttpCode(200)
+  async handleSesOutboundWebhook(
+    @Req() request: RawBodyRequest<Request>,
+  ): Promise<void> {
+    if (!isDefined(request.rawBody)) {
+      throw new MessagingWebhookException(
+        'Missing SNS payload',
+        MessagingWebhookExceptionCode.MESSAGING_WEBHOOK_MISSING_REQUEST_BODY,
+      );
+    }
+
+    await this.sesOutboundWebhookRouterService.route(request.rawBody);
+  }
+}
@@ -0,0 +1,30 @@
+import { Module } from '@nestjs/common';
+
+import { EmailingDomainModule } from 'src/engine/core-modules/emailing-domain/emailing-domain.module';
+import { EmailingModule } from 'src/modules/emailing/emailing.module';
+import { MessagingWebhooksController } from 'src/modules/messaging-webhooks/messaging-webhooks.controller';
+import { SesInboundMailHandlerService } from 'src/modules/messaging-webhooks/services/ses-inbound-mail-handler.service';
+import { SesInboundUnsubscribeHandlerService } from 'src/modules/messaging-webhooks/services/ses-inbound-unsubscribe-handler.service';
+import { SesInboundWebhookRouterService } from 'src/modules/messaging-webhooks/services/ses-inbound-webhook-router.service';
+import { SesOutboundSendingStateHandlerService } from 'src/modules/messaging-webhooks/services/ses-outbound-sending-state-handler.service';
+import { SesOutboundSuppressionHandlerService } from 'src/modules/messaging-webhooks/services/ses-outbound-suppression-handler.service';
+import { SesOutboundWebhookRouterService } from 'src/modules/messaging-webhooks/services/ses-outbound-webhook-router.service';
+import { SnsSignatureVerifierService } from 'src/modules/messaging-webhooks/services/sns-signature-verifier.service';
+import { SnsSubscriptionConfirmerService } from 'src/modules/messaging-webhooks/services/sns-subscription-confirmer.service';
+import { TwentyConfigModule } from 'src/engine/core-modules/twenty-config/twenty-config.module';
+
+@Module({
+  imports: [TwentyConfigModule, EmailingDomainModule, EmailingModule],
+  controllers: [MessagingWebhooksController],
+  providers: [
+    SnsSignatureVerifierService,
+    SnsSubscriptionConfirmerService,
+    SesInboundMailHandlerService,
+    SesInboundUnsubscribeHandlerService,
+    SesOutboundSendingStateHandlerService,
+    SesOutboundSuppressionHandlerService,
+    SesInboundWebhookRouterService,
+    SesOutboundWebhookRouterService,
+  ],
+})
+export class MessagingWebhooksModule {}
@@ -0,0 +1,84 @@
+import { EmailingDomainTenantStatus } from 'src/engine/core-modules/emailing-domain/drivers/types/emailing-domain-tenant-status.type';
+import { type EmailingDomainTenantStatusService } from 'src/engine/core-modules/emailing-domain/services/emailing-domain-tenant-status.service';
+import { SesOutboundSendingStateHandlerService } from 'src/modules/messaging-webhooks/services/ses-outbound-sending-state-handler.service';
+import { type SesEventBridgeNotification } from 'src/modules/messaging-webhooks/types/ses-event-bridge-notification.type';
+
+describe('SesOutboundSendingStateHandlerService.handle', () => {
+  const setUp = () => {
+    const emailingDomainTenantStatusService = {
+      setTenantStatusForWorkspace: jest.fn().mockResolvedValue(undefined),
+    } as unknown as EmailingDomainTenantStatusService;
+    const service = new SesOutboundSendingStateHandlerService(
+      emailingDomainTenantStatusService,
+    );
+
+    return { service, emailingDomainTenantStatusService };
+  };
+
+  // SES emits `Sending Status Enabled|Disabled` against three resource scopes
+  // (tenant / configuration-set / identity). The handler must mirror the
+  // status onto the workspace's DB column regardless of which scope produced
+  // the event, since every twenty-managed resource shares the same prefix.
+  describe.each([
+    {
+      scope: 'tenant ARN with opaque tenant-id segment',
+      arn: 'arn:aws:ses:us-east-1:123456789012:tenant/twenty-workspace-ws1/9b1deb4d-3b7d-4bad-9bdd-2b0d7b3dcb6d',
+    },
+    {
+      scope: 'configuration-set ARN',
+      arn: 'arn:aws:ses:us-east-1:123456789012:configuration-set/twenty-workspace-ws1',
+    },
+    {
+      scope: 'identity ARN',
+      arn: 'arn:aws:ses:us-east-1:123456789012:identity/twenty-workspace-ws1',
+    },
+  ])('on a $scope', ({ arn }) => {
+    it.each([
+      {
+        detailType: 'Sending Status Disabled' as const,
+        expected: EmailingDomainTenantStatus.PAUSED,
+      },
+      {
+        detailType: 'Sending Status Enabled' as const,
+        expected: EmailingDomainTenantStatus.ACTIVE,
+      },
+    ])(
+      'mirrors "$detailType" to tenantStatus=$expected',
+      async ({ detailType, expected }) => {
+        const { service, emailingDomainTenantStatusService } = setUp();
+
+        const event: SesEventBridgeNotification = {
+          source: 'aws.ses',
+          'detail-type': detailType,
+          resources: [arn],
+        };
+
+        await service.handle(event);
+
+        expect(
+          emailingDomainTenantStatusService.setTenantStatusForWorkspace,
+        ).toHaveBeenCalledWith('ws1', expected);
+      },
+    );
+  });
+
+  // Security-critical: a foreign SES resource that lands on the shared SNS
+  // topic must not be allowed to flip our tenantStatus column. The
+  // workspaceId resolver returns null for any ARN that doesn't carry the
+  // twenty-managed name prefix; the handler must noop in that case.
+  it('does not update any workspace when the ARN does not carry the twenty-managed prefix', async () => {
+    const { service, emailingDomainTenantStatusService } = setUp();
+
+    await service.handle({
+      source: 'aws.ses',
+      'detail-type': 'Sending Status Disabled',
+      resources: [
+        'arn:aws:ses:us-east-1:123456789012:tenant/some-other-prefix/abc',
+      ],
+    });
+
+    expect(
+      emailingDomainTenantStatusService.setTenantStatusForWorkspace,
+    ).not.toHaveBeenCalled();
+  });
+});
@@ -0,0 +1,63 @@
+import { Injectable, Logger } from '@nestjs/common';
+
+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 { SesInboundUnsubscribeHandlerService } from 'src/modules/messaging-webhooks/services/ses-inbound-unsubscribe-handler.service';
+import { type SesInboundNotification } from 'src/modules/messaging-webhooks/types/sns-message.type';
+import { resolveInboundMailIntent } from 'src/modules/messaging-webhooks/utils/resolve-inbound-mail-intent.util';
+import {
+  MessagingInboundEmailImportJob,
+  type MessagingInboundEmailImportJobData,
+} from 'src/modules/messaging/message-import-manager/jobs/messaging-inbound-email-import.job';
+
+@Injectable()
+export class SesInboundMailHandlerService {
+  private readonly logger = new Logger(SesInboundMailHandlerService.name);
+
+  constructor(
+    @InjectMessageQueue(MessageQueue.messagingQueue)
+    private readonly messageQueueService: MessageQueueService,
+    private readonly sesInboundUnsubscribeHandlerService: SesInboundUnsubscribeHandlerService,
+  ) {}
+
+  async handle(
+    notification: SesInboundNotification,
+    snsMessageId: string,
+  ): Promise<void> {
+    switch (resolveInboundMailIntent(notification)) {
+      case 'UNSUBSCRIBE':
+        await this.sesInboundUnsubscribeHandlerService.handle(notification);
+
+        return;
+      case 'IMPORT':
+        await this.enqueueInboundEmailImport(notification, snsMessageId);
+
+        return;
+    }
+  }
+
+  private async enqueueInboundEmailImport(
+    notification: SesInboundNotification,
+    snsMessageId: string,
+  ): Promise<void> {
+    const { receipt } = notification;
+
+    if (receipt?.action?.type !== 'S3') {
+      this.logger.warn(
+        `SNS message ${snsMessageId} has unsupported action type ${receipt?.action?.type}`,
+      );
+
+      return;
+    }
+
+    await this.messageQueueService.add<MessagingInboundEmailImportJobData>(
+      MessagingInboundEmailImportJob.name,
+      {
+        s3Key: receipt.action.objectKey,
+        envelopeRecipients: receipt.recipients,
+      },
+      { id: snsMessageId },
+    );
+  }
+}
@@ -0,0 +1,52 @@
+import { Injectable, Logger } from '@nestjs/common';
+
+import { isNonEmptyString } from '@sniptt/guards';
+import { isDefined } from 'twenty-shared/utils';
+
+import { MessageSuppressionService } from 'src/modules/emailing/services/message-suppression.service';
+import { UnsubscribeTokenService } from 'src/engine/core-modules/emailing-domain/services/unsubscribe-token.service';
+import { MessageSuppressionReason } from 'src/engine/core-modules/emailing-domain/types/message-suppression-reason.type';
+import { MessageSuppressionSource } from 'src/engine/core-modules/emailing-domain/types/message-suppression-source.type';
+import { type SesInboundNotification } from 'src/modules/messaging-webhooks/types/sns-message.type';
+
+@Injectable()
+export class SesInboundUnsubscribeHandlerService {
+  private readonly logger = new Logger(
+    SesInboundUnsubscribeHandlerService.name,
+  );
+
+  constructor(
+    private readonly unsubscribeTokenService: UnsubscribeTokenService,
+    private readonly messageSuppressionService: MessageSuppressionService,
+  ) {}
+
+  async handle(notification: SesInboundNotification): Promise<void> {
+    const subject = notification.mail?.commonHeaders?.subject;
+
+    if (!isNonEmptyString(subject)) {
+      this.logger.warn('Unsubscribe email received without a token subject');
+
+      return;
+    }
+
+    const payload = this.unsubscribeTokenService.verify(subject.trim());
+
+    if (!isDefined(payload)) {
+      this.logger.warn('Unsubscribe email received with an invalid token');
+
+      return;
+    }
+
+    if (payload.preview === true) {
+      return;
+    }
+
+    await this.messageSuppressionService.suppress({
+      workspaceId: payload.workspaceId,
+      emailAddress: payload.emailAddress,
+      reason: MessageSuppressionReason.UNSUBSCRIBE,
+      source: MessageSuppressionSource.SYSTEM,
+      unsubscribeTopicId: payload.unsubscribeTopicId ?? null,
+    });
+  }
+}
@@ -0,0 +1,62 @@
+import { Injectable } from '@nestjs/common';
+
+import type SnsPayloadValidator from 'sns-payload-validator';
+import { isDefined, parseJson } from 'twenty-shared/utils';
+
+import { MessagingWebhookExceptionCode } from 'src/modules/messaging-webhooks/messaging-webhook-exception-code.enum';
+import { MessagingWebhookException } from 'src/modules/messaging-webhooks/messaging-webhook.exception';
+import { SesInboundMailHandlerService } from 'src/modules/messaging-webhooks/services/ses-inbound-mail-handler.service';
+import { SnsSignatureVerifierService } from 'src/modules/messaging-webhooks/services/sns-signature-verifier.service';
+import { SnsSubscriptionConfirmerService } from 'src/modules/messaging-webhooks/services/sns-subscription-confirmer.service';
+import { type SesInboundNotification } from 'src/modules/messaging-webhooks/types/sns-message.type';
+
+type SnsPayload = SnsPayloadValidator.SnsPayload;
+
+@Injectable()
+export class SesInboundWebhookRouterService {
+  constructor(
+    private readonly snsSignatureVerifierService: SnsSignatureVerifierService,
+    private readonly snsSubscriptionConfirmerService: SnsSubscriptionConfirmerService,
+    private readonly sesInboundMailHandlerService: SesInboundMailHandlerService,
+  ) {}
+
+  async route(rawBody: Buffer): Promise<void> {
+    const payload = parseJson<SnsPayload>(rawBody.toString('utf8'));
+
+    if (!isDefined(payload)) {
+      throw new MessagingWebhookException(
+        'Invalid SNS payload',
+        MessagingWebhookExceptionCode.MESSAGING_WEBHOOK_INVALID_PAYLOAD,
+      );
+    }
+
+    await this.snsSignatureVerifierService.assertAllowedAndSigned(payload);
+
+    if (
+      payload.Type === 'SubscriptionConfirmation' ||
+      payload.Type === 'UnsubscribeConfirmation'
+    ) {
+      await this.snsSubscriptionConfirmerService.confirm(payload.SubscribeURL);
+
+      return;
+    }
+
+    if (payload.Type !== 'Notification') {
+      return;
+    }
+
+    const notification = parseJson<SesInboundNotification>(payload.Message);
+
+    if (!isDefined(notification)) {
+      throw new MessagingWebhookException(
+        'Invalid SNS notification message',
+        MessagingWebhookExceptionCode.MESSAGING_WEBHOOK_INVALID_PAYLOAD,
+      );
+    }
+
+    await this.sesInboundMailHandlerService.handle(
+      notification,
+      payload.MessageId,
+    );
+  }
+}
@@ -0,0 +1,40 @@
+import { Injectable, Logger } from '@nestjs/common';
+
+import { EmailingDomainTenantStatus } from 'src/engine/core-modules/emailing-domain/drivers/types/emailing-domain-tenant-status.type';
+import { EmailingDomainTenantStatusService } from 'src/engine/core-modules/emailing-domain/services/emailing-domain-tenant-status.service';
+import { type SesEventBridgeNotification } from 'src/modules/messaging-webhooks/types/ses-event-bridge-notification.type';
+import { resolveWorkspaceIdFromAwsSesResources } from 'src/modules/messaging-webhooks/utils/resolve-workspace-id-from-aws-ses-resources.util';
+import { isDefined } from 'twenty-shared/utils';
+
+@Injectable()
+export class SesOutboundSendingStateHandlerService {
+  private readonly logger = new Logger(
+    SesOutboundSendingStateHandlerService.name,
+  );
+
+  constructor(
+    private readonly emailingDomainTenantStatusService: EmailingDomainTenantStatusService,
+  ) {}
+
+  async handle(event: SesEventBridgeNotification): Promise<void> {
+    const targetStatus =
+      event['detail-type'] === 'Sending Status Enabled'
+        ? EmailingDomainTenantStatus.ACTIVE
+        : EmailingDomainTenantStatus.PAUSED;
+
+    const workspaceId = resolveWorkspaceIdFromAwsSesResources(event.resources);
+
+    if (!isDefined(workspaceId)) {
+      this.logger.warn(
+        `Could not resolve workspaceId from SES sending-state event resources: ${JSON.stringify(event.resources)}`,
+      );
+
+      return;
+    }
+
+    await this.emailingDomainTenantStatusService.setTenantStatusForWorkspace(
+      workspaceId,
+      targetStatus,
+    );
+  }
+}
@@ -0,0 +1,129 @@
+import { Injectable, Logger } from '@nestjs/common';
+
+import { isDefined, isNonEmptyArray } from 'twenty-shared/utils';
+
+import { CAMPAIGN_MESSAGE_DELIVERY_STATUS } from 'src/engine/core-modules/emailing-domain/constants/campaign.constant';
+import { MessageCampaignService } from 'src/modules/emailing/services/message-campaign.service';
+import { MessageSuppressionService } from 'src/modules/emailing/services/message-suppression.service';
+import { MessageSuppressionReason } from 'src/engine/core-modules/emailing-domain/types/message-suppression-reason.type';
+import { MessageSuppressionSource } from 'src/engine/core-modules/emailing-domain/types/message-suppression-source.type';
+import { type SesEventBridgeNotification } from 'src/modules/messaging-webhooks/types/ses-event-bridge-notification.type';
+import { resolveWorkspaceIdFromAwsSesResources } from 'src/modules/messaging-webhooks/utils/resolve-workspace-id-from-aws-ses-resources.util';
+
+@Injectable()
+export class SesOutboundSuppressionHandlerService {
+  private readonly logger = new Logger(
+    SesOutboundSuppressionHandlerService.name,
+  );
+
+  constructor(
+    private readonly messageSuppressionService: MessageSuppressionService,
+    private readonly messageCampaignService: MessageCampaignService,
+  ) {}
+
+  async handle(event: SesEventBridgeNotification): Promise<void> {
+    const suppression = this.resolveSuppression(event);
+
+    if (!isDefined(suppression)) {
+      return;
+    }
+
+    const workspaceId = resolveWorkspaceIdFromAwsSesResources(event.resources);
+
+    if (!isDefined(workspaceId)) {
+      this.logger.warn(
+        `Could not resolve workspaceId from SES ${event['detail-type']} event resources: ${JSON.stringify(event.resources)}`,
+      );
+
+      return;
+    }
+
+    const providerMessageId = event.detail?.mail?.messageId;
+
+    if (isDefined(providerMessageId)) {
+      await this.messageCampaignService.recordDeliveryFailureByProviderMessageId(
+        {
+          workspaceId,
+          providerMessageId,
+          deliveryStatus:
+            event['detail-type'] === 'Email Complaint Received'
+              ? CAMPAIGN_MESSAGE_DELIVERY_STATUS.COMPLAINED
+              : CAMPAIGN_MESSAGE_DELIVERY_STATUS.BOUNCED,
+        },
+      );
+    }
+
+    const results = await Promise.allSettled(
+      suppression.emailAddresses.map((emailAddress) =>
+        this.messageSuppressionService.suppress({
+          workspaceId,
+          emailAddress,
+          reason: suppression.reason,
+          source: MessageSuppressionSource.WEBHOOK,
+          providerEventId: suppression.providerEventId,
+        }),
+      ),
+    );
+
+    if (results.some((result) => result.status === 'rejected')) {
+      throw new Error(
+        `Failed to suppress one or more recipients for ${event['detail-type']} event in workspace ${workspaceId}`,
+      );
+    }
+  }
+
+  private resolveSuppression(event: SesEventBridgeNotification): {
+    reason: MessageSuppressionReason;
+    emailAddresses: string[];
+    providerEventId: string | null;
+  } | null {
+    if (event['detail-type'] === 'Email Bounced') {
+      const bounce = event.detail?.bounce;
+
+      if (bounce?.bounceType !== 'Permanent') {
+        return null;
+      }
+
+      const emailAddresses = this.extractRecipientAddresses(
+        bounce.bouncedRecipients,
+      );
+
+      if (!isNonEmptyArray(emailAddresses)) {
+        return null;
+      }
+
+      return {
+        reason: MessageSuppressionReason.BOUNCE,
+        emailAddresses,
+        providerEventId: bounce.feedbackId ?? null,
+      };
+    }
+
+    if (event['detail-type'] === 'Email Complaint Received') {
+      const complaint = event.detail?.complaint;
+      const emailAddresses = this.extractRecipientAddresses(
+        complaint?.complainedRecipients,
+      );
+
+      if (!isNonEmptyArray(emailAddresses)) {
+        return null;
+      }
+
+      return {
+        reason: MessageSuppressionReason.COMPLAINT,
+        emailAddresses,
+        providerEventId: complaint?.feedbackId ?? null,
+      };
+    }
+
+    return null;
+  }
+
+  private extractRecipientAddresses = (
+    recipients: { emailAddress: string }[] | undefined,
+  ): string[] => {
+    return (recipients ?? [])
+      .map((recipient) => recipient.emailAddress)
+      .filter(isDefined);
+  };
+}
@@ -0,0 +1,70 @@
+import { Injectable } from '@nestjs/common';
+
+import type SnsPayloadValidator from 'sns-payload-validator';
+import { isDefined, parseJson } from 'twenty-shared/utils';
+
+import { MessagingWebhookExceptionCode } from 'src/modules/messaging-webhooks/messaging-webhook-exception-code.enum';
+import { MessagingWebhookException } from 'src/modules/messaging-webhooks/messaging-webhook.exception';
+import { SesOutboundSendingStateHandlerService } from 'src/modules/messaging-webhooks/services/ses-outbound-sending-state-handler.service';
+import { SesOutboundSuppressionHandlerService } from 'src/modules/messaging-webhooks/services/ses-outbound-suppression-handler.service';
+import { SnsSignatureVerifierService } from 'src/modules/messaging-webhooks/services/sns-signature-verifier.service';
+import { SnsSubscriptionConfirmerService } from 'src/modules/messaging-webhooks/services/sns-subscription-confirmer.service';
+import { type SesEventBridgeNotification } from 'src/modules/messaging-webhooks/types/ses-event-bridge-notification.type';
+
+type SnsPayload = SnsPayloadValidator.SnsPayload;
+
+@Injectable()
+export class SesOutboundWebhookRouterService {
+  constructor(
+    private readonly snsSignatureVerifierService: SnsSignatureVerifierService,
+    private readonly snsSubscriptionConfirmerService: SnsSubscriptionConfirmerService,
+    private readonly sesOutboundSendingStateHandlerService: SesOutboundSendingStateHandlerService,
+    private readonly sesOutboundSuppressionHandlerService: SesOutboundSuppressionHandlerService,
+  ) {}
+
+  async route(rawBody: Buffer): Promise<void> {
+    const payload = parseJson<SnsPayload>(rawBody.toString('utf8'));
+
+    if (!isDefined(payload)) {
+      throw new MessagingWebhookException(
+        'Invalid SNS payload',
+        MessagingWebhookExceptionCode.MESSAGING_WEBHOOK_INVALID_PAYLOAD,
+      );
+    }
+
+    await this.snsSignatureVerifierService.assertAllowedAndSigned(payload);
+
+    if (
+      payload.Type === 'SubscriptionConfirmation' ||
+      payload.Type === 'UnsubscribeConfirmation'
+    ) {
+      await this.snsSubscriptionConfirmerService.confirm(payload.SubscribeURL);
+
+      return;
+    }
+
+    if (payload.Type !== 'Notification') {
+      return;
+    }
+
+    const event = parseJson<SesEventBridgeNotification>(payload.Message);
+
+    if (!isDefined(event)) {
+      throw new MessagingWebhookException(
+        'Invalid SNS notification message',
+        MessagingWebhookExceptionCode.MESSAGING_WEBHOOK_INVALID_PAYLOAD,
+      );
+    }
+
+    if (
+      event['detail-type'] === 'Email Bounced' ||
+      event['detail-type'] === 'Email Complaint Received'
+    ) {
+      await this.sesOutboundSuppressionHandlerService.handle(event);
+
+      return;
+    }
+
+    await this.sesOutboundSendingStateHandlerService.handle(event);
+  }
+}
@@ -0,0 +1,58 @@
+import { Injectable, Logger } from '@nestjs/common';
+
+import SnsPayloadValidator from 'sns-payload-validator';
+
+import { MessagingWebhookExceptionCode } from 'src/modules/messaging-webhooks/messaging-webhook-exception-code.enum';
+import { MessagingWebhookException } from 'src/modules/messaging-webhooks/messaging-webhook.exception';
+import { TwentyConfigService } from 'src/engine/core-modules/twenty-config/twenty-config.service';
+
+type SnsPayload = SnsPayloadValidator.SnsPayload;
+
+@Injectable()
+export class SnsSignatureVerifierService {
+  private readonly logger = new Logger(SnsSignatureVerifierService.name);
+  private readonly validator = new SnsPayloadValidator();
+
+  constructor(private readonly twentyConfigService: TwentyConfigService) {}
+
+  async assertAllowedAndSigned(payload: SnsPayload): Promise<void> {
+    if (!this.isTopicAllowlisted(payload.TopicArn)) {
+      this.logger.warn(`SNS topic ${payload.TopicArn} is not in allowlist`);
+
+      throw new MessagingWebhookException(
+        'SNS topic not allowed',
+        MessagingWebhookExceptionCode.MESSAGING_WEBHOOK_FORBIDDEN_TOPIC,
+      );
+    }
+
+    try {
+      await this.validator.validate(payload);
+    } catch (error) {
+      const errorMessage =
+        error instanceof Error ? error.message : String(error);
+
+      this.logger.warn(`SNS signature verification failed: ${errorMessage}`);
+
+      throw new MessagingWebhookException(
+        'SNS signature invalid',
+        MessagingWebhookExceptionCode.MESSAGING_WEBHOOK_INVALID_SIGNATURE,
+      );
+    }
+  }
+
+  private isTopicAllowlisted(topicArn: string): boolean {
+    const allowlist = this.twentyConfigService.get(
+      'SES_SNS_TOPIC_ARN_ALLOWLIST',
+    );
+
+    if (typeof allowlist !== 'string' || allowlist.trim() === '') {
+      return false;
+    }
+
+    return allowlist
+      .split(',')
+      .map((entry) => entry.trim())
+      .filter((entry) => entry.length > 0)
+      .includes(topicArn);
+  }
+}
@@ -0,0 +1,39 @@
+import { Injectable, Logger } from '@nestjs/common';
+
+import { MessagingWebhookExceptionCode } from 'src/modules/messaging-webhooks/messaging-webhook-exception-code.enum';
+import { MessagingWebhookException } from 'src/modules/messaging-webhooks/messaging-webhook.exception';
+
+const SNS_SUBSCRIBE_URL_PATTERN =
+  /^https:\/\/sns\.[a-z0-9-]+\.amazonaws\.com\//;
+
+@Injectable()
+export class SnsSubscriptionConfirmerService {
+  private readonly logger = new Logger(SnsSubscriptionConfirmerService.name);
+
+  async confirm(subscribeUrl: string | undefined): Promise<void> {
+    if (!subscribeUrl) {
+      throw new MessagingWebhookException(
+        'Missing SubscribeURL on SNS subscription confirmation',
+        MessagingWebhookExceptionCode.MESSAGING_WEBHOOK_INVALID_PAYLOAD,
+      );
+    }
+
+    if (!SNS_SUBSCRIBE_URL_PATTERN.test(subscribeUrl)) {
+      throw new MessagingWebhookException(
+        `Refusing to fetch non-AWS SubscribeURL: ${subscribeUrl}`,
+        MessagingWebhookExceptionCode.MESSAGING_WEBHOOK_INVALID_SUBSCRIBE_URL,
+      );
+    }
+
+    const response = await fetch(subscribeUrl);
+
+    if (!response.ok) {
+      throw new MessagingWebhookException(
+        `Failed to confirm SNS subscription via ${subscribeUrl}: ${response.status}`,
+        MessagingWebhookExceptionCode.MESSAGING_WEBHOOK_SUBSCRIPTION_CONFIRMATION_FAILED,
+      );
+    }
+
+    this.logger.log(`Confirmed SNS subscription via ${subscribeUrl}`);
+  }
+}
@@ -0,0 +1,37 @@
+type SesEventBridgeDetailType =
+  | 'Sending Status Enabled'
+  | 'Sending Status Disabled'
+  | 'Email Bounced'
+  | 'Email Complaint Received';
+
+type SesEventBridgeRecipient = {
+  emailAddress: string;
+};
+
+export type SesEventBridgeNotification = {
+  source: 'aws.ses';
+  'detail-type': SesEventBridgeDetailType;
+  resources?: string[];
+  detail?: {
+    version?: string;
+    data?: {
+      origin?: string;
+      record?: {
+        status?: 'ENABLED' | 'DISABLED';
+        cause?: string;
+      };
+    };
+    bounce?: {
+      bounceType?: 'Permanent' | 'Transient' | 'Undetermined';
+      feedbackId?: string;
+      bouncedRecipients?: SesEventBridgeRecipient[];
+    };
+    complaint?: {
+      feedbackId?: string;
+      complainedRecipients?: SesEventBridgeRecipient[];
+    };
+    mail?: {
+      messageId?: string;
+    };
+  };
+};
@@ -0,0 +1 @@
+export type SesInboundMailIntent = 'UNSUBSCRIBE' | 'IMPORT';
@@ -0,0 +1,5 @@
+import { type SESMessage } from 'aws-lambda';
+
+export type SesInboundNotification = SESMessage & {
+  notificationType?: string;
+};
@@ -0,0 +1,53 @@
+import { parseWorkspaceIdFromAwsSesResourceArn } from 'src/modules/messaging-webhooks/utils/parse-workspace-id-from-aws-ses-resource-arn.util';
+
+describe('parseWorkspaceIdFromAwsSesResourceArn', () => {
+  // Tenant ARNs have an AWS-assigned opaque id segment after the tenant name
+  // that must be discarded; configuration-set and identity ARNs do not. A
+  // single helper has to handle both shapes consistently.
+  it.each([
+    {
+      label: 'tenant ARN (drops the AWS-assigned tenant-id segment)',
+      arn: 'arn:aws:ses:us-east-1:123456789012:tenant/twenty-workspace-ws1/9b1deb4d-3b7d-4bad-9bdd-2b0d7b3dcb6d',
+    },
+    {
+      label: 'configuration-set ARN (single segment after resource type)',
+      arn: 'arn:aws:ses:us-east-1:123456789012:configuration-set/twenty-workspace-ws1',
+    },
+    {
+      label: 'identity ARN (single segment after resource type)',
+      arn: 'arn:aws:ses:us-east-1:123456789012:identity/twenty-workspace-ws1',
+    },
+  ])('extracts the workspaceId from a $label', ({ arn }) => {
+    expect(parseWorkspaceIdFromAwsSesResourceArn(arn)).toBe('ws1');
+  });
+
+  it('preserves the workspaceId verbatim when it is a UUID', () => {
+    expect(
+      parseWorkspaceIdFromAwsSesResourceArn(
+        'arn:aws:ses:us-east-1:123456789012:tenant/twenty-workspace-20202020-cb1b-4e35-b50f-2bbd09c3b1ee/9b1deb4d',
+      ),
+    ).toBe('20202020-cb1b-4e35-b50f-2bbd09c3b1ee');
+  });
+
+  // The prefix-check is the only guard preventing cross-tenant updates from
+  // foreign SES resources hitting the same SNS topic; an empty workspaceId
+  // (resource named exactly "twenty-workspace-") would otherwise produce a
+  // catastrophic empty WHERE clause downstream.
+  it.each([
+    {
+      label: 'foreign prefix',
+      arn: 'arn:aws:ses:us-east-1:123456789012:tenant/some-other-prefix/abc',
+    },
+    {
+      label: 'empty workspaceId after the prefix',
+      arn: 'arn:aws:ses:us-east-1:123456789012:tenant/twenty-workspace-/abc',
+    },
+    {
+      label: 'malformed ARN with no resource segment',
+      arn: 'arn:aws:ses:us-east-1:123456789012:tenant',
+    },
+    { label: 'empty string', arn: '' },
+  ])('returns null for $label', ({ arn }) => {
+    expect(parseWorkspaceIdFromAwsSesResourceArn(arn)).toBeNull();
+  });
+});
@@ -0,0 +1,24 @@
+import { assertUnreachable } from 'twenty-shared/utils';
+
+import { MessagingWebhookExceptionCode } from 'src/modules/messaging-webhooks/messaging-webhook-exception-code.enum';
+import { type MessagingWebhookException } from 'src/modules/messaging-webhooks/messaging-webhook.exception';
+
+export const getMessagingWebhookExceptionStatusCode = (
+  exception: MessagingWebhookException,
+): 400 | 403 | 500 => {
+  switch (exception.code) {
+    case MessagingWebhookExceptionCode.MESSAGING_WEBHOOK_MISSING_REQUEST_BODY:
+    case MessagingWebhookExceptionCode.MESSAGING_WEBHOOK_INVALID_PAYLOAD:
+    case MessagingWebhookExceptionCode.MESSAGING_WEBHOOK_INVALID_SUBSCRIBE_URL:
+      return 400;
+    case MessagingWebhookExceptionCode.MESSAGING_WEBHOOK_FORBIDDEN_TOPIC:
+    case MessagingWebhookExceptionCode.MESSAGING_WEBHOOK_INVALID_SIGNATURE:
+      return 403;
+    case MessagingWebhookExceptionCode.MESSAGING_WEBHOOK_SUBSCRIPTION_CONFIRMATION_FAILED:
+    case MessagingWebhookExceptionCode.MESSAGING_WEBHOOK_UNHANDLED_ERROR:
+      return 500;
+    default: {
+      return assertUnreachable(exception.code);
+    }
+  }
+};
@@ -0,0 +1,29 @@
+import { AWS_SES_RESOURCE_NAME_PREFIX } from 'src/engine/core-modules/emailing-domain/drivers/aws-ses/constants/aws-ses-resource-name-prefix.constant';
+import { isDefined } from 'twenty-shared/utils';
+
+export const parseWorkspaceIdFromAwsSesResourceArn = (
+  resourceArn: string,
+): string | null => {
+  const slashIndex = resourceArn.indexOf('/');
+
+  if (slashIndex === -1) {
+    return null;
+  }
+
+  const afterPrefix = resourceArn.slice(slashIndex + 1);
+  const resourceName = afterPrefix.split('/')[0];
+
+  if (!isDefined(resourceName)) {
+    return null;
+  }
+
+  const expectedPrefix = `${AWS_SES_RESOURCE_NAME_PREFIX}-`;
+
+  if (!resourceName.startsWith(expectedPrefix)) {
+    return null;
+  }
+
+  const workspaceId = resourceName.slice(expectedPrefix.length);
+
+  return workspaceId.length > 0 ? workspaceId : null;
+};
@@ -0,0 +1,16 @@
+import { UNSUBSCRIBE_MAILBOX_LOCAL_PART } from 'src/engine/core-modules/emailing-domain/constants/unsubscribe-mailbox.constant';
+import { type SesInboundMailIntent } from 'src/modules/messaging-webhooks/types/ses-inbound-mail-intent.type';
+import { type SesInboundNotification } from 'src/modules/messaging-webhooks/types/sns-message.type';
+
+export const resolveInboundMailIntent = (
+  notification: SesInboundNotification,
+): SesInboundMailIntent => {
+  const isAddressedToUnsubscribeMailbox = (
+    notification.receipt?.recipients ?? []
+  ).some(
+    (recipient) =>
+      recipient.split('@')[0]?.toLowerCase() === UNSUBSCRIBE_MAILBOX_LOCAL_PART,
+  );
+
+  return isAddressedToUnsubscribeMailbox ? 'UNSUBSCRIBE' : 'IMPORT';
+};
@@ -0,0 +1,21 @@
+import { isDefined, isNonEmptyArray } from 'twenty-shared/utils';
+
+import { parseWorkspaceIdFromAwsSesResourceArn } from 'src/modules/messaging-webhooks/utils/parse-workspace-id-from-aws-ses-resource-arn.util';
+
+export const resolveWorkspaceIdFromAwsSesResources = (
+  resources: string[] | undefined,
+): string | null => {
+  if (!isNonEmptyArray(resources)) {
+    return null;
+  }
+
+  for (const resourceArn of resources) {
+    const workspaceId = parseWorkspaceIdFromAwsSesResourceArn(resourceArn);
+
+    if (isDefined(workspaceId)) {
+      return workspaceId;
+    }
+  }
+
+  return null;
+};
				`@@ -0,0 +1 @@`
				`export type SesInboundMailIntent = 'UNSUBSCRIBE' \| 'IMPORT';`