houseassassin/twenty
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: harden email-group SES provisioning, cleanup, and inbound replay (#21046)

Browse Source 
## Changes

**Provisioning idempotent** (`aws-ses-register-domain.service.ts`)
- Each SES create call (`CreateConfigurationSet`, event destination,
contact list, tenant association) now swallow `AlreadyExistsException`
via `.send().catch()`.
- Retry after partial failure re-run every step, no blow up on "already
exists". Before: one existing resource kill whole provision.

**Workspace delete clean up cloud** (`workspace.service.ts`,
`emailing-domain-workspace-cleanup.job.ts`,
`emailing-domain.service.ts`)
- On workspace delete, fetch domain list first, pass domains to cleanup
job.
- Cleanup now loop `driver.cleanupDomain(domain)` per domain +
`deprovisionWorkspace`. Tear down SES identity/tenant/config-set, not
just delete DB rows.
- Before: DB rows gone, SES resources orphaned forever. Now: cloud match
DB.

**Inbound replay dedupe** (`ses-inbound-mail-handler.service.ts`)
- Use `snsMessageId` as job id. SNS deliver same message twice → second
is no-op. No duplicate inbound email import.
This commit is contained in:
neo773
2026-05-29 16:01:03 +05:30
committed by GitHub
parent 08b1c5738d
commit 10c0bed462
7 changed files with 142 additions and 144 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/twenty-server/src/engine/core-modules/emailing-domain/drivers/aws-ses/services/__tests__/aws-ses-register-domain.service.spec.ts
+33 -42
View File
@@ -1,10 +1,9 @@
import {
AlreadyExistsException,
CreateConfigurationSetCommand,
CreateConfigurationSetEventDestinationCommand,
CreateContactListCommand,
CreateTenantResourceAssociationCommand,
GetConfigurationSetCommand,
NotFoundException,
PutEmailIdentityMailFromAttributesCommand,
} from '@aws-sdk/client-sesv2';
@@ -26,10 +25,10 @@ describe('AwsSesRegisterDomainService', () => {
contactListName: 'twenty-workspace-ws1',
};
const buildNotFound = () =>
new NotFoundException({
$metadata: { httpStatusCode: 404 },
message: 'Configuration set not found.',
const buildAlreadyExists = () =>
new AlreadyExistsException({
$metadata: { httpStatusCode: 409 },
message: 'Resource already exists.',
});
const setUp = () => {
@@ -43,33 +42,7 @@ describe('AwsSesRegisterDomainService', () => {
};
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 () => {
it('creates every workspace-scoped resource', async () => {
const { service, send } = setUp();
send.mockResolvedValue({});
@@ -80,20 +53,38 @@ describe('AwsSesRegisterDomainService', () => {
([command]) => command.constructor.name,
);
expect(commandTypes).toEqual([GetConfigurationSetCommand.name]);
expect(commandTypes).toEqual([
CreateConfigurationSetCommand.name,
CreateConfigurationSetEventDestinationCommand.name,
CreateContactListCommand.name,
CreateTenantResourceAssociationCommand.name,
]);
});
it('propagates non-NotFound AWS errors raised by the existence probe', async () => {
it('ignores AlreadyExistsException per resource so a retry re-runs every step', async () => {
const { service, send } = setUp();
send.mockRejectedValue(buildAlreadyExists());
await service.provisionWorkspaceResources(provisionInput, config);
const commandTypes = send.mock.calls.map(
([command]) => command.constructor.name,
);
expect(commandTypes).toEqual([
CreateConfigurationSetCommand.name,
CreateConfigurationSetEventDestinationCommand.name,
CreateContactListCommand.name,
CreateTenantResourceAssociationCommand.name,
]);
});
it('propagates AWS errors that are not AlreadyExistsException', async () => {
const { service, send } = setUp();
const fatalError = new Error('Boom');
send.mockImplementation(async (command) => {
if (command instanceof GetConfigurationSetCommand) {
throw fatalError;
}
return {};
});
send.mockRejectedValue(fatalError);
await expect(
service.provisionWorkspaceResources(provisionInput, config),
packages/twenty-server/src/engine/core-modules/emailing-domain/drivers/aws-ses/services/aws-ses-register-domain.service.ts
+73 -80
View File
@@ -1,14 +1,12 @@
import { Injectable, Logger } from '@nestjs/common';
import {
AlreadyExistsException,
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';
@@ -35,69 +33,84 @@ export class AwsSesRegisterDomainService {
): 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 CreateConfigurationSetCommand({
ConfigurationSetName: input.configurationSetName,
ReputationOptions: { ReputationMetricsEnabled: true },
SendingOptions: { SendingEnabled: true },
SuppressionOptions: { SuppressedReasons: ['BOUNCE', 'COMPLAINT'] },
Tags: [{ Key: 'managed-by', Value: 'twenty' }],
}),
)
.catch((error) => {
if (!(error instanceof AlreadyExistsException)) {
throw error;
}
});
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',
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 },
},
],
Tags: [{ Key: 'managed-by', Value: 'twenty' }],
}),
);
}),
)
.catch((error) => {
if (!(error instanceof AlreadyExistsException)) {
throw error;
}
});
await sesClient.send(
new CreateTenantResourceAssociationCommand({
TenantName: input.tenantName,
ResourceArn: configurationSetArn,
}),
);
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' }],
}),
)
.catch((error) => {
if (!(error instanceof AlreadyExistsException)) {
throw error;
}
});
await sesClient
.send(
new CreateTenantResourceAssociationCommand({
TenantName: input.tenantName,
ResourceArn: configurationSetArn,
}),
)
.catch((error) => {
if (!(error instanceof AlreadyExistsException)) {
throw error;
}
});
this.logger.log(
`Provisioned workspace resources for tenant ${input.tenantName}`,
@@ -117,24 +130,4 @@ export class AwsSesRegisterDomainService {
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;
}
}
}
packages/twenty-server/src/engine/core-modules/emailing-domain/jobs/emailing-domain-workspace-cleanup.job.ts
+4 -2
View File
@@ -5,6 +5,7 @@ import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queu
export type EmailingDomainWorkspaceCleanupJobData = {
workspaceId: string;
domains: string[];
};
@Processor(MessageQueue.deleteCascadeQueue)
@@ -13,11 +14,12 @@ export class EmailingDomainWorkspaceCleanupJob {
@Process(EmailingDomainWorkspaceCleanupJob.name)
async handle(data: EmailingDomainWorkspaceCleanupJobData): Promise<void> {
const { workspaceId } = data;
const { workspaceId, domains } = data;
try {
await this.emailingDomainService.cleanupAllEmailingDomainsForWorkspace(
await this.emailingDomainService.cleanupEmailingDomainsForWorkspace(
workspaceId,
domains,
);
} catch (error) {
throw new Error(
packages/twenty-server/src/engine/core-modules/emailing-domain/services/emailing-domain.service.ts
+19 -19
View File
@@ -87,18 +87,30 @@ export class EmailingDomainService {
});
}
async cleanupAllEmailingDomainsForWorkspace(
async cleanupEmailingDomainsForWorkspace(
workspaceId: string,
domains: string[],
): Promise<void> {
const emailingDomains =
await this.emailingDomainRepository.find(workspaceId);
const emailingDomainDriver =
this.emailingDomainDriverFactory.getCurrentDriver();
for (const emailingDomain of emailingDomains) {
await this.deleteRemoteEmailingDomain(emailingDomain);
if (domains.length === 0) {
return;
}
await this.deprovisionRemoteWorkspace(workspaceId);
await this.emailingDomainRepository.delete(workspaceId, {});
const results = await Promise.allSettled(
domains.map((domain) =>
emailingDomainDriver.cleanupDomain({ domain, workspaceId }),
),
);
await emailingDomainDriver.deprovisionWorkspace(workspaceId);
if (results.some((result) => result.status === 'rejected')) {
throw new Error(
`Failed to clean up one or more emailing domains for workspace ${workspaceId}`,
);
}
}
async getEmailingDomains(
@@ -220,16 +232,4 @@ export class EmailingDomainService {
);
}
}
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}`,
);
}
}
}
packages/twenty-server/src/engine/core-modules/messaging-webhooks/services/ses-inbound-mail-handler.service.ts
+1
View File
@@ -38,6 +38,7 @@ export class SesInboundMailHandlerService {
s3Key: receipt.action.objectKey,
envelopeRecipients: receipt.recipients,
},
{ id: snsMessageId },
);
}
}
packages/twenty-server/src/engine/core-modules/workspace/services/__tests__/workspace.service.spec.ts
+3
View File
@@ -160,6 +160,9 @@ describe('WorkspaceService', () => {
delete: jest.fn().mockResolvedValue({ affected: 0 }),
},
}),
getRepository: jest.fn().mockReturnValue({
find: jest.fn().mockResolvedValue([]),
}),
},
},
],
packages/twenty-server/src/engine/core-modules/workspace/services/workspace.service.ts
+9 -1
View File
@@ -22,6 +22,7 @@ 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 { EmailingDomainEntity } from 'src/engine/core-modules/emailing-domain/emailing-domain.entity';
import {
EmailingDomainWorkspaceCleanupJob,
type EmailingDomainWorkspaceCleanupJobData,
@@ -512,9 +513,16 @@ export class WorkspaceService extends TypeOrmQueryService<WorkspaceEntity> {
{ workspaceId: id },
);
const emailingDomains = await this.coreDataSource
.getRepository(EmailingDomainEntity)
.find({ where: { workspaceId: id } });
await this.messageQueueService.add<EmailingDomainWorkspaceCleanupJobData>(
EmailingDomainWorkspaceCleanupJob.name,
{ workspaceId: id },
{
workspaceId: id,
domains: emailingDomains.map((emailingDomain) => emailingDomain.domain),
},
);
if (workspace.customDomain) {