Add db event emitter in twenty orm (#13167)

## Context
Add an eventEmitter instance to twenty datasources so we can emit DB
events.
Add input and output formatting to twenty orm (formatData, formatResult)
Those 2 elements simplified existing logic when we interact with the
ORM, input will be formatted by the ORM so we can directly use
field-like structure instead of column-like. The output will be
formatted, for builder queries it will be in `result.generatedMaps`
where `result.raw` preserves the previous column-like structure.

Important change: We now have an authContext that we can pass when we
get a repository, this will be used for the different events emitted in
the ORM. We also removed the caching for repositories as it was not
scaling well and not necessary imho

Note: An upcoming PR should handle the onDelete: cascade behavior where
we send DESTROY events in cascade when there is an onDelete: CASCADE on
the FK.

---------

Co-authored-by: Charles Bochet <charles@twenty.com>

packages/twenty-server/src/modules/workflow/workflow-status/jobs/__tests__/workflow-statuses-update.job.spec.ts

@@ -5,7 +5,6 @@ import { ObjectMetadataEntity } from 'src/engine/metadata-modules/object-metadat
 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 { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager';
-import { WorkspaceEventEmitter } from 'src/engine/workspace-event-emitter/workspace-event-emitter';
 import { WorkflowVersionStatus } from 'src/modules/workflow/common/standard-objects/workflow-version.workspace-entity';
 import { WorkflowStatus } from 'src/modules/workflow/common/standard-objects/workflow.workspace-entity';
 import {
@@ -54,10 +53,6 @@ describe('WorkflowStatusesUpdate', () => {
     findOneOrFail: jest.fn(),
   };
 
-  const mockWorkspaceEventEmitter = {
-    emitDatabaseBatchEvent: jest.fn(),
-  };
-
   beforeEach(async () => {
     const module: TestingModule = await Test.createTestingModule({
       providers: [
@@ -70,10 +65,6 @@ describe('WorkflowStatusesUpdate', () => {
           provide: ServerlessFunctionService,
           useValue: mockServerlessFunctionService,
         },
-        {
-          provide: WorkspaceEventEmitter,
-          useValue: mockWorkspaceEventEmitter,
-        },
         {
           provide: getRepositoryToken(ObjectMetadataEntity, 'core'),
           useValue: {
@@ -125,9 +116,6 @@ describe('WorkflowStatusesUpdate', () => {
 
         expect(mockWorkflowRepository.findOneOrFail).toHaveBeenCalledTimes(1);
         expect(mockWorkflowRepository.update).toHaveBeenCalledTimes(0);
-        expect(
-          mockWorkspaceEventEmitter.emitDatabaseBatchEvent,
-        ).toHaveBeenCalledTimes(0);
       });
 
       it('when no draft yet, update statuses', async () => {
@@ -155,9 +143,6 @@ describe('WorkflowStatusesUpdate', () => {
           { id: '1' },
           { statuses: [WorkflowStatus.DRAFT, WorkflowStatus.ACTIVE] },
         );
-        expect(
-          mockWorkspaceEventEmitter.emitDatabaseBatchEvent,
-        ).toHaveBeenCalledTimes(1);
       });
     });
 
@@ -202,9 +187,6 @@ describe('WorkflowStatusesUpdate', () => {
           mockWorkflowVersionRepository.findOneOrFail,
         ).toHaveBeenCalledTimes(1);
         expect(mockWorkflowRepository.update).toHaveBeenCalledTimes(0);
-        expect(
-          mockWorkspaceEventEmitter.emitDatabaseBatchEvent,
-        ).toHaveBeenCalledTimes(0);
       });
 
       test('when WorkflowVersionStatus.DRAFT to WorkflowVersionStatus.ACTIVE, should activate and publish serverless functions', async () => {
@@ -286,9 +268,6 @@ describe('WorkflowStatusesUpdate', () => {
           { id: '1' },
           { statuses: [WorkflowStatus.ACTIVE] },
         );
-        expect(
-          mockWorkspaceEventEmitter.emitDatabaseBatchEvent,
-        ).toHaveBeenCalledTimes(1);
       });
     });
 
@@ -338,9 +317,6 @@ describe('WorkflowStatusesUpdate', () => {
           { id: '1' },
           { statuses: [] },
         );
-        expect(
-          mockWorkspaceEventEmitter.emitDatabaseBatchEvent,
-        ).toHaveBeenCalledTimes(1);
       });
     });
   });

packages/twenty-server/src/modules/workflow/workflow-status/jobs/workflow-statuses-update.job.ts

@@ -1,20 +1,15 @@
 import { Logger, Scope } from '@nestjs/common';
-import { InjectRepository } from '@nestjs/typeorm';
 
 import isEqual from 'lodash.isequal';
 import { isDefined } from 'twenty-shared/utils';
-import { In, Repository } from 'typeorm';
+import { In } from 'typeorm';
 
-import { DatabaseEventAction } from 'src/engine/api/graphql/graphql-query-runner/enums/database-event-action';
 import { Process } from 'src/engine/core-modules/message-queue/decorators/process.decorator';
 import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.decorator';
 import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
-import { ObjectMetadataEntity } from 'src/engine/metadata-modules/object-metadata/object-metadata.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 { WorkspaceRepository } from 'src/engine/twenty-orm/repository/workspace.repository';
 import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager';
-import { WorkspaceEventEmitter } from 'src/engine/workspace-event-emitter/workspace-event-emitter';
 import {
   WorkflowVersionStatus,
   WorkflowVersionWorkspaceEntity,
@@ -72,22 +67,10 @@ export class WorkflowStatusesUpdateJob {
   constructor(
     private readonly twentyORMGlobalManager: TwentyORMGlobalManager,
     private readonly serverlessFunctionService: ServerlessFunctionService,
-    private readonly workspaceEventEmitter: WorkspaceEventEmitter,
-    @InjectRepository(ObjectMetadataEntity, 'core')
-    protected readonly objectMetadataRepository: Repository<ObjectMetadataEntity>,
-    @InjectRepository(ServerlessFunctionEntity, 'core')
-    private readonly serverlessFunctionRepository: Repository<ServerlessFunctionEntity>,
   ) {}
 
   @Process(WorkflowStatusesUpdateJob.name)
   async handle(event: WorkflowVersionBatchEvent): Promise<void> {
-    const workflowObjectMetadata =
-      await this.objectMetadataRepository.findOneOrFail({
-        where: {
-          nameSingular: 'workflow',
-        },
-      });
-
     switch (event.type) {
       case WorkflowVersionEventType.CREATE:
       case WorkflowVersionEventType.DELETE:
@@ -95,7 +78,6 @@ export class WorkflowStatusesUpdateJob {
           event.workflowIds.map((workflowId) =>
             this.handleWorkflowVersionCreatedOrDeleted({
               workflowId,
-              workflowObjectMetadata,
               workspaceId: event.workspaceId,
             }),
           ),
@@ -106,7 +88,6 @@ export class WorkflowStatusesUpdateJob {
           event.statusUpdates.map((statusUpdate) =>
             this.handleWorkflowVersionStatusUpdated({
               statusUpdate,
-              workflowObjectMetadata,
               workspaceId: event.workspaceId,
             }),
           ),
@@ -119,11 +100,9 @@ export class WorkflowStatusesUpdateJob {
 
   private async handleWorkflowVersionCreatedOrDeleted({
     workflowId,
-    workflowObjectMetadata,
     workspaceId,
   }: {
     workflowId: string;
-    workflowObjectMetadata: ObjectMetadataEntity;
     workspaceId: string;
   }): Promise<void> {
     const workflowRepository =
@@ -163,13 +142,6 @@ export class WorkflowStatusesUpdateJob {
         statuses: newWorkflowStatuses,
       },
     );
-
-    this.emitWorkflowStatusUpdatedEvent({
-      currentWorkflow: previousWorkflow,
-      workflowObjectMetadata,
-      newWorkflowStatuses,
-      workspaceId,
-    });
   }
 
   private async handlePublishServerlessFunction({
@@ -222,11 +194,9 @@ export class WorkflowStatusesUpdateJob {
 
   private async handleWorkflowVersionStatusUpdated({
     statusUpdate,
-    workflowObjectMetadata,
     workspaceId,
   }: {
     statusUpdate: WorkflowVersionStatusUpdate;
-    workflowObjectMetadata: ObjectMetadataEntity;
     workspaceId: string;
   }): Promise<void> {
     const workflowRepository =
@@ -277,51 +247,6 @@ export class WorkflowStatusesUpdateJob {
         statuses: newWorkflowStatuses,
       },
     );
-
-    this.emitWorkflowStatusUpdatedEvent({
-      currentWorkflow: workflow,
-      workflowObjectMetadata,
-      newWorkflowStatuses,
-      workspaceId,
-    });
-  }
-
-  private emitWorkflowStatusUpdatedEvent({
-    currentWorkflow,
-    workflowObjectMetadata,
-    newWorkflowStatuses,
-    workspaceId,
-  }: {
-    currentWorkflow: WorkflowWorkspaceEntity;
-    workflowObjectMetadata: ObjectMetadataEntity;
-    newWorkflowStatuses: WorkflowStatus[];
-    workspaceId: string;
-  }) {
-    this.workspaceEventEmitter.emitDatabaseBatchEvent({
-      objectMetadataNameSingular: workflowObjectMetadata.nameSingular,
-      action: DatabaseEventAction.UPDATED,
-      events: [
-        {
-          recordId: currentWorkflow.id,
-          objectMetadata: workflowObjectMetadata,
-          properties: {
-            before: currentWorkflow,
-            after: {
-              ...currentWorkflow,
-              statuses: newWorkflowStatuses,
-            },
-            updatedFields: ['statuses'],
-            diff: {
-              statuses: {
-                before: currentWorkflow.statuses,
-                after: newWorkflowStatuses,
-              },
-            },
-          },
-        },
-      ],
-      workspaceId,
-    });
   }
 
   private async getWorkflowStatuses({