feat(workflow): backfill workspace workflowVersion into core (phase A) (#22663)

## workflowVersion -> core, Phase A Follows #21674 (Phase 0, merged). Base: `main`. Populates core `workflowVersion` and keeps it in sync with the workspace object, so a later phase can switch reads to core. Reads stay on the workspace object in this PR. ### 1. Backfill (upgrade command) `BackfillWorkflowVersionToCoreCommand`, a `@RegisteredWorkspaceCommand('2.20.0', ...)`. Per workspace, reads all workspace `workflowVersion` records and upserts them into core, preserving ids (idempotent), dry-run aware. ### 2. Dual-write (always on, not flag-gated) `WorkflowVersionCoreDualWriteListener` hooks `@OnDatabaseBatchEvent('workflowVersion', CREATED/UPDATED/DELETED)` (same mechanism as the existing workflow-version status listener) and mirrors every mutation into core. Sync failures are logged, never break the user's write; drift is repaired by re-running the backfill command. Dual-write is deliberately not behind a flag: reading from core (next phase) is only safe if core has been continuously in sync since the backfill. An always-on mirror makes "core is fresh" an invariant, so the read switch becomes a plain flag flip. The cost is one extra upsert on infrequent workflowVersion writes. `IS_WORKFLOW_VERSION_IN_CORE_ENABLED` is reserved for the read switch (Phase B): dispatch from the `workflowAutomatedTriggerMaps` cache, runner and builder reading trigger/steps from core. Until then it gates nothing. Both the backfill and the listener go through a single `WorkflowVersionCoreSyncService` (`upsertToCore`/`deleteFromCore`): the workspace-to-core mapping (`trigger` -> `triggers[]`, plus `steps`, `status`, `workflowId`) and `workflowAutomatedTriggerMaps` invalidation live in one place. ### Rollout plan (following phases) - **B, read switch (flag per workspace):** reads move to core; writes keep flowing workspace -> listener -> core. Rollback = flip the flag back, workspace never stopped being source of truth. - **C, contract (code change):** write paths write trigger/steps to core directly; workspace `workflowVersion` stays as a thin shell (nav/relations/search) but drops the trigger/steps columns; listener and flag removed. ### Not in this PR - Reconciliation tooling beyond re-running the backfill. - The read switch (Phase B).
2026-07-08 18:45:57 +02:00
parent 0ea0c23556
commit ca90a9358f
7 changed files with 306 additions and 4 deletions
@@ -0,0 +1,125 @@
+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 { WorkflowVersionCoreSyncService } from 'src/engine/core-modules/workflow/services/workflow-version-core-sync.service';
+import { type CustomWorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/custom-workspace-batch-event.type';
+import { type WorkflowVersionWorkspaceEntity } from 'src/modules/workflow/common/standard-objects/workflow-version.workspace-entity';
+
+@Injectable()
+export class WorkflowVersionCoreDualWriteListener {
+  constructor(
+    private readonly exceptionHandlerService: ExceptionHandlerService,
+    private readonly workflowVersionCoreSyncService: WorkflowVersionCoreSyncService,
+  ) {}
+
+  @OnDatabaseBatchEvent('workflowVersion', DatabaseEventAction.CREATED)
+  async handleCreated(
+    batchEvent: CustomWorkspaceEventBatch<
+      ObjectRecordCreateEvent<WorkflowVersionWorkspaceEntity>
+    >,
+  ): Promise<void> {
+    await this.upsertToCore(
+      batchEvent.workspaceId,
+      batchEvent.events.map((event) => event.properties.after),
+    );
+  }
+
+  @OnDatabaseBatchEvent('workflowVersion', DatabaseEventAction.UPDATED)
+  async handleUpdated(
+    batchEvent: CustomWorkspaceEventBatch<
+      ObjectRecordUpdateEvent<WorkflowVersionWorkspaceEntity>
+    >,
+  ): Promise<void> {
+    await this.upsertToCore(
+      batchEvent.workspaceId,
+      batchEvent.events.map((event) => event.properties.after),
+    );
+  }
+
+  @OnDatabaseBatchEvent('workflowVersion', DatabaseEventAction.RESTORED)
+  async handleRestored(
+    batchEvent: CustomWorkspaceEventBatch<
+      ObjectRecordRestoreEvent<WorkflowVersionWorkspaceEntity>
+    >,
+  ): Promise<void> {
+    await this.upsertToCore(
+      batchEvent.workspaceId,
+      batchEvent.events.map((event) => event.properties.after),
+    );
+  }
+
+  @OnDatabaseBatchEvent('workflowVersion', DatabaseEventAction.DELETED)
+  async handleDeleted(
+    batchEvent: CustomWorkspaceEventBatch<
+      ObjectRecordDeleteEvent<WorkflowVersionWorkspaceEntity>
+    >,
+  ): Promise<void> {
+    await this.deleteFromCore(
+      batchEvent.workspaceId,
+      batchEvent.events.map((event) => event.properties.before.id),
+    );
+  }
+
+  @OnDatabaseBatchEvent('workflowVersion', DatabaseEventAction.DESTROYED)
+  async handleDestroyed(
+    batchEvent: CustomWorkspaceEventBatch<
+      ObjectRecordDestroyEvent<WorkflowVersionWorkspaceEntity>
+    >,
+  ): Promise<void> {
+    await this.deleteFromCore(
+      batchEvent.workspaceId,
+      batchEvent.events.map((event) => event.properties.before.id),
+    );
+  }
+
+  private async upsertToCore(
+    workspaceId: string | undefined,
+    workflowVersions: WorkflowVersionWorkspaceEntity[],
+  ): Promise<void> {
+    if (!isDefined(workspaceId)) {
+      return;
+    }
+
+    try {
+      await this.workflowVersionCoreSyncService.upsertToCore(
+        workspaceId,
+        workflowVersions,
+      );
+    } catch (error) {
+      this.exceptionHandlerService.captureExceptions([error], {
+        workspace: { id: workspaceId },
+      });
+    }
+  }
+
+  private async deleteFromCore(
+    workspaceId: string | undefined,
+    workflowVersionIds: string[],
+  ): Promise<void> {
+    if (!isDefined(workspaceId)) {
+      return;
+    }
+
+    try {
+      await this.workflowVersionCoreSyncService.deleteFromCore(
+        workspaceId,
+        workflowVersionIds,
+      );
+    } catch (error) {
+      this.exceptionHandlerService.captureExceptions([error], {
+        workspace: { id: workspaceId },
+      });
+    }
+  }
+}
@@ -0,0 +1,10 @@
+import { Module } from '@nestjs/common';
+
+import { WorkflowVersionCoreModule } from 'src/engine/core-modules/workflow/workflow-version-core.module';
+import { WorkflowVersionCoreDualWriteListener } from 'src/modules/workflow/workflow-version-core-sync/listeners/workflow-version-core-dual-write.listener';
+
+@Module({
+  imports: [WorkflowVersionCoreModule],
+  providers: [WorkflowVersionCoreDualWriteListener],
+})
+export class WorkflowVersionCoreSyncModule {}
@@ -2,8 +2,13 @@ import { Module } from '@nestjs/common';

 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';

@Module({
-  imports: [WorkflowTriggerModule, WorkflowStatusModule],
+  imports: [
+    WorkflowTriggerModule,
+    WorkflowStatusModule,
+    WorkflowVersionCoreSyncModule,
+  ],
 })
 export class WorkflowModule {}