Refactor global datasource part 3 (#16447)
## Context Following https://github.com/twentyhq/twenty/pull/16399 Now using the new global orm manager everywhere and returning a GlobalDatasource/WorkspaceDatasource based on a feature flag. This means we now need to wrap all our ORM calls within executeInWorkspaceContext callback (at least for now) so the global datasource can dynamically hydrate its context via the new store (the global datasource does not store anything related to workspaces as it is now a unique singleton). If feature flag is off it still uses local data stored in the workspace datasource.
This commit is contained in:
Weiko
+3
-3
@@ -7,7 +7,7 @@ import { ActiveOrSuspendedWorkspacesMigrationCommandRunner } from 'src/database/
|
||||
import { RunOnWorkspaceArgs } from 'src/database/commands/command-runners/workspaces-migration.command-runner';
|
||||
import { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity';
|
||||
import { DataSourceService } from 'src/engine/metadata-modules/data-source/data-source.service';
|
||||
import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager';
|
||||
import { GlobalWorkspaceOrmManager } from 'src/engine/twenty-orm/global-workspace-datasource/global-workspace-orm.manager';
|
||||
import { MessagingMessageCleanerService } from 'src/modules/messaging/message-cleaner/services/messaging-message-cleaner.service';
|
||||
|
||||
@Command({
|
||||
@@ -18,11 +18,11 @@ export class MessagingMessageCleanerRemoveOrphansCommand extends ActiveOrSuspend
|
||||
constructor(
|
||||
@InjectRepository(WorkspaceEntity)
|
||||
protected readonly workspaceRepository: Repository<WorkspaceEntity>,
|
||||
protected readonly twentyORMGlobalManager: TwentyORMGlobalManager,
|
||||
protected readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager,
|
||||
private readonly messagingMessageCleanerService: MessagingMessageCleanerService,
|
||||
protected readonly dataSourceService: DataSourceService,
|
||||
) {
|
||||
super(workspaceRepository, twentyORMGlobalManager, dataSourceService);
|
||||
super(workspaceRepository, globalWorkspaceOrmManager, dataSourceService);
|
||||
}
|
||||
|
||||
override async runOnWorkspace({
|
||||
|
||||
+47
-37
@@ -3,7 +3,8 @@ import { Logger } from '@nestjs/common';
|
||||
import { Command, CommandRunner, Option } from 'nest-commander';
|
||||
import { isDefined } from 'twenty-shared/utils';
|
||||
|
||||
import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager';
|
||||
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 { MessageChannelSyncStatusService } from 'src/modules/messaging/common/services/message-channel-sync-status.service';
|
||||
import { type MessageChannelWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity';
|
||||
import { MessagingMessageCleanerService } from 'src/modules/messaging/message-cleaner/services/messaging-message-cleaner.service';
|
||||
@@ -22,7 +23,7 @@ export class MessagingResetChannelCommand extends CommandRunner {
|
||||
private readonly logger = new Logger(MessagingResetChannelCommand.name);
|
||||
|
||||
constructor(
|
||||
private readonly twentyORMGlobalManager: TwentyORMGlobalManager,
|
||||
private readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager,
|
||||
private readonly messagingChannelSyncStatusService: MessageChannelSyncStatusService,
|
||||
private readonly messagingMessageCleanerService: MessagingMessageCleanerService,
|
||||
) {
|
||||
@@ -35,44 +36,53 @@ export class MessagingResetChannelCommand extends CommandRunner {
|
||||
): Promise<void> {
|
||||
const { workspaceId, messageChannelId } = options;
|
||||
|
||||
const messageChannelRepository =
|
||||
await this.twentyORMGlobalManager.getRepositoryForWorkspace<MessageChannelWorkspaceEntity>(
|
||||
workspaceId,
|
||||
'messageChannel',
|
||||
);
|
||||
const authContext = buildSystemAuthContext(workspaceId);
|
||||
|
||||
this.logger.log(
|
||||
`No message channel ID provided, resetting all message channels in workspace ${workspaceId}`,
|
||||
);
|
||||
await this.globalWorkspaceOrmManager.executeInWorkspaceContext(
|
||||
authContext,
|
||||
async () => {
|
||||
const messageChannelRepository =
|
||||
await this.globalWorkspaceOrmManager.getRepository<MessageChannelWorkspaceEntity>(
|
||||
workspaceId,
|
||||
'messageChannel',
|
||||
);
|
||||
|
||||
const messageChannels = await messageChannelRepository.find({
|
||||
where: {
|
||||
...(isDefined(messageChannelId) ? { id: messageChannelId } : {}),
|
||||
this.logger.log(
|
||||
`No message channel ID provided, resetting all message channels in workspace ${workspaceId}`,
|
||||
);
|
||||
|
||||
const messageChannels = await messageChannelRepository.find({
|
||||
where: {
|
||||
...(isDefined(messageChannelId) ? { id: messageChannelId } : {}),
|
||||
},
|
||||
});
|
||||
|
||||
if (messageChannels.length === 0) {
|
||||
this.logger.log(
|
||||
`No message channels found in workspace ${workspaceId}`,
|
||||
);
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
this.logger.log(
|
||||
`Found ${messageChannels.length} message channels to reset`,
|
||||
);
|
||||
|
||||
for (const messageChannel of messageChannels) {
|
||||
await this.messagingChannelSyncStatusService.resetAndMarkAsMessagesListFetchPending(
|
||||
[messageChannel.id],
|
||||
workspaceId,
|
||||
);
|
||||
await this.messagingMessageCleanerService.cleanOrphanMessagesAndThreads(
|
||||
workspaceId,
|
||||
);
|
||||
}
|
||||
|
||||
this.logger.log(
|
||||
`Successfully reset all ${messageChannels.length} message channels in workspace ${workspaceId}`,
|
||||
);
|
||||
},
|
||||
});
|
||||
|
||||
if (messageChannels.length === 0) {
|
||||
this.logger.log(`No message channels found in workspace ${workspaceId}`);
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
this.logger.log(
|
||||
`Found ${messageChannels.length} message channels to reset`,
|
||||
);
|
||||
|
||||
for (const messageChannel of messageChannels) {
|
||||
await this.messagingChannelSyncStatusService.resetAndMarkAsMessagesListFetchPending(
|
||||
[messageChannel.id],
|
||||
workspaceId,
|
||||
);
|
||||
await this.messagingMessageCleanerService.cleanOrphanMessagesAndThreads(
|
||||
workspaceId,
|
||||
);
|
||||
}
|
||||
|
||||
this.logger.log(
|
||||
`Successfully reset all ${messageChannels.length} message channels in workspace ${workspaceId}`,
|
||||
);
|
||||
}
|
||||
|
||||
|
||||
+165
-148
@@ -4,7 +4,8 @@ import chunk from 'lodash.chunk';
|
||||
import { In, IsNull } from 'typeorm';
|
||||
|
||||
import { type WorkspaceEntityManager } from 'src/engine/twenty-orm/entity-manager/workspace-entity-manager';
|
||||
import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager';
|
||||
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 { MessageChannelMessageAssociationWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-channel-message-association.workspace-entity';
|
||||
import { type MessageThreadWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-thread.workspace-entity';
|
||||
import { type MessageWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message.workspace-entity';
|
||||
@@ -14,7 +15,7 @@ import { deleteUsingPagination } from 'src/modules/messaging/message-cleaner/uti
|
||||
export class MessagingMessageCleanerService {
|
||||
private readonly logger = new Logger(MessagingMessageCleanerService.name);
|
||||
constructor(
|
||||
private readonly twentyORMGlobalManager: TwentyORMGlobalManager,
|
||||
private readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager,
|
||||
) {}
|
||||
|
||||
async deleteMessagesChannelMessageAssociationsAndRelatedOrphans({
|
||||
@@ -26,183 +27,199 @@ export class MessagingMessageCleanerService {
|
||||
messageExternalIds: string[];
|
||||
messageChannelId: string;
|
||||
}) {
|
||||
const messageRepository =
|
||||
await this.twentyORMGlobalManager.getRepositoryForWorkspace<MessageWorkspaceEntity>(
|
||||
workspaceId,
|
||||
'message',
|
||||
);
|
||||
const authContext = buildSystemAuthContext(workspaceId);
|
||||
|
||||
const messageChannelMessageAssociationRepository =
|
||||
await this.twentyORMGlobalManager.getRepositoryForWorkspace<MessageChannelMessageAssociationWorkspaceEntity>(
|
||||
workspaceId,
|
||||
'messageChannelMessageAssociation',
|
||||
);
|
||||
await this.globalWorkspaceOrmManager.executeInWorkspaceContext(
|
||||
authContext,
|
||||
async () => {
|
||||
const messageRepository =
|
||||
await this.globalWorkspaceOrmManager.getRepository<MessageWorkspaceEntity>(
|
||||
workspaceId,
|
||||
'message',
|
||||
);
|
||||
|
||||
const messageThreadRepository =
|
||||
await this.twentyORMGlobalManager.getRepositoryForWorkspace<MessageThreadWorkspaceEntity>(
|
||||
workspaceId,
|
||||
'messageThread',
|
||||
);
|
||||
const messageChannelMessageAssociationRepository =
|
||||
await this.globalWorkspaceOrmManager.getRepository<MessageChannelMessageAssociationWorkspaceEntity>(
|
||||
workspaceId,
|
||||
'messageChannelMessageAssociation',
|
||||
);
|
||||
|
||||
const messageExternalIdsChunks = chunk(messageExternalIds, 500);
|
||||
const messageThreadRepository =
|
||||
await this.globalWorkspaceOrmManager.getRepository<MessageThreadWorkspaceEntity>(
|
||||
workspaceId,
|
||||
'messageThread',
|
||||
);
|
||||
|
||||
for (const messageExternalIdsChunk of messageExternalIdsChunks) {
|
||||
const messageChannelMessageAssociationsToDelete =
|
||||
await messageChannelMessageAssociationRepository.find({
|
||||
where: {
|
||||
messageExternalId: In(messageExternalIdsChunk),
|
||||
messageChannelId,
|
||||
},
|
||||
});
|
||||
const messageExternalIdsChunks = chunk(messageExternalIds, 500);
|
||||
|
||||
if (messageChannelMessageAssociationsToDelete.length <= 0) {
|
||||
continue;
|
||||
}
|
||||
for (const messageExternalIdsChunk of messageExternalIdsChunks) {
|
||||
const messageChannelMessageAssociationsToDelete =
|
||||
await messageChannelMessageAssociationRepository.find({
|
||||
where: {
|
||||
messageExternalId: In(messageExternalIdsChunk),
|
||||
messageChannelId,
|
||||
},
|
||||
});
|
||||
|
||||
await messageChannelMessageAssociationRepository.delete(
|
||||
messageChannelMessageAssociationsToDelete.map(({ id }) => id),
|
||||
);
|
||||
if (messageChannelMessageAssociationsToDelete.length <= 0) {
|
||||
continue;
|
||||
}
|
||||
|
||||
this.logger.log(
|
||||
`WorkspaceId: ${workspaceId} Deleting ${messageChannelMessageAssociationsToDelete.length} message channel message associations`,
|
||||
);
|
||||
await messageChannelMessageAssociationRepository.delete(
|
||||
messageChannelMessageAssociationsToDelete.map(({ id }) => id),
|
||||
);
|
||||
|
||||
const orphanMessages = await messageRepository.find({
|
||||
where: {
|
||||
id: In(
|
||||
messageChannelMessageAssociationsToDelete.map(
|
||||
({ messageId }) => messageId,
|
||||
),
|
||||
),
|
||||
messageChannelMessageAssociations: {
|
||||
id: IsNull(),
|
||||
},
|
||||
},
|
||||
});
|
||||
this.logger.log(
|
||||
`WorkspaceId: ${workspaceId} Deleting ${messageChannelMessageAssociationsToDelete.length} message channel message associations`,
|
||||
);
|
||||
|
||||
if (orphanMessages.length <= 0) {
|
||||
continue;
|
||||
}
|
||||
const orphanMessages = await messageRepository.find({
|
||||
where: {
|
||||
id: In(
|
||||
messageChannelMessageAssociationsToDelete.map(
|
||||
({ messageId }) => messageId,
|
||||
),
|
||||
),
|
||||
messageChannelMessageAssociations: {
|
||||
id: IsNull(),
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
this.logger.log(
|
||||
`WorkspaceId: ${workspaceId} Deleting ${orphanMessages.length} orphan messages`,
|
||||
);
|
||||
if (orphanMessages.length <= 0) {
|
||||
continue;
|
||||
}
|
||||
|
||||
await messageRepository.delete(orphanMessages.map(({ id }) => id));
|
||||
this.logger.log(
|
||||
`WorkspaceId: ${workspaceId} Deleting ${orphanMessages.length} orphan messages`,
|
||||
);
|
||||
|
||||
const orphanMessageThreads = await messageThreadRepository.find({
|
||||
where: {
|
||||
id: In(orphanMessages.map(({ messageThreadId }) => messageThreadId)),
|
||||
messages: {
|
||||
id: IsNull(),
|
||||
},
|
||||
},
|
||||
});
|
||||
await messageRepository.delete(orphanMessages.map(({ id }) => id));
|
||||
|
||||
if (orphanMessageThreads.length <= 0) {
|
||||
continue;
|
||||
}
|
||||
const orphanMessageThreads = await messageThreadRepository.find({
|
||||
where: {
|
||||
id: In(
|
||||
orphanMessages.map(({ messageThreadId }) => messageThreadId),
|
||||
),
|
||||
messages: {
|
||||
id: IsNull(),
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
this.logger.log(
|
||||
`WorkspaceId: ${workspaceId} Deleting ${orphanMessageThreads.length} orphan message threads`,
|
||||
);
|
||||
if (orphanMessageThreads.length <= 0) {
|
||||
continue;
|
||||
}
|
||||
|
||||
await messageThreadRepository.delete(
|
||||
orphanMessageThreads.map(({ id }) => id),
|
||||
);
|
||||
}
|
||||
this.logger.log(
|
||||
`WorkspaceId: ${workspaceId} Deleting ${orphanMessageThreads.length} orphan message threads`,
|
||||
);
|
||||
|
||||
await messageThreadRepository.delete(
|
||||
orphanMessageThreads.map(({ id }) => id),
|
||||
);
|
||||
}
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
public async cleanOrphanMessagesAndThreads(workspaceId: string) {
|
||||
const messageThreadRepository =
|
||||
await this.twentyORMGlobalManager.getRepositoryForWorkspace<MessageThreadWorkspaceEntity>(
|
||||
workspaceId,
|
||||
'messageThread',
|
||||
);
|
||||
const authContext = buildSystemAuthContext(workspaceId);
|
||||
|
||||
const messageRepository =
|
||||
await this.twentyORMGlobalManager.getRepositoryForWorkspace<MessageWorkspaceEntity>(
|
||||
workspaceId,
|
||||
'message',
|
||||
);
|
||||
await this.globalWorkspaceOrmManager.executeInWorkspaceContext(
|
||||
authContext,
|
||||
async () => {
|
||||
const messageThreadRepository =
|
||||
await this.globalWorkspaceOrmManager.getRepository<MessageThreadWorkspaceEntity>(
|
||||
workspaceId,
|
||||
'messageThread',
|
||||
);
|
||||
|
||||
const workspaceDataSource =
|
||||
await this.twentyORMGlobalManager.getDataSourceForWorkspace({
|
||||
workspaceId,
|
||||
});
|
||||
const messageRepository =
|
||||
await this.globalWorkspaceOrmManager.getRepository<MessageWorkspaceEntity>(
|
||||
workspaceId,
|
||||
'message',
|
||||
);
|
||||
|
||||
await workspaceDataSource.transaction(
|
||||
async (transactionManager: WorkspaceEntityManager) => {
|
||||
await deleteUsingPagination(
|
||||
workspaceId,
|
||||
500,
|
||||
async (
|
||||
limit: number,
|
||||
offset: number,
|
||||
_workspaceId: string,
|
||||
transactionManager: WorkspaceEntityManager,
|
||||
) => {
|
||||
const nonAssociatedMessages = await messageRepository.find(
|
||||
{
|
||||
where: {
|
||||
messageChannelMessageAssociations: {
|
||||
id: IsNull(),
|
||||
const workspaceDataSource =
|
||||
await this.globalWorkspaceOrmManager.getDataSourceForWorkspace(
|
||||
workspaceId,
|
||||
);
|
||||
|
||||
await workspaceDataSource.transaction(
|
||||
async (transactionManager: WorkspaceEntityManager) => {
|
||||
await deleteUsingPagination(
|
||||
workspaceId,
|
||||
500,
|
||||
async (
|
||||
limit: number,
|
||||
offset: number,
|
||||
_workspaceId: string,
|
||||
transactionManager: WorkspaceEntityManager,
|
||||
) => {
|
||||
const nonAssociatedMessages = await messageRepository.find(
|
||||
{
|
||||
where: {
|
||||
messageChannelMessageAssociations: {
|
||||
id: IsNull(),
|
||||
},
|
||||
},
|
||||
take: limit,
|
||||
skip: offset,
|
||||
relations: ['messageChannelMessageAssociations'],
|
||||
},
|
||||
},
|
||||
take: limit,
|
||||
skip: offset,
|
||||
relations: ['messageChannelMessageAssociations'],
|
||||
transactionManager,
|
||||
);
|
||||
|
||||
return nonAssociatedMessages.map(({ id }) => id);
|
||||
},
|
||||
async (
|
||||
ids: string[],
|
||||
workspaceId: string,
|
||||
transactionManager?: WorkspaceEntityManager,
|
||||
) => {
|
||||
this.logger.log(
|
||||
`WorkspaceId: ${workspaceId} Deleting ${ids.length} messages from message cleaner`,
|
||||
);
|
||||
await messageRepository.delete(ids, transactionManager);
|
||||
},
|
||||
transactionManager,
|
||||
);
|
||||
|
||||
return nonAssociatedMessages.map(({ id }) => id);
|
||||
},
|
||||
async (
|
||||
ids: string[],
|
||||
workspaceId: string,
|
||||
transactionManager?: WorkspaceEntityManager,
|
||||
) => {
|
||||
this.logger.log(
|
||||
`WorkspaceId: ${workspaceId} Deleting ${ids.length} messages from message cleaner`,
|
||||
);
|
||||
await messageRepository.delete(ids, transactionManager);
|
||||
},
|
||||
transactionManager,
|
||||
);
|
||||
|
||||
await deleteUsingPagination(
|
||||
workspaceId,
|
||||
500,
|
||||
async (
|
||||
limit: number,
|
||||
offset: number,
|
||||
_workspaceId: string,
|
||||
transactionManager?: WorkspaceEntityManager,
|
||||
) => {
|
||||
const orphanThreads = await messageThreadRepository.find(
|
||||
{
|
||||
where: {
|
||||
messages: {
|
||||
id: IsNull(),
|
||||
await deleteUsingPagination(
|
||||
workspaceId,
|
||||
500,
|
||||
async (
|
||||
limit: number,
|
||||
offset: number,
|
||||
_workspaceId: string,
|
||||
transactionManager?: WorkspaceEntityManager,
|
||||
) => {
|
||||
const orphanThreads = await messageThreadRepository.find(
|
||||
{
|
||||
where: {
|
||||
messages: {
|
||||
id: IsNull(),
|
||||
},
|
||||
},
|
||||
take: limit,
|
||||
skip: offset,
|
||||
},
|
||||
},
|
||||
take: limit,
|
||||
skip: offset,
|
||||
transactionManager,
|
||||
);
|
||||
|
||||
return orphanThreads.map(({ id }) => id);
|
||||
},
|
||||
async (
|
||||
ids: string[],
|
||||
_workspaceId: string,
|
||||
transactionManager?: WorkspaceEntityManager,
|
||||
) => {
|
||||
await messageThreadRepository.delete(ids, transactionManager);
|
||||
},
|
||||
transactionManager,
|
||||
);
|
||||
|
||||
return orphanThreads.map(({ id }) => id);
|
||||
},
|
||||
async (
|
||||
ids: string[],
|
||||
_workspaceId: string,
|
||||
transactionManager?: WorkspaceEntityManager,
|
||||
) => {
|
||||
await messageThreadRepository.delete(ids, transactionManager);
|
||||
},
|
||||
transactionManager,
|
||||
);
|
||||
},
|
||||
);
|
||||
|
||||
Reference in New Issue
Block a user