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

bugfix: escape destroyed objects on workers (#9719)

Browse Source 
# This PR

- Fixes #9358 

@FelixMalfait please check this workaround

---------

Co-authored-by: Félix Malfait <felix@twenty.com>
This commit is contained in:
P A C - 先生
2025-01-23 17:29:54 +02:00
committed by GitHub
parent bbb0c9a761
commit bbd3af108b
11 changed files with 134 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files

77
packages/twenty-server/src/engine/api/graphql/workspace-query-runner/listeners/entity-events-to-db.listener.ts
View File

@ -1,17 +1,21 @@
import { Injectable } from '@nestjs/common';
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';
import { ObjectRecordCreateEvent } from 'src/engine/core-modules/event-emitter/types/object-record-create.event';
import { ObjectRecordDeleteEvent } from 'src/engine/core-modules/event-emitter/types/object-record-delete.event';
import { ObjectRecordDestroyEvent } from 'src/engine/core-modules/event-emitter/types/object-record-destroy.event';
import { ObjectRecordEvent } from 'src/engine/core-modules/event-emitter/types/object-record-event.event';
import { ObjectRecordNonDestructiveEvent } from 'src/engine/core-modules/event-emitter/types/object-record-non-destructive-event';
import { ObjectRecordRestoreEvent } from 'src/engine/core-modules/event-emitter/types/object-record-restore.event';
import { ObjectRecordUpdateEvent } from 'src/engine/core-modules/event-emitter/types/object-record-update.event';
import { ObjectRecordBaseEvent } from 'src/engine/core-modules/event-emitter/types/object-record.base.event';
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 { CreateAuditLogFromInternalEvent } from 'src/modules/timeline/jobs/create-audit-log-from-internal-event';
import { UpsertTimelineActivityFromInternalEvent } from 'src/modules/timeline/jobs/upsert-timeline-activity-from-internal-event.job';
import { OnDatabaseBatchEvent } from 'src/engine/api/graphql/graphql-query-runner/decorators/on-database-batch-event.decorator';
import { CallWebhookJobsJob } from 'src/modules/webhook/jobs/call-webhook-jobs.job';
import { DatabaseEventAction } from 'src/engine/api/graphql/graphql-query-runner/enums/database-event-action';
@Injectable()
export class EntityEventsToDbListener {
@ -24,47 +28,66 @@ export class EntityEventsToDbListener {
@OnDatabaseBatchEvent('*', DatabaseEventAction.CREATED)
async handleCreate(batchEvent: WorkspaceEventBatch<ObjectRecordCreateEvent>) {
return this.handle(batchEvent);
return this.handleEvent(batchEvent, DatabaseEventAction.CREATED);
}
@OnDatabaseBatchEvent('*', DatabaseEventAction.UPDATED)
async handleUpdate(batchEvent: WorkspaceEventBatch<ObjectRecordUpdateEvent>) {
return this.handle(batchEvent);
return this.handleEvent(batchEvent, DatabaseEventAction.UPDATED);
}
@OnDatabaseBatchEvent('*', DatabaseEventAction.DELETED)
async handleDelete(batchEvent: WorkspaceEventBatch<ObjectRecordUpdateEvent>) {
return this.handle(batchEvent);
async handleDelete(batchEvent: WorkspaceEventBatch<ObjectRecordDeleteEvent>) {
return this.handleEvent(batchEvent, DatabaseEventAction.DELETED);
}
@OnDatabaseBatchEvent('*', DatabaseEventAction.RESTORED)
async handleRestore(
batchEvent: WorkspaceEventBatch<ObjectRecordRestoreEvent>,
) {
return this.handleEvent(batchEvent, DatabaseEventAction.RESTORED);
}
@OnDatabaseBatchEvent('*', DatabaseEventAction.DESTROYED)
async handleDestroy(
batchEvent: WorkspaceEventBatch<ObjectRecordUpdateEvent>,
batchEvent: WorkspaceEventBatch<ObjectRecordDestroyEvent>,
) {
return this.handle(batchEvent);
return this.handleEvent(batchEvent, DatabaseEventAction.DESTROYED);
}
private async handle(batchEvent: WorkspaceEventBatch<ObjectRecordBaseEvent>) {
private async handleEvent<T extends ObjectRecordEvent>(
batchEvent: WorkspaceEventBatch<T>,
action: DatabaseEventAction,
) {
const filteredEvents = batchEvent.events.filter(
(event) => event.objectMetadata?.isAuditLogged,
);
await this.entityEventsToDbQueueService.add<
WorkspaceEventBatch<ObjectRecordBaseEvent>
>(CreateAuditLogFromInternalEvent.name, {
...batchEvent,
events: filteredEvents,
});
await this.entityEventsToDbQueueService.add<
WorkspaceEventBatch<ObjectRecordBaseEvent>
>(UpsertTimelineActivityFromInternalEvent.name, {
...batchEvent,
events: filteredEvents,
});
await this.webhookQueueService.add<
WorkspaceEventBatch<ObjectRecordBaseEvent>
>(CallWebhookJobsJob.name, batchEvent, { retryLimit: 3 });
await Promise.all([
this.webhookQueueService.add<WorkspaceEventBatch<T>>(
CallWebhookJobsJob.name,
batchEvent,
{
retryLimit: 3,
},
),
this.entityEventsToDbQueueService.add<WorkspaceEventBatch<T>>(
CreateAuditLogFromInternalEvent.name,
{
...batchEvent,
events: filteredEvents,
},
),
...(action !== DatabaseEventAction.DESTROYED
? [
this.entityEventsToDbQueueService.add<
WorkspaceEventBatch<ObjectRecordNonDestructiveEvent>
>(UpsertTimelineActivityFromInternalEvent.name, {
...batchEvent,
events: filteredEvents,
}),
]
: []),
]);
}
}

2
packages/twenty-server/src/engine/core-modules/auth/auth.module.ts
View File

@ -17,6 +17,7 @@ import { MicrosoftAPIsService } from 'src/engine/core-modules/auth/services/micr
// import { OAuthService } from 'src/engine/core-modules/auth/services/oauth.service';
import { ResetPasswordService } from 'src/engine/core-modules/auth/services/reset-password.service';
import { SignInUpService } from 'src/engine/core-modules/auth/services/sign-in-up.service';
import { SocialSsoService } from 'src/engine/core-modules/auth/services/social-sso.service';
import { SamlAuthStrategy } from 'src/engine/core-modules/auth/strategies/saml.auth.strategy';
import { AccessTokenService } from 'src/engine/core-modules/auth/token/services/access-token.service';
import { LoginTokenService } from 'src/engine/core-modules/auth/token/services/login-token.service';
@ -43,7 +44,6 @@ import { DataSourceModule } from 'src/engine/metadata-modules/data-source/data-s
import { WorkspaceDataSourceModule } from 'src/engine/workspace-datasource/workspace-datasource.module';
import { WorkspaceManagerModule } from 'src/engine/workspace-manager/workspace-manager.module';
import { ConnectedAccountModule } from 'src/modules/connected-account/connected-account.module';
import { SocialSsoService } from 'src/engine/core-modules/auth/services/social-sso.service';
import { AuthResolver } from './auth.resolver';

12
packages/twenty-server/src/engine/core-modules/event-emitter/types/object-record-event.event.ts Normal file
View File

@ -0,0 +1,12 @@
import { ObjectRecordCreateEvent } from 'src/engine/core-modules/event-emitter/types/object-record-create.event';
import { ObjectRecordDeleteEvent } from 'src/engine/core-modules/event-emitter/types/object-record-delete.event';
import { ObjectRecordDestroyEvent } from 'src/engine/core-modules/event-emitter/types/object-record-destroy.event';
import { ObjectRecordRestoreEvent } from 'src/engine/core-modules/event-emitter/types/object-record-restore.event';
import { ObjectRecordUpdateEvent } from 'src/engine/core-modules/event-emitter/types/object-record-update.event';
export type ObjectRecordEvent<T = object> =
| ObjectRecordUpdateEvent<T>
| ObjectRecordDeleteEvent<T>
| ObjectRecordCreateEvent<T>
| ObjectRecordDestroyEvent<T>
| ObjectRecordRestoreEvent<T>;

10
packages/twenty-server/src/engine/core-modules/event-emitter/types/object-record-non-destructive-event.ts Normal file
View File

@ -0,0 +1,10 @@
import { ObjectRecordCreateEvent } from 'src/engine/core-modules/event-emitter/types/object-record-create.event';
import { ObjectRecordDeleteEvent } from 'src/engine/core-modules/event-emitter/types/object-record-delete.event';
import { ObjectRecordRestoreEvent } from 'src/engine/core-modules/event-emitter/types/object-record-restore.event';
import { ObjectRecordUpdateEvent } from 'src/engine/core-modules/event-emitter/types/object-record-update.event';
export type ObjectRecordNonDestructiveEvent =
| ObjectRecordCreateEvent
| ObjectRecordUpdateEvent
| ObjectRecordDeleteEvent
| ObjectRecordRestoreEvent;

6
packages/twenty-server/src/engine/core-modules/event-emitter/types/object-record-restore.event.ts
View File

@ -1,9 +1,9 @@
import { ObjectRecordBaseEvent } from 'src/engine/core-modules/event-emitter/types/object-record.base.event';
import { ObjectRecordCreateEvent } from 'src/engine/core-modules/event-emitter/types/object-record-create.event';
export class ObjectRecordRestoreEvent<
T = object,
> extends ObjectRecordBaseEvent<T> {
> extends ObjectRecordCreateEvent<T> {
properties: {
before: T;
after: T;
};
}

4
packages/twenty-server/src/engine/core-modules/event-emitter/types/object-record-update.event.ts
View File

@ -1,13 +1,13 @@
import { ObjectRecordBaseEvent } from 'src/engine/core-modules/event-emitter/types/object-record.base.event';
import { ObjectRecordDiff } from 'src/engine/core-modules/event-emitter/types/object-record-diff';
import { ObjectRecordBaseEvent } from 'src/engine/core-modules/event-emitter/types/object-record.base.event';
export class ObjectRecordUpdateEvent<
T = object,
> extends ObjectRecordBaseEvent<T> {
properties: {
updatedFields?: string[];
diff?: Partial<ObjectRecordDiff<T>>;
before: T;
after: T;
diff?: Partial<ObjectRecordDiff<T>>;
};
}

4
packages/twenty-server/src/modules/messaging/message-participant-manager/listeners/message-participant-person.listener.ts
View File

@ -1,5 +1,7 @@
import { Injectable } from '@nestjs/common';
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';
import { ObjectRecordCreateEvent } from 'src/engine/core-modules/event-emitter/types/object-record-create.event';
import { ObjectRecordUpdateEvent } from 'src/engine/core-modules/event-emitter/types/object-record-update.event';
import { objectRecordChangedProperties as objectRecordUpdateEventChangedProperties } from 'src/engine/core-modules/event-emitter/utils/object-record-changed-properties.util';
@ -16,8 +18,6 @@ import {
MessageParticipantUnmatchParticipantJobData,
} from 'src/modules/messaging/message-participant-manager/jobs/message-participant-unmatch-participant.job';
import { PersonWorkspaceEntity } from 'src/modules/person/standard-objects/person.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';
@Injectable()
export class MessageParticipantPersonListener {

18
packages/twenty-server/src/modules/timeline/jobs/create-audit-log-from-internal-event.ts
View File

@ -1,4 +1,4 @@
import { ObjectRecordBaseEvent } from 'src/engine/core-modules/event-emitter/types/object-record.base.event';
import { ObjectRecordEvent } from 'src/engine/core-modules/event-emitter/types/object-record-event.event';
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';
@ -20,7 +20,7 @@ export class CreateAuditLogFromInternalEvent {
@Process(CreateAuditLogFromInternalEvent.name)
async handle(
workspaceEventBatch: WorkspaceEventBatch<ObjectRecordBaseEvent>,
workspaceEventBatch: WorkspaceEventBatch<ObjectRecordEvent>,
): Promise<void> {
for (const eventData of workspaceEventBatch.events) {
let workspaceMemberId: string | null = null;
@ -34,16 +34,14 @@ export class CreateAuditLogFromInternalEvent {
workspaceMemberId = workspaceMember.id;
}
if (eventData.properties.diff) {
// we remove "before" and "after" property for a cleaner/slimmer event payload
eventData.properties = {
diff: eventData.properties.diff,
};
}
await this.auditLogRepository.insert(
workspaceEventBatch.name,
eventData.properties,
'diff' in eventData.properties
? {
// we remove "before" and "after" property for a cleaner/slimmer event payload
diff: eventData.properties.diff,
}
: eventData.properties,
workspaceMemberId,
workspaceEventBatch.name.split('.')[0],
eventData.objectMetadata.id,

25
packages/twenty-server/src/modules/timeline/jobs/upsert-timeline-activity-from-internal-event.job.ts
View File

@ -1,4 +1,4 @@
import { ObjectRecordBaseEvent } from 'src/engine/core-modules/event-emitter/types/object-record.base.event';
import { ObjectRecordNonDestructiveEvent } from 'src/engine/core-modules/event-emitter/types/object-record-non-destructive-event';
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';
@ -7,7 +7,6 @@ import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/wo
import { TimelineActivityService } from 'src/modules/timeline/services/timeline-activity.service';
import { WorkspaceMemberRepository } from 'src/modules/workspace-member/repositories/workspace-member.repository';
import { WorkspaceMemberWorkspaceEntity } from 'src/modules/workspace-member/standard-objects/workspace-member.workspace-entity';
import { TimelineActivityWorkspaceEntity } from 'src/modules/timeline/standard-objects/timeline-activity.workspace-entity';
@Processor(MessageQueue.entityEventsToDbQueue)
export class UpsertTimelineActivityFromInternalEvent {
@ -19,9 +18,7 @@ export class UpsertTimelineActivityFromInternalEvent {
@Process(UpsertTimelineActivityFromInternalEvent.name)
async handle(
workspaceEventBatch: WorkspaceEventBatch<
ObjectRecordBaseEvent<TimelineActivityWorkspaceEntity>
>,
workspaceEventBatch: WorkspaceEventBatch<ObjectRecordNonDestructiveEvent>,
): Promise<void> {
for (const eventData of workspaceEventBatch.events) {
if (eventData.userId) {
@ -33,13 +30,6 @@ export class UpsertTimelineActivityFromInternalEvent {
eventData.workspaceMemberId = workspaceMember.id;
}
if (eventData.properties.diff) {
// we remove "before" and "after" property for a cleaner/slimmer event payload
eventData.properties = {
diff: eventData.properties.diff,
};
}
// Temporary
// We ignore every that is not a LinkedObject or a Business Object
if (
@ -51,7 +41,16 @@ export class UpsertTimelineActivityFromInternalEvent {
}
await this.timelineActivityService.upsertEvent({
event: eventData,
event:
// we remove "before" and "after" property for a cleaner/slimmer event payload
'diff' in eventData.properties && eventData.properties.diff
? {
...eventData,
properties: {
diff: eventData.properties.diff,
},
}
: eventData,
eventName: workspaceEventBatch.name,
workspaceId: workspaceEventBatch.workspaceId,
});

38
packages/twenty-server/src/modules/timeline/services/timeline-activity.service.ts
View File

@ -1,19 +1,20 @@
import { Injectable } from '@nestjs/common';
import { ObjectRecordNonDestructiveEvent } from 'src/engine/core-modules/event-emitter/types/object-record-non-destructive-event';
import { ObjectRecordBaseEvent } from 'src/engine/core-modules/event-emitter/types/object-record.base.event';
import { InjectObjectMetadataRepository } from 'src/engine/object-metadata-repository/object-metadata-repository.decorator';
import { WorkspaceDataSourceService } from 'src/engine/workspace-datasource/workspace-datasource.service';
import { TimelineActivityRepository } from 'src/modules/timeline/repositiories/timeline-activity.repository';
import { TimelineActivityWorkspaceEntity } from 'src/modules/timeline/standard-objects/timeline-activity.workspace-entity';
type TimelineActivity =
ObjectRecordBaseEvent<TimelineActivityWorkspaceEntity> & {
name: string;
objectName?: string;
linkedRecordCachedName?: string;
linkedRecordId?: string;
linkedObjectMetadataId?: string;
};
type TimelineActivity = Omit<ObjectRecordNonDestructiveEvent, 'properties'> & {
name: string;
objectName?: string;
linkedRecordCachedName?: string;
linkedRecordId?: string;
linkedObjectMetadataId?: string;
properties: Record<string, any>; // more relaxed conditions than for internal events
};
@Injectable()
export class TimelineActivityService {
@ -33,7 +34,7 @@ export class TimelineActivityService {
eventName,
workspaceId,
}: {
event: ObjectRecordBaseEvent<TimelineActivityWorkspaceEntity>;
event: ObjectRecordBaseEvent;
eventName: string;
workspaceId: string;
}) {
@ -65,7 +66,7 @@ export class TimelineActivityService {
workspaceId,
eventName,
}: {
event: ObjectRecordBaseEvent<TimelineActivityWorkspaceEntity>;
event: ObjectRecordBaseEvent;
workspaceId: string;
eventName: string;
}): Promise<TimelineActivity[] | undefined> {
@ -78,7 +79,10 @@ export class TimelineActivityService {
// 2 timelines, one for the linked object and one for the task/note
if (linkedTimelineActivities && linkedTimelineActivities?.length > 0)
return [...linkedTimelineActivities, { ...event, name: eventName }];
return [
...linkedTimelineActivities,
{ ...event, name: eventName },
] satisfies TimelineActivity[];
}
if (
@ -93,7 +97,7 @@ export class TimelineActivityService {
});
}
return [{ ...event, name: eventName }];
return [{ ...event, name: eventName }] satisfies TimelineActivity[];
}
private async getLinkedTimelineActivities({
@ -101,7 +105,7 @@ export class TimelineActivityService {
workspaceId,
eventName,
}: {
event: ObjectRecordBaseEvent<TimelineActivityWorkspaceEntity>;
event: ObjectRecordBaseEvent;
workspaceId: string;
eventName: string;
}): Promise<TimelineActivity[] | undefined> {
@ -146,7 +150,7 @@ export class TimelineActivityService {
eventName,
workspaceId,
}: {
event: ObjectRecordBaseEvent<TimelineActivityWorkspaceEntity>;
event: ObjectRecordBaseEvent;
dataSourceSchema: string;
activityType: string;
eventName: string;
@ -195,7 +199,7 @@ export class TimelineActivityService {
linkedRecordCachedName: activity[0].title,
linkedRecordId: activity[0].id,
linkedObjectMetadataId: event.objectMetadata.id,
} as TimelineActivity;
} satisfies TimelineActivity;
})
.filter((event): event is TimelineActivity => event !== undefined);
}
@ -207,7 +211,7 @@ export class TimelineActivityService {
eventName,
workspaceId,
}: {
event: ObjectRecordBaseEvent<TimelineActivityWorkspaceEntity>;
event: ObjectRecordBaseEvent;
dataSourceSchema: string;
activityType: string;
eventName: string;
@ -258,7 +262,7 @@ export class TimelineActivityService {
linkedRecordCachedName: activity[0].title,
linkedRecordId: activity[0].id,
linkedObjectMetadataId: activityObjectMetadataId,
} as TimelineActivity,
} satisfies TimelineActivity,
];
}
}

18
packages/twenty-server/src/modules/webhook/jobs/call-webhook-jobs.job.ts
View File

@ -2,19 +2,19 @@ import { Logger } from '@nestjs/common';
import { ArrayContains } from 'typeorm';
import { ObjectRecordEvent } from 'src/engine/core-modules/event-emitter/types/object-record-event.event';
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';
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 { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service';
import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager';
import { WebhookWorkspaceEntity } from 'src/modules/webhook/standard-objects/webhook.workspace-entity';
import { ObjectRecordBaseEvent } from 'src/engine/core-modules/event-emitter/types/object-record.base.event';
import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type';
import {
CallWebhookJob,
CallWebhookJobData,
} from 'src/modules/webhook/jobs/call-webhook.job';
import { WebhookWorkspaceEntity } from 'src/modules/webhook/standard-objects/webhook.workspace-entity';
import { removeSecretFromWebhookRecord } from 'src/utils/remove-secret-from-webhook-record';
@Processor(MessageQueue.webhookQueue)
@ -29,7 +29,7 @@ export class CallWebhookJobsJob {
@Process(CallWebhookJobsJob.name)
async handle(
workspaceEventBatch: WorkspaceEventBatch<ObjectRecordBaseEvent>,
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
@ -60,8 +60,16 @@ export class CallWebhookJobsJob {
nameSingular: eventData.objectMetadata.nameSingular,
};
const workspaceId = workspaceEventBatch.workspaceId;
const record = eventData.properties.after || eventData.properties.before;
const updatedFields = eventData.properties.updatedFields;
const record =
'after' in eventData.properties
? eventData.properties.after
: 'before' in eventData.properties
? eventData.properties.before
: {};
const updatedFields =
'updatedFields' in eventData.properties
? eventData.properties.updatedFields
: undefined;
const isWebhookEvent = nameSingular === 'webhook';
const sanitizedRecord = removeSecretFromWebhookRecord(