feat: add email forwarding message channel (#19535)

## Summary

- Add email forwarding as a new message channel type, allowing users to
forward emails from addresses like `support@mycompany.com` into Twenty
- Inbound emails arrive via S3 (SES → S3 bucket), are polled by a cron
job, parsed, routed to the correct workspace/channel, and persisted as
messages
- Dedicated settings page at `/settings/accounts/new-email-forwarding`
where users provide their source email handle and receive a unique
forwarding address
- Forwarding channels bypass the IMAP/mailbox sync state machine — they
skip cron-driven sync, relaunch, and message-list-fetch lifecycle stages
- Forwarding address section shown at the top of the Emails settings
page so users can find/copy their addresses after initial setup
- Tab names for forwarding channels display the user-provided handle
(e.g. `support@mycompany.com`) instead of the internal routing address
- Shared utilities extracted from IMAP driver: `extractThreadId`,
`extractParticipants`, `extractAddresses` to avoid code duplication
- Uses the existing S3 bucket (STORAGE_S3_*) with `inbound-email/`
prefix — no separate bucket needed
- Feature gated behind `isEmailForwardingEnabled` client config
(requires `INBOUND_EMAIL_DOMAIN` + S3 storage)

## New backend modules

- `InboundEmailS3ClientProvider` — lazy-initialized S3 client using
existing storage config
- `InboundEmailStorageService` — S3 operations (get, move to
processed/unmatched/failed)
- `InboundEmailParserService` — RFC 822 parsing via `postal-mime`,
builds `MessageWithParticipants`
- `InboundEmailImportService` — orchestrates download → parse → route →
persist → archive
- `MessagingInboundEmailPollCronJob` — polls S3 `incoming/` prefix,
enqueues import jobs
- `CreateEmailForwardingChannelInput` DTO — accepts user-provided
`handle`

## New frontend components

- `SettingsAccountsNewEmailForwardingChannel` — dedicated page with
handle input form + forwarding address result
- `SettingsAccountsEmailForwardingSection` — forwarding address list on
the Emails settings page
- `useConnectedAccountHandleMap` — shared hook for account ID → handle
lookup
- `useCreateEmailForwardingChannel` — mutation hook accepting handle
parameter

## Test plan

- [x] 17 unit tests for inbound email import service (all outcomes:
imported, unmatched, loop_dropped, unconfigured, parse_failed,
persist_failed)
- [x] 16 tests for `computeSyncStatus` including EMAIL_FORWARDING cases
- [x] 11 tests for `extractEnvelopeRecipient` utility
- [x] TypeScript typechecks pass for both twenty-server and twenty-front
- [x] Lint passes for both packages
- [ ] Manual: create forwarding channel, verify forwarding address
generated
- [ ] Manual: send email to forwarding address, verify it appears in
Twenty

https://claude.ai/code/session_01KpyF6p4cUEnuaT4h8DP5Pm

---------

Co-authored-by: Claude <noreply@anthropic.com>
Co-authored-by: Copilot Autofix powered by AI <223894421+github-code-quality[bot]@users.noreply.github.com>
Co-authored-by: claude[bot] <41898282+claude[bot]@users.noreply.github.com>
Co-authored-by: neo773 <62795688+neo773@users.noreply.github.com>
Co-authored-by: neo773 <neo773@protonmail.com>

packages/twenty-server/src/engine/core-modules/auth/services/auth.service.ts

@@ -1140,6 +1140,7 @@ export class AuthService {
         return [];
       case ConnectedAccountProvider.IMAP_SMTP_CALDAV:
         return [];
+      case ConnectedAccountProvider.EMAIL_GROUP:
       case ConnectedAccountProvider.APP:
         return [];
       default:

packages/twenty-server/src/engine/core-modules/client-config/client-config.controller.spec.ts

@@ -94,6 +94,7 @@ describe('ClientConfigController', () => {
         isGoogleCalendarEnabled: false,
         isConfigVariablesInDbEnabled: false,
         isImapSmtpCaldavEnabled: false,
+        isEmailGroupEnabled: false,
         calendarBookingPageId: undefined,
         isTwoFactorAuthenticationEnabled: false,
         allowRequestsToTwentyIcons: true,

packages/twenty-server/src/engine/core-modules/client-config/client-config.entity.ts

@@ -309,6 +309,9 @@ export class ClientConfig {
   @Field(() => Boolean)
   isImapSmtpCaldavEnabled: boolean;
 
+  @Field(() => Boolean)
+  isEmailGroupEnabled: boolean;
+
   @Field(() => Boolean)
   allowRequestsToTwentyIcons: boolean;
 

packages/twenty-server/src/engine/core-modules/client-config/services/client-config.service.spec.ts

@@ -171,6 +171,7 @@ describe('ClientConfigService', () => {
         isGoogleCalendarEnabled: true,
         isConfigVariablesInDbEnabled: false,
         isImapSmtpCaldavEnabled: false,
+        isEmailGroupEnabled: false,
         allowRequestsToTwentyIcons: false,
         calendarBookingPageId: 'team/twenty/talk-to-us',
         isCloudflareIntegrationEnabled: false,

packages/twenty-server/src/engine/core-modules/client-config/services/client-config.service.ts

@@ -4,6 +4,8 @@ import { isNonEmptyString } from '@sniptt/guards';
 import { isDefined } from 'twenty-shared/utils';
 import { type AiSdkPackage } from 'twenty-shared/ai';
 
+import { StorageDriverType } from 'src/engine/core-modules/file-storage/interfaces/file-storage.interface';
+
 import {
   AI_SDK_ANTHROPIC,
   AI_SDK_BEDROCK,
@@ -247,6 +249,10 @@ export class ClientConfigService {
       isImapSmtpCaldavEnabled: this.twentyConfigService.get(
         'IS_IMAP_SMTP_CALDAV_ENABLED',
       ),
+      isEmailGroupEnabled:
+        this.twentyConfigService.get('STORAGE_TYPE') ===
+          StorageDriverType.S_3 &&
+        isNonEmptyString(this.twentyConfigService.get('INBOUND_EMAIL_DOMAIN')),
       allowRequestsToTwentyIcons: this.twentyConfigService.get(
         'ALLOW_REQUESTS_TO_TWENTY_ICONS',
       ),

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

@@ -47,6 +47,7 @@ import { LogicFunctionModule } from 'src/engine/core-modules/logic-function/logi
 import { MessageQueueModule } from 'src/engine/core-modules/message-queue/message-queue.module';
 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/engine/core-modules/messaging-webhooks/messaging-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';
@@ -89,6 +90,7 @@ import { FileModule } from './file/file.module';
     AuthModule,
     BillingModule,
     BillingWebhookModule,
+    MessagingWebhooksModule,
     UsageModule,
     ClientConfigModule,
     FeatureFlagModule,

packages/twenty-server/src/engine/core-modules/messaging-webhooks/messaging-webhooks.controller.ts

@@ -0,0 +1,67 @@
+import {
+  BadRequestException,
+  Controller,
+  HttpCode,
+  Post,
+  type RawBodyRequest,
+  Req,
+  UseGuards,
+} from '@nestjs/common';
+
+import { type Request } from 'express';
+import type SnsPayloadValidator from 'sns-payload-validator';
+
+import { MessagingWebhookDispatcherService } from 'src/engine/core-modules/messaging-webhooks/services/messaging-webhook-dispatcher.service';
+import { SnsSignatureVerifierService } from 'src/engine/core-modules/messaging-webhooks/services/sns-signature-verifier.service';
+import { NoPermissionGuard } from 'src/engine/guards/no-permission.guard';
+import { PublicEndpointGuard } from 'src/engine/guards/public-endpoint.guard';
+
+type SnsPayload = SnsPayloadValidator.SnsPayload;
+
+@Controller()
+export class MessagingWebhooksController {
+  constructor(
+    private readonly snsSignatureVerifierService: SnsSignatureVerifierService,
+    private readonly messagingWebhookDispatcherService: MessagingWebhookDispatcherService,
+  ) {}
+
+  @Post(['webhooks/messaging/ses'])
+  @UseGuards(PublicEndpointGuard, NoPermissionGuard)
+  @HttpCode(200)
+  async handleSesWebhook(
+    @Req() request: RawBodyRequest<Request>,
+  ): Promise<void> {
+    if (!request.rawBody) {
+      throw new BadRequestException('Missing SNS payload');
+    }
+
+    const payload = this.parseSnsPayload(request.rawBody);
+
+    await this.snsSignatureVerifierService.assertAllowedAndSigned(payload);
+
+    if (
+      payload.Type === 'SubscriptionConfirmation' ||
+      payload.Type === 'UnsubscribeConfirmation'
+    ) {
+      await this.messagingWebhookDispatcherService.confirmSnsSubscription(
+        payload.SubscribeURL,
+      );
+
+      return;
+    }
+
+    if (payload.Type === 'Notification') {
+      await this.messagingWebhookDispatcherService.dispatchSnsNotification(
+        payload,
+      );
+    }
+  }
+
+  private parseSnsPayload(rawBody: Buffer): SnsPayload {
+    try {
+      return JSON.parse(rawBody.toString('utf8')) as SnsPayload;
+    } catch {
+      throw new BadRequestException('Invalid SNS payload');
+    }
+  }
+}

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

@@ -0,0 +1,13 @@
+import { Module } from '@nestjs/common';
+
+import { MessagingWebhooksController } from 'src/engine/core-modules/messaging-webhooks/messaging-webhooks.controller';
+import { MessagingWebhookDispatcherService } from 'src/engine/core-modules/messaging-webhooks/services/messaging-webhook-dispatcher.service';
+import { SnsSignatureVerifierService } from 'src/engine/core-modules/messaging-webhooks/services/sns-signature-verifier.service';
+import { TwentyConfigModule } from 'src/engine/core-modules/twenty-config/twenty-config.module';
+
+@Module({
+  imports: [TwentyConfigModule],
+  controllers: [MessagingWebhooksController],
+  providers: [SnsSignatureVerifierService, MessagingWebhookDispatcherService],
+})
+export class MessagingWebhooksModule {}

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

@@ -0,0 +1,99 @@
+import { Injectable, Logger } from '@nestjs/common';
+
+import type SnsPayloadValidator from 'sns-payload-validator';
+
+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 { type SesInboundNotification } from 'src/engine/core-modules/messaging-webhooks/types/sns-message.type';
+import {
+  MessagingInboundEmailImportJob,
+  type MessagingInboundEmailImportJobData,
+} from 'src/modules/messaging/message-import-manager/jobs/messaging-inbound-email-import.job';
+
+type SnsPayload = SnsPayloadValidator.SnsPayload;
+
+@Injectable()
+export class MessagingWebhookDispatcherService {
+  private readonly logger = new Logger(MessagingWebhookDispatcherService.name);
+
+  constructor(
+    @InjectMessageQueue(MessageQueue.messagingQueue)
+    private readonly messageQueueService: MessageQueueService,
+  ) {}
+
+  private static readonly SNS_SUBSCRIBE_URL_PATTERN =
+    /^https:\/\/sns\.[a-z0-9-]+\.amazonaws\.com\//;
+
+  async confirmSnsSubscription(
+    subscribeUrl: string | undefined,
+  ): Promise<void> {
+    if (!subscribeUrl) {
+      return;
+    }
+
+    if (
+      !MessagingWebhookDispatcherService.SNS_SUBSCRIBE_URL_PATTERN.test(
+        subscribeUrl,
+      )
+    ) {
+      this.logger.error(
+        `Refusing to fetch non-AWS SubscribeURL: ${subscribeUrl}`,
+      );
+
+      return;
+    }
+
+    const response = await fetch(subscribeUrl);
+
+    if (!response.ok) {
+      this.logger.error(
+        `Failed to confirm SNS subscription via ${subscribeUrl}: ${response.status}`,
+      );
+
+      return;
+    }
+
+    this.logger.log(`Confirmed SNS subscription via ${subscribeUrl}`);
+  }
+
+  async dispatchSnsNotification(payload: SnsPayload): Promise<void> {
+    const notification = this.parseSesInboundNotification(payload.Message);
+
+    if (!notification) {
+      this.logger.warn(
+        `SNS message ${payload.MessageId} has invalid JSON body`,
+      );
+
+      return;
+    }
+
+    const { receipt } = notification;
+
+    if (receipt.action.type !== 'S3') {
+      this.logger.warn(
+        `SNS message ${payload.MessageId} has unsupported action type ${receipt.action.type}`,
+      );
+
+      return;
+    }
+
+    await this.messageQueueService.add<MessagingInboundEmailImportJobData>(
+      MessagingInboundEmailImportJob.name,
+      {
+        s3Key: receipt.action.objectKey,
+        envelopeRecipients: receipt.recipients,
+      },
+    );
+  }
+
+  private parseSesInboundNotification(
+    rawJson: string,
+  ): SesInboundNotification | null {
+    try {
+      return JSON.parse(rawJson) as SesInboundNotification;
+    } catch {
+      return null;
+    }
+  }
+}

packages/twenty-server/src/engine/core-modules/messaging-webhooks/services/sns-signature-verifier.service.ts

@@ -0,0 +1,50 @@
+import { ForbiddenException, Injectable, Logger } from '@nestjs/common';
+
+import SnsPayloadValidator from 'sns-payload-validator';
+
+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 ForbiddenException('SNS topic not allowed');
+    }
+
+    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 ForbiddenException('SNS signature invalid');
+    }
+  }
+
+  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);
+  }
+}

packages/twenty-server/src/engine/core-modules/messaging-webhooks/types/sns-message.type.ts

@@ -0,0 +1,5 @@
+import { type SESMessage } from 'aws-lambda';
+
+export type SesInboundNotification = SESMessage & {
+  notificationType?: string;
+};

packages/twenty-server/src/engine/core-modules/twenty-config/config-variables.ts

@@ -1652,6 +1652,24 @@ export class ConfigVariables {
   @IsOptional()
   AWS_SES_ACCOUNT_ID: string;
 
+  @ConfigVariablesMetadata({
+    group: ConfigVariablesGroup.AWS_SES_SETTINGS,
+    description:
+      'Domain used for email group inbound mail (the right-hand side of ch_xxx@<domain>). Required to enable email group channels.',
+    type: ConfigVariableType.STRING,
+  })
+  @IsOptional()
+  INBOUND_EMAIL_DOMAIN: string;
+
+  @ConfigVariablesMetadata({
+    group: ConfigVariablesGroup.AWS_SES_SETTINGS,
+    description:
+      'Comma-separated list of SNS topic ARNs accepted by the inbound-email webhook (e.g. arn:aws:sns:us-east-1:123:my-inbound).',
+    type: ConfigVariableType.STRING,
+  })
+  @IsOptional()
+  SES_SNS_TOPIC_ARN_ALLOWLIST: string;
+
   @ConfigVariablesMetadata({
     group: ConfigVariablesGroup.ADVANCED_SETTINGS,
     description: 'Timeout in milliseconds for primary database queries',