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

8643 fix sentry error (#8644)

Browse Source 
- fixes missing data in event payload when adding a new workspaceMember
- add strong typing to database event emitters
This commit is contained in:
martmull
2024-11-21 17:09:36 +01:00
committed by GitHub
parent 395da91071
commit 39373b4a28
61 changed files with 460 additions and 311 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
packages/twenty-server/src/engine/api/graphql/graphql-query-runner/decorators/on-custom-batch-event.decorator.ts Normal file
View File

@ -0,0 +1,13 @@
import { OnEvent } from '@nestjs/event-emitter';
import { CustomEventName } from 'src/engine/workspace-event-emitter/types/custom-event-name.type';
export function OnCustomBatchEvent(event: CustomEventName): MethodDecorator {
return (
target: object,
propertyKey: string,
descriptor: PropertyDescriptor,
) => {
OnEvent(event)(target, propertyKey, descriptor);
};
}

2
packages/twenty-server/src/engine/api/graphql/graphql-query-runner/decorators/on-database-event.decorator.ts → packages/twenty-server/src/engine/api/graphql/graphql-query-runner/decorators/on-database-batch-event.decorator.ts
View File

@ -2,7 +2,7 @@ import { OnEvent } from '@nestjs/event-emitter';
import { DatabaseEventAction } from 'src/engine/api/graphql/graphql-query-runner/enums/database-event-action'; import { DatabaseEventAction } from 'src/engine/api/graphql/graphql-query-runner/enums/database-event-action';
export function OnDatabaseEvent( export function OnDatabaseBatchEvent(
object: string, object: string,
action: DatabaseEventAction, action: DatabaseEventAction,
): MethodDecorator { ): MethodDecorator {

44
packages/twenty-server/src/engine/api/graphql/graphql-query-runner/services/api-event-emitter.service.ts
View File

@ -17,9 +17,10 @@ export class ApiEventEmitterService {
authContext: AuthContext, authContext: AuthContext,
objectMetadataItem: ObjectMetadataInterface, objectMetadataItem: ObjectMetadataInterface,
): void { ): void {
this.workspaceEventEmitter.emit( this.workspaceEventEmitter.emitDatabaseBatchEvent({
`${objectMetadataItem.nameSingular}.${DatabaseEventAction.CREATED}`, objectMetadataNameSingular: objectMetadataItem.nameSingular,
records.map((record) => ({ action: DatabaseEventAction.CREATED,
events: records.map((record) => ({
userId: authContext.user?.id, userId: authContext.user?.id,
recordId: record.id, recordId: record.id,
objectMetadata: objectMetadataItem, objectMetadata: objectMetadataItem,
@ -28,8 +29,8 @@ export class ApiEventEmitterService {
after: this.removeGraphQLAndNestedProperties(record), after: this.removeGraphQLAndNestedProperties(record),
}, },
})), })),
authContext.workspace.id, workspaceId: authContext.workspace.id,
); });
} }
public emitUpdateEvents<T extends ObjectRecord>( public emitUpdateEvents<T extends ObjectRecord>(
@ -47,9 +48,10 @@ export class ApiEventEmitterService {
{}, {},
); );
this.workspaceEventEmitter.emit( this.workspaceEventEmitter.emitDatabaseBatchEvent({
`${objectMetadataItem.nameSingular}.${DatabaseEventAction.UPDATED}`, objectMetadataNameSingular: objectMetadataItem.nameSingular,
records.map((record) => { action: DatabaseEventAction.UPDATED,
events: records.map((record) => {
const before = this.removeGraphQLAndNestedProperties( const before = this.removeGraphQLAndNestedProperties(
mappedExistingRecords[record.id], mappedExistingRecords[record.id],
); );
@ -73,8 +75,8 @@ export class ApiEventEmitterService {
}, },
}; };
}), }),
authContext.workspace.id, workspaceId: authContext.workspace.id,
); });
} }
public emitDeletedEvents<T extends ObjectRecord>( public emitDeletedEvents<T extends ObjectRecord>(
@ -82,9 +84,10 @@ export class ApiEventEmitterService {
authContext: AuthContext, authContext: AuthContext,
objectMetadataItem: ObjectMetadataInterface, objectMetadataItem: ObjectMetadataInterface,
): void { ): void {
this.workspaceEventEmitter.emit( this.workspaceEventEmitter.emitDatabaseBatchEvent({
`${objectMetadataItem.nameSingular}.${DatabaseEventAction.DELETED}`, objectMetadataNameSingular: objectMetadataItem.nameSingular,
records.map((record) => { action: DatabaseEventAction.DELETED,
events: records.map((record) => {
return { return {
userId: authContext.user?.id, userId: authContext.user?.id,
recordId: record.id, recordId: record.id,
@ -95,8 +98,8 @@ export class ApiEventEmitterService {
}, },
}; };
}), }),
authContext.workspace.id, workspaceId: authContext.workspace.id,
); });
} }
public emitDestroyEvents<T extends ObjectRecord>( public emitDestroyEvents<T extends ObjectRecord>(
@ -104,9 +107,10 @@ export class ApiEventEmitterService {
authContext: AuthContext, authContext: AuthContext,
objectMetadataItem: ObjectMetadataInterface, objectMetadataItem: ObjectMetadataInterface,
): void { ): void {
this.workspaceEventEmitter.emit( this.workspaceEventEmitter.emitDatabaseBatchEvent({
`${objectMetadataItem.nameSingular}.${DatabaseEventAction.DESTROYED}`, objectMetadataNameSingular: objectMetadataItem.nameSingular,
records.map((record) => { action: DatabaseEventAction.DESTROYED,
events: records.map((record) => {
return { return {
userId: authContext.user?.id, userId: authContext.user?.id,
recordId: record.id, recordId: record.id,
@ -117,8 +121,8 @@ export class ApiEventEmitterService {
}, },
}; };
}), }),
authContext.workspace.id, workspaceId: authContext.workspace.id,
); });
} }
private removeGraphQLAndNestedProperties<T extends ObjectRecord>(record: T) { private removeGraphQLAndNestedProperties<T extends ObjectRecord>(record: T) {

1
packages/twenty-server/src/engine/api/graphql/workspace-query-runner/constants/user-signup-event-name.constants.ts Normal file
View File

@ -0,0 +1 @@
export const USER_SIGNUP_EVENT_NAME = 'user_signup';

5
packages/twenty-server/src/engine/api/graphql/workspace-query-runner/jobs/record-position-backfill.job.ts
View File

@ -16,6 +16,9 @@ export class RecordPositionBackfillJob {
@Process(RecordPositionBackfillJob.name) @Process(RecordPositionBackfillJob.name)
async handle(data: RecordPositionBackfillJobData): Promise<void> { async handle(data: RecordPositionBackfillJobData): Promise<void> {
this.recordPositionBackfillService.backfill(data.workspaceId, data.dryRun); await this.recordPositionBackfillService.backfill(
data.workspaceId,
data.dryRun,
);
} }
} }

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

@ -6,10 +6,10 @@ import { ObjectRecordBaseEvent } from 'src/engine/core-modules/event-emitter/typ
import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator'; 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 { 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 { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service';
import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/workspace-event.type'; 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 { 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 { UpsertTimelineActivityFromInternalEvent } from 'src/modules/timeline/jobs/upsert-timeline-activity-from-internal-event.job';
import { OnDatabaseEvent } from 'src/engine/api/graphql/graphql-query-runner/decorators/on-database-event.decorator'; 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 { CallWebhookJobsJob } from 'src/modules/webhook/jobs/call-webhook-jobs.job';
import { DatabaseEventAction } from 'src/engine/api/graphql/graphql-query-runner/enums/database-event-action'; import { DatabaseEventAction } from 'src/engine/api/graphql/graphql-query-runner/enums/database-event-action';
@ -22,55 +22,49 @@ export class EntityEventsToDbListener {
private readonly webhookQueueService: MessageQueueService, private readonly webhookQueueService: MessageQueueService,
) {} ) {}
@OnDatabaseEvent('*', DatabaseEventAction.CREATED) @OnDatabaseBatchEvent('*', DatabaseEventAction.CREATED)
async handleCreate( async handleCreate(batchEvent: WorkspaceEventBatch<ObjectRecordCreateEvent>) {
payload: WorkspaceEventBatch<ObjectRecordCreateEvent<any>>, return this.handle(batchEvent);
) {
return this.handle(payload);
} }
@OnDatabaseEvent('*', DatabaseEventAction.UPDATED) @OnDatabaseBatchEvent('*', DatabaseEventAction.UPDATED)
async handleUpdate( async handleUpdate(batchEvent: WorkspaceEventBatch<ObjectRecordUpdateEvent>) {
payload: WorkspaceEventBatch<ObjectRecordUpdateEvent<any>>, return this.handle(batchEvent);
) {
return this.handle(payload);
} }
@OnDatabaseEvent('*', DatabaseEventAction.DELETED) @OnDatabaseBatchEvent('*', DatabaseEventAction.DELETED)
async handleDelete( async handleDelete(batchEvent: WorkspaceEventBatch<ObjectRecordUpdateEvent>) {
payload: WorkspaceEventBatch<ObjectRecordUpdateEvent<any>>, return this.handle(batchEvent);
) {
return this.handle(payload);
} }
@OnDatabaseEvent('*', DatabaseEventAction.DESTROYED) @OnDatabaseBatchEvent('*', DatabaseEventAction.DESTROYED)
async handleDestroy( async handleDestroy(
payload: WorkspaceEventBatch<ObjectRecordUpdateEvent<any>>, batchEvent: WorkspaceEventBatch<ObjectRecordUpdateEvent>,
) { ) {
return this.handle(payload); return this.handle(batchEvent);
} }
private async handle(payload: WorkspaceEventBatch<ObjectRecordBaseEvent>) { private async handle(batchEvent: WorkspaceEventBatch<ObjectRecordBaseEvent>) {
const filteredEvents = payload.events.filter( const filteredEvents = batchEvent.events.filter(
(event) => event.objectMetadata?.isAuditLogged, (event) => event.objectMetadata?.isAuditLogged,
); );
await this.entityEventsToDbQueueService.add< await this.entityEventsToDbQueueService.add<
WorkspaceEventBatch<ObjectRecordBaseEvent> WorkspaceEventBatch<ObjectRecordBaseEvent>
>(CreateAuditLogFromInternalEvent.name, { >(CreateAuditLogFromInternalEvent.name, {
...payload, ...batchEvent,
events: filteredEvents, events: filteredEvents,
}); });
await this.entityEventsToDbQueueService.add< await this.entityEventsToDbQueueService.add<
WorkspaceEventBatch<ObjectRecordBaseEvent> WorkspaceEventBatch<ObjectRecordBaseEvent>
>(UpsertTimelineActivityFromInternalEvent.name, { >(UpsertTimelineActivityFromInternalEvent.name, {
...payload, ...batchEvent,
events: filteredEvents, events: filteredEvents,
}); });
await this.webhookQueueService.add< await this.webhookQueueService.add<
WorkspaceEventBatch<ObjectRecordBaseEvent> WorkspaceEventBatch<ObjectRecordBaseEvent>
>(CallWebhookJobsJob.name, payload, { retryLimit: 3 }); >(CallWebhookJobsJob.name, batchEvent, { retryLimit: 3 });
} }
} }

21
packages/twenty-server/src/engine/api/graphql/workspace-query-runner/listeners/telemetry.listener.ts
View File

@ -1,12 +1,13 @@
import { Injectable } from '@nestjs/common'; import { Injectable } from '@nestjs/common';
import { OnEvent } from '@nestjs/event-emitter';
import { OnDatabaseEvent } from 'src/engine/api/graphql/graphql-query-runner/decorators/on-database-event.decorator'; 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 { DatabaseEventAction } from 'src/engine/api/graphql/graphql-query-runner/enums/database-event-action';
import { AnalyticsService } from 'src/engine/core-modules/analytics/analytics.service'; import { AnalyticsService } from 'src/engine/core-modules/analytics/analytics.service';
import { ObjectRecordCreateEvent } from 'src/engine/core-modules/event-emitter/types/object-record-create.event'; import { ObjectRecordCreateEvent } from 'src/engine/core-modules/event-emitter/types/object-record-create.event';
import { TelemetryService } from 'src/engine/core-modules/telemetry/telemetry.service'; import { TelemetryService } from 'src/engine/core-modules/telemetry/telemetry.service';
import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/workspace-event.type'; import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type';
import { USER_SIGNUP_EVENT_NAME } from 'src/engine/api/graphql/workspace-query-runner/constants/user-signup-event-name.constants';
import { OnCustomBatchEvent } from 'src/engine/api/graphql/graphql-query-runner/decorators/on-custom-batch-event.decorator';
@Injectable() @Injectable()
export class TelemetryListener { export class TelemetryListener {
@ -15,10 +16,8 @@ export class TelemetryListener {
private readonly telemetryService: TelemetryService, private readonly telemetryService: TelemetryService,
) {} ) {}
@OnDatabaseEvent('*', DatabaseEventAction.CREATED) @OnDatabaseBatchEvent('*', DatabaseEventAction.CREATED)
async handleAllCreate( async handleAllCreate(payload: WorkspaceEventBatch<ObjectRecordCreateEvent>) {
payload: WorkspaceEventBatch<ObjectRecordCreateEvent<any>>,
) {
await Promise.all( await Promise.all(
payload.events.map((eventPayload) => payload.events.map((eventPayload) =>
this.analyticsService.create( this.analyticsService.create(
@ -33,15 +32,15 @@ export class TelemetryListener {
); );
} }
@OnEvent('user.signup') @OnCustomBatchEvent(USER_SIGNUP_EVENT_NAME)
async handleUserSignup( async handleUserSignup(
payload: WorkspaceEventBatch<ObjectRecordCreateEvent<any>>, payload: WorkspaceEventBatch<ObjectRecordCreateEvent>,
) { ) {
await Promise.all( await Promise.all(
payload.events.map(async (eventPayload) => { payload.events.map(async (eventPayload) => {
this.analyticsService.create( this.analyticsService.create(
{ {
action: 'user.signup', action: USER_SIGNUP_EVENT_NAME,
payload: {}, payload: {},
}, },
eventPayload.userId, eventPayload.userId,
@ -50,7 +49,7 @@ export class TelemetryListener {
this.telemetryService.create( this.telemetryService.create(
{ {
action: 'user.signup', action: USER_SIGNUP_EVENT_NAME,
payload: { payload: {
payload, payload,
userId: undefined, userId: undefined,

8
packages/twenty-server/src/engine/core-modules/billing/listeners/billing-workspace-member.listener.ts
View File

@ -9,9 +9,9 @@ import { ObjectRecordCreateEvent } from 'src/engine/core-modules/event-emitter/t
import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator'; 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 { 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 { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service';
import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/workspace-event.type'; import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type';
import { WorkspaceMemberWorkspaceEntity } from 'src/modules/workspace-member/standard-objects/workspace-member.workspace-entity'; import { WorkspaceMemberWorkspaceEntity } from 'src/modules/workspace-member/standard-objects/workspace-member.workspace-entity';
import { OnDatabaseEvent } from 'src/engine/api/graphql/graphql-query-runner/decorators/on-database-event.decorator'; 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 { DatabaseEventAction } from 'src/engine/api/graphql/graphql-query-runner/enums/database-event-action';
@Injectable() @Injectable()
@ -22,8 +22,8 @@ export class BillingWorkspaceMemberListener {
private readonly environmentService: EnvironmentService, private readonly environmentService: EnvironmentService,
) {} ) {}
@OnDatabaseEvent('workspaceMember', DatabaseEventAction.CREATED) @OnDatabaseBatchEvent('workspaceMember', DatabaseEventAction.CREATED)
@OnDatabaseEvent('workspaceMember', DatabaseEventAction.DELETED) @OnDatabaseBatchEvent('workspaceMember', DatabaseEventAction.DELETED)
async handleCreateOrDeleteEvent( async handleCreateOrDeleteEvent(
payload: WorkspaceEventBatch< payload: WorkspaceEventBatch<
ObjectRecordCreateEvent<WorkspaceMemberWorkspaceEntity> ObjectRecordCreateEvent<WorkspaceMemberWorkspaceEntity>

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

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

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

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

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

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

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

@ -0,0 +1,3 @@
export type ObjectRecordDiff<T> = {
[K in keyof T]: { before: T[K]; after: T[K] };
};

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

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

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

@ -1,9 +1,18 @@
import { ObjectMetadataInterface } from 'src/engine/metadata-modules/field-metadata/interfaces/object-metadata.interface'; import { ObjectMetadataInterface } from 'src/engine/metadata-modules/field-metadata/interfaces/object-metadata.interface';
export class ObjectRecordBaseEvent { import { ObjectRecordDiff } from 'src/engine/core-modules/event-emitter/types/object-record-diff';
type Properties<T> = {
updatedFields?: string[];
before?: T;
after?: T;
diff?: Partial<ObjectRecordDiff<T>>;
};
export class ObjectRecordBaseEvent<T = object> {
recordId: string; recordId: string;
userId?: string; userId?: string;
workspaceMemberId?: string; workspaceMemberId?: string;
objectMetadata: ObjectMetadataInterface; objectMetadata: ObjectMetadataInterface;
properties: any; properties: Properties<T>;
} }

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

@ -11,6 +11,7 @@ import { WorkspaceDataSourceModule } from 'src/engine/workspace-datasource/works
import { User } from 'src/engine/core-modules/user/user.entity'; import { User } from 'src/engine/core-modules/user/user.entity';
import { AppToken } from 'src/engine/core-modules/app-token/app-token.entity'; import { AppToken } from 'src/engine/core-modules/app-token/app-token.entity';
import { WorkspaceInvitationModule } from 'src/engine/core-modules/workspace-invitation/workspace-invitation.module'; import { WorkspaceInvitationModule } from 'src/engine/core-modules/workspace-invitation/workspace-invitation.module';
import { ObjectMetadataEntity } from 'src/engine/metadata-modules/object-metadata/object-metadata.entity';
@Module({ @Module({
imports: [ imports: [
@ -20,6 +21,7 @@ import { WorkspaceInvitationModule } from 'src/engine/core-modules/workspace-inv
[User, UserWorkspace, AppToken], [User, UserWorkspace, AppToken],
'core', 'core',
), ),
NestjsQueryTypeOrmModule.forFeature([ObjectMetadataEntity], 'metadata'),
TypeORMModule, TypeORMModule,
DataSourceModule, DataSourceModule,
WorkspaceDataSourceModule, WorkspaceDataSourceModule,

45
packages/twenty-server/src/engine/core-modules/user-workspace/user-workspace.service.ts
View File

@ -9,16 +9,16 @@ import {
AppToken, AppToken,
AppTokenType, AppTokenType,
} from 'src/engine/core-modules/app-token/app-token.entity'; } from 'src/engine/core-modules/app-token/app-token.entity';
import { ObjectRecordCreateEvent } from 'src/engine/core-modules/event-emitter/types/object-record-create.event';
import { UserWorkspace } from 'src/engine/core-modules/user-workspace/user-workspace.entity'; import { UserWorkspace } from 'src/engine/core-modules/user-workspace/user-workspace.entity';
import { User } from 'src/engine/core-modules/user/user.entity'; import { User } from 'src/engine/core-modules/user/user.entity';
import { WorkspaceInvitationService } from 'src/engine/core-modules/workspace-invitation/services/workspace-invitation.service'; import { WorkspaceInvitationService } from 'src/engine/core-modules/workspace-invitation/services/workspace-invitation.service';
import { Workspace } from 'src/engine/core-modules/workspace/workspace.entity'; import { Workspace } from 'src/engine/core-modules/workspace/workspace.entity';
import { DataSourceService } from 'src/engine/metadata-modules/data-source/data-source.service'; import { DataSourceService } from 'src/engine/metadata-modules/data-source/data-source.service';
import { WorkspaceEventEmitter } from 'src/engine/workspace-event-emitter/workspace-event-emitter'; import { WorkspaceEventEmitter } from 'src/engine/workspace-event-emitter/workspace-event-emitter';
import { WorkspaceMemberWorkspaceEntity } from 'src/modules/workspace-member/standard-objects/workspace-member.workspace-entity';
import { assert } from 'src/utils/assert'; import { assert } from 'src/utils/assert';
import { DatabaseEventAction } from 'src/engine/api/graphql/graphql-query-runner/enums/database-event-action'; import { DatabaseEventAction } from 'src/engine/api/graphql/graphql-query-runner/enums/database-event-action';
import { ObjectMetadataEntity } from 'src/engine/metadata-modules/object-metadata/object-metadata.entity';
import { USER_SIGNUP_EVENT_NAME } from 'src/engine/api/graphql/workspace-query-runner/constants/user-signup-event-name.constants';
export class UserWorkspaceService extends TypeOrmQueryService<UserWorkspace> { export class UserWorkspaceService extends TypeOrmQueryService<UserWorkspace> {
constructor( constructor(
@ -28,6 +28,8 @@ export class UserWorkspaceService extends TypeOrmQueryService<UserWorkspace> {
private readonly userRepository: Repository<User>, private readonly userRepository: Repository<User>,
@InjectRepository(AppToken, 'core') @InjectRepository(AppToken, 'core')
private readonly appTokenRepository: Repository<AppToken>, private readonly appTokenRepository: Repository<AppToken>,
@InjectRepository(ObjectMetadataEntity, 'metadata')
private readonly objectMetadataRepository: Repository<ObjectMetadataEntity>,
private readonly dataSourceService: DataSourceService, private readonly dataSourceService: DataSourceService,
private readonly typeORMService: TypeORMService, private readonly typeORMService: TypeORMService,
private readonly workspaceInvitationService: WorkspaceInvitationService, private readonly workspaceInvitationService: WorkspaceInvitationService,
@ -42,11 +44,11 @@ export class UserWorkspaceService extends TypeOrmQueryService<UserWorkspace> {
workspaceId, workspaceId,
}); });
const payload = new ObjectRecordCreateEvent<UserWorkspace>(); this.workspaceEventEmitter.emitCustomBatchEvent(
USER_SIGNUP_EVENT_NAME,
payload.userId = userId; [{ userId }],
workspaceId,
this.workspaceEventEmitter.emit('user.signup', [payload], workspaceId); );
return this.userWorkspaceRepository.save(userWorkspace); return this.userWorkspaceRepository.save(userWorkspace);
} }
@ -80,19 +82,26 @@ export class UserWorkspaceService extends TypeOrmQueryService<UserWorkspace> {
workspaceMember.length === 1, workspaceMember.length === 1,
`Error while creating workspace member ${user.email} on workspace ${workspaceId}`, `Error while creating workspace member ${user.email} on workspace ${workspaceId}`,
); );
const payload = const objectMetadata = await this.objectMetadataRepository.findOneOrFail({
new ObjectRecordCreateEvent<WorkspaceMemberWorkspaceEntity>(); where: {
nameSingular: 'workspaceMember',
},
});
payload.properties = { this.workspaceEventEmitter.emitDatabaseBatchEvent({
after: workspaceMember[0], objectMetadataNameSingular: 'workspaceMember',
}; action: DatabaseEventAction.CREATED,
payload.recordId = workspaceMember[0].id; events: [
{
this.workspaceEventEmitter.emit( recordId: workspaceMember[0].id,
`workspaceMember.${DatabaseEventAction.CREATED}`, objectMetadata,
[payload], properties: {
after: workspaceMember[0],
},
},
],
workspaceId, workspaceId,
); });
} }
async addUserToWorkspace(user: User, workspace: Workspace) { async addUserToWorkspace(user: User, workspace: Workspace) {

5
packages/twenty-server/src/engine/core-modules/user/services/__tests__/user.service.spec.ts
View File

@ -9,6 +9,7 @@ import { WorkspaceService } from 'src/engine/core-modules/workspace/services/wor
import { DataSourceService } from 'src/engine/metadata-modules/data-source/data-source.service'; import { DataSourceService } from 'src/engine/metadata-modules/data-source/data-source.service';
import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager'; import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager';
import { WorkspaceEventEmitter } from 'src/engine/workspace-event-emitter/workspace-event-emitter'; import { WorkspaceEventEmitter } from 'src/engine/workspace-event-emitter/workspace-event-emitter';
import { ObjectMetadataEntity } from 'src/engine/metadata-modules/object-metadata/object-metadata.entity';
describe('UserService', () => { describe('UserService', () => {
let service: UserService; let service: UserService;
@ -25,6 +26,10 @@ describe('UserService', () => {
provide: getRepositoryToken(UserWorkspace, 'core'), provide: getRepositoryToken(UserWorkspace, 'core'),
useValue: {}, useValue: {},
}, },
{
provide: getRepositoryToken(ObjectMetadataEntity, 'metadata'),
useValue: {},
},
{ {
provide: DataSourceService, provide: DataSourceService,
useValue: {}, useValue: {},

36
packages/twenty-server/src/engine/core-modules/user/services/user.service.ts
View File

@ -6,7 +6,6 @@ import { TypeOrmQueryService } from '@ptc-org/nestjs-query-typeorm';
import { Repository } from 'typeorm'; import { Repository } from 'typeorm';
import { TypeORMService } from 'src/database/typeorm/typeorm.service'; import { TypeORMService } from 'src/database/typeorm/typeorm.service';
import { ObjectRecordDeleteEvent } from 'src/engine/core-modules/event-emitter/types/object-record-delete.event';
import { User } from 'src/engine/core-modules/user/user.entity'; import { User } from 'src/engine/core-modules/user/user.entity';
import { WorkspaceService } from 'src/engine/core-modules/workspace/services/workspace.service'; import { WorkspaceService } from 'src/engine/core-modules/workspace/services/workspace.service';
import { import {
@ -18,12 +17,15 @@ import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.
import { WorkspaceEventEmitter } from 'src/engine/workspace-event-emitter/workspace-event-emitter'; import { WorkspaceEventEmitter } from 'src/engine/workspace-event-emitter/workspace-event-emitter';
import { WorkspaceMemberWorkspaceEntity } from 'src/modules/workspace-member/standard-objects/workspace-member.workspace-entity'; import { WorkspaceMemberWorkspaceEntity } from 'src/modules/workspace-member/standard-objects/workspace-member.workspace-entity';
import { DatabaseEventAction } from 'src/engine/api/graphql/graphql-query-runner/enums/database-event-action'; import { DatabaseEventAction } from 'src/engine/api/graphql/graphql-query-runner/enums/database-event-action';
import { ObjectMetadataEntity } from 'src/engine/metadata-modules/object-metadata/object-metadata.entity';
// eslint-disable-next-line @nx/workspace-inject-workspace-repository // eslint-disable-next-line @nx/workspace-inject-workspace-repository
export class UserService extends TypeOrmQueryService<User> { export class UserService extends TypeOrmQueryService<User> {
constructor( constructor(
@InjectRepository(User, 'core') @InjectRepository(User, 'core')
private readonly userRepository: Repository<User>, private readonly userRepository: Repository<User>,
@InjectRepository(ObjectMetadataEntity, 'metadata')
private readonly objectMetadataRepository: Repository<ObjectMetadataEntity>,
private readonly dataSourceService: DataSourceService, private readonly dataSourceService: DataSourceService,
private readonly typeORMService: TypeORMService, private readonly typeORMService: TypeORMService,
private readonly workspaceEventEmitter: WorkspaceEventEmitter, private readonly workspaceEventEmitter: WorkspaceEventEmitter,
@ -44,13 +46,11 @@ export class UserService extends TypeOrmQueryService<User> {
'workspaceMember', 'workspaceMember',
); );
const workspaceMember = await workspaceMemberRepository.findOne({ return await workspaceMemberRepository.findOne({
where: { where: {
userId: user.id, userId: user.id,
}, },
}); });
return workspaceMember;
} }
async loadWorkspaceMembers(workspace: Workspace) { async loadWorkspaceMembers(workspace: Workspace) {
@ -107,19 +107,27 @@ export class UserService extends TypeOrmQueryService<User> {
await workspaceDataSource?.query( await workspaceDataSource?.query(
`DELETE FROM ${dataSourceMetadata.schema}."workspaceMember" WHERE "userId" = '${userId}'`, `DELETE FROM ${dataSourceMetadata.schema}."workspaceMember" WHERE "userId" = '${userId}'`,
); );
const payload =
new ObjectRecordDeleteEvent<WorkspaceMemberWorkspaceEntity>();
payload.properties = { const objectMetadata = await this.objectMetadataRepository.findOneOrFail({
before: workspaceMember, where: {
}; nameSingular: 'workspaceMember',
payload.recordId = workspaceMember.id; },
});
this.workspaceEventEmitter.emit( this.workspaceEventEmitter.emitDatabaseBatchEvent({
`workspaceMember.${DatabaseEventAction.DELETED}`, objectMetadataNameSingular: 'workspaceMember',
[payload], action: DatabaseEventAction.DELETED,
events: [
{
recordId: workspaceMember.id,
objectMetadata,
properties: {
before: workspaceMember,
},
},
],
workspaceId, workspaceId,
); });
return user; return user;
} }

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

@ -17,6 +17,7 @@ import { User } from 'src/engine/core-modules/user/user.entity';
import { UserResolver } from 'src/engine/core-modules/user/user.resolver'; import { UserResolver } from 'src/engine/core-modules/user/user.resolver';
import { WorkspaceModule } from 'src/engine/core-modules/workspace/workspace.module'; import { WorkspaceModule } from 'src/engine/core-modules/workspace/workspace.module';
import { DataSourceModule } from 'src/engine/metadata-modules/data-source/data-source.module'; import { DataSourceModule } from 'src/engine/metadata-modules/data-source/data-source.module';
import { ObjectMetadataEntity } from 'src/engine/metadata-modules/object-metadata/object-metadata.entity';
import { userAutoResolverOpts } from './user.auto-resolver-opts'; import { userAutoResolverOpts } from './user.auto-resolver-opts';
@ -32,6 +33,7 @@ import { UserService } from './services/user.service';
], ],
resolvers: userAutoResolverOpts, resolvers: userAutoResolverOpts,
}), }),
NestjsQueryTypeOrmModule.forFeature([ObjectMetadataEntity], 'metadata'),
DataSourceModule, DataSourceModule,
FileUploadModule, FileUploadModule,
WorkspaceModule, WorkspaceModule,

8
packages/twenty-server/src/engine/core-modules/workspace/workspace-workspace-member.listener.ts
View File

@ -10,9 +10,9 @@ import { ObjectRecordUpdateEvent } from 'src/engine/core-modules/event-emitter/t
import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator'; 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 { 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 { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service';
import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/workspace-event.type'; import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type';
import { WorkspaceMemberWorkspaceEntity } from 'src/modules/workspace-member/standard-objects/workspace-member.workspace-entity'; import { WorkspaceMemberWorkspaceEntity } from 'src/modules/workspace-member/standard-objects/workspace-member.workspace-entity';
import { OnDatabaseEvent } from 'src/engine/api/graphql/graphql-query-runner/decorators/on-database-event.decorator'; 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 { DatabaseEventAction } from 'src/engine/api/graphql/graphql-query-runner/enums/database-event-action';
@Injectable() @Injectable()
@ -23,7 +23,7 @@ export class WorkspaceWorkspaceMemberListener {
private readonly messageQueueService: MessageQueueService, private readonly messageQueueService: MessageQueueService,
) {} ) {}
@OnDatabaseEvent('workspaceMember', DatabaseEventAction.UPDATED) @OnDatabaseBatchEvent('workspaceMember', DatabaseEventAction.UPDATED)
async handleUpdateEvent( async handleUpdateEvent(
payload: WorkspaceEventBatch< payload: WorkspaceEventBatch<
ObjectRecordUpdateEvent<WorkspaceMemberWorkspaceEntity> ObjectRecordUpdateEvent<WorkspaceMemberWorkspaceEntity>
@ -51,7 +51,7 @@ export class WorkspaceWorkspaceMemberListener {
); );
} }
@OnDatabaseEvent('workspaceMember', DatabaseEventAction.DELETED) @OnDatabaseBatchEvent('workspaceMember', DatabaseEventAction.DELETED)
async handleDeleteEvent( async handleDeleteEvent(
payload: WorkspaceEventBatch< payload: WorkspaceEventBatch<
ObjectRecordDeleteEvent<WorkspaceMemberWorkspaceEntity> ObjectRecordDeleteEvent<WorkspaceMemberWorkspaceEntity>

2
packages/twenty-server/src/engine/metadata-modules/serverless-function/constants/serverless-function-published.ts
View File

@ -1 +1 @@
export const SERVERLESS_FUNCTION_PUBLISHED = 'serverlessFunction.published'; export const SERVERLESS_FUNCTION_PUBLISHED = 'serverless_function_published';

12
packages/twenty-server/src/engine/metadata-modules/serverless-function/listeners/serverless-function-publication.listener.ts
View File

@ -1,5 +1,4 @@
import { Injectable } from '@nestjs/common'; import { Injectable } from '@nestjs/common';
import { OnEvent } from '@nestjs/event-emitter';
import { InjectRepository } from '@nestjs/typeorm'; import { InjectRepository } from '@nestjs/typeorm';
import { join } from 'path'; import { join } from 'path';
@ -10,8 +9,9 @@ import { INDEX_FILE_NAME } from 'src/engine/core-modules/serverless/drivers/cons
import { SERVERLESS_FUNCTION_PUBLISHED } from 'src/engine/metadata-modules/serverless-function/constants/serverless-function-published'; import { SERVERLESS_FUNCTION_PUBLISHED } from 'src/engine/metadata-modules/serverless-function/constants/serverless-function-published';
import { ServerlessFunctionEntity } from 'src/engine/metadata-modules/serverless-function/serverless-function.entity'; import { ServerlessFunctionEntity } from 'src/engine/metadata-modules/serverless-function/serverless-function.entity';
import { ServerlessFunctionService } from 'src/engine/metadata-modules/serverless-function/serverless-function.service'; import { ServerlessFunctionService } from 'src/engine/metadata-modules/serverless-function/serverless-function.service';
import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/workspace-event.type'; import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type';
import { CodeIntrospectionService } from 'src/modules/code-introspection/code-introspection.service'; import { CodeIntrospectionService } from 'src/modules/code-introspection/code-introspection.service';
import { OnCustomBatchEvent } from 'src/engine/api/graphql/graphql-query-runner/decorators/on-custom-batch-event.decorator';
@Injectable() @Injectable()
export class ServerlessFunctionPublicationListener { export class ServerlessFunctionPublicationListener {
@ -22,17 +22,17 @@ export class ServerlessFunctionPublicationListener {
private readonly serverlessFunctionRepository: Repository<ServerlessFunctionEntity>, private readonly serverlessFunctionRepository: Repository<ServerlessFunctionEntity>,
) {} ) {}
@OnEvent(SERVERLESS_FUNCTION_PUBLISHED) @OnCustomBatchEvent(SERVERLESS_FUNCTION_PUBLISHED)
async handle( async handle(
payload: WorkspaceEventBatch<{ batchEvent: WorkspaceEventBatch<{
serverlessFunctionId: string; serverlessFunctionId: string;
serverlessFunctionVersion: string; serverlessFunctionVersion: string;
}>, }>,
): Promise<void> { ): Promise<void> {
for (const event of payload.events) { for (const event of batchEvent.events) {
const sourceCode = const sourceCode =
await this.serverlessFunctionService.getServerlessFunctionSourceCode( await this.serverlessFunctionService.getServerlessFunctionSourceCode(
payload.workspaceId, batchEvent.workspaceId,
event.serverlessFunctionId, event.serverlessFunctionId,
event.serverlessFunctionVersion, event.serverlessFunctionVersion,
); );

2
packages/twenty-server/src/engine/metadata-modules/serverless-function/serverless-function.service.ts
View File

@ -194,7 +194,7 @@ export class ServerlessFunctionService {
}, },
); );
this.workspaceEventEmitter.emit( this.workspaceEventEmitter.emitCustomBatchEvent(
SERVERLESS_FUNCTION_PUBLISHED, SERVERLESS_FUNCTION_PUBLISHED,
[ [
{ {

1
packages/twenty-server/src/engine/workspace-event-emitter/types/custom-event-name.type.ts Normal file
View File

@ -0,0 +1 @@
export type CustomEventName = `${string}_${string}`;

0
packages/twenty-server/src/engine/workspace-event-emitter/workspace-event.type.ts → packages/twenty-server/src/engine/workspace-event-emitter/types/workspace-event.type.ts
View File

48
packages/twenty-server/src/engine/workspace-event-emitter/workspace-event-emitter.ts
View File

@ -1,21 +1,61 @@
import { Injectable } from '@nestjs/common'; import { Injectable } from '@nestjs/common';
import { EventEmitter2 } from '@nestjs/event-emitter'; import { EventEmitter2 } from '@nestjs/event-emitter';
import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/workspace-event.type'; 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 { 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 { DatabaseEventAction } from 'src/engine/api/graphql/graphql-query-runner/enums/database-event-action';
import { CustomEventName } from 'src/engine/workspace-event-emitter/types/custom-event-name.type';
type ActionEventMap<T> = {
[DatabaseEventAction.CREATED]: ObjectRecordCreateEvent<T>;
[DatabaseEventAction.UPDATED]: ObjectRecordUpdateEvent<T>;
[DatabaseEventAction.DELETED]: ObjectRecordDeleteEvent<T>;
[DatabaseEventAction.DESTROYED]: ObjectRecordDestroyEvent<T>;
};
@Injectable() @Injectable()
export class WorkspaceEventEmitter { export class WorkspaceEventEmitter {
constructor(private readonly eventEmitter: EventEmitter2) {} constructor(private readonly eventEmitter: EventEmitter2) {}
public emit(eventName: string, events: any[], workspaceId: string) { public emitDatabaseBatchEvent<T, A extends keyof ActionEventMap<T>>({
objectMetadataNameSingular,
action,
events,
workspaceId,
}: {
objectMetadataNameSingular: string;
action: A;
events: ActionEventMap<T>[A][];
workspaceId: string;
}) {
if (!events.length) { if (!events.length) {
return; return;
} }
return this.eventEmitter.emit(eventName, { const eventName = `${objectMetadataNameSingular}.${action}`;
this.eventEmitter.emit(eventName, {
name: eventName, name: eventName,
workspaceId, workspaceId,
events, events,
} satisfies WorkspaceEventBatch<any>); });
}
public emitCustomBatchEvent(
eventName: CustomEventName,
events: object[],
workspaceId: string,
) {
if (!events.length) {
return;
}
this.eventEmitter.emit(eventName, {
name: eventName,
workspaceId,
events,
});
} }
} }

2
packages/twenty-server/src/modules/calendar/blocklist-manager/jobs/blocklist-item-delete-calendar-events.job.ts
View File

@ -7,7 +7,7 @@ import { Process } from 'src/engine/core-modules/message-queue/decorators/proces
import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.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 { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
import { TwentyORMManager } from 'src/engine/twenty-orm/twenty-orm.manager'; import { TwentyORMManager } from 'src/engine/twenty-orm/twenty-orm.manager';
import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/workspace-event.type'; import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type';
import { BlocklistWorkspaceEntity } from 'src/modules/blocklist/standard-objects/blocklist.workspace-entity'; import { BlocklistWorkspaceEntity } from 'src/modules/blocklist/standard-objects/blocklist.workspace-entity';
import { CalendarEventCleanerService } from 'src/modules/calendar/calendar-event-cleaner/services/calendar-event-cleaner.service'; import { CalendarEventCleanerService } from 'src/modules/calendar/calendar-event-cleaner/services/calendar-event-cleaner.service';
import { CalendarChannelEventAssociationWorkspaceEntity } from 'src/modules/calendar/common/standard-objects/calendar-channel-event-association.workspace-entity'; import { CalendarChannelEventAssociationWorkspaceEntity } from 'src/modules/calendar/common/standard-objects/calendar-channel-event-association.workspace-entity';

2
packages/twenty-server/src/modules/calendar/blocklist-manager/jobs/blocklist-reimport-calendar-events.job.ts
View File

@ -7,7 +7,7 @@ import { Process } from 'src/engine/core-modules/message-queue/decorators/proces
import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.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 { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
import { TwentyORMManager } from 'src/engine/twenty-orm/twenty-orm.manager'; import { TwentyORMManager } from 'src/engine/twenty-orm/twenty-orm.manager';
import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/workspace-event.type'; import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type';
import { BlocklistWorkspaceEntity } from 'src/modules/blocklist/standard-objects/blocklist.workspace-entity'; import { BlocklistWorkspaceEntity } from 'src/modules/blocklist/standard-objects/blocklist.workspace-entity';
import { CalendarChannelSyncStatusService } from 'src/modules/calendar/common/services/calendar-channel-sync-status.service'; import { CalendarChannelSyncStatusService } from 'src/modules/calendar/common/services/calendar-channel-sync-status.service';
import { import {

10
packages/twenty-server/src/modules/calendar/blocklist-manager/listeners/calendar-blocklist.listener.ts
View File

@ -6,7 +6,7 @@ import { ObjectRecordUpdateEvent } from 'src/engine/core-modules/event-emitter/t
import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator'; 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 { 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 { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service';
import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/workspace-event.type'; import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type';
import { BlocklistWorkspaceEntity } from 'src/modules/blocklist/standard-objects/blocklist.workspace-entity'; import { BlocklistWorkspaceEntity } from 'src/modules/blocklist/standard-objects/blocklist.workspace-entity';
import { import {
BlocklistItemDeleteCalendarEventsJob, BlocklistItemDeleteCalendarEventsJob,
@ -16,7 +16,7 @@ import {
BlocklistReimportCalendarEventsJob, BlocklistReimportCalendarEventsJob,
BlocklistReimportCalendarEventsJobData, BlocklistReimportCalendarEventsJobData,
} from 'src/modules/calendar/blocklist-manager/jobs/blocklist-reimport-calendar-events.job'; } from 'src/modules/calendar/blocklist-manager/jobs/blocklist-reimport-calendar-events.job';
import { OnDatabaseEvent } from 'src/engine/api/graphql/graphql-query-runner/decorators/on-database-event.decorator'; 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 { DatabaseEventAction } from 'src/engine/api/graphql/graphql-query-runner/enums/database-event-action';
@Injectable() @Injectable()
@ -26,7 +26,7 @@ export class CalendarBlocklistListener {
private readonly messageQueueService: MessageQueueService, private readonly messageQueueService: MessageQueueService,
) {} ) {}
@OnDatabaseEvent('blocklist', DatabaseEventAction.CREATED) @OnDatabaseBatchEvent('blocklist', DatabaseEventAction.CREATED)
async handleCreatedEvent( async handleCreatedEvent(
payload: WorkspaceEventBatch< payload: WorkspaceEventBatch<
ObjectRecordCreateEvent<BlocklistWorkspaceEntity> ObjectRecordCreateEvent<BlocklistWorkspaceEntity>
@ -38,7 +38,7 @@ export class CalendarBlocklistListener {
); );
} }
@OnDatabaseEvent('blocklist', DatabaseEventAction.DELETED) @OnDatabaseBatchEvent('blocklist', DatabaseEventAction.DELETED)
async handleDeletedEvent( async handleDeletedEvent(
payload: WorkspaceEventBatch< payload: WorkspaceEventBatch<
ObjectRecordDeleteEvent<BlocklistWorkspaceEntity> ObjectRecordDeleteEvent<BlocklistWorkspaceEntity>
@ -50,7 +50,7 @@ export class CalendarBlocklistListener {
); );
} }
@OnDatabaseEvent('blocklist', DatabaseEventAction.UPDATED) @OnDatabaseBatchEvent('blocklist', DatabaseEventAction.UPDATED)
async handleUpdatedEvent( async handleUpdatedEvent(
payload: WorkspaceEventBatch< payload: WorkspaceEventBatch<
ObjectRecordUpdateEvent<BlocklistWorkspaceEntity> ObjectRecordUpdateEvent<BlocklistWorkspaceEntity>

6
packages/twenty-server/src/modules/calendar/calendar-event-cleaner/listeners/calendar-event-cleaner-connected-account.listener.ts
View File

@ -4,13 +4,13 @@ import { ObjectRecordDeleteEvent } from 'src/engine/core-modules/event-emitter/t
import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator'; 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 { 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 { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service';
import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/workspace-event.type'; import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type';
import { import {
DeleteConnectedAccountAssociatedCalendarDataJob, DeleteConnectedAccountAssociatedCalendarDataJob,
DeleteConnectedAccountAssociatedCalendarDataJobData, DeleteConnectedAccountAssociatedCalendarDataJobData,
} from 'src/modules/calendar/calendar-event-cleaner/jobs/delete-connected-account-associated-calendar-data.job'; } from 'src/modules/calendar/calendar-event-cleaner/jobs/delete-connected-account-associated-calendar-data.job';
import { ConnectedAccountWorkspaceEntity } from 'src/modules/connected-account/standard-objects/connected-account.workspace-entity'; import { ConnectedAccountWorkspaceEntity } from 'src/modules/connected-account/standard-objects/connected-account.workspace-entity';
import { OnDatabaseEvent } from 'src/engine/api/graphql/graphql-query-runner/decorators/on-database-event.decorator'; 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 { DatabaseEventAction } from 'src/engine/api/graphql/graphql-query-runner/enums/database-event-action';
@Injectable() @Injectable()
@ -20,7 +20,7 @@ export class CalendarEventCleanerConnectedAccountListener {
private readonly calendarQueueService: MessageQueueService, private readonly calendarQueueService: MessageQueueService,
) {} ) {}
@OnDatabaseEvent('connectedAccount', DatabaseEventAction.DESTROYED) @OnDatabaseBatchEvent('connectedAccount', DatabaseEventAction.DESTROYED)
async handleDestroyedEvent( async handleDestroyedEvent(
payload: WorkspaceEventBatch< payload: WorkspaceEventBatch<
ObjectRecordDeleteEvent<ConnectedAccountWorkspaceEntity> ObjectRecordDeleteEvent<ConnectedAccountWorkspaceEntity>

8
packages/twenty-server/src/modules/calendar/calendar-event-participant-manager/listeners/calendar-event-participant-person.listener.ts
View File

@ -6,7 +6,7 @@ import { objectRecordChangedProperties as objectRecordUpdateEventChangedProperti
import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator'; 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 { 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 { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service';
import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/workspace-event.type'; import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type';
import { import {
CalendarEventParticipantMatchParticipantJob, CalendarEventParticipantMatchParticipantJob,
CalendarEventParticipantMatchParticipantJobData, CalendarEventParticipantMatchParticipantJobData,
@ -16,7 +16,7 @@ import {
CalendarEventParticipantUnmatchParticipantJobData, CalendarEventParticipantUnmatchParticipantJobData,
} from 'src/modules/calendar/calendar-event-participant-manager/jobs/calendar-event-participant-unmatch-participant.job'; } from 'src/modules/calendar/calendar-event-participant-manager/jobs/calendar-event-participant-unmatch-participant.job';
import { PersonWorkspaceEntity } from 'src/modules/person/standard-objects/person.workspace-entity'; import { PersonWorkspaceEntity } from 'src/modules/person/standard-objects/person.workspace-entity';
import { OnDatabaseEvent } from 'src/engine/api/graphql/graphql-query-runner/decorators/on-database-event.decorator'; 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 { DatabaseEventAction } from 'src/engine/api/graphql/graphql-query-runner/enums/database-event-action';
@Injectable() @Injectable()
@ -26,7 +26,7 @@ export class CalendarEventParticipantPersonListener {
private readonly messageQueueService: MessageQueueService, private readonly messageQueueService: MessageQueueService,
) {} ) {}
@OnDatabaseEvent('person', DatabaseEventAction.CREATED) @OnDatabaseBatchEvent('person', DatabaseEventAction.CREATED)
async handleCreatedEvent( async handleCreatedEvent(
payload: WorkspaceEventBatch< payload: WorkspaceEventBatch<
ObjectRecordCreateEvent<PersonWorkspaceEntity> ObjectRecordCreateEvent<PersonWorkspaceEntity>
@ -49,7 +49,7 @@ export class CalendarEventParticipantPersonListener {
} }
} }
@OnDatabaseEvent('person', DatabaseEventAction.UPDATED) @OnDatabaseBatchEvent('person', DatabaseEventAction.UPDATED)
async handleUpdatedEvent( async handleUpdatedEvent(
payload: WorkspaceEventBatch< payload: WorkspaceEventBatch<
ObjectRecordUpdateEvent<PersonWorkspaceEntity> ObjectRecordUpdateEvent<PersonWorkspaceEntity>

8
packages/twenty-server/src/modules/calendar/calendar-event-participant-manager/listeners/calendar-event-participant-workspace-member.listener.ts
View File

@ -6,7 +6,7 @@ import { objectRecordChangedProperties as objectRecordUpdateEventChangedProperti
import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator'; 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 { 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 { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service';
import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/workspace-event.type'; import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type';
import { import {
CalendarEventParticipantMatchParticipantJob, CalendarEventParticipantMatchParticipantJob,
CalendarEventParticipantMatchParticipantJobData, CalendarEventParticipantMatchParticipantJobData,
@ -16,7 +16,7 @@ import {
CalendarEventParticipantUnmatchParticipantJobData, CalendarEventParticipantUnmatchParticipantJobData,
} from 'src/modules/calendar/calendar-event-participant-manager/jobs/calendar-event-participant-unmatch-participant.job'; } from 'src/modules/calendar/calendar-event-participant-manager/jobs/calendar-event-participant-unmatch-participant.job';
import { WorkspaceMemberWorkspaceEntity } from 'src/modules/workspace-member/standard-objects/workspace-member.workspace-entity'; import { WorkspaceMemberWorkspaceEntity } from 'src/modules/workspace-member/standard-objects/workspace-member.workspace-entity';
import { OnDatabaseEvent } from 'src/engine/api/graphql/graphql-query-runner/decorators/on-database-event.decorator'; 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 { DatabaseEventAction } from 'src/engine/api/graphql/graphql-query-runner/enums/database-event-action';
@Injectable() @Injectable()
@ -26,7 +26,7 @@ export class CalendarEventParticipantWorkspaceMemberListener {
private readonly messageQueueService: MessageQueueService, private readonly messageQueueService: MessageQueueService,
) {} ) {}
@OnDatabaseEvent('workspaceMember', DatabaseEventAction.CREATED) @OnDatabaseBatchEvent('workspaceMember', DatabaseEventAction.CREATED)
async handleCreatedEvent( async handleCreatedEvent(
payload: WorkspaceEventBatch< payload: WorkspaceEventBatch<
ObjectRecordCreateEvent<WorkspaceMemberWorkspaceEntity> ObjectRecordCreateEvent<WorkspaceMemberWorkspaceEntity>
@ -48,7 +48,7 @@ export class CalendarEventParticipantWorkspaceMemberListener {
} }
} }
@OnDatabaseEvent('workspaceMember', DatabaseEventAction.UPDATED) @OnDatabaseBatchEvent('workspaceMember', DatabaseEventAction.UPDATED)
async handleUpdatedEvent( async handleUpdatedEvent(
payload: WorkspaceEventBatch< payload: WorkspaceEventBatch<
ObjectRecordUpdateEvent<WorkspaceMemberWorkspaceEntity> ObjectRecordUpdateEvent<WorkspaceMemberWorkspaceEntity>

12
packages/twenty-server/src/modules/calendar/calendar-event-participant-manager/listeners/calendar-event-participant.listener.ts
View File

@ -1,5 +1,4 @@
import { Injectable } from '@nestjs/common'; import { Injectable } from '@nestjs/common';
import { OnEvent } from '@nestjs/event-emitter';
import { InjectRepository } from '@nestjs/typeorm'; import { InjectRepository } from '@nestjs/typeorm';
import { Repository } from 'typeorm'; import { Repository } from 'typeorm';
@ -7,10 +6,11 @@ import { Repository } from 'typeorm';
import { ObjectMetadataEntity } from 'src/engine/metadata-modules/object-metadata/object-metadata.entity'; import { ObjectMetadataEntity } from 'src/engine/metadata-modules/object-metadata/object-metadata.entity';
import { InjectObjectMetadataRepository } from 'src/engine/object-metadata-repository/object-metadata-repository.decorator'; import { InjectObjectMetadataRepository } from 'src/engine/object-metadata-repository/object-metadata-repository.decorator';
import { WorkspaceDataSourceService } from 'src/engine/workspace-datasource/workspace-datasource.service'; import { WorkspaceDataSourceService } from 'src/engine/workspace-datasource/workspace-datasource.service';
import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/workspace-event.type'; import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type';
import { CalendarEventParticipantWorkspaceEntity } from 'src/modules/calendar/common/standard-objects/calendar-event-participant.workspace-entity'; import { CalendarEventParticipantWorkspaceEntity } from 'src/modules/calendar/common/standard-objects/calendar-event-participant.workspace-entity';
import { TimelineActivityRepository } from 'src/modules/timeline/repositiories/timeline-activity.repository'; import { TimelineActivityRepository } from 'src/modules/timeline/repositiories/timeline-activity.repository';
import { TimelineActivityWorkspaceEntity } from 'src/modules/timeline/standard-objects/timeline-activity.workspace-entity'; import { TimelineActivityWorkspaceEntity } from 'src/modules/timeline/standard-objects/timeline-activity.workspace-entity';
import { OnCustomBatchEvent } from 'src/engine/api/graphql/graphql-query-runner/decorators/on-custom-batch-event.decorator';
@Injectable() @Injectable()
export class CalendarEventParticipantListener { export class CalendarEventParticipantListener {
@ -22,17 +22,17 @@ export class CalendarEventParticipantListener {
private readonly objectMetadataRepository: Repository<ObjectMetadataEntity>, private readonly objectMetadataRepository: Repository<ObjectMetadataEntity>,
) {} ) {}
@OnEvent('calendarEventParticipant.matched') @OnCustomBatchEvent('calendarEventParticipant_matched')
public async handleCalendarEventParticipantMatchedEvent( public async handleCalendarEventParticipantMatchedEvent(
payload: WorkspaceEventBatch<{ batchEvent: WorkspaceEventBatch<{
workspaceMemberId: string; workspaceMemberId: string;
participants: CalendarEventParticipantWorkspaceEntity[]; participants: CalendarEventParticipantWorkspaceEntity[];
}>, }>,
): Promise<void> { ): Promise<void> {
const workspaceId = payload.workspaceId; const workspaceId = batchEvent.workspaceId;
// TODO: Refactor to insertTimelineActivitiesForObject once // TODO: Refactor to insertTimelineActivitiesForObject once
for (const eventPayload of payload.events) { for (const eventPayload of batchEvent.events) {
const calendarEventParticipants = eventPayload.participants; const calendarEventParticipants = eventPayload.participants;
const workspaceMemberId = eventPayload.workspaceMemberId; const workspaceMemberId = eventPayload.workspaceMemberId;

6
packages/twenty-server/src/modules/connected-account/listeners/connected-account.listener.ts
View File

@ -2,11 +2,11 @@ import { Injectable } from '@nestjs/common';
import { ObjectRecordDeleteEvent } from 'src/engine/core-modules/event-emitter/types/object-record-delete.event'; import { ObjectRecordDeleteEvent } from 'src/engine/core-modules/event-emitter/types/object-record-delete.event';
import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager'; import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager';
import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/workspace-event.type'; import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type';
import { AccountsToReconnectService } from 'src/modules/connected-account/services/accounts-to-reconnect.service'; import { AccountsToReconnectService } from 'src/modules/connected-account/services/accounts-to-reconnect.service';
import { ConnectedAccountWorkspaceEntity } from 'src/modules/connected-account/standard-objects/connected-account.workspace-entity'; import { ConnectedAccountWorkspaceEntity } from 'src/modules/connected-account/standard-objects/connected-account.workspace-entity';
import { WorkspaceMemberWorkspaceEntity } from 'src/modules/workspace-member/standard-objects/workspace-member.workspace-entity'; import { WorkspaceMemberWorkspaceEntity } from 'src/modules/workspace-member/standard-objects/workspace-member.workspace-entity';
import { OnDatabaseEvent } from 'src/engine/api/graphql/graphql-query-runner/decorators/on-database-event.decorator'; 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 { DatabaseEventAction } from 'src/engine/api/graphql/graphql-query-runner/enums/database-event-action';
@Injectable() @Injectable()
@ -16,7 +16,7 @@ export class ConnectedAccountListener {
private readonly accountsToReconnectService: AccountsToReconnectService, private readonly accountsToReconnectService: AccountsToReconnectService,
) {} ) {}
@OnDatabaseEvent('connectedAccount', DatabaseEventAction.DESTROYED) @OnDatabaseBatchEvent('connectedAccount', DatabaseEventAction.DESTROYED)
async handleDestroyedEvent( async handleDestroyedEvent(
payload: WorkspaceEventBatch< payload: WorkspaceEventBatch<
ObjectRecordDeleteEvent<ConnectedAccountWorkspaceEntity> ObjectRecordDeleteEvent<ConnectedAccountWorkspaceEntity>

39
packages/twenty-server/src/modules/connected-account/query-hooks/connected-account-delete-one.pre-query.hook.ts
View File

@ -1,13 +1,17 @@
import { InjectRepository } from '@nestjs/typeorm';
import { Repository } from 'typeorm';
import { WorkspaceQueryHookInstance } from 'src/engine/api/graphql/workspace-query-runner/workspace-query-hook/interfaces/workspace-query-hook.interface'; import { WorkspaceQueryHookInstance } from 'src/engine/api/graphql/workspace-query-runner/workspace-query-hook/interfaces/workspace-query-hook.interface';
import { DeleteOneResolverArgs } from 'src/engine/api/graphql/workspace-resolver-builder/interfaces/workspace-resolvers-builder.interface'; import { DeleteOneResolverArgs } from 'src/engine/api/graphql/workspace-resolver-builder/interfaces/workspace-resolvers-builder.interface';
import { WorkspaceQueryHook } from 'src/engine/api/graphql/workspace-query-runner/workspace-query-hook/decorators/workspace-query-hook.decorator'; import { WorkspaceQueryHook } from 'src/engine/api/graphql/workspace-query-runner/workspace-query-hook/decorators/workspace-query-hook.decorator';
import { AuthContext } from 'src/engine/core-modules/auth/types/auth-context.type'; import { AuthContext } from 'src/engine/core-modules/auth/types/auth-context.type';
import { ObjectRecordDeleteEvent } from 'src/engine/core-modules/event-emitter/types/object-record-delete.event';
import { TwentyORMManager } from 'src/engine/twenty-orm/twenty-orm.manager'; import { TwentyORMManager } from 'src/engine/twenty-orm/twenty-orm.manager';
import { WorkspaceEventEmitter } from 'src/engine/workspace-event-emitter/workspace-event-emitter'; import { WorkspaceEventEmitter } from 'src/engine/workspace-event-emitter/workspace-event-emitter';
import { MessageChannelWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity'; import { MessageChannelWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity';
import { DatabaseEventAction } from 'src/engine/api/graphql/graphql-query-runner/enums/database-event-action'; import { DatabaseEventAction } from 'src/engine/api/graphql/graphql-query-runner/enums/database-event-action';
import { ObjectMetadataEntity } from 'src/engine/metadata-modules/object-metadata/object-metadata.entity';
@WorkspaceQueryHook(`connectedAccount.destroyOne`) @WorkspaceQueryHook(`connectedAccount.destroyOne`)
export class ConnectedAccountDeleteOnePreQueryHook export class ConnectedAccountDeleteOnePreQueryHook
@ -16,6 +20,8 @@ export class ConnectedAccountDeleteOnePreQueryHook
constructor( constructor(
private readonly twentyORMManager: TwentyORMManager, private readonly twentyORMManager: TwentyORMManager,
private readonly workspaceEventEmitter: WorkspaceEventEmitter, private readonly workspaceEventEmitter: WorkspaceEventEmitter,
@InjectRepository(ObjectMetadataEntity, 'metadata')
private readonly objectMetadataRepository: Repository<ObjectMetadataEntity>,
) {} ) {}
async execute( async execute(
@ -34,19 +40,24 @@ export class ConnectedAccountDeleteOnePreQueryHook
connectedAccountId, connectedAccountId,
}); });
this.workspaceEventEmitter.emit( const objectMetadata = await this.objectMetadataRepository.findOneOrFail({
`messageChannel.${DatabaseEventAction.DESTROYED}`, where: {
messageChannels.map( nameSingular: 'messageChannel',
(messageChannel) => },
({ });
recordId: messageChannel.id,
}) satisfies Pick< this.workspaceEventEmitter.emitDatabaseBatchEvent({
ObjectRecordDeleteEvent<MessageChannelWorkspaceEntity>, objectMetadataNameSingular: 'messageChannel',
'recordId' action: DatabaseEventAction.DESTROYED,
>, events: messageChannels.map((messageChannel) => ({
), recordId: messageChannel.id,
authContext.workspace.id, objectMetadata,
); properties: {
before: messageChannel,
},
})),
workspaceId: authContext.workspace.id,
});
return payload; return payload;
} }

7
packages/twenty-server/src/modules/connected-account/query-hooks/connected-account-query-hook.module.ts
View File

@ -1,9 +1,14 @@
import { Module } from '@nestjs/common'; import { Module } from '@nestjs/common';
import { NestjsQueryTypeOrmModule } from '@ptc-org/nestjs-query-typeorm';
import { ConnectedAccountDeleteOnePreQueryHook } from 'src/modules/connected-account/query-hooks/connected-account-delete-one.pre-query.hook'; import { ConnectedAccountDeleteOnePreQueryHook } from 'src/modules/connected-account/query-hooks/connected-account-delete-one.pre-query.hook';
import { ObjectMetadataEntity } from 'src/engine/metadata-modules/object-metadata/object-metadata.entity';
@Module({ @Module({
imports: [], imports: [
NestjsQueryTypeOrmModule.forFeature([ObjectMetadataEntity], 'metadata'),
],
providers: [ConnectedAccountDeleteOnePreQueryHook], providers: [ConnectedAccountDeleteOnePreQueryHook],
}) })
export class ConnectedAccountQueryHookModule {} export class ConnectedAccountQueryHookModule {}

6
packages/twenty-server/src/modules/contact-creation-manager/listeners/auto-companies-and-contacts-creation-calendar-channel.listener.ts
View File

@ -5,13 +5,13 @@ import { objectRecordChangedProperties } from 'src/engine/core-modules/event-emi
import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator'; 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 { 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 { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service';
import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/workspace-event.type'; import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type';
import { import {
CalendarCreateCompanyAndContactAfterSyncJob, CalendarCreateCompanyAndContactAfterSyncJob,
CalendarCreateCompanyAndContactAfterSyncJobData, CalendarCreateCompanyAndContactAfterSyncJobData,
} from 'src/modules/calendar/calendar-event-participant-manager/jobs/calendar-create-company-and-contact-after-sync.job'; } from 'src/modules/calendar/calendar-event-participant-manager/jobs/calendar-create-company-and-contact-after-sync.job';
import { MessageChannelWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity'; import { MessageChannelWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity';
import { OnDatabaseEvent } from 'src/engine/api/graphql/graphql-query-runner/decorators/on-database-event.decorator'; 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 { DatabaseEventAction } from 'src/engine/api/graphql/graphql-query-runner/enums/database-event-action';
@Injectable() @Injectable()
@ -21,7 +21,7 @@ export class AutoCompaniesAndContactsCreationCalendarChannelListener {
private readonly messageQueueService: MessageQueueService, private readonly messageQueueService: MessageQueueService,
) {} ) {}
@OnDatabaseEvent('calendarChannel', DatabaseEventAction.UPDATED) @OnDatabaseBatchEvent('calendarChannel', DatabaseEventAction.UPDATED)
async handleUpdatedEvent( async handleUpdatedEvent(
payload: WorkspaceEventBatch< payload: WorkspaceEventBatch<
ObjectRecordUpdateEvent<MessageChannelWorkspaceEntity> ObjectRecordUpdateEvent<MessageChannelWorkspaceEntity>

6
packages/twenty-server/src/modules/contact-creation-manager/listeners/auto-companies-and-contacts-creation-message-channel.listener.ts
View File

@ -5,13 +5,13 @@ import { objectRecordChangedProperties } from 'src/engine/core-modules/event-emi
import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator'; 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 { 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 { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service';
import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/workspace-event.type'; import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type';
import { MessageChannelWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity'; import { MessageChannelWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity';
import { import {
MessagingCreateCompanyAndContactAfterSyncJob, MessagingCreateCompanyAndContactAfterSyncJob,
MessagingCreateCompanyAndContactAfterSyncJobData, MessagingCreateCompanyAndContactAfterSyncJobData,
} from 'src/modules/messaging/message-participant-manager/jobs/messaging-create-company-and-contact-after-sync.job'; } from 'src/modules/messaging/message-participant-manager/jobs/messaging-create-company-and-contact-after-sync.job';
import { OnDatabaseEvent } from 'src/engine/api/graphql/graphql-query-runner/decorators/on-database-event.decorator'; 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 { DatabaseEventAction } from 'src/engine/api/graphql/graphql-query-runner/enums/database-event-action';
@Injectable() @Injectable()
@ -21,7 +21,7 @@ export class AutoCompaniesAndContactsCreationMessageChannelListener {
private readonly messageQueueService: MessageQueueService, private readonly messageQueueService: MessageQueueService,
) {} ) {}
@OnDatabaseEvent('messageChannel', DatabaseEventAction.UPDATED) @OnDatabaseBatchEvent('messageChannel', DatabaseEventAction.UPDATED)
async handleUpdatedEvent( async handleUpdatedEvent(
payload: WorkspaceEventBatch< payload: WorkspaceEventBatch<
ObjectRecordUpdateEvent<MessageChannelWorkspaceEntity> ObjectRecordUpdateEvent<MessageChannelWorkspaceEntity>

27
packages/twenty-server/src/modules/contact-creation-manager/services/create-company-and-contact.service.ts
View File

@ -5,7 +5,6 @@ import chunk from 'lodash.chunk';
import compact from 'lodash.compact'; import compact from 'lodash.compact';
import { Any, EntityManager, Repository } from 'typeorm'; import { Any, EntityManager, Repository } from 'typeorm';
import { ObjectRecordCreateEvent } from 'src/engine/core-modules/event-emitter/types/object-record-create.event';
import { FieldActorSource } from 'src/engine/metadata-modules/field-metadata/composite-types/actor.composite-type'; import { FieldActorSource } from 'src/engine/metadata-modules/field-metadata/composite-types/actor.composite-type';
import { FieldMetadataEntity } from 'src/engine/metadata-modules/field-metadata/field-metadata.entity'; import { FieldMetadataEntity } from 'src/engine/metadata-modules/field-metadata/field-metadata.entity';
import { ObjectMetadataEntity } from 'src/engine/metadata-modules/object-metadata/object-metadata.entity'; import { ObjectMetadataEntity } from 'src/engine/metadata-modules/object-metadata/object-metadata.entity';
@ -195,21 +194,19 @@ export class CreateCompanyAndContactService {
source, source,
); );
this.workspaceEventEmitter.emit( this.workspaceEventEmitter.emitDatabaseBatchEvent({
`person.${DatabaseEventAction.CREATED}`, objectMetadataNameSingular: 'person',
createdPeople.map( action: DatabaseEventAction.CREATED,
(createdPerson) => events: createdPeople.map((createdPerson) => ({
({ // Fix ' as string': TypeORM typing issue... id is always returned when using save
// FixMe: TypeORM typing issue... id is always returned when using save recordId: createdPerson.id as string,
recordId: createdPerson.id as string, objectMetadata,
objectMetadata, properties: {
properties: { after: createdPerson,
after: createdPerson, },
}, })),
}) satisfies ObjectRecordCreateEvent<any>,
),
workspaceId, workspaceId,
); });
} }
} }
} }

7
packages/twenty-server/src/modules/favorite-folder/listeners/favorite-folder.listener.ts
View File

@ -1,13 +1,14 @@
import { Injectable } from '@nestjs/common'; import { Injectable } from '@nestjs/common';
import { OnEvent } from '@nestjs/event-emitter';
import { ObjectRecordDeleteEvent } from 'src/engine/core-modules/event-emitter/types/object-record-delete.event'; import { ObjectRecordDeleteEvent } from 'src/engine/core-modules/event-emitter/types/object-record-delete.event';
import { FeatureFlagKey } from 'src/engine/core-modules/feature-flag/enums/feature-flag-key.enum'; import { FeatureFlagKey } from 'src/engine/core-modules/feature-flag/enums/feature-flag-key.enum';
import { FeatureFlagService } from 'src/engine/core-modules/feature-flag/services/feature-flag.service'; import { FeatureFlagService } from 'src/engine/core-modules/feature-flag/services/feature-flag.service';
import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager'; import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager';
import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/workspace-event.type'; import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type';
import { FavoriteFolderWorkspaceEntity } from 'src/modules/favorite-folder/standard-objects/favorite-folder.workspace-entity'; import { FavoriteFolderWorkspaceEntity } from 'src/modules/favorite-folder/standard-objects/favorite-folder.workspace-entity';
import { FavoriteWorkspaceEntity } from 'src/modules/favorite/standard-objects/favorite.workspace-entity'; import { FavoriteWorkspaceEntity } from 'src/modules/favorite/standard-objects/favorite.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() @Injectable()
export class FavoriteFolderDeletionListener { export class FavoriteFolderDeletionListener {
@ -16,7 +17,7 @@ export class FavoriteFolderDeletionListener {
private readonly featureFlagService: FeatureFlagService, private readonly featureFlagService: FeatureFlagService,
) {} ) {}
@OnEvent('favoriteFolder.deleted') @OnDatabaseBatchEvent('favoriteFolder', DatabaseEventAction.DELETED)
async handleDelete( async handleDelete(
payload: WorkspaceEventBatch< payload: WorkspaceEventBatch<
ObjectRecordDeleteEvent<FavoriteFolderWorkspaceEntity> ObjectRecordDeleteEvent<FavoriteFolderWorkspaceEntity>

10
packages/twenty-server/src/modules/match-participant/match-participant.service.ts
View File

@ -119,8 +119,8 @@ export class MatchParticipantService<
transactionManager, transactionManager,
); );
this.workspaceEventEmitter.emit( this.workspaceEventEmitter.emitCustomBatchEvent(
`${objectMetadataName}.matched`, `${objectMetadataName}_matched`,
[ [
{ {
workspaceMemberId: null, workspaceMemberId: null,
@ -174,12 +174,12 @@ export class MatchParticipantService<
}, },
}); });
this.workspaceEventEmitter.emit( this.workspaceEventEmitter.emitCustomBatchEvent(
`${objectMetadataName}.matched`, `${objectMetadataName}_matched`,
[ [
{ {
workspaceId, workspaceId,
name: `${objectMetadataName}.matched`, name: `${objectMetadataName}_matched`,
workspaceMemberId: null, workspaceMemberId: null,
participants: updatedParticipants, participants: updatedParticipants,
}, },

2
packages/twenty-server/src/modules/messaging/blocklist-manager/jobs/messaging-blocklist-item-delete-messages.job.ts
View File

@ -7,7 +7,7 @@ import { Process } from 'src/engine/core-modules/message-queue/decorators/proces
import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.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 { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
import { TwentyORMManager } from 'src/engine/twenty-orm/twenty-orm.manager'; import { TwentyORMManager } from 'src/engine/twenty-orm/twenty-orm.manager';
import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/workspace-event.type'; import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type';
import { BlocklistWorkspaceEntity } from 'src/modules/blocklist/standard-objects/blocklist.workspace-entity'; import { BlocklistWorkspaceEntity } from 'src/modules/blocklist/standard-objects/blocklist.workspace-entity';
import { MessageChannelMessageAssociationWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-channel-message-association.workspace-entity'; import { MessageChannelMessageAssociationWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-channel-message-association.workspace-entity';
import { MessageChannelWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity'; import { MessageChannelWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity';

2
packages/twenty-server/src/modules/messaging/blocklist-manager/jobs/messaging-blocklist-reimport-messages.job.ts
View File

@ -7,7 +7,7 @@ import { Process } from 'src/engine/core-modules/message-queue/decorators/proces
import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.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 { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
import { TwentyORMManager } from 'src/engine/twenty-orm/twenty-orm.manager'; import { TwentyORMManager } from 'src/engine/twenty-orm/twenty-orm.manager';
import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/workspace-event.type'; import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type';
import { BlocklistWorkspaceEntity } from 'src/modules/blocklist/standard-objects/blocklist.workspace-entity'; import { BlocklistWorkspaceEntity } from 'src/modules/blocklist/standard-objects/blocklist.workspace-entity';
import { MessageChannelSyncStatusService } from 'src/modules/messaging/common/services/message-channel-sync-status.service'; import { MessageChannelSyncStatusService } from 'src/modules/messaging/common/services/message-channel-sync-status.service';
import { import {

10
packages/twenty-server/src/modules/messaging/blocklist-manager/listeners/messaging-blocklist.listener.ts
View File

@ -6,7 +6,7 @@ import { ObjectRecordUpdateEvent } from 'src/engine/core-modules/event-emitter/t
import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator'; 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 { 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 { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service';
import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/workspace-event.type'; import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type';
import { BlocklistWorkspaceEntity } from 'src/modules/blocklist/standard-objects/blocklist.workspace-entity'; import { BlocklistWorkspaceEntity } from 'src/modules/blocklist/standard-objects/blocklist.workspace-entity';
import { import {
BlocklistItemDeleteMessagesJob, BlocklistItemDeleteMessagesJob,
@ -16,7 +16,7 @@ import {
BlocklistReimportMessagesJob, BlocklistReimportMessagesJob,
BlocklistReimportMessagesJobData, BlocklistReimportMessagesJobData,
} from 'src/modules/messaging/blocklist-manager/jobs/messaging-blocklist-reimport-messages.job'; } from 'src/modules/messaging/blocklist-manager/jobs/messaging-blocklist-reimport-messages.job';
import { OnDatabaseEvent } from 'src/engine/api/graphql/graphql-query-runner/decorators/on-database-event.decorator'; 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 { DatabaseEventAction } from 'src/engine/api/graphql/graphql-query-runner/enums/database-event-action';
@Injectable({ scope: Scope.REQUEST }) @Injectable({ scope: Scope.REQUEST })
@ -26,7 +26,7 @@ export class MessagingBlocklistListener {
private readonly messageQueueService: MessageQueueService, private readonly messageQueueService: MessageQueueService,
) {} ) {}
@OnDatabaseEvent('blocklist', DatabaseEventAction.CREATED) @OnDatabaseBatchEvent('blocklist', DatabaseEventAction.CREATED)
async handleCreatedEvent( async handleCreatedEvent(
payload: WorkspaceEventBatch< payload: WorkspaceEventBatch<
ObjectRecordCreateEvent<BlocklistWorkspaceEntity> ObjectRecordCreateEvent<BlocklistWorkspaceEntity>
@ -38,7 +38,7 @@ export class MessagingBlocklistListener {
); );
} }
@OnDatabaseEvent('blocklist', DatabaseEventAction.CREATED) @OnDatabaseBatchEvent('blocklist', DatabaseEventAction.CREATED)
async handleDeletedEvent( async handleDeletedEvent(
payload: WorkspaceEventBatch< payload: WorkspaceEventBatch<
ObjectRecordDeleteEvent<BlocklistWorkspaceEntity> ObjectRecordDeleteEvent<BlocklistWorkspaceEntity>
@ -50,7 +50,7 @@ export class MessagingBlocklistListener {
); );
} }
@OnDatabaseEvent('blocklist', DatabaseEventAction.UPDATED) @OnDatabaseBatchEvent('blocklist', DatabaseEventAction.UPDATED)
async handleUpdatedEvent( async handleUpdatedEvent(
payload: WorkspaceEventBatch< payload: WorkspaceEventBatch<
ObjectRecordUpdateEvent<BlocklistWorkspaceEntity> ObjectRecordUpdateEvent<BlocklistWorkspaceEntity>

6
packages/twenty-server/src/modules/messaging/message-cleaner/listeners/messaging-message-cleaner-connected-account.listener.ts
View File

@ -4,13 +4,13 @@ import { ObjectRecordDeleteEvent } from 'src/engine/core-modules/event-emitter/t
import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator'; 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 { 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 { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service';
import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/workspace-event.type'; import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type';
import { ConnectedAccountWorkspaceEntity } from 'src/modules/connected-account/standard-objects/connected-account.workspace-entity'; import { ConnectedAccountWorkspaceEntity } from 'src/modules/connected-account/standard-objects/connected-account.workspace-entity';
import { import {
MessagingConnectedAccountDeletionCleanupJob, MessagingConnectedAccountDeletionCleanupJob,
MessagingConnectedAccountDeletionCleanupJobData, MessagingConnectedAccountDeletionCleanupJobData,
} from 'src/modules/messaging/message-cleaner/jobs/messaging-connected-account-deletion-cleanup.job'; } from 'src/modules/messaging/message-cleaner/jobs/messaging-connected-account-deletion-cleanup.job';
import { OnDatabaseEvent } from 'src/engine/api/graphql/graphql-query-runner/decorators/on-database-event.decorator'; 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 { DatabaseEventAction } from 'src/engine/api/graphql/graphql-query-runner/enums/database-event-action';
@Injectable() @Injectable()
@ -20,7 +20,7 @@ export class MessagingMessageCleanerConnectedAccountListener {
private readonly messageQueueService: MessageQueueService, private readonly messageQueueService: MessageQueueService,
) {} ) {}
@OnDatabaseEvent('connectedAccount', DatabaseEventAction.DESTROYED) @OnDatabaseBatchEvent('connectedAccount', DatabaseEventAction.DESTROYED)
async handleDestroyedEvent( async handleDestroyedEvent(
payload: WorkspaceEventBatch< payload: WorkspaceEventBatch<
ObjectRecordDeleteEvent<ConnectedAccountWorkspaceEntity> ObjectRecordDeleteEvent<ConnectedAccountWorkspaceEntity>

6
packages/twenty-server/src/modules/messaging/message-import-manager/listeners/messaging-import-manager-message-channel.listener.ts
View File

@ -4,13 +4,13 @@ import { ObjectRecordDeleteEvent } from 'src/engine/core-modules/event-emitter/t
import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator'; 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 { 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 { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service';
import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/workspace-event.type'; import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type';
import { MessageChannelWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity'; import { MessageChannelWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-channel.workspace-entity';
import { import {
MessagingCleanCacheJob, MessagingCleanCacheJob,
MessagingCleanCacheJobData, MessagingCleanCacheJobData,
} from 'src/modules/messaging/message-import-manager/jobs/messaging-clean-cache'; } from 'src/modules/messaging/message-import-manager/jobs/messaging-clean-cache';
import { OnDatabaseEvent } from 'src/engine/api/graphql/graphql-query-runner/decorators/on-database-event.decorator'; 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 { DatabaseEventAction } from 'src/engine/api/graphql/graphql-query-runner/enums/database-event-action';
@Injectable() @Injectable()
@ -20,7 +20,7 @@ export class MessagingMessageImportManagerMessageChannelListener {
private readonly messageQueueService: MessageQueueService, private readonly messageQueueService: MessageQueueService,
) {} ) {}
@OnDatabaseEvent('messageChannel', DatabaseEventAction.DESTROYED) @OnDatabaseBatchEvent('messageChannel', DatabaseEventAction.DESTROYED)
async handleDestroyedEvent( async handleDestroyedEvent(
payload: WorkspaceEventBatch< payload: WorkspaceEventBatch<
ObjectRecordDeleteEvent<MessageChannelWorkspaceEntity> ObjectRecordDeleteEvent<MessageChannelWorkspaceEntity>

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

@ -6,7 +6,7 @@ import { objectRecordChangedProperties as objectRecordUpdateEventChangedProperti
import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator'; 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 { 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 { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service';
import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/workspace-event.type'; import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type';
import { import {
MessageParticipantMatchParticipantJob, MessageParticipantMatchParticipantJob,
MessageParticipantMatchParticipantJobData, MessageParticipantMatchParticipantJobData,
@ -16,7 +16,7 @@ import {
MessageParticipantUnmatchParticipantJobData, MessageParticipantUnmatchParticipantJobData,
} from 'src/modules/messaging/message-participant-manager/jobs/message-participant-unmatch-participant.job'; } 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 { PersonWorkspaceEntity } from 'src/modules/person/standard-objects/person.workspace-entity';
import { OnDatabaseEvent } from 'src/engine/api/graphql/graphql-query-runner/decorators/on-database-event.decorator'; 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 { DatabaseEventAction } from 'src/engine/api/graphql/graphql-query-runner/enums/database-event-action';
@Injectable() @Injectable()
@ -26,7 +26,7 @@ export class MessageParticipantPersonListener {
private readonly messageQueueService: MessageQueueService, private readonly messageQueueService: MessageQueueService,
) {} ) {}
@OnDatabaseEvent('person', DatabaseEventAction.CREATED) @OnDatabaseBatchEvent('person', DatabaseEventAction.CREATED)
async handleCreatedEvent( async handleCreatedEvent(
payload: WorkspaceEventBatch< payload: WorkspaceEventBatch<
ObjectRecordCreateEvent<PersonWorkspaceEntity> ObjectRecordCreateEvent<PersonWorkspaceEntity>
@ -48,7 +48,7 @@ export class MessageParticipantPersonListener {
} }
} }
@OnDatabaseEvent('person', DatabaseEventAction.UPDATED) @OnDatabaseBatchEvent('person', DatabaseEventAction.UPDATED)
async handleUpdatedEvent( async handleUpdatedEvent(
payload: WorkspaceEventBatch< payload: WorkspaceEventBatch<
ObjectRecordUpdateEvent<PersonWorkspaceEntity> ObjectRecordUpdateEvent<PersonWorkspaceEntity>

8
packages/twenty-server/src/modules/messaging/message-participant-manager/listeners/message-participant-workspace-member.listener.ts
View File

@ -13,7 +13,7 @@ import { objectRecordChangedProperties as objectRecordUpdateEventChangedProperti
import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator'; 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 { 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 { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service';
import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/workspace-event.type'; import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type';
import { import {
MessageParticipantMatchParticipantJob, MessageParticipantMatchParticipantJob,
MessageParticipantMatchParticipantJobData, MessageParticipantMatchParticipantJobData,
@ -23,7 +23,7 @@ import {
MessageParticipantUnmatchParticipantJobData, MessageParticipantUnmatchParticipantJobData,
} from 'src/modules/messaging/message-participant-manager/jobs/message-participant-unmatch-participant.job'; } from 'src/modules/messaging/message-participant-manager/jobs/message-participant-unmatch-participant.job';
import { WorkspaceMemberWorkspaceEntity } from 'src/modules/workspace-member/standard-objects/workspace-member.workspace-entity'; import { WorkspaceMemberWorkspaceEntity } from 'src/modules/workspace-member/standard-objects/workspace-member.workspace-entity';
import { OnDatabaseEvent } from 'src/engine/api/graphql/graphql-query-runner/decorators/on-database-event.decorator'; 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 { DatabaseEventAction } from 'src/engine/api/graphql/graphql-query-runner/enums/database-event-action';
@Injectable() @Injectable()
@ -35,7 +35,7 @@ export class MessageParticipantWorkspaceMemberListener {
private readonly workspaceRepository: Repository<Workspace>, private readonly workspaceRepository: Repository<Workspace>,
) {} ) {}
@OnDatabaseEvent('workspaceMember', DatabaseEventAction.CREATED) @OnDatabaseBatchEvent('workspaceMember', DatabaseEventAction.CREATED)
async handleCreatedEvent( async handleCreatedEvent(
payload: WorkspaceEventBatch< payload: WorkspaceEventBatch<
ObjectRecordCreateEvent<WorkspaceMemberWorkspaceEntity> ObjectRecordCreateEvent<WorkspaceMemberWorkspaceEntity>
@ -68,7 +68,7 @@ export class MessageParticipantWorkspaceMemberListener {
} }
} }
@OnDatabaseEvent('workspaceMember', DatabaseEventAction.UPDATED) @OnDatabaseBatchEvent('workspaceMember', DatabaseEventAction.UPDATED)
async handleUpdatedEvent( async handleUpdatedEvent(
payload: WorkspaceEventBatch< payload: WorkspaceEventBatch<
ObjectRecordUpdateEvent<WorkspaceMemberWorkspaceEntity> ObjectRecordUpdateEvent<WorkspaceMemberWorkspaceEntity>

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

@ -1,5 +1,4 @@
import { Injectable } from '@nestjs/common'; import { Injectable } from '@nestjs/common';
import { OnEvent } from '@nestjs/event-emitter';
import { InjectRepository } from '@nestjs/typeorm'; import { InjectRepository } from '@nestjs/typeorm';
import { Repository } from 'typeorm'; import { Repository } from 'typeorm';
@ -7,10 +6,11 @@ import { Repository } from 'typeorm';
import { ObjectMetadataEntity } from 'src/engine/metadata-modules/object-metadata/object-metadata.entity'; import { ObjectMetadataEntity } from 'src/engine/metadata-modules/object-metadata/object-metadata.entity';
import { InjectObjectMetadataRepository } from 'src/engine/object-metadata-repository/object-metadata-repository.decorator'; import { InjectObjectMetadataRepository } from 'src/engine/object-metadata-repository/object-metadata-repository.decorator';
import { WorkspaceDataSourceService } from 'src/engine/workspace-datasource/workspace-datasource.service'; import { WorkspaceDataSourceService } from 'src/engine/workspace-datasource/workspace-datasource.service';
import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/workspace-event.type'; import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type';
import { MessageParticipantWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-participant.workspace-entity'; import { MessageParticipantWorkspaceEntity } from 'src/modules/messaging/common/standard-objects/message-participant.workspace-entity';
import { TimelineActivityRepository } from 'src/modules/timeline/repositiories/timeline-activity.repository'; import { TimelineActivityRepository } from 'src/modules/timeline/repositiories/timeline-activity.repository';
import { TimelineActivityWorkspaceEntity } from 'src/modules/timeline/standard-objects/timeline-activity.workspace-entity'; import { TimelineActivityWorkspaceEntity } from 'src/modules/timeline/standard-objects/timeline-activity.workspace-entity';
import { OnCustomBatchEvent } from 'src/engine/api/graphql/graphql-query-runner/decorators/on-custom-batch-event.decorator';
@Injectable() @Injectable()
export class MessageParticipantListener { export class MessageParticipantListener {
@ -22,28 +22,28 @@ export class MessageParticipantListener {
private readonly objectMetadataRepository: Repository<ObjectMetadataEntity>, private readonly objectMetadataRepository: Repository<ObjectMetadataEntity>,
) {} ) {}
@OnEvent('messageParticipant.matched') @OnCustomBatchEvent('messageParticipant_matched')
public async handleMessageParticipantMatched( public async handleMessageParticipantMatched(
payload: WorkspaceEventBatch<{ batchEvent: WorkspaceEventBatch<{
workspaceMemberId: string; workspaceMemberId: string;
participants: MessageParticipantWorkspaceEntity[]; participants: MessageParticipantWorkspaceEntity[];
}>, }>,
): Promise<void> { ): Promise<void> {
// TODO: Refactor to insertTimelineActivitiesForObject once // TODO: Refactor to insertTimelineActivitiesForObject once
for (const eventPayload of payload.events) { for (const eventPayload of batchEvent.events) {
const messageParticipants = eventPayload.participants ?? []; const messageParticipants = eventPayload.participants ?? [];
// TODO: move to a job? // TODO: move to a job?
const dataSourceSchema = this.workspaceDataSourceService.getSchemaName( const dataSourceSchema = this.workspaceDataSourceService.getSchemaName(
payload.workspaceId, batchEvent.workspaceId,
); );
const messageObjectMetadata = const messageObjectMetadata =
await this.objectMetadataRepository.findOneOrFail({ await this.objectMetadataRepository.findOneOrFail({
where: { where: {
nameSingular: 'message', nameSingular: 'message',
workspaceId: payload.workspaceId, workspaceId: batchEvent.workspaceId,
}, },
}); });
@ -64,12 +64,12 @@ export class MessageParticipantListener {
objectName: 'message', objectName: 'message',
recordId: participant.personId, recordId: participant.personId,
workspaceMemberId: eventPayload.workspaceMemberId, workspaceMemberId: eventPayload.workspaceMemberId,
workspaceId: payload.workspaceId, workspaceId: batchEvent.workspaceId,
linkedObjectMetadataId: messageObjectMetadata.id, linkedObjectMetadataId: messageObjectMetadata.id,
linkedRecordId: participant.messageId, linkedRecordId: participant.messageId,
linkedRecordCachedName: '', linkedRecordCachedName: '',
})), })),
payload.workspaceId, batchEvent.workspaceId,
); );
} }
} }

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

@ -3,7 +3,7 @@ import { Process } from 'src/engine/core-modules/message-queue/decorators/proces
import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.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 { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
import { InjectObjectMetadataRepository } from 'src/engine/object-metadata-repository/object-metadata-repository.decorator'; import { InjectObjectMetadataRepository } from 'src/engine/object-metadata-repository/object-metadata-repository.decorator';
import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/workspace-event.type'; import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type';
import { AuditLogRepository } from 'src/modules/timeline/repositiories/audit-log.repository'; import { AuditLogRepository } from 'src/modules/timeline/repositiories/audit-log.repository';
import { AuditLogWorkspaceEntity } from 'src/modules/timeline/standard-objects/audit-log.workspace-entity'; import { AuditLogWorkspaceEntity } from 'src/modules/timeline/standard-objects/audit-log.workspace-entity';
import { WorkspaceMemberRepository } from 'src/modules/workspace-member/repositories/workspace-member.repository'; import { WorkspaceMemberRepository } from 'src/modules/workspace-member/repositories/workspace-member.repository';

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

@ -3,10 +3,11 @@ import { Process } from 'src/engine/core-modules/message-queue/decorators/proces
import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.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 { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
import { InjectObjectMetadataRepository } from 'src/engine/object-metadata-repository/object-metadata-repository.decorator'; import { InjectObjectMetadataRepository } from 'src/engine/object-metadata-repository/object-metadata-repository.decorator';
import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/workspace-event.type'; import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type';
import { TimelineActivityService } from 'src/modules/timeline/services/timeline-activity.service'; import { TimelineActivityService } from 'src/modules/timeline/services/timeline-activity.service';
import { WorkspaceMemberRepository } from 'src/modules/workspace-member/repositories/workspace-member.repository'; 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 { 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) @Processor(MessageQueue.entityEventsToDbQueue)
export class UpsertTimelineActivityFromInternalEvent { export class UpsertTimelineActivityFromInternalEvent {
@ -18,7 +19,9 @@ export class UpsertTimelineActivityFromInternalEvent {
@Process(UpsertTimelineActivityFromInternalEvent.name) @Process(UpsertTimelineActivityFromInternalEvent.name)
async handle( async handle(
workspaceEventBatch: WorkspaceEventBatch<ObjectRecordBaseEvent>, workspaceEventBatch: WorkspaceEventBatch<
ObjectRecordBaseEvent<TimelineActivityWorkspaceEntity>
>,
): Promise<void> { ): Promise<void> {
for (const eventData of workspaceEventBatch.events) { for (const eventData of workspaceEventBatch.events) {
if (eventData.userId) { if (eventData.userId) {

2
packages/twenty-server/src/modules/timeline/repositiories/audit-log.repository.ts
View File

@ -10,7 +10,7 @@ export class AuditLogRepository {
public async insert( public async insert(
name: string, name: string,
properties: string, properties: object | null,
workspaceMemberId: string | null, workspaceMemberId: string | null,
objectName: string, objectName: string,
objectMetadataId: string, objectMetadataId: string,

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

@ -6,13 +6,14 @@ import { WorkspaceDataSourceService } from 'src/engine/workspace-datasource/work
import { TimelineActivityRepository } from 'src/modules/timeline/repositiories/timeline-activity.repository'; import { TimelineActivityRepository } from 'src/modules/timeline/repositiories/timeline-activity.repository';
import { TimelineActivityWorkspaceEntity } from 'src/modules/timeline/standard-objects/timeline-activity.workspace-entity'; import { TimelineActivityWorkspaceEntity } from 'src/modules/timeline/standard-objects/timeline-activity.workspace-entity';
type TimelineActivity = ObjectRecordBaseEvent & { type TimelineActivity =
name: string; ObjectRecordBaseEvent<TimelineActivityWorkspaceEntity> & {
objectName?: string; name: string;
linkedRecordCachedName?: string; objectName?: string;
linkedRecordId?: string; linkedRecordCachedName?: string;
linkedObjectMetadataId?: string; linkedRecordId?: string;
}; linkedObjectMetadataId?: string;
};
@Injectable() @Injectable()
export class TimelineActivityService { export class TimelineActivityService {
@ -32,7 +33,7 @@ export class TimelineActivityService {
eventName, eventName,
workspaceId, workspaceId,
}: { }: {
event: ObjectRecordBaseEvent; event: ObjectRecordBaseEvent<TimelineActivityWorkspaceEntity>;
eventName: string; eventName: string;
workspaceId: string; workspaceId: string;
}) { }) {
@ -64,7 +65,7 @@ export class TimelineActivityService {
workspaceId, workspaceId,
eventName, eventName,
}: { }: {
event: ObjectRecordBaseEvent; event: ObjectRecordBaseEvent<TimelineActivityWorkspaceEntity>;
workspaceId: string; workspaceId: string;
eventName: string; eventName: string;
}): Promise<TimelineActivity[] | undefined> { }): Promise<TimelineActivity[] | undefined> {
@ -100,7 +101,7 @@ export class TimelineActivityService {
workspaceId, workspaceId,
eventName, eventName,
}: { }: {
event: ObjectRecordBaseEvent; event: ObjectRecordBaseEvent<TimelineActivityWorkspaceEntity>;
workspaceId: string; workspaceId: string;
eventName: string; eventName: string;
}): Promise<TimelineActivity[] | undefined> { }): Promise<TimelineActivity[] | undefined> {
@ -145,7 +146,7 @@ export class TimelineActivityService {
eventName, eventName,
workspaceId, workspaceId,
}: { }: {
event: ObjectRecordBaseEvent; event: ObjectRecordBaseEvent<TimelineActivityWorkspaceEntity>;
dataSourceSchema: string; dataSourceSchema: string;
activityType: string; activityType: string;
eventName: string; eventName: string;
@ -206,7 +207,7 @@ export class TimelineActivityService {
eventName, eventName,
workspaceId, workspaceId,
}: { }: {
event: ObjectRecordBaseEvent; event: ObjectRecordBaseEvent<TimelineActivityWorkspaceEntity>;
dataSourceSchema: string; dataSourceSchema: string;
activityType: string; activityType: string;
eventName: string; eventName: string;

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

@ -10,7 +10,7 @@ import { MessageQueueService } from 'src/engine/core-modules/message-queue/servi
import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager'; import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager';
import { WebhookWorkspaceEntity } from 'src/modules/webhook/standard-objects/webhook.workspace-entity'; 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 { ObjectRecordBaseEvent } from 'src/engine/core-modules/event-emitter/types/object-record.base.event';
import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/workspace-event.type'; import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type';
import { import {
CallWebhookJob, CallWebhookJob,
CallWebhookJobData, CallWebhookJobData,

14
packages/twenty-server/src/modules/workflow/common/query-hooks/workflow-create-many.post-query.hook.ts
View File

@ -7,7 +7,6 @@ import { WorkspaceQueryPostHookInstance } from 'src/engine/api/graphql/workspace
import { WorkspaceQueryHook } from 'src/engine/api/graphql/workspace-query-runner/workspace-query-hook/decorators/workspace-query-hook.decorator'; import { WorkspaceQueryHook } from 'src/engine/api/graphql/workspace-query-runner/workspace-query-hook/decorators/workspace-query-hook.decorator';
import { WorkspaceQueryHookType } from 'src/engine/api/graphql/workspace-query-runner/workspace-query-hook/types/workspace-query-hook.type'; import { WorkspaceQueryHookType } from 'src/engine/api/graphql/workspace-query-runner/workspace-query-hook/types/workspace-query-hook.type';
import { AuthContext } from 'src/engine/core-modules/auth/types/auth-context.type'; import { AuthContext } from 'src/engine/core-modules/auth/types/auth-context.type';
import { ObjectRecordCreateEvent } from 'src/engine/core-modules/event-emitter/types/object-record-create.event';
import { ObjectMetadataEntity } from 'src/engine/metadata-modules/object-metadata/object-metadata.entity'; import { ObjectMetadataEntity } from 'src/engine/metadata-modules/object-metadata/object-metadata.entity';
import { TwentyORMManager } from 'src/engine/twenty-orm/twenty-orm.manager'; import { TwentyORMManager } from 'src/engine/twenty-orm/twenty-orm.manager';
import { WorkspaceEventEmitter } from 'src/engine/workspace-event-emitter/workspace-event-emitter'; import { WorkspaceEventEmitter } from 'src/engine/workspace-event-emitter/workspace-event-emitter';
@ -62,9 +61,10 @@ export class WorkflowCreateManyPostQueryHook
}, },
}); });
this.workspaceEventEmitter.emit( this.workspaceEventEmitter.emitDatabaseBatchEvent({
`workflowVersion.${DatabaseEventAction.CREATED}`, objectMetadataNameSingular: 'workflowVersion',
workflowVersionsToCreate.map((workflowVersionToCreate) => { action: DatabaseEventAction.CREATED,
events: workflowVersionsToCreate.map((workflowVersionToCreate) => {
return { return {
userId: authContext.user?.id, userId: authContext.user?.id,
recordId: workflowVersionToCreate.id, recordId: workflowVersionToCreate.id,
@ -72,9 +72,9 @@ export class WorkflowCreateManyPostQueryHook
properties: { properties: {
after: workflowVersionToCreate, after: workflowVersionToCreate,
}, },
} satisfies ObjectRecordCreateEvent<any>; };
}), }),
authContext.workspace.id, workspaceId: authContext.workspace.id,
); });
} }
} }

14
packages/twenty-server/src/modules/workflow/common/query-hooks/workflow-create-one.post-query.hook.ts
View File

@ -7,7 +7,6 @@ import { WorkspaceQueryPostHookInstance } from 'src/engine/api/graphql/workspace
import { WorkspaceQueryHook } from 'src/engine/api/graphql/workspace-query-runner/workspace-query-hook/decorators/workspace-query-hook.decorator'; import { WorkspaceQueryHook } from 'src/engine/api/graphql/workspace-query-runner/workspace-query-hook/decorators/workspace-query-hook.decorator';
import { WorkspaceQueryHookType } from 'src/engine/api/graphql/workspace-query-runner/workspace-query-hook/types/workspace-query-hook.type'; import { WorkspaceQueryHookType } from 'src/engine/api/graphql/workspace-query-runner/workspace-query-hook/types/workspace-query-hook.type';
import { AuthContext } from 'src/engine/core-modules/auth/types/auth-context.type'; import { AuthContext } from 'src/engine/core-modules/auth/types/auth-context.type';
import { ObjectRecordCreateEvent } from 'src/engine/core-modules/event-emitter/types/object-record-create.event';
import { ObjectMetadataEntity } from 'src/engine/metadata-modules/object-metadata/object-metadata.entity'; import { ObjectMetadataEntity } from 'src/engine/metadata-modules/object-metadata/object-metadata.entity';
import { TwentyORMManager } from 'src/engine/twenty-orm/twenty-orm.manager'; import { TwentyORMManager } from 'src/engine/twenty-orm/twenty-orm.manager';
import { WorkspaceEventEmitter } from 'src/engine/workspace-event-emitter/workspace-event-emitter'; import { WorkspaceEventEmitter } from 'src/engine/workspace-event-emitter/workspace-event-emitter';
@ -58,9 +57,10 @@ export class WorkflowCreateOnePostQueryHook
}, },
}); });
this.workspaceEventEmitter.emit( this.workspaceEventEmitter.emitDatabaseBatchEvent({
`workflowVersion.${DatabaseEventAction.CREATED}`, objectMetadataNameSingular: 'workflowVersion',
[ action: DatabaseEventAction.CREATED,
events: [
{ {
userId: authContext.user?.id, userId: authContext.user?.id,
recordId: workflowVersionToCreate.id, recordId: workflowVersionToCreate.id,
@ -68,9 +68,9 @@ export class WorkflowCreateOnePostQueryHook
properties: { properties: {
after: workflowVersionToCreate, after: workflowVersionToCreate,
}, },
} satisfies ObjectRecordCreateEvent<any>, },
], ],
authContext.workspace.id, workspaceId: authContext.workspace.id,
); });
} }
} }

2
packages/twenty-server/src/modules/workflow/workflow-status/constants/workflow-version-status-updated.constants.ts Normal file
View File

@ -0,0 +1,2 @@
export const WORKFLOW_VERSION_STATUS_UPDATED =
'workflow_version_status_updated';

30
packages/twenty-server/src/modules/workflow/workflow-status/listeners/workflow-version-status.listener.ts
View File

@ -5,7 +5,7 @@ import { ObjectRecordDeleteEvent } from 'src/engine/core-modules/event-emitter/t
import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator'; 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 { 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 { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service';
import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/workspace-event.type'; import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type';
import { import {
WorkflowVersionStatus, WorkflowVersionStatus,
WorkflowVersionWorkspaceEntity, WorkflowVersionWorkspaceEntity,
@ -16,8 +16,10 @@ import {
WorkflowVersionEventType, WorkflowVersionEventType,
WorkflowVersionStatusUpdate, WorkflowVersionStatusUpdate,
} from 'src/modules/workflow/workflow-status/jobs/workflow-statuses-update.job'; } from 'src/modules/workflow/workflow-status/jobs/workflow-statuses-update.job';
import { OnDatabaseEvent } from 'src/engine/api/graphql/graphql-query-runner/decorators/on-database-event.decorator'; 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 { DatabaseEventAction } from 'src/engine/api/graphql/graphql-query-runner/enums/database-event-action';
import { WORKFLOW_VERSION_STATUS_UPDATED } from 'src/modules/workflow/workflow-status/constants/workflow-version-status-updated.constants';
import { OnCustomBatchEvent } from 'src/engine/api/graphql/graphql-query-runner/decorators/on-custom-batch-event.decorator';
@Injectable() @Injectable()
export class WorkflowVersionStatusListener { export class WorkflowVersionStatusListener {
@ -26,13 +28,13 @@ export class WorkflowVersionStatusListener {
private readonly messageQueueService: MessageQueueService, private readonly messageQueueService: MessageQueueService,
) {} ) {}
@OnDatabaseEvent('workflowVersion', DatabaseEventAction.CREATED) @OnDatabaseBatchEvent('workflowVersion', DatabaseEventAction.CREATED)
async handleWorkflowVersionCreated( async handleWorkflowVersionCreated(
payload: WorkspaceEventBatch< batchEvent: WorkspaceEventBatch<
ObjectRecordCreateEvent<WorkflowVersionWorkspaceEntity> ObjectRecordCreateEvent<WorkflowVersionWorkspaceEntity>
>, >,
): Promise<void> { ): Promise<void> {
const workflowIds = payload.events const workflowIds = batchEvent.events
.filter( .filter(
(event) => (event) =>
!event.properties.after.status || !event.properties.after.status ||
@ -48,33 +50,33 @@ export class WorkflowVersionStatusListener {
WorkflowStatusesUpdateJob.name, WorkflowStatusesUpdateJob.name,
{ {
type: WorkflowVersionEventType.CREATE, type: WorkflowVersionEventType.CREATE,
workspaceId: payload.workspaceId, workspaceId: batchEvent.workspaceId,
workflowIds, workflowIds,
}, },
); );
} }
@OnDatabaseEvent('workflowVersion', DatabaseEventAction.UPDATED) @OnCustomBatchEvent(WORKFLOW_VERSION_STATUS_UPDATED)
async handleWorkflowVersionUpdated( async handleWorkflowVersionUpdated(
payload: WorkspaceEventBatch<WorkflowVersionStatusUpdate>, batchEvent: WorkspaceEventBatch<WorkflowVersionStatusUpdate>,
): Promise<void> { ): Promise<void> {
await this.messageQueueService.add<WorkflowVersionBatchEvent>( await this.messageQueueService.add<WorkflowVersionBatchEvent>(
WorkflowStatusesUpdateJob.name, WorkflowStatusesUpdateJob.name,
{ {
type: WorkflowVersionEventType.STATUS_UPDATE, type: WorkflowVersionEventType.STATUS_UPDATE,
workspaceId: payload.workspaceId, workspaceId: batchEvent.workspaceId,
statusUpdates: payload.events, statusUpdates: batchEvent.events,
}, },
); );
} }
@OnDatabaseEvent('workflowVersion', DatabaseEventAction.DELETED) @OnDatabaseBatchEvent('workflowVersion', DatabaseEventAction.DELETED)
async handleWorkflowVersionDeleted( async handleWorkflowVersionDeleted(
payload: WorkspaceEventBatch< batchEvent: WorkspaceEventBatch<
ObjectRecordDeleteEvent<WorkflowVersionWorkspaceEntity> ObjectRecordDeleteEvent<WorkflowVersionWorkspaceEntity>
>, >,
): Promise<void> { ): Promise<void> {
const workflowIds = payload.events const workflowIds = batchEvent.events
.filter( .filter(
(event) => (event) =>
event.properties.before.status === WorkflowVersionStatus.DRAFT, event.properties.before.status === WorkflowVersionStatus.DRAFT,
@ -89,7 +91,7 @@ export class WorkflowVersionStatusListener {
WorkflowStatusesUpdateJob.name, WorkflowStatusesUpdateJob.name,
{ {
type: WorkflowVersionEventType.DELETE, type: WorkflowVersionEventType.DELETE,
workspaceId: payload.workspaceId, workspaceId: batchEvent.workspaceId,
workflowIds, workflowIds,
}, },
); );

28
packages/twenty-server/src/modules/workflow/workflow-trigger/database-event-trigger/listeners/database-event-trigger.listener.ts
View File

@ -10,13 +10,13 @@ import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decora
import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants'; 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 { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service';
import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager'; import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager';
import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/workspace-event.type'; import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type';
import { WorkflowEventListenerWorkspaceEntity } from 'src/modules/workflow/common/standard-objects/workflow-event-listener.workspace-entity'; import { WorkflowEventListenerWorkspaceEntity } from 'src/modules/workflow/common/standard-objects/workflow-event-listener.workspace-entity';
import { import {
WorkflowEventTriggerJob, WorkflowEventTriggerJob,
WorkflowEventTriggerJobData, WorkflowEventTriggerJobData,
} from 'src/modules/workflow/workflow-trigger/jobs/workflow-event-trigger.job'; } from 'src/modules/workflow/workflow-trigger/jobs/workflow-event-trigger.job';
import { OnDatabaseEvent } from 'src/engine/api/graphql/graphql-query-runner/decorators/on-database-event.decorator'; 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 { DatabaseEventAction } from 'src/engine/api/graphql/graphql-query-runner/enums/database-event-action';
@Injectable() @Injectable()
@ -30,40 +30,40 @@ export class DatabaseEventTriggerListener {
private readonly isFeatureFlagEnabledService: FeatureFlagService, private readonly isFeatureFlagEnabledService: FeatureFlagService,
) {} ) {}
@OnDatabaseEvent('*', DatabaseEventAction.CREATED) @OnDatabaseBatchEvent('*', DatabaseEventAction.CREATED)
async handleObjectRecordCreateEvent( async handleObjectRecordCreateEvent(
payload: WorkspaceEventBatch<ObjectRecordCreateEvent<any>>, payload: WorkspaceEventBatch<ObjectRecordCreateEvent>,
) { ) {
await this.handleEvent(payload); await this.handleEvent(payload);
} }
@OnDatabaseEvent('*', DatabaseEventAction.UPDATED) @OnDatabaseBatchEvent('*', DatabaseEventAction.UPDATED)
async handleObjectRecordUpdateEvent( async handleObjectRecordUpdateEvent(
payload: WorkspaceEventBatch<ObjectRecordUpdateEvent<any>>, payload: WorkspaceEventBatch<ObjectRecordUpdateEvent>,
) { ) {
await this.handleEvent(payload); await this.handleEvent(payload);
} }
@OnDatabaseEvent('*', DatabaseEventAction.DELETED) @OnDatabaseBatchEvent('*', DatabaseEventAction.DELETED)
async handleObjectRecordDeleteEvent( async handleObjectRecordDeleteEvent(
payload: WorkspaceEventBatch<ObjectRecordDeleteEvent<any>>, payload: WorkspaceEventBatch<ObjectRecordDeleteEvent>,
) { ) {
await this.handleEvent(payload); await this.handleEvent(payload);
} }
@OnDatabaseEvent('*', DatabaseEventAction.DESTROYED) @OnDatabaseBatchEvent('*', DatabaseEventAction.DESTROYED)
async handleObjectRecordDestroyEvent( async handleObjectRecordDestroyEvent(
payload: WorkspaceEventBatch<ObjectRecordDestroyEvent<any>>, payload: WorkspaceEventBatch<ObjectRecordDestroyEvent>,
) { ) {
await this.handleEvent(payload); await this.handleEvent(payload);
} }
private async handleEvent( private async handleEvent(
payload: WorkspaceEventBatch< payload: WorkspaceEventBatch<
| ObjectRecordCreateEvent<any> | ObjectRecordCreateEvent
| ObjectRecordUpdateEvent<any> | ObjectRecordUpdateEvent
| ObjectRecordDeleteEvent<any> | ObjectRecordDeleteEvent
| ObjectRecordDestroyEvent<any> | ObjectRecordDestroyEvent
>, >,
) { ) {
const workspaceId = payload.workspaceId; const workspaceId = payload.workspaceId;

4
packages/twenty-server/src/modules/workflow/workflow-trigger/workflow-trigger.module.ts
View File

@ -1,17 +1,21 @@
import { Module } from '@nestjs/common'; import { Module } from '@nestjs/common';
import { NestjsQueryTypeOrmModule } from '@ptc-org/nestjs-query-typeorm';
import { ScopedWorkspaceContextFactory } from 'src/engine/twenty-orm/factories/scoped-workspace-context.factory'; import { ScopedWorkspaceContextFactory } from 'src/engine/twenty-orm/factories/scoped-workspace-context.factory';
import { WorkflowCommonModule } from 'src/modules/workflow/common/workflow-common.module'; import { WorkflowCommonModule } from 'src/modules/workflow/common/workflow-common.module';
import { WorkflowRunnerModule } from 'src/modules/workflow/workflow-runner/workflow-runner.module'; import { WorkflowRunnerModule } from 'src/modules/workflow/workflow-runner/workflow-runner.module';
import { DatabaseEventTriggerModule } from 'src/modules/workflow/workflow-trigger/database-event-trigger/database-event-trigger.module'; import { DatabaseEventTriggerModule } from 'src/modules/workflow/workflow-trigger/database-event-trigger/database-event-trigger.module';
import { WorkflowEventTriggerJob } from 'src/modules/workflow/workflow-trigger/jobs/workflow-event-trigger.job'; import { WorkflowEventTriggerJob } from 'src/modules/workflow/workflow-trigger/jobs/workflow-event-trigger.job';
import { WorkflowTriggerWorkspaceService } from 'src/modules/workflow/workflow-trigger/workspace-services/workflow-trigger.workspace-service'; import { WorkflowTriggerWorkspaceService } from 'src/modules/workflow/workflow-trigger/workspace-services/workflow-trigger.workspace-service';
import { ObjectMetadataEntity } from 'src/engine/metadata-modules/object-metadata/object-metadata.entity';
@Module({ @Module({
imports: [ imports: [
WorkflowCommonModule, WorkflowCommonModule,
WorkflowRunnerModule, WorkflowRunnerModule,
DatabaseEventTriggerModule, DatabaseEventTriggerModule,
NestjsQueryTypeOrmModule.forFeature([ObjectMetadataEntity], 'metadata'),
], ],
providers: [ providers: [
WorkflowTriggerWorkspaceService, WorkflowTriggerWorkspaceService,

85
packages/twenty-server/src/modules/workflow/workflow-trigger/workspace-services/workflow-trigger.workspace-service.ts
View File

@ -1,6 +1,7 @@
import { Injectable } from '@nestjs/common'; import { Injectable } from '@nestjs/common';
import { InjectRepository } from '@nestjs/typeorm';
import { EntityManager } from 'typeorm'; import { EntityManager, Repository } from 'typeorm';
import { buildCreatedByFromWorkspaceMember } from 'src/engine/core-modules/actor/utils/build-created-by-from-workspace-member.util'; import { buildCreatedByFromWorkspaceMember } from 'src/engine/core-modules/actor/utils/build-created-by-from-workspace-member.util';
import { User } from 'src/engine/core-modules/user/user.entity'; import { User } from 'src/engine/core-modules/user/user.entity';
@ -16,7 +17,6 @@ import { WorkflowWorkspaceEntity } from 'src/modules/workflow/common/standard-ob
import { assertWorkflowVersionTriggerIsDefined } from 'src/modules/workflow/common/utils/assert-workflow-version-trigger-is-defined.util'; import { assertWorkflowVersionTriggerIsDefined } from 'src/modules/workflow/common/utils/assert-workflow-version-trigger-is-defined.util';
import { WorkflowCommonWorkspaceService } from 'src/modules/workflow/common/workspace-services/workflow-common.workspace-service'; import { WorkflowCommonWorkspaceService } from 'src/modules/workflow/common/workspace-services/workflow-common.workspace-service';
import { WorkflowRunnerWorkspaceService } from 'src/modules/workflow/workflow-runner/workspace-services/workflow-runner.workspace-service'; import { WorkflowRunnerWorkspaceService } from 'src/modules/workflow/workflow-runner/workspace-services/workflow-runner.workspace-service';
import { WorkflowVersionStatusUpdate } from 'src/modules/workflow/workflow-status/jobs/workflow-statuses-update.job';
import { DatabaseEventTriggerService } from 'src/modules/workflow/workflow-trigger/database-event-trigger/database-event-trigger.service'; import { DatabaseEventTriggerService } from 'src/modules/workflow/workflow-trigger/database-event-trigger/database-event-trigger.service';
import { import {
WorkflowTriggerException, WorkflowTriggerException,
@ -26,6 +26,8 @@ import { WorkflowTriggerType } from 'src/modules/workflow/workflow-trigger/types
import { assertVersionCanBeActivated } from 'src/modules/workflow/workflow-trigger/utils/assert-version-can-be-activated.util'; import { assertVersionCanBeActivated } from 'src/modules/workflow/workflow-trigger/utils/assert-version-can-be-activated.util';
import { assertNever } from 'src/utils/assert'; import { assertNever } from 'src/utils/assert';
import { DatabaseEventAction } from 'src/engine/api/graphql/graphql-query-runner/enums/database-event-action'; import { DatabaseEventAction } from 'src/engine/api/graphql/graphql-query-runner/enums/database-event-action';
import { WORKFLOW_VERSION_STATUS_UPDATED } from 'src/modules/workflow/workflow-status/constants/workflow-version-status-updated.constants';
import { ObjectMetadataEntity } from 'src/engine/metadata-modules/object-metadata/object-metadata.entity';
@Injectable() @Injectable()
export class WorkflowTriggerWorkspaceService { export class WorkflowTriggerWorkspaceService {
@ -36,14 +38,11 @@ export class WorkflowTriggerWorkspaceService {
private readonly workflowRunnerWorkspaceService: WorkflowRunnerWorkspaceService, private readonly workflowRunnerWorkspaceService: WorkflowRunnerWorkspaceService,
private readonly databaseEventTriggerService: DatabaseEventTriggerService, private readonly databaseEventTriggerService: DatabaseEventTriggerService,
private readonly workspaceEventEmitter: WorkspaceEventEmitter, private readonly workspaceEventEmitter: WorkspaceEventEmitter,
@InjectRepository(ObjectMetadataEntity, 'metadata')
private readonly objectMetadataRepository: Repository<ObjectMetadataEntity>,
) {} ) {}
async runWorkflowVersion( private getWorkspaceId() {
workflowVersionId: string,
payload: object,
workspaceMemberId: string,
user: User,
) {
const workspaceId = this.scopedWorkspaceContextFactory.create().workspaceId; const workspaceId = this.scopedWorkspaceContextFactory.create().workspaceId;
if (!workspaceId) { if (!workspaceId) {
@ -53,12 +52,21 @@ export class WorkflowTriggerWorkspaceService {
); );
} }
return workspaceId;
}
async runWorkflowVersion(
workflowVersionId: string,
payload: object,
workspaceMemberId: string,
user: User,
) {
await this.workflowCommonWorkspaceService.getWorkflowVersionOrFail( await this.workflowCommonWorkspaceService.getWorkflowVersionOrFail(
workflowVersionId, workflowVersionId,
); );
return await this.workflowRunnerWorkspaceService.run( return await this.workflowRunnerWorkspaceService.run(
workspaceId, this.getWorkspaceId(),
workflowVersionId, workflowVersionId,
payload, payload,
buildCreatedByFromWorkspaceMember(workspaceMemberId, user), buildCreatedByFromWorkspaceMember(workspaceMemberId, user),
@ -246,10 +254,10 @@ export class WorkflowTriggerWorkspaceService {
manager, manager,
); );
this.emitStatusUpdateEventOrThrow( await this.emitStatusUpdateEvents(
workflowVersion.workflowId, workflowVersion,
workflowVersion.status,
WorkflowVersionStatus.ACTIVE, WorkflowVersionStatus.ACTIVE,
this.getWorkspaceId(),
); );
} }
@ -271,10 +279,10 @@ export class WorkflowTriggerWorkspaceService {
manager, manager,
); );
this.emitStatusUpdateEventOrThrow( await this.emitStatusUpdateEvents(
workflowVersion.workflowId, workflowVersion,
workflowVersion.status,
WorkflowVersionStatus.DEACTIVATED, WorkflowVersionStatus.DEACTIVATED,
this.getWorkspaceId(),
); );
} }
@ -348,28 +356,45 @@ export class WorkflowTriggerWorkspaceService {
} }
} }
private emitStatusUpdateEventOrThrow( private async emitStatusUpdateEvents(
workflowId: string, workflowVersion: WorkflowVersionWorkspaceEntity,
previousStatus: WorkflowVersionStatus,
newStatus: WorkflowVersionStatus, newStatus: WorkflowVersionStatus,
workspaceId: string,
) { ) {
const workspaceId = this.scopedWorkspaceContextFactory.create().workspaceId; const objectMetadata = await this.objectMetadataRepository.findOneOrFail({
where: {
nameSingular: 'workflowVersion',
},
});
if (!workspaceId) { this.workspaceEventEmitter.emitDatabaseBatchEvent({
throw new WorkflowTriggerException( objectMetadataNameSingular: 'workflowVersion',
'No workspace id found', action: DatabaseEventAction.UPDATED,
WorkflowTriggerExceptionCode.INTERNAL_ERROR, events: [
); {
} recordId: workflowVersion.id,
objectMetadata,
properties: {
before: workflowVersion,
after: { ...workflowVersion, status: newStatus },
updatedFields: ['status'],
diff: {
status: { before: workflowVersion.status, after: newStatus },
},
},
},
],
workspaceId,
});
this.workspaceEventEmitter.emit( this.workspaceEventEmitter.emitCustomBatchEvent(
`workflowVersion.${DatabaseEventAction.UPDATED}`, WORKFLOW_VERSION_STATUS_UPDATED,
[ [
{ {
workflowId, workflowId: workflowVersion.workflowId,
previousStatus, previousStatus: workflowVersion.status,
newStatus, newStatus,
} satisfies WorkflowVersionStatusUpdate, },
], ],
workspaceId, workspaceId,
); );