Ses outbound followup (#20610)

This pull request unifies outbound with inbound under the new feature and the new email groups feature. These are workspace level shared inboxes that are shared between all workspace members. outbound sending with SES works, we only listen for tenant status events, rest is managed by AWS PR refactors old code and webhook to be split for outbound and inbound for proper separation | Area | Change | |---|---| | AWS SES driver | Split into `AwsSesRegisterDomainService` (tenant + identity + DKIM + MAIL FROM + configuration-set + EventBridge dest + contact list) and `AwsSesSendEmailService` (SendEmail). | | Reputation webhook | New `/webhooks/messaging/ses/outbound` route. SES → EventBridge (`Sending Status Enabled/Disabled` on default bus) → SNS → router → `SesOutboundSendingStateHandlerService` updates `emailing_domain.tenantStatus`. | | Inbound webhook | Refactored into `SesInboundWebhookRouterService` + `SesInboundMailHandlerService`. Shared `SnsSignatureVerifierService` + `SnsSubscriptionConfirmerService` across both routes. | | Global uniqueness | New migration + instance command: `emailing_domain.domain` is now globally unique (one tenant per domain across workspaces). | | Tenant status | New `emailing_domain.tenantStatus` column (`ACTIVE` / `PAUSED`) + `EmailingDomainTenantStatusService`. | | Send-email mutation | New `sendEmailViaDomain` GraphQL mutation + DTOs. | | Cleanup | `EmailingDomainWorkspaceCleanupJob` wired into `WorkspaceService.deleteWorkspace` — tears down SES tenant association + identity on workspace delete. | | Settings UI | Rewritten around reusable `SettingsTableListSection`. "Email Group" → "Email Handle" rename. New cells for status/source/forwarding. Outbound domains surfaced on workspace settings page. | ### Env vars (new) All in `config-variables.ts`, group `AWS_SES_SETTINGS`, all optional: - `AWS_SES_REGION` — `@IsAWSRegion`, consumed by `AwsSesClientProvider` + driver factory - `AWS_SES_ACCOUNT_ID` — used for ARN construction in driver factory - `SES_SNS_TOPIC_ARN_ALLOWLIST` — **shared** by inbound + outbound webhook routers, comma-separated list of accepted SNS topic ARNs (verified via `sns-payload-validator`) ### Migrations - `1778862608620-add-emailing-domain-tenant-status` (fast) — adds `tenantStatus` column. - `1778865501791-unique-emailing-domain-globally` (slow, idempotent) — enforces global uniqueness on `domain`. - Instance commands bumped to `2.5`. ### Infra dependency Two coupled twenty-infra PRs: - `ses-inbound-email` — receipt-rule + inbound SNS topic + S3 bucket policy + KMS grant + `email_group_*` outputs. - `ses-outbound-tf` — EventBridge rule + outbound SNS topic + SES IAM policy + outbound `webhook_url` subscription. **Based on `ses-inbound-email`.** Merge order: inbound first, then outbound. Outbound PR's chart edit owns the comma-joined `SES_SNS_TOPIC_ARN_ALLOWLIST` value (both ARNs). Features lives under `/settings/general` <img width="1496" height="845" alt="SCR-20260519-ofhi-2" src="https://github.com/user-attachments/assets/a025485a-09f7-4131-91cd-0067690ff18d" /> --------- Co-authored-by: Félix Malfait <felix.malfait@gmail.com> Co-authored-by: claude[bot] <41898282+claude[bot]@users.noreply.github.com> Co-authored-by: Félix Malfait <FelixMalfait@users.noreply.github.com>
2026-05-27 23:08:44 +05:30
parent 0702e72e3f
commit c5606212f2
80 changed files with 3155 additions and 1341 deletions
@@ -0,0 +1 @@
+export const AWS_SES_EVENT_BUS_NAME = 'default';
@@ -0,0 +1 @@
+export const AWS_SES_MAIL_FROM_SUBDOMAIN = 'bounce';
@@ -0,0 +1 @@
+export const AWS_SES_MARKETING_TOPIC_NAME = 'marketing';
@@ -0,0 +1 @@
+export const AWS_SES_RESOURCE_NAME_PREFIX = 'twenty-workspace';
@@ -27,11 +27,11 @@ export class AwsSesClientProvider {
        'AWS_SES_SESSION_TOKEN',
      );

-      if (accessKeyId && secretAccessKey && sessionToken) {
+      if (accessKeyId && secretAccessKey) {
        config.credentials = {
          accessKeyId,
          secretAccessKey,
-          sessionToken,
+          ...(sessionToken ? { sessionToken } : {}),
        };
      }

@@ -0,0 +1,132 @@
+import {
+  CreateConfigurationSetCommand,
+  CreateConfigurationSetEventDestinationCommand,
+  CreateContactListCommand,
+  CreateTenantResourceAssociationCommand,
+  GetConfigurationSetCommand,
+  NotFoundException,
+  PutEmailIdentityMailFromAttributesCommand,
+} from '@aws-sdk/client-sesv2';
+
+import { type AwsSesClientProvider } from 'src/engine/core-modules/emailing-domain/drivers/aws-ses/providers/aws-ses-client.provider';
+import { AwsSesRegisterDomainService } from 'src/engine/core-modules/emailing-domain/drivers/aws-ses/services/aws-ses-register-domain.service';
+import { type AwsSesDriverConfig } from 'src/engine/core-modules/emailing-domain/drivers/interfaces/driver-config.interface';
+import { EmailingDomainDriver } from 'src/engine/core-modules/emailing-domain/drivers/types/emailing-domain-driver.type';
+
+describe('AwsSesRegisterDomainService', () => {
+  const config: AwsSesDriverConfig = {
+    driver: EmailingDomainDriver.AWS_SES,
+    region: 'us-east-1',
+    accountId: '123456789012',
+  };
+
+  const provisionInput = {
+    tenantName: 'twenty-workspace-ws1',
+    configurationSetName: 'twenty-workspace-ws1',
+    contactListName: 'twenty-workspace-ws1',
+  };
+
+  const buildNotFound = () =>
+    new NotFoundException({
+      $metadata: { httpStatusCode: 404 },
+      message: 'Configuration set not found.',
+    });
+
+  const setUp = () => {
+    const send = jest.fn();
+    const clientProvider = {
+      getSESClient: () => ({ send }),
+    } as unknown as AwsSesClientProvider;
+    const service = new AwsSesRegisterDomainService(clientProvider);
+
+    return { service, send };
+  };
+
+  describe('provisionWorkspaceResources', () => {
+    it('creates every workspace-scoped resource when the configuration set does not yet exist', async () => {
+      const { service, send } = setUp();
+
+      send.mockImplementation(async (command) => {
+        if (command instanceof GetConfigurationSetCommand) {
+          throw buildNotFound();
+        }
+
+        return {};
+      });
+
+      await service.provisionWorkspaceResources(provisionInput, config);
+
+      const commandTypes = send.mock.calls.map(
+        ([command]) => command.constructor.name,
+      );
+
+      expect(commandTypes).toEqual([
+        GetConfigurationSetCommand.name,
+        CreateConfigurationSetCommand.name,
+        CreateConfigurationSetEventDestinationCommand.name,
+        CreateContactListCommand.name,
+        CreateTenantResourceAssociationCommand.name,
+      ]);
+    });
+
+    it('issues no creates when the configuration set already exists', async () => {
+      const { service, send } = setUp();
+
+      send.mockResolvedValue({});
+
+      await service.provisionWorkspaceResources(provisionInput, config);
+
+      const commandTypes = send.mock.calls.map(
+        ([command]) => command.constructor.name,
+      );
+
+      expect(commandTypes).toEqual([GetConfigurationSetCommand.name]);
+    });
+
+    it('propagates non-NotFound AWS errors raised by the existence probe', async () => {
+      const { service, send } = setUp();
+      const fatalError = new Error('Boom');
+
+      send.mockImplementation(async (command) => {
+        if (command instanceof GetConfigurationSetCommand) {
+          throw fatalError;
+        }
+
+        return {};
+      });
+
+      await expect(
+        service.provisionWorkspaceResources(provisionInput, config),
+      ).rejects.toBe(fatalError);
+    });
+  });
+
+  describe('registerDomain', () => {
+    it('configures custom MAIL FROM using the bounce subdomain', async () => {
+      const { service, send } = setUp();
+
+      send.mockResolvedValue({});
+
+      await service.registerDomain('mail.example.com');
+
+      const commandTypes = send.mock.calls.map(
+        ([command]) => command.constructor.name,
+      );
+
+      expect(commandTypes).toEqual([
+        PutEmailIdentityMailFromAttributesCommand.name,
+      ]);
+
+      const mailFromCall = send.mock.calls.find(
+        ([command]) =>
+          command instanceof PutEmailIdentityMailFromAttributesCommand,
+      );
+
+      expect(mailFromCall?.[0].input).toMatchObject({
+        EmailIdentity: 'mail.example.com',
+        MailFromDomain: 'bounce.mail.example.com',
+        BehaviorOnMxFailure: 'USE_DEFAULT_VALUE',
+      });
+    });
+  });
+});
@@ -0,0 +1,113 @@
+import { SendEmailCommand } from '@aws-sdk/client-sesv2';
+
+import { type AwsSesClientProvider } from 'src/engine/core-modules/emailing-domain/drivers/aws-ses/providers/aws-ses-client.provider';
+import { type AwsSesHandleErrorService } from 'src/engine/core-modules/emailing-domain/drivers/aws-ses/services/aws-ses-handle-error.service';
+import { AwsSesSendEmailService } from 'src/engine/core-modules/emailing-domain/drivers/aws-ses/services/aws-ses-send-email.service';
+import {
+  EmailingDomainDriverException,
+  EmailingDomainDriverExceptionCode,
+} from 'src/engine/core-modules/emailing-domain/drivers/exceptions/emailing-domain-driver.exception';
+
+describe('AwsSesSendEmailService', () => {
+  const baseInput = {
+    workspaceId: 'ws1',
+    domain: 'mail.example.com',
+    from: 'noreply@mail.example.com',
+    to: ['user@example.com'],
+    subject: 'Hello',
+    text: 'World',
+  };
+
+  const baseContext = {
+    tenantName: 'twenty-workspace-ws1',
+    configurationSetName: 'twenty-workspace-ws1',
+    contactListName: 'twenty-workspace-ws1',
+  };
+
+  const setUp = () => {
+    const send = jest.fn();
+    const clientProvider = {
+      getSESClient: () => ({ send }),
+    } as unknown as AwsSesClientProvider;
+    const handleErrorService = {
+      handleAwsSesError: jest.fn((error) => {
+        throw error;
+      }),
+    } as unknown as AwsSesHandleErrorService;
+    const service = new AwsSesSendEmailService(
+      clientProvider,
+      handleErrorService,
+    );
+
+    return { service, send, handleErrorService };
+  };
+
+  it('should call SendEmail with tenant, config set, and list management options', async () => {
+    const { service, send } = setUp();
+
+    send.mockResolvedValue({ MessageId: 'msg-1' });
+
+    const result = await service.sendEmail(baseInput, baseContext);
+
+    expect(result.messageId).toBe('msg-1');
+
+    const [command] = send.mock.calls[0];
+
+    expect(command).toBeInstanceOf(SendEmailCommand);
+    expect(command.input).toMatchObject({
+      FromEmailAddress: 'noreply@mail.example.com',
+      Destination: { ToAddresses: ['user@example.com'] },
+      ConfigurationSetName: 'twenty-workspace-ws1',
+      TenantName: 'twenty-workspace-ws1',
+      ListManagementOptions: {
+        ContactListName: 'twenty-workspace-ws1',
+        TopicName: 'marketing',
+      },
+    });
+    expect(command.input.EmailTags).toEqual(
+      expect.arrayContaining([
+        { Name: 'workspace', Value: 'ws1' },
+        { Name: 'domain', Value: 'mail.example.com' },
+      ]),
+    );
+  });
+
+  it('should throw when SES returns no MessageId', async () => {
+    const { service, send } = setUp();
+
+    send.mockResolvedValue({});
+
+    await expect(service.sendEmail(baseInput, baseContext)).rejects.toThrow(
+      EmailingDomainDriverException,
+    );
+  });
+
+  it('should reject empty recipient list before calling SES', async () => {
+    const { service, send } = setUp();
+
+    await expect(
+      service.sendEmail({ ...baseInput, to: [] }, baseContext),
+    ).rejects.toMatchObject({
+      code: EmailingDomainDriverExceptionCode.CONFIGURATION_ERROR,
+    });
+    expect(send).not.toHaveBeenCalled();
+  });
+
+  it('should route AWS errors through the error handler', async () => {
+    const { service, send, handleErrorService } = setUp();
+    const awsError = Object.assign(new Error('Rejected'), {
+      name: 'MessageRejected',
+      $metadata: { httpStatusCode: 400 },
+    });
+
+    send.mockRejectedValue(awsError);
+
+    await expect(service.sendEmail(baseInput, baseContext)).rejects.toBe(
+      awsError,
+    );
+    expect(handleErrorService.handleAwsSesError).toHaveBeenCalledWith(
+      awsError,
+      'sendEmail',
+    );
+  });
+});
@@ -1,24 +1,37 @@
 import { Logger } from '@nestjs/common';

 import {
+  AlreadyExistsException,
  CreateEmailIdentityCommand,
  CreateTenantCommand,
  CreateTenantResourceAssociationCommand,
+  DeleteConfigurationSetCommand,
+  DeleteContactListCommand,
+  DeleteEmailIdentityCommand,
+  DeleteTenantCommand,
+  DeleteTenantResourceAssociationCommand,
  GetEmailIdentityCommand,
+  NotFoundException,
  PutEmailIdentityDkimAttributesCommand,
 } from '@aws-sdk/client-sesv2';

 import { type AwsSesDriverConfig } from 'src/engine/core-modules/emailing-domain/drivers/interfaces/driver-config.interface';
 import {
-  type DomainStatusInput,
-  type DomainVerificationInput,
  type EmailingDomainDriverInterface,
+  type EmailingDomainResourceInput,
  type EmailingDomainVerificationResult,
 } from 'src/engine/core-modules/emailing-domain/drivers/interfaces/emailing-domain-driver.interface';
+import {
+  type EmailingDomainSendEmailInput,
+  type EmailingDomainSendEmailResult,
+} from 'src/engine/core-modules/emailing-domain/drivers/types/send-email';

+import { AWS_SES_RESOURCE_NAME_PREFIX } from 'src/engine/core-modules/emailing-domain/drivers/aws-ses/constants/aws-ses-resource-name-prefix.constant';
 import { type AwsSesClientProvider } from 'src/engine/core-modules/emailing-domain/drivers/aws-ses/providers/aws-ses-client.provider';
+import { AwsSesRegisterDomainService } from 'src/engine/core-modules/emailing-domain/drivers/aws-ses/services/aws-ses-register-domain.service';
 import { type AwsSesHandleErrorService } from 'src/engine/core-modules/emailing-domain/drivers/aws-ses/services/aws-ses-handle-error.service';
-import { EmailingDomainStatus } from 'src/engine/core-modules/emailing-domain/drivers/types/emailing-domain';
+import { type AwsSesSendEmailService } from 'src/engine/core-modules/emailing-domain/drivers/aws-ses/services/aws-ses-send-email.service';
+import { EmailingDomainStatus } from 'src/engine/core-modules/emailing-domain/drivers/types/emailing-domain-status.type';
 import { type VerificationRecordDTO } from 'src/engine/core-modules/emailing-domain/dtos/verification-record.dto';

 export class AwsSesDriver implements EmailingDomainDriverInterface {
@@ -28,17 +41,17 @@ export class AwsSesDriver implements EmailingDomainDriverInterface {
    private readonly config: AwsSesDriverConfig,
    private readonly awsSesClientProvider: AwsSesClientProvider,
    private readonly awsSesHandleErrorService: AwsSesHandleErrorService,
+    private readonly awsSesRegisterDomainService: AwsSesRegisterDomainService,
+    private readonly awsSesSendEmailService: AwsSesSendEmailService,
  ) {}

  async verifyDomain(
-    input: DomainVerificationInput,
+    input: EmailingDomainResourceInput,
  ): Promise<EmailingDomainVerificationResult> {
    try {
      this.logger.log(`Starting domain verification for: ${input.domain}`);

-      const tenantName = this.generateTenantName(input.workspaceId);
-
-      await this.ensureTenantExists(tenantName);
+      const tenantName = this.buildTenantName(input.workspaceId);

      const { isVerified, verificationRecords } =
        await this.createOrUpdateEmailIdentity(input.domain, tenantName);
@@ -51,7 +64,6 @@ export class AwsSesDriver implements EmailingDomainDriverInterface {
        status: isVerified
          ? EmailingDomainStatus.VERIFIED
          : EmailingDomainStatus.PENDING,
-        verifiedAt: isVerified ? new Date() : null,
        verificationRecords,
      };
    } catch (error) {
@@ -61,7 +73,7 @@ export class AwsSesDriver implements EmailingDomainDriverInterface {
  }

  async getDomainStatus(
-    input: DomainStatusInput,
+    input: EmailingDomainResourceInput,
  ): Promise<EmailingDomainVerificationResult> {
    try {
      this.logger.log(`Getting domain status for: ${input.domain}`);
@@ -75,7 +87,6 @@ export class AwsSesDriver implements EmailingDomainDriverInterface {
      const identityResponse = await sesClient.send(getIdentityCommand);

      const status = this.determineVerificationStatus(identityResponse);
-      const isFullyVerified = status === EmailingDomainStatus.VERIFIED;
      const verificationRecords = this.buildVerificationRecords(
        input.domain,
        identityResponse.DkimAttributes?.Tokens || [],
@@ -83,14 +94,12 @@ export class AwsSesDriver implements EmailingDomainDriverInterface {

      return {
        status,
-        verifiedAt: isFullyVerified ? new Date() : null,
        verificationRecords,
      };
    } catch (error) {
-      if (error.name === 'NotFoundException') {
+      if (error instanceof NotFoundException) {
        return {
          status: EmailingDomainStatus.FAILED,
-          verifiedAt: null,
          verificationRecords: [],
        };
      }
@@ -102,8 +111,109 @@ export class AwsSesDriver implements EmailingDomainDriverInterface {
    }
  }

-  private generateTenantName(workspaceId: string): string {
-    return `twenty-workspace-${workspaceId}`;
+  async provisionWorkspace(workspaceId: string): Promise<void> {
+    const tenantName = this.buildTenantName(workspaceId);
+
+    await this.ensureTenantExists(tenantName);
+
+    await this.awsSesRegisterDomainService.provisionWorkspaceResources(
+      {
+        tenantName,
+        configurationSetName: this.buildConfigurationSetName(workspaceId),
+        contactListName: this.buildContactListName(workspaceId),
+      },
+      this.config,
+    );
+  }
+
+  async registerDomain(input: EmailingDomainResourceInput): Promise<void> {
+    await this.awsSesRegisterDomainService.registerDomain(input.domain);
+  }
+
+  async sendEmail(
+    input: EmailingDomainSendEmailInput,
+  ): Promise<EmailingDomainSendEmailResult> {
+    return this.awsSesSendEmailService.sendEmail(input, {
+      tenantName: this.buildTenantName(input.workspaceId),
+      configurationSetName: this.buildConfigurationSetName(input.workspaceId),
+      contactListName: this.buildContactListName(input.workspaceId),
+    });
+  }
+
+  async cleanupDomain(input: EmailingDomainResourceInput): Promise<void> {
+    const sesClient = this.awsSesClientProvider.getSESClient();
+    const tenantName = this.buildTenantName(input.workspaceId);
+    const identityArn = `arn:aws:ses:${this.config.region}:${this.config.accountId}:identity/${input.domain}`;
+
+    await sesClient
+      .send(
+        new DeleteTenantResourceAssociationCommand({
+          TenantName: tenantName,
+          ResourceArn: identityArn,
+        }),
+      )
+      .catch((error) => {
+        if (!(error instanceof NotFoundException)) throw error;
+      });
+
+    await sesClient
+      .send(new DeleteEmailIdentityCommand({ EmailIdentity: input.domain }))
+      .catch((error) => {
+        if (!(error instanceof NotFoundException)) throw error;
+      });
+  }
+
+  async deprovisionWorkspace(workspaceId: string): Promise<void> {
+    const sesClient = this.awsSesClientProvider.getSESClient();
+    const tenantName = this.buildTenantName(workspaceId);
+    const configurationSetName = this.buildConfigurationSetName(workspaceId);
+    const contactListName = this.buildContactListName(workspaceId);
+    const configurationSetArn = `arn:aws:ses:${this.config.region}:${this.config.accountId}:configuration-set/${configurationSetName}`;
+
+    await sesClient
+      .send(
+        new DeleteTenantResourceAssociationCommand({
+          TenantName: tenantName,
+          ResourceArn: configurationSetArn,
+        }),
+      )
+      .catch((error) => {
+        if (!(error instanceof NotFoundException)) throw error;
+      });
+
+    await sesClient
+      .send(
+        new DeleteConfigurationSetCommand({
+          ConfigurationSetName: configurationSetName,
+        }),
+      )
+      .catch((error) => {
+        if (!(error instanceof NotFoundException)) throw error;
+      });
+
+    await sesClient
+      .send(new DeleteContactListCommand({ ContactListName: contactListName }))
+      .catch((error) => {
+        if (!(error instanceof NotFoundException)) throw error;
+      });
+
+    await sesClient
+      .send(new DeleteTenantCommand({ TenantName: tenantName }))
+      .catch((error) => {
+        if (!(error instanceof NotFoundException)) throw error;
+      });
+  }
+
+  private buildTenantName(workspaceId: string): string {
+    return `${AWS_SES_RESOURCE_NAME_PREFIX}-${workspaceId}`;
+  }
+
+  private buildConfigurationSetName(workspaceId: string): string {
+    return `${AWS_SES_RESOURCE_NAME_PREFIX}-${workspaceId}`;
+  }
+
+  private buildContactListName(workspaceId: string): string {
+    return `${AWS_SES_RESOURCE_NAME_PREFIX}-${workspaceId}`;
  }

  private async ensureTenantExists(tenantName: string): Promise<void> {
@@ -113,7 +223,7 @@ export class AwsSesDriver implements EmailingDomainDriverInterface {
      await sesClient.send(new CreateTenantCommand({ TenantName: tenantName }));
      this.logger.log(`Created tenant: ${tenantName}`);
    } catch (error) {
-      if (error.name === 'AlreadyExistsException') {
+      if (error instanceof AlreadyExistsException) {
        this.logger.log(`Tenant already exists: ${tenantName}`);

        return;
@@ -143,13 +253,11 @@ export class AwsSesDriver implements EmailingDomainDriverInterface {
        existingIdentity.DkimAttributes?.Tokens || [],
      );

-      if (!isVerified) {
-        await this.associateResourceWithTenant(domain, tenantName);
-      }
+      await this.associateResourceWithTenant(domain, tenantName);

      return { isVerified, verificationRecords };
    } catch (error) {
-      if (error.name === 'NotFoundException') {
+      if (error instanceof NotFoundException) {
        return await this.createNewEmailIdentity(domain, tenantName);
      }
      throw error;
@@ -201,7 +309,7 @@ export class AwsSesDriver implements EmailingDomainDriverInterface {
      );
      this.logger.log(`Associated domain ${domain} with tenant ${tenantName}`);
    } catch (error) {
-      if (error.name === 'AlreadyExistsException') {
+      if (error instanceof AlreadyExistsException) {
        this.logger.log(
          `Domain ${domain} already associated with tenant ${tenantName}`,
        );
@@ -0,0 +1,140 @@
+import { Injectable, Logger } from '@nestjs/common';
+
+import {
+  CreateConfigurationSetCommand,
+  CreateConfigurationSetEventDestinationCommand,
+  CreateContactListCommand,
+  CreateTenantResourceAssociationCommand,
+  GetConfigurationSetCommand,
+  NotFoundException,
+  PutEmailIdentityMailFromAttributesCommand,
+  type SESv2Client,
+} from '@aws-sdk/client-sesv2';
+import { type AwsSesDriverConfig } from 'src/engine/core-modules/emailing-domain/drivers/interfaces/driver-config.interface';
+
+import { AWS_SES_EVENT_BUS_NAME } from 'src/engine/core-modules/emailing-domain/drivers/aws-ses/constants/aws-ses-event-bus-name.constant';
+import { AWS_SES_MAIL_FROM_SUBDOMAIN } from 'src/engine/core-modules/emailing-domain/drivers/aws-ses/constants/aws-ses-mail-from-subdomain.constant';
+import { AWS_SES_MARKETING_TOPIC_NAME } from 'src/engine/core-modules/emailing-domain/drivers/aws-ses/constants/aws-ses-marketing-topic-name.constant';
+import { AwsSesClientProvider } from 'src/engine/core-modules/emailing-domain/drivers/aws-ses/providers/aws-ses-client.provider';
+
+type ProvisionWorkspaceInput = {
+  tenantName: string;
+  configurationSetName: string;
+  contactListName: string;
+};
+
+@Injectable()
+export class AwsSesRegisterDomainService {
+  private readonly logger = new Logger(AwsSesRegisterDomainService.name);
+
+  constructor(private readonly awsSesClientProvider: AwsSesClientProvider) {}
+
+  async provisionWorkspaceResources(
+    input: ProvisionWorkspaceInput,
+    config: AwsSesDriverConfig,
+  ): Promise<void> {
+    const sesClient = this.awsSesClientProvider.getSESClient();
+
+    const isAlreadyProvisioned = await this.isWorkspaceProvisioned(
+      sesClient,
+      input.configurationSetName,
+    );
+
+    if (isAlreadyProvisioned) {
+      return;
+    }
+
+    const eventBusArn = `arn:aws:events:${config.region}:${config.accountId}:event-bus/${AWS_SES_EVENT_BUS_NAME}`;
+    const configurationSetArn = `arn:aws:ses:${config.region}:${config.accountId}:configuration-set/${input.configurationSetName}`;
+
+    await sesClient.send(
+      new CreateConfigurationSetCommand({
+        ConfigurationSetName: input.configurationSetName,
+        ReputationOptions: { ReputationMetricsEnabled: true },
+        SendingOptions: { SendingEnabled: true },
+        SuppressionOptions: { SuppressedReasons: ['BOUNCE', 'COMPLAINT'] },
+        Tags: [{ Key: 'managed-by', Value: 'twenty' }],
+      }),
+    );
+
+    await sesClient.send(
+      new CreateConfigurationSetEventDestinationCommand({
+        ConfigurationSetName: input.configurationSetName,
+        EventDestinationName: 'twenty-eventbridge',
+        EventDestination: {
+          Enabled: true,
+          MatchingEventTypes: [
+            'SEND',
+            'DELIVERY',
+            'BOUNCE',
+            'COMPLAINT',
+            'REJECT',
+            'RENDERING_FAILURE',
+            'DELIVERY_DELAY',
+            'SUBSCRIPTION',
+          ],
+          EventBridgeDestination: { EventBusArn: eventBusArn },
+        },
+      }),
+    );
+
+    await sesClient.send(
+      new CreateContactListCommand({
+        ContactListName: input.contactListName,
+        Topics: [
+          {
+            TopicName: AWS_SES_MARKETING_TOPIC_NAME,
+            DisplayName: 'Marketing',
+            DefaultSubscriptionStatus: 'OPT_IN',
+          },
+        ],
+        Tags: [{ Key: 'managed-by', Value: 'twenty' }],
+      }),
+    );
+
+    await sesClient.send(
+      new CreateTenantResourceAssociationCommand({
+        TenantName: input.tenantName,
+        ResourceArn: configurationSetArn,
+      }),
+    );
+
+    this.logger.log(
+      `Provisioned workspace resources for tenant ${input.tenantName}`,
+    );
+  }
+
+  async registerDomain(domain: string): Promise<void> {
+    const sesClient = this.awsSesClientProvider.getSESClient();
+
+    await sesClient.send(
+      new PutEmailIdentityMailFromAttributesCommand({
+        EmailIdentity: domain,
+        MailFromDomain: `${AWS_SES_MAIL_FROM_SUBDOMAIN}.${domain}`,
+        BehaviorOnMxFailure: 'USE_DEFAULT_VALUE',
+      }),
+    );
+
+    this.logger.log(`Registered MAIL FROM for domain ${domain}`);
+  }
+
+  private async isWorkspaceProvisioned(
+    sesClient: SESv2Client,
+    configurationSetName: string,
+  ): Promise<boolean> {
+    try {
+      await sesClient.send(
+        new GetConfigurationSetCommand({
+          ConfigurationSetName: configurationSetName,
+        }),
+      );
+
+      return true;
+    } catch (error) {
+      if (error instanceof NotFoundException) {
+        return false;
+      }
+      throw error;
+    }
+  }
+}
@@ -0,0 +1,108 @@
+import { Injectable, Logger } from '@nestjs/common';
+
+import { SendEmailCommand } from '@aws-sdk/client-sesv2';
+import { isDefined, isNonEmptyArray } from 'twenty-shared/utils';
+
+import {
+  type EmailingDomainSendEmailInput,
+  type EmailingDomainSendEmailResult,
+} from 'src/engine/core-modules/emailing-domain/drivers/types/send-email';
+
+import { AWS_SES_MARKETING_TOPIC_NAME } from 'src/engine/core-modules/emailing-domain/drivers/aws-ses/constants/aws-ses-marketing-topic-name.constant';
+import { AwsSesClientProvider } from 'src/engine/core-modules/emailing-domain/drivers/aws-ses/providers/aws-ses-client.provider';
+import { AwsSesHandleErrorService } from 'src/engine/core-modules/emailing-domain/drivers/aws-ses/services/aws-ses-handle-error.service';
+import {
+  EmailingDomainDriverException,
+  EmailingDomainDriverExceptionCode,
+} from 'src/engine/core-modules/emailing-domain/drivers/exceptions/emailing-domain-driver.exception';
+
+type SendEmailContext = {
+  tenantName: string;
+  configurationSetName: string;
+  contactListName: string;
+};
+
+@Injectable()
+export class AwsSesSendEmailService {
+  private readonly logger = new Logger(AwsSesSendEmailService.name);
+
+  constructor(
+    private readonly awsSesClientProvider: AwsSesClientProvider,
+    private readonly awsSesHandleErrorService: AwsSesHandleErrorService,
+  ) {}
+
+  async sendEmail(
+    input: EmailingDomainSendEmailInput,
+    context: SendEmailContext,
+  ): Promise<EmailingDomainSendEmailResult> {
+    if (!isNonEmptyArray(input.to)) {
+      throw new EmailingDomainDriverException(
+        'sendEmail requires at least one recipient',
+        EmailingDomainDriverExceptionCode.CONFIGURATION_ERROR,
+      );
+    }
+
+    try {
+      const sesClient = this.awsSesClientProvider.getSESClient();
+
+      const response = await sesClient.send(
+        new SendEmailCommand({
+          FromEmailAddress: input.from,
+          Destination: {
+            ToAddresses: input.to,
+            CcAddresses: input.cc,
+            BccAddresses: input.bcc,
+          },
+          ReplyToAddresses: input.replyTo,
+          Content: {
+            Simple: {
+              Subject: { Data: input.subject, Charset: 'UTF-8' },
+              Body: {
+                Text: { Data: input.text, Charset: 'UTF-8' },
+                Html: isDefined(input.html)
+                  ? { Data: input.html, Charset: 'UTF-8' }
+                  : undefined,
+              },
+              Attachments: isNonEmptyArray(input.attachments)
+                ? input.attachments.map((attachment) => ({
+                    FileName: attachment.filename,
+                    RawContent: attachment.content,
+                    ContentType: attachment.contentType,
+                    ContentDisposition: 'ATTACHMENT',
+                  }))
+                : undefined,
+            },
+          },
+          ConfigurationSetName: context.configurationSetName,
+          TenantName: context.tenantName,
+          ListManagementOptions: {
+            ContactListName: context.contactListName,
+            TopicName: AWS_SES_MARKETING_TOPIC_NAME,
+          },
+          EmailTags: [
+            { Name: 'workspace', Value: input.workspaceId },
+            { Name: 'domain', Value: input.domain },
+          ],
+        }),
+      );
+
+      if (!isDefined(response.MessageId)) {
+        throw new EmailingDomainDriverException(
+          'SES returned no MessageId',
+          EmailingDomainDriverExceptionCode.UNKNOWN,
+        );
+      }
+
+      this.logger.log(
+        `Sent email ${response.MessageId} from ${input.from} (tenant ${context.tenantName})`,
+      );
+
+      return { messageId: response.MessageId };
+    } catch (error) {
+      if (error instanceof EmailingDomainDriverException) {
+        throw error;
+      }
+      this.awsSesHandleErrorService.handleAwsSesError(error, 'sendEmail');
+    }
+  }
+}
@@ -4,9 +4,11 @@ import { type AwsSesDriverConfig } from 'src/engine/core-modules/emailing-domain
 import { type EmailingDomainDriverInterface } from 'src/engine/core-modules/emailing-domain/drivers/interfaces/emailing-domain-driver.interface';

 import { AwsSesClientProvider } from 'src/engine/core-modules/emailing-domain/drivers/aws-ses/providers/aws-ses-client.provider';
+import { AwsSesRegisterDomainService } from 'src/engine/core-modules/emailing-domain/drivers/aws-ses/services/aws-ses-register-domain.service';
 import { AwsSesDriver } from 'src/engine/core-modules/emailing-domain/drivers/aws-ses/services/aws-ses-driver.service';
 import { AwsSesHandleErrorService } from 'src/engine/core-modules/emailing-domain/drivers/aws-ses/services/aws-ses-handle-error.service';
-import { EmailingDomainDriver } from 'src/engine/core-modules/emailing-domain/drivers/types/emailing-domain';
+import { AwsSesSendEmailService } from 'src/engine/core-modules/emailing-domain/drivers/aws-ses/services/aws-ses-send-email.service';
+import { EmailingDomainDriver } from 'src/engine/core-modules/emailing-domain/drivers/types/emailing-domain-driver.type';
 import { DriverFactoryBase } from 'src/engine/core-modules/twenty-config/dynamic-factory.base';
 import { ConfigVariablesGroup } from 'src/engine/core-modules/twenty-config/enums/config-variables-group.enum';
 import { ConfigGroupHashService } from 'src/engine/core-modules/twenty-config/services/config-group-hash.service';
@@ -19,6 +21,8 @@ export class EmailingDomainDriverFactory extends DriverFactoryBase<EmailingDomai
    configGroupHashService: ConfigGroupHashService,
    private readonly awsSesClientProvider: AwsSesClientProvider,
    private readonly awsSesHandleErrorService: AwsSesHandleErrorService,
+    private readonly awsSesRegisterDomainService: AwsSesRegisterDomainService,
+    private readonly awsSesSendEmailService: AwsSesSendEmailService,
  ) {
    super(twentyConfigService, configGroupHashService);
  }
@@ -67,6 +71,8 @@ export class EmailingDomainDriverFactory extends DriverFactoryBase<EmailingDomai
          awsConfig,
          this.awsSesClientProvider,
          this.awsSesHandleErrorService,
+          this.awsSesRegisterDomainService,
+          this.awsSesSendEmailService,
        );
      }

@@ -10,6 +10,7 @@ export enum EmailingDomainDriverExceptionCode {
  TEMPORARY_ERROR = 'TEMPORARY_ERROR',
  INSUFFICIENT_PERMISSIONS = 'INSUFFICIENT_PERMISSIONS',
  CONFIGURATION_ERROR = 'CONFIGURATION_ERROR',
+  SENDING_SUSPENDED = 'SENDING_SUSPENDED',
  UNKNOWN = 'UNKNOWN',
 }

@@ -23,6 +24,8 @@ const getEmailingDomainDriverExceptionUserFriendlyMessage = (
      return msg`Insufficient permissions for email domain.`;
    case EmailingDomainDriverExceptionCode.CONFIGURATION_ERROR:
      return msg`Email domain configuration error.`;
+    case EmailingDomainDriverExceptionCode.SENDING_SUSPENDED:
+      return msg`Sending is currently suspended for this email domain.`;
    case EmailingDomainDriverExceptionCode.TEMPORARY_ERROR:
    case EmailingDomainDriverExceptionCode.UNKNOWN:
      return STANDARD_ERROR_MESSAGE;
@@ -1,4 +1,4 @@
-import { type EmailingDomainDriver } from 'src/engine/core-modules/emailing-domain/drivers/types/emailing-domain';
+import { type EmailingDomainDriver } from 'src/engine/core-modules/emailing-domain/drivers/types/emailing-domain-driver.type';

 export interface BaseDriverConfig {
  driver: EmailingDomainDriver;
@@ -1,12 +1,11 @@
-import { type EmailingDomainStatus } from 'src/engine/core-modules/emailing-domain/drivers/types/emailing-domain';
+import {
+  type EmailingDomainSendEmailInput,
+  type EmailingDomainSendEmailResult,
+} from 'src/engine/core-modules/emailing-domain/drivers/types/send-email';
+import { type EmailingDomainStatus } from 'src/engine/core-modules/emailing-domain/drivers/types/emailing-domain-status.type';
 import { type VerificationRecord } from 'src/engine/core-modules/emailing-domain/drivers/types/verifications-record';

-export type DomainVerificationInput = {
-  domain: string;
-  workspaceId: string;
-};
-
-export type DomainStatusInput = {
+export type EmailingDomainResourceInput = {
  domain: string;
  workspaceId: string;
 };
@@ -14,14 +13,20 @@ export type DomainStatusInput = {
 export type EmailingDomainVerificationResult = {
  status: EmailingDomainStatus;
  verificationRecords: VerificationRecord[];
-  verifiedAt: Date | null;
 };

 export interface EmailingDomainDriverInterface {
+  provisionWorkspace(workspaceId: string): Promise<void>;
+  deprovisionWorkspace(workspaceId: string): Promise<void>;
  verifyDomain(
-    input: DomainVerificationInput,
+    input: EmailingDomainResourceInput,
  ): Promise<EmailingDomainVerificationResult>;
  getDomainStatus(
-    input: DomainStatusInput,
+    input: EmailingDomainResourceInput,
  ): Promise<EmailingDomainVerificationResult>;
+  registerDomain(input: EmailingDomainResourceInput): Promise<void>;
+  cleanupDomain(input: EmailingDomainResourceInput): Promise<void>;
+  sendEmail(
+    input: EmailingDomainSendEmailInput,
+  ): Promise<EmailingDomainSendEmailResult>;
 }
@@ -0,0 +1,3 @@
+export enum EmailingDomainDriver {
+  AWS_SES = 'AWS_SES',
+}
@@ -1,7 +1,3 @@
-export enum EmailingDomainDriver {
-  AWS_SES = 'AWS_SES',
-}
-
 export enum EmailingDomainStatus {
  PENDING = 'PENDING',
  VERIFIED = 'VERIFIED',
@@ -0,0 +1,5 @@
+export enum EmailingDomainTenantStatus {
+  ACTIVE = 'ACTIVE',
+  PAUSED = 'PAUSED',
+  PERMANENTLY_SUSPENDED = 'PERMANENTLY_SUSPENDED',
+}
@@ -0,0 +1,26 @@
+export type EmailingDomainAttachment = {
+  filename: string;
+  content: Buffer;
+  contentType: string;
+};
+
+export type EmailingDomainEmailContent = {
+  from: string;
+  to: string[];
+  cc?: string[];
+  bcc?: string[];
+  subject: string;
+  text: string;
+  html?: string;
+  replyTo?: string[];
+  attachments?: EmailingDomainAttachment[];
+};
+
+export type EmailingDomainSendEmailInput = EmailingDomainEmailContent & {
+  workspaceId: string;
+  domain: string;
+};
+
+export type EmailingDomainSendEmailResult = {
+  messageId: string;
+};
@@ -3,10 +3,8 @@ import { Field, ObjectType, registerEnumType } from '@nestjs/graphql';
 import { IDField } from '@ptc-org/nestjs-query-graphql';

 import { UUIDScalarType } from 'src/engine/api/graphql/workspace-schema-builder/graphql-types/scalars';
-import {
-  EmailingDomainDriver,
-  EmailingDomainStatus,
-} from 'src/engine/core-modules/emailing-domain/drivers/types/emailing-domain';
+import { EmailingDomainDriver } from 'src/engine/core-modules/emailing-domain/drivers/types/emailing-domain-driver.type';
+import { EmailingDomainStatus } from 'src/engine/core-modules/emailing-domain/drivers/types/emailing-domain-status.type';
 import { VerificationRecordDTO } from 'src/engine/core-modules/emailing-domain/dtos/verification-record.dto';

 registerEnumType(EmailingDomainDriver, {
@@ -0,0 +1,7 @@
+import { Field, ObjectType } from '@nestjs/graphql';
+
+@ObjectType('SendEmailViaDomainOutput')
+export class SendEmailViaDomainOutputDTO {
+  @Field(() => String)
+  messageId: string;
+}
@@ -0,0 +1,59 @@
+import { Field, InputType } from '@nestjs/graphql';
+
+import {
+  ArrayMinSize,
+  IsArray,
+  IsEmail,
+  IsOptional,
+  IsString,
+  MinLength,
+} from 'class-validator';
+
+@InputType()
+export class SendEmailViaDomainInput {
+  @Field(() => String)
+  @IsString()
+  emailingDomainId: string;
+
+  @Field(() => [String])
+  @IsArray()
+  @ArrayMinSize(1)
+  @IsEmail({}, { each: true })
+  to: string[];
+
+  @Field(() => [String], { nullable: true })
+  @IsOptional()
+  @IsArray()
+  @IsEmail({}, { each: true })
+  cc?: string[];
+
+  @Field(() => [String], { nullable: true })
+  @IsOptional()
+  @IsArray()
+  @IsEmail({}, { each: true })
+  bcc?: string[];
+
+  @Field(() => String)
+  @IsString()
+  @MinLength(1)
+  subject: string;
+
+  @Field(() => String)
+  @IsString()
+  text: string;
+
+  @Field(() => String, { nullable: true })
+  @IsOptional()
+  @IsString()
+  html?: string;
+
+  @Field(() => String)
+  @IsEmail()
+  from: string;
+
+  @Field(() => [String], { nullable: true })
+  @IsOptional()
+  @IsArray()
+  @IsEmail({}, { each: true })
+  replyTo?: string[];
+}
@@ -9,19 +9,15 @@ import {
  UpdateDateColumn,
 } from 'typeorm';

-import {
-  EmailingDomainDriver,
-  EmailingDomainStatus,
-} from 'src/engine/core-modules/emailing-domain/drivers/types/emailing-domain';
+import { EmailingDomainDriver } from 'src/engine/core-modules/emailing-domain/drivers/types/emailing-domain-driver.type';
+import { EmailingDomainStatus } from 'src/engine/core-modules/emailing-domain/drivers/types/emailing-domain-status.type';
+import { EmailingDomainTenantStatus } from 'src/engine/core-modules/emailing-domain/drivers/types/emailing-domain-tenant-status.type';
 import { VerificationRecord } from 'src/engine/core-modules/emailing-domain/drivers/types/verifications-record';
 import { WorkspaceRelatedEntity } from 'src/engine/workspace-manager/types/workspace-related-entity';

@Entity({ name: 'emailingDomain', schema: 'core' })
@ObjectType('EmailingDomain')
-@Unique('IDX_EMAILING_DOMAIN_DOMAIN_WORKSPACE_ID_UNIQUE', [
-  'domain',
-  'workspaceId',
-])
+@Unique('IDX_EMAILING_DOMAIN_DOMAIN_UNIQUE', ['domain'])
 export class EmailingDomainEntity extends WorkspaceRelatedEntity {
  @PrimaryGeneratedColumn('uuid')
  id: string;
@@ -55,4 +51,12 @@ export class EmailingDomainEntity extends WorkspaceRelatedEntity {

  @Column({ type: 'timestamptz', nullable: true })
  verifiedAt: Date | null;
+
+  @Column({
+    type: 'enum',
+    enum: Object.values(EmailingDomainTenantStatus),
+    default: EmailingDomainTenantStatus.ACTIVE,
+    nullable: false,
+  })
+  tenantStatus: EmailingDomainTenantStatus;
 }
@@ -4,26 +4,36 @@ import { NestjsQueryTypeOrmModule } from '@ptc-org/nestjs-query-typeorm';

 import { TypeORMModule } from 'src/database/typeorm/typeorm.module';
 import { AwsSesClientProvider } from 'src/engine/core-modules/emailing-domain/drivers/aws-ses/providers/aws-ses-client.provider';
+import { AwsSesRegisterDomainService } from 'src/engine/core-modules/emailing-domain/drivers/aws-ses/services/aws-ses-register-domain.service';
 import { AwsSesHandleErrorService } from 'src/engine/core-modules/emailing-domain/drivers/aws-ses/services/aws-ses-handle-error.service';
+import { AwsSesSendEmailService } from 'src/engine/core-modules/emailing-domain/drivers/aws-ses/services/aws-ses-send-email.service';
 import { EmailingDomainDriverFactory } from 'src/engine/core-modules/emailing-domain/drivers/emailing-domain-driver.factory';
 import { EmailingDomainEntity } from 'src/engine/core-modules/emailing-domain/emailing-domain.entity';
 import { EmailingDomainResolver } from 'src/engine/core-modules/emailing-domain/emailing-domain.resolver';
+import { EmailingDomainWorkspaceCleanupJob } from 'src/engine/core-modules/emailing-domain/jobs/emailing-domain-workspace-cleanup.job';
+import { EmailingDomainTenantStatusService } from 'src/engine/core-modules/emailing-domain/services/emailing-domain-tenant-status.service';
 import { EmailingDomainService } from 'src/engine/core-modules/emailing-domain/services/emailing-domain.service';
+import { FeatureFlagModule } from 'src/engine/core-modules/feature-flag/feature-flag.module';
 import { PermissionsModule } from 'src/engine/metadata-modules/permissions/permissions.module';
 import { provideWorkspaceScopedRepository } from 'src/engine/twenty-orm/workspace-scoped-repository/provide-workspace-scoped-repository';
@Module({
  imports: [
    TypeORMModule,
    NestjsQueryTypeOrmModule.forFeature([EmailingDomainEntity]),
+    FeatureFlagModule,
    PermissionsModule,
  ],
-  exports: [EmailingDomainService],
+  exports: [EmailingDomainService, EmailingDomainTenantStatusService],
  providers: [
    EmailingDomainService,
+    EmailingDomainTenantStatusService,
    EmailingDomainResolver,
    EmailingDomainDriverFactory,
+    EmailingDomainWorkspaceCleanupJob,
    AwsSesClientProvider,
    AwsSesHandleErrorService,
+    AwsSesRegisterDomainService,
+    AwsSesSendEmailService,
    provideWorkspaceScopedRepository(EmailingDomainEntity),
  ],
 })
@@ -2,19 +2,27 @@ import { UseGuards, UsePipes } from '@nestjs/common';
 import { Args, Mutation, Query } from '@nestjs/graphql';

 import { PermissionFlagType } from 'twenty-shared/constants';
+import { FeatureFlagKey } from 'twenty-shared/types';

 import { MetadataResolver } from 'src/engine/api/graphql/graphql-config/decorators/metadata-resolver.decorator';
-import { EmailingDomainDriver } from 'src/engine/core-modules/emailing-domain/drivers/types/emailing-domain';
+import { EmailingDomainDriver } from 'src/engine/core-modules/emailing-domain/drivers/types/emailing-domain-driver.type';
 import { EmailingDomainDTO } from 'src/engine/core-modules/emailing-domain/dtos/emailing-domain.dto';
+import { SendEmailViaDomainOutputDTO } from 'src/engine/core-modules/emailing-domain/dtos/send-email-via-domain-output.dto';
+import { SendEmailViaDomainInput } from 'src/engine/core-modules/emailing-domain/dtos/send-email-via-domain.input';
 import { EmailingDomainService } from 'src/engine/core-modules/emailing-domain/services/emailing-domain.service';
 import { ResolverValidationPipe } from 'src/engine/core-modules/graphql/pipes/resolver-validation.pipe';
 import { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity';
 import { AuthWorkspace } from 'src/engine/decorators/auth/auth-workspace.decorator';
+import {
+  FeatureFlagGuard,
+  RequireFeatureFlag,
+} from 'src/engine/guards/feature-flag.guard';
 import { SettingsPermissionGuard } from 'src/engine/guards/settings-permission.guard';
 import { WorkspaceAuthGuard } from 'src/engine/guards/workspace-auth.guard';

@UseGuards(
  WorkspaceAuthGuard,
+  FeatureFlagGuard,
  SettingsPermissionGuard(PermissionFlagType.WORKSPACE),
 )
@UsePipes(ResolverValidationPipe)
@@ -23,6 +31,7 @@ export class EmailingDomainResolver {
  constructor(private readonly emailingDomainService: EmailingDomainService) {}

  @Mutation(() => EmailingDomainDTO)
+  @RequireFeatureFlag(FeatureFlagKey.IS_EMAIL_GROUP_ENABLED)
  async createEmailingDomain(
    @Args('domain') domain: string,
    @Args('driver') driver: EmailingDomainDriver,
@@ -39,6 +48,7 @@ export class EmailingDomainResolver {
  }

  @Mutation(() => Boolean)
+  @RequireFeatureFlag(FeatureFlagKey.IS_EMAIL_GROUP_ENABLED)
  async deleteEmailingDomain(
    @Args('id') id: string,
    @AuthWorkspace() currentWorkspace: WorkspaceEntity,
@@ -49,6 +59,7 @@ export class EmailingDomainResolver {
  }

  @Mutation(() => EmailingDomainDTO)
+  @RequireFeatureFlag(FeatureFlagKey.IS_EMAIL_GROUP_ENABLED)
  async verifyEmailingDomain(
    @Args('id') id: string,
    @AuthWorkspace() currentWorkspace: WorkspaceEntity,
@@ -62,7 +73,24 @@ export class EmailingDomainResolver {
    return emailingDomain;
  }

+  @Mutation(() => SendEmailViaDomainOutputDTO)
+  @RequireFeatureFlag(FeatureFlagKey.IS_EMAIL_GROUP_ENABLED)
+  async sendEmailViaEmailingDomain(
+    @Args('input') input: SendEmailViaDomainInput,
+    @AuthWorkspace() currentWorkspace: WorkspaceEntity,
+  ): Promise<SendEmailViaDomainOutputDTO> {
+    const { emailingDomainId, ...content } = input;
+    const result = await this.emailingDomainService.sendEmail(
+      currentWorkspace.id,
+      emailingDomainId,
+      content,
+    );
+
+    return { messageId: result.messageId };
+  }
+
  @Query(() => [EmailingDomainDTO])
+  @RequireFeatureFlag(FeatureFlagKey.IS_EMAIL_GROUP_ENABLED)
  async getEmailingDomains(
    @AuthWorkspace() currentWorkspace: WorkspaceEntity,
  ): Promise<EmailingDomainDTO[]> {
@@ -0,0 +1,28 @@
+import { EmailingDomainService } from 'src/engine/core-modules/emailing-domain/services/emailing-domain.service';
+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';
+
+export type EmailingDomainWorkspaceCleanupJobData = {
+  workspaceId: string;
+};
+
+@Processor(MessageQueue.deleteCascadeQueue)
+export class EmailingDomainWorkspaceCleanupJob {
+  constructor(private readonly emailingDomainService: EmailingDomainService) {}
+
+  @Process(EmailingDomainWorkspaceCleanupJob.name)
+  async handle(data: EmailingDomainWorkspaceCleanupJobData): Promise<void> {
+    const { workspaceId } = data;
+
+    try {
+      await this.emailingDomainService.cleanupAllEmailingDomainsForWorkspace(
+        workspaceId,
+      );
+    } catch (error) {
+      throw new Error(
+        `[${EmailingDomainWorkspaceCleanupJob.name}] Cannot cleanup emailing domains - ${workspaceId} - ${error?.message || error}`,
+      );
+    }
+  }
+}
@@ -0,0 +1,97 @@
+import { EmailingDomainDriverExceptionCode } from 'src/engine/core-modules/emailing-domain/drivers/exceptions/emailing-domain-driver.exception';
+import { type EmailingDomainDriverFactory } from 'src/engine/core-modules/emailing-domain/drivers/emailing-domain-driver.factory';
+import { EmailingDomainStatus } from 'src/engine/core-modules/emailing-domain/drivers/types/emailing-domain-status.type';
+import { EmailingDomainTenantStatus } from 'src/engine/core-modules/emailing-domain/drivers/types/emailing-domain-tenant-status.type';
+import { type EmailingDomainEntity } from 'src/engine/core-modules/emailing-domain/emailing-domain.entity';
+import { EmailingDomainService } from 'src/engine/core-modules/emailing-domain/services/emailing-domain.service';
+import { type WorkspaceScopedRepository } from 'src/engine/twenty-orm/workspace-scoped-repository/workspace-scoped-repository';
+
+describe('EmailingDomainService.sendEmail', () => {
+  const buildEmailingDomain = (
+    overrides: Partial<EmailingDomainEntity> = {},
+  ): EmailingDomainEntity =>
+    ({
+      id: 'domain-1',
+      workspaceId: 'ws1',
+      domain: 'mail.example.com',
+      status: EmailingDomainStatus.VERIFIED,
+      tenantStatus: EmailingDomainTenantStatus.ACTIVE,
+      ...overrides,
+    }) as EmailingDomainEntity;
+
+  const buildEmailContent = () => ({
+    from: 'hello@mail.example.com',
+    to: ['user@example.com'],
+    subject: 'Hi',
+    text: 'Body',
+  });
+
+  const setUp = (emailingDomain: EmailingDomainEntity) => {
+    const sendEmail = jest.fn().mockResolvedValue({ messageId: 'msg-1' });
+    const repository = {
+      findOne: jest.fn().mockResolvedValue(emailingDomain),
+    } as unknown as WorkspaceScopedRepository<EmailingDomainEntity>;
+    const factory = {
+      getCurrentDriver: () => ({ sendEmail }),
+    } as unknown as EmailingDomainDriverFactory;
+    const service = new EmailingDomainService(repository, factory);
+
+    return { service, sendEmail };
+  };
+
+  it('delegates to the driver when the domain is verified and the tenant is active', async () => {
+    const { service, sendEmail } = setUp(buildEmailingDomain());
+
+    const result = await service.sendEmail(
+      'ws1',
+      'domain-1',
+      buildEmailContent(),
+    );
+
+    expect(result.messageId).toBe('msg-1');
+    expect(sendEmail).toHaveBeenCalledWith(
+      expect.objectContaining({
+        workspaceId: 'ws1',
+        domain: 'mail.example.com',
+        from: 'hello@mail.example.com',
+      }),
+    );
+  });
+
+  it.each([
+    EmailingDomainTenantStatus.PAUSED,
+    EmailingDomainTenantStatus.PERMANENTLY_SUSPENDED,
+  ])(
+    'rejects sending with SENDING_SUSPENDED when tenantStatus is %s, without calling the driver',
+    async (tenantStatus) => {
+      const { service, sendEmail } = setUp(
+        buildEmailingDomain({ tenantStatus }),
+      );
+
+      await expect(
+        service.sendEmail('ws1', 'domain-1', buildEmailContent()),
+      ).rejects.toMatchObject({
+        code: EmailingDomainDriverExceptionCode.SENDING_SUSPENDED,
+      });
+      expect(sendEmail).not.toHaveBeenCalled();
+    },
+  );
+
+  // Verification is a precondition for the tenant-status check: a domain that
+  // has not been verified should surface a CONFIGURATION_ERROR rather than
+  // leaking the tenant pause state to callers who couldn't have used it anyway.
+  it('reports the verification failure before the tenant-status failure', async () => {
+    const { service } = setUp(
+      buildEmailingDomain({
+        status: EmailingDomainStatus.PENDING,
+        tenantStatus: EmailingDomainTenantStatus.PAUSED,
+      }),
+    );
+
+    await expect(
+      service.sendEmail('ws1', 'domain-1', buildEmailContent()),
+    ).rejects.toMatchObject({
+      code: EmailingDomainDriverExceptionCode.CONFIGURATION_ERROR,
+    });
+  });
+});
@@ -0,0 +1,35 @@
+import { Injectable, Logger } from '@nestjs/common';
+
+import { Not } from 'typeorm';
+
+import { EmailingDomainTenantStatus } from 'src/engine/core-modules/emailing-domain/drivers/types/emailing-domain-tenant-status.type';
+import { EmailingDomainEntity } from 'src/engine/core-modules/emailing-domain/emailing-domain.entity';
+import { InjectWorkspaceScopedRepository } from 'src/engine/twenty-orm/workspace-scoped-repository/inject-workspace-scoped-repository.decorator';
+import { WorkspaceScopedRepository } from 'src/engine/twenty-orm/workspace-scoped-repository/workspace-scoped-repository';
+
+@Injectable()
+export class EmailingDomainTenantStatusService {
+  private readonly logger = new Logger(EmailingDomainTenantStatusService.name);
+
+  constructor(
+    @InjectWorkspaceScopedRepository(EmailingDomainEntity)
+    private readonly emailingDomainRepository: WorkspaceScopedRepository<EmailingDomainEntity>,
+  ) {}
+
+  async setTenantStatusForWorkspace(
+    workspaceId: string,
+    tenantStatus: EmailingDomainTenantStatus,
+  ): Promise<void> {
+    const { affected } = await this.emailingDomainRepository.update(
+      workspaceId,
+      {
+        tenantStatus: Not(EmailingDomainTenantStatus.PERMANENTLY_SUSPENDED),
+      },
+      { tenantStatus },
+    );
+
+    this.logger.log(
+      `Workspace ${workspaceId}: ${affected ?? 0} domain(s) -> ${tenantStatus}`,
+    );
+  }
+}
@@ -1,16 +1,25 @@
-import { Injectable } from '@nestjs/common';
+import { Injectable, Logger } from '@nestjs/common';

-import { EmailingDomainDriverFactory } from 'src/engine/core-modules/emailing-domain/drivers/emailing-domain-driver.factory';
 import {
-  EmailingDomainDriver,
-  EmailingDomainStatus,
-} from 'src/engine/core-modules/emailing-domain/drivers/types/emailing-domain';
+  EmailingDomainDriverException,
+  EmailingDomainDriverExceptionCode,
+} from 'src/engine/core-modules/emailing-domain/drivers/exceptions/emailing-domain-driver.exception';
+import { EmailingDomainDriverFactory } from 'src/engine/core-modules/emailing-domain/drivers/emailing-domain-driver.factory';
+import { EmailingDomainDriver } from 'src/engine/core-modules/emailing-domain/drivers/types/emailing-domain-driver.type';
+import { EmailingDomainStatus } from 'src/engine/core-modules/emailing-domain/drivers/types/emailing-domain-status.type';
+import { EmailingDomainTenantStatus } from 'src/engine/core-modules/emailing-domain/drivers/types/emailing-domain-tenant-status.type';
+import {
+  type EmailingDomainEmailContent,
+  type EmailingDomainSendEmailResult,
+} from 'src/engine/core-modules/emailing-domain/drivers/types/send-email';
 import { EmailingDomainEntity } from 'src/engine/core-modules/emailing-domain/emailing-domain.entity';
 import { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity';
 import { InjectWorkspaceScopedRepository } from 'src/engine/twenty-orm/workspace-scoped-repository/inject-workspace-scoped-repository.decorator';
 import { WorkspaceScopedRepository } from 'src/engine/twenty-orm/workspace-scoped-repository/workspace-scoped-repository';
@Injectable()
 export class EmailingDomainService {
+  private readonly logger = new Logger(EmailingDomainService.name);
+
  constructor(
    @InjectWorkspaceScopedRepository(EmailingDomainEntity)
    private readonly emailingDomainRepository: WorkspaceScopedRepository<EmailingDomainEntity>,
@@ -19,30 +28,47 @@ export class EmailingDomainService {

  async createEmailingDomain(
    domain: string,
-    driver: EmailingDomainDriver,
+    driverType: EmailingDomainDriver,
    workspace: WorkspaceEntity,
  ): Promise<EmailingDomainEntity> {
-    const existingDomain = await this.emailingDomainRepository.findOne(
+    const existingEmailingDomain = await this.emailingDomainRepository.findOne(
      workspace.id,
      {
        where: { domain },
      },
    );

-    if (existingDomain) {
-      throw new Error('Emailing domain already exists for this workspace');
+    if (existingEmailingDomain) {
+      throw new EmailingDomainDriverException(
+        'Emailing domain already exists for this workspace',
+        EmailingDomainDriverExceptionCode.CONFIGURATION_ERROR,
+      );
    }

-    const driverInstance = this.emailingDomainDriverFactory.getCurrentDriver();
-    const verificationResult = await driverInstance.verifyDomain({
+    const emailingDomainDriver =
+      this.emailingDomainDriverFactory.getCurrentDriver();
+
+    await emailingDomainDriver.provisionWorkspace(workspace.id);
+
+    const verificationResult = await emailingDomainDriver.verifyDomain({
      domain,
      workspaceId: workspace.id,
    });

+    await emailingDomainDriver.registerDomain({
+      domain,
+      workspaceId: workspace.id,
+    });
+
+    const isVerifiedOnCreation =
+      verificationResult.status === EmailingDomainStatus.VERIFIED;
+
    return this.emailingDomainRepository.save(workspace.id, {
      domain,
-      driver,
-      ...verificationResult,
+      driver: driverType,
+      status: verificationResult.status,
+      verificationRecords: verificationResult.verificationRecords,
+      verifiedAt: isVerifiedOnCreation ? new Date() : null,
    });
  }

@@ -50,22 +76,31 @@ export class EmailingDomainService {
    workspace: WorkspaceEntity,
    emailingDomainId: string,
  ): Promise<void> {
-    const emailingDomain = await this.emailingDomainRepository.findOne(
+    const emailingDomain = await this.findEmailingDomainByIdOrThrow(
      workspace.id,
-      {
-        where: { id: emailingDomainId },
-      },
+      emailingDomainId,
    );

-    if (!emailingDomain) {
-      throw new Error('Emailing domain not found');
-    }
-
+    await this.deleteRemoteEmailingDomain(emailingDomain);
    await this.emailingDomainRepository.delete(workspace.id, {
      id: emailingDomain.id,
    });
  }

+  async cleanupAllEmailingDomainsForWorkspace(
+    workspaceId: string,
+  ): Promise<void> {
+    const emailingDomains =
+      await this.emailingDomainRepository.find(workspaceId);
+
+    for (const emailingDomain of emailingDomains) {
+      await this.deleteRemoteEmailingDomain(emailingDomain);
+    }
+
+    await this.deprovisionRemoteWorkspace(workspaceId);
+    await this.emailingDomainRepository.delete(workspaceId, {});
+  }
+
  async getEmailingDomains(
    workspace: WorkspaceEntity,
  ): Promise<EmailingDomainEntity[]> {
@@ -74,88 +109,127 @@ export class EmailingDomainService {
    });
  }

-  async getEmailingDomain(
-    workspace: WorkspaceEntity,
-    emailingDomainId: string,
-  ): Promise<EmailingDomainEntity | null> {
-    return this.emailingDomainRepository.findOne(workspace.id, {
-      where: { id: emailingDomainId },
-    });
-  }
-
  async verifyEmailingDomain(
    workspace: WorkspaceEntity,
    emailingDomainId: string,
  ): Promise<EmailingDomainEntity> {
-    const emailingDomain = await this.getEmailingDomain(
-      workspace,
+    const emailingDomain = await this.findEmailingDomainByIdOrThrow(
+      workspace.id,
      emailingDomainId,
    );

-    if (!emailingDomain) {
-      throw new Error('Emailing domain not found');
-    }
+    const emailingDomainDriver =
+      this.emailingDomainDriverFactory.getCurrentDriver();

-    if (emailingDomain.status === EmailingDomainStatus.VERIFIED) {
-      throw new Error('Emailing domain is already verified');
-    }
-
-    const driver = this.emailingDomainDriverFactory.getCurrentDriver();
-    const verificationResult = await driver.verifyDomain({
+    const verificationResult = await emailingDomainDriver.verifyDomain({
      domain: emailingDomain.domain,
      workspaceId: emailingDomain.workspaceId,
    });

-    return this.emailingDomainRepository.save(workspace.id, {
-      ...emailingDomain,
-      ...verificationResult,
-    });
-  }
-
-  async syncEmailingDomain(
-    workspace: WorkspaceEntity,
-    emailingDomainId: string,
-  ): Promise<EmailingDomainEntity> {
-    const emailingDomain = await this.getEmailingDomain(
-      workspace,
-      emailingDomainId,
-    );
-
-    if (!emailingDomain) {
-      throw new Error('Emailing domain not found');
-    }
+    const hasJustBecomeVerified =
+      emailingDomain.status !== EmailingDomainStatus.VERIFIED &&
+      verificationResult.status === EmailingDomainStatus.VERIFIED;

    await this.emailingDomainRepository.update(
      workspace.id,
-      { id: emailingDomainId },
+      { id: emailingDomain.id },
      {
-        verificationRecords: emailingDomain.verificationRecords,
-        status: EmailingDomainStatus.PENDING,
+        status: verificationResult.status,
+        verificationRecords: verificationResult.verificationRecords,
+        ...(hasJustBecomeVerified ? { verifiedAt: new Date() } : {}),
      },
    );

+    return this.emailingDomainRepository.findOneOrFail(workspace.id, {
+      where: { id: emailingDomain.id },
+    });
+  }
+
+  async sendEmail(
+    workspaceId: string,
+    emailingDomainId: string,
+    emailContent: EmailingDomainEmailContent,
+  ): Promise<EmailingDomainSendEmailResult> {
+    const emailingDomain = await this.findEmailingDomainByIdOrThrow(
+      workspaceId,
+      emailingDomainId,
+    );
+
+    if (emailingDomain.status !== EmailingDomainStatus.VERIFIED) {
+      throw new EmailingDomainDriverException(
+        `Emailing domain is not verified (status: ${emailingDomain.status})`,
+        EmailingDomainDriverExceptionCode.CONFIGURATION_ERROR,
+      );
+    }
+
+    if (emailingDomain.tenantStatus !== EmailingDomainTenantStatus.ACTIVE) {
+      throw new EmailingDomainDriverException(
+        `Sending is suspended for emailing domain ${emailingDomain.domain} (tenantStatus: ${emailingDomain.tenantStatus})`,
+        EmailingDomainDriverExceptionCode.SENDING_SUSPENDED,
+      );
+    }
+
+    const fromAddressDomain = emailContent.from.split('@')[1]?.toLowerCase();
+
+    if (fromAddressDomain !== emailingDomain.domain.toLowerCase()) {
+      throw new EmailingDomainDriverException(
+        `From address ${emailContent.from} does not match verified domain ${emailingDomain.domain}`,
+        EmailingDomainDriverExceptionCode.CONFIGURATION_ERROR,
+      );
+    }
+
+    return this.emailingDomainDriverFactory.getCurrentDriver().sendEmail({
+      ...emailContent,
+      workspaceId,
+      domain: emailingDomain.domain,
+    });
+  }
+
+  private async findEmailingDomainByIdOrThrow(
+    workspaceId: string,
+    emailingDomainId: string,
+  ): Promise<EmailingDomainEntity> {
+    const emailingDomain = await this.emailingDomainRepository.findOne(
+      workspaceId,
+      {
+        where: { id: emailingDomainId },
+      },
+    );
+
+    if (!emailingDomain) {
+      throw new EmailingDomainDriverException(
+        'Emailing domain not found',
+        EmailingDomainDriverExceptionCode.NOT_FOUND,
+      );
+    }
+
+    return emailingDomain;
+  }
+
+  private async deleteRemoteEmailingDomain(
+    emailingDomain: EmailingDomainEntity,
+  ): Promise<void> {
    try {
-      const driver = this.emailingDomainDriverFactory.getCurrentDriver();
-      const statusResult = await driver.getDomainStatus({
+      await this.emailingDomainDriverFactory.getCurrentDriver().cleanupDomain({
        domain: emailingDomain.domain,
        workspaceId: emailingDomain.workspaceId,
      });
-
-      return this.emailingDomainRepository.save(workspace.id, {
-        ...emailingDomain,
-        ...statusResult,
-      });
    } catch (error) {
-      await this.emailingDomainRepository.update(
-        workspace.id,
-        { id: emailingDomainId },
-        {
-          verificationRecords: emailingDomain.verificationRecords,
-          status: emailingDomain.status,
-        },
+      this.logger.warn(
+        `Remote cleanup for emailing domain ${emailingDomain.domain} (workspace ${emailingDomain.workspaceId}) failed: ${error}`,
      );
+    }
+  }

-      throw error;
+  private async deprovisionRemoteWorkspace(workspaceId: string): Promise<void> {
+    try {
+      await this.emailingDomainDriverFactory
+        .getCurrentDriver()
+        .deprovisionWorkspace(workspaceId);
+    } catch (error) {
+      this.logger.warn(
+        `Remote deprovision for emailing domain workspace ${workspaceId} failed: ${error}`,
+      );
    }
  }
 }
@@ -13,6 +13,7 @@ import { UpdateSubscriptionQuantityJob } from 'src/engine/core-modules/billing/j
 import { StripeModule } from 'src/engine/core-modules/billing/stripe/stripe.module';
 import { EmailSenderJob } from 'src/engine/core-modules/email/email-sender.job';
 import { EmailModule } from 'src/engine/core-modules/email/email.module';
+import { EmailingDomainModule } from 'src/engine/core-modules/emailing-domain/emailing-domain.module';
 import { EnterpriseModule } from 'src/engine/core-modules/enterprise/enterprise.module';
 import { FeatureFlagModule } from 'src/engine/core-modules/feature-flag/feature-flag.module';
 import { GenerateSdkClientJob } from 'src/engine/core-modules/sdk-client/jobs/generate-sdk-client.job';
@@ -79,6 +80,7 @@ import { WorkflowModule } from 'src/modules/workflow/workflow.module';
    AiChatModule,
    LogicFunctionModule,
    EnterpriseModule,
+    EmailingDomainModule,
  ],
  providers: [
    CleanSuspendedWorkspacesJob,
@@ -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/engine/core-modules/messaging-webhooks/messaging-webhook.exception';
+import { getMessagingWebhookExceptionStatusCode } from 'src/engine/core-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/engine/core-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),
+    });
+  }
+}
@@ -1,67 +1,61 @@
 import {
-  BadRequestException,
  Controller,
  HttpCode,
  Post,
  type RawBodyRequest,
  Req,
+  UseFilters,
  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 { MessagingWebhookApiExceptionFilter } from 'src/engine/core-modules/messaging-webhooks/filters/messaging-webhook-api-exception.filter';
+import { MessagingWebhookExceptionCode } from 'src/engine/core-modules/messaging-webhooks/messaging-webhook-exception-code.enum';
+import { MessagingWebhookException } from 'src/engine/core-modules/messaging-webhooks/messaging-webhook.exception';
+import { SesInboundWebhookRouterService } from 'src/engine/core-modules/messaging-webhooks/services/ses-inbound-webhook-router.service';
+import { SesOutboundWebhookRouterService } from 'src/engine/core-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';
-
-type SnsPayload = SnsPayloadValidator.SnsPayload;
+import { isDefined } from 'twenty-shared/utils';

@Controller()
+@UseFilters(MessagingWebhookApiExceptionFilter)
 export class MessagingWebhooksController {
  constructor(
-    private readonly snsSignatureVerifierService: SnsSignatureVerifierService,
-    private readonly messagingWebhookDispatcherService: MessagingWebhookDispatcherService,
+    private readonly sesInboundWebhookRouterService: SesInboundWebhookRouterService,
+    private readonly sesOutboundWebhookRouterService: SesOutboundWebhookRouterService,
  ) {}

-  @Post(['webhooks/messaging/ses'])
+  @Post(['webhooks/messaging/ses/inbound'])
  @UseGuards(PublicEndpointGuard, NoPermissionGuard)
  @HttpCode(200)
-  async handleSesWebhook(
+  async handleSesInboundWebhook(
    @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,
+    if (!isDefined(request.rawBody)) {
+      throw new MessagingWebhookException(
+        'Missing SNS payload',
+        MessagingWebhookExceptionCode.MESSAGING_WEBHOOK_MISSING_REQUEST_BODY,
      );
    }
+
+    await this.sesInboundWebhookRouterService.route(request.rawBody);
  }

-  private parseSnsPayload(rawBody: Buffer): SnsPayload {
-    try {
-      return JSON.parse(rawBody.toString('utf8')) as SnsPayload;
-    } catch {
-      throw new BadRequestException('Invalid SNS payload');
+  @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);
  }
 }
@@ -1,13 +1,25 @@
 import { Module } from '@nestjs/common';

+import { EmailingDomainModule } from 'src/engine/core-modules/emailing-domain/emailing-domain.module';
 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 { SesInboundMailHandlerService } from 'src/engine/core-modules/messaging-webhooks/services/ses-inbound-mail-handler.service';
+import { SesInboundWebhookRouterService } from 'src/engine/core-modules/messaging-webhooks/services/ses-inbound-webhook-router.service';
+import { SesOutboundSendingStateHandlerService } from 'src/engine/core-modules/messaging-webhooks/services/ses-outbound-sending-state-handler.service';
+import { SesOutboundWebhookRouterService } from 'src/engine/core-modules/messaging-webhooks/services/ses-outbound-webhook-router.service';
 import { SnsSignatureVerifierService } from 'src/engine/core-modules/messaging-webhooks/services/sns-signature-verifier.service';
+import { SnsSubscriptionConfirmerService } from 'src/engine/core-modules/messaging-webhooks/services/sns-subscription-confirmer.service';
 import { TwentyConfigModule } from 'src/engine/core-modules/twenty-config/twenty-config.module';

@Module({
-  imports: [TwentyConfigModule],
+  imports: [TwentyConfigModule, EmailingDomainModule],
  controllers: [MessagingWebhooksController],
-  providers: [SnsSignatureVerifierService, MessagingWebhookDispatcherService],
+  providers: [
+    SnsSignatureVerifierService,
+    SnsSubscriptionConfirmerService,
+    SesInboundMailHandlerService,
+    SesOutboundSendingStateHandlerService,
+    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/engine/core-modules/messaging-webhooks/services/ses-outbound-sending-state-handler.service';
+import { type SesEventBridgeNotification } from 'src/engine/core-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();
+  });
+});
@@ -1,99 +0,0 @@
-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;
-    }
-  }
-}
@@ -0,0 +1,43 @@
+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 { 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';
+
+@Injectable()
+export class SesInboundMailHandlerService {
+  private readonly logger = new Logger(SesInboundMailHandlerService.name);
+
+  constructor(
+    @InjectMessageQueue(MessageQueue.messagingQueue)
+    private readonly messageQueueService: MessageQueueService,
+  ) {}
+
+  async handle(
+    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,
+      },
+    );
+  }
+}
@@ -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/engine/core-modules/messaging-webhooks/messaging-webhook-exception-code.enum';
+import { MessagingWebhookException } from 'src/engine/core-modules/messaging-webhooks/messaging-webhook.exception';
+import { SesInboundMailHandlerService } from 'src/engine/core-modules/messaging-webhooks/services/ses-inbound-mail-handler.service';
+import { SnsSignatureVerifierService } from 'src/engine/core-modules/messaging-webhooks/services/sns-signature-verifier.service';
+import { SnsSubscriptionConfirmerService } from 'src/engine/core-modules/messaging-webhooks/services/sns-subscription-confirmer.service';
+import { type SesInboundNotification } from 'src/engine/core-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,58 @@
+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/engine/core-modules/messaging-webhooks/types/ses-event-bridge-notification.type';
+import { parseWorkspaceIdFromAwsSesResourceArn } from 'src/engine/core-modules/messaging-webhooks/utils/parse-workspace-id-from-aws-ses-resource-arn.util';
+import { isDefined, isNonEmptyArray } 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 = this.resolveWorkspaceIdFromResources(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,
+    );
+  }
+
+  private resolveWorkspaceIdFromResources(
+    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,59 @@
+import { Injectable } from '@nestjs/common';
+
+import type SnsPayloadValidator from 'sns-payload-validator';
+import { isDefined, parseJson } from 'twenty-shared/utils';
+
+import { MessagingWebhookExceptionCode } from 'src/engine/core-modules/messaging-webhooks/messaging-webhook-exception-code.enum';
+import { MessagingWebhookException } from 'src/engine/core-modules/messaging-webhooks/messaging-webhook.exception';
+import { SesOutboundSendingStateHandlerService } from 'src/engine/core-modules/messaging-webhooks/services/ses-outbound-sending-state-handler.service';
+import { SnsSignatureVerifierService } from 'src/engine/core-modules/messaging-webhooks/services/sns-signature-verifier.service';
+import { SnsSubscriptionConfirmerService } from 'src/engine/core-modules/messaging-webhooks/services/sns-subscription-confirmer.service';
+import { type SesEventBridgeNotification } from 'src/engine/core-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,
+  ) {}
+
+  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,
+      );
+    }
+
+    await this.sesOutboundSendingStateHandlerService.handle(event);
+  }
+}
@@ -1,7 +1,9 @@
-import { ForbiddenException, Injectable, Logger } from '@nestjs/common';
+import { Injectable, Logger } from '@nestjs/common';

 import SnsPayloadValidator from 'sns-payload-validator';

+import { MessagingWebhookExceptionCode } from 'src/engine/core-modules/messaging-webhooks/messaging-webhook-exception-code.enum';
+import { MessagingWebhookException } from 'src/engine/core-modules/messaging-webhooks/messaging-webhook.exception';
 import { TwentyConfigService } from 'src/engine/core-modules/twenty-config/twenty-config.service';

 type SnsPayload = SnsPayloadValidator.SnsPayload;
@@ -17,7 +19,10 @@ export class SnsSignatureVerifierService {
    if (!this.isTopicAllowlisted(payload.TopicArn)) {
      this.logger.warn(`SNS topic ${payload.TopicArn} is not in allowlist`);

-      throw new ForbiddenException('SNS topic not allowed');
+      throw new MessagingWebhookException(
+        'SNS topic not allowed',
+        MessagingWebhookExceptionCode.MESSAGING_WEBHOOK_FORBIDDEN_TOPIC,
+      );
    }

    try {
@@ -28,7 +33,10 @@ export class SnsSignatureVerifierService {

      this.logger.warn(`SNS signature verification failed: ${errorMessage}`);

-      throw new ForbiddenException('SNS signature invalid');
+      throw new MessagingWebhookException(
+        'SNS signature invalid',
+        MessagingWebhookExceptionCode.MESSAGING_WEBHOOK_INVALID_SIGNATURE,
+      );
    }
  }

@@ -0,0 +1,39 @@
+import { Injectable, Logger } from '@nestjs/common';
+
+import { MessagingWebhookExceptionCode } from 'src/engine/core-modules/messaging-webhooks/messaging-webhook-exception-code.enum';
+import { MessagingWebhookException } from 'src/engine/core-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,15 @@
+export type SesEventBridgeNotification = {
+  source: 'aws.ses';
+  'detail-type': 'Sending Status Enabled' | 'Sending Status Disabled';
+  resources?: string[];
+  detail?: {
+    version?: string;
+    data?: {
+      origin?: string;
+      record?: {
+        status?: 'ENABLED' | 'DISABLED';
+        cause?: string;
+      };
+    };
+  };
+};
@@ -0,0 +1,53 @@
+import { parseWorkspaceIdFromAwsSesResourceArn } from 'src/engine/core-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/engine/core-modules/messaging-webhooks/messaging-webhook-exception-code.enum';
+import { type MessagingWebhookException } from 'src/engine/core-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;
+};
@@ -22,6 +22,10 @@ import { CustomDomainManagerService } from 'src/engine/core-modules/domain/custo
 import { SubdomainManagerService } from 'src/engine/core-modules/domain/subdomain-manager/services/subdomain-manager.service';
 import { ExceptionHandlerService } from 'src/engine/core-modules/exception-handler/exception-handler.service';
 import { FeatureFlagService } from 'src/engine/core-modules/feature-flag/services/feature-flag.service';
+import {
+  EmailingDomainWorkspaceCleanupJob,
+  type EmailingDomainWorkspaceCleanupJobData,
+} from 'src/engine/core-modules/emailing-domain/jobs/emailing-domain-workspace-cleanup.job';
 import { FileCorePictureService } from 'src/engine/core-modules/file/file-core-picture/services/file-core-picture.service';
 import {
  FileWorkspaceFolderDeletionJob,
@@ -508,6 +512,11 @@ export class WorkspaceService extends TypeOrmQueryService<WorkspaceEntity> {
      { workspaceId: id },
    );

+    await this.messageQueueService.add<EmailingDomainWorkspaceCleanupJobData>(
+      EmailingDomainWorkspaceCleanupJob.name,
+      { workspaceId: id },
+    );
+
    if (workspace.customDomain) {
      await this.dnsManagerService.deleteHostnameSilently(
        workspace.customDomain,
				`@@ -0,0 +1 @@`
				`export const AWS_SES_EVENT_BUS_NAME = 'default';`
				`@@ -0,0 +1 @@`
				`export const AWS_SES_MAIL_FROM_SUBDOMAIN = 'bounce';`
				`@@ -0,0 +1 @@`
				`export const AWS_SES_MARKETING_TOPIC_NAME = 'marketing';`
				`@@ -0,0 +1 @@`
				`export const AWS_SES_RESOURCE_NAME_PREFIX = 'twenty-workspace';`