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

Improve messaging sync performances (#13549)

Browse Source 
This PR aims to improve performances as we are making A LOT of queries
after having added events emission to the ORM layer. While investigating
issues, I've come across multiple problems I will describe below

## Add index on workspace.activationStatus as we are querying it a lot

As per title

## Add logs on core datasource destroy

It seems that we have postgres connection pool destruction in
production. I cannot reproduce locally but I suspect the Query Timeouts
to be the root cause. I'm fixing most of the Query Timeouts cause in
this PR but I'm adding the logs so we have more information in case in
keeps happening in production.

## GraphQL query runner createMany

It was using a for loop on each record. This is as issue has we will
emit an event separately for each record => we should always try to
batch events.
Replacing by a save on all records. Note that this is not perfect as we
should avoid using save (bad performances), and use insert + updateMany
instead. As I have follow up discussions regarding permissions, I
haven't replaced it by insert and updateMany yet. Using save instead of
a for loop makes the problem less worrying.

## Introduce updateMany in ORM @Weiko @ijreilly FYI

.save(manyRecords) is bad as it's querying the data for no good reason
(and doing a select for each record...).
We already have .insert(), i'm introducing .updateMany()

I think our ORM layer should be simplified a lot but I'm not starting
the refacto yet

## Fixing ORM @Weiko @ijreilly  FYI
- Fixing events emission in delete function
- Fixing events emission in insert function
- make sure everything is batched

## Events performance @Weiko @ijreilly FYI

Do not emit timelineActivity db events as this does not seem useful and
is quite heavy

## Messaging and Calendar performance @bosiraphael FYI

Rework many functions to make sure they are batched. This does not touch
the driver layer and I have heavily tested it. (multiple account, with
multiple channels, common thread, common messages, etc...)

## Workflow Trigger relation Fetch performance @martmull FYI

Improved performance by batching
This commit is contained in:
Charles Bochet
2025-08-03 23:56:44 +02:00
committed by GitHub
parent 996a471a4e
commit dc3c194f3e
37 changed files with 1231 additions and 1766 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/twenty-server/src/engine/twenty-orm/entity-manager/workspace-entity-manager.ts
+55 -26
View File
@@ -134,7 +134,7 @@ export class WorkspaceEntityManager extends EntityManager {
shouldBypassPermissionChecks: false,
objectRecordsPermissions: {},
},
): SelectQueryBuilder<Entity> | WorkspaceSelectQueryBuilder<Entity> {
): WorkspaceSelectQueryBuilder<Entity> {
let queryBuilder: SelectQueryBuilder<Entity>;
if (alias) {
@@ -173,14 +173,15 @@ export class WorkspaceEntityManager extends EntityManager {
selectedColumns: string[] = [],
permissionOptions?: PermissionOptions,
): Promise<InsertResult> {
const metadata = this.connection.getMetadata(target);
return this.createQueryBuilder(
undefined,
undefined,
target,
metadata.name,
undefined,
permissionOptions,
)
.insert()
.into(target)
.values(entity)
.returning(selectedColumns)
.execute();
@@ -316,6 +317,29 @@ export class WorkspaceEntityManager extends EntityManager {
}
}
public updateMany<Entity extends ObjectLiteral>(
target: EntityTarget<Entity>,
inputs: {
criteria: string;
partialEntity: QueryDeepPartialEntity<Entity>;
}[],
permissionOptions?: PermissionOptions,
selectedColumns?: string[],
): Promise<UpdateResult> {
const metadata = this.connection.getMetadata(target);
return this.createQueryBuilder(
target,
metadata.name,
undefined,
permissionOptions,
)
.update()
.setManyInputs(inputs)
.returning(selectedColumns ?? [])
.execute();
}
override increment<Entity extends ObjectLiteral>(
target: EntityTarget<Entity>,
criteria: object,
@@ -1077,7 +1101,7 @@ export class WorkspaceEntityManager extends EntityManager {
entityTarget,
);
const updatedEntities = isDefined(relationNestedConfig)
const entityWithConnectedRelations = isDefined(relationNestedConfig)
? await relationNestedQueries.processRelationNestedQueries({
entities: entityArray,
relationNestedConfig,
@@ -1090,7 +1114,9 @@ export class WorkspaceEntityManager extends EntityManager {
})
: entityArray;
const entityIds = entityArray.map((e) => (e as { id: string }).id);
const entityIds = entityArray
.map((entity) => (entity as { id: string }).id)
.filter(isDefined);
const beforeUpdate = await this.find(
entityTarget,
{
@@ -1114,7 +1140,7 @@ export class WorkspaceEntityManager extends EntityManager {
);
const formattedEntityOrEntities = formatData(
updatedEntities,
entityWithConnectedRelations,
objectMetadataItem,
);
@@ -1150,26 +1176,29 @@ export class WorkspaceEntityManager extends EntityManager {
this.internalContext.objectMetadataMaps,
);
for (const entity of formattedResult) {
const isUpdate = beforeUpdateMapById[entity.id];
const updatedEntities = formattedResult.filter(
(entity) => beforeUpdateMapById[entity.id],
);
const createdEntities = formattedResult.filter(
(entity) => !beforeUpdateMapById[entity.id],
);
if (isUpdate) {
await this.internalContext.eventEmitterService.emitMutationEvent({
action: DatabaseEventAction.UPDATED,
objectMetadataItem,
workspaceId: this.internalContext.workspaceId,
entities: [entity],
beforeEntities: beforeUpdateMapById[entity.id],
});
} else {
await this.internalContext.eventEmitterService.emitMutationEvent({
action: DatabaseEventAction.CREATED,
objectMetadataItem,
workspaceId: this.internalContext.workspaceId,
entities: [entity],
});
}
}
await this.internalContext.eventEmitterService.emitMutationEvent({
action: DatabaseEventAction.UPDATED,
objectMetadataItem,
workspaceId: this.internalContext.workspaceId,
entities: updatedEntities,
beforeEntities: updatedEntities.map(
(entity) => beforeUpdateMapById[entity.id],
),
});
await this.internalContext.eventEmitterService.emitMutationEvent({
action: DatabaseEventAction.CREATED,
objectMetadataItem,
workspaceId: this.internalContext.workspaceId,
entities: createdEntities,
});
const isFieldPermissionsEnabled =
this.getFeatureFlagMap().IS_FIELDS_PERMISSIONS_ENABLED;
packages/twenty-server/src/engine/twenty-orm/factories/workspace-datasource.factory.ts
+4 -4
View File
@@ -63,15 +63,15 @@ export class WorkspaceDatasourceFactory {
);
if (isPoolSharingEnabled) {
this.logger.debug(
`PromiseMemoizer Event: A WorkspaceDataSource (using shared pool) is being cleared. Actual pool closure managed by PgPoolSharedService. Not calling dataSource.destroy().`,
this.logger.log(
`PromiseMemoizer Event: A WorkspaceDataSource for workspace ${dataSource.internalContext.workspaceId} is being cleared. Actual pool closure managed by PgPoolSharedService. Not calling dataSource.destroy().`,
);
// We should NOT call dataSource.destroy() here, because that would end
// the shared pool, potentially affecting other active users of that pool.
// The PgPoolSharedService is responsible for the lifecycle of shared pools.
} else {
this.logger.debug(
`PromiseMemoizer Event: A WorkspaceDataSource (using dedicated pool) is being cleared. Calling safelyDestroyDataSource.`,
this.logger.log(
`PromiseMemoizer Event: A WorkspaceDataSource for workspace ${dataSource.internalContext.workspaceId} is being cleared. Calling safelyDestroyDataSource.`,
);
await this.safelyDestroyDataSource(dataSource);
}
packages/twenty-server/src/engine/twenty-orm/pg-shared-pool/pg-shared-pool.module.ts
+3
View File
@@ -1,5 +1,6 @@
import {
Global,
Logger,
Module,
OnApplicationShutdown,
OnModuleInit,
@@ -19,6 +20,7 @@ import { PgPoolSharedService } from 'src/engine/twenty-orm/pg-shared-pool/pg-sha
})
export class PgPoolSharedModule implements OnModuleInit, OnApplicationShutdown {
constructor(private readonly pgPoolSharedService: PgPoolSharedService) {}
private readonly logger = new Logger(PgPoolSharedModule.name);
/**
* Initialize the pool sharing service when the module is initialized
@@ -31,6 +33,7 @@ export class PgPoolSharedModule implements OnModuleInit, OnApplicationShutdown {
* Clean up any resources when the application shuts down
*/
async onApplicationShutdown() {
this.logger.log('Shutting down PgPoolSharedModule');
await this.pgPoolSharedService.onApplicationShutdown();
}
}
packages/twenty-server/src/engine/twenty-orm/pg-shared-pool/pg-shared-pool.service.ts
+1
View File
@@ -140,6 +140,7 @@ export class PgPoolSharedService {
*/
async onApplicationShutdown(): Promise<void> {
this.stopStatsLogging();
this.logger.log('onApplicationShutdown called in PgPoolSharedService');
await this.closeAllPools();
}
packages/twenty-server/src/engine/twenty-orm/repository/workspace-delete-query-builder.ts
+22 -1
View File
@@ -78,6 +78,21 @@ export class WorkspaceDeleteQueryBuilder<
this.internalContext,
);
const eventSelectQueryBuilder = new WorkspaceSelectQueryBuilder(
this as unknown as WorkspaceSelectQueryBuilder<T>,
this.objectRecordsPermissions,
this.internalContext,
true,
this.authContext,
this.featureFlagMap,
);
eventSelectQueryBuilder.expressionMap.wheres = this.expressionMap.wheres;
eventSelectQueryBuilder.expressionMap.aliases = this.expressionMap.aliases;
eventSelectQueryBuilder.setParameters(this.getParameters());
const before = await eventSelectQueryBuilder.getOne();
const result = await super.execute();
const formattedResult = formatResult<T[]>(
@@ -86,11 +101,17 @@ export class WorkspaceDeleteQueryBuilder<
this.internalContext.objectMetadataMaps,
);
const formattedBefore = formatResult<T[]>(
before,
objectMetadata,
this.internalContext.objectMetadataMaps,
);
await this.internalContext.eventEmitterService.emitMutationEvent({
action: DatabaseEventAction.DESTROYED,
objectMetadataItem: objectMetadata,
workspaceId: this.internalContext.workspaceId,
entities: formattedResult,
entities: formattedBefore,
authContext: this.authContext,
});
packages/twenty-server/src/engine/twenty-orm/repository/workspace-insert-query-builder.ts
+17 -2
View File
@@ -137,9 +137,24 @@ export class WorkspaceInsertQueryBuilder<
}
const result = await super.execute();
const eventSelectQueryBuilder = new WorkspaceSelectQueryBuilder(
this as unknown as WorkspaceSelectQueryBuilder<T>,
this.objectRecordsPermissions,
this.internalContext,
true,
this.authContext,
this.featureFlagMap,
);
eventSelectQueryBuilder.expressionMap.aliases = this.expressionMap.aliases;
eventSelectQueryBuilder.whereInIds(
result.identifiers.map((identifier) => identifier.id),
);
const afterResult = await eventSelectQueryBuilder.getMany();
const formattedResult = formatResult<T[]>(
result.raw,
afterResult,
objectMetadata,
this.internalContext.objectMetadataMaps,
);
@@ -153,7 +168,7 @@ export class WorkspaceInsertQueryBuilder<
});
return {
raw: result.raw,
raw: afterResult,
generatedMaps: formattedResult,
identifiers: result.identifiers,
};
packages/twenty-server/src/engine/twenty-orm/repository/workspace-update-query-builder.ts
+133
View File
@@ -42,6 +42,10 @@ export class WorkspaceUpdateQueryBuilder<
private relationNestedConfig:
| [RelationConnectQueryConfig[], RelationDisconnectQueryFieldsByEntityIndex]
| null;
private manyInputs: {
criteria: string;
partialEntity: QueryDeepPartialEntity<T>;
}[];
constructor(
queryBuilder: UpdateQueryBuilder<T>,
@@ -76,6 +80,10 @@ export class WorkspaceUpdateQueryBuilder<
}
override async execute(): Promise<UpdateResult> {
if (this.manyInputs) {
return this.executeMany();
}
validateQueryIsPermittedOrThrow({
expressionMap: this.expressionMap,
objectRecordsPermissions: this.objectRecordsPermissions,
@@ -166,6 +174,120 @@ export class WorkspaceUpdateQueryBuilder<
};
}
public async executeMany(): Promise<UpdateResult> {
for (const input of this.manyInputs) {
const fakeExpressionMapToValidatePermissions = Object.assign(
{},
this.expressionMap,
{
wheres: input.criteria,
valuesSet: input.partialEntity,
},
);
validateQueryIsPermittedOrThrow({
expressionMap: fakeExpressionMapToValidatePermissions,
objectRecordsPermissions: this.objectRecordsPermissions,
objectMetadataMaps: this.internalContext.objectMetadataMaps,
shouldBypassPermissionChecks: this.shouldBypassPermissionChecks,
isFieldPermissionsEnabled:
this.featureFlagMap?.[FeatureFlagKey.IS_FIELDS_PERMISSIONS_ENABLED],
});
}
const mainAliasTarget = this.getMainAliasTarget();
const objectMetadata = getObjectMetadataFromEntityTarget(
mainAliasTarget,
this.internalContext,
);
const eventSelectQueryBuilder = new WorkspaceSelectQueryBuilder(
this as unknown as WorkspaceSelectQueryBuilder<T>,
this.objectRecordsPermissions,
this.internalContext,
true,
this.authContext,
this.featureFlagMap,
);
eventSelectQueryBuilder.whereInIds(
this.manyInputs.map((input) => input.criteria),
);
eventSelectQueryBuilder.expressionMap.aliases = this.expressionMap.aliases;
eventSelectQueryBuilder.setParameters(this.getParameters());
const beforeRecords = await eventSelectQueryBuilder.getMany();
const formattedBefore = formatResult<T[]>(
beforeRecords,
objectMetadata,
this.internalContext.objectMetadataMaps,
);
const results: UpdateResult[] = [];
for (const input of this.manyInputs) {
this.expressionMap.valuesSet = input.partialEntity;
this.where({ id: input.criteria });
const nestedRelationQueryBuilder = new WorkspaceSelectQueryBuilder(
this as unknown as WorkspaceSelectQueryBuilder<T>,
this.objectRecordsPermissions,
this.internalContext,
this.shouldBypassPermissionChecks,
this.authContext,
);
if (isDefined(this.relationNestedConfig)) {
const updatedValues =
await this.relationNestedQueries.processRelationNestedQueries({
entities: input.partialEntity as
| QueryDeepPartialEntityWithNestedRelationFields<T>
| QueryDeepPartialEntityWithNestedRelationFields<T>[],
relationNestedConfig: this.relationNestedConfig,
queryBuilder: nestedRelationQueryBuilder,
});
this.expressionMap.valuesSet =
updatedValues.length === 1 ? updatedValues[0] : updatedValues;
}
const result = await super.execute();
results.push(result);
}
const afterRecords = await eventSelectQueryBuilder.getMany();
const formattedAfter = formatResult<T[]>(
afterRecords,
objectMetadata,
this.internalContext.objectMetadataMaps,
);
await this.internalContext.eventEmitterService.emitMutationEvent({
action: DatabaseEventAction.UPDATED,
objectMetadataItem: objectMetadata,
workspaceId: this.internalContext.workspaceId,
entities: formattedAfter,
beforeEntities: formattedBefore,
authContext: this.authContext,
});
const formattedResults = formatResult<T[]>(
results.map((result) => result.raw),
objectMetadata,
this.internalContext.objectMetadataMaps,
);
return {
raw: results.map((result) => result.raw),
generatedMaps: formattedResults,
affected: results.length,
};
}
override set(
_values:
| QueryDeepPartialEntityWithNestedRelationFields<T>
@@ -243,4 +365,15 @@ export class WorkspaceUpdateQueryBuilder<
return mainAliasTarget;
}
public setManyInputs(
inputs: {
criteria: string;
partialEntity: QueryDeepPartialEntity<T>;
}[],
): this {
this.manyInputs = inputs;
return this;
}
}
packages/twenty-server/src/engine/twenty-orm/repository/workspace.repository.ts
+26
View File
@@ -595,6 +595,32 @@ export class WorkspaceRepository<
);
}
// Experimental method to allow batch update and batch event emission
async updateMany(
inputs: {
criteria: string;
partialEntity: QueryDeepPartialEntity<T>;
}[],
entityManager?: WorkspaceEntityManager,
selectedColumns?: string[],
): Promise<UpdateResult> {
const manager = entityManager || this.manager;
const permissionOptions = {
shouldBypassPermissionChecks: this.shouldBypassPermissionChecks,
objectRecordsPermissions: this.objectRecordsPermissions,
};
const results = await manager.updateMany(
this.target,
inputs,
permissionOptions,
selectedColumns,
);
return results;
}
override async upsert(
entityOrEntities: QueryDeepPartialEntity<T> | QueryDeepPartialEntity<T>[],
conflictPathsOrOptions: string[] | UpsertOptions<T>,
packages/twenty-server/src/engine/twenty-orm/utils/format-result.util.ts
+1 -2
View File
@@ -13,14 +13,13 @@ import { getCompositeFieldMetadataCollection } from 'src/engine/twenty-orm/utils
import { isFieldMetadataEntityOfType } from 'src/engine/utils/is-field-metadata-of-type.util';
import { isDate } from 'src/utils/date/isDate';
import { isValidDate } from 'src/utils/date/isValidDate';
export function formatResult<T>(
// eslint-disable-next-line @typescript-eslint/no-explicit-any
data: any,
objectMetadataItemWithFieldMaps: ObjectMetadataItemWithFieldMaps | undefined,
objectMetadataMaps: ObjectMetadataMaps,
): T {
if (!data) {
if (!isDefined(data)) {
return data;
}