wip(messaging): scaffold email forwarding channel inbound pipeline

Adds EMAIL_FORWARDING as a new MessageChannelType and ConnectedAccountProvider,
along with the S3-polling inbound email driver skeleton (storage, parser,
import service, driver module) and supporting config variables. Provider
switches across auth, refresh tokens, email aliases, and outbound routing
are updated to handle the new enum value.

This is a work-in-progress commit; the poll cron job, list-fetch cron
filter, channel creation mutation, frontend UI and tests are not yet
implemented.

https://claude.ai/code/session_01SM4RuPAL9p8CTmt3xvKgp3

packages/twenty-front/src/modules/accounts/utils/hasMissingDraftEmailScopes.ts

@@ -27,6 +27,7 @@ export const getMissingDraftEmailScopes = (
     case ConnectedAccountProvider.IMAP_SMTP_CALDAV:
     case ConnectedAccountProvider.OIDC:
     case ConnectedAccountProvider.SAML:
+    case ConnectedAccountProvider.EMAIL_FORWARDING:
       return [];
     default:
       assertUnreachable(

packages/twenty-front/src/modules/ui/field/display/components/ActorDisplay.tsx

@@ -30,6 +30,7 @@ const PROVIDERS_ICON_MAPPING = {
     [ConnectedAccountProvider.IMAP_SMTP_CALDAV]: IconMail,
     [ConnectedAccountProvider.OIDC]: IconMail,
     [ConnectedAccountProvider.SAML]: IconMail,
+    [ConnectedAccountProvider.EMAIL_FORWARDING]: IconMail,
     default: IconMail,
   },
   CALENDAR: {

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

@@ -1127,6 +1127,8 @@ export class AuthService {
         return [];
       case ConnectedAccountProvider.IMAP_SMTP_CALDAV:
         return [];
+      case ConnectedAccountProvider.EMAIL_FORWARDING:
+        return [];
       default:
         throw new Error(
           `Unsupported connected account provider: ${provider satisfies never}`,

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

@@ -1622,6 +1622,43 @@ export class ConfigVariables {
   @IsOptional()
   AWS_SES_ACCOUNT_ID: string;
 
+  @ConfigVariablesMetadata({
+    group: ConfigVariablesGroup.AWS_SES_SETTINGS,
+    description:
+      'Domain used for inbound email forwarding addresses (e.g. in.twenty.com). Each email forwarding channel is assigned an opaque local part at this domain.',
+    type: ConfigVariableType.STRING,
+  })
+  @IsOptional()
+  INBOUND_EMAIL_DOMAIN: string;
+
+  @ConfigVariablesMetadata({
+    group: ConfigVariablesGroup.AWS_SES_SETTINGS,
+    description:
+      'S3 bucket name where SES writes incoming raw RFC822 messages for the email forwarding feature. Messages are read from the "incoming/" prefix and moved to "processed/", "unmatched/" or "failed/" after handling.',
+    type: ConfigVariableType.STRING,
+  })
+  @IsOptional()
+  INBOUND_EMAIL_S3_BUCKET: string;
+
+  @ConfigVariablesMetadata({
+    group: ConfigVariablesGroup.AWS_SES_SETTINGS,
+    description:
+      'AWS region of the inbound email S3 bucket. Defaults to AWS_SES_REGION when unset.',
+    type: ConfigVariableType.STRING,
+  })
+  @IsOptional()
+  INBOUND_EMAIL_S3_REGION: string;
+
+  @ConfigVariablesMetadata({
+    group: ConfigVariablesGroup.AWS_SES_SETTINGS,
+    description:
+      'Maximum number of S3 keys processed per inbound email poll tick. Acts as a soft time limit; remaining keys are picked up by the next tick.',
+    type: ConfigVariableType.NUMBER,
+  })
+  @CastToPositiveNumber()
+  @IsOptional()
+  INBOUND_EMAIL_POLL_BATCH_SIZE: number = 500;
+
   @ConfigVariablesMetadata({
     group: ConfigVariablesGroup.ADVANCED_SETTINGS,
     description: 'Timeout in milliseconds for primary database queries',

packages/twenty-server/src/engine/workspace-manager/twenty-standard-application/utils/field-metadata/compute-message-channel-standard-flat-field-metadata.util.ts

@@ -286,6 +286,13 @@ export const buildMessageChannelStandardFlatFieldMetadatas = ({
           position: 1,
           color: 'blue',
         },
+        {
+          id: '20202020-7f22-4e58-aa33-9c3e2c72ab10',
+          value: MessageChannelType.EMAIL_FORWARDING,
+          label: i18nLabel(msg`Email Forwarding`),
+          position: 2,
+          color: 'turquoise',
+        },
       ],
     },
     standardObjectMetadataRelatedEntityIds,

packages/twenty-server/src/modules/connected-account/email-alias-manager/services/email-alias-manager.service.ts

@@ -43,6 +43,7 @@ export class EmailAliasManagerService {
       case ConnectedAccountProvider.IMAP_SMTP_CALDAV:
       case ConnectedAccountProvider.OIDC:
       case ConnectedAccountProvider.SAML:
+      case ConnectedAccountProvider.EMAIL_FORWARDING:
         handleAliases = [];
         break;
       default:

packages/twenty-server/src/modules/connected-account/refresh-tokens-manager/services/connected-account-refresh-tokens.service.ts

@@ -117,6 +117,7 @@ export class ConnectedAccountRefreshTokensService {
       case ConnectedAccountProvider.IMAP_SMTP_CALDAV:
       case ConnectedAccountProvider.OIDC:
       case ConnectedAccountProvider.SAML:
+      case ConnectedAccountProvider.EMAIL_FORWARDING:
         return true;
       default:
         return assertUnreachable(
@@ -144,6 +145,7 @@ export class ConnectedAccountRefreshTokensService {
         case ConnectedAccountProvider.IMAP_SMTP_CALDAV:
         case ConnectedAccountProvider.OIDC:
         case ConnectedAccountProvider.SAML:
+        case ConnectedAccountProvider.EMAIL_FORWARDING:
           throw new ConnectedAccountRefreshAccessTokenException(
             `Token refresh is not supported for ${connectedAccount.provider} provider for connected account ${connectedAccount.id} in workspace ${workspaceId}`,
             ConnectedAccountRefreshAccessTokenExceptionCode.PROVIDER_NOT_SUPPORTED,

packages/twenty-server/src/modules/messaging/message-import-manager/drivers/inbound-email/constants/inbound-email.constants.ts

@@ -0,0 +1,15 @@
+export const INBOUND_EMAIL_S3_PREFIXES = {
+  INCOMING: 'incoming/',
+  PROCESSED: 'processed/',
+  UNMATCHED: 'unmatched/',
+  FAILED: 'failed/',
+} as const;
+
+// Header added to outbound messages sent to a group via an email forwarding
+// channel. When the forwarding service echoes our own send back as an inbound
+// message, we use this header to drop it and avoid a loop.
+export const INBOUND_EMAIL_ORIGIN_HEADER = 'x-twenty-origin';
+
+// Prefix used for the opaque local part of the forwarding address.
+// The full address is <PREFIX><24 random chars>@<INBOUND_EMAIL_DOMAIN>
+export const INBOUND_EMAIL_LOCAL_PART_PREFIX = 'ch_';

packages/twenty-server/src/modules/messaging/message-import-manager/drivers/inbound-email/messaging-inbound-email-driver.module.ts

@@ -0,0 +1,33 @@
+import { Module } from '@nestjs/common';
+import { TypeOrmModule } from '@nestjs/typeorm';
+
+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 { TwentyConfigModule } from 'src/engine/core-modules/twenty-config/twenty-config.module';
+import { WorkspaceDataSourceModule } from 'src/engine/workspace-datasource/workspace-datasource.module';
+import { MessagingCommonModule } from 'src/modules/messaging/common/messaging-common.module';
+import { InboundEmailS3ClientProvider } from 'src/modules/messaging/message-import-manager/drivers/inbound-email/providers/inbound-email-s3-client.provider';
+import { InboundEmailImportService } from 'src/modules/messaging/message-import-manager/drivers/inbound-email/services/inbound-email-import.service';
+import { InboundEmailParserService } from 'src/modules/messaging/message-import-manager/drivers/inbound-email/services/inbound-email-parser.service';
+import { InboundEmailStorageService } from 'src/modules/messaging/message-import-manager/drivers/inbound-email/services/inbound-email-storage.service';
+
+@Module({
+  imports: [
+    TypeOrmModule.forFeature([MessageChannelEntity, ConnectedAccountEntity]),
+    TwentyConfigModule,
+    WorkspaceDataSourceModule,
+    MessagingCommonModule,
+  ],
+  providers: [
+    InboundEmailS3ClientProvider,
+    InboundEmailStorageService,
+    InboundEmailParserService,
+    InboundEmailImportService,
+  ],
+  exports: [
+    InboundEmailImportService,
+    InboundEmailStorageService,
+    InboundEmailS3ClientProvider,
+  ],
+})
+export class MessagingInboundEmailDriverModule {}

packages/twenty-server/src/modules/messaging/message-import-manager/drivers/inbound-email/providers/inbound-email-s3-client.provider.ts

@@ -0,0 +1,71 @@
+import { Injectable, Logger } from '@nestjs/common';
+
+import { S3Client, type S3ClientConfig } from '@aws-sdk/client-s3';
+
+import { TwentyConfigService } from 'src/engine/core-modules/twenty-config/twenty-config.service';
+
+// Thin wrapper that lazily builds an S3 client using the inbound email config
+// (INBOUND_EMAIL_S3_REGION falls back to AWS_SES_REGION, and AWS_SES_*
+// credentials are reused). Returning `null` when the bucket is unconfigured
+// lets callers short-circuit without throwing.
+@Injectable()
+export class InboundEmailS3ClientProvider {
+  private readonly logger = new Logger(InboundEmailS3ClientProvider.name);
+  private s3Client: S3Client | null = null;
+
+  constructor(private readonly twentyConfigService: TwentyConfigService) {}
+
+  getBucketName(): string | null {
+    const bucket = this.twentyConfigService.get('INBOUND_EMAIL_S3_BUCKET');
+
+    return bucket && bucket.length > 0 ? bucket : null;
+  }
+
+  isConfigured(): boolean {
+    return this.getBucketName() !== null;
+  }
+
+  getClient(): S3Client | null {
+    const bucket = this.getBucketName();
+
+    if (!bucket) {
+      return null;
+    }
+
+    if (!this.s3Client) {
+      const region =
+        this.twentyConfigService.get('INBOUND_EMAIL_S3_REGION') ||
+        this.twentyConfigService.get('AWS_SES_REGION');
+
+      if (!region) {
+        this.logger.warn(
+          'Inbound email bucket is configured but no region is set. Set INBOUND_EMAIL_S3_REGION or AWS_SES_REGION.',
+        );
+
+        return null;
+      }
+
+      const config: S3ClientConfig = { region };
+
+      const accessKeyId = this.twentyConfigService.get('AWS_SES_ACCESS_KEY_ID');
+      const secretAccessKey = this.twentyConfigService.get(
+        'AWS_SES_SECRET_ACCESS_KEY',
+      );
+      const sessionToken = this.twentyConfigService.get(
+        'AWS_SES_SESSION_TOKEN',
+      );
+
+      if (accessKeyId && secretAccessKey) {
+        config.credentials = {
+          accessKeyId,
+          secretAccessKey,
+          ...(sessionToken ? { sessionToken } : {}),
+        };
+      }
+
+      this.s3Client = new S3Client(config);
+    }
+
+    return this.s3Client;
+  }
+}

packages/twenty-server/src/modules/messaging/message-import-manager/drivers/inbound-email/services/inbound-email-import.service.ts

@@ -0,0 +1,198 @@
+import { Injectable, Logger } from '@nestjs/common';
+import { InjectRepository } from '@nestjs/typeorm';
+
+import { Repository } from 'typeorm';
+import { MessageChannelType } from 'twenty-shared/types';
+
+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 { GlobalWorkspaceOrmManager } from 'src/engine/twenty-orm/global-workspace-datasource/global-workspace-orm.manager';
+import { buildSystemAuthContext } from 'src/engine/twenty-orm/utils/build-system-auth-context.util';
+import { TwentyConfigService } from 'src/engine/core-modules/twenty-config/twenty-config.service';
+import {
+  InboundEmailParserService,
+  type InboundEmailParseResult,
+} from 'src/modules/messaging/message-import-manager/drivers/inbound-email/services/inbound-email-parser.service';
+import { InboundEmailStorageService } from 'src/modules/messaging/message-import-manager/drivers/inbound-email/services/inbound-email-storage.service';
+import {
+  extractEnvelopeRecipientForDomain,
+  extractLocalPart,
+} from 'src/modules/messaging/message-import-manager/drivers/inbound-email/utils/extract-envelope-recipient.util';
+import { MessagingSaveMessagesAndEnqueueContactCreationService } from 'src/modules/messaging/message-import-manager/services/messaging-save-messages-and-enqueue-contact-creation.service';
+
+export type InboundEmailImportOutcome =
+  | 'imported'
+  | 'unmatched'
+  | 'duplicate'
+  | 'loop_dropped'
+  | 'unconfigured'
+  | 'parse_failed'
+  | 'persist_failed';
+
+@Injectable()
+export class InboundEmailImportService {
+  private readonly logger = new Logger(InboundEmailImportService.name);
+
+  constructor(
+    private readonly storage: InboundEmailStorageService,
+    private readonly parser: InboundEmailParserService,
+    private readonly twentyConfigService: TwentyConfigService,
+    private readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager,
+    @InjectRepository(MessageChannelEntity)
+    private readonly messageChannelRepository: Repository<MessageChannelEntity>,
+    @InjectRepository(ConnectedAccountEntity)
+    private readonly connectedAccountRepository: Repository<ConnectedAccountEntity>,
+    private readonly saveMessagesService: MessagingSaveMessagesAndEnqueueContactCreationService,
+  ) {}
+
+  async importFromS3Key(s3Key: string): Promise<InboundEmailImportOutcome> {
+    const domain = this.twentyConfigService.get('INBOUND_EMAIL_DOMAIN');
+
+    if (!domain) {
+      this.logger.warn(
+        'INBOUND_EMAIL_DOMAIN is not configured; leaving inbound message in place',
+      );
+
+      return 'unconfigured';
+    }
+
+    let raw: Buffer;
+
+    try {
+      raw = await this.storage.getRawMessage(s3Key);
+    } catch (error) {
+      this.logger.error(
+        `Failed to fetch inbound email S3 object ${s3Key}: ${(error as Error).message}`,
+      );
+      await this.safeMoveToFailed(s3Key);
+
+      return 'persist_failed';
+    }
+
+    let parseResult: InboundEmailParseResult;
+
+    try {
+      parseResult = await this.parser.parseRawMessage(raw, s3Key);
+    } catch (error) {
+      this.logger.error(
+        `Failed to parse inbound email ${s3Key}: ${(error as Error).message}`,
+      );
+      await this.safeMoveToFailed(s3Key);
+
+      return 'parse_failed';
+    }
+
+    const envelopeRecipient = extractEnvelopeRecipientForDomain(
+      parseResult.parsed,
+      domain,
+    );
+
+    if (!envelopeRecipient) {
+      this.logger.log(
+        `Inbound email ${s3Key} has no recipient at ${domain}; archiving as unmatched`,
+      );
+      await this.storage.moveToUnmatched(s3Key);
+
+      return 'unmatched';
+    }
+
+    const localPart = extractLocalPart(envelopeRecipient);
+
+    const channel = await this.findChannelByLocalPart(localPart);
+
+    if (!channel) {
+      this.logger.log(
+        `No email forwarding channel matches inbound address ${envelopeRecipient}`,
+      );
+      await this.storage.moveToUnmatched(s3Key);
+
+      return 'unmatched';
+    }
+
+    // Loop prevention: messages we sent ourselves carry X-Twenty-Origin with
+    // our workspace id. If the group echoes our send back, we drop it.
+    if (parseResult.originWorkspaceId === channel.workspaceId) {
+      this.logger.log(
+        `Dropping inbound email ${s3Key} as self-echo from workspace ${channel.workspaceId}`,
+      );
+      await this.storage.moveToProcessed(s3Key);
+
+      return 'loop_dropped';
+    }
+
+    try {
+      await this.persistMessage(channel, parseResult);
+    } catch (error) {
+      this.logger.error(
+        `Failed to persist inbound email ${s3Key}: ${(error as Error).message}`,
+      );
+      await this.safeMoveToFailed(s3Key);
+
+      return 'persist_failed';
+    }
+
+    await this.storage.moveToProcessed(s3Key);
+
+    return 'imported';
+  }
+
+  private async findChannelByLocalPart(
+    localPart: string,
+  ): Promise<MessageChannelEntity | null> {
+    const domain = this.twentyConfigService.get('INBOUND_EMAIL_DOMAIN');
+
+    if (!domain) {
+      return null;
+    }
+
+    const fullAddress = `${localPart}@${domain}`.toLowerCase();
+
+    // Email forwarding channels are uniquely keyed on handle + type.
+    return this.messageChannelRepository.findOne({
+      where: {
+        handle: fullAddress,
+        type: MessageChannelType.EMAIL_FORWARDING,
+      },
+      relations: { connectedAccount: true },
+    });
+  }
+
+  private async persistMessage(
+    channel: MessageChannelEntity,
+    parseResult: InboundEmailParseResult,
+  ): Promise<void> {
+    const connectedAccount = await this.connectedAccountRepository.findOne({
+      where: { id: channel.connectedAccountId },
+    });
+
+    if (!connectedAccount) {
+      throw new Error(
+        `Connected account ${channel.connectedAccountId} missing for email forwarding channel ${channel.id}`,
+      );
+    }
+
+    const authContext = buildSystemAuthContext(channel.workspaceId);
+
+    await this.globalWorkspaceOrmManager.executeInWorkspaceContext(
+      async () => {
+        await this.saveMessagesService.saveMessagesAndEnqueueContactCreation(
+          [parseResult.message],
+          channel,
+          connectedAccount,
+          channel.workspaceId,
+        );
+      },
+      authContext,
+    );
+  }
+
+  private async safeMoveToFailed(s3Key: string): Promise<void> {
+    try {
+      await this.storage.moveToFailed(s3Key);
+    } catch (moveError) {
+      this.logger.error(
+        `Failed to move ${s3Key} to failed/: ${(moveError as Error).message}`,
+      );
+    }
+  }
+}

packages/twenty-server/src/modules/messaging/message-import-manager/drivers/inbound-email/services/inbound-email-parser.service.ts

@@ -0,0 +1,122 @@
+import { Injectable, Logger } from '@nestjs/common';
+
+import PostalMime, {
+  type Address,
+  type Email as ParsedEmail,
+} from 'postal-mime';
+import { MessageDirection } from 'src/modules/messaging/common/enums/message-direction.enum';
+import { MessageParticipantRole } from 'twenty-shared/types';
+
+import { INBOUND_EMAIL_ORIGIN_HEADER } from 'src/modules/messaging/message-import-manager/drivers/inbound-email/constants/inbound-email.constants';
+import { type EmailAddress } from 'src/modules/messaging/message-import-manager/types/email-address';
+import { type MessageWithParticipants } from 'src/modules/messaging/message-import-manager/types/message';
+import { formatAddressObjectAsParticipants } from 'src/modules/messaging/message-import-manager/utils/format-address-object-as-participants.util';
+import { sanitizeString } from 'src/modules/messaging/message-import-manager/utils/sanitize-string.util';
+
+export type InboundEmailParseResult = {
+  parsed: ParsedEmail;
+  message: MessageWithParticipants;
+  originWorkspaceId: string | null;
+};
+
+@Injectable()
+export class InboundEmailParserService {
+  private readonly logger = new Logger(InboundEmailParserService.name);
+
+  async parseRawMessage(
+    raw: Buffer,
+    s3Key: string,
+  ): Promise<InboundEmailParseResult> {
+    const parsed = await PostalMime.parse(raw);
+
+    const externalId = parsed.messageId?.trim() || `s3:${s3Key}`;
+    const messageThreadExternalId = this.extractThreadId(parsed, externalId);
+
+    const text = sanitizeString(parsed.text ?? '');
+
+    const message: MessageWithParticipants = {
+      externalId,
+      messageThreadExternalId,
+      headerMessageId: parsed.messageId || externalId,
+      subject: sanitizeString(parsed.subject || ''),
+      text,
+      receivedAt: parsed.date ? new Date(parsed.date) : new Date(),
+      direction: MessageDirection.INCOMING,
+      attachments: (parsed.attachments || []).map((attachment) => ({
+        filename: attachment.filename || 'unnamed-attachment',
+      })),
+      participants: this.extractParticipants(parsed),
+    };
+
+    return {
+      parsed,
+      message,
+      originWorkspaceId: this.extractOriginWorkspaceId(parsed),
+    };
+  }
+
+  private extractThreadId(parsed: ParsedEmail, fallback: string): string {
+    if (Array.isArray(parsed.references) && parsed.references[0]?.trim()) {
+      return parsed.references[0].trim();
+    }
+
+    if (parsed.inReplyTo) {
+      const inReplyTo = String(parsed.inReplyTo).trim();
+
+      if (inReplyTo) {
+        return inReplyTo;
+      }
+    }
+
+    return fallback;
+  }
+
+  private extractParticipants(parsed: ParsedEmail) {
+    const addressFields = [
+      { field: parsed.from, role: MessageParticipantRole.FROM },
+      { field: parsed.to, role: MessageParticipantRole.TO },
+      { field: parsed.cc, role: MessageParticipantRole.CC },
+      { field: parsed.bcc, role: MessageParticipantRole.BCC },
+    ] as const;
+
+    return addressFields.flatMap(({ field, role }) =>
+      formatAddressObjectAsParticipants(this.extractAddresses(field), role),
+    );
+  }
+
+  private extractAddresses(
+    address: Address | Address[] | undefined,
+  ): EmailAddress[] {
+    if (!address) {
+      return [];
+    }
+
+    const addresses = Array.isArray(address) ? address : [address];
+
+    const mailboxes = addresses.flatMap((addr) =>
+      addr.address ? [addr] : (addr.group ?? []),
+    );
+
+    return mailboxes
+      .filter((mailbox) => !!mailbox.address)
+      .map((mailbox) => ({
+        address: mailbox.address as string,
+        name: sanitizeString(mailbox.name || ''),
+      }));
+  }
+
+  private extractOriginWorkspaceId(parsed: ParsedEmail): string | null {
+    const headers = (parsed.headers ?? []) as Array<{
+      key?: string;
+      value?: string;
+    }>;
+
+    for (const header of headers) {
+      if ((header.key ?? '').toLowerCase() === INBOUND_EMAIL_ORIGIN_HEADER) {
+        return (header.value ?? '').trim() || null;
+      }
+    }
+
+    return null;
+  }
+}

packages/twenty-server/src/modules/messaging/message-import-manager/drivers/inbound-email/services/inbound-email-storage.service.ts

@@ -0,0 +1,133 @@
+import { Injectable, Logger } from '@nestjs/common';
+
+import {
+  CopyObjectCommand,
+  DeleteObjectCommand,
+  GetObjectCommand,
+  ListObjectsV2Command,
+  type ListObjectsV2CommandOutput,
+} from '@aws-sdk/client-s3';
+import { Readable } from 'stream';
+
+import { INBOUND_EMAIL_S3_PREFIXES } from 'src/modules/messaging/message-import-manager/drivers/inbound-email/constants/inbound-email.constants';
+import { InboundEmailS3ClientProvider } from 'src/modules/messaging/message-import-manager/drivers/inbound-email/providers/inbound-email-s3-client.provider';
+
+export type InboundEmailListResult = {
+  keys: string[];
+  isTruncated: boolean;
+};
+
+@Injectable()
+export class InboundEmailStorageService {
+  private readonly logger = new Logger(InboundEmailStorageService.name);
+
+  constructor(
+    private readonly s3ClientProvider: InboundEmailS3ClientProvider,
+  ) {}
+
+  async listIncoming(maxKeys: number): Promise<InboundEmailListResult> {
+    const client = this.s3ClientProvider.getClient();
+    const bucket = this.s3ClientProvider.getBucketName();
+
+    if (!client || !bucket) {
+      return { keys: [], isTruncated: false };
+    }
+
+    const command = new ListObjectsV2Command({
+      Bucket: bucket,
+      Prefix: INBOUND_EMAIL_S3_PREFIXES.INCOMING,
+      MaxKeys: maxKeys,
+    });
+
+    const response: ListObjectsV2CommandOutput = await client.send(command);
+
+    const keys =
+      response.Contents?.map((content) => content.Key).filter(
+        (key): key is string => typeof key === 'string',
+      ) ?? [];
+
+    return {
+      keys,
+      isTruncated: response.IsTruncated === true,
+    };
+  }
+
+  async getRawMessage(key: string): Promise<Buffer> {
+    const client = this.s3ClientProvider.getClient();
+    const bucket = this.s3ClientProvider.getBucketName();
+
+    if (!client || !bucket) {
+      throw new Error('Inbound email S3 bucket is not configured');
+    }
+
+    const command = new GetObjectCommand({ Bucket: bucket, Key: key });
+    const response = await client.send(command);
+
+    if (!response.Body) {
+      throw new Error(`Empty body for S3 key ${key}`);
+    }
+
+    return this.streamToBuffer(response.Body as Readable);
+  }
+
+  async moveToProcessed(key: string): Promise<void> {
+    await this.moveKey(key, this.buildArchiveKey(key, 'PROCESSED'));
+  }
+
+  async moveToUnmatched(key: string): Promise<void> {
+    await this.moveKey(key, this.buildArchiveKey(key, 'UNMATCHED'));
+  }
+
+  async moveToFailed(key: string): Promise<void> {
+    await this.moveKey(key, this.buildArchiveKey(key, 'FAILED'));
+  }
+
+  private buildArchiveKey(
+    originalKey: string,
+    destination: 'PROCESSED' | 'UNMATCHED' | 'FAILED',
+  ): string {
+    const bare = originalKey.startsWith(INBOUND_EMAIL_S3_PREFIXES.INCOMING)
+      ? originalKey.slice(INBOUND_EMAIL_S3_PREFIXES.INCOMING.length)
+      : originalKey;
+
+    // YYYY-MM-DD date prefix so old archives can be pruned by lifecycle rules.
+    const datePrefix = new Date().toISOString().slice(0, 10);
+
+    return `${INBOUND_EMAIL_S3_PREFIXES[destination]}${datePrefix}/${bare}`;
+  }
+
+  private async moveKey(sourceKey: string, destKey: string): Promise<void> {
+    const client = this.s3ClientProvider.getClient();
+    const bucket = this.s3ClientProvider.getBucketName();
+
+    if (!client || !bucket) {
+      this.logger.warn(
+        `Cannot move inbound email S3 key ${sourceKey}: bucket not configured`,
+      );
+
+      return;
+    }
+
+    await client.send(
+      new CopyObjectCommand({
+        Bucket: bucket,
+        CopySource: `${bucket}/${sourceKey}`,
+        Key: destKey,
+      }),
+    );
+
+    await client.send(
+      new DeleteObjectCommand({ Bucket: bucket, Key: sourceKey }),
+    );
+  }
+
+  private async streamToBuffer(stream: Readable): Promise<Buffer> {
+    const chunks: Buffer[] = [];
+
+    for await (const chunk of stream) {
+      chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk));
+    }
+
+    return Buffer.concat(chunks);
+  }
+}

packages/twenty-server/src/modules/messaging/message-import-manager/drivers/inbound-email/utils/extract-envelope-recipient.util.ts

@@ -0,0 +1,82 @@
+import { type Email as ParsedEmail } from 'postal-mime';
+
+// Best-effort extraction of the envelope recipient (the address the message
+// was actually delivered to, as opposed to what appears in To/Cc). SES writes
+// the envelope recipient into the Received header's "for" clause, so we check
+// that first and fall back to scanning To/Cc/Delivered-To/X-Original-To for
+// anything at the given domain.
+export const extractEnvelopeRecipientForDomain = (
+  parsed: ParsedEmail,
+  domain: string,
+): string | null => {
+  const normalizedDomain = domain.trim().toLowerCase();
+
+  if (!normalizedDomain) {
+    return null;
+  }
+
+  const candidates: string[] = [];
+
+  const headers = (parsed.headers ?? []) as Array<{
+    key?: string;
+    value?: string;
+  }>;
+
+  for (const header of headers) {
+    const key = (header.key ?? '').toLowerCase();
+
+    if (!header.value) {
+      continue;
+    }
+
+    if (key === 'received') {
+      const match = /\bfor\s+<?([^\s<>;]+)>?/i.exec(header.value);
+
+      if (match?.[1]) {
+        candidates.push(match[1]);
+      }
+    }
+
+    if (key === 'delivered-to' || key === 'x-original-to') {
+      candidates.push(header.value);
+    }
+  }
+
+  const addressFields = [parsed.to, parsed.cc, parsed.bcc];
+
+  for (const field of addressFields) {
+    if (!field) continue;
+
+    const list = Array.isArray(field) ? field : [field];
+
+    for (const entry of list) {
+      if (entry.address) {
+        candidates.push(entry.address);
+      }
+
+      if (Array.isArray(entry.group)) {
+        for (const groupEntry of entry.group) {
+          if (groupEntry.address) {
+            candidates.push(groupEntry.address);
+          }
+        }
+      }
+    }
+  }
+
+  for (const candidate of candidates) {
+    const normalized = candidate.trim().toLowerCase();
+
+    if (normalized.endsWith(`@${normalizedDomain}`)) {
+      return normalized;
+    }
+  }
+
+  return null;
+};
+
+export const extractLocalPart = (address: string): string => {
+  const atIndex = address.indexOf('@');
+
+  return atIndex === -1 ? address : address.slice(0, atIndex);
+};

packages/twenty-server/src/modules/messaging/message-outbound-manager/services/messaging-message-outbound.service.ts

@@ -40,6 +40,7 @@ export class MessagingMessageOutboundService {
         );
       case ConnectedAccountProvider.OIDC:
       case ConnectedAccountProvider.SAML:
+      case ConnectedAccountProvider.EMAIL_FORWARDING:
         throw new Error(
           `Provider ${connectedAccount.provider} does not support sending messages`,
         );
@@ -73,6 +74,7 @@ export class MessagingMessageOutboundService {
         );
       case ConnectedAccountProvider.OIDC:
       case ConnectedAccountProvider.SAML:
+      case ConnectedAccountProvider.EMAIL_FORWARDING:
         throw new Error(
           `Provider ${connectedAccount.provider} does not support creating drafts`,
         );

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

@@ -4,4 +4,5 @@ export enum ConnectedAccountProvider {
   IMAP_SMTP_CALDAV = 'imap_smtp_caldav',
   OIDC = 'oidc',
   SAML = 'saml',
+  EMAIL_FORWARDING = 'email_forwarding',
 }

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

@@ -1,4 +1,5 @@
 export enum MessageChannelType {
   EMAIL = 'EMAIL',
   SMS = 'SMS',
+  EMAIL_FORWARDING = 'EMAIL_FORWARDING',
 }