diff --git a/packages/twenty-server/src/database/commands/upgrade-version-command/2-23/2-23-upgrade-version-command.module.ts b/packages/twenty-server/src/database/commands/upgrade-version-command/2-23/2-23-upgrade-version-command.module.ts new file mode 100644 index 0000000000..21510f74b3 --- /dev/null +++ b/packages/twenty-server/src/database/commands/upgrade-version-command/2-23/2-23-upgrade-version-command.module.ts @@ -0,0 +1,22 @@ +import { Module } from '@nestjs/common'; + +import { WorkspaceIteratorModule } from 'src/database/commands/command-runners/workspace-iterator.module'; +import { AddWorkflowCoreSoftRefFieldCommand } from 'src/database/commands/upgrade-version-command/2-23/2-23-workspace-command-1784286706000-add-workflow-core-soft-ref-field.command'; +import { BackfillWorkflowCoreLinksCommand } from 'src/database/commands/upgrade-version-command/2-23/2-23-workspace-command-1784286707000-backfill-workflow-core-links.command'; +import { ApplicationModule } from 'src/engine/core-modules/application/application.module'; +import { WorkspaceCacheModule } from 'src/engine/workspace-cache/workspace-cache.module'; +import { WorkspaceMigrationModule } from 'src/engine/workspace-manager/workspace-migration/workspace-migration.module'; + +@Module({ + imports: [ + ApplicationModule, + WorkspaceCacheModule, + WorkspaceMigrationModule, + WorkspaceIteratorModule, + ], + providers: [ + AddWorkflowCoreSoftRefFieldCommand, + BackfillWorkflowCoreLinksCommand, + ], +}) +export class V2_23_UpgradeVersionCommandModule {} diff --git a/packages/twenty-server/src/database/commands/upgrade-version-command/2-23/2-23-workspace-command-1784286706000-add-workflow-core-soft-ref-field.command.ts b/packages/twenty-server/src/database/commands/upgrade-version-command/2-23/2-23-workspace-command-1784286706000-add-workflow-core-soft-ref-field.command.ts new file mode 100644 index 0000000000..31e45b9201 --- /dev/null +++ b/packages/twenty-server/src/database/commands/upgrade-version-command/2-23/2-23-workspace-command-1784286706000-add-workflow-core-soft-ref-field.command.ts @@ -0,0 +1,141 @@ +import { Command } from 'nest-commander'; +import { STANDARD_OBJECTS } from 'twenty-shared/metadata'; +import { isDefined } from 'twenty-shared/utils'; + +import { ProvisionedWorkspaceCommandRunner } from 'src/database/commands/command-runners/provisioned-workspace.command-runner'; +import { WorkspaceIteratorService } from 'src/database/commands/command-runners/workspace-iterator.service'; +import { type RunOnWorkspaceArgs } from 'src/database/commands/command-runners/workspace.command-runner'; +import { ApplicationService } from 'src/engine/core-modules/application/application.service'; +import { RegisteredWorkspaceCommand } from 'src/engine/core-modules/upgrade/decorators/registered-workspace-command.decorator'; +import { findFlatEntityByUniversalIdentifier } from 'src/engine/metadata-modules/flat-entity/utils/find-flat-entity-by-universal-identifier.util'; +import { type FlatFieldMetadata } from 'src/engine/metadata-modules/flat-field-metadata/types/flat-field-metadata.type'; +import { type FlatObjectMetadata } from 'src/engine/metadata-modules/flat-object-metadata/types/flat-object-metadata.type'; +import { computeTwentyStandardApplicationAllFlatEntityMaps } from 'src/engine/workspace-manager/twenty-standard-application/utils/twenty-standard-application-all-flat-entity-maps.constant'; +import { WorkspaceCacheService } from 'src/engine/workspace-cache/services/workspace-cache.service'; +import { WorkspaceMigrationValidateBuildAndRunService } from 'src/engine/workspace-manager/workspace-migration/services/workspace-migration-validate-build-and-run-service'; + +const WORKFLOW = STANDARD_OBJECTS.workflow; +const CORE_WORKFLOW_ID_FIELD_UNIVERSAL_IDENTIFIER = + WORKFLOW.fields.coreWorkflowId.universalIdentifier; + +@RegisteredWorkspaceCommand('2.23.0', 1784286706000) +@Command({ + name: 'upgrade:2-23:add-workflow-core-soft-ref-field', + description: + 'Add the workflow.coreWorkflowId system field on existing workspaces that predate it', +}) +export class AddWorkflowCoreSoftRefFieldCommand extends ProvisionedWorkspaceCommandRunner { + constructor( + protected readonly workspaceIteratorService: WorkspaceIteratorService, + private readonly applicationService: ApplicationService, + private readonly workspaceCacheService: WorkspaceCacheService, + private readonly workspaceMigrationValidateBuildAndRunService: WorkspaceMigrationValidateBuildAndRunService, + ) { + super(workspaceIteratorService); + } + + override async runOnWorkspace({ + workspaceId, + options, + }: RunOnWorkspaceArgs): Promise { + const isDryRun = options.dryRun ?? false; + + const { flatFieldMetadataMaps, flatObjectMetadataMaps } = + await this.workspaceCacheService.getOrRecompute(workspaceId, [ + 'flatFieldMetadataMaps', + 'flatObjectMetadataMaps', + ]); + + const workflowObjectMetadata = + findFlatEntityByUniversalIdentifier({ + flatEntityMaps: flatObjectMetadataMaps, + universalIdentifier: WORKFLOW.universalIdentifier, + }); + + if (!isDefined(workflowObjectMetadata)) { + this.logger.log( + `workflow object does not exist for workspace ${workspaceId}, skipping`, + ); + + return; + } + + if ( + isDefined( + flatFieldMetadataMaps.byUniversalIdentifier[ + CORE_WORKFLOW_ID_FIELD_UNIVERSAL_IDENTIFIER + ], + ) + ) { + return; + } + + const { twentyStandardFlatApplication } = + await this.applicationService.findWorkspaceTwentyStandardAndCustomApplicationOrThrow( + { workspaceId }, + ); + + const { allFlatEntityMaps: standardAllFlatEntityMaps } = + computeTwentyStandardApplicationAllFlatEntityMaps({ + now: new Date().toISOString(), + workspaceId, + twentyStandardApplicationId: twentyStandardFlatApplication.id, + }); + + const standardField = findFlatEntityByUniversalIdentifier( + { + flatEntityMaps: standardAllFlatEntityMaps.flatFieldMetadataMaps, + universalIdentifier: CORE_WORKFLOW_ID_FIELD_UNIVERSAL_IDENTIFIER, + }, + ); + + if (!isDefined(standardField)) { + throw new Error( + 'Standard application is missing workflow field coreWorkflowId', + ); + } + + if (isDryRun) { + this.logger.log( + `[DRY RUN] Would add coreWorkflowId field for workspace ${workspaceId}`, + ); + + return; + } + + const flatFieldMetadataToCreate: FlatFieldMetadata = { + ...standardField, + viewFieldIds: [], + viewFieldUniversalIdentifiers: [], + }; + + const result = + await this.workspaceMigrationValidateBuildAndRunService.validateBuildAndRunLegacyWorkspaceMigration( + { + isSystemBuild: true, + workspaceId, + applicationUniversalIdentifier: + twentyStandardFlatApplication.universalIdentifier, + allFlatEntityOperationByMetadataName: { + fieldMetadata: { + flatEntityToCreate: [flatFieldMetadataToCreate], + flatEntityToDelete: [], + flatEntityToUpdate: [], + }, + }, + }, + ); + + if (result.status === 'fail') { + this.logger.error( + `Failed to add coreWorkflowId field:\n${JSON.stringify(result, null, 2)}`, + ); + + throw new Error( + `Failed to add coreWorkflowId field for workspace ${workspaceId}`, + ); + } + + this.logger.log(`Added coreWorkflowId field for workspace ${workspaceId}`); + } +} diff --git a/packages/twenty-server/src/database/commands/upgrade-version-command/2-23/2-23-workspace-command-1784286707000-backfill-workflow-core-links.command.ts b/packages/twenty-server/src/database/commands/upgrade-version-command/2-23/2-23-workspace-command-1784286707000-backfill-workflow-core-links.command.ts new file mode 100644 index 0000000000..14a5f72102 --- /dev/null +++ b/packages/twenty-server/src/database/commands/upgrade-version-command/2-23/2-23-workspace-command-1784286707000-backfill-workflow-core-links.command.ts @@ -0,0 +1,137 @@ +import { Command } from 'nest-commander'; +import { isDefined } from 'twenty-shared/utils'; +import { EntityMetadataNotFoundError } from 'typeorm/error/EntityMetadataNotFoundError'; +import { v4 as uuidv4 } from 'uuid'; + +import { ProvisionedWorkspaceCommandRunner } from 'src/database/commands/command-runners/provisioned-workspace.command-runner'; +import { WorkspaceIteratorService } from 'src/database/commands/command-runners/workspace-iterator.service'; +import { type RunOnWorkspaceArgs } from 'src/database/commands/command-runners/workspace.command-runner'; +import { RegisteredWorkspaceCommand } from 'src/engine/core-modules/upgrade/decorators/registered-workspace-command.decorator'; +import { type WorkflowWorkspaceEntity } from 'src/modules/workflow/common/standard-objects/workflow.workspace-entity'; + +// Full rebuild of the core workflow rows for a workspace once +// coreWorkflowId is provisioned. Per workspace, in one transaction: +// wipe every core row (clearing all pre-soft-ref / stale-id leftovers), insert +// a fresh own-id row for every workspace workflow, and write that id back onto +// the workspace record. Because it re-links every record, no workflow is left +// pointing at a deleted core row, so the dual-write's update path stays correct. +@RegisteredWorkspaceCommand('2.23.0', 1784286707000) +@Command({ + name: 'upgrade:2-23:backfill-workflow-core-links', + description: + 'Rebuild core workflow rows for each workspace and link every workspace record via coreWorkflowId', +}) +export class BackfillWorkflowCoreLinksCommand extends ProvisionedWorkspaceCommandRunner { + constructor( + protected readonly workspaceIteratorService: WorkspaceIteratorService, + ) { + super(workspaceIteratorService); + } + + override async runOnWorkspace({ + workspaceId, + options, + dataSource, + }: RunOnWorkspaceArgs): Promise { + if (!isDefined(dataSource)) { + this.logger.log( + `No workspace data source for workspace ${workspaceId}, skipping`, + ); + + return; + } + + let workspaceWorkflows: WorkflowWorkspaceEntity[]; + + try { + const workflowRepository = + dataSource.getRepository('workflow', { + shouldBypassPermissionChecks: true, + }); + + workspaceWorkflows = await workflowRepository.find(); + } catch (error) { + if (error instanceof EntityMetadataNotFoundError) { + this.logger.log( + `workflow object does not exist for workspace ${workspaceId}, skipping`, + ); + + return; + } + + throw error; + } + + if (options.dryRun === true) { + this.logger.log( + `[DRY RUN] Would rebuild ${workspaceWorkflows.length} core workflow row(s) for workspace ${workspaceId}`, + ); + + return; + } + + const queryRunner = dataSource.createQueryRunner(); + + await queryRunner.connect(); + + try { + const [workspace] = await queryRunner.query( + `SELECT "databaseSchema", "workspaceCustomApplicationId" FROM core."workspace" WHERE id = $1`, + [workspaceId], + ); + + if (!isDefined(workspace?.workspaceCustomApplicationId)) { + throw new Error( + `Workspace custom application not found for workspace ${workspaceId}`, + ); + } + + const schema: string = workspace.databaseSchema; + const applicationId: string = workspace.workspaceCustomApplicationId; + + await queryRunner.startTransaction(); + + try { + await queryRunner.query( + `DELETE FROM core."workflow" WHERE "workspaceId" = $1`, + [workspaceId], + ); + + for (const workflow of workspaceWorkflows) { + const coreWorkflowId = uuidv4(); + + await queryRunner.query( + `INSERT INTO core."workflow" + (id, "workspaceId", "universalIdentifier", "applicationId", name, "lastPublishedVersionId") + VALUES ($1, $2, $3, $4, $5, $6)`, + [ + coreWorkflowId, + workspaceId, + uuidv4(), + applicationId, + workflow.name ?? null, + workflow.lastPublishedVersionId ?? null, + ], + ); + + await queryRunner.query( + `UPDATE "${schema}"."workflow" SET "coreWorkflowId" = $1 WHERE id = $2`, + [coreWorkflowId, workflow.id], + ); + } + + await queryRunner.commitTransaction(); + } catch (error) { + await queryRunner.rollbackTransaction(); + + throw error; + } + + this.logger.log( + `Rebuilt ${workspaceWorkflows.length} core workflow row(s) for workspace ${workspaceId}`, + ); + } finally { + await queryRunner.release(); + } + } +} diff --git a/packages/twenty-server/src/database/commands/upgrade-version-command/workspace-command-provider.module.ts b/packages/twenty-server/src/database/commands/upgrade-version-command/workspace-command-provider.module.ts index 83f1e28315..e374b69fff 100644 --- a/packages/twenty-server/src/database/commands/upgrade-version-command/workspace-command-provider.module.ts +++ b/packages/twenty-server/src/database/commands/upgrade-version-command/workspace-command-provider.module.ts @@ -23,6 +23,7 @@ import { V2_19_UpgradeVersionCommandModule } from 'src/database/commands/upgrade import { V2_20_UpgradeVersionCommandModule } from 'src/database/commands/upgrade-version-command/2-20/2-20-upgrade-version-command.module'; import { V2_21_UpgradeVersionCommandModule } from 'src/database/commands/upgrade-version-command/2-21/2-21-upgrade-version-command.module'; import { V2_22_UpgradeVersionCommandModule } from 'src/database/commands/upgrade-version-command/2-22/2-22-upgrade-version-command.module'; +import { V2_23_UpgradeVersionCommandModule } from 'src/database/commands/upgrade-version-command/2-23/2-23-upgrade-version-command.module'; @Module({ imports: [ @@ -49,6 +50,7 @@ import { V2_22_UpgradeVersionCommandModule } from 'src/database/commands/upgrade V2_20_UpgradeVersionCommandModule, V2_21_UpgradeVersionCommandModule, V2_22_UpgradeVersionCommandModule, + V2_23_UpgradeVersionCommandModule, ], }) export class WorkspaceCommandProviderModule {} diff --git a/packages/twenty-server/src/engine/core-modules/workflow/services/__tests__/workflow-core-sync.service.spec.ts b/packages/twenty-server/src/engine/core-modules/workflow/services/__tests__/workflow-core-sync.service.spec.ts new file mode 100644 index 0000000000..ddaf079668 --- /dev/null +++ b/packages/twenty-server/src/engine/core-modules/workflow/services/__tests__/workflow-core-sync.service.spec.ts @@ -0,0 +1,117 @@ +import { STANDARD_OBJECTS } from 'twenty-shared/metadata'; +import { type Repository } from 'typeorm'; + +import { type GlobalWorkspaceOrmManager } from 'src/engine/twenty-orm/global-workspace-datasource/global-workspace-orm.manager'; +import { type WorkspaceScopedRepository } from 'src/engine/twenty-orm/workspace-scoped-repository/workspace-scoped-repository'; +import { type WorkspaceCacheService } from 'src/engine/workspace-cache/services/workspace-cache.service'; +import { WorkflowCoreSyncService } from 'src/engine/core-modules/workflow/services/workflow-core-sync.service'; +import { type WorkflowEntity } from 'src/engine/core-modules/workflow/entities/workflow.entity'; +import { type WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity'; +import { type WorkflowWorkspaceEntity } from 'src/modules/workflow/common/standard-objects/workflow.workspace-entity'; + +const CORE_WORKFLOW_ID_FIELD = + STANDARD_OBJECTS.workflow.fields.coreWorkflowId.universalIdentifier; + +describe('WorkflowCoreSyncService', () => { + const workspaceId = '20202020-1c25-4d02-bf25-6aeccf7ea419'; + + let service: WorkflowCoreSyncService; + let workflowRepository: { upsert: jest.Mock; delete: jest.Mock }; + let workspaceRepository: { findOne: jest.Mock }; + let globalWorkspaceOrmManager: { executeInWorkspaceContext: jest.Mock }; + let workspaceCacheService: { getOrRecompute: jest.Mock }; + + const buildWorkflow = ( + overrides: Partial = {}, + ): WorkflowWorkspaceEntity => + ({ + id: '1dddc806-4144-5020-898f-b1ab287b89d5', + name: 'My workflow', + // Event payloads represent unset uuids as empty strings, not null. + lastPublishedVersionId: '', + coreWorkflowId: '', + ...overrides, + }) as unknown as WorkflowWorkspaceEntity; + + const mockFieldPresence = (present: boolean) => + workspaceCacheService.getOrRecompute.mockResolvedValue({ + flatFieldMetadataMaps: { + byUniversalIdentifier: present ? { [CORE_WORKFLOW_ID_FIELD]: {} } : {}, + }, + }); + + const upsertedRows = (): Array<{ + id: string; + lastPublishedVersionId: string | null; + }> => workflowRepository.upsert.mock.calls[0][1]; + + beforeEach(() => { + workflowRepository = { upsert: jest.fn(), delete: jest.fn() }; + workspaceRepository = { + findOne: jest.fn().mockResolvedValue({ + id: workspaceId, + workspaceCustomApplicationId: 'application-1', + }), + }; + globalWorkspaceOrmManager = { + executeInWorkspaceContext: jest.fn().mockResolvedValue(undefined), + }; + workspaceCacheService = { getOrRecompute: jest.fn() }; + + service = new WorkflowCoreSyncService( + workflowRepository as unknown as WorkspaceScopedRepository, + workspaceRepository as unknown as Repository, + globalWorkspaceOrmManager as unknown as GlobalWorkspaceOrmManager, + workspaceCacheService as unknown as WorkspaceCacheService, + ); + + jest.spyOn(service['logger'], 'warn').mockImplementation(); + }); + + afterEach(() => { + jest.clearAllMocks(); + }); + + it('skips the core id write-back when the workspace lacks the coreWorkflowId field', async () => { + mockFieldPresence(false); + + await expect( + service.upsertToCore(workspaceId, [buildWorkflow()]), + ).resolves.toBeUndefined(); + + expect(workflowRepository.upsert).toHaveBeenCalledTimes(1); + expect( + globalWorkspaceOrmManager.executeInWorkspaceContext, + ).not.toHaveBeenCalled(); + }); + + it('generates an id, normalizes empty-string uuids and writes back for an unlinked workflow', async () => { + mockFieldPresence(true); + + await service.upsertToCore(workspaceId, [buildWorkflow()]); + + expect(upsertedRows()[0].id).not.toBe(''); + expect(upsertedRows()[0].id.length).toBeGreaterThan(0); + // lastPublishedVersionId ('') must be normalized to null for the uuid column. + expect(upsertedRows()[0].lastPublishedVersionId).toBeNull(); + expect( + globalWorkspaceOrmManager.executeInWorkspaceContext, + ).toHaveBeenCalledTimes(1); + }); + + it('reuses the existing id and does not write back for a linked workflow', async () => { + mockFieldPresence(true); + const coreWorkflowId = 'e2b1c0d4-0000-4000-8000-000000000000'; + + await service.upsertToCore(workspaceId, [ + buildWorkflow({ + coreWorkflowId, + } as Partial), + ]); + + expect(upsertedRows()[0].id).toBe(coreWorkflowId); + expect( + globalWorkspaceOrmManager.executeInWorkspaceContext, + ).not.toHaveBeenCalled(); + }); +}); diff --git a/packages/twenty-server/src/engine/core-modules/workflow/services/__tests__/workflow-version-core-sync.service.spec.ts b/packages/twenty-server/src/engine/core-modules/workflow/services/__tests__/workflow-version-core-sync.service.spec.ts index a0cac3c708..742c741206 100644 --- a/packages/twenty-server/src/engine/core-modules/workflow/services/__tests__/workflow-version-core-sync.service.spec.ts +++ b/packages/twenty-server/src/engine/core-modules/workflow/services/__tests__/workflow-version-core-sync.service.spec.ts @@ -25,14 +25,18 @@ describe('WorkflowVersionCoreSyncService', () => { invalidateAndRecompute: jest.Mock; }; - const buildVersion = (): WorkflowVersionWorkspaceEntity => + const buildVersion = ( + overrides: Partial = {}, + ): WorkflowVersionWorkspaceEntity => ({ id: '1dddc806-4144-5020-898f-b1ab287b89d5', workflowId: 'c95c78b4-48d2-56f6-8e15-36ff8572f1d8', status: 'DRAFT', trigger: null, steps: null, - coreWorkflowVersionId: null, + // Event payloads represent unset uuids as empty strings, not null. + coreWorkflowVersionId: '', + ...overrides, }) as unknown as WorkflowVersionWorkspaceEntity; const mockFieldPresence = (present: boolean) => @@ -42,6 +46,9 @@ describe('WorkflowVersionCoreSyncService', () => { }, }); + const upsertedRows = (): Array<{ id: string }> => + workflowVersionRepository.upsert.mock.calls[0][1]; + beforeEach(() => { workflowVersionRepository = { upsert: jest.fn(), delete: jest.fn() }; workspaceRepository = { @@ -85,13 +92,32 @@ describe('WorkflowVersionCoreSyncService', () => { ).not.toHaveBeenCalled(); }); - it('writes the core id back when the field is present', async () => { + it('generates an id and writes it back for an unlinked version (empty-string id)', async () => { mockFieldPresence(true); await service.upsertToCore(workspaceId, [buildVersion()]); + // Empty-string coreWorkflowVersionId is treated as unlinked, not passed through. + expect(upsertedRows()[0].id).not.toBe(''); + expect(upsertedRows()[0].id.length).toBeGreaterThan(0); expect( globalWorkspaceOrmManager.executeInWorkspaceContext, ).toHaveBeenCalledTimes(1); }); + + it('reuses the existing id and does not write back for a linked version', async () => { + mockFieldPresence(true); + const coreWorkflowVersionId = 'e2b1c0d4-0000-4000-8000-000000000000'; + + await service.upsertToCore(workspaceId, [ + buildVersion({ + coreWorkflowVersionId, + } as Partial), + ]); + + expect(upsertedRows()[0].id).toBe(coreWorkflowVersionId); + expect( + globalWorkspaceOrmManager.executeInWorkspaceContext, + ).not.toHaveBeenCalled(); + }); }); diff --git a/packages/twenty-server/src/engine/core-modules/workflow/services/workflow-core-sync.service.ts b/packages/twenty-server/src/engine/core-modules/workflow/services/workflow-core-sync.service.ts new file mode 100644 index 0000000000..04d5ebdb6c --- /dev/null +++ b/packages/twenty-server/src/engine/core-modules/workflow/services/workflow-core-sync.service.ts @@ -0,0 +1,153 @@ +import { Injectable, Logger } from '@nestjs/common'; +import { InjectRepository } from '@nestjs/typeorm'; + +import { isNonEmptyString } from '@sniptt/guards'; +import { STANDARD_OBJECTS } from 'twenty-shared/metadata'; +import { isDefined } from 'twenty-shared/utils'; +import { In, Repository } from 'typeorm'; +import { v4 as uuidv4 } from 'uuid'; + +import { WorkflowEntity } from 'src/engine/core-modules/workflow/entities/workflow.entity'; +import { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity'; +import { GlobalWorkspaceOrmManager } from 'src/engine/twenty-orm/global-workspace-datasource/global-workspace-orm.manager'; +import { InjectWorkspaceScopedRepository } from 'src/engine/twenty-orm/workspace-scoped-repository/inject-workspace-scoped-repository.decorator'; +import { WorkspaceScopedRepository } from 'src/engine/twenty-orm/workspace-scoped-repository/workspace-scoped-repository'; +import { buildSystemAuthContext } from 'src/engine/twenty-orm/utils/build-system-auth-context.util'; +import { WorkspaceCacheService } from 'src/engine/workspace-cache/services/workspace-cache.service'; +import { type WorkflowWorkspaceEntity } from 'src/modules/workflow/common/standard-objects/workflow.workspace-entity'; + +@Injectable() +export class WorkflowCoreSyncService { + private readonly logger = new Logger(WorkflowCoreSyncService.name); + + constructor( + @InjectWorkspaceScopedRepository(WorkflowEntity) + private readonly coreWorkflowRepository: WorkspaceScopedRepository, + @InjectRepository(WorkspaceEntity) + private readonly workspaceRepository: Repository, + private readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager, + private readonly workspaceCacheService: WorkspaceCacheService, + ) {} + + async upsertToCore( + workspaceId: string, + workflows: WorkflowWorkspaceEntity[], + ): Promise { + if (workflows.length === 0) { + return; + } + + const applicationId = await this.getCustomApplicationIdOrThrow(workspaceId); + + const coreWorkflowIdByWorkspaceRecordId = new Map(); + + const coreRows = workflows.map((workflow) => { + const coreWorkflowId = isNonEmptyString(workflow.coreWorkflowId) + ? workflow.coreWorkflowId + : uuidv4(); + + if (!isNonEmptyString(workflow.coreWorkflowId)) { + coreWorkflowIdByWorkspaceRecordId.set(workflow.id, coreWorkflowId); + } + + return { + id: coreWorkflowId, + name: workflow.name ?? null, + lastPublishedVersionId: isNonEmptyString( + workflow.lastPublishedVersionId, + ) + ? workflow.lastPublishedVersionId + : null, + universalIdentifier: uuidv4(), + applicationId, + }; + }); + + await this.coreWorkflowRepository.upsert(workspaceId, coreRows, ['id']); + + await this.writeBackCoreWorkflowIds( + workspaceId, + coreWorkflowIdByWorkspaceRecordId, + ); + } + + async deleteFromCore( + workspaceId: string, + coreWorkflowIds: string[], + ): Promise { + if (coreWorkflowIds.length === 0) { + return; + } + + await this.coreWorkflowRepository.delete(workspaceId, { + id: In(coreWorkflowIds), + }); + } + + private async writeBackCoreWorkflowIds( + workspaceId: string, + coreWorkflowIdByWorkspaceRecordId: Map, + ): Promise { + if (coreWorkflowIdByWorkspaceRecordId.size === 0) { + return; + } + + if (!(await this.workspaceHasCoreWorkflowIdField(workspaceId))) { + this.logger.warn( + `workflow.coreWorkflowId field missing for workspace ${workspaceId}, skipping core id write-back`, + ); + + return; + } + + await this.globalWorkspaceOrmManager.executeInWorkspaceContext(async () => { + const workspaceWorkflowRepository = + await this.globalWorkspaceOrmManager.getRepository( + workspaceId, + 'workflow', + { shouldBypassPermissionChecks: true }, + ); + + for (const [ + workspaceRecordId, + coreWorkflowId, + ] of coreWorkflowIdByWorkspaceRecordId) { + await workspaceWorkflowRepository.update(workspaceRecordId, { + coreWorkflowId, + }); + } + }, buildSystemAuthContext(workspaceId)); + } + + private async workspaceHasCoreWorkflowIdField( + workspaceId: string, + ): Promise { + const { flatFieldMetadataMaps } = + await this.workspaceCacheService.getOrRecompute(workspaceId, [ + 'flatFieldMetadataMaps', + ]); + + return isDefined( + flatFieldMetadataMaps.byUniversalIdentifier[ + STANDARD_OBJECTS.workflow.fields.coreWorkflowId.universalIdentifier + ], + ); + } + + private async getCustomApplicationIdOrThrow( + workspaceId: string, + ): Promise { + const workspace = await this.workspaceRepository.findOne({ + where: { id: workspaceId }, + select: ['id', 'workspaceCustomApplicationId'], + }); + + if (!isDefined(workspace?.workspaceCustomApplicationId)) { + throw new Error( + `Workspace custom application not found for workspace ${workspaceId}`, + ); + } + + return workspace.workspaceCustomApplicationId; + } +} diff --git a/packages/twenty-server/src/engine/core-modules/workflow/services/workflow-version-core-sync.service.ts b/packages/twenty-server/src/engine/core-modules/workflow/services/workflow-version-core-sync.service.ts index e5eac82983..2fcf4dff59 100644 --- a/packages/twenty-server/src/engine/core-modules/workflow/services/workflow-version-core-sync.service.ts +++ b/packages/twenty-server/src/engine/core-modules/workflow/services/workflow-version-core-sync.service.ts @@ -1,10 +1,11 @@ import { Injectable, Logger } from '@nestjs/common'; import { InjectRepository } from '@nestjs/typeorm'; +import { isNonEmptyString } from '@sniptt/guards'; import { STANDARD_OBJECTS } from 'twenty-shared/metadata'; import { isDefined } from 'twenty-shared/utils'; import { In, Repository } from 'typeorm'; -import { v4 as uuidv4, v5 as uuidv5 } from 'uuid'; +import { v4 as uuidv4 } from 'uuid'; import { WorkflowVersionEntity, @@ -18,12 +19,6 @@ import { buildSystemAuthContext } from 'src/engine/twenty-orm/utils/build-system import { WorkspaceCacheService } from 'src/engine/workspace-cache/services/workspace-cache.service'; import { type WorkflowVersionWorkspaceEntity } from 'src/modules/workflow/common/standard-objects/workflow-version.workspace-entity'; -// Deriving the core id deterministically from workspaceId + record id makes the -// upsert idempotent across retries: a failed write-back re-derives the same id -// instead of orphaning a row or hitting the one-active-per-workflow index. -const CORE_WORKFLOW_VERSION_ID_NAMESPACE = - 'f4988927-0a5c-453a-a262-0bd136d7fdaf'; - @Injectable() export class WorkflowVersionCoreSyncService { private readonly logger = new Logger(WorkflowVersionCoreSyncService.name); @@ -50,14 +45,13 @@ export class WorkflowVersionCoreSyncService { const coreVersionIdByWorkspaceRecordId = new Map(); const coreRows = workflowVersions.map((workflowVersion) => { - const coreWorkflowVersionId = - workflowVersion.coreWorkflowVersionId ?? - uuidv5( - `${workspaceId}:${workflowVersion.id}`, - CORE_WORKFLOW_VERSION_ID_NAMESPACE, - ); + const coreWorkflowVersionId = isNonEmptyString( + workflowVersion.coreWorkflowVersionId, + ) + ? workflowVersion.coreWorkflowVersionId + : uuidv4(); - if (!isDefined(workflowVersion.coreWorkflowVersionId)) { + if (!isNonEmptyString(workflowVersion.coreWorkflowVersionId)) { coreVersionIdByWorkspaceRecordId.set( workflowVersion.id, coreWorkflowVersionId, @@ -77,11 +71,6 @@ export class WorkflowVersionCoreSyncService { }; }); - await this.purgeSharedIdCoreRows( - workspaceId, - Array.from(coreVersionIdByWorkspaceRecordId.keys()), - ); - await this.coreWorkflowVersionRepository.upsert(workspaceId, coreRows, [ 'id', ]); @@ -109,22 +98,6 @@ export class WorkflowVersionCoreSyncService { await this.invalidateAutomatedTriggerMaps(workspaceId); } - // The pre-soft-ref model wrote core rows with id === workspace record id. - // Delete those before creating own-id rows, otherwise re-migration orphans - // them and a second active row collides on the one-active-per-workflow index. - private async purgeSharedIdCoreRows( - workspaceId: string, - workspaceRecordIds: string[], - ): Promise { - if (workspaceRecordIds.length === 0) { - return; - } - - await this.coreWorkflowVersionRepository.delete(workspaceId, { - id: In(workspaceRecordIds), - }); - } - private async writeBackCoreVersionIds( workspaceId: string, coreVersionIdByWorkspaceRecordId: Map, diff --git a/packages/twenty-server/src/engine/core-modules/workflow/workflow-core.module.ts b/packages/twenty-server/src/engine/core-modules/workflow/workflow-core.module.ts new file mode 100644 index 0000000000..7b96d81180 --- /dev/null +++ b/packages/twenty-server/src/engine/core-modules/workflow/workflow-core.module.ts @@ -0,0 +1,21 @@ +import { Module } from '@nestjs/common'; +import { TypeOrmModule } from '@nestjs/typeorm'; + +import { WorkflowEntity } from 'src/engine/core-modules/workflow/entities/workflow.entity'; +import { WorkflowCoreSyncService } from 'src/engine/core-modules/workflow/services/workflow-core-sync.service'; +import { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity'; +import { provideWorkspaceScopedRepository } from 'src/engine/twenty-orm/workspace-scoped-repository/provide-workspace-scoped-repository'; +import { WorkspaceCacheModule } from 'src/engine/workspace-cache/workspace-cache.module'; + +@Module({ + imports: [ + TypeOrmModule.forFeature([WorkflowEntity, WorkspaceEntity]), + WorkspaceCacheModule, + ], + providers: [ + WorkflowCoreSyncService, + provideWorkspaceScopedRepository(WorkflowEntity), + ], + exports: [TypeOrmModule, WorkflowCoreSyncService], +}) +export class WorkflowCoreModule {} diff --git a/packages/twenty-server/src/modules/workflow/workflow-core-sync/listeners/workflow-core-dual-write.listener.ts b/packages/twenty-server/src/modules/workflow/workflow-core-sync/listeners/workflow-core-dual-write.listener.ts new file mode 100644 index 0000000000..278677fd4b --- /dev/null +++ b/packages/twenty-server/src/modules/workflow/workflow-core-sync/listeners/workflow-core-dual-write.listener.ts @@ -0,0 +1,126 @@ +import { Injectable } from '@nestjs/common'; + +import { + type ObjectRecordCreateEvent, + type ObjectRecordDeleteEvent, + type ObjectRecordDestroyEvent, + type ObjectRecordRestoreEvent, + type ObjectRecordUpdateEvent, +} from 'twenty-shared/database-events'; +import { isDefined } from 'twenty-shared/utils'; + +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 { ExceptionHandlerService } from 'src/engine/core-modules/exception-handler/exception-handler.service'; +import { WorkflowCoreSyncService } from 'src/engine/core-modules/workflow/services/workflow-core-sync.service'; +import { type CustomWorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/custom-workspace-batch-event.type'; +import { type WorkflowWorkspaceEntity } from 'src/modules/workflow/common/standard-objects/workflow.workspace-entity'; + +@Injectable() +export class WorkflowCoreDualWriteListener { + constructor( + private readonly exceptionHandlerService: ExceptionHandlerService, + private readonly workflowCoreSyncService: WorkflowCoreSyncService, + ) {} + + @OnDatabaseBatchEvent('workflow', DatabaseEventAction.CREATED) + async handleCreated( + batchEvent: CustomWorkspaceEventBatch< + ObjectRecordCreateEvent + >, + ): Promise { + await this.upsertToCore( + batchEvent.workspaceId, + batchEvent.events.map((event) => event.properties.after), + ); + } + + @OnDatabaseBatchEvent('workflow', DatabaseEventAction.UPDATED) + async handleUpdated( + batchEvent: CustomWorkspaceEventBatch< + ObjectRecordUpdateEvent + >, + ): Promise { + await this.upsertToCore( + batchEvent.workspaceId, + batchEvent.events.map((event) => event.properties.after), + ); + } + + @OnDatabaseBatchEvent('workflow', DatabaseEventAction.RESTORED) + async handleRestored( + batchEvent: CustomWorkspaceEventBatch< + ObjectRecordRestoreEvent + >, + ): Promise { + await this.upsertToCore( + batchEvent.workspaceId, + batchEvent.events.map((event) => event.properties.after), + ); + } + + @OnDatabaseBatchEvent('workflow', DatabaseEventAction.DELETED) + async handleDeleted( + batchEvent: CustomWorkspaceEventBatch< + ObjectRecordDeleteEvent + >, + ): Promise { + await this.deleteFromCore( + batchEvent.workspaceId, + batchEvent.events + .map((event) => event.properties.before.coreWorkflowId) + .filter(isDefined), + ); + } + + @OnDatabaseBatchEvent('workflow', DatabaseEventAction.DESTROYED) + async handleDestroyed( + batchEvent: CustomWorkspaceEventBatch< + ObjectRecordDestroyEvent + >, + ): Promise { + await this.deleteFromCore( + batchEvent.workspaceId, + batchEvent.events + .map((event) => event.properties.before.coreWorkflowId) + .filter(isDefined), + ); + } + + private async upsertToCore( + workspaceId: string | undefined, + workflows: WorkflowWorkspaceEntity[], + ): Promise { + if (!isDefined(workspaceId)) { + return; + } + + try { + await this.workflowCoreSyncService.upsertToCore(workspaceId, workflows); + } catch (error) { + this.exceptionHandlerService.captureExceptions([error], { + workspace: { id: workspaceId }, + }); + } + } + + private async deleteFromCore( + workspaceId: string | undefined, + coreWorkflowIds: string[], + ): Promise { + if (!isDefined(workspaceId)) { + return; + } + + try { + await this.workflowCoreSyncService.deleteFromCore( + workspaceId, + coreWorkflowIds, + ); + } catch (error) { + this.exceptionHandlerService.captureExceptions([error], { + workspace: { id: workspaceId }, + }); + } + } +} diff --git a/packages/twenty-server/src/modules/workflow/workflow-core-sync/workflow-core-sync.module.ts b/packages/twenty-server/src/modules/workflow/workflow-core-sync/workflow-core-sync.module.ts new file mode 100644 index 0000000000..fef95bb85c --- /dev/null +++ b/packages/twenty-server/src/modules/workflow/workflow-core-sync/workflow-core-sync.module.ts @@ -0,0 +1,10 @@ +import { Module } from '@nestjs/common'; + +import { WorkflowCoreModule } from 'src/engine/core-modules/workflow/workflow-core.module'; +import { WorkflowCoreDualWriteListener } from 'src/modules/workflow/workflow-core-sync/listeners/workflow-core-dual-write.listener'; + +@Module({ + imports: [WorkflowCoreModule], + providers: [WorkflowCoreDualWriteListener], +}) +export class WorkflowCoreSyncModule {} diff --git a/packages/twenty-server/src/modules/workflow/workflow.module.ts b/packages/twenty-server/src/modules/workflow/workflow.module.ts index 485874e4e6..12b79c9117 100644 --- a/packages/twenty-server/src/modules/workflow/workflow.module.ts +++ b/packages/twenty-server/src/modules/workflow/workflow.module.ts @@ -1,5 +1,6 @@ import { Module } from '@nestjs/common'; +import { WorkflowCoreSyncModule } from 'src/modules/workflow/workflow-core-sync/workflow-core-sync.module'; import { WorkflowStatusModule } from 'src/modules/workflow/workflow-status/workflow-status.module'; import { WorkflowTriggerModule } from 'src/modules/workflow/workflow-trigger/workflow-trigger.module'; import { WorkflowVersionCoreSyncModule } from 'src/modules/workflow/workflow-version-core-sync/workflow-version-core-sync.module'; @@ -9,6 +10,7 @@ import { WorkflowVersionCoreSyncModule } from 'src/modules/workflow/workflow-ver WorkflowTriggerModule, WorkflowStatusModule, WorkflowVersionCoreSyncModule, + WorkflowCoreSyncModule, ], }) export class WorkflowModule {}