houseassassin/twenty
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(workflow): mirror all workflowVersion writes to core in-transaction, drop async create/update dual-write (#23243)

Browse Source 
Switches the workflowVersion -> core dual-write from the async,
best-effort listener to a **transactional mirror**, wires every
content-write funnel through it, and drops the async create/update
handlers. Core can no longer drift from the workspace on any covered
path: the core copy commits or rolls back atomically with the workspace
write.

## Helper (`WorkflowVersionCoreSyncService`)
Two entry points, both writing `core.workflowVersion` and stamping the
`coreWorkflowVersionId` soft-ref on the **caller's transaction
manager**:

- `writeWorkflowVersionAndMirror(workspaceId, write)` - for funnels that
don't own a transaction. Opens a workspace queryRunner, runs the
caller's workspace write on that manager, re-reads the row, mirrors to
core in the same tx, commits, then invalidates the trigger-map cache
post-commit.
- `mirrorWorkflowVersionWrite({ workspaceId, entityManager,
workflowVersion })` - for funnels that already own a queryRunner tx
(activation / deactivation / delete-cascade); they just gain this one
call on their existing manager before commit.

`invalidateAutomatedTriggerMaps` is public so tx-owning callers run it
post-commit.

## Funnels wired (all content writes now mirror in-transaction)
- `updateWorkflowVersionStepsAndTrigger` (central builder step/trigger
edit)
- edge create/delete (4 trigger/step writes)
- `createDraftFromWorkflowVersion` (update + insert),
`duplicateWorkflow` (content update), `updateWorkflowVersionPositions`
- iterator / if-else empty-node step writes
- `workflow.createOne` / `createMany` post-hooks (v1 draft insert)
- AI `create_complete_workflow` tool (v1 insert)
- activation / deactivation status writes (ACTIVE / ARCHIVED /
DEACTIVATED) on their existing tx
- `deactivateVersionOnDelete` cascade (ACTIVE -> DEACTIVATED) on its
existing tx

Direct `workflowVersion.createOne/createMany` is forbidden by a
pre-hook, so there is no un-funneled create path.

## Async listener trimmed
`handleCreated` and `handleUpdated` are removed: every create/update now
mirrors in-transaction, so the post-commit handlers were redundant and
were the source of the rollback-drift (an edit that rolls back still
emitted an `UPDATED` event carrying the uncommitted payload, which the
async listener wrote to core).

`handleRestored`, `handleDeleted`, `handleDestroyed` are **kept**.
Soft-delete/restore go through the generic ORM (not a funnel):
`handleDeleted` drops the core row on soft-delete, and `handleRestored`
recreates it on restore (the restore path just calls
`workflowVersionRepository.restore()`). Removing the restore handler
would leave restored versions with no core row, so the delete/restore
pair stays async.

## Why the core write is raw SQL
`core.workflowVersion` is on the core DataSource, not the workspace
DataSource, so a repository can't be pointed at it from the workspace
queryRunner's manager. But both schemas are one Postgres DB and a
queryRunner is a single connection, so a schema-qualified `INSERT INTO
core."workflowVersion" ... ON CONFLICT` on that manager participates in
the workspace transaction (the prefill util's pattern). The workspace
write and soft-ref write-back go through the ORM's
`repository.update(criteria, data, undefined, queryRunner.manager)`, so
the source-of-truth workspace write keeps its ORM machinery (actor
stamping, search vector, events); only the dumb core mirror is raw,
confined to the helper.

**jsonb:** the raw insert has no entity transformer, so
`triggers`/`steps` are passed as JSON strings (Postgres parses them) -
the inverse of the prefill core insert (the #23204 double-encode trap),
covered by the test. `universalIdentifier` is minted only for a new
link; on conflict only `triggers`/`steps`/`status` are updated.

## Test
In-process integration test: opens a workspace queryRunner, calls the
helper, asserts the core row is visible inside the tx with correct
native jsonb, rolls back, asserts the core row is gone - empirically
confirming one workspace queryRunner writes `core.*` in the same tx.

## Review feedback addressed
- **Duplicate atomicity:** `duplicateWorkflow` now inserts the draft
version with its final steps/trigger inside
`writeWorkflowVersionAndMirror`, so a mirror failure rolls back the
whole version instead of leaving an unmirrored empty draft.
- **Delete cascade:** `deactivateVersionOnDelete` moved its command-menu
cleanup after commit, so a rolled-back deactivation can no longer strip
the menu item from a still-active version.
- **Rollback drift / unlinked rows:** resolved structurally by the
in-transaction mirror + removal of the async CREATED/UPDATED handlers,
and by the soft-ref-field guard that skips mirroring (returns null) on
workspaces without `coreWorkflowVersionId`.
- **Unit suites:** the step-helpers, step-operations and edge suites now
provide a `WorkflowVersionCoreSyncService` mock whose
`writeWorkflowVersionAndMirror` runs the write callback against the test
repo; all three pass (35 tests).

## Verification
Rebased onto `origin/main`. `nx typecheck twenty-server` is green, the
three affected unit suites pass (35 tests), and every changed file
passes type-aware oxlint + oxfmt. Integration suites were failing only
on behind-main DB-init drift (`core.keyValuePair` / data-migration),
which the rebase resolves. Live end-to-end test on a running instance
still pending.
This commit is contained in:
Thomas Trompette
2026-07-27 11:17:17 +02:00
committed by GitHub
parent a96dc335ab
commit 46a3a83866
24 changed files with 888 additions and 287 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/twenty-server/src/engine/core-modules/workflow/services/workflow-version-core-sync.service.ts
+189 -3
View File
@@ -12,7 +12,9 @@ import {
WorkflowVersionStatus,
} from 'src/engine/core-modules/workflow/entities/workflow-version.entity';
import { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity';
import { type WorkspaceEntityManager } from 'src/engine/twenty-orm/entity-manager/workspace-entity-manager';
import { GlobalWorkspaceOrmManager } from 'src/engine/twenty-orm/global-workspace-datasource/global-workspace-orm.manager';
import { WorkspaceRepository } from 'src/engine/twenty-orm/repository/workspace.repository';
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';
@@ -98,6 +100,171 @@ export class WorkflowVersionCoreSyncService {
await this.invalidateAutomatedTriggerMaps(workspaceId);
}
async mirrorWorkflowVersionWrite({
workspaceId,
entityManager,
workflowVersion,
applicationId,
}: {
workspaceId: string;
entityManager: WorkspaceEntityManager;
workflowVersion: WorkflowVersionWorkspaceEntity;
applicationId?: string;
}): Promise<{ coreWorkflowVersionId: string } | null> {
if (!(await this.workspaceHasCoreWorkflowVersionIdField(workspaceId))) {
this.logger.warn(
`workflowVersion.coreWorkflowVersionId field missing for workspace ${workspaceId}, skipping transactional core mirror`,
);
return null;
}
const resolvedApplicationId =
applicationId ?? (await this.getCustomApplicationIdOrThrow(workspaceId));
const isNewLink = !isNonEmptyString(workflowVersion.coreWorkflowVersionId);
const coreWorkflowVersionId = isNonEmptyString(
workflowVersion.coreWorkflowVersionId,
)
? workflowVersion.coreWorkflowVersionId
: uuidv4();
const queryRunner = entityManager.queryRunner;
if (!isDefined(queryRunner)) {
throw new Error(
'Transactional core mirror requires a transaction-scoped entity manager',
);
}
await queryRunner.query(
`INSERT INTO core."workflowVersion"
("id", "workspaceId", "workflowId", "triggers", "steps", "status", "universalIdentifier", "applicationId")
VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
ON CONFLICT ("id") DO UPDATE SET
"triggers" = EXCLUDED."triggers",
"steps" = EXCLUDED."steps",
"status" = EXCLUDED."status"`,
[
coreWorkflowVersionId,
workspaceId,
workflowVersion.workflowId,
isDefined(workflowVersion.trigger)
? JSON.stringify([workflowVersion.trigger])
: null,
isDefined(workflowVersion.steps)
? JSON.stringify(workflowVersion.steps)
: null,
workflowVersion.status,
uuidv4(),
resolvedApplicationId,
],
);
if (isNewLink) {
await this.writeBackCoreVersionIdOnManager(
workspaceId,
workflowVersion.id,
coreWorkflowVersionId,
entityManager,
);
}
return { coreWorkflowVersionId };
}
async mirrorWorkflowVersionWrites({
workspaceId,
entityManager,
workflowVersions,
}: {
workspaceId: string;
entityManager: WorkspaceEntityManager;
workflowVersions: WorkflowVersionWorkspaceEntity[];
}): Promise<Map<string, string>> {
const coreIdByWorkspaceRecordId = new Map<string, string>();
if (workflowVersions.length === 0) {
return coreIdByWorkspaceRecordId;
}
const applicationId = await this.getCustomApplicationIdOrThrow(workspaceId);
for (const workflowVersion of workflowVersions) {
const result = await this.mirrorWorkflowVersionWrite({
workspaceId,
entityManager,
workflowVersion,
applicationId,
});
if (isDefined(result)) {
coreIdByWorkspaceRecordId.set(
workflowVersion.id,
result.coreWorkflowVersionId,
);
}
}
return coreIdByWorkspaceRecordId;
}
async writeWorkflowVersionAndMirror(
workspaceId: string,
write: (
workflowVersionRepository: WorkspaceRepository<WorkflowVersionWorkspaceEntity>,
entityManager: WorkspaceEntityManager,
) => Promise<string>,
): Promise<void> {
await this.globalWorkspaceOrmManager.executeInWorkspaceContext(async () => {
const workflowVersionRepository =
await this.globalWorkspaceOrmManager.getRepository<WorkflowVersionWorkspaceEntity>(
workspaceId,
'workflowVersion',
{ shouldBypassPermissionChecks: true },
);
const dataSource =
await this.globalWorkspaceOrmManager.getGlobalWorkspaceDataSource();
const queryRunner = dataSource.createQueryRunner();
await queryRunner.connect();
await queryRunner.startTransaction();
try {
const workflowVersionId = await write(
workflowVersionRepository,
queryRunner.manager,
);
const workflowVersion = await workflowVersionRepository.findOne(
{ where: { id: workflowVersionId } },
queryRunner.manager,
);
if (isDefined(workflowVersion)) {
await this.mirrorWorkflowVersionWrite({
workspaceId,
entityManager: queryRunner.manager,
workflowVersion,
});
}
await queryRunner.commitTransaction();
} catch (error) {
if (queryRunner.isTransactionActive) {
await queryRunner.rollbackTransaction();
}
throw error;
} finally {
await queryRunner.release();
}
}, buildSystemAuthContext(workspaceId));
await this.invalidateAutomatedTriggerMaps(workspaceId);
}
private async writeBackCoreVersionIds(
workspaceId: string,
coreVersionIdByWorkspaceRecordId: Map<string, string>,
@@ -133,6 +300,27 @@ export class WorkflowVersionCoreSyncService {
}, buildSystemAuthContext(workspaceId));
}
private async writeBackCoreVersionIdOnManager(
workspaceId: string,
workflowVersionId: string,
coreWorkflowVersionId: string,
entityManager: WorkspaceEntityManager,
): Promise<void> {
const workspaceWorkflowVersionRepository =
await this.globalWorkspaceOrmManager.getRepository<WorkflowVersionWorkspaceEntity>(
workspaceId,
'workflowVersion',
{ shouldBypassPermissionChecks: true },
);
await workspaceWorkflowVersionRepository.update(
{ id: workflowVersionId },
{ coreWorkflowVersionId },
undefined,
entityManager,
);
}
private async workspaceHasCoreWorkflowVersionIdField(
workspaceId: string,
): Promise<boolean> {
@@ -166,9 +354,7 @@ export class WorkflowVersionCoreSyncService {
return workspace.workspaceCustomApplicationId;
}
private async invalidateAutomatedTriggerMaps(
workspaceId: string,
): Promise<void> {
async invalidateAutomatedTriggerMaps(workspaceId: string): Promise<void> {
await this.workspaceCacheService.invalidateAndRecompute(workspaceId, [
'workflowAutomatedTriggerMaps',
]);
packages/twenty-server/src/modules/workflow/common/query-hooks/workflow-create-many.post-query.hook.ts
+35 -28
View File
@@ -6,6 +6,7 @@ import { WorkspaceQueryHook } from 'src/engine/api/graphql/workspace-query-runne
import { WorkspaceQueryHookType } from 'src/engine/api/graphql/workspace-query-runner/workspace-query-hook/types/workspace-query-hook.type';
import { type WorkspaceAuthContext } from 'src/engine/core-modules/auth/types/workspace-auth-context.type';
import { RecordPositionService } from 'src/engine/core-modules/record-position/services/record-position.service';
import { WorkflowVersionCoreSyncService } from 'src/engine/core-modules/workflow/services/workflow-version-core-sync.service';
import { WorkspaceNotFoundDefaultError } from 'src/engine/core-modules/workspace/workspace.exception';
import { GlobalWorkspaceOrmManager } from 'src/engine/twenty-orm/global-workspace-datasource/global-workspace-orm.manager';
import {
@@ -22,6 +23,7 @@ export class WorkflowCreateManyPostQueryHook implements WorkspacePostQueryHookIn
constructor(
private readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager,
private readonly recordPositionService: RecordPositionService,
private readonly workflowVersionCoreSyncService: WorkflowVersionCoreSyncService,
) {}
async execute(
@@ -33,34 +35,39 @@ export class WorkflowCreateManyPostQueryHook implements WorkspacePostQueryHookIn
assertIsDefinedOrThrow(workspace, WorkspaceNotFoundDefaultError);
await this.globalWorkspaceOrmManager.executeInWorkspaceContext(async () => {
const workflowVersionRepository =
await this.globalWorkspaceOrmManager.getRepository<WorkflowVersionWorkspaceEntity>(
workspace.id,
'workflowVersion',
);
const position = await this.recordPositionService.buildRecordPosition({
value: 'first',
objectMetadata: {
isCustom: false,
nameSingular: 'workflowVersion',
},
workspaceId: workspace.id,
});
const workflowVersionsToCreate = payload.map((workflow) => ({
workflowId: workflow.id,
status: WorkflowVersionStatus.DRAFT,
name: 'v1',
position,
}));
await Promise.all(
workflowVersionsToCreate.map((workflowVersion) => {
return workflowVersionRepository.insert(workflowVersion);
}),
const position =
await this.globalWorkspaceOrmManager.executeInWorkspaceContext(
() =>
this.recordPositionService.buildRecordPosition({
value: 'first',
objectMetadata: {
isCustom: false,
nameSingular: 'workflowVersion',
},
workspaceId: workspace.id,
}),
authContext,
);
}, authContext);
for (const workflow of payload) {
await this.workflowVersionCoreSyncService.writeWorkflowVersionAndMirror(
workspace.id,
async (workflowVersionRepository, entityManager) => {
const insertResult = await workflowVersionRepository.insert(
{
workflowId: workflow.id,
status: WorkflowVersionStatus.DRAFT,
name: 'v1',
position,
},
entityManager,
);
return (
insertResult.generatedMaps[0] as WorkflowVersionWorkspaceEntity
).id;
},
);
}
}
}
packages/twenty-server/src/modules/workflow/common/query-hooks/workflow-create-one.post-query.hook.ts
+26 -23
View File
@@ -6,8 +6,8 @@ import { WorkspaceQueryHook } from 'src/engine/api/graphql/workspace-query-runne
import { WorkspaceQueryHookType } from 'src/engine/api/graphql/workspace-query-runner/workspace-query-hook/types/workspace-query-hook.type';
import { type WorkspaceAuthContext } from 'src/engine/core-modules/auth/types/workspace-auth-context.type';
import { RecordPositionService } from 'src/engine/core-modules/record-position/services/record-position.service';
import { WorkflowVersionCoreSyncService } from 'src/engine/core-modules/workflow/services/workflow-version-core-sync.service';
import { WorkspaceNotFoundDefaultError } from 'src/engine/core-modules/workspace/workspace.exception';
import { GlobalWorkspaceOrmManager } from 'src/engine/twenty-orm/global-workspace-datasource/global-workspace-orm.manager';
import {
WorkflowVersionStatus,
type WorkflowVersionWorkspaceEntity,
@@ -20,8 +20,8 @@ import { type WorkflowWorkspaceEntity } from 'src/modules/workflow/common/standa
})
export class WorkflowCreateOnePostQueryHook implements WorkspacePostQueryHookInstance {
constructor(
private readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager,
private readonly recordPositionService: RecordPositionService,
private readonly workflowVersionCoreSyncService: WorkflowVersionCoreSyncService,
) {}
async execute(
@@ -35,28 +35,31 @@ export class WorkflowCreateOnePostQueryHook implements WorkspacePostQueryHookIns
const workflow = payload[0];
await this.globalWorkspaceOrmManager.executeInWorkspaceContext(async () => {
const workflowVersionRepository =
await this.globalWorkspaceOrmManager.getRepository<WorkflowVersionWorkspaceEntity>(
workspace.id,
'workflowVersion',
await this.workflowVersionCoreSyncService.writeWorkflowVersionAndMirror(
workspace.id,
async (workflowVersionRepository, entityManager) => {
const position = await this.recordPositionService.buildRecordPosition({
value: 'first',
objectMetadata: {
isCustom: false,
nameSingular: 'workflowVersion',
},
workspaceId: workspace.id,
});
const insertResult = await workflowVersionRepository.insert(
{
workflowId: workflow.id,
status: WorkflowVersionStatus.DRAFT,
name: 'v1',
position,
},
entityManager,
);
const position = await this.recordPositionService.buildRecordPosition({
value: 'first',
objectMetadata: {
isCustom: false,
nameSingular: 'workflowVersion',
},
workspaceId: workspace.id,
});
await workflowVersionRepository.insert({
workflowId: workflow.id,
status: WorkflowVersionStatus.DRAFT,
name: 'v1',
position,
});
}, authContext);
return (insertResult.generatedMaps[0] as WorkflowVersionWorkspaceEntity)
.id;
},
);
}
}
packages/twenty-server/src/modules/workflow/common/query-hooks/workflow-query-hook.module.ts
+2
View File
@@ -3,6 +3,7 @@ import { Module } from '@nestjs/common';
import { CommandMenuItemModule } from 'src/engine/metadata-modules/command-menu-item/command-menu-item.module';
import { FeatureFlagModule } from 'src/engine/core-modules/feature-flag/feature-flag.module';
import { RecordPositionModule } from 'src/engine/core-modules/record-position/record-position.module';
import { WorkflowVersionCoreModule } from 'src/engine/core-modules/workflow/workflow-version-core.module';
import { WorkspaceManyOrAllFlatEntityMapsCacheModule } from 'src/engine/metadata-modules/flat-entity/services/workspace-many-or-all-flat-entity-maps-cache.module';
import { LogicFunctionModule } from 'src/engine/metadata-modules/logic-function/logic-function.module';
import { ObjectMetadataModule } from 'src/engine/metadata-modules/object-metadata/object-metadata.module';
@@ -49,6 +50,7 @@ import { WorkflowVersionValidationWorkspaceService } from 'src/modules/workflow/
CodeStepBuildModule,
CommandMenuItemModule,
FeatureFlagModule,
WorkflowVersionCoreModule,
],
providers: [
WorkflowCreateOnePreQueryHook,
packages/twenty-server/src/modules/workflow/common/workflow-common.module.ts
+2
View File
@@ -4,6 +4,7 @@ import { CommandMenuItemModule } from 'src/engine/metadata-modules/command-menu-
import { FeatureFlagModule } from 'src/engine/core-modules/feature-flag/feature-flag.module';
import { WorkspaceManyOrAllFlatEntityMapsCacheModule } from 'src/engine/metadata-modules/flat-entity/services/workspace-many-or-all-flat-entity-maps-cache.module';
import { LogicFunctionModule } from 'src/engine/metadata-modules/logic-function/logic-function.module';
import { WorkflowVersionCoreModule } from 'src/engine/core-modules/workflow/workflow-version-core.module';
import { WorkflowQueryHookModule } from 'src/modules/workflow/common/query-hooks/workflow-query-hook.module';
import { WorkflowCommonWorkspaceService } from 'src/modules/workflow/common/workspace-services/workflow-common.workspace-service';
@@ -14,6 +15,7 @@ import { WorkflowCommonWorkspaceService } from 'src/modules/workflow/common/work
WorkspaceManyOrAllFlatEntityMapsCacheModule,
CommandMenuItemModule,
FeatureFlagModule,
WorkflowVersionCoreModule,
],
providers: [WorkflowCommonWorkspaceService],
exports: [WorkflowCommonWorkspaceService],
packages/twenty-server/src/modules/workflow/common/workspace-services/workflow-common.workspace-service.ts
+24 -9
View File
@@ -4,6 +4,7 @@ import { isDefined, isValidUuid } from 'twenty-shared/utils';
import { In } from 'typeorm';
import { type WorkspaceAuthContext } from 'src/engine/core-modules/auth/types/workspace-auth-context.type';
import { WorkflowVersionCoreSyncService } from 'src/engine/core-modules/workflow/services/workflow-version-core-sync.service';
import { CommandMenuItemService } from 'src/engine/metadata-modules/command-menu-item/command-menu-item.service';
import { WorkspaceManyOrAllFlatEntityMapsCacheService } from 'src/engine/metadata-modules/flat-entity/services/workspace-many-or-all-flat-entity-maps-cache.service';
import { type FlatEntityMaps } from 'src/engine/metadata-modules/flat-entity/types/flat-entity-maps.type';
@@ -56,6 +57,7 @@ export class WorkflowCommonWorkspaceService {
private readonly logicFunctionFromSourceService: LogicFunctionFromSourceService,
private readonly workspaceManyOrAllFlatEntityMapsCacheService: WorkspaceManyOrAllFlatEntityMapsCacheService,
private readonly commandMenuItemService: CommandMenuItemService,
private readonly workflowVersionCoreSyncService: WorkflowVersionCoreSyncService,
) {}
async getWorkflowVersionOrFail({
@@ -371,15 +373,6 @@ export class WorkflowCommonWorkspaceService {
withDeleted: true,
});
for (const workflowVersion of workflowVersions) {
if (workflowVersion.status === WorkflowVersionStatus.ACTIVE) {
await this.cleanupCommandMenuItemForVersion(
workflowVersion.id,
workspaceId,
);
}
}
const workspaceDataSource =
await this.globalWorkspaceOrmManager.getGlobalWorkspaceDataSource();
@@ -421,6 +414,15 @@ export class WorkflowCommonWorkspaceService {
undefined,
queryRunner.manager,
);
await this.workflowVersionCoreSyncService.mirrorWorkflowVersionWrite({
workspaceId,
entityManager: queryRunner.manager,
workflowVersion: {
...workflowVersion,
status: WorkflowVersionStatus.DEACTIVATED,
},
});
}
}
@@ -434,6 +436,19 @@ export class WorkflowCommonWorkspaceService {
} finally {
await queryRunner.release();
}
for (const workflowVersion of workflowVersions) {
if (workflowVersion.status === WorkflowVersionStatus.ACTIVE) {
await this.cleanupCommandMenuItemForVersion(
workflowVersion.id,
workspaceId,
);
}
}
await this.workflowVersionCoreSyncService.invalidateAutomatedTriggerMaps(
workspaceId,
);
}
private async cleanupCommandMenuItemForVersion(
packages/twenty-server/src/modules/workflow/workflow-builder/workflow-version-edge/__tests__/workflow-version-edge.workspace-service.spec.ts
+121 -68
View File
@@ -2,6 +2,7 @@ import { Test, type TestingModule } from '@nestjs/testing';
import { TRIGGER_STEP_ID, WorkflowActionType } from 'twenty-shared/workflow';
import { WorkflowVersionCoreSyncService } from 'src/engine/core-modules/workflow/services/workflow-version-core-sync.service';
import { GlobalWorkspaceOrmManager } from 'src/engine/twenty-orm/global-workspace-datasource/global-workspace-orm.manager';
import { type WorkspaceRepository } from 'src/engine/twenty-orm/repository/workspace.repository';
import { type WorkflowVersionWorkspaceEntity } from 'src/modules/workflow/common/standard-objects/workflow-version.workspace-entity';
@@ -113,6 +114,18 @@ describe('WorkflowVersionEdgeWorkspaceService', () => {
provide: WorkflowCommonWorkspaceService,
useValue: workflowCommonWorkspaceService,
},
{
provide: WorkflowVersionCoreSyncService,
useValue: {
writeWorkflowVersionAndMirror: jest.fn(
async (_workspaceId: string, write: any) => {
await write(mockWorkflowVersionWorkspaceRepository, {});
},
),
mirrorWorkflowVersionWrite: jest.fn(),
invalidateAutomatedTriggerMaps: jest.fn(),
},
},
],
}).compile();
@@ -175,12 +188,17 @@ describe('WorkflowVersionEdgeWorkspaceService', () => {
expect(
mockWorkflowVersionWorkspaceRepository.update,
).toHaveBeenCalledWith(mockWorkflowVersionId, {
trigger: {
...mockTrigger,
nextStepIds: ['step-1', 'step-3'],
).toHaveBeenCalledWith(
mockWorkflowVersionId,
{
trigger: {
...mockTrigger,
nextStepIds: ['step-1', 'step-3'],
},
},
});
undefined,
expect.anything(),
);
expect(result).toEqual({
triggerDiff: [
@@ -258,18 +276,23 @@ describe('WorkflowVersionEdgeWorkspaceService', () => {
expect(
mockWorkflowVersionWorkspaceRepository.update,
).toHaveBeenCalledWith(mockWorkflowVersionId, {
steps: expect.arrayContaining([
expect.objectContaining({
id: 'iterator-step',
settings: expect.objectContaining({
input: expect.objectContaining({
initialLoopStepIds: ['step-1', 'step-3'],
).toHaveBeenCalledWith(
mockWorkflowVersionId,
{
steps: expect.arrayContaining([
expect.objectContaining({
id: 'iterator-step',
settings: expect.objectContaining({
input: expect.objectContaining({
initialLoopStepIds: ['step-1', 'step-3'],
}),
}),
}),
}),
]),
});
]),
},
undefined,
expect.anything(),
);
expect(result).toEqual({
stepsDiff: [
@@ -350,14 +373,19 @@ describe('WorkflowVersionEdgeWorkspaceService', () => {
expect(
mockWorkflowVersionWorkspaceRepository.update,
).toHaveBeenCalledWith(mockWorkflowVersionId, {
steps: expect.arrayContaining([
expect.objectContaining({
id: 'iterator-step',
nextStepIds: ['step-2', 'step-3'],
}),
]),
});
).toHaveBeenCalledWith(
mockWorkflowVersionId,
{
steps: expect.arrayContaining([
expect.objectContaining({
id: 'iterator-step',
nextStepIds: ['step-2', 'step-3'],
}),
]),
},
undefined,
expect.anything(),
);
expect(result).toEqual({
stepsDiff: [
@@ -382,18 +410,23 @@ describe('WorkflowVersionEdgeWorkspaceService', () => {
expect(
mockWorkflowVersionWorkspaceRepository.update,
).toHaveBeenCalledWith(mockWorkflowVersionId, {
steps: mockSteps.map((step) => {
if (step.id === 'step-2') {
return {
...step,
nextStepIds: ['step-3'],
};
}
).toHaveBeenCalledWith(
mockWorkflowVersionId,
{
steps: mockSteps.map((step) => {
if (step.id === 'step-2') {
return {
...step,
nextStepIds: ['step-3'],
};
}
return step;
}),
});
return step;
}),
},
undefined,
expect.anything(),
);
expect(result).toEqual({
stepsDiff: [
@@ -474,12 +507,17 @@ describe('WorkflowVersionEdgeWorkspaceService', () => {
expect(
mockWorkflowVersionWorkspaceRepository.update,
).toHaveBeenCalledWith(mockWorkflowVersionId, {
trigger: {
...mockTrigger,
nextStepIds: [],
).toHaveBeenCalledWith(
mockWorkflowVersionId,
{
trigger: {
...mockTrigger,
nextStepIds: [],
},
},
});
undefined,
expect.anything(),
);
expect(result).toEqual({
triggerDiff: [
@@ -530,18 +568,23 @@ describe('WorkflowVersionEdgeWorkspaceService', () => {
expect(
mockWorkflowVersionWorkspaceRepository.update,
).toHaveBeenCalledWith(mockWorkflowVersionId, {
steps: mockSteps.map((step) => {
if (step.id === 'step-1') {
return {
...step,
nextStepIds: [],
};
}
).toHaveBeenCalledWith(
mockWorkflowVersionId,
{
steps: mockSteps.map((step) => {
if (step.id === 'step-1') {
return {
...step,
nextStepIds: [],
};
}
return step;
}),
});
return step;
}),
},
undefined,
expect.anything(),
);
expect(result).toEqual({
stepsDiff: [
@@ -654,18 +697,23 @@ describe('WorkflowVersionEdgeWorkspaceService', () => {
expect(
mockWorkflowVersionWorkspaceRepository.update,
).toHaveBeenCalledWith(mockWorkflowVersionId, {
steps: expect.arrayContaining([
expect.objectContaining({
id: 'iterator-step',
settings: expect.objectContaining({
input: expect.objectContaining({
initialLoopStepIds: ['step-3'],
).toHaveBeenCalledWith(
mockWorkflowVersionId,
{
steps: expect.arrayContaining([
expect.objectContaining({
id: 'iterator-step',
settings: expect.objectContaining({
input: expect.objectContaining({
initialLoopStepIds: ['step-3'],
}),
}),
}),
}),
]),
});
]),
},
undefined,
expect.anything(),
);
expect(result).toEqual({
stepsDiff: [
@@ -739,14 +787,19 @@ describe('WorkflowVersionEdgeWorkspaceService', () => {
expect(
mockWorkflowVersionWorkspaceRepository.update,
).toHaveBeenCalledWith(mockWorkflowVersionId, {
steps: expect.arrayContaining([
expect.objectContaining({
id: 'iterator-step',
nextStepIds: [],
}),
]),
});
).toHaveBeenCalledWith(
mockWorkflowVersionId,
{
steps: expect.arrayContaining([
expect.objectContaining({
id: 'iterator-step',
nextStepIds: [],
}),
]),
},
undefined,
expect.anything(),
);
expect(result).toEqual({
stepsDiff: [
packages/twenty-server/src/modules/workflow/workflow-builder/workflow-version-edge/workflow-version-edge.module.ts
+2 -1
View File
@@ -1,10 +1,11 @@
import { Module } from '@nestjs/common';
import { WorkflowVersionCoreModule } from 'src/engine/core-modules/workflow/workflow-version-core.module';
import { WorkflowCommonModule } from 'src/modules/workflow/common/workflow-common.module';
import { WorkflowVersionEdgeWorkspaceService } from 'src/modules/workflow/workflow-builder/workflow-version-edge/workflow-version-edge.workspace-service';
@Module({
imports: [WorkflowCommonModule],
imports: [WorkflowCommonModule, WorkflowVersionCoreModule],
providers: [WorkflowVersionEdgeWorkspaceService],
exports: [WorkflowVersionEdgeWorkspaceService],
})
packages/twenty-server/src/modules/workflow/workflow-builder/workflow-version-edge/workflow-version-edge.workspace-service.ts
+66 -39
View File
@@ -4,8 +4,8 @@ import { isDefined } from 'twenty-shared/utils';
import { TRIGGER_STEP_ID, WorkflowActionType } from 'twenty-shared/workflow';
import { type WorkflowVersionStepChangesDTO } from 'src/engine/core-modules/workflow/dtos/workflow-version-step-changes.dto';
import { WorkflowVersionCoreSyncService } from 'src/engine/core-modules/workflow/services/workflow-version-core-sync.service';
import { GlobalWorkspaceOrmManager } from 'src/engine/twenty-orm/global-workspace-datasource/global-workspace-orm.manager';
import { WorkspaceRepository } from 'src/engine/twenty-orm/repository/workspace.repository';
import { buildSystemAuthContext } from 'src/engine/twenty-orm/utils/build-system-auth-context.util';
import {
WorkflowVersionEdgeException,
@@ -24,6 +24,7 @@ export class WorkflowVersionEdgeWorkspaceService {
constructor(
private readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager,
private readonly workflowCommonWorkspaceService: WorkflowCommonWorkspaceService,
private readonly workflowVersionCoreSyncService: WorkflowVersionCoreSyncService,
) {}
async createWorkflowVersionEdge({
@@ -43,13 +44,6 @@ export class WorkflowVersionEdgeWorkspaceService {
return this.globalWorkspaceOrmManager.executeInWorkspaceContext(
async () => {
const workflowVersionRepository =
await this.globalWorkspaceOrmManager.getRepository<WorkflowVersionWorkspaceEntity>(
workspaceId,
'workflowVersion',
{ shouldBypassPermissionChecks: true },
);
const workflowVersion =
await this.workflowCommonWorkspaceService.getWorkflowVersionOrFail({
workflowVersionId,
@@ -78,7 +72,7 @@ export class WorkflowVersionEdgeWorkspaceService {
steps,
target,
workflowVersion,
workflowVersionRepository,
workspaceId,
});
} else {
return this.createStepEdge({
@@ -88,7 +82,7 @@ export class WorkflowVersionEdgeWorkspaceService {
target,
sourceConnectionOptions,
workflowVersion,
workflowVersionRepository,
workspaceId,
});
}
},
@@ -113,13 +107,6 @@ export class WorkflowVersionEdgeWorkspaceService {
return this.globalWorkspaceOrmManager.executeInWorkspaceContext(
async () => {
const workflowVersionRepository =
await this.globalWorkspaceOrmManager.getRepository<WorkflowVersionWorkspaceEntity>(
workspaceId,
'workflowVersion',
{ shouldBypassPermissionChecks: true },
);
const workflowVersion =
await this.workflowCommonWorkspaceService.getWorkflowVersionOrFail({
workflowVersionId,
@@ -148,7 +135,7 @@ export class WorkflowVersionEdgeWorkspaceService {
steps,
target,
workflowVersion,
workflowVersionRepository,
workspaceId,
});
} else {
return this.deleteStepEdge({
@@ -157,7 +144,7 @@ export class WorkflowVersionEdgeWorkspaceService {
source,
target,
workflowVersion,
workflowVersionRepository,
workspaceId,
sourceConnectionOptions,
});
}
@@ -171,13 +158,13 @@ export class WorkflowVersionEdgeWorkspaceService {
steps,
target,
workflowVersion,
workflowVersionRepository,
workspaceId,
}: {
trigger: WorkflowTrigger | null;
steps: WorkflowAction[];
target: string;
workflowVersion: WorkflowVersionWorkspaceEntity;
workflowVersionRepository: WorkspaceRepository<WorkflowVersionWorkspaceEntity>;
workspaceId: string;
}): Promise<WorkflowVersionStepChangesDTO> {
if (!isDefined(trigger)) {
throw new WorkflowVersionEdgeException(
@@ -198,9 +185,19 @@ export class WorkflowVersionEdgeWorkspaceService {
nextStepIds: [...(trigger.nextStepIds ?? []), target],
};
await workflowVersionRepository.update(workflowVersion.id, {
trigger: updatedTrigger,
});
await this.workflowVersionCoreSyncService.writeWorkflowVersionAndMirror(
workspaceId,
async (workflowVersionRepository, entityManager) => {
await workflowVersionRepository.update(
workflowVersion.id,
{ trigger: updatedTrigger },
undefined,
entityManager,
);
return workflowVersion.id;
},
);
return computeWorkflowVersionStepChanges({
existingTrigger: trigger,
@@ -215,7 +212,7 @@ export class WorkflowVersionEdgeWorkspaceService {
source,
target,
workflowVersion,
workflowVersionRepository,
workspaceId,
sourceConnectionOptions,
}: {
trigger: WorkflowTrigger | null;
@@ -223,7 +220,7 @@ export class WorkflowVersionEdgeWorkspaceService {
source: string;
target: string;
workflowVersion: WorkflowVersionWorkspaceEntity;
workflowVersionRepository: WorkspaceRepository<WorkflowVersionWorkspaceEntity>;
workspaceId: string;
sourceConnectionOptions?: WorkflowStepConnectionOptions;
}): Promise<WorkflowVersionStepChangesDTO> {
const sourceStep = steps.find((step) => step.id === source);
@@ -264,9 +261,19 @@ export class WorkflowVersionEdgeWorkspaceService {
});
if (shouldPersist) {
await workflowVersionRepository.update(workflowVersion.id, {
steps: updatedSteps,
});
await this.workflowVersionCoreSyncService.writeWorkflowVersionAndMirror(
workspaceId,
async (workflowVersionRepository, entityManager) => {
await workflowVersionRepository.update(
workflowVersion.id,
{ steps: updatedSteps },
undefined,
entityManager,
);
return workflowVersion.id;
},
);
}
return computeWorkflowVersionStepChanges({
@@ -372,13 +379,13 @@ export class WorkflowVersionEdgeWorkspaceService {
steps,
target,
workflowVersion,
workflowVersionRepository,
workspaceId,
}: {
trigger: WorkflowTrigger | null;
steps: WorkflowAction[];
target: string;
workflowVersion: WorkflowVersionWorkspaceEntity;
workflowVersionRepository: WorkspaceRepository<WorkflowVersionWorkspaceEntity>;
workspaceId: string;
}): Promise<WorkflowVersionStepChangesDTO> {
if (!isDefined(trigger)) {
throw new WorkflowVersionEdgeException(
@@ -401,9 +408,19 @@ export class WorkflowVersionEdgeWorkspaceService {
),
};
await workflowVersionRepository.update(workflowVersion.id, {
trigger: updatedTrigger,
});
await this.workflowVersionCoreSyncService.writeWorkflowVersionAndMirror(
workspaceId,
async (workflowVersionRepository, entityManager) => {
await workflowVersionRepository.update(
workflowVersion.id,
{ trigger: updatedTrigger },
undefined,
entityManager,
);
return workflowVersion.id;
},
);
return computeWorkflowVersionStepChanges({
existingTrigger: trigger,
@@ -418,7 +435,7 @@ export class WorkflowVersionEdgeWorkspaceService {
source,
target,
workflowVersion,
workflowVersionRepository,
workspaceId,
sourceConnectionOptions,
}: {
trigger: WorkflowTrigger | null;
@@ -426,7 +443,7 @@ export class WorkflowVersionEdgeWorkspaceService {
source: string;
target: string;
workflowVersion: WorkflowVersionWorkspaceEntity;
workflowVersionRepository: WorkspaceRepository<WorkflowVersionWorkspaceEntity>;
workspaceId: string;
sourceConnectionOptions?: WorkflowStepConnectionOptions;
}): Promise<WorkflowVersionStepChangesDTO> {
const sourceStep = steps.find((step) => step.id === source);
@@ -481,9 +498,19 @@ export class WorkflowVersionEdgeWorkspaceService {
return step;
});
await workflowVersionRepository.update(workflowVersion.id, {
steps: updatedSteps,
});
await this.workflowVersionCoreSyncService.writeWorkflowVersionAndMirror(
workspaceId,
async (workflowVersionRepository, entityManager) => {
await workflowVersionRepository.update(
workflowVersion.id,
{ steps: updatedSteps },
undefined,
entityManager,
);
return workflowVersion.id;
},
);
return computeWorkflowVersionStepChanges({
existingTrigger: trigger,
packages/twenty-server/src/modules/workflow/workflow-builder/workflow-version-step/__tests__/workflow-version-step-operations.workspace-service.spec.ts
+20
View File
@@ -6,6 +6,7 @@ import { SEED_WORKFLOW_ACTION_TRIGGER_SETTINGS } from 'twenty-shared/logic-funct
import { AiAgentRoleService } from 'src/engine/metadata-modules/ai/ai-agent-role/ai-agent-role.service';
import { AgentService } from 'src/engine/metadata-modules/ai/ai-agent/agent.service';
import { WorkflowVersionCoreSyncService } from 'src/engine/core-modules/workflow/services/workflow-version-core-sync.service';
import { createEmptyAllFlatEntityMaps } from 'src/engine/metadata-modules/flat-entity/constant/create-empty-all-flat-entity-maps.constant';
import { WorkspaceManyOrAllFlatEntityMapsCacheService } from 'src/engine/metadata-modules/flat-entity/services/workspace-many-or-all-flat-entity-maps-cache.service';
import { LogicFunctionRuntime } from 'src/engine/metadata-modules/logic-function/logic-function.entity';
@@ -170,6 +171,25 @@ describe('WorkflowVersionStepOperationsWorkspaceService', () => {
.mockResolvedValue(createEmptyAllFlatEntityMaps()),
},
},
{
provide: WorkflowVersionCoreSyncService,
useValue: {
writeWorkflowVersionAndMirror: jest.fn(
async (_workspaceId: string, write: any) => {
const scopedRepository =
(await globalWorkspaceOrmManager.getRepository(
_workspaceId,
'workflowVersion',
{ shouldBypassPermissionChecks: true },
)) ?? {};
return write(scopedRepository, {});
},
),
mirrorWorkflowVersionWrite: jest.fn(),
invalidateAutomatedTriggerMaps: jest.fn(),
},
},
],
}).compile();
packages/twenty-server/src/modules/workflow/workflow-builder/workflow-version-step/__tests__/workflow-version-step.workspace-service.spec.ts
+13
View File
@@ -2,6 +2,7 @@ import { Test, type TestingModule } from '@nestjs/testing';
import { TRIGGER_STEP_ID, WorkflowActionType } from 'twenty-shared/workflow';
import { WorkflowVersionCoreSyncService } from 'src/engine/core-modules/workflow/services/workflow-version-core-sync.service';
import { GlobalWorkspaceOrmManager } from 'src/engine/twenty-orm/global-workspace-datasource/global-workspace-orm.manager';
import { type WorkspaceRepository } from 'src/engine/twenty-orm/repository/workspace.repository';
import { type WorkflowVersionWorkspaceEntity } from 'src/modules/workflow/common/standard-objects/workflow-version.workspace-entity';
@@ -167,6 +168,18 @@ describe('WorkflowVersionStepWorkspaceService', () => {
.mockResolvedValue(mockWorkflowVersion),
},
},
{
provide: WorkflowVersionCoreSyncService,
useValue: {
writeWorkflowVersionAndMirror: jest.fn(
async (_workspaceId: string, write: any) => {
await write(mockWorkflowVersionWorkspaceRepository, {});
},
),
mirrorWorkflowVersionWrite: jest.fn(),
invalidateAutomatedTriggerMaps: jest.fn(),
},
},
],
}).compile();
packages/twenty-server/src/modules/workflow/workflow-builder/workflow-version-step/workflow-version-step-helpers.workspace-service.ts
+25 -25
View File
@@ -1,7 +1,6 @@
import { Injectable } from '@nestjs/common';
import { GlobalWorkspaceOrmManager } from 'src/engine/twenty-orm/global-workspace-datasource/global-workspace-orm.manager';
import { buildSystemAuthContext } from 'src/engine/twenty-orm/utils/build-system-auth-context.util';
import { WorkflowVersionCoreSyncService } from 'src/engine/core-modules/workflow/services/workflow-version-core-sync.service';
import { type WorkflowVersionWorkspaceEntity } from 'src/modules/workflow/common/standard-objects/workflow-version.workspace-entity';
import { assertWorkflowVersionIsDraft } from 'src/modules/workflow/common/utils/assert-workflow-version-is-draft.util';
import { WorkflowCommonWorkspaceService } from 'src/modules/workflow/common/workspace-services/workflow-common.workspace-service';
@@ -11,8 +10,8 @@ import { type WorkflowTrigger } from 'src/modules/workflow/workflow-trigger/type
@Injectable()
export class WorkflowVersionStepHelpersWorkspaceService {
constructor(
private readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager,
private readonly workflowCommonWorkspaceService: WorkflowCommonWorkspaceService,
private readonly workflowVersionCoreSyncService: WorkflowVersionCoreSyncService,
) {}
async getValidatedDraftWorkflowVersion({
@@ -44,30 +43,31 @@ export class WorkflowVersionStepHelpersWorkspaceService {
steps?: WorkflowAction[] | null;
trigger?: WorkflowTrigger | null;
}): Promise<void> {
const authContext = buildSystemAuthContext(workspaceId);
const updateData: Pick<
Partial<WorkflowVersionWorkspaceEntity>,
'steps' | 'trigger'
> = {};
await this.globalWorkspaceOrmManager.executeInWorkspaceContext(async () => {
const workflowVersionRepository =
await this.globalWorkspaceOrmManager.getRepository<WorkflowVersionWorkspaceEntity>(
workspaceId,
'workflowVersion',
{ shouldBypassPermissionChecks: true },
if (steps !== undefined) {
updateData.steps = steps;
}
if (trigger !== undefined) {
updateData.trigger = trigger;
}
await this.workflowVersionCoreSyncService.writeWorkflowVersionAndMirror(
workspaceId,
async (workflowVersionRepository, entityManager) => {
await workflowVersionRepository.update(
workflowVersionId,
updateData,
undefined,
entityManager,
);
const updateData: Pick<
Partial<WorkflowVersionWorkspaceEntity>,
'steps' | 'trigger'
> = {};
if (steps !== undefined) {
updateData.steps = steps;
}
if (trigger !== undefined) {
updateData.trigger = trigger;
}
await workflowVersionRepository.update(workflowVersionId, updateData);
}, authContext);
return workflowVersionId;
},
);
}
}
packages/twenty-server/src/modules/workflow/workflow-builder/workflow-version-step/workflow-version-step-operations.workspace-service.ts
+28 -6
View File
@@ -19,6 +19,7 @@ import { v4 } from 'uuid';
import { getFlatFieldsFromFlatObjectMetadata } from 'src/engine/api/graphql/workspace-schema-builder/utils/get-flat-fields-for-flat-object-metadata.util';
import { type WorkflowStepPositionInput } from 'src/engine/core-modules/workflow/dtos/update-workflow-step-position.input';
import { WorkflowVersionCoreSyncService } from 'src/engine/core-modules/workflow/services/workflow-version-core-sync.service';
import { AiAgentRoleService } from 'src/engine/metadata-modules/ai/ai-agent-role/ai-agent-role.service';
import { AgentService } from 'src/engine/metadata-modules/ai/ai-agent/agent.service';
import { WorkspaceManyOrAllFlatEntityMapsCacheService } from 'src/engine/metadata-modules/flat-entity/services/workspace-many-or-all-flat-entity-maps-cache.service';
@@ -84,6 +85,7 @@ export class WorkflowVersionStepOperationsWorkspaceService {
private readonly aiAgentRoleService: AiAgentRoleService,
private readonly workspaceCacheService: WorkspaceCacheService,
private readonly flatEntityMapsCacheService: WorkspaceManyOrAllFlatEntityMapsCacheService,
private readonly workflowVersionCoreSyncService: WorkflowVersionCoreSyncService,
) {}
async runWorkflowVersionStepDeletionSideEffects({
@@ -930,9 +932,19 @@ export class WorkflowVersionStepOperationsWorkspaceService {
},
};
await workflowVersionRepository.update(workflowVersion.id, {
steps: [...existingSteps, emptyNodeStep],
});
await this.workflowVersionCoreSyncService.writeWorkflowVersionAndMirror(
workspaceId,
async (scopedRepository, entityManager) => {
await scopedRepository.update(
workflowVersion.id,
{ steps: [...existingSteps, emptyNodeStep] },
undefined,
entityManager,
);
return workflowVersion.id;
},
);
return emptyNodeStep;
},
@@ -1012,9 +1024,19 @@ export class WorkflowVersionStepOperationsWorkspaceService {
},
};
await workflowVersionRepository.update(workflowVersion.id, {
steps: [...existingSteps, ifEmptyNode, elseEmptyNode],
});
await this.workflowVersionCoreSyncService.writeWorkflowVersionAndMirror(
workspaceId,
async (scopedRepository, entityManager) => {
await scopedRepository.update(
workflowVersion.id,
{ steps: [...existingSteps, ifEmptyNode, elseEmptyNode] },
undefined,
entityManager,
);
return workflowVersion.id;
},
);
const ifFilterGroupId = v4();
packages/twenty-server/src/modules/workflow/workflow-builder/workflow-version-step/workflow-version-step.module.ts
+2
View File
@@ -1,6 +1,7 @@
import { Module } from '@nestjs/common';
import { TypeOrmModule } from '@nestjs/typeorm';
import { WorkflowVersionCoreModule } from 'src/engine/core-modules/workflow/workflow-version-core.module';
import { AiAgentRoleModule } from 'src/engine/metadata-modules/ai/ai-agent-role/ai-agent-role.module';
import { AiAgentModule } from 'src/engine/metadata-modules/ai/ai-agent/ai-agent.module';
import { WorkspaceManyOrAllFlatEntityMapsCacheModule } from 'src/engine/metadata-modules/flat-entity/services/workspace-many-or-all-flat-entity-maps-cache.module';
@@ -30,6 +31,7 @@ import { WorkflowVersionStepWorkspaceService } from 'src/modules/workflow/workfl
WorkspaceCacheModule,
TypeOrmModule.forFeature([ObjectMetadataEntity, RoleTargetEntity]),
WorkspaceManyOrAllFlatEntityMapsCacheModule,
WorkflowVersionCoreModule,
],
providers: [
WorkflowVersionStepWorkspaceService,
packages/twenty-server/src/modules/workflow/workflow-builder/workflow-version/workflow-version.module.ts
+2
View File
@@ -2,6 +2,7 @@ import { Module } from '@nestjs/common';
import { CacheLockModule } from 'src/engine/core-modules/cache-lock/cache-lock.module';
import { RecordPositionModule } from 'src/engine/core-modules/record-position/record-position.module';
import { WorkflowVersionCoreModule } from 'src/engine/core-modules/workflow/workflow-version-core.module';
import { LogicFunctionModule } from 'src/engine/metadata-modules/logic-function/logic-function.module';
import { WorkflowCommonModule } from 'src/modules/workflow/common/workflow-common.module';
import { WorkflowSchemaModule } from 'src/modules/workflow/workflow-builder/workflow-schema/workflow-schema.module';
@@ -16,6 +17,7 @@ import { WorkflowVersionWorkspaceService } from 'src/modules/workflow/workflow-b
WorkflowCommonModule,
RecordPositionModule,
CacheLockModule,
WorkflowVersionCoreModule,
],
providers: [WorkflowVersionWorkspaceService],
exports: [WorkflowVersionWorkspaceService],
packages/twenty-server/src/modules/workflow/workflow-builder/workflow-version/workflow-version.workspace-service.ts
+90 -29
View File
@@ -12,6 +12,7 @@ import {
import { WithLock } from 'src/engine/core-modules/cache-lock/with-lock.decorator';
import { RecordPositionService } from 'src/engine/core-modules/record-position/services/record-position.service';
import { type WorkflowStepPositionUpdateInput } from 'src/engine/core-modules/workflow/dtos/update-workflow-step-position-update.input';
import { WorkflowVersionCoreSyncService } from 'src/engine/core-modules/workflow/services/workflow-version-core-sync.service';
import { GlobalWorkspaceOrmManager } from 'src/engine/twenty-orm/global-workspace-datasource/global-workspace-orm.manager';
import { buildSystemAuthContext } from 'src/engine/twenty-orm/utils/build-system-auth-context.util';
import {
@@ -42,6 +43,7 @@ export class WorkflowVersionWorkspaceService {
private readonly workflowVersionStepOperationsWorkspaceService: WorkflowVersionStepOperationsWorkspaceService,
private readonly recordPositionService: RecordPositionService,
private readonly workflowCommonWorkspaceService: WorkflowCommonWorkspaceService,
private readonly workflowVersionCoreSyncService: WorkflowVersionCoreSyncService,
) {}
@WithLock('workflowId')
@@ -105,10 +107,22 @@ export class WorkflowVersionWorkspaceService {
if (isDefined(existingDraftVersion)) {
assertWorkflowVersionIsDraft(existingDraftVersion);
await workflowVersionRepository.update(existingDraftVersion.id, {
steps: newWorkflowVersionSteps,
trigger: newWorkflowVersionTrigger,
});
await this.workflowVersionCoreSyncService.writeWorkflowVersionAndMirror(
workspaceId,
async (scopedRepository, entityManager) => {
await scopedRepository.update(
existingDraftVersion.id,
{
steps: newWorkflowVersionSteps,
trigger: newWorkflowVersionTrigger,
},
undefined,
entityManager,
);
return existingDraftVersion.id;
},
);
return {
...existingDraftVersion,
@@ -133,17 +147,36 @@ export class WorkflowVersionWorkspaceService {
workspaceId,
});
const insertResult = await workflowVersionRepository.insert({
workflowId,
name: `v${workflowVersionsCount + 1}`,
status: WorkflowVersionStatus.DRAFT,
steps: newWorkflowVersionSteps,
trigger: newWorkflowVersionTrigger,
position,
});
let draftWorkflowVersion: WorkflowVersionWorkspaceEntity | undefined;
const draftWorkflowVersion = insertResult
.generatedMaps[0] as WorkflowVersionWorkspaceEntity;
await this.workflowVersionCoreSyncService.writeWorkflowVersionAndMirror(
workspaceId,
async (scopedRepository, entityManager) => {
const insertResult = await scopedRepository.insert(
{
workflowId,
name: `v${workflowVersionsCount + 1}`,
status: WorkflowVersionStatus.DRAFT,
steps: newWorkflowVersionSteps,
trigger: newWorkflowVersionTrigger,
position,
},
entityManager,
);
draftWorkflowVersion = insertResult
.generatedMaps[0] as WorkflowVersionWorkspaceEntity;
return draftWorkflowVersion.id;
},
);
if (!isDefined(draftWorkflowVersion)) {
throw new WorkflowVersionStepException(
'Failed to create draft workflow version',
WorkflowVersionStepExceptionCode.NOT_FOUND,
);
}
return {
...draftWorkflowVersion,
@@ -243,16 +276,6 @@ export class WorkflowVersionWorkspaceService {
workspaceId,
});
const insertVersionResult = await workflowVersionRepository.insert({
workflowId: newWorkflowId,
name: 'v1',
status: WorkflowVersionStatus.DRAFT,
position: versionPosition,
});
const newDraftVersion = insertVersionResult
.generatedMaps[0] as WorkflowVersionWorkspaceEntity;
const newTrigger = sourceVersion.trigger;
const sourceToClonedPairs: Array<{
source: WorkflowAction;
@@ -312,10 +335,36 @@ export class WorkflowVersionWorkspaceService {
},
);
await workflowVersionRepository.update(newDraftVersion.id, {
steps: remappedSteps,
trigger: remappedTrigger,
});
let newDraftVersion: WorkflowVersionWorkspaceEntity | undefined;
await this.workflowVersionCoreSyncService.writeWorkflowVersionAndMirror(
workspaceId,
async (scopedRepository, entityManager) => {
const insertVersionResult = await scopedRepository.insert(
{
workflowId: newWorkflowId,
name: 'v1',
status: WorkflowVersionStatus.DRAFT,
position: versionPosition,
steps: remappedSteps,
trigger: remappedTrigger,
},
entityManager,
);
newDraftVersion = insertVersionResult
.generatedMaps[0] as WorkflowVersionWorkspaceEntity;
return newDraftVersion.id;
},
);
if (!isDefined(newDraftVersion)) {
throw new WorkflowVersionStepException(
'Failed to duplicate workflow version',
WorkflowVersionStepExceptionCode.NOT_FOUND,
);
}
return {
...newDraftVersion,
@@ -387,7 +436,19 @@ export class WorkflowVersionWorkspaceService {
...(!isDefined(updatedSteps) ? {} : { steps: updatedSteps }),
};
await workflowVersionRepository.update(workflowVersionId, updatePayload);
await this.workflowVersionCoreSyncService.writeWorkflowVersionAndMirror(
workspaceId,
async (scopedRepository, entityManager) => {
await scopedRepository.update(
workflowVersionId,
updatePayload,
undefined,
entityManager,
);
return workflowVersionId;
},
);
}, authContext);
}
packages/twenty-server/src/modules/workflow/workflow-tools/services/workflow-tool.workspace-service.ts
+3
View File
@@ -3,6 +3,7 @@ import { Injectable } from '@nestjs/common';
import { type ToolSet } from 'ai';
import { RecordPositionService } from 'src/engine/core-modules/record-position/services/record-position.service';
import { WorkflowVersionCoreSyncService } from 'src/engine/core-modules/workflow/services/workflow-version-core-sync.service';
import { AgentService } from 'src/engine/metadata-modules/ai/ai-agent/agent.service';
import { WorkspaceManyOrAllFlatEntityMapsCacheService } from 'src/engine/metadata-modules/flat-entity/services/workspace-many-or-all-flat-entity-maps-cache.service';
import { LogicFunctionFromSourceService } from 'src/engine/metadata-modules/logic-function/services/logic-function-from-source.service';
@@ -58,6 +59,7 @@ export class WorkflowToolWorkspaceService {
flatEntityMapsCacheService: WorkspaceManyOrAllFlatEntityMapsCacheService,
agentService: AgentService,
workflowCommonService: WorkflowCommonWorkspaceService,
workflowVersionCoreSyncService: WorkflowVersionCoreSyncService,
) {
this.deps = {
workflowVersionStepService,
@@ -73,6 +75,7 @@ export class WorkflowToolWorkspaceService {
flatEntityMapsCacheService,
agentService,
workflowCommonService,
workflowVersionCoreSyncService,
};
}
packages/twenty-server/src/modules/workflow/workflow-tools/tools/create-complete-workflow.tool.ts
+30 -30
View File
@@ -59,6 +59,7 @@ type CreateCompleteWorkflowToolDeps = Pick<
| 'globalWorkspaceOrmManager'
| 'recordPositionService'
| 'workflowValidationService'
| 'workflowVersionCoreSyncService'
>;
type CreateCompleteWorkflowToolContext = WorkflowToolContext & {
@@ -264,40 +265,39 @@ const createWorkflowVersion = async ({
trigger: WorkflowTrigger;
steps: WorkflowAction[];
}): Promise<string> => {
const authContext = buildSystemAuthContext(context.workspaceId);
const workflowVersionId = uuidv4();
return deps.globalWorkspaceOrmManager.executeInWorkspaceContext(async () => {
const workflowVersionRepository =
await deps.globalWorkspaceOrmManager.getRepository(
context.workspaceId,
'workflowVersion',
context.rolePermissionConfig,
await deps.workflowVersionCoreSyncService.writeWorkflowVersionAndMirror(
context.workspaceId,
async (workflowVersionRepository, entityManager) => {
const versionPosition =
await deps.recordPositionService.buildRecordPosition({
value: 'first',
objectMetadata: {
isCustom: false,
nameSingular: 'workflowVersion',
},
workspaceId: context.workspaceId,
});
await workflowVersionRepository.insert(
{
id: workflowVersionId,
workflowId,
name: 'v1',
status: WorkflowVersionStatus.DRAFT,
trigger,
steps,
position: versionPosition,
},
entityManager,
);
const versionPosition =
await deps.recordPositionService.buildRecordPosition({
value: 'first',
objectMetadata: {
isCustom: false,
nameSingular: 'workflowVersion',
},
workspaceId: context.workspaceId,
});
return workflowVersionId;
},
);
const workflowVersion = {
id: uuidv4(),
workflowId,
name: 'v1',
status: WorkflowVersionStatus.DRAFT,
trigger,
steps,
position: versionPosition,
};
await workflowVersionRepository.insert(workflowVersion);
return workflowVersion.id;
}, authContext);
return workflowVersionId;
};
const updateWorkflowStatus = async ({
packages/twenty-server/src/modules/workflow/workflow-tools/types/workflow-tool-dependencies.type.ts
+2
View File
@@ -1,4 +1,5 @@
import type { RecordPositionService } from 'src/engine/core-modules/record-position/services/record-position.service';
import type { WorkflowVersionCoreSyncService } from 'src/engine/core-modules/workflow/services/workflow-version-core-sync.service';
import type { AgentService } from 'src/engine/metadata-modules/ai/ai-agent/agent.service';
import type { LogicFunctionFromSourceService } from 'src/engine/metadata-modules/logic-function/services/logic-function-from-source.service';
import type { WorkspaceManyOrAllFlatEntityMapsCacheService } from 'src/engine/metadata-modules/flat-entity/services/workspace-many-or-all-flat-entity-maps-cache.service';
@@ -26,6 +27,7 @@ export type WorkflowToolDependencies = {
flatEntityMapsCacheService: WorkspaceManyOrAllFlatEntityMapsCacheService;
agentService: AgentService;
workflowCommonService: WorkflowCommonWorkspaceService;
workflowVersionCoreSyncService: WorkflowVersionCoreSyncService;
};
export type WorkflowToolContext = {
packages/twenty-server/src/modules/workflow/workflow-tools/workflow-tools.module.ts
+2
View File
@@ -5,6 +5,7 @@ import { WORKFLOW_TOOL_SERVICE_TOKEN } from 'src/engine/core-modules/tool-provid
import { AiAgentModule } from 'src/engine/metadata-modules/ai/ai-agent/ai-agent.module';
import { WorkspaceManyOrAllFlatEntityMapsCacheModule } from 'src/engine/metadata-modules/flat-entity/services/workspace-many-or-all-flat-entity-maps-cache.module';
import { LogicFunctionModule } from 'src/engine/metadata-modules/logic-function/logic-function.module';
import { WorkflowVersionCoreModule } from 'src/engine/core-modules/workflow/workflow-version-core.module';
import { WorkflowCommonModule } from 'src/modules/workflow/common/workflow-common.module';
import { WorkflowSchemaModule } from 'src/modules/workflow/workflow-builder/workflow-schema/workflow-schema.module';
import { WorkflowValidationModule } from 'src/modules/workflow/workflow-builder/workflow-validation/workflow-validation.module';
@@ -31,6 +32,7 @@ import { WorkflowToolWorkspaceService } from './services/workflow-tool.workspace
LogicFunctionModule,
WorkspaceManyOrAllFlatEntityMapsCacheModule,
AiAgentModule,
WorkflowVersionCoreModule,
],
providers: [
WorkflowToolWorkspaceService,
packages/twenty-server/src/modules/workflow/workflow-trigger/workflow-trigger.module.ts
+2
View File
@@ -4,6 +4,7 @@ import { CacheStorageModule } from 'src/engine/core-modules/cache-storage/cache-
import { FeatureFlagModule } from 'src/engine/core-modules/feature-flag/feature-flag.module';
import { CommandMenuItemModule } from 'src/engine/metadata-modules/command-menu-item/command-menu-item.module';
import { LogicFunctionModule } from 'src/engine/metadata-modules/logic-function/logic-function.module';
import { WorkflowVersionCoreModule } from 'src/engine/core-modules/workflow/workflow-version-core.module';
import { WorkflowCommonModule } from 'src/modules/workflow/common/workflow-common.module';
import { CodeStepBuildModule } from 'src/modules/workflow/workflow-builder/workflow-version-step/code-step/code-step-build.module';
import { WorkflowCoreConsistencyModule } from 'src/modules/workflow/workflow-core-consistency/workflow-core-consistency.module';
@@ -23,6 +24,7 @@ import { WorkflowTriggerWorkspaceService } from 'src/modules/workflow/workflow-t
CommandMenuItemModule,
FeatureFlagModule,
LogicFunctionModule,
WorkflowVersionCoreModule,
],
providers: [WorkflowTriggerWorkspaceService, WorkflowTriggerJob],
exports: [WorkflowTriggerWorkspaceService],
packages/twenty-server/src/modules/workflow/workflow-trigger/workspace-services/workflow-trigger.workspace-service.ts
+53
View File
@@ -7,6 +7,7 @@ import { WorkflowActionType } from 'twenty-shared/workflow';
import { InjectCacheStorage } from 'src/engine/core-modules/cache-storage/decorators/cache-storage.decorator';
import { CacheStorageService } from 'src/engine/core-modules/cache-storage/services/cache-storage.service';
import { CacheStorageNamespace } from 'src/engine/core-modules/cache-storage/types/cache-storage-namespace.enum';
import { WorkflowVersionCoreSyncService } from 'src/engine/core-modules/workflow/services/workflow-version-core-sync.service';
import { CommandMenuItemService } from 'src/engine/metadata-modules/command-menu-item/command-menu-item.service';
import { CommandMenuItemAvailabilityType } from 'src/engine/metadata-modules/command-menu-item/enums/command-menu-item-availability-type.enum';
import { EngineComponentKey } from 'src/engine/metadata-modules/command-menu-item/enums/engine-component-key.enum';
@@ -61,6 +62,7 @@ export class WorkflowTriggerWorkspaceService {
private readonly automatedTriggerWorkspaceService: AutomatedTriggerWorkspaceService,
private readonly workspaceEventEmitter: WorkspaceEventEmitter,
private readonly commandMenuItemService: CommandMenuItemService,
private readonly workflowVersionCoreSyncService: WorkflowVersionCoreSyncService,
@InjectCacheStorage(CacheStorageNamespace.ModuleWorkflow)
private readonly cacheStorageService: CacheStorageService,
) {}
@@ -243,6 +245,28 @@ export class WorkflowTriggerWorkspaceService {
);
}
private async mirrorVersionStatusChangeInTransaction(
workflowVersionId: string,
workspaceId: string,
workflowVersionRepository: WorkspaceRepository<WorkflowVersionWorkspaceEntity>,
entityManager: WorkspaceEntityManager,
): Promise<void> {
const workflowVersion = await workflowVersionRepository.findOne(
{ where: { id: workflowVersionId } },
entityManager,
);
if (!isDefined(workflowVersion)) {
return;
}
await this.workflowVersionCoreSyncService.mirrorWorkflowVersionWrite({
workspaceId,
entityManager,
workflowVersion,
});
}
private async performActivationSteps(
workflow: WorkflowWorkspaceEntity,
workflowVersion: WorkflowVersionWorkspaceEntity,
@@ -286,6 +310,13 @@ export class WorkflowTriggerWorkspaceService {
undefined,
queryRunner.manager,
);
await this.mirrorVersionStatusChangeInTransaction(
workflow.lastPublishedVersionId,
workspaceId,
workflowVersionRepository,
queryRunner.manager,
);
}
await workflowRepository.update(
@@ -323,6 +354,13 @@ export class WorkflowTriggerWorkspaceService {
queryRunner.manager,
);
await this.mirrorVersionStatusChangeInTransaction(
workflowVersion.id,
workspaceId,
workflowVersionRepository,
queryRunner.manager,
);
await this.enableAutomatedTrigger(workflowVersion, workspaceId, {
entityManager: queryRunner.manager,
});
@@ -338,6 +376,10 @@ export class WorkflowTriggerWorkspaceService {
await queryRunner.release();
}
await this.workflowVersionCoreSyncService.invalidateAutomatedTriggerMaps(
workspaceId,
);
await this.emitStatusUpdateEvents(
workflowVersion,
WorkflowVersionStatus.ACTIVE,
@@ -381,6 +423,13 @@ export class WorkflowTriggerWorkspaceService {
queryRunner.manager,
);
await this.mirrorVersionStatusChangeInTransaction(
workflowVersion.id,
workspaceId,
workflowVersionRepository,
queryRunner.manager,
);
await this.disableAutomatedTrigger(workflowVersion, workspaceId, {
entityManager: queryRunner.manager,
});
@@ -396,6 +445,10 @@ export class WorkflowTriggerWorkspaceService {
await queryRunner.release();
}
await this.workflowVersionCoreSyncService.invalidateAutomatedTriggerMaps(
workspaceId,
);
await this.emitStatusUpdateEvents(
workflowVersion,
WorkflowVersionStatus.DEACTIVATED,
packages/twenty-server/src/modules/workflow/workflow-version-core-sync/listeners/workflow-version-core-dual-write.listener.ts
-26
View File
@@ -1,11 +1,9 @@
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';
@@ -23,30 +21,6 @@ export class WorkflowVersionCoreDualWriteListener {
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<
packages/twenty-server/test/integration/graphql/suites/workflow/duplicate-workflow.integration-spec.ts
+149
View File
@@ -0,0 +1,149 @@
import request from 'supertest';
import { updateWorkflowVersionTrigger } from 'test/integration/graphql/suites/workflow/utils/update-workflow-version-trigger.util';
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 });
describe('duplicateWorkflow (e2e)', () => {
let sourceWorkflowId: string;
let sourceVersionId: string;
let duplicatedWorkflowId: string | undefined;
beforeAll(async () => {
const createResponse = await graphql(`
mutation {
createWorkflow(data: { name: "Duplicate Source" }) {
id
}
}
`);
expect(createResponse.body.errors).toBeUndefined();
sourceWorkflowId = createResponse.body.data.createWorkflow.id;
const getResponse = await graphql(
`
query GetWorkflow($id: UUID!) {
workflow(filter: { id: { eq: $id } }) {
id
versions {
edges {
node {
id
}
}
}
}
}
`,
{ id: sourceWorkflowId },
);
sourceVersionId = getResponse.body.data.workflow.versions.edges[0].node.id;
await updateWorkflowVersionTrigger({
workflowVersionId: sourceVersionId,
trigger: {
name: 'Manual Trigger',
type: 'MANUAL',
settings: { outputSchema: {} },
nextStepIds: [],
position: { x: 0, y: 0 },
},
});
const stepResponse = await graphql(
`
mutation CreateWorkflowVersionStep(
$input: CreateWorkflowVersionStepInput!
) {
createWorkflowVersionStep(input: $input) {
stepsDiff
}
}
`,
{
input: {
workflowVersionId: sourceVersionId,
stepType: 'FIND_RECORDS',
parentStepId: 'trigger',
position: { x: 200, y: 0 },
},
},
);
expect(stepResponse.body.errors).toBeUndefined();
});
afterAll(async () => {
for (const id of [duplicatedWorkflowId, sourceWorkflowId]) {
if (id) {
await graphql(
`
mutation DestroyWorkflow($id: ID!) {
destroyWorkflow(id: $id) {
id
}
}
`,
{ id },
);
}
}
});
it('duplicates the workflow and mirrors the new draft version to core', async () => {
const response = await graphql(
`
mutation DuplicateWorkflow($input: DuplicateWorkflowInput!) {
duplicateWorkflow(input: $input) {
id
workflowId
status
trigger
steps
}
}
`,
{
input: {
workflowIdToDuplicate: sourceWorkflowId,
workflowVersionIdToCopy: sourceVersionId,
},
},
);
expect(response.body.errors).toBeUndefined();
const duplicated = response.body.data.duplicateWorkflow;
duplicatedWorkflowId = duplicated?.workflowId;
expect(duplicated.id).not.toBe(sourceVersionId);
expect(duplicated.workflowId).not.toBe(sourceWorkflowId);
expect(duplicated.status).toBe('DRAFT');
expect(duplicated.trigger?.type).toBe('MANUAL');
expect(Array.isArray(duplicated.steps)).toBe(true);
expect(duplicated.steps.length).toBeGreaterThan(0);
const coreRows = await global.testDataSource.query(
`SELECT "id", "steps", "triggers", "status" FROM core."workflowVersion"
WHERE "workspaceId" = $1 AND "workflowId" = $2`,
[SEED_APPLE_WORKSPACE_ID, duplicated.workflowId],
);
expect(coreRows).toHaveLength(1);
expect(coreRows[0].status).toBe('DRAFT');
expect(Array.isArray(coreRows[0].steps)).toBe(true);
expect(coreRows[0].steps.length).toBe(duplicated.steps.length);
expect(Array.isArray(coreRows[0].triggers)).toBe(true);
expect(coreRows[0].triggers[0].type).toBe('MANUAL');
});
});