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

[messaging] Fix thread cleaner service subqueries (#4416)

Browse Source 
* [messaging] Fix thread cleaner service subqueries

* add pagination

* various fixes

* Fix thread merging

* fix

* fix
This commit is contained in:
Weiko
2024-03-12 17:49:45 +01:00
committed by GitHub
parent 91f4e1a853
commit 4476f5215b
17 changed files with 311 additions and 115 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
packages/twenty-server/src/core/auth/services/google-gmail.service.ts
View File

@ -79,7 +79,6 @@ export class GoogleGmailService {
connectedAccountId,
},
{
id: connectedAccountId,
retryLimit: 2,
},
);

18
packages/twenty-server/src/workspace/messaging/jobs/gmail-full-sync.job.ts
View File

@ -26,10 +26,20 @@ export class GmailFullSyncJob implements MessageQueueJob<GmailFullSyncJobData> {
data.connectedAccountId
} ${data.nextPageToken ? `and ${data.nextPageToken} pageToken` : ''}`,
);
await this.gmailRefreshAccessTokenService.refreshAndSaveAccessToken(
data.workspaceId,
data.connectedAccountId,
);
try {
await this.gmailRefreshAccessTokenService.refreshAndSaveAccessToken(
data.workspaceId,
data.connectedAccountId,
);
} catch (e) {
this.logger.error(
`Error refreshing access token for connected account ${data.connectedAccountId} in workspace ${data.workspaceId}`,
e,
);
return;
}
await this.gmailFullSyncService.fetchConnectedAccountThreads(
data.workspaceId,

18
packages/twenty-server/src/workspace/messaging/jobs/gmail-partial-sync.job.ts
View File

@ -25,10 +25,20 @@ export class GmailPartialSyncJob
this.logger.log(
`gmail partial-sync for workspace ${data.workspaceId} and account ${data.connectedAccountId}`,
);
await this.gmailRefreshAccessTokenService.refreshAndSaveAccessToken(
data.workspaceId,
data.connectedAccountId,
);
try {
await this.gmailRefreshAccessTokenService.refreshAndSaveAccessToken(
data.workspaceId,
data.connectedAccountId,
);
} catch (e) {
this.logger.error(
`Error refreshing access token for connected account ${data.connectedAccountId} in workspace ${data.workspaceId}`,
e,
);
return;
}
await this.gmailPartialSyncService.fetchConnectedAccountThreads(
data.workspaceId,

24
packages/twenty-server/src/workspace/messaging/repositories/connected-account/connected-account.service.ts
View File

@ -43,11 +43,11 @@ export class ConnectedAccountService {
);
}
public async getByIdOrFail(
public async getById(
connectedAccountId: string,
workspaceId: string,
transactionManager?: EntityManager,
): Promise<ObjectRecord<ConnectedAccountObjectMetadata>> {
): Promise<ObjectRecord<ConnectedAccountObjectMetadata> | undefined> {
const dataSourceSchema =
this.workspaceDataSourceService.getSchemaName(workspaceId);
@ -59,13 +59,27 @@ export class ConnectedAccountService {
transactionManager,
);
if (!connectedAccounts || connectedAccounts.length === 0) {
return connectedAccounts[0];
}
public async getByIdOrFail(
connectedAccountId: string,
workspaceId: string,
transactionManager?: EntityManager,
): Promise<ObjectRecord<ConnectedAccountObjectMetadata>> {
const connectedAccount = await this.getById(
connectedAccountId,
workspaceId,
transactionManager,
);
if (!connectedAccount) {
throw new NotFoundException(
`No connected account found for id ${connectedAccountId} in workspace ${workspaceId}`,
`Connected account with id ${connectedAccountId} not found in workspace ${workspaceId}`,
);
}
return connectedAccounts[0];
return connectedAccount;
}
public async updateLastSyncHistoryId(

12
packages/twenty-server/src/workspace/messaging/repositories/message-channel-message-association/message-channel-message-association.service.ts
View File

@ -84,18 +84,6 @@ export class MessageChannelMessageAssociationService {
);
}
public async deleteByMessageChannelId(
messageChannelId: string,
workspaceId: string,
transactionManager?: EntityManager,
) {
this.deleteByMessageChannelIds(
[messageChannelId],
workspaceId,
transactionManager,
);
}
public async deleteByMessageChannelIds(
messageChannelIds: string[],
workspaceId: string,

18
packages/twenty-server/src/workspace/messaging/repositories/message-channel/message-channel.service.ts
View File

@ -32,17 +32,29 @@ export class MessageChannelService {
connectedAccountId: string,
workspaceId: string,
): Promise<ObjectRecord<MessageChannelObjectMetadata>> {
const messageChannels = await this.getByConnectedAccountId(
const messageChannel = await this.getFirstByConnectedAccountId(
connectedAccountId,
workspaceId,
);
if (!messageChannels || messageChannels.length === 0) {
if (!messageChannel) {
throw new Error(
`No message channel found for connected account ${connectedAccountId} in workspace ${workspaceId}`,
`Message channel for connected account ${connectedAccountId} not found in workspace ${workspaceId}`,
);
}
return messageChannel;
}
public async getFirstByConnectedAccountId(
connectedAccountId: string,
workspaceId: string,
): Promise<ObjectRecord<MessageChannelObjectMetadata> | undefined> {
const messageChannels = await this.getByConnectedAccountId(
connectedAccountId,
workspaceId,
);
return messageChannels[0];
}

9
packages/twenty-server/src/workspace/messaging/repositories/message-thread/message-thread.module.ts
View File

@ -1,11 +1,16 @@
import { Module } from '@nestjs/common';
import { Module, forwardRef } from '@nestjs/common';
import { MessageChannelMessageAssociationModule } from 'src/workspace/messaging/repositories/message-channel-message-association/message-channel-message-assocation.module';
import { MessageThreadService } from 'src/workspace/messaging/repositories/message-thread/message-thread.service';
import { MessageModule } from 'src/workspace/messaging/repositories/message/message.module';
import { WorkspaceDataSourceModule } from 'src/workspace/workspace-datasource/workspace-datasource.module';
@Module({
imports: [WorkspaceDataSourceModule, MessageChannelMessageAssociationModule],
imports: [
WorkspaceDataSourceModule,
MessageChannelMessageAssociationModule,
forwardRef(() => MessageModule),
],
providers: [MessageThreadService],
exports: [MessageThreadService],
})

46
packages/twenty-server/src/workspace/messaging/repositories/message-thread/message-thread.service.ts
View File

@ -1,4 +1,4 @@
import { Injectable } from '@nestjs/common';
import { Inject, Injectable, forwardRef } from '@nestjs/common';
import { EntityManager } from 'typeorm';
import { v4 } from 'uuid';
@ -6,33 +6,38 @@ import { v4 } from 'uuid';
import { WorkspaceDataSourceService } from 'src/workspace/workspace-datasource/workspace-datasource.service';
import { DataSourceEntity } from 'src/metadata/data-source/data-source.entity';
import { MessageChannelMessageAssociationService } from 'src/workspace/messaging/repositories/message-channel-message-association/message-channel-message-association.service';
import { MessageThreadObjectMetadata } from 'src/workspace/workspace-sync-metadata/standard-objects/message-thread.object-metadata';
import { ObjectRecord } from 'src/workspace/workspace-sync-metadata/types/object-record';
import { MessageService } from 'src/workspace/messaging/repositories/message/message.service';
@Injectable()
export class MessageThreadService {
constructor(
private readonly messageChannelMessageAssociationService: MessageChannelMessageAssociationService,
private readonly workspaceDataSourceService: WorkspaceDataSourceService,
@Inject(forwardRef(() => MessageService))
private readonly messageService: MessageService,
) {}
public async getOrphanThreads(
public async getOrphanThreadIdsPaginated(
limit: number,
offset: number,
workspaceId: string,
transactionManager?: EntityManager,
): Promise<ObjectRecord<MessageThreadObjectMetadata>[]> {
): Promise<string[]> {
const dataSourceSchema =
this.workspaceDataSourceService.getSchemaName(workspaceId);
return await this.workspaceDataSourceService.executeRawQuery(
`SELECT mt.* FROM ${dataSourceSchema}."messageThread" mt
WHERE NOT EXISTS (
SELECT 1 FROM ${dataSourceSchema}."message" m
WHERE m."messageThreadId" = mt.id
)`,
[],
const orphanThreads = await this.workspaceDataSourceService.executeRawQuery(
`SELECT mt.id
FROM ${dataSourceSchema}."messageThread" mt
LEFT JOIN ${dataSourceSchema}."message" m ON mt.id = m."messageThreadId"
WHERE m."messageThreadId" IS NULL
LIMIT $1 OFFSET $2`,
[limit, offset],
workspaceId,
transactionManager,
);
return orphanThreads.map(({ id }) => id);
}
public async deleteByIds(
@ -52,11 +57,13 @@ export class MessageThreadService {
}
public async saveMessageThreadOrReturnExistingMessageThread(
headerMessageId: string,
messageThreadExternalId: string,
dataSourceMetadata: DataSourceEntity,
workspaceId: string,
manager: EntityManager,
) {
// Check if message thread already exists via threadExternalId
const existingMessageChannelMessageAssociationByMessageThreadExternalId =
await this.messageChannelMessageAssociationService.getFirstByMessageThreadExternalId(
messageThreadExternalId,
@ -71,6 +78,21 @@ export class MessageThreadService {
return Promise.resolve(existingMessageThread);
}
// Check if message thread already exists via existing message headerMessageId
const existingMessageWithSameHeaderMessageId =
await this.messageService.getFirstOrNullByHeaderMessageId(
headerMessageId,
workspaceId,
manager,
);
if (existingMessageWithSameHeaderMessageId) {
return Promise.resolve(
existingMessageWithSameHeaderMessageId.messageThreadId,
);
}
// If message thread does not exist, create new message thread
const newMessageThreadId = v4();
await manager.query(

4
packages/twenty-server/src/workspace/messaging/repositories/message/message.module.ts
View File

@ -1,4 +1,4 @@
import { Module } from '@nestjs/common';
import { Module, forwardRef } from '@nestjs/common';
import { MessageChannelModule } from 'src/workspace/messaging/repositories/message-channel/message-channel.module';
import { MessageChannelMessageAssociationModule } from 'src/workspace/messaging/repositories/message-channel-message-association/message-channel-message-assocation.module';
@ -11,7 +11,7 @@ import { CreateCompaniesAndContactsModule } from 'src/workspace/messaging/servic
@Module({
imports: [
WorkspaceDataSourceModule,
MessageThreadModule,
forwardRef(() => MessageThreadModule),
MessageParticipantModule,
MessageChannelMessageAssociationModule,
MessageChannelModule,

70
packages/twenty-server/src/workspace/messaging/repositories/message/message.service.ts
View File

@ -1,4 +1,4 @@
import { Injectable } from '@nestjs/common';
import { Inject, Injectable, Logger, forwardRef } from '@nestjs/common';
import { DataSource, EntityManager } from 'typeorm';
import { v4 } from 'uuid';
@ -9,42 +9,47 @@ import { ObjectRecord } from 'src/workspace/workspace-sync-metadata/types/object
import { DataSourceEntity } from 'src/metadata/data-source/data-source.entity';
import { GmailMessage } from 'src/workspace/messaging/types/gmail-message';
import { ConnectedAccountObjectMetadata } from 'src/workspace/workspace-sync-metadata/standard-objects/connected-account.object-metadata';
import { MessageChannelService } from 'src/workspace/messaging/repositories/message-channel/message-channel.service';
import { MessageChannelMessageAssociationService } from 'src/workspace/messaging/repositories/message-channel-message-association/message-channel-message-association.service';
import { MessageParticipantService } from 'src/workspace/messaging/repositories/message-participant/message-participant.service';
import { MessageThreadService } from 'src/workspace/messaging/repositories/message-thread/message-thread.service';
import { CreateCompaniesAndContactsService } from 'src/workspace/messaging/services/create-companies-and-contacts/create-companies-and-contacts.service';
import { MessageChannelService } from 'src/workspace/messaging/repositories/message-channel/message-channel.service';
@Injectable()
export class MessageService {
private readonly logger = new Logger(MessageService.name);
constructor(
private readonly workspaceDataSourceService: WorkspaceDataSourceService,
private readonly messageChannelMessageAssociationService: MessageChannelMessageAssociationService,
@Inject(forwardRef(() => MessageThreadService))
private readonly messageThreadService: MessageThreadService,
private readonly messageParticipantService: MessageParticipantService,
private readonly messageChannelService: MessageChannelService,
private readonly createCompaniesAndContactsService: CreateCompaniesAndContactsService,
) {}
public async getNonAssociatedMessages(
public async getNonAssociatedMessageIdsPaginated(
limit: number,
offset: number,
workspaceId: string,
transactionManager?: EntityManager,
): Promise<ObjectRecord<MessageObjectMetadata>[]> {
): Promise<string[]> {
const dataSourceSchema =
this.workspaceDataSourceService.getSchemaName(workspaceId);
return await this.workspaceDataSourceService.executeRawQuery(
`SELECT m.* FROM ${dataSourceSchema}."message" m
WHERE NOT EXISTS (
SELECT 1 FROM ${dataSourceSchema}."messageChannelMessageAssociation" mcma
WHERE mcma."messageId" = m.id
)`,
[],
workspaceId,
transactionManager,
);
const nonAssociatedMessages =
await this.workspaceDataSourceService.executeRawQuery(
`SELECT m.id FROM ${dataSourceSchema}."message" m
LEFT JOIN ${dataSourceSchema}."messageChannelMessageAssociation" mcma
ON m.id = mcma."messageId"
WHERE mcma.id IS NULL
LIMIT $1 OFFSET $2`,
[limit, offset],
workspaceId,
transactionManager,
);
return nonAssociatedMessages.map(({ id }) => id);
}
public async getFirstByHeaderMessageId(
public async getFirstOrNullByHeaderMessageId(
headerMessageId: string,
workspaceId: string,
transactionManager?: EntityManager,
@ -125,9 +130,32 @@ export class MessageService {
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.messageChannelService.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.messageChannelMessageAssociationService.countByMessageExternalIdsAndMessageChannelId(
[message.externalId],
@ -140,8 +168,10 @@ export class MessageService {
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,
dataSourceMetadata,
workspaceId,
@ -193,7 +223,7 @@ export class MessageService {
workspaceId: string,
manager: EntityManager,
): Promise<string> {
const existingMessage = await this.getFirstByHeaderMessageId(
const existingMessage = await this.getFirstOrNullByHeaderMessageId(
message.headerMessageId,
workspaceId,
);

22
packages/twenty-server/src/workspace/messaging/services/gmail-full-sync.service.ts
View File

@ -46,11 +46,19 @@ export class GmailFullSyncService {
connectedAccountId: string,
nextPageToken?: string,
): Promise<void> {
const connectedAccount = await this.connectedAccountService.getByIdOrFail(
const connectedAccount = await this.connectedAccountService.getById(
connectedAccountId,
workspaceId,
);
if (!connectedAccount) {
this.logger.error(
`Connected account ${connectedAccountId} not found in workspace ${workspaceId} during full-sync`,
);
return;
}
const accessToken = connectedAccount.accessToken;
const refreshToken = connectedAccount.refreshToken;
const workspaceMemberId = connectedAccount.accountOwnerId;
@ -62,11 +70,19 @@ export class GmailFullSyncService {
}
const gmailMessageChannel =
await this.messageChannelService.getFirstByConnectedAccountIdOrFail(
await this.messageChannelService.getFirstByConnectedAccountId(
connectedAccountId,
workspaceId,
);
if (!gmailMessageChannel) {
this.logger.error(
`No message channel found for connected account ${connectedAccountId} in workspace ${workspaceId} during full-syn`,
);
return;
}
const gmailMessageChannelId = gmailMessageChannel.id;
const gmailClient =
@ -173,7 +189,7 @@ export class GmailFullSyncService {
);
if (messagesToSave.length > 0) {
this.saveMessagesAndCreateContactsService.saveMessagesAndCreateContacts(
await this.saveMessagesAndCreateContactsService.saveMessagesAndCreateContacts(
messagesToSave,
connectedAccount,
workspaceId,

20
packages/twenty-server/src/workspace/messaging/services/gmail-partial-sync.service.ts
View File

@ -48,11 +48,19 @@ export class GmailPartialSyncService {
connectedAccountId: string,
maxResults = 500,
): Promise<void> {
const connectedAccount = await this.connectedAccountService.getByIdOrFail(
const connectedAccount = await this.connectedAccountService.getById(
connectedAccountId,
workspaceId,
);
if (!connectedAccount) {
this.logger.error(
`Connected account ${connectedAccountId} not found in workspace ${workspaceId} during partial-sync`,
);
return;
}
const lastSyncHistoryId = connectedAccount.lastSyncHistoryId;
if (!lastSyncHistoryId) {
@ -135,11 +143,19 @@ export class GmailPartialSyncService {
}
const gmailMessageChannel =
await this.messageChannelService.getFirstByConnectedAccountIdOrFail(
await this.messageChannelService.getFirstByConnectedAccountId(
connectedAccountId,
workspaceId,
);
if (!gmailMessageChannel) {
this.logger.error(
`No message channel found for connected account ${connectedAccountId} in workspace ${workspaceId} during partial-sync`,
);
return;
}
const gmailMessageChannelId = gmailMessageChannel.id;
const { messagesAdded, messagesDeleted } =

8
packages/twenty-server/src/workspace/messaging/services/gmail-refresh-access-token.service.ts
View File

@ -16,11 +16,17 @@ export class GmailRefreshAccessTokenService {
workspaceId: string,
connectedAccountId: string,
): Promise<void> {
const connectedAccount = await this.connectedAccountService.getByIdOrFail(
const connectedAccount = await this.connectedAccountService.getById(
connectedAccountId,
workspaceId,
);
if (!connectedAccount) {
throw new Error(
`No connected account found for ${connectedAccountId} in workspace ${workspaceId}`,
);
}
const refreshToken = connectedAccount.refreshToken;
if (!refreshToken) {

15
packages/twenty-server/src/workspace/messaging/services/save-messages-and-create-contacts.service.ts
View File

@ -59,12 +59,23 @@ export class SaveMessagesAndCreateContactsService {
} in ${endTime - startTime}ms`,
);
const isContactAutoCreationEnabled =
await this.messageChannelService.getIsContactAutoCreationEnabledByConnectedAccountIdOrFail(
const gmailMessageChannel =
await this.messageChannelService.getFirstByConnectedAccountId(
connectedAccount.id,
workspaceId,
);
if (!gmailMessageChannel) {
this.logger.error(
`No message channel found for connected account ${connectedAccount.id} in workspace ${workspaceId} in saveMessagesAndCreateContacts`,
);
return;
}
const isContactAutoCreationEnabled =
gmailMessageChannel.isContactAutoCreationEnabled;
const participantsWithMessageId: ParticipantWithMessageId[] =
messagesToSave.flatMap((message) => {
const messageId = messageExternalIdsAndIdsMap.get(message.externalId);

59
packages/twenty-server/src/workspace/messaging/services/thread-cleaner/thread-cleaner.service.ts
View File

@ -4,6 +4,7 @@ import { TypeORMService } from 'src/database/typeorm/typeorm.service';
import { DataSourceService } from 'src/metadata/data-source/data-source.service';
import { MessageThreadService } from 'src/workspace/messaging/repositories/message-thread/message-thread.service';
import { MessageService } from 'src/workspace/messaging/repositories/message/message.service';
import { deleteUsingPagination } from 'src/workspace/messaging/services/thread-cleaner/utils/delete-using-pagination.util';
@Injectable()
export class ThreadCleanerService {
@ -15,48 +16,22 @@ export class ThreadCleanerService {
) {}
public async cleanWorkspaceThreads(workspaceId: string) {
const dataSourceMetadata =
await this.dataSourceService.getLastDataSourceMetadataFromWorkspaceIdOrFail(
workspaceId,
);
await deleteUsingPagination(
workspaceId,
500,
this.messageService.getNonAssociatedMessageIdsPaginated.bind(
this.messageService,
),
this.messageService.deleteByIds.bind(this.messageService),
);
const workspaceDataSource =
await this.typeORMService.connectToDataSource(dataSourceMetadata);
await workspaceDataSource?.transaction(async (transactionManager) => {
const messagesToDelete =
await this.messageService.getNonAssociatedMessages(
workspaceId,
transactionManager,
);
const messageIdsToDelete = messagesToDelete.map(({ id }) => id);
if (messageIdsToDelete.length > 0) {
await this.messageService.deleteByIds(
messageIdsToDelete,
workspaceId,
transactionManager,
);
}
const messageThreadsToDelete =
await this.messageThreadService.getOrphanThreads(
workspaceId,
transactionManager,
);
const messageThreadToDeleteIds = messageThreadsToDelete.map(
({ id }) => id,
);
if (messageThreadToDeleteIds.length > 0) {
await this.messageThreadService.deleteByIds(
messageThreadToDeleteIds,
workspaceId,
transactionManager,
);
}
});
await deleteUsingPagination(
workspaceId,
500,
this.messageThreadService.getOrphanThreadIdsPaginated.bind(
this.messageThreadService,
),
this.messageThreadService.deleteByIds.bind(this.messageThreadService),
);
}
}

44
packages/twenty-server/src/workspace/messaging/services/thread-cleaner/utils/delete-using-pagination.util.spec.ts Normal file
View File

@ -0,0 +1,44 @@
import { deleteUsingPagination } from './delete-using-pagination.util';
describe('deleteUsingPagination', () => {
it('should delete items using pagination', async () => {
const workspaceId = 'workspace123';
const batchSize = 10;
const getterPaginated = jest
.fn()
.mockResolvedValueOnce(['id1', 'id2'])
.mockResolvedValueOnce([]);
const deleter = jest.fn();
const transactionManager = undefined;
await deleteUsingPagination(
workspaceId,
batchSize,
getterPaginated,
deleter,
transactionManager,
);
expect(getterPaginated).toHaveBeenNthCalledWith(
1,
batchSize,
0,
workspaceId,
transactionManager,
);
expect(getterPaginated).toHaveBeenNthCalledWith(
2,
batchSize,
batchSize,
workspaceId,
transactionManager,
);
expect(deleter).toHaveBeenNthCalledWith(
1,
['id1', 'id2'],
workspaceId,
transactionManager,
);
expect(deleter).toHaveBeenCalledTimes(1);
});
});

38
packages/twenty-server/src/workspace/messaging/services/thread-cleaner/utils/delete-using-pagination.util.ts Normal file
View File

@ -0,0 +1,38 @@
import { EntityManager } from 'typeorm';
export const deleteUsingPagination = async (
workspaceId: string,
batchSize: number,
getterPaginated: (
limit: number,
offset: number,
workspaceId: string,
transactionManager?: EntityManager,
) => Promise<string[]>,
deleter: (
ids: string[],
workspaceId: string,
transactionManager?: EntityManager,
) => Promise<void>,
transactionManager?: EntityManager,
) => {
let offset = 0;
let hasMoreData = true;
while (hasMoreData) {
const idsToDelete = await getterPaginated(
batchSize,
offset,
workspaceId,
transactionManager,
);
if (idsToDelete.length > 0) {
await deleter(idsToDelete, workspaceId, transactionManager);
} else {
hasMoreData = false;
}
offset += batchSize;
}
};