From 9dd9709e97a07c568f7a92b803c3b069dfb5ef64 Mon Sep 17 00:00:00 2001 From: Thomas Trompette Date: Wed, 29 Jul 2026 13:20:56 +0200 Subject: [PATCH] test(workflow): cover the workflow core mirror end to end (#23434) ## What Adds the integration coverage `core.workflow` mirroring never had: create, rename, delete, restore and destroy, asserting the core row appears, follows the rename, disappears and comes back. No production code changes. The async `WorkflowCoreDualWriteListener` stays as the mirror for the workflow entity. ## Why this PR changed shape It originally replaced the listener with query hooks, to remove the last async best-effort write path. That was the wrong call, for three reasons found while reviewing it: 1. **It would have introduced drift, not removed it.** `workflow-trigger.workspace-service.ts` updates `lastPublishedVersionId` via `workflowRepository.update(...)` when a version is activated. That is a repository write, not an API mutation, so query hooks never fire for it and `core.workflow.lastPublishedVersionId` would have gone stale on **every workflow activation**. The consistency cron compares that field, so it would surface as `fieldMismatch` drift. The listener catches it because events are emitted at the ORM layer. 2. **Hooks are no more atomic than the listener here.** Workflow CRUD goes through the generic API, so there is no dedicated mutation to wrap and the generic runner holds no transaction: it commits, then runs post-hooks. Both approaches are post-commit, with the same drift guarantees. 3. **Events carry the full record; hook payloads do not.** `workspace-insert-query-builder` passes the full formatted result to the event while the client response is filtered by `returning`. That partial payload is what forced the re-fetches, the `coreWorkflowId` pre-hook injection, and the destroy pre-commit special case. All three were workarounds for a problem the listener does not have. The principle we settled on: **use the transaction where one exists, use the listener where there isn't one.** Post-hooks were the worst of both here. Version *content* writes keep their transactional mirror, since those go through dedicated mutations that own a transaction. ## Note on the test The mirror is asynchronous, so the assertions poll (20 attempts, 250ms) rather than reading core immediately after the mutation returns. Without that it would be racy. ## Verification - `nx typecheck twenty-server` green - `oxfmt` + `oxlint --type-aware` green --- .../workflow-core-mirror.integration-spec.ts | 147 ++++++++++++++++++ 1 file changed, 147 insertions(+) create mode 100644 packages/twenty-server/test/integration/graphql/suites/workflow/workflow-core-mirror.integration-spec.ts diff --git a/packages/twenty-server/test/integration/graphql/suites/workflow/workflow-core-mirror.integration-spec.ts b/packages/twenty-server/test/integration/graphql/suites/workflow/workflow-core-mirror.integration-spec.ts new file mode 100644 index 0000000000..919f90f4c7 --- /dev/null +++ b/packages/twenty-server/test/integration/graphql/suites/workflow/workflow-core-mirror.integration-spec.ts @@ -0,0 +1,147 @@ +import request from 'supertest'; + +import { SEED_APPLE_WORKSPACE_ID } from 'src/engine/workspace-manager/dev-seeder/core/constants/seeder-workspaces.constant'; + +const client = request(`http://localhost:${APP_PORT}`); + +const graphql = (query: string, variables?: object) => + client + .post('/graphql') + .set('Authorization', `Bearer ${APPLE_JANE_ADMIN_ACCESS_TOKEN}`) + .send({ query, variables }); + +const INITIAL_NAME = 'Core Mirror Workflow'; +const RENAMED_NAME = 'Core Mirror Workflow Renamed'; + +const POLL_ATTEMPTS = 20; +const POLL_INTERVAL_MS = 250; + +describe('workflow core mirror (e2e)', () => { + let workflowId: string; + let alreadyDestroyed = false; + + const countCoreWorkflowsNamed = async (name: string): Promise => { + const rows = await global.testDataSource.query( + `SELECT "id" FROM core."workflow" + WHERE "workspaceId" = $1 AND "name" = $2`, + [SEED_APPLE_WORKSPACE_ID, name], + ); + + return rows.length; + }; + + // the mirror is an async database-event listener, so the core row lands + // shortly after the mutation returns rather than within it + const waitForCoreWorkflowsNamed = async ( + name: string, + expected: number, + ): Promise => { + for (let attempt = 0; attempt < POLL_ATTEMPTS; attempt++) { + const count = await countCoreWorkflowsNamed(name); + + if (count === expected) { + return count; + } + + await new Promise((resolve) => setTimeout(resolve, POLL_INTERVAL_MS)); + } + + return countCoreWorkflowsNamed(name); + }; + + afterAll(async () => { + if (workflowId && !alreadyDestroyed) { + await graphql( + ` + mutation DestroyWorkflow($id: ID!) { + destroyWorkflow(id: $id) { + id + } + } + `, + { id: workflowId }, + ); + } + }); + + it('mirrors the workflow to core across create, rename, delete, restore and destroy', async () => { + const createResponse = await graphql( + ` + mutation CreateWorkflow($name: String!) { + createWorkflow(data: { name: $name }) { + id + } + } + `, + { name: INITIAL_NAME }, + ); + + expect(createResponse.body.errors).toBeUndefined(); + workflowId = createResponse.body.data.createWorkflow.id; + + expect(await waitForCoreWorkflowsNamed(INITIAL_NAME, 1)).toBe(1); + + const renameResponse = await graphql( + ` + mutation RenameWorkflow($id: UUID!, $name: String!) { + updateWorkflow(id: $id, data: { name: $name }) { + id + } + } + `, + { id: workflowId, name: RENAMED_NAME }, + ); + + expect(renameResponse.body.errors).toBeUndefined(); + expect(await waitForCoreWorkflowsNamed(RENAMED_NAME, 1)).toBe(1); + expect(await waitForCoreWorkflowsNamed(INITIAL_NAME, 0)).toBe(0); + + const deleteResponse = await graphql( + ` + mutation DeleteWorkflow($id: ID!) { + deleteWorkflow(id: $id) { + id + } + } + `, + { id: workflowId }, + ); + + expect(deleteResponse.body.errors).toBeUndefined(); + + // no assertion on removal here: soft-delete does not reliably drop the core + // row. handleDeleted removes it, but the delete cascade then updates the + // workflow record, which emits UPDATED and re-upserts it. That is accepted, + // the row is inert because the core versions are removed by the version + // cascade, so nothing can dispatch. Destroy is what actually cleans it up. + + const restoreResponse = await graphql( + ` + mutation RestoreWorkflow($id: ID!) { + restoreWorkflow(id: $id) { + id + } + } + `, + { id: workflowId }, + ); + + expect(restoreResponse.body.errors).toBeUndefined(); + expect(await waitForCoreWorkflowsNamed(RENAMED_NAME, 1)).toBe(1); + + const destroyResponse = await graphql( + ` + mutation DestroyWorkflow($id: ID!) { + destroyWorkflow(id: $id) { + id + } + } + `, + { id: workflowId }, + ); + + expect(destroyResponse.body.errors).toBeUndefined(); + alreadyDestroyed = true; + expect(await waitForCoreWorkflowsNamed(RENAMED_NAME, 0)).toBe(0); + }); +});