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

1658 post mortem 0710 send batch events in webhook (#15022)

Browse Source 
Use concurrent calls

---------

Co-authored-by: Charles Bochet <charles@twenty.com>
This commit is contained in:
martmull
2025-10-11 12:48:40 +02:00
committed by GitHub
parent 68c86871dd
commit c681fb7fb6
58 changed files with 920 additions and 521 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/twenty-server/src/engine/core-modules/audit/jobs/create-audit-log-from-internal-event.ts
+5 -5
View File
@@ -7,7 +7,7 @@ import { type ObjectRecordEvent } from 'src/engine/core-modules/event-emitter/ty
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 { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type';
import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event-batch.type';
@Processor(MessageQueue.entityEventsToDbQueue)
export class CreateAuditLogFromInternalEvent {
@@ -37,25 +37,25 @@ export class CreateAuditLogFromInternalEvent {
auditService.createObjectEvent(OBJECT_RECORD_UPDATED_EVENT, {
...eventProperties,
recordId: eventData.recordId,
objectMetadataId: eventData.objectMetadata.id,
objectMetadataId: workspaceEventBatch.objectMetadata.id,
});
} else if (workspaceEventBatch.name.endsWith('.created')) {
auditService.createObjectEvent(OBJECT_RECORD_CREATED_EVENT, {
...eventProperties,
recordId: eventData.recordId,
objectMetadataId: eventData.objectMetadata.id,
objectMetadataId: workspaceEventBatch.objectMetadata.id,
});
} else if (workspaceEventBatch.name.endsWith('.deleted')) {
auditService.createObjectEvent(OBJECT_RECORD_DELETED_EVENT, {
...eventProperties,
recordId: eventData.recordId,
objectMetadataId: eventData.objectMetadata.id,
objectMetadataId: workspaceEventBatch.objectMetadata.id,
});
} else if (workspaceEventBatch.name.endsWith('.upserted')) {
auditService.createObjectEvent(OBJECT_RECORD_UPSERTED_EVENT, {
...eventProperties,
recordId: eventData.recordId,
objectMetadataId: eventData.objectMetadata.id,
objectMetadataId: workspaceEventBatch.objectMetadata.id,
});
}
}
packages/twenty-server/src/engine/core-modules/billing/listeners/billing-feature-used.listener.ts
+8 -2
View File
@@ -2,12 +2,14 @@
import { Injectable } from '@nestjs/common';
import { isDefined } from 'twenty-shared/utils';
import { OnCustomBatchEvent } from 'src/engine/api/graphql/graphql-query-runner/decorators/on-custom-batch-event.decorator';
import { BILLING_FEATURE_USED } from 'src/engine/core-modules/billing/constants/billing-feature-used.constant';
import { BillingUsageService } from 'src/engine/core-modules/billing/services/billing-usage.service';
import { type BillingUsageEvent } from 'src/engine/core-modules/billing/types/billing-usage-event.type';
import { TwentyConfigService } from 'src/engine/core-modules/twenty-config/twenty-config.service';
import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type';
import { CustomWorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/custom-workspace-batch-event.type';
@Injectable()
export class BillingFeatureUsedListener {
@@ -18,8 +20,12 @@ export class BillingFeatureUsedListener {
@OnCustomBatchEvent(BILLING_FEATURE_USED)
async handleBillingFeatureUsedEvent(
payload: WorkspaceEventBatch<BillingUsageEvent>,
payload: CustomWorkspaceEventBatch<BillingUsageEvent>,
) {
if (!isDefined(payload.workspaceId)) {
return;
}
if (!this.twentyConfigService.get('IS_BILLING_ENABLED')) {
return;
}
packages/twenty-server/src/engine/core-modules/billing/listeners/billing-workspace-member.listener.ts
+1 -1
View File
@@ -13,7 +13,7 @@ import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decora
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 { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type';
import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event-batch.type';
import { type WorkspaceMemberWorkspaceEntity } from 'src/modules/workspace-member/standard-objects/workspace-member.workspace-entity';
@Injectable()
packages/twenty-server/src/engine/core-modules/event-emitter/types/object-record.base.event.ts
-2
View File
@@ -1,5 +1,4 @@
import { type ObjectRecordDiff } from 'src/engine/core-modules/event-emitter/types/object-record-diff';
import { type ObjectMetadataEntity } from 'src/engine/metadata-modules/object-metadata/object-metadata.entity';
type Properties<T> = {
updatedFields?: string[];
@@ -12,6 +11,5 @@ export class ObjectRecordBaseEvent<T = object> {
recordId: string;
userId?: string;
workspaceMemberId?: string;
objectMetadata: Omit<ObjectMetadataEntity, 'indexMetadatas'>;
properties: Properties<T>;
}
packages/twenty-server/src/engine/core-modules/file/listeners/file-attachment.listener.ts
+1 -1
View File
@@ -10,7 +10,7 @@ import {
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 { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type';
import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event-batch.type';
import { type AttachmentWorkspaceEntity } from 'src/modules/attachment/standard-objects/attachment.workspace-entity';
@Injectable()
packages/twenty-server/src/engine/core-modules/file/listeners/file-workspace-member.listener.ts
+1 -1
View File
@@ -10,7 +10,7 @@ import {
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 { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type';
import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event-batch.type';
import { type WorkspaceMemberWorkspaceEntity } from 'src/modules/workspace-member/standard-objects/workspace-member.workspace-entity';
@Injectable()
packages/twenty-server/src/engine/core-modules/webhook/jobs/call-webhook-jobs.job.ts
+18 -49
View File
@@ -1,6 +1,6 @@
import { Logger } from '@nestjs/common';
import { isDefined } from 'twenty-shared/utils';
import chunk from 'lodash.chunk';
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';
@@ -11,11 +11,12 @@ import {
CallWebhookJob,
type CallWebhookJobData,
} from 'src/engine/core-modules/webhook/jobs/call-webhook.job';
import { type ObjectRecordEventForWebhook } from 'src/engine/core-modules/webhook/types/object-record-event-for-webhook.type';
import { WebhookService } from 'src/engine/core-modules/webhook/webhook.service';
import { type ObjectMetadataEntity } from 'src/engine/metadata-modules/object-metadata/object-metadata.entity';
import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type';
import { removeSecretFromWebhookRecord } from 'src/utils/remove-secret-from-webhook-record';
import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event-batch.type';
import type { ObjectRecordEvent } from 'src/engine/core-modules/event-emitter/types/object-record-event.event';
import { transformEventBatchToWebhookEvents } from 'src/engine/core-modules/webhook/utils/transform-event-batch-to-webhook-events';
const WEBHOOK_JOBS_CHUNK_SIZE = 20;
@Processor(MessageQueue.webhookQueue)
export class CallWebhookJobsJob {
@@ -28,7 +29,7 @@ export class CallWebhookJobsJob {
@Process(CallWebhookJobsJob.name)
async handle(
workspaceEventBatch: WorkspaceEventBatch<ObjectRecordEventForWebhook>,
workspaceEventBatch: WorkspaceEventBatch<ObjectRecordEvent>,
): Promise<void> {
// If you change that function, double check it does not break Zapier
// trigger in packages/twenty-zapier/src/triggers/trigger_record.ts
@@ -47,51 +48,19 @@ export class CallWebhookJobsJob {
],
);
for (const eventData of workspaceEventBatch.events) {
const eventName = workspaceEventBatch.name;
const objectMetadata: Pick<ObjectMetadataEntity, 'id' | 'nameSingular'> =
{
id: eventData.objectMetadata.id,
nameSingular: eventData.objectMetadata.nameSingular,
};
const workspaceId = workspaceEventBatch.workspaceId;
const record =
'after' in eventData.properties && isDefined(eventData.properties.after)
? eventData.properties.after
: 'before' in eventData.properties &&
isDefined(eventData.properties.before)
? eventData.properties.before
: {};
const updatedFields =
'updatedFields' in eventData.properties
? eventData.properties.updatedFields
: undefined;
const webhookEvents = transformEventBatchToWebhookEvents({
workspaceEventBatch,
webhooks,
});
const isWebhookEvent = nameSingular === 'webhook';
const sanitizedRecord = removeSecretFromWebhookRecord(
record,
isWebhookEvent,
const webhookEventsChunks = chunk(webhookEvents, WEBHOOK_JOBS_CHUNK_SIZE);
for (const webhookEventsChunk of webhookEventsChunks) {
await this.messageQueueService.add<CallWebhookJobData[]>(
CallWebhookJob.name,
webhookEventsChunk,
{ retryLimit: 3 },
);
webhooks.forEach((webhook) => {
const webhookData = {
targetUrl: webhook.targetUrl,
secret: webhook.secret,
eventName,
objectMetadata,
workspaceId,
webhookId: webhook.id,
eventDate: new Date(),
record: sanitizedRecord,
...(updatedFields && { updatedFields }),
};
this.messageQueueService.add<CallWebhookJobData>(
CallWebhookJob.name,
webhookData,
{ retryLimit: 3 },
);
});
}
}
}
packages/twenty-server/src/engine/core-modules/webhook/jobs/call-webhook.job.ts
+9 -1
View File
@@ -42,7 +42,15 @@ export class CallWebhookJob {
}
@Process(CallWebhookJob.name)
async handle(data: CallWebhookJobData): Promise<void> {
async handle(webhookJobEvents: CallWebhookJobData[]): Promise<void> {
await Promise.all(
webhookJobEvents.map(
async (webhookJobEvent) => await this.callWebhook(webhookJobEvent),
),
);
}
private async callWebhook(data: CallWebhookJobData): Promise<void> {
const commonPayload = {
url: data.targetUrl,
webhookId: data.webhookId,
packages/twenty-server/src/engine/core-modules/webhook/types/object-record-event-for-webhook.type.ts
-9
View File
@@ -1,9 +0,0 @@
import { type ObjectRecordEvent } from 'src/engine/core-modules/event-emitter/types/object-record-event.event';
import { type ObjectMetadataEntity } from 'src/engine/metadata-modules/object-metadata/object-metadata.entity';
export type ObjectRecordEventForWebhook = Omit<
ObjectRecordEvent,
'objectMetadata'
> & {
objectMetadata: Pick<ObjectMetadataEntity, 'id' | 'nameSingular'>;
};
packages/twenty-server/src/engine/core-modules/webhook/types/workspace-event-batch-for-webhook.type.ts
+9
View File
@@ -0,0 +1,9 @@
import { type ObjectMetadataEntity } from 'src/engine/metadata-modules/object-metadata/object-metadata.entity';
import { type WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event-batch.type';
export type WorkspaceEventBatchForWebhook<WorkspaceEvent> = Omit<
WorkspaceEventBatch<WorkspaceEvent>,
'objectMetadata'
> & {
objectMetadata: Pick<ObjectMetadataEntity, 'id' | 'nameSingular'>;
};
packages/twenty-server/src/engine/core-modules/webhook/utils/__tests__/transform-event-batch-to-webhook-events.spec.ts
+235
View File
@@ -0,0 +1,235 @@
import type { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event-batch.type';
import type { ObjectRecordEvent } from 'src/engine/core-modules/event-emitter/types/object-record-event.event';
import type { Webhook } from 'src/engine/core-modules/webhook/webhook.entity';
import { transformEventBatchToWebhookEvents } from 'src/engine/core-modules/webhook/utils/transform-event-batch-to-webhook-events';
import { getMockObjectMetadataEntity } from 'src/utils/__test__/get-object-metadata-entity.mock';
const mockObjectMetadata = getMockObjectMetadataEntity({
id: 'id',
nameSingular: 'nameSingular',
namePlural: 'namePlural',
workspaceId: 'workspaceId',
});
describe('transformEventBatchToWebhookEvents', () => {
it('should transform properly', () => {
const workspaceEventBatch = {
workspaceId: 'workspaceId',
objectMetadata: mockObjectMetadata,
name: 'objectNameSingular.created',
events: [
{
recordId: 'recordId-1',
properties: {
after: {
id: 'id-1',
nameSingular: 'nameSingular-1',
},
},
},
{
recordId: 'recordId-2',
properties: {
before: {
id: 'id-2',
nameSingular: 'nameSingular-2',
},
},
},
{
recordId: 'recordId-3',
properties: {
after: {
id: 'id-3',
nameSingular: 'nameSingular-3',
secret: 'secret-3',
},
updatedFields: ['nameSingular'],
},
},
],
} as WorkspaceEventBatch<ObjectRecordEvent>;
const webhooks = [
{
id: 'webhook-id',
targetUrl: 'targetUrl',
secret: 'secret',
},
{
id: 'webhook-id-2',
targetUrl: 'targetUrl-2',
secret: 'secret-2',
},
] as Webhook[];
const result = transformEventBatchToWebhookEvents({
workspaceEventBatch,
webhooks,
});
const expectedResultWithoutEventDate = [
{
targetUrl: 'targetUrl',
eventName: 'objectNameSingular.created',
objectMetadata: {
id: mockObjectMetadata.id,
nameSingular: mockObjectMetadata.nameSingular,
},
workspaceId: 'workspaceId',
webhookId: 'webhook-id',
record: {
id: 'id-1',
nameSingular: 'nameSingular-1',
},
secret: 'secret',
},
{
targetUrl: 'targetUrl',
eventName: 'objectNameSingular.created',
objectMetadata: {
id: mockObjectMetadata.id,
nameSingular: mockObjectMetadata.nameSingular,
},
workspaceId: 'workspaceId',
webhookId: 'webhook-id',
record: {
id: 'id-2',
nameSingular: 'nameSingular-2',
},
secret: 'secret',
},
{
targetUrl: 'targetUrl',
eventName: 'objectNameSingular.created',
objectMetadata: {
id: mockObjectMetadata.id,
nameSingular: mockObjectMetadata.nameSingular,
},
workspaceId: 'workspaceId',
webhookId: 'webhook-id',
record: {
id: 'id-3',
nameSingular: 'nameSingular-3',
secret: 'secret-3',
},
updatedFields: ['nameSingular'],
secret: 'secret',
},
{
targetUrl: 'targetUrl-2',
eventName: 'objectNameSingular.created',
objectMetadata: {
id: mockObjectMetadata.id,
nameSingular: mockObjectMetadata.nameSingular,
},
workspaceId: 'workspaceId',
webhookId: 'webhook-id-2',
record: {
id: 'id-1',
nameSingular: 'nameSingular-1',
},
secret: 'secret-2',
},
{
targetUrl: 'targetUrl-2',
eventName: 'objectNameSingular.created',
objectMetadata: {
id: mockObjectMetadata.id,
nameSingular: mockObjectMetadata.nameSingular,
},
workspaceId: 'workspaceId',
webhookId: 'webhook-id-2',
record: {
id: 'id-2',
nameSingular: 'nameSingular-2',
},
secret: 'secret-2',
},
{
targetUrl: 'targetUrl-2',
eventName: 'objectNameSingular.created',
objectMetadata: {
id: mockObjectMetadata.id,
nameSingular: mockObjectMetadata.nameSingular,
},
workspaceId: 'workspaceId',
webhookId: 'webhook-id-2',
record: {
id: 'id-3',
nameSingular: 'nameSingular-3',
secret: 'secret-3',
},
updatedFields: ['nameSingular'],
secret: 'secret-2',
},
];
const resultWithoutEventDate = result.map((event) => {
const { eventDate: _, ...eventWithoutEventDate } = event;
return eventWithoutEventDate;
});
expect(resultWithoutEventDate).toEqual(expectedResultWithoutEventDate);
});
it('should sanitize records properly', () => {
const workspaceEventBatch = {
workspaceId: 'workspaceId',
objectMetadata: mockObjectMetadata,
name: 'webhook.created',
events: [
{
recordId: 'recordId-1',
properties: {
after: {
id: 'id-1',
targetUrl: 'targetUrl-1',
secret: 'secret-1',
},
},
},
],
} as WorkspaceEventBatch<ObjectRecordEvent>;
const webhooks = [
{
id: 'webhook-id',
targetUrl: 'targetUrl',
secret: 'secret',
},
] as Webhook[];
const result = transformEventBatchToWebhookEvents({
workspaceEventBatch,
webhooks,
});
const expectedResultWithoutEventDate = [
{
targetUrl: 'targetUrl',
eventName: 'webhook.created',
objectMetadata: {
id: mockObjectMetadata.id,
nameSingular: mockObjectMetadata.nameSingular,
},
workspaceId: 'workspaceId',
webhookId: 'webhook-id',
record: {
id: 'id-1',
targetUrl: 'targetUrl-1',
// No secret
},
secret: 'secret',
},
];
const resultWithoutEventDate = result.map((event) => {
const { eventDate: _, ...eventWithoutEventDate } = event;
return eventWithoutEventDate;
});
expect(resultWithoutEventDate).toEqual(expectedResultWithoutEventDate);
});
});
packages/twenty-server/src/engine/core-modules/webhook/utils/__tests__/transform-event-to-webhook-event.spec.ts
+34
View File
@@ -0,0 +1,34 @@
import { transformEventToWebhookEvent } from 'src/engine/core-modules/webhook/utils/transform-event-to-webhook-event';
import type { ObjectRecordEvent } from 'src/engine/core-modules/event-emitter/types/object-record-event.event';
describe('transformEventToWebhookEvent', () => {
it('should transform event to webhook event', () => {
const record = {
recordId: 'recordId',
properties: {
after: {
id: 'id',
nameSingular: 'nameSingular',
secret: 'secret',
},
updatedFields: ['nameSingular'],
},
} as ObjectRecordEvent;
const expectedResult = {
record: {
id: 'id',
nameSingular: 'nameSingular',
secret: 'secret',
},
updatedFields: ['nameSingular'],
};
expect(
transformEventToWebhookEvent({
eventName: 'nameSingular.created',
event: record,
}),
).toEqual(expectedResult);
});
});
packages/twenty-server/src/engine/core-modules/webhook/utils/transform-event-batch-to-webhook-events.ts
+49
View File
@@ -0,0 +1,49 @@
import { type WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event-batch.type';
import { type CallWebhookJobData } from 'src/engine/core-modules/webhook/jobs/call-webhook.job';
import { type Webhook } from 'src/engine/core-modules/webhook/webhook.entity';
import type { ObjectRecordEvent } from 'src/engine/core-modules/event-emitter/types/object-record-event.event';
import { transformEventToWebhookEvent } from 'src/engine/core-modules/webhook/utils/transform-event-to-webhook-event';
export const transformEventBatchToWebhookEvents = ({
workspaceEventBatch,
webhooks,
}: {
workspaceEventBatch: WorkspaceEventBatch<ObjectRecordEvent>;
webhooks: Webhook[];
}): CallWebhookJobData[] => {
const result: CallWebhookJobData[] = [];
for (const webhook of webhooks) {
const targetUrl = webhook.targetUrl;
const eventName = workspaceEventBatch.name;
const objectMetadataForWebhook = {
id: workspaceEventBatch.objectMetadata.id,
nameSingular: workspaceEventBatch.objectMetadata.nameSingular,
};
const workspaceId = workspaceEventBatch.workspaceId;
const webhookId = webhook.id;
const eventDate = new Date();
const secret = webhook.secret;
for (const eventData of workspaceEventBatch.events) {
const { record, updatedFields } = transformEventToWebhookEvent({
eventName: workspaceEventBatch.name,
event: eventData,
});
result.push({
targetUrl,
eventName,
objectMetadata: objectMetadataForWebhook,
workspaceId,
webhookId,
eventDate,
record,
...(updatedFields && { updatedFields }),
secret,
});
}
}
return result;
};
packages/twenty-server/src/engine/core-modules/webhook/utils/transform-event-to-webhook-event.ts
+34
View File
@@ -0,0 +1,34 @@
import { isDefined } from 'twenty-shared/utils';
import type { ObjectRecordEvent } from 'src/engine/core-modules/event-emitter/types/object-record-event.event';
import { removeSecretFromWebhookRecord } from 'src/utils/remove-secret-from-webhook-record';
export const transformEventToWebhookEvent = ({
eventName,
event,
}: {
eventName: string;
event: ObjectRecordEvent;
}) => {
const [nameSingular, _] = eventName.split('.');
const record =
'after' in event.properties && isDefined(event.properties.after)
? event.properties.after
: 'before' in event.properties && isDefined(event.properties.before)
? event.properties.before
: {};
const updatedFields =
'updatedFields' in event.properties
? event.properties.updatedFields
: undefined;
const isWebhookEvent = nameSingular === 'webhook';
const sanitizedRecord = removeSecretFromWebhookRecord(record, isWebhookEvent);
return {
record: sanitizedRecord,
...(updatedFields && { updatedFields }),
};
};
packages/twenty-server/src/engine/core-modules/workspace/workspace-workspace-member.listener.ts
+1 -1
View File
@@ -10,7 +10,7 @@ import { type ObjectRecordUpdateEvent } from 'src/engine/core-modules/event-emit
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 { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type';
import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event-batch.type';
import { type WorkspaceMemberWorkspaceEntity } from 'src/modules/workspace-member/standard-objects/workspace-member.workspace-entity';
import { OnDatabaseBatchEvent } from 'src/engine/api/graphql/graphql-query-runner/decorators/on-database-batch-event.decorator';
import { DatabaseEventAction } from 'src/engine/api/graphql/graphql-query-runner/enums/database-event-action';