feat: add email forwarding message channel
Adds a new EMAIL_FORWARDING message channel type that lets users receive emails from Google Groups and Microsoft 365 shared mailboxes via SES inbound routing to S3. Backend: - New enum values: MessageChannelType.EMAIL_FORWARDING, ConnectedAccountProvider.EMAIL_FORWARDING - S3-based inbound email pipeline: poll cron -> parse MIME -> extract envelope recipient -> match channel -> save messages - Loop prevention via X-Twenty-Origin header on outbound mail - createEmailForwardingChannel mutation generates ch_<hex>@<domain> addresses - Exhaustive switch coverage across auth, refresh tokens, outbound, drafts, email aliases Frontend: - "Add Email Forwarding" card in account settings - Modal showing generated forwarding address with copy-to-clipboard - GraphQL mutation + hook for channel creation Tests: - extractEnvelopeRecipient util: Received header, Delivered-To, X-Original-To, To/Cc fallback, domain mismatch - InboundEmailImportService: imported, unmatched, loop_dropped, unconfigured, parse_failed, persist_failed outcomes https://claude.ai/code/session_01KpyF6p4cUEnuaT4h8DP5Pm
This commit is contained in:
Claude
@@ -27,6 +27,7 @@ export const getMissingDraftEmailScopes = (
|
||||
case ConnectedAccountProvider.IMAP_SMTP_CALDAV:
|
||||
case ConnectedAccountProvider.OIDC:
|
||||
case ConnectedAccountProvider.SAML:
|
||||
case ConnectedAccountProvider.EMAIL_FORWARDING:
|
||||
return [];
|
||||
default:
|
||||
assertUnreachable(
|
||||
|
||||
+123
@@ -0,0 +1,123 @@
|
||||
import { useCopyToClipboard } from '~/hooks/useCopyToClipboard';
|
||||
import { ModalStatefulWrapper } from '@/ui/layout/modal/components/ModalStatefulWrapper';
|
||||
import { useModal } from '@/ui/layout/modal/hooks/useModal';
|
||||
import { styled } from '@linaria/react';
|
||||
import { useLingui } from '@lingui/react/macro';
|
||||
import { IconCopy, IconMail } from 'twenty-ui/display';
|
||||
import { Button } from 'twenty-ui/input';
|
||||
import { themeCssVariables } from 'twenty-ui/theme-constants';
|
||||
|
||||
export const EMAIL_FORWARDING_MODAL_ID = 'email-forwarding-address-modal';
|
||||
|
||||
const StyledModalContent = styled.div`
|
||||
display: flex;
|
||||
flex-direction: column;
|
||||
gap: ${themeCssVariables.spacing[6]};
|
||||
padding: ${themeCssVariables.spacing[6]};
|
||||
`;
|
||||
|
||||
const StyledTitle = styled.h2`
|
||||
color: ${themeCssVariables.font.color.primary};
|
||||
font-size: ${themeCssVariables.font.size.lg};
|
||||
font-weight: ${themeCssVariables.font.weight.semiBold};
|
||||
margin: 0;
|
||||
`;
|
||||
|
||||
const StyledDescription = styled.p`
|
||||
color: ${themeCssVariables.font.color.secondary};
|
||||
font-size: ${themeCssVariables.font.size.sm};
|
||||
line-height: 1.5;
|
||||
margin: 0;
|
||||
`;
|
||||
|
||||
const StyledAddressContainer = styled.div`
|
||||
align-items: center;
|
||||
background: ${themeCssVariables.background.transparent.lighter};
|
||||
border: 1px solid ${themeCssVariables.border.color.medium};
|
||||
border-radius: ${themeCssVariables.border.radius.sm};
|
||||
display: flex;
|
||||
gap: ${themeCssVariables.spacing[2]};
|
||||
padding: ${themeCssVariables.spacing[3]} ${themeCssVariables.spacing[4]};
|
||||
`;
|
||||
|
||||
const StyledAddress = styled.span`
|
||||
color: ${themeCssVariables.font.color.primary};
|
||||
flex: 1;
|
||||
font-family: monospace;
|
||||
font-size: ${themeCssVariables.font.size.sm};
|
||||
overflow: hidden;
|
||||
text-overflow: ellipsis;
|
||||
white-space: nowrap;
|
||||
`;
|
||||
|
||||
const StyledInstructions = styled.ol`
|
||||
color: ${themeCssVariables.font.color.secondary};
|
||||
font-size: ${themeCssVariables.font.size.sm};
|
||||
line-height: 1.6;
|
||||
margin: 0;
|
||||
padding-left: ${themeCssVariables.spacing[6]};
|
||||
`;
|
||||
|
||||
const StyledButtonContainer = styled.div`
|
||||
display: flex;
|
||||
justify-content: flex-end;
|
||||
`;
|
||||
|
||||
type SettingsAccountsEmailForwardingModalProps = {
|
||||
forwardingAddress: string;
|
||||
};
|
||||
|
||||
export const SettingsAccountsEmailForwardingModal = ({
|
||||
forwardingAddress,
|
||||
}: SettingsAccountsEmailForwardingModalProps) => {
|
||||
const { t } = useLingui();
|
||||
const { copyToClipboard } = useCopyToClipboard();
|
||||
const { closeModal } = useModal();
|
||||
|
||||
return (
|
||||
<ModalStatefulWrapper
|
||||
modalInstanceId={EMAIL_FORWARDING_MODAL_ID}
|
||||
size="medium"
|
||||
isClosable
|
||||
>
|
||||
<StyledModalContent>
|
||||
<StyledTitle>{t`Email Forwarding Channel Created`}</StyledTitle>
|
||||
<StyledDescription>
|
||||
{t`Add this address to a Google Group or Microsoft 365 shared mailbox to start receiving emails in Twenty.`}
|
||||
</StyledDescription>
|
||||
|
||||
<StyledAddressContainer>
|
||||
<IconMail size={16} />
|
||||
<StyledAddress>{forwardingAddress}</StyledAddress>
|
||||
<Button
|
||||
Icon={IconCopy}
|
||||
title={t`Copy`}
|
||||
variant="secondary"
|
||||
size="small"
|
||||
onClick={() =>
|
||||
copyToClipboard(
|
||||
forwardingAddress,
|
||||
t`Forwarding address copied to clipboard`,
|
||||
)
|
||||
}
|
||||
/>
|
||||
</StyledAddressContainer>
|
||||
|
||||
<StyledInstructions>
|
||||
<li>{t`Copy the forwarding address above`}</li>
|
||||
<li>{t`Add it as a member of your Google Group or Microsoft 365 shared mailbox`}</li>
|
||||
<li>{t`Emails sent to the group will automatically appear in Twenty`}</li>
|
||||
</StyledInstructions>
|
||||
|
||||
<StyledButtonContainer>
|
||||
<Button
|
||||
title={t`Done`}
|
||||
variant="primary"
|
||||
size="small"
|
||||
onClick={() => closeModal(EMAIL_FORWARDING_MODAL_ID)}
|
||||
/>
|
||||
</StyledButtonContainer>
|
||||
</StyledModalContent>
|
||||
</ModalStatefulWrapper>
|
||||
);
|
||||
};
|
||||
+76
-27
@@ -3,16 +3,23 @@ import { isGoogleMessagingEnabledState } from '@/client-config/states/isGoogleMe
|
||||
import { isImapSmtpCaldavEnabledState } from '@/client-config/states/isImapSmtpCaldavEnabledState';
|
||||
import { isMicrosoftCalendarEnabledState } from '@/client-config/states/isMicrosoftCalendarEnabledState';
|
||||
import { isMicrosoftMessagingEnabledState } from '@/client-config/states/isMicrosoftMessagingEnabledState';
|
||||
import {
|
||||
EMAIL_FORWARDING_MODAL_ID,
|
||||
SettingsAccountsEmailForwardingModal,
|
||||
} from '@/settings/accounts/components/SettingsAccountsEmailForwardingModal';
|
||||
import { useCreateEmailForwardingChannel } from '@/settings/accounts/hooks/useCreateEmailForwardingChannel';
|
||||
import { useTriggerApisOAuth } from '@/settings/accounts/hooks/useTriggerApiOAuth';
|
||||
import { SettingsCard } from '@/settings/components/SettingsCard';
|
||||
import { useSnackBar } from '@/ui/feedback/snack-bar-manager/hooks/useSnackBar';
|
||||
import { useModal } from '@/ui/layout/modal/hooks/useModal';
|
||||
import { styled } from '@linaria/react';
|
||||
import { useLingui } from '@lingui/react/macro';
|
||||
import { useAtomStateValue } from '@/ui/utilities/state/jotai/hooks/useAtomStateValue';
|
||||
import { useCallback, useContext, useState } from 'react';
|
||||
import { ConnectedAccountProvider, SettingsPath } from 'twenty-shared/types';
|
||||
import { getSettingsPath } from 'twenty-shared/utils';
|
||||
import { IconAt, IconGoogle, IconMicrosoft } from 'twenty-ui/display';
|
||||
import { IconAt, IconGoogle, IconMail, IconMicrosoft } from 'twenty-ui/display';
|
||||
import { UndecoratedLink } from 'twenty-ui/navigation';
|
||||
import { useContext } from 'react';
|
||||
import { ThemeContext, themeCssVariables } from 'twenty-ui/theme-constants';
|
||||
|
||||
const StyledCardsContainer = styled.div`
|
||||
@@ -26,6 +33,14 @@ export const SettingsAccountsListEmptyStateCard = () => {
|
||||
const { triggerApisOAuth } = useTriggerApisOAuth();
|
||||
|
||||
const { t } = useLingui();
|
||||
const { enqueueErrorSnackBar } = useSnackBar();
|
||||
const { openModal } = useModal();
|
||||
const { createEmailForwardingChannel, loading: isCreatingForwarding } =
|
||||
useCreateEmailForwardingChannel();
|
||||
const [forwardingAddress, setForwardingAddress] = useState<string | null>(
|
||||
null,
|
||||
);
|
||||
|
||||
const isGoogleMessagingEnabled = useAtomStateValue(
|
||||
isGoogleMessagingEnabledState,
|
||||
);
|
||||
@@ -45,34 +60,68 @@ export const SettingsAccountsListEmptyStateCard = () => {
|
||||
isImapSmtpCaldavEnabledState,
|
||||
);
|
||||
|
||||
const handleCreateEmailForwardingChannel = useCallback(async () => {
|
||||
try {
|
||||
const result = await createEmailForwardingChannel();
|
||||
|
||||
const address =
|
||||
result.data?.createEmailForwardingChannel.forwardingAddress;
|
||||
|
||||
if (address) {
|
||||
setForwardingAddress(address);
|
||||
openModal(EMAIL_FORWARDING_MODAL_ID);
|
||||
}
|
||||
} catch {
|
||||
enqueueErrorSnackBar({
|
||||
message: t`Failed to create email forwarding channel. Email forwarding may not be configured on this server.`,
|
||||
});
|
||||
}
|
||||
}, [createEmailForwardingChannel, openModal, enqueueErrorSnackBar, t]);
|
||||
|
||||
return (
|
||||
<StyledCardsContainer>
|
||||
{(isGoogleMessagingEnabled || isGoogleCalendarEnabled) && (
|
||||
<SettingsCard
|
||||
Icon={<IconGoogle size={theme.icon.size.md} />}
|
||||
title={t`Connect with Google`}
|
||||
onClick={() => triggerApisOAuth(ConnectedAccountProvider.GOOGLE)}
|
||||
/>
|
||||
)}
|
||||
|
||||
{(isMicrosoftMessagingEnabled || isMicrosoftCalendarEnabled) && (
|
||||
<SettingsCard
|
||||
Icon={<IconMicrosoft size={theme.icon.size.md} />}
|
||||
title={t`Connect with Microsoft`}
|
||||
onClick={() => triggerApisOAuth(ConnectedAccountProvider.MICROSOFT)}
|
||||
/>
|
||||
)}
|
||||
|
||||
{isImapSmtpCaldavEnabled && (
|
||||
<UndecoratedLink
|
||||
to={getSettingsPath(SettingsPath.NewImapSmtpCaldavConnection)}
|
||||
>
|
||||
<>
|
||||
<StyledCardsContainer>
|
||||
{(isGoogleMessagingEnabled || isGoogleCalendarEnabled) && (
|
||||
<SettingsCard
|
||||
Icon={<IconAt size={theme.icon.size.md} />}
|
||||
title={t`Connect Account`}
|
||||
Icon={<IconGoogle size={theme.icon.size.md} />}
|
||||
title={t`Connect with Google`}
|
||||
onClick={() => triggerApisOAuth(ConnectedAccountProvider.GOOGLE)}
|
||||
/>
|
||||
</UndecoratedLink>
|
||||
)}
|
||||
|
||||
{(isMicrosoftMessagingEnabled || isMicrosoftCalendarEnabled) && (
|
||||
<SettingsCard
|
||||
Icon={<IconMicrosoft size={theme.icon.size.md} />}
|
||||
title={t`Connect with Microsoft`}
|
||||
onClick={() => triggerApisOAuth(ConnectedAccountProvider.MICROSOFT)}
|
||||
/>
|
||||
)}
|
||||
|
||||
{isImapSmtpCaldavEnabled && (
|
||||
<UndecoratedLink
|
||||
to={getSettingsPath(SettingsPath.NewImapSmtpCaldavConnection)}
|
||||
>
|
||||
<SettingsCard
|
||||
Icon={<IconAt size={theme.icon.size.md} />}
|
||||
title={t`Connect Account`}
|
||||
/>
|
||||
</UndecoratedLink>
|
||||
)}
|
||||
|
||||
<SettingsCard
|
||||
Icon={<IconMail size={theme.icon.size.md} />}
|
||||
title={t`Add Email Forwarding`}
|
||||
description={t`Receive emails from a shared mailbox or Google Group`}
|
||||
disabled={isCreatingForwarding}
|
||||
onClick={handleCreateEmailForwardingChannel}
|
||||
/>
|
||||
</StyledCardsContainer>
|
||||
|
||||
{forwardingAddress && (
|
||||
<SettingsAccountsEmailForwardingModal
|
||||
forwardingAddress={forwardingAddress}
|
||||
/>
|
||||
)}
|
||||
</StyledCardsContainer>
|
||||
</>
|
||||
);
|
||||
};
|
||||
|
||||
+18
@@ -0,0 +1,18 @@
|
||||
import { gql } from '@apollo/client';
|
||||
|
||||
export const CREATE_EMAIL_FORWARDING_CHANNEL = gql`
|
||||
mutation CreateEmailForwardingChannel {
|
||||
createEmailForwardingChannel {
|
||||
messageChannel {
|
||||
id
|
||||
handle
|
||||
visibility
|
||||
type
|
||||
isSyncEnabled
|
||||
excludeGroupEmails
|
||||
contactAutoCreationPolicy
|
||||
}
|
||||
forwardingAddress
|
||||
}
|
||||
}
|
||||
`;
|
||||
+27
@@ -0,0 +1,27 @@
|
||||
import { useMutation } from '@apollo/client/react';
|
||||
|
||||
import { CREATE_EMAIL_FORWARDING_CHANNEL } from '@/settings/accounts/graphql/mutations/createEmailForwardingChannel';
|
||||
|
||||
type CreateEmailForwardingChannelResult = {
|
||||
createEmailForwardingChannel: {
|
||||
messageChannel: {
|
||||
id: string;
|
||||
handle: string;
|
||||
visibility: string;
|
||||
type: string;
|
||||
isSyncEnabled: boolean;
|
||||
excludeGroupEmails: boolean;
|
||||
contactAutoCreationPolicy: string;
|
||||
};
|
||||
forwardingAddress: string;
|
||||
};
|
||||
};
|
||||
|
||||
export const useCreateEmailForwardingChannel = () => {
|
||||
const [createEmailForwardingChannel, { loading, error }] =
|
||||
useMutation<CreateEmailForwardingChannelResult>(
|
||||
CREATE_EMAIL_FORWARDING_CHANNEL,
|
||||
);
|
||||
|
||||
return { createEmailForwardingChannel, loading, error };
|
||||
};
|
||||
@@ -30,6 +30,7 @@ const PROVIDERS_ICON_MAPPING = {
|
||||
[ConnectedAccountProvider.IMAP_SMTP_CALDAV]: IconMail,
|
||||
[ConnectedAccountProvider.OIDC]: IconMail,
|
||||
[ConnectedAccountProvider.SAML]: IconMail,
|
||||
[ConnectedAccountProvider.EMAIL_FORWARDING]: IconMail,
|
||||
default: IconMail,
|
||||
},
|
||||
CALENDAR: {
|
||||
|
||||
@@ -1127,6 +1127,8 @@ export class AuthService {
|
||||
return [];
|
||||
case ConnectedAccountProvider.IMAP_SMTP_CALDAV:
|
||||
return [];
|
||||
case ConnectedAccountProvider.EMAIL_FORWARDING:
|
||||
return [];
|
||||
default:
|
||||
throw new Error(
|
||||
`Unsupported connected account provider: ${provider satisfies never}`,
|
||||
|
||||
@@ -1622,6 +1622,43 @@ export class ConfigVariables {
|
||||
@IsOptional()
|
||||
AWS_SES_ACCOUNT_ID: string;
|
||||
|
||||
@ConfigVariablesMetadata({
|
||||
group: ConfigVariablesGroup.AWS_SES_SETTINGS,
|
||||
description:
|
||||
'Domain used for inbound email forwarding (the right-hand side of ch_xxx@<domain>). Required to enable email forwarding channels.',
|
||||
type: ConfigVariableType.STRING,
|
||||
})
|
||||
@IsOptional()
|
||||
INBOUND_EMAIL_DOMAIN: string;
|
||||
|
||||
@ConfigVariablesMetadata({
|
||||
group: ConfigVariablesGroup.AWS_SES_SETTINGS,
|
||||
description:
|
||||
'S3 bucket where SES inbound action writes raw messages. Required to enable email forwarding channels.',
|
||||
type: ConfigVariableType.STRING,
|
||||
})
|
||||
@IsOptional()
|
||||
INBOUND_EMAIL_S3_BUCKET: string;
|
||||
|
||||
@ConfigVariablesMetadata({
|
||||
group: ConfigVariablesGroup.AWS_SES_SETTINGS,
|
||||
description:
|
||||
'Region for the inbound-email S3 bucket. Defaults to AWS_SES_REGION when unset.',
|
||||
type: ConfigVariableType.STRING,
|
||||
})
|
||||
@IsOptional()
|
||||
INBOUND_EMAIL_S3_REGION: string;
|
||||
|
||||
@ConfigVariablesMetadata({
|
||||
group: ConfigVariablesGroup.AWS_SES_SETTINGS,
|
||||
description:
|
||||
'Maximum number of S3 keys listed per poll of the inbound-email bucket.',
|
||||
type: ConfigVariableType.NUMBER,
|
||||
})
|
||||
@CastToPositiveNumber()
|
||||
@IsOptional()
|
||||
INBOUND_EMAIL_POLL_BATCH_SIZE: number = 500;
|
||||
|
||||
@ConfigVariablesMetadata({
|
||||
group: ConfigVariablesGroup.ADVANCED_SETTINGS,
|
||||
description: 'Timeout in milliseconds for primary database queries',
|
||||
|
||||
+12
@@ -0,0 +1,12 @@
|
||||
import { Field, ObjectType } from '@nestjs/graphql';
|
||||
|
||||
import { MessageChannelDTO } from 'src/engine/metadata-modules/message-channel/dtos/message-channel.dto';
|
||||
|
||||
@ObjectType('CreateEmailForwardingChannelOutput')
|
||||
export class CreateEmailForwardingChannelOutput {
|
||||
@Field(() => MessageChannelDTO)
|
||||
messageChannel: MessageChannelDTO;
|
||||
|
||||
@Field()
|
||||
forwardingAddress: string;
|
||||
}
|
||||
+67
@@ -1,21 +1,33 @@
|
||||
import { randomBytes } from 'crypto';
|
||||
|
||||
import { Injectable } from '@nestjs/common';
|
||||
import { InjectRepository } from '@nestjs/typeorm';
|
||||
|
||||
import { isNonEmptyString } from '@sniptt/guards';
|
||||
import { In, Repository } from 'typeorm';
|
||||
|
||||
import {
|
||||
ConnectedAccountProvider,
|
||||
MessageChannelContactAutoCreationPolicy,
|
||||
MessageChannelPendingGroupEmailsAction,
|
||||
MessageChannelSyncStage,
|
||||
MessageChannelType,
|
||||
MessageChannelVisibility,
|
||||
} from 'twenty-shared/types';
|
||||
|
||||
import { TwentyConfigService } from 'src/engine/core-modules/twenty-config/twenty-config.service';
|
||||
import { ConnectedAccountMetadataService } from 'src/engine/metadata-modules/connected-account/connected-account-metadata.service';
|
||||
import { CreateEmailForwardingChannelOutput } from 'src/engine/metadata-modules/message-channel/dtos/create-email-forwarding-channel.output';
|
||||
import { MessageChannelDTO } from 'src/engine/metadata-modules/message-channel/dtos/message-channel.dto';
|
||||
import { MessageChannelEntity } from 'src/engine/metadata-modules/message-channel/entities/message-channel.entity';
|
||||
import {
|
||||
MessageChannelException,
|
||||
MessageChannelExceptionCode,
|
||||
} from 'src/engine/metadata-modules/message-channel/message-channel.exception';
|
||||
import {
|
||||
INBOUND_EMAIL_LOCAL_PART_PREFIX,
|
||||
INBOUND_EMAIL_LOCAL_PART_RANDOM_BYTES,
|
||||
} from 'src/modules/messaging/message-import-manager/drivers/inbound-email/constants/inbound-email.constants';
|
||||
|
||||
@Injectable()
|
||||
export class MessageChannelMetadataService {
|
||||
@@ -23,6 +35,7 @@ export class MessageChannelMetadataService {
|
||||
@InjectRepository(MessageChannelEntity)
|
||||
private readonly repository: Repository<MessageChannelEntity>,
|
||||
private readonly connectedAccountMetadataService: ConnectedAccountMetadataService,
|
||||
private readonly twentyConfigService: TwentyConfigService,
|
||||
) {}
|
||||
|
||||
async findAll(workspaceId: string): Promise<MessageChannelDTO[]> {
|
||||
@@ -172,6 +185,60 @@ export class MessageChannelMetadataService {
|
||||
return this.repository.findOneOrFail({ where: { id, workspaceId } });
|
||||
}
|
||||
|
||||
async createEmailForwardingChannel({
|
||||
userWorkspaceId,
|
||||
workspaceId,
|
||||
}: {
|
||||
userWorkspaceId: string;
|
||||
workspaceId: string;
|
||||
}): Promise<CreateEmailForwardingChannelOutput> {
|
||||
const inboundEmailDomain = this.twentyConfigService.get(
|
||||
'INBOUND_EMAIL_DOMAIN',
|
||||
);
|
||||
|
||||
if (!isNonEmptyString(inboundEmailDomain)) {
|
||||
throw new MessageChannelException(
|
||||
'Email forwarding is not configured: INBOUND_EMAIL_DOMAIN is not set',
|
||||
MessageChannelExceptionCode.EMAIL_FORWARDING_NOT_CONFIGURED,
|
||||
);
|
||||
}
|
||||
|
||||
const localPart =
|
||||
INBOUND_EMAIL_LOCAL_PART_PREFIX +
|
||||
randomBytes(INBOUND_EMAIL_LOCAL_PART_RANDOM_BYTES).toString('hex');
|
||||
|
||||
const forwardingAddress = `${localPart}@${inboundEmailDomain}`;
|
||||
|
||||
const connectedAccount =
|
||||
await this.connectedAccountMetadataService.create({
|
||||
workspaceId,
|
||||
handle: forwardingAddress,
|
||||
provider: ConnectedAccountProvider.EMAIL_FORWARDING,
|
||||
userWorkspaceId,
|
||||
accessToken: null,
|
||||
refreshToken: null,
|
||||
});
|
||||
|
||||
const messageChannel = await this.create({
|
||||
workspaceId,
|
||||
handle: forwardingAddress,
|
||||
connectedAccountId: connectedAccount.id,
|
||||
type: MessageChannelType.EMAIL_FORWARDING,
|
||||
visibility: MessageChannelVisibility.SHARE_EVERYTHING,
|
||||
syncStage: MessageChannelSyncStage.PENDING_CONFIGURATION,
|
||||
isSyncEnabled: true,
|
||||
isContactAutoCreationEnabled: true,
|
||||
contactAutoCreationPolicy:
|
||||
MessageChannelContactAutoCreationPolicy.SENT_AND_RECEIVED,
|
||||
excludeGroupEmails: false,
|
||||
excludeNonProfessionalEmails: false,
|
||||
pendingGroupEmailsAction:
|
||||
MessageChannelPendingGroupEmailsAction.NONE,
|
||||
});
|
||||
|
||||
return { messageChannel, forwardingAddress };
|
||||
}
|
||||
|
||||
async delete({
|
||||
id,
|
||||
workspaceId,
|
||||
|
||||
+3
@@ -8,6 +8,7 @@ export enum MessageChannelExceptionCode {
|
||||
MESSAGE_CHANNEL_NOT_FOUND = 'MESSAGE_CHANNEL_NOT_FOUND',
|
||||
INVALID_MESSAGE_CHANNEL_INPUT = 'INVALID_MESSAGE_CHANNEL_INPUT',
|
||||
MESSAGE_CHANNEL_OWNERSHIP_VIOLATION = 'MESSAGE_CHANNEL_OWNERSHIP_VIOLATION',
|
||||
EMAIL_FORWARDING_NOT_CONFIGURED = 'EMAIL_FORWARDING_NOT_CONFIGURED',
|
||||
}
|
||||
|
||||
const getMessageChannelExceptionUserFriendlyMessage = (
|
||||
@@ -20,6 +21,8 @@ const getMessageChannelExceptionUserFriendlyMessage = (
|
||||
return msg`Invalid message channel input.`;
|
||||
case MessageChannelExceptionCode.MESSAGE_CHANNEL_OWNERSHIP_VIOLATION:
|
||||
return msg`You do not have access to this message channel.`;
|
||||
case MessageChannelExceptionCode.EMAIL_FORWARDING_NOT_CONFIGURED:
|
||||
return msg`Email forwarding is not configured on this server.`;
|
||||
default:
|
||||
assertUnreachable(code);
|
||||
}
|
||||
|
||||
+13
@@ -19,6 +19,7 @@ import { AuthUserWorkspaceId } from 'src/engine/decorators/auth/auth-user-worksp
|
||||
import { AuthWorkspace } from 'src/engine/decorators/auth/auth-workspace.decorator';
|
||||
import { NoPermissionGuard } from 'src/engine/guards/no-permission.guard';
|
||||
import { WorkspaceAuthGuard } from 'src/engine/guards/workspace-auth.guard';
|
||||
import { CreateEmailForwardingChannelOutput } from 'src/engine/metadata-modules/message-channel/dtos/create-email-forwarding-channel.output';
|
||||
import { MessageChannelDTO } from 'src/engine/metadata-modules/message-channel/dtos/message-channel.dto';
|
||||
import { UpdateMessageChannelInput } from 'src/engine/metadata-modules/message-channel/dtos/update-message-channel.input';
|
||||
import {
|
||||
@@ -130,4 +131,16 @@ export class MessageChannelResolver {
|
||||
data: input.update,
|
||||
});
|
||||
}
|
||||
|
||||
@Mutation(() => CreateEmailForwardingChannelOutput)
|
||||
@UseGuards(NoPermissionGuard)
|
||||
async createEmailForwardingChannel(
|
||||
@AuthWorkspace() workspace: WorkspaceEntity,
|
||||
@AuthUserWorkspaceId() userWorkspaceId: string,
|
||||
): Promise<CreateEmailForwardingChannelOutput> {
|
||||
return this.messageChannelMetadataService.createEmailForwardingChannel({
|
||||
userWorkspaceId,
|
||||
workspaceId: workspace.id,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
+2
@@ -23,6 +23,8 @@ export const messageChannelGraphqlApiExceptionHandler = (error: Error) => {
|
||||
throw new UserInputError(error);
|
||||
case MessageChannelExceptionCode.MESSAGE_CHANNEL_OWNERSHIP_VIOLATION:
|
||||
throw new ForbiddenError(error);
|
||||
case MessageChannelExceptionCode.EMAIL_FORWARDING_NOT_CONFIGURED:
|
||||
throw new UserInputError(error);
|
||||
default: {
|
||||
return assertUnreachable(error.code);
|
||||
}
|
||||
|
||||
+7
@@ -286,6 +286,13 @@ export const buildMessageChannelStandardFlatFieldMetadatas = ({
|
||||
position: 1,
|
||||
color: 'blue',
|
||||
},
|
||||
{
|
||||
id: '20202020-7f22-4e58-aa33-9c3e2c72ab10',
|
||||
value: MessageChannelType.EMAIL_FORWARDING,
|
||||
label: i18nLabel(msg`Email Forwarding`),
|
||||
position: 2,
|
||||
color: 'turquoise',
|
||||
},
|
||||
],
|
||||
},
|
||||
standardObjectMetadataRelatedEntityIds,
|
||||
|
||||
+1
@@ -43,6 +43,7 @@ export class EmailAliasManagerService {
|
||||
case ConnectedAccountProvider.IMAP_SMTP_CALDAV:
|
||||
case ConnectedAccountProvider.OIDC:
|
||||
case ConnectedAccountProvider.SAML:
|
||||
case ConnectedAccountProvider.EMAIL_FORWARDING:
|
||||
handleAliases = [];
|
||||
break;
|
||||
default:
|
||||
|
||||
+2
@@ -117,6 +117,7 @@ export class ConnectedAccountRefreshTokensService {
|
||||
case ConnectedAccountProvider.IMAP_SMTP_CALDAV:
|
||||
case ConnectedAccountProvider.OIDC:
|
||||
case ConnectedAccountProvider.SAML:
|
||||
case ConnectedAccountProvider.EMAIL_FORWARDING:
|
||||
return true;
|
||||
default:
|
||||
return assertUnreachable(
|
||||
@@ -144,6 +145,7 @@ export class ConnectedAccountRefreshTokensService {
|
||||
case ConnectedAccountProvider.IMAP_SMTP_CALDAV:
|
||||
case ConnectedAccountProvider.OIDC:
|
||||
case ConnectedAccountProvider.SAML:
|
||||
case ConnectedAccountProvider.EMAIL_FORWARDING:
|
||||
throw new ConnectedAccountRefreshAccessTokenException(
|
||||
`Token refresh is not supported for ${connectedAccount.provider} provider for connected account ${connectedAccount.id} in workspace ${workspaceId}`,
|
||||
ConnectedAccountRefreshAccessTokenExceptionCode.PROVIDER_NOT_SUPPORTED,
|
||||
|
||||
+33
@@ -0,0 +1,33 @@
|
||||
import { Command, CommandRunner } from 'nest-commander';
|
||||
|
||||
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 {
|
||||
MESSAGING_INBOUND_EMAIL_POLL_CRON_PATTERN,
|
||||
MessagingInboundEmailPollCronJob,
|
||||
} from 'src/modules/messaging/message-import-manager/crons/jobs/messaging-inbound-email-poll.cron.job';
|
||||
|
||||
@Command({
|
||||
name: 'cron:messaging:inbound-email-poll',
|
||||
description:
|
||||
'Starts a cron job to poll the SES inbound S3 bucket and enqueue one import job per message.',
|
||||
})
|
||||
export class MessagingInboundEmailPollCronCommand extends CommandRunner {
|
||||
constructor(
|
||||
@InjectMessageQueue(MessageQueue.cronQueue)
|
||||
private readonly messageQueueService: MessageQueueService,
|
||||
) {
|
||||
super();
|
||||
}
|
||||
|
||||
async run(): Promise<void> {
|
||||
await this.messageQueueService.addCron<undefined>({
|
||||
jobName: MessagingInboundEmailPollCronJob.name,
|
||||
data: undefined,
|
||||
options: {
|
||||
repeat: { pattern: MESSAGING_INBOUND_EMAIL_POLL_CRON_PATTERN },
|
||||
},
|
||||
});
|
||||
}
|
||||
}
|
||||
+69
@@ -0,0 +1,69 @@
|
||||
import { Logger } from '@nestjs/common';
|
||||
|
||||
import { SentryCronMonitor } from 'src/engine/core-modules/cron/sentry-cron-monitor.decorator';
|
||||
import { ExceptionHandlerService } from 'src/engine/core-modules/exception-handler/exception-handler.service';
|
||||
import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator';
|
||||
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';
|
||||
import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service';
|
||||
import { TwentyConfigService } from 'src/engine/core-modules/twenty-config/twenty-config.service';
|
||||
import { InboundEmailS3ClientProvider } from 'src/modules/messaging/message-import-manager/drivers/inbound-email/providers/inbound-email-s3-client.provider';
|
||||
import { InboundEmailStorageService } from 'src/modules/messaging/message-import-manager/drivers/inbound-email/services/inbound-email-storage.service';
|
||||
import {
|
||||
MessagingInboundEmailImportJob,
|
||||
type MessagingInboundEmailImportJobData,
|
||||
} from 'src/modules/messaging/message-import-manager/jobs/messaging-inbound-email-import.job';
|
||||
|
||||
// BullMQ cron patterns are minute-based at the finest; every minute is
|
||||
// fine for a forwarding inbox since SES writes are eventually consistent
|
||||
// anyway. Single-leader election comes from BullMQ's repeatable-job dedupe,
|
||||
// so multiple workers can run this cron safely.
|
||||
export const MESSAGING_INBOUND_EMAIL_POLL_CRON_PATTERN = '* * * * *';
|
||||
|
||||
@Processor(MessageQueue.cronQueue)
|
||||
export class MessagingInboundEmailPollCronJob {
|
||||
private readonly logger = new Logger(MessagingInboundEmailPollCronJob.name);
|
||||
|
||||
constructor(
|
||||
@InjectMessageQueue(MessageQueue.messagingQueue)
|
||||
private readonly messageQueueService: MessageQueueService,
|
||||
private readonly inboundEmailS3ClientProvider: InboundEmailS3ClientProvider,
|
||||
private readonly inboundEmailStorageService: InboundEmailStorageService,
|
||||
private readonly twentyConfigService: TwentyConfigService,
|
||||
private readonly exceptionHandlerService: ExceptionHandlerService,
|
||||
) {}
|
||||
|
||||
@Process(MessagingInboundEmailPollCronJob.name)
|
||||
@SentryCronMonitor(
|
||||
MessagingInboundEmailPollCronJob.name,
|
||||
MESSAGING_INBOUND_EMAIL_POLL_CRON_PATTERN,
|
||||
)
|
||||
async handle(): Promise<void> {
|
||||
if (!this.inboundEmailS3ClientProvider.isConfigured()) {
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
const batchSize = this.twentyConfigService.get(
|
||||
'INBOUND_EMAIL_POLL_BATCH_SIZE',
|
||||
);
|
||||
const keys = await this.inboundEmailStorageService.listIncoming(batchSize);
|
||||
|
||||
if (keys.length === 0) {
|
||||
return;
|
||||
}
|
||||
|
||||
this.logger.log(`Enqueuing ${keys.length} inbound emails for import`);
|
||||
|
||||
for (const s3Key of keys) {
|
||||
await this.messageQueueService.add<MessagingInboundEmailImportJobData>(
|
||||
MessagingInboundEmailImportJob.name,
|
||||
{ s3Key },
|
||||
);
|
||||
}
|
||||
} catch (error) {
|
||||
this.exceptionHandlerService.captureExceptions([error]);
|
||||
}
|
||||
}
|
||||
}
|
||||
+5
-2
@@ -3,7 +3,7 @@ import { InjectDataSource, InjectRepository } from '@nestjs/typeorm';
|
||||
import { WorkspaceActivationStatus } from 'twenty-shared/workspace';
|
||||
import { DataSource, Repository } from 'typeorm';
|
||||
|
||||
import { MessageChannelSyncStage } from 'twenty-shared/types';
|
||||
import { MessageChannelSyncStage, MessageChannelType } from 'twenty-shared/types';
|
||||
import { SentryCronMonitor } from 'src/engine/core-modules/cron/sentry-cron-monitor.decorator';
|
||||
import { ExceptionHandlerService } from 'src/engine/core-modules/exception-handler/exception-handler.service';
|
||||
import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator';
|
||||
@@ -47,9 +47,12 @@ export class MessagingMessageListFetchCronJob {
|
||||
try {
|
||||
const now = new Date().toISOString();
|
||||
|
||||
// Email forwarding channels are driven by the S3 poll cron, not by
|
||||
// this "fetch from provider" path — skip them explicitly so their
|
||||
// syncStage never transitions into MESSAGE_LIST_FETCH_SCHEDULED.
|
||||
const [messageChannels] = await this.coreDataSource.query(
|
||||
`UPDATE core."messageChannel" SET "syncStage" = '${MessageChannelSyncStage.MESSAGE_LIST_FETCH_SCHEDULED}', "syncStageStartedAt" = COALESCE("syncStageStartedAt", '${now}')
|
||||
WHERE "workspaceId" = '${activeWorkspace.id}' AND "isSyncEnabled" = true AND "syncStage" = '${MessageChannelSyncStage.MESSAGE_LIST_FETCH_PENDING}' RETURNING *`,
|
||||
WHERE "workspaceId" = '${activeWorkspace.id}' AND "isSyncEnabled" = true AND "syncStage" = '${MessageChannelSyncStage.MESSAGE_LIST_FETCH_PENDING}' AND "type" != '${MessageChannelType.EMAIL_FORWARDING}' RETURNING *`,
|
||||
);
|
||||
|
||||
for (const messageChannel of messageChannels) {
|
||||
|
||||
+20
@@ -0,0 +1,20 @@
|
||||
// S3 key prefixes used to shard incoming/processed/unmatched/failed mail.
|
||||
// The bucket itself IS the work queue: presence of a key in "incoming/"
|
||||
// means the object is pending import. Moving an object to any other prefix
|
||||
// removes it from the queue, which gives us at-least-once semantics without
|
||||
// needing a separate dedupe store.
|
||||
export const INBOUND_EMAIL_S3_PREFIXES = {
|
||||
incoming: 'incoming/',
|
||||
processed: 'processed/',
|
||||
unmatched: 'unmatched/',
|
||||
failed: 'failed/',
|
||||
} as const;
|
||||
|
||||
// Custom header stamped on outbound mail so we can drop echoes when group
|
||||
// replies hit the inbound pipeline. The value is the workspace id that sent
|
||||
// the message, which lets us scope the loop check to a single workspace.
|
||||
export const X_TWENTY_ORIGIN_HEADER = 'x-twenty-origin';
|
||||
|
||||
// Local part prefix for forwarding-channel addresses: "ch_" + 24 hex chars.
|
||||
export const INBOUND_EMAIL_LOCAL_PART_PREFIX = 'ch_';
|
||||
export const INBOUND_EMAIL_LOCAL_PART_RANDOM_BYTES = 12;
|
||||
+33
@@ -0,0 +1,33 @@
|
||||
import { Module } from '@nestjs/common';
|
||||
import { TypeOrmModule } from '@nestjs/typeorm';
|
||||
|
||||
import { TwentyConfigModule } from 'src/engine/core-modules/twenty-config/twenty-config.module';
|
||||
import { ConnectedAccountEntity } from 'src/engine/metadata-modules/connected-account/entities/connected-account.entity';
|
||||
import { MessageChannelEntity } from 'src/engine/metadata-modules/message-channel/entities/message-channel.entity';
|
||||
import { WorkspaceDataSourceModule } from 'src/engine/workspace-datasource/workspace-datasource.module';
|
||||
import { InboundEmailS3ClientProvider } from 'src/modules/messaging/message-import-manager/drivers/inbound-email/providers/inbound-email-s3-client.provider';
|
||||
import { InboundEmailParserService } from 'src/modules/messaging/message-import-manager/drivers/inbound-email/services/inbound-email-parser.service';
|
||||
import { InboundEmailStorageService } from 'src/modules/messaging/message-import-manager/drivers/inbound-email/services/inbound-email-storage.service';
|
||||
|
||||
// Thin driver module that provides the low-level primitives for inbound
|
||||
// email: S3 client, storage (list/get/move), and MIME parser. The import
|
||||
// orchestration service lives in MessagingImportManagerModule so it can
|
||||
// reuse the save-messages pipeline without a circular import.
|
||||
@Module({
|
||||
imports: [
|
||||
TwentyConfigModule,
|
||||
WorkspaceDataSourceModule,
|
||||
TypeOrmModule.forFeature([MessageChannelEntity, ConnectedAccountEntity]),
|
||||
],
|
||||
providers: [
|
||||
InboundEmailS3ClientProvider,
|
||||
InboundEmailStorageService,
|
||||
InboundEmailParserService,
|
||||
],
|
||||
exports: [
|
||||
InboundEmailS3ClientProvider,
|
||||
InboundEmailStorageService,
|
||||
InboundEmailParserService,
|
||||
],
|
||||
})
|
||||
export class MessagingInboundEmailDriverModule {}
|
||||
+91
@@ -0,0 +1,91 @@
|
||||
import { Injectable, Logger } from '@nestjs/common';
|
||||
|
||||
import { S3Client, type S3ClientConfig } from '@aws-sdk/client-s3';
|
||||
import { isNonEmptyString } from '@sniptt/guards';
|
||||
|
||||
import { TwentyConfigService } from 'src/engine/core-modules/twenty-config/twenty-config.service';
|
||||
|
||||
// Lazy S3 client: constructed on first call so the server boots cleanly in
|
||||
// workspaces that don't enable email forwarding. Credentials are reused from
|
||||
// the AWS_SES_* config block — SES inbound action already writes to this
|
||||
// bucket, so the same principal can read it.
|
||||
@Injectable()
|
||||
export class InboundEmailS3ClientProvider {
|
||||
private readonly logger = new Logger(InboundEmailS3ClientProvider.name);
|
||||
private s3Client: S3Client | null = null;
|
||||
|
||||
constructor(private readonly twentyConfigService: TwentyConfigService) {}
|
||||
|
||||
isConfigured(): boolean {
|
||||
const bucket = this.twentyConfigService.get('INBOUND_EMAIL_S3_BUCKET');
|
||||
const domain = this.twentyConfigService.get('INBOUND_EMAIL_DOMAIN');
|
||||
|
||||
return isNonEmptyString(bucket) && isNonEmptyString(domain);
|
||||
}
|
||||
|
||||
getBucket(): string {
|
||||
const bucket = this.twentyConfigService.get('INBOUND_EMAIL_S3_BUCKET');
|
||||
|
||||
if (!isNonEmptyString(bucket)) {
|
||||
throw new Error(
|
||||
'INBOUND_EMAIL_S3_BUCKET is not configured; email forwarding is disabled.',
|
||||
);
|
||||
}
|
||||
|
||||
return bucket;
|
||||
}
|
||||
|
||||
getDomain(): string {
|
||||
const domain = this.twentyConfigService.get('INBOUND_EMAIL_DOMAIN');
|
||||
|
||||
if (!isNonEmptyString(domain)) {
|
||||
throw new Error(
|
||||
'INBOUND_EMAIL_DOMAIN is not configured; email forwarding is disabled.',
|
||||
);
|
||||
}
|
||||
|
||||
return domain;
|
||||
}
|
||||
|
||||
getClient(): S3Client {
|
||||
if (this.s3Client) {
|
||||
return this.s3Client;
|
||||
}
|
||||
|
||||
const region =
|
||||
this.twentyConfigService.get('INBOUND_EMAIL_S3_REGION') ||
|
||||
this.twentyConfigService.get('AWS_SES_REGION');
|
||||
|
||||
if (!isNonEmptyString(region)) {
|
||||
throw new Error(
|
||||
'INBOUND_EMAIL_S3_REGION or AWS_SES_REGION must be set to use email forwarding.',
|
||||
);
|
||||
}
|
||||
|
||||
const config: S3ClientConfig = { region };
|
||||
|
||||
const accessKeyId = this.twentyConfigService.get('AWS_SES_ACCESS_KEY_ID');
|
||||
const secretAccessKey = this.twentyConfigService.get(
|
||||
'AWS_SES_SECRET_ACCESS_KEY',
|
||||
);
|
||||
const sessionToken = this.twentyConfigService.get('AWS_SES_SESSION_TOKEN');
|
||||
|
||||
if (
|
||||
isNonEmptyString(accessKeyId) &&
|
||||
isNonEmptyString(secretAccessKey) &&
|
||||
isNonEmptyString(sessionToken)
|
||||
) {
|
||||
config.credentials = { accessKeyId, secretAccessKey, sessionToken };
|
||||
} else if (
|
||||
isNonEmptyString(accessKeyId) &&
|
||||
isNonEmptyString(secretAccessKey)
|
||||
) {
|
||||
config.credentials = { accessKeyId, secretAccessKey };
|
||||
}
|
||||
|
||||
this.s3Client = new S3Client(config);
|
||||
this.logger.log(`Inbound-email S3 client initialized in region ${region}`);
|
||||
|
||||
return this.s3Client;
|
||||
}
|
||||
}
|
||||
+230
@@ -0,0 +1,230 @@
|
||||
import { type Repository } from 'typeorm';
|
||||
|
||||
import { type ConnectedAccountEntity } from 'src/engine/metadata-modules/connected-account/entities/connected-account.entity';
|
||||
import { type MessageChannelEntity } from 'src/engine/metadata-modules/message-channel/entities/message-channel.entity';
|
||||
import { type GlobalWorkspaceOrmManager } from 'src/engine/twenty-orm/global-workspace-datasource/global-workspace-orm.manager';
|
||||
import { type InboundEmailParserService } from 'src/modules/messaging/message-import-manager/drivers/inbound-email/services/inbound-email-parser.service';
|
||||
import { type InboundEmailS3ClientProvider } from 'src/modules/messaging/message-import-manager/drivers/inbound-email/providers/inbound-email-s3-client.provider';
|
||||
import { type InboundEmailStorageService } from 'src/modules/messaging/message-import-manager/drivers/inbound-email/services/inbound-email-storage.service';
|
||||
import {
|
||||
InboundEmailImportService,
|
||||
type InboundEmailImportOutcome,
|
||||
} from 'src/modules/messaging/message-import-manager/drivers/inbound-email/services/inbound-email-import.service';
|
||||
import { type MessagingSaveMessagesAndEnqueueContactCreationService } from 'src/modules/messaging/message-import-manager/services/messaging-save-messages-and-enqueue-contact-creation.service';
|
||||
import { MessageChannelType } from 'twenty-shared/types';
|
||||
|
||||
const TEST_S3_KEY = 'incoming/test-email-001';
|
||||
const TEST_DOMAIN = 'in.twenty.com';
|
||||
const TEST_WORKSPACE_ID = 'ws-123';
|
||||
const TEST_CHANNEL_ID = 'ch-456';
|
||||
const TEST_CONNECTED_ACCOUNT_ID = 'ca-789';
|
||||
const TEST_HANDLE = `ch_abc123@${TEST_DOMAIN}`;
|
||||
|
||||
const buildMockChannel = (): Partial<MessageChannelEntity> => ({
|
||||
id: TEST_CHANNEL_ID,
|
||||
handle: TEST_HANDLE,
|
||||
type: MessageChannelType.EMAIL_FORWARDING,
|
||||
workspaceId: TEST_WORKSPACE_ID,
|
||||
connectedAccountId: TEST_CONNECTED_ACCOUNT_ID,
|
||||
});
|
||||
|
||||
const buildMockConnectedAccount = (): Partial<ConnectedAccountEntity> => ({
|
||||
id: TEST_CONNECTED_ACCOUNT_ID,
|
||||
workspaceId: TEST_WORKSPACE_ID,
|
||||
});
|
||||
|
||||
const buildMockParsedResult = (
|
||||
overrides: { originWorkspaceId?: string | null } = {},
|
||||
) => ({
|
||||
parsed: {
|
||||
headers: [
|
||||
{
|
||||
key: 'delivered-to',
|
||||
originalKey: 'Delivered-To',
|
||||
value: TEST_HANDLE,
|
||||
},
|
||||
],
|
||||
to: [{ address: TEST_HANDLE, name: '' }],
|
||||
from: { address: 'sender@example.com', name: 'Sender' },
|
||||
subject: 'Test',
|
||||
text: 'body',
|
||||
messageId: '<test@example.com>',
|
||||
date: new Date().toISOString(),
|
||||
cc: [],
|
||||
bcc: [],
|
||||
html: '',
|
||||
attachments: [],
|
||||
},
|
||||
originWorkspaceId: overrides.originWorkspaceId ?? null,
|
||||
message: {
|
||||
externalId: `inbound-email:${TEST_S3_KEY}`,
|
||||
messageThreadExternalId: '<test@example.com>',
|
||||
headerMessageId: '<test@example.com>',
|
||||
subject: 'Test',
|
||||
text: 'body',
|
||||
receivedAt: new Date(),
|
||||
direction: 'incoming',
|
||||
attachments: [],
|
||||
participants: [],
|
||||
},
|
||||
});
|
||||
|
||||
describe('InboundEmailImportService', () => {
|
||||
let service: InboundEmailImportService;
|
||||
let s3ClientProvider: jest.Mocked<InboundEmailS3ClientProvider>;
|
||||
let storageService: jest.Mocked<InboundEmailStorageService>;
|
||||
let parserService: jest.Mocked<InboundEmailParserService>;
|
||||
let globalWorkspaceOrmManager: jest.Mocked<GlobalWorkspaceOrmManager>;
|
||||
let saveMessagesService: jest.Mocked<MessagingSaveMessagesAndEnqueueContactCreationService>;
|
||||
let messageChannelRepo: jest.Mocked<Repository<MessageChannelEntity>>;
|
||||
let connectedAccountRepo: jest.Mocked<Repository<ConnectedAccountEntity>>;
|
||||
|
||||
beforeEach(() => {
|
||||
s3ClientProvider = {
|
||||
isConfigured: jest.fn().mockReturnValue(true),
|
||||
getDomain: jest.fn().mockReturnValue(TEST_DOMAIN),
|
||||
getBucket: jest.fn(),
|
||||
getClient: jest.fn(),
|
||||
} as unknown as jest.Mocked<InboundEmailS3ClientProvider>;
|
||||
|
||||
storageService = {
|
||||
getRawMessage: jest.fn().mockResolvedValue(Buffer.from('raw-email')),
|
||||
moveToProcessed: jest.fn().mockResolvedValue(undefined),
|
||||
moveToUnmatched: jest.fn().mockResolvedValue(undefined),
|
||||
moveToFailed: jest.fn().mockResolvedValue(undefined),
|
||||
listIncoming: jest.fn(),
|
||||
} as unknown as jest.Mocked<InboundEmailStorageService>;
|
||||
|
||||
parserService = {
|
||||
parse: jest.fn().mockResolvedValue(buildMockParsedResult()),
|
||||
} as unknown as jest.Mocked<InboundEmailParserService>;
|
||||
|
||||
globalWorkspaceOrmManager = {
|
||||
executeInWorkspaceContext: jest
|
||||
.fn()
|
||||
.mockImplementation(async (fn: () => Promise<void>) => fn()),
|
||||
} as unknown as jest.Mocked<GlobalWorkspaceOrmManager>;
|
||||
|
||||
saveMessagesService = {
|
||||
saveMessagesAndEnqueueContactCreation: jest
|
||||
.fn()
|
||||
.mockResolvedValue(undefined),
|
||||
} as unknown as jest.Mocked<MessagingSaveMessagesAndEnqueueContactCreationService>;
|
||||
|
||||
messageChannelRepo = {
|
||||
findOne: jest.fn().mockResolvedValue(buildMockChannel()),
|
||||
} as unknown as jest.Mocked<Repository<MessageChannelEntity>>;
|
||||
|
||||
connectedAccountRepo = {
|
||||
findOne: jest.fn().mockResolvedValue(buildMockConnectedAccount()),
|
||||
} as unknown as jest.Mocked<Repository<ConnectedAccountEntity>>;
|
||||
|
||||
service = new InboundEmailImportService(
|
||||
s3ClientProvider,
|
||||
storageService,
|
||||
parserService,
|
||||
globalWorkspaceOrmManager,
|
||||
saveMessagesService,
|
||||
messageChannelRepo,
|
||||
connectedAccountRepo,
|
||||
);
|
||||
});
|
||||
|
||||
it('should return "imported" when everything succeeds', async () => {
|
||||
const outcome: InboundEmailImportOutcome =
|
||||
await service.importFromS3Key(TEST_S3_KEY);
|
||||
|
||||
expect(outcome).toEqual({
|
||||
kind: 'imported',
|
||||
workspaceId: TEST_WORKSPACE_ID,
|
||||
messageChannelId: TEST_CHANNEL_ID,
|
||||
});
|
||||
expect(storageService.moveToProcessed).toHaveBeenCalledWith(TEST_S3_KEY);
|
||||
expect(
|
||||
saveMessagesService.saveMessagesAndEnqueueContactCreation,
|
||||
).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('should return "unconfigured" when S3 is not configured', async () => {
|
||||
s3ClientProvider.isConfigured.mockReturnValue(false);
|
||||
|
||||
const outcome = await service.importFromS3Key(TEST_S3_KEY);
|
||||
|
||||
expect(outcome).toEqual({ kind: 'unconfigured' });
|
||||
expect(storageService.getRawMessage).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('should return "unmatched" when no channel matches the recipient', async () => {
|
||||
messageChannelRepo.findOne.mockResolvedValue(null);
|
||||
|
||||
const outcome = await service.importFromS3Key(TEST_S3_KEY);
|
||||
|
||||
expect(outcome.kind).toBe('unmatched');
|
||||
expect(storageService.moveToUnmatched).toHaveBeenCalledWith(TEST_S3_KEY);
|
||||
});
|
||||
|
||||
it('should return "loop_dropped" when X-Twenty-Origin matches workspace', async () => {
|
||||
parserService.parse.mockResolvedValue(
|
||||
buildMockParsedResult({ originWorkspaceId: TEST_WORKSPACE_ID }) as never,
|
||||
);
|
||||
|
||||
const outcome = await service.importFromS3Key(TEST_S3_KEY);
|
||||
|
||||
expect(outcome).toEqual({
|
||||
kind: 'loop_dropped',
|
||||
workspaceId: TEST_WORKSPACE_ID,
|
||||
});
|
||||
expect(storageService.moveToProcessed).toHaveBeenCalledWith(TEST_S3_KEY);
|
||||
});
|
||||
|
||||
it('should return "parse_failed" when download fails', async () => {
|
||||
storageService.getRawMessage.mockRejectedValue(
|
||||
new Error('S3 download error'),
|
||||
);
|
||||
|
||||
const outcome = await service.importFromS3Key(TEST_S3_KEY);
|
||||
|
||||
expect(outcome).toEqual({
|
||||
kind: 'parse_failed',
|
||||
error: 'S3 download error',
|
||||
});
|
||||
});
|
||||
|
||||
it('should return "parse_failed" when parsing fails', async () => {
|
||||
parserService.parse.mockRejectedValue(new Error('Invalid MIME'));
|
||||
|
||||
const outcome = await service.importFromS3Key(TEST_S3_KEY);
|
||||
|
||||
expect(outcome).toEqual({
|
||||
kind: 'parse_failed',
|
||||
error: 'Invalid MIME',
|
||||
});
|
||||
expect(storageService.moveToFailed).toHaveBeenCalledWith(TEST_S3_KEY);
|
||||
});
|
||||
|
||||
it('should return "persist_failed" when save throws', async () => {
|
||||
saveMessagesService.saveMessagesAndEnqueueContactCreation.mockRejectedValue(
|
||||
new Error('DB error'),
|
||||
);
|
||||
|
||||
const outcome = await service.importFromS3Key(TEST_S3_KEY);
|
||||
|
||||
expect(outcome).toEqual({
|
||||
kind: 'persist_failed',
|
||||
error: 'DB error',
|
||||
});
|
||||
expect(storageService.moveToFailed).toHaveBeenCalledWith(TEST_S3_KEY);
|
||||
});
|
||||
|
||||
it('should return "persist_failed" when connected account is missing', async () => {
|
||||
connectedAccountRepo.findOne.mockResolvedValue(null);
|
||||
|
||||
const outcome = await service.importFromS3Key(TEST_S3_KEY);
|
||||
|
||||
expect(outcome).toEqual({
|
||||
kind: 'persist_failed',
|
||||
error: 'connected_account_missing',
|
||||
});
|
||||
expect(storageService.moveToFailed).toHaveBeenCalledWith(TEST_S3_KEY);
|
||||
});
|
||||
});
|
||||
+197
@@ -0,0 +1,197 @@
|
||||
import { Injectable, Logger } from '@nestjs/common';
|
||||
import { InjectRepository } from '@nestjs/typeorm';
|
||||
|
||||
import { MessageChannelType } from 'twenty-shared/types';
|
||||
import { Repository } from 'typeorm';
|
||||
|
||||
import { ConnectedAccountEntity } from 'src/engine/metadata-modules/connected-account/entities/connected-account.entity';
|
||||
import { MessageChannelEntity } from 'src/engine/metadata-modules/message-channel/entities/message-channel.entity';
|
||||
import { GlobalWorkspaceOrmManager } from 'src/engine/twenty-orm/global-workspace-datasource/global-workspace-orm.manager';
|
||||
import { buildSystemAuthContext } from 'src/engine/twenty-orm/utils/build-system-auth-context.util';
|
||||
import { InboundEmailParserService } from 'src/modules/messaging/message-import-manager/drivers/inbound-email/services/inbound-email-parser.service';
|
||||
import { InboundEmailS3ClientProvider } from 'src/modules/messaging/message-import-manager/drivers/inbound-email/providers/inbound-email-s3-client.provider';
|
||||
import { InboundEmailStorageService } from 'src/modules/messaging/message-import-manager/drivers/inbound-email/services/inbound-email-storage.service';
|
||||
import { extractEnvelopeRecipient } from 'src/modules/messaging/message-import-manager/drivers/inbound-email/utils/extract-envelope-recipient.util';
|
||||
import { MessagingSaveMessagesAndEnqueueContactCreationService } from 'src/modules/messaging/message-import-manager/services/messaging-save-messages-and-enqueue-contact-creation.service';
|
||||
|
||||
export type InboundEmailImportOutcome =
|
||||
| { kind: 'imported'; workspaceId: string; messageChannelId: string }
|
||||
| { kind: 'unmatched'; recipient: string | null }
|
||||
| { kind: 'loop_dropped'; workspaceId: string }
|
||||
| { kind: 'unconfigured' }
|
||||
| { kind: 'parse_failed'; error: string }
|
||||
| { kind: 'persist_failed'; error: string };
|
||||
|
||||
// Orchestrates the "pull one S3 object, import it" cycle. Kept independent
|
||||
// of BullMQ so it can be unit-tested directly; the job is a thin shell.
|
||||
@Injectable()
|
||||
export class InboundEmailImportService {
|
||||
private readonly logger = new Logger(InboundEmailImportService.name);
|
||||
|
||||
constructor(
|
||||
private readonly inboundEmailS3ClientProvider: InboundEmailS3ClientProvider,
|
||||
private readonly inboundEmailStorageService: InboundEmailStorageService,
|
||||
private readonly inboundEmailParserService: InboundEmailParserService,
|
||||
private readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager,
|
||||
private readonly messagingSaveMessagesAndEnqueueContactCreationService: MessagingSaveMessagesAndEnqueueContactCreationService,
|
||||
@InjectRepository(MessageChannelEntity)
|
||||
private readonly messageChannelRepository: Repository<MessageChannelEntity>,
|
||||
@InjectRepository(ConnectedAccountEntity)
|
||||
private readonly connectedAccountRepository: Repository<ConnectedAccountEntity>,
|
||||
) {}
|
||||
|
||||
async importFromS3Key(s3Key: string): Promise<InboundEmailImportOutcome> {
|
||||
if (!this.inboundEmailS3ClientProvider.isConfigured()) {
|
||||
this.logger.warn(
|
||||
`Skipping inbound email import for ${s3Key}: forwarding is not configured.`,
|
||||
);
|
||||
|
||||
return { kind: 'unconfigured' };
|
||||
}
|
||||
|
||||
const inboundDomain = this.inboundEmailS3ClientProvider.getDomain();
|
||||
|
||||
let rawMessage: Buffer;
|
||||
|
||||
try {
|
||||
rawMessage = await this.inboundEmailStorageService.getRawMessage(s3Key);
|
||||
} catch (error) {
|
||||
const message = error instanceof Error ? error.message : String(error);
|
||||
|
||||
this.logger.error(
|
||||
`Failed to download inbound email from S3 key ${s3Key}: ${message}`,
|
||||
);
|
||||
|
||||
return { kind: 'parse_failed', error: message };
|
||||
}
|
||||
|
||||
let parsedInbound;
|
||||
|
||||
try {
|
||||
parsedInbound = await this.inboundEmailParserService.parse(
|
||||
rawMessage,
|
||||
s3Key,
|
||||
);
|
||||
} catch (error) {
|
||||
const message = error instanceof Error ? error.message : String(error);
|
||||
|
||||
this.logger.error(
|
||||
`Failed to parse inbound email ${s3Key}: ${message}`,
|
||||
);
|
||||
await this.safeMove(s3Key, 'failed');
|
||||
|
||||
return { kind: 'parse_failed', error: message };
|
||||
}
|
||||
|
||||
const recipient = extractEnvelopeRecipient(parsedInbound.parsed, inboundDomain);
|
||||
|
||||
if (!recipient) {
|
||||
this.logger.warn(
|
||||
`No recipient at ${inboundDomain} found for inbound email ${s3Key}`,
|
||||
);
|
||||
await this.safeMove(s3Key, 'unmatched');
|
||||
|
||||
return { kind: 'unmatched', recipient: null };
|
||||
}
|
||||
|
||||
const messageChannel = await this.messageChannelRepository.findOne({
|
||||
where: {
|
||||
handle: recipient,
|
||||
type: MessageChannelType.EMAIL_FORWARDING,
|
||||
},
|
||||
});
|
||||
|
||||
if (!messageChannel) {
|
||||
this.logger.warn(
|
||||
`No forwarding channel matches recipient ${recipient} (key ${s3Key})`,
|
||||
);
|
||||
await this.safeMove(s3Key, 'unmatched');
|
||||
|
||||
return { kind: 'unmatched', recipient };
|
||||
}
|
||||
|
||||
const workspaceId = messageChannel.workspaceId;
|
||||
|
||||
if (
|
||||
parsedInbound.originWorkspaceId &&
|
||||
parsedInbound.originWorkspaceId === workspaceId
|
||||
) {
|
||||
this.logger.log(
|
||||
`Dropping loopback email ${s3Key} for workspace ${workspaceId}`,
|
||||
);
|
||||
await this.safeMove(s3Key, 'processed');
|
||||
|
||||
return { kind: 'loop_dropped', workspaceId };
|
||||
}
|
||||
|
||||
const connectedAccount = await this.connectedAccountRepository.findOne({
|
||||
where: {
|
||||
id: messageChannel.connectedAccountId,
|
||||
workspaceId,
|
||||
},
|
||||
});
|
||||
|
||||
if (!connectedAccount) {
|
||||
this.logger.error(
|
||||
`Forwarding channel ${messageChannel.id} has no connected account`,
|
||||
);
|
||||
await this.safeMove(s3Key, 'failed');
|
||||
|
||||
return { kind: 'persist_failed', error: 'connected_account_missing' };
|
||||
}
|
||||
|
||||
const authContext = buildSystemAuthContext(workspaceId);
|
||||
|
||||
try {
|
||||
await this.globalWorkspaceOrmManager.executeInWorkspaceContext(
|
||||
async () => {
|
||||
await this.messagingSaveMessagesAndEnqueueContactCreationService.saveMessagesAndEnqueueContactCreation(
|
||||
[parsedInbound.message],
|
||||
messageChannel,
|
||||
connectedAccount,
|
||||
workspaceId,
|
||||
);
|
||||
},
|
||||
authContext,
|
||||
);
|
||||
} catch (error) {
|
||||
const message = error instanceof Error ? error.message : String(error);
|
||||
|
||||
this.logger.error(
|
||||
`Failed to persist inbound email ${s3Key} for workspace ${workspaceId}: ${message}`,
|
||||
);
|
||||
await this.safeMove(s3Key, 'failed');
|
||||
|
||||
return { kind: 'persist_failed', error: message };
|
||||
}
|
||||
|
||||
await this.safeMove(s3Key, 'processed');
|
||||
|
||||
return {
|
||||
kind: 'imported',
|
||||
workspaceId,
|
||||
messageChannelId: messageChannel.id,
|
||||
};
|
||||
}
|
||||
|
||||
private async safeMove(
|
||||
s3Key: string,
|
||||
destination: 'processed' | 'unmatched' | 'failed',
|
||||
): Promise<void> {
|
||||
try {
|
||||
if (destination === 'processed') {
|
||||
await this.inboundEmailStorageService.moveToProcessed(s3Key);
|
||||
} else if (destination === 'unmatched') {
|
||||
await this.inboundEmailStorageService.moveToUnmatched(s3Key);
|
||||
} else {
|
||||
await this.inboundEmailStorageService.moveToFailed(s3Key);
|
||||
}
|
||||
} catch (error) {
|
||||
this.logger.error(
|
||||
`Failed to archive inbound email ${s3Key} to ${destination}: ${
|
||||
error instanceof Error ? error.message : String(error)
|
||||
}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
+137
@@ -0,0 +1,137 @@
|
||||
import { Injectable, Logger } from '@nestjs/common';
|
||||
|
||||
import PostalMime, { type Email as ParsedEmail, type Address } from 'postal-mime';
|
||||
import { MessageDirection } from 'src/modules/messaging/common/enums/message-direction.enum';
|
||||
import { MessageParticipantRole } from 'twenty-shared/types';
|
||||
|
||||
import { type EmailAddress } from 'src/modules/messaging/message-import-manager/types/email-address';
|
||||
import { type MessageWithParticipants } from 'src/modules/messaging/message-import-manager/types/message';
|
||||
import { formatAddressObjectAsParticipants } from 'src/modules/messaging/message-import-manager/utils/format-address-object-as-participants.util';
|
||||
import { sanitizeString } from 'src/modules/messaging/message-import-manager/utils/sanitize-string.util';
|
||||
import { X_TWENTY_ORIGIN_HEADER } from 'src/modules/messaging/message-import-manager/drivers/inbound-email/constants/inbound-email.constants';
|
||||
|
||||
export type ParsedInboundMessage = {
|
||||
parsed: ParsedEmail;
|
||||
originWorkspaceId: string | null;
|
||||
message: MessageWithParticipants;
|
||||
};
|
||||
|
||||
@Injectable()
|
||||
export class InboundEmailParserService {
|
||||
private readonly logger = new Logger(InboundEmailParserService.name);
|
||||
|
||||
async parse(
|
||||
rawMessage: Buffer,
|
||||
s3Key: string,
|
||||
): Promise<ParsedInboundMessage> {
|
||||
const parsed = await PostalMime.parse(rawMessage);
|
||||
|
||||
const originWorkspaceId = this.extractOriginWorkspaceId(parsed);
|
||||
const message = this.buildMessage(parsed, s3Key);
|
||||
|
||||
return { parsed, originWorkspaceId, message };
|
||||
}
|
||||
|
||||
private extractOriginWorkspaceId(parsed: ParsedEmail): string | null {
|
||||
const header = parsed.headers?.find(
|
||||
(h) => h.key?.toLowerCase() === X_TWENTY_ORIGIN_HEADER,
|
||||
);
|
||||
|
||||
if (!header?.value) {
|
||||
return null;
|
||||
}
|
||||
|
||||
return header.value.trim();
|
||||
}
|
||||
|
||||
private buildMessage(
|
||||
parsed: ParsedEmail,
|
||||
s3Key: string,
|
||||
): MessageWithParticipants {
|
||||
return {
|
||||
// The S3 key is the natural external id: it uniquely identifies the
|
||||
// object that produced this message and survives retries.
|
||||
externalId: `inbound-email:${s3Key}`,
|
||||
messageThreadExternalId: this.extractThreadId(parsed),
|
||||
headerMessageId:
|
||||
parsed.messageId?.trim() || `inbound-${s3Key}`,
|
||||
subject: sanitizeString(parsed.subject || ''),
|
||||
text: sanitizeString(parsed.text || ''),
|
||||
receivedAt: parsed.date ? new Date(parsed.date) : new Date(),
|
||||
// Forwarding channels are inbound-only — every message is incoming.
|
||||
direction: MessageDirection.INCOMING,
|
||||
attachments: (parsed.attachments || []).map((attachment) => ({
|
||||
filename: attachment.filename || 'unnamed-attachment',
|
||||
})),
|
||||
participants: this.extractParticipants(parsed),
|
||||
};
|
||||
}
|
||||
|
||||
private extractThreadId(parsed: ParsedEmail): string {
|
||||
const references = parsed.references;
|
||||
|
||||
if (typeof references === 'string' && references.trim()) {
|
||||
const first = references.trim().split(/\s+/)[0];
|
||||
|
||||
if (first) {
|
||||
return first;
|
||||
}
|
||||
}
|
||||
|
||||
if (Array.isArray(references) && references.length > 0) {
|
||||
const first = String(references[0]).trim();
|
||||
|
||||
if (first) {
|
||||
return first;
|
||||
}
|
||||
}
|
||||
|
||||
if (parsed.inReplyTo) {
|
||||
const inReplyTo = String(parsed.inReplyTo).trim();
|
||||
|
||||
if (inReplyTo) {
|
||||
return inReplyTo;
|
||||
}
|
||||
}
|
||||
|
||||
if (parsed.messageId?.trim()) {
|
||||
return parsed.messageId.trim();
|
||||
}
|
||||
|
||||
return `thread-${Date.now()}-${Math.random().toString(36).slice(2, 11)}`;
|
||||
}
|
||||
|
||||
private extractParticipants(parsed: ParsedEmail) {
|
||||
const addressFields = [
|
||||
{ field: parsed.from, role: MessageParticipantRole.FROM },
|
||||
{ field: parsed.to, role: MessageParticipantRole.TO },
|
||||
{ field: parsed.cc, role: MessageParticipantRole.CC },
|
||||
{ field: parsed.bcc, role: MessageParticipantRole.BCC },
|
||||
] as const;
|
||||
|
||||
return addressFields.flatMap(({ field, role }) =>
|
||||
formatAddressObjectAsParticipants(this.extractAddresses(field), role),
|
||||
);
|
||||
}
|
||||
|
||||
private extractAddresses(
|
||||
address: Address | Address[] | undefined,
|
||||
): EmailAddress[] {
|
||||
if (!address) {
|
||||
return [];
|
||||
}
|
||||
|
||||
const addresses = Array.isArray(address) ? address : [address];
|
||||
|
||||
const mailboxes = addresses.flatMap((addr) =>
|
||||
addr.address ? [addr] : (addr.group ?? []),
|
||||
);
|
||||
|
||||
return mailboxes
|
||||
.filter((mailbox) => mailbox.address)
|
||||
.map((mailbox) => ({
|
||||
address: mailbox.address,
|
||||
name: sanitizeString(mailbox.name || ''),
|
||||
}));
|
||||
}
|
||||
}
|
||||
+125
@@ -0,0 +1,125 @@
|
||||
import { Injectable, Logger } from '@nestjs/common';
|
||||
|
||||
import {
|
||||
CopyObjectCommand,
|
||||
DeleteObjectCommand,
|
||||
GetObjectCommand,
|
||||
ListObjectsV2Command,
|
||||
} from '@aws-sdk/client-s3';
|
||||
import { Readable } from 'stream';
|
||||
|
||||
import { INBOUND_EMAIL_S3_PREFIXES } from 'src/modules/messaging/message-import-manager/drivers/inbound-email/constants/inbound-email.constants';
|
||||
import { InboundEmailS3ClientProvider } from 'src/modules/messaging/message-import-manager/drivers/inbound-email/providers/inbound-email-s3-client.provider';
|
||||
|
||||
// Thin wrapper around S3 that implements the "bucket-as-queue" pattern:
|
||||
// listIncoming => pull work, getRawMessage => download, moveTo* => ack.
|
||||
// Archive destinations are date-partitioned so operators can browse the
|
||||
// bucket by day when triaging failures.
|
||||
@Injectable()
|
||||
export class InboundEmailStorageService {
|
||||
private readonly logger = new Logger(InboundEmailStorageService.name);
|
||||
|
||||
constructor(
|
||||
private readonly inboundEmailS3ClientProvider: InboundEmailS3ClientProvider,
|
||||
) {}
|
||||
|
||||
async listIncoming(maxKeys: number): Promise<string[]> {
|
||||
const client = this.inboundEmailS3ClientProvider.getClient();
|
||||
const Bucket = this.inboundEmailS3ClientProvider.getBucket();
|
||||
|
||||
const response = await client.send(
|
||||
new ListObjectsV2Command({
|
||||
Bucket,
|
||||
Prefix: INBOUND_EMAIL_S3_PREFIXES.incoming,
|
||||
MaxKeys: maxKeys,
|
||||
}),
|
||||
);
|
||||
|
||||
const keys = (response.Contents ?? [])
|
||||
.map((object) => object.Key)
|
||||
.filter((key): key is string => typeof key === 'string')
|
||||
.filter((key) => key !== INBOUND_EMAIL_S3_PREFIXES.incoming);
|
||||
|
||||
return keys;
|
||||
}
|
||||
|
||||
async getRawMessage(key: string): Promise<Buffer> {
|
||||
const client = this.inboundEmailS3ClientProvider.getClient();
|
||||
const Bucket = this.inboundEmailS3ClientProvider.getBucket();
|
||||
|
||||
const response = await client.send(
|
||||
new GetObjectCommand({ Bucket, Key: key }),
|
||||
);
|
||||
|
||||
if (!response.Body) {
|
||||
throw new Error(`S3 object ${key} has no body`);
|
||||
}
|
||||
|
||||
// AWS SDK v3 returns Body as a stream in Node; collect it to a Buffer so
|
||||
// postal-mime can parse it in one shot.
|
||||
const stream = response.Body as Readable;
|
||||
const chunks: Buffer[] = [];
|
||||
|
||||
for await (const chunk of stream) {
|
||||
chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk));
|
||||
}
|
||||
|
||||
return Buffer.concat(chunks);
|
||||
}
|
||||
|
||||
async moveToProcessed(key: string): Promise<void> {
|
||||
await this.moveToArchive(key, INBOUND_EMAIL_S3_PREFIXES.processed);
|
||||
}
|
||||
|
||||
async moveToUnmatched(key: string): Promise<void> {
|
||||
await this.moveToArchive(key, INBOUND_EMAIL_S3_PREFIXES.unmatched);
|
||||
}
|
||||
|
||||
async moveToFailed(key: string): Promise<void> {
|
||||
await this.moveToArchive(key, INBOUND_EMAIL_S3_PREFIXES.failed);
|
||||
}
|
||||
|
||||
private async moveToArchive(
|
||||
key: string,
|
||||
destinationPrefix: string,
|
||||
): Promise<void> {
|
||||
const client = this.inboundEmailS3ClientProvider.getClient();
|
||||
const Bucket = this.inboundEmailS3ClientProvider.getBucket();
|
||||
|
||||
const destinationKey = this.buildArchiveKey(key, destinationPrefix);
|
||||
|
||||
try {
|
||||
await client.send(
|
||||
new CopyObjectCommand({
|
||||
Bucket,
|
||||
CopySource: `${Bucket}/${encodeURIComponent(key)}`,
|
||||
Key: destinationKey,
|
||||
}),
|
||||
);
|
||||
|
||||
await client.send(
|
||||
new DeleteObjectCommand({ Bucket, Key: key }),
|
||||
);
|
||||
} catch (error) {
|
||||
this.logger.error(
|
||||
`Failed to move S3 key ${key} to ${destinationKey}: ${
|
||||
error instanceof Error ? error.message : String(error)
|
||||
}`,
|
||||
);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
buildArchiveKey(sourceKey: string, destinationPrefix: string): string {
|
||||
const bareKey = sourceKey.startsWith(INBOUND_EMAIL_S3_PREFIXES.incoming)
|
||||
? sourceKey.slice(INBOUND_EMAIL_S3_PREFIXES.incoming.length)
|
||||
: sourceKey;
|
||||
|
||||
const now = new Date();
|
||||
const year = now.getUTCFullYear();
|
||||
const month = String(now.getUTCMonth() + 1).padStart(2, '0');
|
||||
const day = String(now.getUTCDate()).padStart(2, '0');
|
||||
|
||||
return `${destinationPrefix}${year}-${month}-${day}/${bareKey}`;
|
||||
}
|
||||
}
|
||||
+136
@@ -0,0 +1,136 @@
|
||||
import { type Email as ParsedEmail, type Header } from 'postal-mime';
|
||||
|
||||
import { extractEnvelopeRecipient } from 'src/modules/messaging/message-import-manager/drivers/inbound-email/utils/extract-envelope-recipient.util';
|
||||
|
||||
const INBOUND_DOMAIN = 'in.twenty.com';
|
||||
|
||||
const header = (key: string, value: string): Header => ({
|
||||
key: key.toLowerCase(),
|
||||
originalKey: key,
|
||||
value,
|
||||
});
|
||||
|
||||
const buildParsedEmail = (
|
||||
overrides: Partial<ParsedEmail> = {},
|
||||
): ParsedEmail => ({
|
||||
headers: [],
|
||||
subject: 'test',
|
||||
from: { address: 'sender@example.com', name: 'Sender' },
|
||||
to: [],
|
||||
cc: [],
|
||||
bcc: [],
|
||||
date: new Date().toISOString(),
|
||||
messageId: '<test@example.com>',
|
||||
html: '',
|
||||
text: 'test body',
|
||||
attachments: [],
|
||||
...overrides,
|
||||
});
|
||||
|
||||
describe('extractEnvelopeRecipient', () => {
|
||||
it('should extract recipient from Received header "for <addr>" clause', () => {
|
||||
const parsed = buildParsedEmail({
|
||||
headers: [
|
||||
header(
|
||||
'Received',
|
||||
'from mx.google.com by 10.0.0.1 for <ch_abc123@in.twenty.com>; Mon, 01 Jan 2024 00:00:00 +0000',
|
||||
),
|
||||
],
|
||||
});
|
||||
|
||||
expect(extractEnvelopeRecipient(parsed, INBOUND_DOMAIN)).toBe(
|
||||
'ch_abc123@in.twenty.com',
|
||||
);
|
||||
});
|
||||
|
||||
it('should extract from Delivered-To header', () => {
|
||||
const parsed = buildParsedEmail({
|
||||
headers: [
|
||||
header('Delivered-To', 'ch_def456@in.twenty.com'),
|
||||
],
|
||||
});
|
||||
|
||||
expect(extractEnvelopeRecipient(parsed, INBOUND_DOMAIN)).toBe(
|
||||
'ch_def456@in.twenty.com',
|
||||
);
|
||||
});
|
||||
|
||||
it('should extract from X-Original-To header', () => {
|
||||
const parsed = buildParsedEmail({
|
||||
headers: [
|
||||
header('X-Original-To', 'ch_ghi789@in.twenty.com'),
|
||||
],
|
||||
});
|
||||
|
||||
expect(extractEnvelopeRecipient(parsed, INBOUND_DOMAIN)).toBe(
|
||||
'ch_ghi789@in.twenty.com',
|
||||
);
|
||||
});
|
||||
|
||||
it('should fall back to To field when no delivery headers match', () => {
|
||||
const parsed = buildParsedEmail({
|
||||
to: [{ address: 'ch_tofield@in.twenty.com', name: 'Channel' }],
|
||||
});
|
||||
|
||||
expect(extractEnvelopeRecipient(parsed, INBOUND_DOMAIN)).toBe(
|
||||
'ch_tofield@in.twenty.com',
|
||||
);
|
||||
});
|
||||
|
||||
it('should fall back to Cc field', () => {
|
||||
const parsed = buildParsedEmail({
|
||||
to: [{ address: 'someone@other.com', name: '' }],
|
||||
cc: [{ address: 'ch_ccfield@in.twenty.com', name: '' }],
|
||||
});
|
||||
|
||||
expect(extractEnvelopeRecipient(parsed, INBOUND_DOMAIN)).toBe(
|
||||
'ch_ccfield@in.twenty.com',
|
||||
);
|
||||
});
|
||||
|
||||
it('should return null when no address matches the inbound domain', () => {
|
||||
const parsed = buildParsedEmail({
|
||||
to: [{ address: 'user@example.com', name: '' }],
|
||||
headers: [
|
||||
header('Delivered-To', 'user@other-domain.com'),
|
||||
],
|
||||
});
|
||||
|
||||
expect(extractEnvelopeRecipient(parsed, INBOUND_DOMAIN)).toBeNull();
|
||||
});
|
||||
|
||||
it('should be case-insensitive on domain match', () => {
|
||||
const parsed = buildParsedEmail({
|
||||
headers: [
|
||||
header('Delivered-To', 'ch_upper@IN.TWENTY.COM'),
|
||||
],
|
||||
});
|
||||
|
||||
expect(extractEnvelopeRecipient(parsed, INBOUND_DOMAIN)).toBe(
|
||||
'ch_upper@in.twenty.com',
|
||||
);
|
||||
});
|
||||
|
||||
it('should prefer Received header over Delivered-To', () => {
|
||||
const parsed = buildParsedEmail({
|
||||
headers: [
|
||||
header('Received', 'from mx for <ch_received@in.twenty.com>; date'),
|
||||
header('Delivered-To', 'ch_deliveredto@in.twenty.com'),
|
||||
],
|
||||
});
|
||||
|
||||
expect(extractEnvelopeRecipient(parsed, INBOUND_DOMAIN)).toBe(
|
||||
'ch_received@in.twenty.com',
|
||||
);
|
||||
});
|
||||
|
||||
it('should handle empty headers gracefully', () => {
|
||||
const parsed = buildParsedEmail({
|
||||
headers: [],
|
||||
to: [],
|
||||
cc: [],
|
||||
});
|
||||
|
||||
expect(extractEnvelopeRecipient(parsed, INBOUND_DOMAIN)).toBeNull();
|
||||
});
|
||||
});
|
||||
+125
@@ -0,0 +1,125 @@
|
||||
import { type Email as ParsedEmail } from 'postal-mime';
|
||||
|
||||
// SES inbound action writes the raw MIME to S3 without preserving the SMTP
|
||||
// envelope ("RCPT TO"). We recover the envelope recipient by walking the
|
||||
// headers in priority order: delivery headers first (most reliable), then
|
||||
// To/Cc as a fallback. The first address whose domain matches our inbound
|
||||
// domain wins, which is what we use to look up the forwarding channel.
|
||||
|
||||
type HeaderLike =
|
||||
| string
|
||||
| string[]
|
||||
| { key?: string; value?: string }[]
|
||||
| undefined;
|
||||
|
||||
const EMAIL_REGEX = /[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+/g;
|
||||
|
||||
const extractEmailsFromText = (text: string): string[] => {
|
||||
const matches = text.match(EMAIL_REGEX);
|
||||
|
||||
return matches ? matches.map((email) => email.toLowerCase()) : [];
|
||||
};
|
||||
|
||||
const headerValues = (
|
||||
headers: ParsedEmail['headers'] | undefined,
|
||||
name: string,
|
||||
): string[] => {
|
||||
if (!headers) {
|
||||
return [];
|
||||
}
|
||||
|
||||
const target = name.toLowerCase();
|
||||
|
||||
return headers
|
||||
.filter((header) => header.key?.toLowerCase() === target)
|
||||
.map((header) => header.value ?? '');
|
||||
};
|
||||
|
||||
const addressesFromParsedField = (
|
||||
field: ParsedEmail['to'] | ParsedEmail['cc'],
|
||||
): string[] => {
|
||||
if (!field) {
|
||||
return [];
|
||||
}
|
||||
|
||||
const list = Array.isArray(field) ? field : [field];
|
||||
|
||||
return list
|
||||
.flatMap((entry) =>
|
||||
entry.address ? [entry.address] : (entry.group ?? []).map((g) => g.address ?? ''),
|
||||
)
|
||||
.filter((address) => address.length > 0)
|
||||
.map((address) => address.toLowerCase());
|
||||
};
|
||||
|
||||
export const extractEnvelopeRecipient = (
|
||||
parsed: ParsedEmail,
|
||||
inboundDomain: string,
|
||||
rawHeaders?: HeaderLike,
|
||||
): string | null => {
|
||||
const domain = inboundDomain.toLowerCase();
|
||||
const isMatch = (address: string): boolean => address.endsWith(`@${domain}`);
|
||||
|
||||
// SES adds "X-Original-To" and "Delivered-To" on many flows; Received
|
||||
// headers include "for <addr>" which is the closest thing to the envelope
|
||||
// recipient that survives MIME parsing.
|
||||
const receivedValues = headerValues(parsed.headers, 'received');
|
||||
|
||||
for (const received of receivedValues) {
|
||||
const forMatch = received.match(/\bfor\s+<?([^\s>;]+)>?/i);
|
||||
|
||||
if (forMatch) {
|
||||
const candidate = forMatch[1].toLowerCase();
|
||||
|
||||
if (isMatch(candidate)) {
|
||||
return candidate;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const deliveryHeaders = [
|
||||
...headerValues(parsed.headers, 'delivered-to'),
|
||||
...headerValues(parsed.headers, 'x-original-to'),
|
||||
...headerValues(parsed.headers, 'x-forwarded-to'),
|
||||
...headerValues(parsed.headers, 'x-envelope-to'),
|
||||
];
|
||||
|
||||
for (const raw of deliveryHeaders) {
|
||||
for (const candidate of extractEmailsFromText(raw)) {
|
||||
if (isMatch(candidate)) {
|
||||
return candidate;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const toCandidates = [
|
||||
...addressesFromParsedField(parsed.to),
|
||||
...addressesFromParsedField(parsed.cc),
|
||||
];
|
||||
|
||||
for (const candidate of toCandidates) {
|
||||
if (isMatch(candidate)) {
|
||||
return candidate;
|
||||
}
|
||||
}
|
||||
|
||||
// rawHeaders is a last-ditch escape hatch for callers that have the raw
|
||||
// message available and want to scan any arbitrary header line.
|
||||
if (rawHeaders) {
|
||||
const flat = Array.isArray(rawHeaders)
|
||||
? rawHeaders
|
||||
.map((h) => (typeof h === 'string' ? h : (h?.value ?? '')))
|
||||
.join('\n')
|
||||
: rawHeaders;
|
||||
|
||||
if (typeof flat === 'string') {
|
||||
for (const candidate of extractEmailsFromText(flat)) {
|
||||
if (isMatch(candidate)) {
|
||||
return candidate;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return null;
|
||||
};
|
||||
+28
@@ -0,0 +1,28 @@
|
||||
import { Scope } from '@nestjs/common';
|
||||
|
||||
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';
|
||||
import { InboundEmailImportService } from 'src/modules/messaging/message-import-manager/drivers/inbound-email/services/inbound-email-import.service';
|
||||
|
||||
export type MessagingInboundEmailImportJobData = {
|
||||
s3Key: string;
|
||||
};
|
||||
|
||||
// One job per S3 object. Keeping the unit of work small means a single bad
|
||||
// message can be retried or moved to `failed/` without blocking the rest
|
||||
// of the queue, and lets BullMQ do the per-message dedupe/backoff we want.
|
||||
@Processor({
|
||||
queueName: MessageQueue.messagingQueue,
|
||||
scope: Scope.REQUEST,
|
||||
})
|
||||
export class MessagingInboundEmailImportJob {
|
||||
constructor(
|
||||
private readonly inboundEmailImportService: InboundEmailImportService,
|
||||
) {}
|
||||
|
||||
@Process(MessagingInboundEmailImportJob.name)
|
||||
async handle(data: MessagingInboundEmailImportJobData): Promise<void> {
|
||||
await this.inboundEmailImportService.importFromS3Key(data.s3Key);
|
||||
}
|
||||
}
|
||||
+16
@@ -2,8 +2,10 @@ import { Module } from '@nestjs/common';
|
||||
import { TypeOrmModule } from '@nestjs/typeorm';
|
||||
|
||||
import { FeatureFlagModule } from 'src/engine/core-modules/feature-flag/feature-flag.module';
|
||||
import { TwentyConfigModule } from 'src/engine/core-modules/twenty-config/twenty-config.module';
|
||||
import { UserWorkspaceEntity } from 'src/engine/core-modules/user-workspace/user-workspace.entity';
|
||||
import { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity';
|
||||
import { ConnectedAccountEntity } from 'src/engine/metadata-modules/connected-account/entities/connected-account.entity';
|
||||
import { DataSourceEntity } from 'src/engine/metadata-modules/data-source/data-source.entity';
|
||||
import { MessageChannelEntity } from 'src/engine/metadata-modules/message-channel/entities/message-channel.entity';
|
||||
import { MessageFolderEntity } from 'src/engine/metadata-modules/message-folder/entities/message-folder.entity';
|
||||
@@ -19,20 +21,25 @@ import { MessagingMessageCleanerModule } from 'src/modules/messaging/message-cle
|
||||
import { MessagingFolderSyncManagerModule } from 'src/modules/messaging/message-folder-manager/messaging-folder-sync-manager.module';
|
||||
import { MessagingSingleMessageImportCommand } from 'src/modules/messaging/message-import-manager/commands/messaging-single-message-import.command';
|
||||
import { MessagingTriggerMessageListFetchCommand } from 'src/modules/messaging/message-import-manager/commands/messaging-trigger-message-list-fetch.command';
|
||||
import { MessagingInboundEmailPollCronCommand } from 'src/modules/messaging/message-import-manager/crons/commands/messaging-inbound-email-poll.cron.command';
|
||||
import { MessagingMessageListFetchCronCommand } from 'src/modules/messaging/message-import-manager/crons/commands/messaging-message-list-fetch.cron.command';
|
||||
import { MessagingMessagesImportCronCommand } from 'src/modules/messaging/message-import-manager/crons/commands/messaging-messages-import.cron.command';
|
||||
import { MessagingOngoingStaleCronCommand } from 'src/modules/messaging/message-import-manager/crons/commands/messaging-ongoing-stale.cron.command';
|
||||
import { MessagingRelaunchFailedMessageChannelsCronCommand } from 'src/modules/messaging/message-import-manager/crons/commands/messaging-relaunch-failed-message-channels.cron.command';
|
||||
import { MessagingInboundEmailPollCronJob } from 'src/modules/messaging/message-import-manager/crons/jobs/messaging-inbound-email-poll.cron.job';
|
||||
import { MessagingMessageListFetchCronJob } from 'src/modules/messaging/message-import-manager/crons/jobs/messaging-message-list-fetch.cron.job';
|
||||
import { MessagingMessagesImportCronJob } from 'src/modules/messaging/message-import-manager/crons/jobs/messaging-messages-import.cron.job';
|
||||
import { MessagingOngoingStaleCronJob } from 'src/modules/messaging/message-import-manager/crons/jobs/messaging-ongoing-stale.cron.job';
|
||||
import { MessagingRelaunchFailedMessageChannelsCronJob } from 'src/modules/messaging/message-import-manager/crons/jobs/messaging-relaunch-failed-message-channels.cron.job';
|
||||
import { MessagingGmailDriverModule } from 'src/modules/messaging/message-import-manager/drivers/gmail/messaging-gmail-driver.module';
|
||||
import { MessagingIMAPDriverModule } from 'src/modules/messaging/message-import-manager/drivers/imap/messaging-imap-driver.module';
|
||||
import { MessagingInboundEmailDriverModule } from 'src/modules/messaging/message-import-manager/drivers/inbound-email/messaging-inbound-email-driver.module';
|
||||
import { InboundEmailImportService } from 'src/modules/messaging/message-import-manager/drivers/inbound-email/services/inbound-email-import.service';
|
||||
import { MessagingMicrosoftDriverModule } from 'src/modules/messaging/message-import-manager/drivers/microsoft/messaging-microsoft-driver.module';
|
||||
import { MessagingSmtpDriverModule } from 'src/modules/messaging/message-import-manager/drivers/smtp/messaging-smtp-driver.module';
|
||||
import { MessagingAddSingleMessageToCacheForImportJob } from 'src/modules/messaging/message-import-manager/jobs/messaging-add-single-message-to-cache-for-import.job';
|
||||
import { MessagingCleanCacheJob } from 'src/modules/messaging/message-import-manager/jobs/messaging-clean-cache';
|
||||
import { MessagingInboundEmailImportJob } from 'src/modules/messaging/message-import-manager/jobs/messaging-inbound-email-import.job';
|
||||
import { MessagingMessageListFetchJob } from 'src/modules/messaging/message-import-manager/jobs/messaging-message-list-fetch.job';
|
||||
import { MessagingMessagesImportJob } from 'src/modules/messaging/message-import-manager/jobs/messaging-messages-import.job';
|
||||
import { MessagingOngoingStaleJob } from 'src/modules/messaging/message-import-manager/jobs/messaging-ongoing-stale.job';
|
||||
@@ -62,7 +69,9 @@ import { MessagingMonitoringModule } from 'src/modules/messaging/monitoring/mess
|
||||
MessagingMicrosoftDriverModule,
|
||||
MessagingIMAPDriverModule,
|
||||
MessagingSmtpDriverModule,
|
||||
MessagingInboundEmailDriverModule,
|
||||
MessagingCommonModule,
|
||||
TwentyConfigModule,
|
||||
TypeOrmModule.forFeature([
|
||||
WorkspaceEntity,
|
||||
DataSourceEntity,
|
||||
@@ -70,6 +79,7 @@ import { MessagingMonitoringModule } from 'src/modules/messaging/monitoring/mess
|
||||
MessageChannelEntity,
|
||||
MessageFolderEntity,
|
||||
UserWorkspaceEntity,
|
||||
ConnectedAccountEntity,
|
||||
]),
|
||||
EmailAliasManagerModule,
|
||||
FeatureFlagModule,
|
||||
@@ -85,16 +95,19 @@ import { MessagingMonitoringModule } from 'src/modules/messaging/monitoring/mess
|
||||
MessagingMessagesImportCronCommand,
|
||||
MessagingOngoingStaleCronCommand,
|
||||
MessagingRelaunchFailedMessageChannelsCronCommand,
|
||||
MessagingInboundEmailPollCronCommand,
|
||||
MessagingSingleMessageImportCommand,
|
||||
MessagingTriggerMessageListFetchCommand,
|
||||
MessagingMessageListFetchJob,
|
||||
MessagingMessagesImportJob,
|
||||
MessagingOngoingStaleJob,
|
||||
MessagingRelaunchFailedMessageChannelJob,
|
||||
MessagingInboundEmailImportJob,
|
||||
MessagingMessageListFetchCronJob,
|
||||
MessagingMessagesImportCronJob,
|
||||
MessagingOngoingStaleCronJob,
|
||||
MessagingRelaunchFailedMessageChannelsCronJob,
|
||||
MessagingInboundEmailPollCronJob,
|
||||
MessagingAddSingleMessageToCacheForImportJob,
|
||||
MessagingCleanCacheJob,
|
||||
MessagingMessageService,
|
||||
@@ -111,6 +124,7 @@ import { MessagingMonitoringModule } from 'src/modules/messaging/monitoring/mess
|
||||
MessagingProcessGroupEmailActionsService,
|
||||
MessagingDeleteFolderMessagesService,
|
||||
MessagingDeleteGroupEmailMessagesService,
|
||||
InboundEmailImportService,
|
||||
],
|
||||
exports: [
|
||||
MessagingAccountAuthenticationService,
|
||||
@@ -118,7 +132,9 @@ import { MessagingMonitoringModule } from 'src/modules/messaging/monitoring/mess
|
||||
MessagingMessagesImportCronCommand,
|
||||
MessagingOngoingStaleCronCommand,
|
||||
MessagingRelaunchFailedMessageChannelsCronCommand,
|
||||
MessagingInboundEmailPollCronCommand,
|
||||
MessagingProcessGroupEmailActionsService,
|
||||
InboundEmailImportService,
|
||||
],
|
||||
})
|
||||
export class MessagingImportManagerModule {}
|
||||
|
||||
+7
@@ -38,6 +38,12 @@ export class MessagingMessageOutboundService {
|
||||
sendMessageInput,
|
||||
connectedAccount,
|
||||
);
|
||||
case ConnectedAccountProvider.EMAIL_FORWARDING:
|
||||
// Forwarding channels are inbound-only: replies should go through the
|
||||
// user's own Gmail/Outlook/IMAP account to avoid masking the sender.
|
||||
throw new Error(
|
||||
'Email forwarding channels are inbound-only; reply using your personal account.',
|
||||
);
|
||||
case ConnectedAccountProvider.OIDC:
|
||||
case ConnectedAccountProvider.SAML:
|
||||
throw new Error(
|
||||
@@ -71,6 +77,7 @@ export class MessagingMessageOutboundService {
|
||||
sendMessageInput,
|
||||
connectedAccount,
|
||||
);
|
||||
case ConnectedAccountProvider.EMAIL_FORWARDING:
|
||||
case ConnectedAccountProvider.OIDC:
|
||||
case ConnectedAccountProvider.SAML:
|
||||
throw new Error(
|
||||
|
||||
+1
@@ -26,6 +26,7 @@ export class SendEmailService {
|
||||
attachments: data.attachments,
|
||||
inReplyTo: data.inReplyTo,
|
||||
threadExternalId: data.threadExternalId,
|
||||
originWorkspaceId: data.connectedAccount.workspaceId,
|
||||
},
|
||||
data.connectedAccount,
|
||||
);
|
||||
|
||||
+3
@@ -14,4 +14,7 @@ export type SendMessageInput = {
|
||||
}[];
|
||||
inReplyTo?: string;
|
||||
threadExternalId?: string;
|
||||
// When set, an X-Twenty-Origin header is stamped on the outbound message.
|
||||
// Used by the inbound-email forwarding pipeline to drop group echoes.
|
||||
originWorkspaceId?: string;
|
||||
};
|
||||
|
||||
+11
@@ -1,9 +1,19 @@
|
||||
import { X_TWENTY_ORIGIN_HEADER } from 'src/modules/messaging/message-import-manager/drivers/inbound-email/constants/inbound-email.constants';
|
||||
import { type SendMessageInput } from 'src/modules/messaging/message-outbound-manager/types/send-message-input.type';
|
||||
|
||||
export const toMailComposerOptions = (
|
||||
from: string,
|
||||
sendMessageInput: SendMessageInput,
|
||||
) => {
|
||||
// Stamp outbound mail with the originating workspace id so the inbound
|
||||
// forwarding pipeline can drop loopbacks when a reply hits the group
|
||||
// address that the forwarding channel is a member of.
|
||||
const headers: Record<string, string> = {};
|
||||
|
||||
if (sendMessageInput.originWorkspaceId) {
|
||||
headers[X_TWENTY_ORIGIN_HEADER] = sendMessageInput.originWorkspaceId;
|
||||
}
|
||||
|
||||
return {
|
||||
from,
|
||||
to: sendMessageInput.to,
|
||||
@@ -12,6 +22,7 @@ export const toMailComposerOptions = (
|
||||
subject: sendMessageInput.subject,
|
||||
text: sendMessageInput.body,
|
||||
html: sendMessageInput.html,
|
||||
...(Object.keys(headers).length > 0 ? { headers } : {}),
|
||||
...(sendMessageInput.attachments && sendMessageInput.attachments.length > 0
|
||||
? {
|
||||
attachments: sendMessageInput.attachments.map((attachment) => ({
|
||||
|
||||
@@ -4,4 +4,5 @@ export enum ConnectedAccountProvider {
|
||||
IMAP_SMTP_CALDAV = 'imap_smtp_caldav',
|
||||
OIDC = 'oidc',
|
||||
SAML = 'saml',
|
||||
EMAIL_FORWARDING = 'email_forwarding',
|
||||
}
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
export enum MessageChannelType {
|
||||
EMAIL = 'EMAIL',
|
||||
SMS = 'SMS',
|
||||
EMAIL_FORWARDING = 'EMAIL_FORWARDING',
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user