5748 Create contacts for emails sent and received by email aliases (#5855)
Closes #5748 - Create feature flag - Add scope `https://www.googleapis.com/auth/profile.emails.read` when connecting an account - Get email aliases with google people API, store them in connectedAccount and refresh them before each message-import - Update the contact creation logic accordingly - Refactor --------- Co-authored-by: Charles Bochet <charles@twenty.com>
This commit is contained in:
bosiraphael
@ -1,17 +1,15 @@
|
||||
import { Injectable, Logger } from '@nestjs/common';
|
||||
import { Injectable } from '@nestjs/common';
|
||||
|
||||
import { DataSource, EntityManager } from 'typeorm';
|
||||
import { EntityManager } from 'typeorm';
|
||||
import { v4 } from 'uuid';
|
||||
|
||||
import { WorkspaceDataSourceService } from 'src/engine/workspace-datasource/workspace-datasource.service';
|
||||
import { InjectObjectMetadataRepository } from 'src/engine/object-metadata-repository/object-metadata-repository.decorator';
|
||||
import { ConnectedAccountWorkspaceEntity } from 'src/modules/connected-account/standard-objects/connected-account.workspace-entity';
|
||||
import { MessageChannelMessageAssociationRepository } from 'src/modules/messaging/common/repositories/message-channel-message-association.repository';
|
||||
import { MessageChannelRepository } from 'src/modules/messaging/common/repositories/message-channel.repository';
|
||||
import { MessageThreadRepository } from 'src/modules/messaging/common/repositories/message-thread.repository';
|
||||
import { MessageRepository } from 'src/modules/messaging/common/repositories/message.repository';
|
||||
import { MessageChannelMessageAssociationWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-channel-message-association.workspace-entity';
|
||||
import { MessageChannelWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity';
|
||||
import { MessageThreadWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-thread.workspace-entity';
|
||||
import { MessageWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message.workspace-entity';
|
||||
import { GmailMessage } from 'src/modules/messaging/message-import-manager/drivers/gmail/types/gmail-message';
|
||||
@ -19,8 +17,6 @@ import { MessagingMessageThreadService } from 'src/modules/messaging/common/serv
|
||||
|
||||
@Injectable()
|
||||
export class MessagingMessageService {
|
||||
private readonly logger = new Logger(MessagingMessageService.name);
|
||||
|
||||
constructor(
|
||||
private readonly workspaceDataSourceService: WorkspaceDataSourceService,
|
||||
@InjectObjectMetadataRepository(
|
||||
@ -29,8 +25,6 @@ export class MessagingMessageService {
|
||||
private readonly messageChannelMessageAssociationRepository: MessageChannelMessageAssociationRepository,
|
||||
@InjectObjectMetadataRepository(MessageWorkspaceEntity)
|
||||
private readonly messageRepository: MessageRepository,
|
||||
@InjectObjectMetadataRepository(MessageChannelWorkspaceEntity)
|
||||
private readonly messageChannelRepository: MessageChannelRepository,
|
||||
@InjectObjectMetadataRepository(MessageThreadWorkspaceEntity)
|
||||
private readonly messageThreadRepository: MessageThreadRepository,
|
||||
private readonly messageThreadService: MessagingMessageThreadService,
|
||||
@ -101,104 +95,6 @@ export class MessagingMessageService {
|
||||
return messageExternalIdsAndIdsMap;
|
||||
}
|
||||
|
||||
public async saveMessages(
|
||||
messages: GmailMessage[],
|
||||
workspaceDataSource: DataSource,
|
||||
connectedAccount: ConnectedAccountWorkspaceEntity,
|
||||
gmailMessageChannelId: string,
|
||||
workspaceId: string,
|
||||
): Promise<Map<string, string>> {
|
||||
const messageExternalIdsAndIdsMap = new Map<string, string>();
|
||||
|
||||
try {
|
||||
let keepImporting = true;
|
||||
|
||||
for (const message of messages) {
|
||||
if (!keepImporting) {
|
||||
break;
|
||||
}
|
||||
|
||||
await workspaceDataSource?.transaction(
|
||||
async (manager: EntityManager) => {
|
||||
const gmailMessageChannel =
|
||||
await this.messageChannelRepository.getByIds(
|
||||
[gmailMessageChannelId],
|
||||
workspaceId,
|
||||
manager,
|
||||
);
|
||||
|
||||
if (gmailMessageChannel.length === 0) {
|
||||
this.logger.error(
|
||||
`No message channel found for connected account ${connectedAccount.id} in workspace ${workspaceId} in saveMessages`,
|
||||
);
|
||||
|
||||
keepImporting = false;
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
const existingMessageChannelMessageAssociationsCount =
|
||||
await this.messageChannelMessageAssociationRepository.countByMessageExternalIdsAndMessageChannelId(
|
||||
[message.externalId],
|
||||
gmailMessageChannelId,
|
||||
workspaceId,
|
||||
manager,
|
||||
);
|
||||
|
||||
if (existingMessageChannelMessageAssociationsCount > 0) {
|
||||
return;
|
||||
}
|
||||
|
||||
// TODO: This does not handle all thread merging use cases and might create orphan threads.
|
||||
const savedOrExistingMessageThreadId =
|
||||
await this.messageThreadService.saveMessageThreadOrReturnExistingMessageThread(
|
||||
message.headerMessageId,
|
||||
message.messageThreadExternalId,
|
||||
workspaceId,
|
||||
manager,
|
||||
);
|
||||
|
||||
if (!savedOrExistingMessageThreadId) {
|
||||
throw new Error(
|
||||
`No message thread found for message ${message.headerMessageId} in workspace ${workspaceId} in saveMessages`,
|
||||
);
|
||||
}
|
||||
|
||||
const savedOrExistingMessageId =
|
||||
await this.saveMessageOrReturnExistingMessage(
|
||||
message,
|
||||
savedOrExistingMessageThreadId,
|
||||
connectedAccount,
|
||||
workspaceId,
|
||||
manager,
|
||||
);
|
||||
|
||||
messageExternalIdsAndIdsMap.set(
|
||||
message.externalId,
|
||||
savedOrExistingMessageId,
|
||||
);
|
||||
|
||||
await this.messageChannelMessageAssociationRepository.insert(
|
||||
gmailMessageChannelId,
|
||||
savedOrExistingMessageId,
|
||||
message.externalId,
|
||||
savedOrExistingMessageThreadId,
|
||||
message.messageThreadExternalId,
|
||||
workspaceId,
|
||||
manager,
|
||||
);
|
||||
},
|
||||
);
|
||||
}
|
||||
} catch (error) {
|
||||
throw new Error(
|
||||
`Error saving connected account ${connectedAccount.id} messages to workspace ${workspaceId}: ${error.message}`,
|
||||
);
|
||||
}
|
||||
|
||||
return messageExternalIdsAndIdsMap;
|
||||
}
|
||||
|
||||
private async saveMessageOrReturnExistingMessage(
|
||||
message: GmailMessage,
|
||||
messageThreadId: string,
|
||||
@ -219,8 +115,11 @@ export class MessagingMessageService {
|
||||
|
||||
const newMessageId = v4();
|
||||
|
||||
const messageDirection =
|
||||
connectedAccount.handle === message.fromHandle ? 'outgoing' : 'incoming';
|
||||
const messageDirection = connectedAccount.emailAliases?.includes(
|
||||
message.fromHandle,
|
||||
)
|
||||
? 'outgoing'
|
||||
: 'incoming';
|
||||
|
||||
const receivedAt = new Date(parseInt(message.internalDate));
|
||||
|
||||
|
||||
@ -58,6 +58,8 @@ export class MessagingSaveMessagesAndEnqueueContactCreationService {
|
||||
value: true,
|
||||
});
|
||||
|
||||
const emailAliases = connectedAccount.emailAliases?.split(',') || [];
|
||||
|
||||
const isContactCreationForSentAndReceivedEmailsEnabled =
|
||||
isContactCreationForSentAndReceivedEmailsEnabledFeatureFlag?.value;
|
||||
|
||||
@ -80,15 +82,21 @@ export class MessagingSaveMessagesAndEnqueueContactCreationService {
|
||||
const messageId = messageExternalIdsAndIdsMap.get(message.externalId);
|
||||
|
||||
return messageId
|
||||
? message.participants.map((participant: Participant) => ({
|
||||
...participant,
|
||||
messageId,
|
||||
shouldCreateContact:
|
||||
messageChannel.isContactAutoCreationEnabled &&
|
||||
(isContactCreationForSentAndReceivedEmailsEnabled ||
|
||||
message.participants.find((p) => p.role === 'from')
|
||||
?.handle === connectedAccount.handle),
|
||||
}))
|
||||
? message.participants.map((participant: Participant) => {
|
||||
const fromHandle =
|
||||
message.participants.find((p) => p.role === 'from')?.handle ||
|
||||
'';
|
||||
|
||||
return {
|
||||
...participant,
|
||||
messageId,
|
||||
shouldCreateContact:
|
||||
messageChannel.isContactAutoCreationEnabled &&
|
||||
(isContactCreationForSentAndReceivedEmailsEnabled ||
|
||||
emailAliases.includes(fromHandle)) &&
|
||||
!emailAliases.includes(participant.handle),
|
||||
};
|
||||
})
|
||||
: [];
|
||||
});
|
||||
|
||||
|
||||
@ -2,8 +2,11 @@ import { Module } from '@nestjs/common';
|
||||
import { TypeOrmModule } from '@nestjs/typeorm';
|
||||
|
||||
import { FeatureFlagEntity } from 'src/engine/core-modules/feature-flag/feature-flag.entity';
|
||||
import { FeatureFlagModule } from 'src/engine/core-modules/feature-flag/feature-flag.module';
|
||||
import { EnvironmentModule } from 'src/engine/integrations/environment/environment.module';
|
||||
import { ObjectMetadataRepositoryModule } from 'src/engine/object-metadata-repository/object-metadata-repository.module';
|
||||
import { EmailAliasManagerModule } from 'src/modules/connected-account/email-alias-manager/email-alias-manager.module';
|
||||
import { OAuth2ClientManagerModule } from 'src/modules/connected-account/oauth2-client-manager/oauth2-client-manager.module';
|
||||
import { GoogleAPIRefreshAccessTokenModule } from 'src/modules/connected-account/services/google-api-refresh-access-token/google-api-refresh-access-token.module';
|
||||
import { BlocklistWorkspaceEntity } from 'src/modules/connected-account/standard-objects/blocklist.workspace-entity';
|
||||
import { ConnectedAccountWorkspaceEntity } from 'src/modules/connected-account/standard-objects/connected-account.workspace-entity';
|
||||
@ -30,6 +33,9 @@ import { MessagingGmailPartialMessageListFetchService } from 'src/modules/messag
|
||||
]),
|
||||
MessagingCommonModule,
|
||||
TypeOrmModule.forFeature([FeatureFlagEntity], 'core'),
|
||||
OAuth2ClientManagerModule,
|
||||
EmailAliasManagerModule,
|
||||
FeatureFlagModule,
|
||||
],
|
||||
providers: [
|
||||
MessagingGmailClientProvider,
|
||||
|
||||
@ -1,16 +1,21 @@
|
||||
import { Injectable } from '@nestjs/common';
|
||||
|
||||
import { OAuth2Client } from 'google-auth-library';
|
||||
import { gmail_v1, google } from 'googleapis';
|
||||
|
||||
import { EnvironmentService } from 'src/engine/integrations/environment/environment.service';
|
||||
import { OAuth2ClientManagerService } from 'src/modules/connected-account/oauth2-client-manager/services/oauth2-client-manager.service';
|
||||
import { ConnectedAccountWorkspaceEntity } from 'src/modules/connected-account/standard-objects/connected-account.workspace-entity';
|
||||
|
||||
@Injectable()
|
||||
export class MessagingGmailClientProvider {
|
||||
constructor(private readonly environmentService: EnvironmentService) {}
|
||||
constructor(
|
||||
private readonly oAuth2ClientManagerService: OAuth2ClientManagerService,
|
||||
) {}
|
||||
|
||||
public async getGmailClient(refreshToken: string): Promise<gmail_v1.Gmail> {
|
||||
const oAuth2Client = await this.getOAuth2Client(refreshToken);
|
||||
public async getGmailClient(
|
||||
connectedAccount: ConnectedAccountWorkspaceEntity,
|
||||
): Promise<gmail_v1.Gmail> {
|
||||
const oAuth2Client =
|
||||
await this.oAuth2ClientManagerService.getOAuth2Client(connectedAccount);
|
||||
|
||||
const gmailClient = google.gmail({
|
||||
version: 'v1',
|
||||
@ -19,22 +24,4 @@ export class MessagingGmailClientProvider {
|
||||
|
||||
return gmailClient;
|
||||
}
|
||||
|
||||
private async getOAuth2Client(refreshToken: string): Promise<OAuth2Client> {
|
||||
const gmailClientId = this.environmentService.get('AUTH_GOOGLE_CLIENT_ID');
|
||||
const gmailClientSecret = this.environmentService.get(
|
||||
'AUTH_GOOGLE_CLIENT_SECRET',
|
||||
);
|
||||
|
||||
const oAuth2Client = new google.auth.OAuth2(
|
||||
gmailClientId,
|
||||
gmailClientSecret,
|
||||
);
|
||||
|
||||
oAuth2Client.setCredentials({
|
||||
refresh_token: refreshToken,
|
||||
});
|
||||
|
||||
return oAuth2Client;
|
||||
}
|
||||
}
|
||||
|
||||
@ -54,9 +54,7 @@ export class MessagingGmailFullMessageListFetchService {
|
||||
);
|
||||
|
||||
const gmailClient: gmail_v1.Gmail =
|
||||
await this.gmailClientProvider.getGmailClient(
|
||||
connectedAccount.refreshToken,
|
||||
);
|
||||
await this.gmailClientProvider.getGmailClient(connectedAccount);
|
||||
|
||||
const { error: gmailError } =
|
||||
await this.fetchAllMessageIdsFromGmailAndStoreInCache(
|
||||
|
||||
@ -20,6 +20,9 @@ import { MessagingGmailFetchMessagesByBatchesService } from 'src/modules/messagi
|
||||
import { MessagingErrorHandlingService } from 'src/modules/messaging/common/services/messaging-error-handling.service';
|
||||
import { MessagingSaveMessagesAndEnqueueContactCreationService } from 'src/modules/messaging/common/services/messaging-save-messages-and-enqueue-contact-creation.service';
|
||||
import { MessageChannelRepository } from 'src/modules/messaging/common/repositories/message-channel.repository';
|
||||
import { EmailAliasManagerService } from 'src/modules/connected-account/email-alias-manager/services/email-alias-manager.service';
|
||||
import { IsFeatureEnabledService } from 'src/engine/core-modules/feature-flag/services/is-feature-enabled.service';
|
||||
import { FeatureFlagKeys } from 'src/engine/core-modules/feature-flag/feature-flag.entity';
|
||||
import { ConnectedAccountRepository } from 'src/modules/connected-account/repositories/connected-account.repository';
|
||||
|
||||
@Injectable()
|
||||
@ -41,6 +44,8 @@ export class MessagingGmailMessagesImportService {
|
||||
private readonly blocklistRepository: BlocklistRepository,
|
||||
@InjectObjectMetadataRepository(MessageChannelWorkspaceEntity)
|
||||
private readonly messageChannelRepository: MessageChannelRepository,
|
||||
private readonly emailAliasManagerService: EmailAliasManagerService,
|
||||
private readonly isFeatureEnabledService: IsFeatureEnabledService,
|
||||
@InjectObjectMetadataRepository(ConnectedAccountWorkspaceEntity)
|
||||
private readonly connectedAccountRepository: ConnectedAccountRepository,
|
||||
) {}
|
||||
@ -78,8 +83,8 @@ export class MessagingGmailMessagesImportService {
|
||||
try {
|
||||
accessToken =
|
||||
await this.googleAPIsRefreshAccessTokenService.refreshAndSaveAccessToken(
|
||||
connectedAccount,
|
||||
workspaceId,
|
||||
connectedAccount.id,
|
||||
);
|
||||
} catch (error) {
|
||||
await this.messagingTelemetryService.track({
|
||||
@ -103,6 +108,30 @@ export class MessagingGmailMessagesImportService {
|
||||
return;
|
||||
}
|
||||
|
||||
if (
|
||||
await this.isFeatureEnabledService.isFeatureEnabled(
|
||||
FeatureFlagKeys.IsMessagingAliasFetchingEnabled,
|
||||
workspaceId,
|
||||
)
|
||||
) {
|
||||
try {
|
||||
await this.emailAliasManagerService.refreshEmailAliases(
|
||||
connectedAccount,
|
||||
workspaceId,
|
||||
);
|
||||
} catch (error) {
|
||||
await this.gmailErrorHandlingService.handleGmailError(
|
||||
{
|
||||
code: error.code,
|
||||
reason: error.message,
|
||||
},
|
||||
'messages-import',
|
||||
messageChannel,
|
||||
workspaceId,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
const messageIdsToFetch =
|
||||
(await this.cacheStorage.setPop(
|
||||
`messages-to-import:${workspaceId}:gmail:${messageChannel.id}`,
|
||||
|
||||
@ -52,9 +52,7 @@ export class MessagingGmailPartialMessageListFetchService {
|
||||
const lastSyncHistoryId = messageChannel.syncCursor;
|
||||
|
||||
const gmailClient: gmail_v1.Gmail =
|
||||
await this.gmailClientProvider.getGmailClient(
|
||||
connectedAccount.refreshToken,
|
||||
);
|
||||
await this.gmailClientProvider.getGmailClient(connectedAccount);
|
||||
|
||||
const { history, historyId, error } =
|
||||
await this.gmailGetHistoryService.getHistory(
|
||||
|
||||
Reference in New Issue
Block a user