feat(emailing): forward SES events to Tatami Monitor (#22407)

Add AwsSesObservabilityService, which adds an SNS event destination to
each workspace's SES configuration set (gated on TATAMI_SNS_TOPIC_ARN)
so deliverability events reach Tatami. Tag sends with tenant_id for
per-workspace breakdowns.

Requires to be merged https://github.com/twentyhq/twenty-infra/pull/765

<!-- This is an auto-generated description by cubic. -->
<a
href="https://cubic.dev/pr/twentyhq/twenty/pull/22407?utm_source=github"
target="_blank" rel="noopener noreferrer"
data-no-image-dialog="true"><picture><source
media="(prefers-color-scheme: dark)"
srcset="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"><source
media="(prefers-color-scheme: light)"
srcset="https://www.cubic.dev/buttons/review-in-cubic-light.svg"><img
alt="Review in cubic"
src="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"></picture></a>
<!-- End of auto-generated description by cubic. -->

packages/twenty-server/src/engine/core-modules/emailing-domain/drivers/aws-ses/services/__tests__/aws-ses-register-domain.service.spec.ts

@@ -7,6 +7,7 @@ import {
 } 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 AwsSesObservabilityService } from 'src/engine/core-modules/emailing-domain/drivers/aws-ses/services/aws-ses-observability.service';
 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';
@@ -34,7 +35,13 @@ describe('AwsSesRegisterDomainService', () => {
     const clientProvider = {
       getSESClient: () => ({ send }),
     } as unknown as AwsSesClientProvider;
-    const service = new AwsSesRegisterDomainService(clientProvider);
+    const awsSesObservabilityService = {
+      addEventDestination: jest.fn().mockResolvedValue(undefined),
+    } as unknown as AwsSesObservabilityService;
+    const service = new AwsSesRegisterDomainService(
+      clientProvider,
+      awsSesObservabilityService,
+    );
 
     return { service, send };
   };

packages/twenty-server/src/engine/core-modules/emailing-domain/drivers/aws-ses/services/aws-ses-observability.service.ts

@@ -0,0 +1,57 @@
+import { Injectable } from '@nestjs/common';
+
+import {
+  AlreadyExistsException,
+  CreateConfigurationSetEventDestinationCommand,
+} from '@aws-sdk/client-sesv2';
+
+import { isDefined } from 'twenty-shared/utils';
+
+import { AwsSesClientProvider } from 'src/engine/core-modules/emailing-domain/drivers/aws-ses/providers/aws-ses-client.provider';
+import { TwentyConfigService } from 'src/engine/core-modules/twenty-config/twenty-config.service';
+
+@Injectable()
+export class AwsSesObservabilityService {
+  constructor(
+    private readonly awsSesClientProvider: AwsSesClientProvider,
+    private readonly twentyConfigService: TwentyConfigService,
+  ) {}
+
+  async addEventDestination(configurationSetName: string): Promise<void> {
+    const tatamiSnsTopicArn = this.twentyConfigService.get(
+      'TATAMI_SNS_TOPIC_ARN',
+    );
+
+    if (!isDefined(tatamiSnsTopicArn)) {
+      return;
+    }
+
+    const sesClient = this.awsSesClientProvider.getSESClient();
+
+    await sesClient
+      .send(
+        new CreateConfigurationSetEventDestinationCommand({
+          ConfigurationSetName: configurationSetName,
+          EventDestinationName: 'tatami-sns',
+          EventDestination: {
+            Enabled: true,
+            MatchingEventTypes: [
+              'SEND',
+              'DELIVERY',
+              'BOUNCE',
+              'COMPLAINT',
+              'REJECT',
+              'RENDERING_FAILURE',
+              'DELIVERY_DELAY',
+            ],
+            SnsDestination: { TopicArn: tatamiSnsTopicArn },
+          },
+        }),
+      )
+      .catch((error) => {
+        if (!(error instanceof AlreadyExistsException)) {
+          throw error;
+        }
+      });
+  }
+}

packages/twenty-server/src/engine/core-modules/emailing-domain/drivers/aws-ses/services/aws-ses-register-domain.service.ts

@@ -12,6 +12,7 @@ import { type AwsSesDriverConfig } from 'src/engine/core-modules/emailing-domain
 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 { AwsSesClientProvider } from 'src/engine/core-modules/emailing-domain/drivers/aws-ses/providers/aws-ses-client.provider';
+import { AwsSesObservabilityService } from 'src/engine/core-modules/emailing-domain/drivers/aws-ses/services/aws-ses-observability.service';
 
 type ProvisionWorkspaceInput = {
   tenantName: string;
@@ -22,7 +23,10 @@ type ProvisionWorkspaceInput = {
 export class AwsSesRegisterDomainService {
   private readonly logger = new Logger(AwsSesRegisterDomainService.name);
 
-  constructor(private readonly awsSesClientProvider: AwsSesClientProvider) {}
+  constructor(
+    private readonly awsSesClientProvider: AwsSesClientProvider,
+    private readonly awsSesObservabilityService: AwsSesObservabilityService,
+  ) {}
 
   async provisionWorkspaceResources(
     input: ProvisionWorkspaceInput,
@@ -89,6 +93,10 @@ export class AwsSesRegisterDomainService {
         }
       });
 
+    await this.awsSesObservabilityService.addEventDestination(
+      input.configurationSetName,
+    );
+
     this.logger.log(
       `Provisioned workspace resources for tenant ${input.tenantName}`,
     );

packages/twenty-server/src/engine/core-modules/emailing-domain/drivers/aws-ses/services/aws-ses-send-email.service.ts

@@ -80,6 +80,7 @@ export class AwsSesSendEmailService {
           EmailTags: [
             { Name: 'workspace', Value: input.workspaceId },
             { Name: 'domain', Value: input.domain },
+            { Name: 'tenant_id', Value: input.workspaceId },
           ],
         }),
       );

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

@@ -5,6 +5,7 @@ import { NestjsQueryTypeOrmModule } from '@ptc-org/nestjs-query-typeorm';
 import { TypeORMModule } from 'src/database/typeorm/typeorm.module';
 import { DnsManagerModule } from 'src/engine/core-modules/dns-manager/dns-manager.module';
 import { AwsSesClientProvider } from 'src/engine/core-modules/emailing-domain/drivers/aws-ses/providers/aws-ses-client.provider';
+import { AwsSesObservabilityService } from 'src/engine/core-modules/emailing-domain/drivers/aws-ses/services/aws-ses-observability.service';
 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';
@@ -47,6 +48,7 @@ import { provideWorkspaceScopedRepository } from 'src/engine/twenty-orm/workspac
     EmailingDomainWorkspaceCleanupJob,
     AwsSesClientProvider,
     AwsSesHandleErrorService,
+    AwsSesObservabilityService,
     AwsSesRegisterDomainService,
     AwsSesSendEmailService,
     LogEmailingDomainDriver,

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

@@ -1959,6 +1959,25 @@ export class ConfigVariables {
   @IsOptional()
   SES_SNS_TOPIC_ARN_ALLOWLIST: string;
 
+  @ConfigVariablesMetadata({
+    group: ConfigVariablesGroup.AWS_SES_SETTINGS,
+    description:
+      'Tatami Monitor ingest webhook URL (SNS HTTPS subscription target for deliverability observability).',
+    type: ConfigVariableType.STRING,
+    isSensitive: true,
+  })
+  @IsOptional()
+  TATAMI_SES_WEBHOOK_URL: string;
+
+  @ConfigVariablesMetadata({
+    group: ConfigVariablesGroup.AWS_SES_SETTINGS,
+    description:
+      'SNS topic ARN that fans out SES events to Tatami Monitor. When set, an SNS event destination is added to each workspace SES configuration set.',
+    type: ConfigVariableType.STRING,
+  })
+  @IsOptional()
+  TATAMI_SNS_TOPIC_ARN: string;
+
   @ConfigVariablesMetadata({
     group: ConfigVariablesGroup.ADVANCED_SETTINGS,
     description: 'Timeout in milliseconds for primary database queries',