feat(workflow): backfill + dual-write for core workflow entity (#22776)

Workflow-side soft-ref sync — the `coreWorkflowId` mirror of the merged version side (#22821 / #22940 / #22944 / #22961). Rebased onto current main; supersedes the original shared-UUID version of this PR. ## What Gives `core.workflow` a per-workspace copy of each workflow (`name`, `lastPublishedVersionId`), soft-reffed from the workspace record via `coreWorkflowId`, so app-shipped workflows have a core home. Does not touch reads/dispatch (that's Phase B). - **Sync service** (`WorkflowCoreSyncService`): core rows get their own id (`uuidv5(workspaceId:recordId)`), the workspace record links via `coreWorkflowId` (written back after the upsert), and the write-back is **guarded** on the `coreWorkflowId` field being present (skips with a warning otherwise — mirrors #22940). Injected repo renamed `coreWorkflowRepository`. - **Dual-write listener** on the `workflow` object: CREATED/UPDATED/RESTORED upsert, DELETED/DESTROYED delete by `coreWorkflowId`. Always-on; failures routed to Sentry so they never break the user write. - **2-20 backfill** (`backfill-workflow-to-core`): reads via the provided `RunOnWorkspaceArgs.dataSource`, upserts each workspace workflow into core. - **2-22 provisioning** (mirrors #22944/#22961): - `add-workflow-core-soft-ref-field`: adds the `coreWorkflowId` system field on existing workspaces (flat-entity legacy migration). - `backfill-workflow-core-links`: full rebuild — per workspace, in one raw-SQL transaction, wipes all `core.workflow` rows, inserts a fresh own-id row per workflow, and re-links every record. (No trigger-map cache to invalidate on `core.workflow`.) Simpler than the version side: `core.workflow` has no one-active-per-workflow index and no trigger-map cache, and it was never backfilled in prod, so there are no legacy shared-id rows. ## Test Fresh `database:reset` + full sequence (2-20 backfill → 2-22 add-field → 2-22 rebuild link): 4/4 workspace records linked via `coreWorkflowId` (id != coreWorkflowId), links resolve, **0 dangling**, no duplicate core rows, names populated. Typecheck + lint + oxfmt clean.
2026-07-20 14:04:32 +02:00
parent c04714b9e0
commit 87729a2822
12 changed files with 768 additions and 38 deletions
@@ -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> = {},
+  ): 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<WorkflowEntity>,
+      workspaceRepository as unknown as Repository<WorkspaceEntity>,
+      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<WorkflowWorkspaceEntity>),
+    ]);
+
+    expect(upsertedRows()[0].id).toBe(coreWorkflowId);
+    expect(
+      globalWorkspaceOrmManager.executeInWorkspaceContext,
+    ).not.toHaveBeenCalled();
+  });
+});
@@ -25,14 +25,18 @@ describe('WorkflowVersionCoreSyncService', () => {
    invalidateAndRecompute: jest.Mock;
  };

-  const buildVersion = (): WorkflowVersionWorkspaceEntity =>
+  const buildVersion = (
+    overrides: Partial<WorkflowVersionWorkspaceEntity> = {},
+  ): 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<WorkflowVersionWorkspaceEntity>),
+    ]);
+
+    expect(upsertedRows()[0].id).toBe(coreWorkflowVersionId);
+    expect(
+      globalWorkspaceOrmManager.executeInWorkspaceContext,
+    ).not.toHaveBeenCalled();
+  });
 });
@@ -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<WorkflowEntity>,
+    @InjectRepository(WorkspaceEntity)
+    private readonly workspaceRepository: Repository<WorkspaceEntity>,
+    private readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager,
+    private readonly workspaceCacheService: WorkspaceCacheService,
+  ) {}
+
+  async upsertToCore(
+    workspaceId: string,
+    workflows: WorkflowWorkspaceEntity[],
+  ): Promise<void> {
+    if (workflows.length === 0) {
+      return;
+    }
+
+    const applicationId = await this.getCustomApplicationIdOrThrow(workspaceId);
+
+    const coreWorkflowIdByWorkspaceRecordId = new Map<string, string>();
+
+    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<void> {
+    if (coreWorkflowIds.length === 0) {
+      return;
+    }
+
+    await this.coreWorkflowRepository.delete(workspaceId, {
+      id: In(coreWorkflowIds),
+    });
+  }
+
+  private async writeBackCoreWorkflowIds(
+    workspaceId: string,
+    coreWorkflowIdByWorkspaceRecordId: Map<string, string>,
+  ): Promise<void> {
+    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<WorkflowWorkspaceEntity>(
+          workspaceId,
+          'workflow',
+          { shouldBypassPermissionChecks: true },
+        );
+
+      for (const [
+        workspaceRecordId,
+        coreWorkflowId,
+      ] of coreWorkflowIdByWorkspaceRecordId) {
+        await workspaceWorkflowRepository.update(workspaceRecordId, {
+          coreWorkflowId,
+        });
+      }
+    }, buildSystemAuthContext(workspaceId));
+  }
+
+  private async workspaceHasCoreWorkflowIdField(
+    workspaceId: string,
+  ): Promise<boolean> {
+    const { flatFieldMetadataMaps } =
+      await this.workspaceCacheService.getOrRecompute(workspaceId, [
+        'flatFieldMetadataMaps',
+      ]);
+
+    return isDefined(
+      flatFieldMetadataMaps.byUniversalIdentifier[
+        STANDARD_OBJECTS.workflow.fields.coreWorkflowId.universalIdentifier
+      ],
+    );
+  }
+
+  private async getCustomApplicationIdOrThrow(
+    workspaceId: string,
+  ): Promise<string> {
+    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;
+  }
+}
@@ -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<string, string>();

    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<void> {
-    if (workspaceRecordIds.length === 0) {
-      return;
-    }
-
-    await this.coreWorkflowVersionRepository.delete(workspaceId, {
-      id: In(workspaceRecordIds),
-    });
-  }
-
  private async writeBackCoreVersionIds(
    workspaceId: string,
    coreVersionIdByWorkspaceRecordId: Map<string, string>,
@@ -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 {}