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

Add WorkspaceAuthContextMiddleware (#17487)

Browse Source 
## Context
Introduces a middleware that automatically sets the workspace auth
context in AsyncLocalStorage for HTTP requests, making it available
throughout the request lifecycle without explicit parameter passing.

The motivation behind this change is to reduce boilerplate and simplify
the developer experience when working with workspace data in HTTP
request handlers.

The Problem (Before)
Every HTTP request handler that needed to access workspace data had to:
- Extract auth-related info from decorators (@AuthWorkspace(),
@AuthUserWorkspaceId(), etc.) in controller/resolver and pass down to
services
- Build or pass the authContext explicitly (sometimes with type
assertion which was flaky)
Then call executeInWorkspaceContext(authContext, async () => { ... })

## Changes
- Add WorkspaceAuthContextMiddleware that extracts auth context from the
request and stores it in AsyncLocalStorage
- Register middleware for GraphQL, metadata, and REST routes (runs after
hydration middlewares)
- Simplify executeInWorkspaceContext signature: fn is now the first
parameter, authContext is optional second
- If authContext is not provided, it's automatically retrieved from the
storage (set by middleware)
- Update all callers (~120 files) to use the new parameter order


- Fixes a bug in search where system auth context was used, bypassing
RLS feature.
This commit is contained in:
Weiko
2026-01-27 18:24:51 +01:00
committed by GitHub
parent dd98146c99
commit 2daebc6d0f
151 changed files with 3743 additions and 4002 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/twenty-server/src/modules/workflow/common/query-hooks/workflow-create-many.post-query.hook.ts
+28 -31
View File
@@ -36,37 +36,34 @@ export class WorkflowCreateManyPostQueryHook
assertIsDefinedOrThrow(workspace, WorkspaceNotFoundDefaultError);
await this.globalWorkspaceOrmManager.executeInWorkspaceContext(
authContext as WorkspaceAuthContext,
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);
}),
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);
}),
);
}, authContext as WorkspaceAuthContext);
}
}
packages/twenty-server/src/modules/workflow/common/query-hooks/workflow-create-one.post-query.hook.ts
+21 -24
View File
@@ -38,31 +38,28 @@ export class WorkflowCreateOnePostQueryHook
const workflow = payload[0];
await this.globalWorkspaceOrmManager.executeInWorkspaceContext(
authContext as WorkspaceAuthContext,
async () => {
const workflowVersionRepository =
await this.globalWorkspaceOrmManager.getRepository<WorkflowVersionWorkspaceEntity>(
workspace.id,
'workflowVersion',
);
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 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,
});
},
);
await workflowVersionRepository.insert({
workflowId: workflow.id,
status: WorkflowVersionStatus.DRAFT,
name: 'v1',
position,
});
}, authContext as WorkspaceAuthContext);
}
}
packages/twenty-server/src/modules/workflow/common/workspace-services/workflow-common.workspace-service.ts
+60 -63
View File
@@ -62,7 +62,6 @@ export class WorkflowCommonWorkspaceService {
const authContext = buildSystemAuthContext(workspaceId);
return this.globalWorkspaceOrmManager.executeInWorkspaceContext(
authContext,
async () => {
const workflowVersionRepository =
await this.globalWorkspaceOrmManager.getRepository<WorkflowVersionWorkspaceEntity>(
@@ -79,6 +78,7 @@ export class WorkflowCommonWorkspaceService {
return this.getValidWorkflowVersionOrFail(workflowVersion);
},
authContext,
);
}
@@ -165,78 +165,75 @@ export class WorkflowCommonWorkspaceService {
}): Promise<void> {
const authContext = buildSystemAuthContext(workspaceId);
await this.globalWorkspaceOrmManager.executeInWorkspaceContext(
authContext,
async () => {
const workflowVersionRepository =
await this.globalWorkspaceOrmManager.getRepository<WorkflowVersionWorkspaceEntity>(
workspaceId,
'workflowVersion',
{ shouldBypassPermissionChecks: true },
);
await this.globalWorkspaceOrmManager.executeInWorkspaceContext(async () => {
const workflowVersionRepository =
await this.globalWorkspaceOrmManager.getRepository<WorkflowVersionWorkspaceEntity>(
workspaceId,
'workflowVersion',
{ shouldBypassPermissionChecks: true },
);
const workflowRunRepository =
await this.globalWorkspaceOrmManager.getRepository<WorkflowRunWorkspaceEntity>(
workspaceId,
'workflowRun',
{ shouldBypassPermissionChecks: true },
);
const workflowRunRepository =
await this.globalWorkspaceOrmManager.getRepository<WorkflowRunWorkspaceEntity>(
workspaceId,
'workflowRun',
{ shouldBypassPermissionChecks: true },
);
const workflowAutomatedTriggerRepository =
await this.globalWorkspaceOrmManager.getRepository<WorkflowAutomatedTriggerWorkspaceEntity>(
workspaceId,
'workflowAutomatedTrigger',
{ shouldBypassPermissionChecks: true },
);
const workflowAutomatedTriggerRepository =
await this.globalWorkspaceOrmManager.getRepository<WorkflowAutomatedTriggerWorkspaceEntity>(
workspaceId,
'workflowAutomatedTrigger',
{ shouldBypassPermissionChecks: true },
);
for (const workflowId of workflowIds) {
switch (operation) {
case 'delete':
await workflowAutomatedTriggerRepository.softDelete({
workflowId,
});
for (const workflowId of workflowIds) {
switch (operation) {
case 'delete':
await workflowAutomatedTriggerRepository.softDelete({
workflowId,
});
await workflowRunRepository.softDelete({
workflowId,
});
await workflowRunRepository.softDelete({
workflowId,
});
await workflowVersionRepository.softDelete({
workflowId,
});
await workflowVersionRepository.softDelete({
workflowId,
});
break;
case 'restore':
await workflowAutomatedTriggerRepository.restore({
workflowId,
});
break;
case 'restore':
await workflowAutomatedTriggerRepository.restore({
workflowId,
});
await workflowRunRepository.restore({
workflowId,
});
await workflowRunRepository.restore({
workflowId,
});
await workflowVersionRepository.restore({
workflowId,
});
await workflowVersionRepository.restore({
workflowId,
});
break;
}
await this.deactivateVersionOnDelete({
workflowVersionRepository,
workflowId,
workspaceId,
operation,
});
await this.handleServerlessFunctionSubEntities({
workflowVersionRepository,
workflowId,
workspaceId,
operation,
});
break;
}
},
);
await this.deactivateVersionOnDelete({
workflowVersionRepository,
workflowId,
workspaceId,
operation,
});
await this.handleServerlessFunctionSubEntities({
workflowVersionRepository,
workflowId,
workspaceId,
operation,
});
}
}, authContext);
}
private async deactivateVersionOnDelete({
packages/twenty-server/src/modules/workflow/common/workspace-services/workflow-version-validation.workspace-service.ts
+51 -56
View File
@@ -48,36 +48,33 @@ export class WorkflowVersionValidationWorkspaceService {
const authContext = buildSystemAuthContext(workspaceId);
await this.globalWorkspaceOrmManager.executeInWorkspaceContext(
authContext,
async () => {
const workflowVersionRepository =
await this.globalWorkspaceOrmManager.getRepository<WorkflowVersionWorkspaceEntity>(
workspaceId,
'workflowVersion',
{ shouldBypassPermissionChecks: true },
);
await this.globalWorkspaceOrmManager.executeInWorkspaceContext(async () => {
const workflowVersionRepository =
await this.globalWorkspaceOrmManager.getRepository<WorkflowVersionWorkspaceEntity>(
workspaceId,
'workflowVersion',
{ shouldBypassPermissionChecks: true },
);
const workflowAlreadyHasDraftVersion =
await workflowVersionRepository.exists({
where: {
workflowId: payload.data.workflowId,
status: WorkflowVersionStatus.DRAFT,
deletedAt: IsNull(),
},
});
const workflowAlreadyHasDraftVersion =
await workflowVersionRepository.exists({
where: {
workflowId: payload.data.workflowId,
status: WorkflowVersionStatus.DRAFT,
deletedAt: IsNull(),
},
});
if (workflowAlreadyHasDraftVersion) {
throw new WorkflowQueryValidationException(
'Cannot create multiple draft versions for the same workflow',
WorkflowQueryValidationExceptionCode.FORBIDDEN,
{
userFriendlyMessage: msg`Cannot create multiple draft versions for the same workflow`,
},
);
}
},
);
if (workflowAlreadyHasDraftVersion) {
throw new WorkflowQueryValidationException(
'Cannot create multiple draft versions for the same workflow',
WorkflowQueryValidationExceptionCode.FORBIDDEN,
{
userFriendlyMessage: msg`Cannot create multiple draft versions for the same workflow`,
},
);
}
}, authContext);
}
async validateWorkflowVersionForUpdateOne({
@@ -130,35 +127,33 @@ export class WorkflowVersionValidationWorkspaceService {
const authContext = buildSystemAuthContext(workspaceId);
await this.globalWorkspaceOrmManager.executeInWorkspaceContext(
authContext,
async () => {
const workflowVersionRepository =
await this.globalWorkspaceOrmManager.getRepository<WorkflowVersionWorkspaceEntity>(
workspaceId,
'workflowVersion',
{ shouldBypassPermissionChecks: true },
);
await this.globalWorkspaceOrmManager.executeInWorkspaceContext(async () => {
const workflowVersionRepository =
await this.globalWorkspaceOrmManager.getRepository<WorkflowVersionWorkspaceEntity>(
workspaceId,
'workflowVersion',
{ shouldBypassPermissionChecks: true },
);
const otherWorkflowVersionsExist =
await workflowVersionRepository.exists({
where: {
workflowId: workflowVersion.workflowId,
deletedAt: IsNull(),
id: Not(workflowVersion.id),
},
});
const otherWorkflowVersionsExist = await workflowVersionRepository.exists(
{
where: {
workflowId: workflowVersion.workflowId,
deletedAt: IsNull(),
id: Not(workflowVersion.id),
},
},
);
if (!otherWorkflowVersionsExist) {
throw new WorkflowQueryValidationException(
'The initial version of a workflow can not be deleted',
WorkflowQueryValidationExceptionCode.FORBIDDEN,
{
userFriendlyMessage: msg`The initial version of a workflow can not be deleted`,
},
);
}
},
);
if (!otherWorkflowVersionsExist) {
throw new WorkflowQueryValidationException(
'The initial version of a workflow can not be deleted',
WorkflowQueryValidationExceptionCode.FORBIDDEN,
{
userFriendlyMessage: msg`The initial version of a workflow can not be deleted`,
},
);
}
}, authContext);
}
}
packages/twenty-server/src/modules/workflow/workflow-builder/workflow-version-edge/__tests__/workflow-version-edge.workspace-service.spec.ts
+3 -1
View File
@@ -91,7 +91,9 @@ describe('WorkflowVersionEdgeWorkspaceService', () => {
globalWorkspaceOrmManager = {
executeInWorkspaceContext: jest
.fn()
.mockImplementation(async (_authContext, callback) => callback()),
.mockImplementation(async (callback: () => any, _authContext?: any) =>
callback(),
),
getRepository: jest
.fn()
.mockResolvedValue(mockWorkflowVersionWorkspaceRepository),
packages/twenty-server/src/modules/workflow/workflow-builder/workflow-version-edge/workflow-version-edge.workspace-service.ts
+2 -2
View File
@@ -45,7 +45,6 @@ export class WorkflowVersionEdgeWorkspaceService {
const authContext = buildSystemAuthContext(workspaceId);
return this.globalWorkspaceOrmManager.executeInWorkspaceContext(
authContext,
async () => {
const workflowVersionRepository =
await this.globalWorkspaceOrmManager.getRepository<WorkflowVersionWorkspaceEntity>(
@@ -96,6 +95,7 @@ export class WorkflowVersionEdgeWorkspaceService {
});
}
},
authContext,
);
}
@@ -115,7 +115,6 @@ export class WorkflowVersionEdgeWorkspaceService {
const authContext = buildSystemAuthContext(workspaceId);
return this.globalWorkspaceOrmManager.executeInWorkspaceContext(
authContext,
async () => {
const workflowVersionRepository =
await this.globalWorkspaceOrmManager.getRepository<WorkflowVersionWorkspaceEntity>(
@@ -166,6 +165,7 @@ export class WorkflowVersionEdgeWorkspaceService {
});
}
},
authContext,
);
}
packages/twenty-server/src/modules/workflow/workflow-builder/workflow-version-step/__tests__/workflow-version-step.workspace-service.spec.ts
+1 -1
View File
@@ -115,7 +115,7 @@ describe('WorkflowVersionStepWorkspaceService', () => {
executeInWorkspaceContext: jest
.fn()
.mockImplementation((_authContext: any, fn: () => any) => fn()),
.mockImplementation((fn: () => any, _authContext?: any) => fn()),
} as unknown as jest.Mocked<GlobalWorkspaceOrmManager>;
const module: TestingModule = await Test.createTestingModule({
packages/twenty-server/src/modules/workflow/workflow-builder/workflow-version-step/workflow-version-step-helpers.workspace-service.ts
+16 -19
View File
@@ -46,28 +46,25 @@ export class WorkflowVersionStepHelpersWorkspaceService {
}): Promise<void> {
const authContext = buildSystemAuthContext(workspaceId);
await this.globalWorkspaceOrmManager.executeInWorkspaceContext(
authContext,
async () => {
const workflowVersionRepository =
await this.globalWorkspaceOrmManager.getRepository<WorkflowVersionWorkspaceEntity>(
workspaceId,
'workflowVersion',
{ shouldBypassPermissionChecks: true },
);
await this.globalWorkspaceOrmManager.executeInWorkspaceContext(async () => {
const workflowVersionRepository =
await this.globalWorkspaceOrmManager.getRepository<WorkflowVersionWorkspaceEntity>(
workspaceId,
'workflowVersion',
{ shouldBypassPermissionChecks: true },
);
const updateData: Partial<WorkflowVersionWorkspaceEntity> = {};
const updateData: Partial<WorkflowVersionWorkspaceEntity> = {};
if (steps !== undefined) {
updateData.steps = steps;
}
if (steps !== undefined) {
updateData.steps = steps;
}
if (trigger !== undefined) {
updateData.trigger = trigger;
}
if (trigger !== undefined) {
updateData.trigger = trigger;
}
await workflowVersionRepository.update(workflowVersionId, updateData);
},
);
await workflowVersionRepository.update(workflowVersionId, updateData);
}, authContext);
}
}
packages/twenty-server/src/modules/workflow/workflow-builder/workflow-version-step/workflow-version-step-operations.workspace-service.ts
+3 -3
View File
@@ -533,7 +533,6 @@ export class WorkflowVersionStepOperationsWorkspaceService {
const authContext = buildSystemAuthContext(workspaceId);
return this.globalWorkspaceOrmManager.executeInWorkspaceContext(
authContext,
async () => {
const responseKeys = Object.keys(response);
@@ -598,6 +597,7 @@ export class WorkflowVersionStepOperationsWorkspaceService {
return acc;
}, {});
},
authContext,
);
}
@@ -727,7 +727,6 @@ export class WorkflowVersionStepOperationsWorkspaceService {
const authContext = buildSystemAuthContext(workspaceId);
return this.globalWorkspaceOrmManager.executeInWorkspaceContext(
authContext,
async () => {
const workflowVersionRepository =
await this.globalWorkspaceOrmManager.getRepository<WorkflowVersionWorkspaceEntity>(
@@ -777,6 +776,7 @@ export class WorkflowVersionStepOperationsWorkspaceService {
return emptyNodeStep;
},
authContext,
);
}
@@ -797,7 +797,6 @@ export class WorkflowVersionStepOperationsWorkspaceService {
const authContext = buildSystemAuthContext(workspaceId);
return this.globalWorkspaceOrmManager.executeInWorkspaceContext(
authContext,
async () => {
const workflowVersionRepository =
await this.globalWorkspaceOrmManager.getRepository<WorkflowVersionWorkspaceEntity>(
@@ -878,6 +877,7 @@ export class WorkflowVersionStepOperationsWorkspaceService {
branches,
};
},
authContext,
);
}
packages/twenty-server/src/modules/workflow/workflow-builder/workflow-version/workflow-version.workspace-service.ts
+45 -51
View File
@@ -50,7 +50,6 @@ export class WorkflowVersionWorkspaceService {
const authContext = buildSystemAuthContext(workspaceId);
return this.globalWorkspaceOrmManager.executeInWorkspaceContext(
authContext,
async () => {
const workflowVersionRepository =
await this.globalWorkspaceOrmManager.getRepository<WorkflowVersionWorkspaceEntity>(
@@ -139,6 +138,7 @@ export class WorkflowVersionWorkspaceService {
trigger: newWorkflowVersionTrigger,
};
},
authContext,
);
}
@@ -154,7 +154,6 @@ export class WorkflowVersionWorkspaceService {
const authContext = buildSystemAuthContext(workspaceId);
return this.globalWorkspaceOrmManager.executeInWorkspaceContext(
authContext,
async () => {
const workflowRepository =
await this.globalWorkspaceOrmManager.getRepository(
@@ -311,6 +310,7 @@ export class WorkflowVersionWorkspaceService {
trigger: remappedTrigger ?? null,
};
},
authContext,
);
}
@@ -325,61 +325,55 @@ export class WorkflowVersionWorkspaceService {
}) {
const authContext = buildSystemAuthContext(workspaceId);
await this.globalWorkspaceOrmManager.executeInWorkspaceContext(
authContext,
async () => {
const workflowVersionRepository =
await this.globalWorkspaceOrmManager.getRepository<WorkflowVersionWorkspaceEntity>(
workspaceId,
'workflowVersion',
{ shouldBypassPermissionChecks: true },
);
const workflowVersion = await workflowVersionRepository.findOneOrFail({
where: {
id: workflowVersionId,
},
});
assertWorkflowVersionIsDraft(workflowVersion);
const triggerPosition = positions.find(
(position) => position.id === TRIGGER_STEP_ID,
await this.globalWorkspaceOrmManager.executeInWorkspaceContext(async () => {
const workflowVersionRepository =
await this.globalWorkspaceOrmManager.getRepository<WorkflowVersionWorkspaceEntity>(
workspaceId,
'workflowVersion',
{ shouldBypassPermissionChecks: true },
);
const updatedTrigger =
isDefined(triggerPosition) && isDefined(workflowVersion.trigger)
? {
...workflowVersion.trigger,
position: triggerPosition.position,
}
: undefined;
const workflowVersion = await workflowVersionRepository.findOneOrFail({
where: {
id: workflowVersionId,
},
});
const updatedSteps = workflowVersion.steps?.map((step) => {
const updatedStep = positions.find(
(position) => position.id === step.id,
);
assertWorkflowVersionIsDraft(workflowVersion);
if (updatedStep) {
return {
...step,
position: updatedStep.position,
};
}
const triggerPosition = positions.find(
(position) => position.id === TRIGGER_STEP_ID,
);
return step;
});
const updatedTrigger =
isDefined(triggerPosition) && isDefined(workflowVersion.trigger)
? {
...workflowVersion.trigger,
position: triggerPosition.position,
}
: undefined;
const updatePayload = {
...(!isDefined(updatedTrigger) ? {} : { trigger: updatedTrigger }),
...(!isDefined(updatedSteps) ? {} : { steps: updatedSteps }),
};
await workflowVersionRepository.update(
workflowVersionId,
updatePayload,
const updatedSteps = workflowVersion.steps?.map((step) => {
const updatedStep = positions.find(
(position) => position.id === step.id,
);
},
);
if (updatedStep) {
return {
...step,
position: updatedStep.position,
};
}
return step;
});
const updatePayload = {
...(!isDefined(updatedTrigger) ? {} : { trigger: updatedTrigger }),
...(!isDefined(updatedSteps) ? {} : { steps: updatedSteps }),
};
await workflowVersionRepository.update(workflowVersionId, updatePayload);
}, authContext);
}
}
packages/twenty-server/src/modules/workflow/workflow-executor/workflow-actions/delay/jobs/resume-delayed-workflow.job.ts
+60 -63
View File
@@ -41,72 +41,69 @@ export class ResumeDelayedWorkflowJob {
}: ResumeDelayedWorkflowJobData): Promise<void> {
const authContext = buildSystemAuthContext(workspaceId);
await this.globalWorkspaceOrmManager.executeInWorkspaceContext(
authContext,
async () => {
try {
const workflowRun =
await this.workflowRunWorkspaceService.getWorkflowRunOrFail({
workflowRunId,
workspaceId,
});
if (workflowRun.status !== WorkflowRunStatus.RUNNING) {
return;
}
const step = workflowRun.state?.flow?.steps?.find(
(step) => step.id === stepId,
);
const stepInfo = workflowRun.state?.stepInfos[stepId];
if (!step || !isWorkflowDelayAction(step)) {
throw new WorkflowRunException(
'Step not found or is not a delay action',
WorkflowRunExceptionCode.INVALID_INPUT,
);
}
if (stepInfo?.status !== StepStatus.PENDING) {
throw new WorkflowRunException(
'Step is not pending',
WorkflowRunExceptionCode.INVALID_INPUT,
);
}
await this.workflowRunWorkspaceService.updateWorkflowRunStepInfo({
stepId,
stepInfo: {
status: StepStatus.SUCCESS,
result: {
success: true,
},
},
workspaceId,
workflowRunId,
});
await this.messageQueueService.add<RunWorkflowJobData>(
RunWorkflowJob.name,
{
workspaceId,
workflowRunId,
lastExecutedStepId: stepId,
},
);
} catch (error) {
await this.workflowRunWorkspaceService.endWorkflowRun({
await this.globalWorkspaceOrmManager.executeInWorkspaceContext(async () => {
try {
const workflowRun =
await this.workflowRunWorkspaceService.getWorkflowRunOrFail({
workflowRunId,
workspaceId,
status: WorkflowRunStatus.FAILED,
error:
error instanceof Error
? error.message
: 'Unknown error during delay resume',
});
if (workflowRun.status !== WorkflowRunStatus.RUNNING) {
return;
}
},
);
const step = workflowRun.state?.flow?.steps?.find(
(step) => step.id === stepId,
);
const stepInfo = workflowRun.state?.stepInfos[stepId];
if (!step || !isWorkflowDelayAction(step)) {
throw new WorkflowRunException(
'Step not found or is not a delay action',
WorkflowRunExceptionCode.INVALID_INPUT,
);
}
if (stepInfo?.status !== StepStatus.PENDING) {
throw new WorkflowRunException(
'Step is not pending',
WorkflowRunExceptionCode.INVALID_INPUT,
);
}
await this.workflowRunWorkspaceService.updateWorkflowRunStepInfo({
stepId,
stepInfo: {
status: StepStatus.SUCCESS,
result: {
success: true,
},
},
workspaceId,
workflowRunId,
});
await this.messageQueueService.add<RunWorkflowJobData>(
RunWorkflowJob.name,
{
workspaceId,
workflowRunId,
lastExecutedStepId: stepId,
},
);
} catch (error) {
await this.workflowRunWorkspaceService.endWorkflowRun({
workflowRunId,
workspaceId,
status: WorkflowRunStatus.FAILED,
error:
error instanceof Error
? error.message
: 'Unknown error during delay resume',
});
}
}, authContext);
}
}
packages/twenty-server/src/modules/workflow/workflow-runner/jobs/run-workflow.job.ts
+19 -22
View File
@@ -39,32 +39,29 @@ export class RunWorkflowJob {
}: RunWorkflowJobData): Promise<void> {
const authContext = buildSystemAuthContext(workspaceId);
await this.globalWorkspaceOrmManager.executeInWorkspaceContext(
authContext,
async () => {
try {
if (lastExecutedStepId) {
await this.resumeWorkflowExecution({
workspaceId,
workflowRunId,
lastExecutedStepId,
});
} else {
await this.startWorkflowExecution({
workflowRunId,
workspaceId,
});
}
} catch (error) {
await this.workflowRunWorkspaceService.endWorkflowRun({
await this.globalWorkspaceOrmManager.executeInWorkspaceContext(async () => {
try {
if (lastExecutedStepId) {
await this.resumeWorkflowExecution({
workspaceId,
workflowRunId,
status: WorkflowRunStatus.FAILED,
error: error.message,
lastExecutedStepId,
});
} else {
await this.startWorkflowExecution({
workflowRunId,
workspaceId,
});
}
},
);
} catch (error) {
await this.workflowRunWorkspaceService.endWorkflowRun({
workspaceId,
workflowRunId,
status: WorkflowRunStatus.FAILED,
error: error.message,
});
}
}, authContext);
}
private async startWorkflowExecution({
packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-clean-workflow-runs.cron.job.ts
+17 -20
View File
@@ -76,11 +76,9 @@ export class WorkflowCleanWorkflowRunsJob {
const schemaName = getWorkspaceSchemaName(workspaceId);
const authContext = buildSystemAuthContext(workspaceId);
await this.globalWorkspaceOrmManager.executeInWorkspaceContext(
authContext,
async () => {
const workflowRunsToDelete = await this.coreDataSource.query(
`
await this.globalWorkspaceOrmManager.executeInWorkspaceContext(async () => {
const workflowRunsToDelete = await this.coreDataSource.query(
`
WITH ranked_runs AS (
SELECT id,
ROW_NUMBER() OVER (
@@ -95,23 +93,22 @@ export class WorkflowCleanWorkflowRunsJob {
WHERE rn > ${NUMBER_OF_WORKFLOW_RUNS_TO_KEEP}
OR "createdAt" < NOW() - INTERVAL '14 days';
`,
);
const workflowRunRepository =
await this.globalWorkspaceOrmManager.getRepository(
workspaceId,
WorkflowRunWorkspaceEntity,
{ shouldBypassPermissionChecks: true },
);
const workflowRunRepository =
await this.globalWorkspaceOrmManager.getRepository(
workspaceId,
WorkflowRunWorkspaceEntity,
{ shouldBypassPermissionChecks: true },
);
for (const workflowRunToDelete of workflowRunsToDelete) {
await workflowRunRepository.delete(workflowRunToDelete.id);
}
for (const workflowRunToDelete of workflowRunsToDelete) {
await workflowRunRepository.delete(workflowRunToDelete.id);
}
this.logger.log(
`Deleted ${workflowRunsToDelete.length} workflow runs for workspace ${workspaceId}`,
);
},
);
this.logger.log(
`Deleted ${workflowRunsToDelete.length} workflow runs for workspace ${workspaceId}`,
);
}, authContext);
}
}
packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-handle-staled-runs.workspace-service.ts
+31 -34
View File
@@ -51,41 +51,38 @@ export class WorkflowHandleStaledRunsWorkspaceService {
private async handleStaledRunsForWorkspace(workspaceId: string) {
const authContext = buildSystemAuthContext(workspaceId);
await this.globalWorkspaceOrmManager.executeInWorkspaceContext(
authContext,
async () => {
const workflowRunRepository =
await this.globalWorkspaceOrmManager.getRepository(
workspaceId,
WorkflowRunWorkspaceEntity,
{ shouldBypassPermissionChecks: true },
);
const oneHourAgo = new Date(Date.now() - 60 * 60 * 1000);
const staledWorkflowRuns = await workflowRunRepository.find({
where: {
status: WorkflowRunStatus.ENQUEUED,
enqueuedAt: Or(LessThan(oneHourAgo), IsNull()),
},
});
if (staledWorkflowRuns.length <= 0) {
return;
}
await workflowRunRepository.update(
staledWorkflowRuns.map((workflowRun) => workflowRun.id),
{
enqueuedAt: null,
status: WorkflowRunStatus.NOT_STARTED,
},
);
await this.workflowThrottlingWorkspaceService.recomputeWorkflowRunNotStartedCount(
await this.globalWorkspaceOrmManager.executeInWorkspaceContext(async () => {
const workflowRunRepository =
await this.globalWorkspaceOrmManager.getRepository(
workspaceId,
WorkflowRunWorkspaceEntity,
{ shouldBypassPermissionChecks: true },
);
},
);
const oneHourAgo = new Date(Date.now() - 60 * 60 * 1000);
const staledWorkflowRuns = await workflowRunRepository.find({
where: {
status: WorkflowRunStatus.ENQUEUED,
enqueuedAt: Or(LessThan(oneHourAgo), IsNull()),
},
});
if (staledWorkflowRuns.length <= 0) {
return;
}
await workflowRunRepository.update(
staledWorkflowRuns.map((workflowRun) => workflowRun.id),
{
enqueuedAt: null,
status: WorkflowRunStatus.NOT_STARTED,
},
);
await this.workflowThrottlingWorkspaceService.recomputeWorkflowRunNotStartedCount(
workspaceId,
);
}, authContext);
}
}
packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-enqueue.workspace-service.ts
+1 -1
View File
@@ -48,7 +48,6 @@ export class WorkflowRunEnqueueWorkspaceService {
const authContext = buildSystemAuthContext(workspaceId);
await this.globalWorkspaceOrmManager.executeInWorkspaceContext(
authContext,
async () => {
const workflowRunRepository =
await this.globalWorkspaceOrmManager.getRepository(
@@ -152,6 +151,7 @@ export class WorkflowRunEnqueueWorkspaceService {
);
}
},
authContext,
);
} catch (error) {
this.metricsService.incrementCounter({
packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-throttling.workspace-service.ts
+2 -2
View File
@@ -80,7 +80,6 @@ export class WorkflowThrottlingWorkspaceService {
const currentlyNotStartedWorkflowRunCount =
await this.globalWorkspaceOrmManager.executeInWorkspaceContext(
authContext,
async () => {
const workflowRunRepository =
await this.globalWorkspaceOrmManager.getRepository(
@@ -95,6 +94,7 @@ export class WorkflowThrottlingWorkspaceService {
},
});
},
authContext,
);
await this.setWorkflowRunNotStartedCount(
@@ -113,7 +113,6 @@ export class WorkflowThrottlingWorkspaceService {
const authContext = buildSystemAuthContext(workspaceId);
return this.globalWorkspaceOrmManager.executeInWorkspaceContext(
authContext,
async () => {
const workflowRunRepository =
await this.globalWorkspaceOrmManager.getRepository(
@@ -128,6 +127,7 @@ export class WorkflowThrottlingWorkspaceService {
},
});
},
authContext,
);
}
packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run/command/delete-workflow-runs.command.ts
+28 -31
View File
@@ -53,38 +53,35 @@ export class DeleteWorkflowRunsCommand extends ActiveOrSuspendedWorkspacesMigrat
}: RunOnWorkspaceArgs): Promise<void> {
const authContext = buildSystemAuthContext(workspaceId);
await this.globalWorkspaceOrmManager.executeInWorkspaceContext(
authContext,
async () => {
try {
const workflowRunRepository =
await this.globalWorkspaceOrmManager.getRepository<WorkflowRunWorkspaceEntity>(
workspaceId,
'workflowRun',
{ shouldBypassPermissionChecks: true },
);
const createdAtCondition = {
createdAt: LessThan(
this.createdBeforeDate || new Date().toISOString(),
),
};
const workflowRunCount = await workflowRunRepository.count({
where: createdAtCondition,
});
if (!options.dryRun && workflowRunCount > 0) {
await workflowRunRepository.delete(createdAtCondition);
}
this.logger.log(
`${options.dryRun ? ' (DRY RUN): ' : ''}Deleted ${workflowRunCount} workflow runs`,
await this.globalWorkspaceOrmManager.executeInWorkspaceContext(async () => {
try {
const workflowRunRepository =
await this.globalWorkspaceOrmManager.getRepository<WorkflowRunWorkspaceEntity>(
workspaceId,
'workflowRun',
{ shouldBypassPermissionChecks: true },
);
} catch (error) {
this.logger.error('Error while deleting workflowRun', error);
const createdAtCondition = {
createdAt: LessThan(
this.createdBeforeDate || new Date().toISOString(),
),
};
const workflowRunCount = await workflowRunRepository.count({
where: createdAtCondition,
});
if (!options.dryRun && workflowRunCount > 0) {
await workflowRunRepository.delete(createdAtCondition);
}
},
);
this.logger.log(
`${options.dryRun ? ' (DRY RUN): ' : ''}Deleted ${workflowRunCount} workflow runs`,
);
} catch (error) {
this.logger.error('Error while deleting workflowRun', error);
}
}, authContext);
}
}
packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run/workflow-run.workspace-service.ts
+27 -30
View File
@@ -57,7 +57,6 @@ export class WorkflowRunWorkspaceService {
const authContext = buildSystemAuthContext(workspaceId);
return this.globalWorkspaceOrmManager.executeInWorkspaceContext(
authContext,
async () => {
const workflowRunRepository =
await this.globalWorkspaceOrmManager.getRepository<WorkflowRunWorkspaceEntity>(
@@ -136,6 +135,7 @@ export class WorkflowRunWorkspaceService {
return workflowRun.id;
},
authContext,
);
}
@@ -346,7 +346,6 @@ export class WorkflowRunWorkspaceService {
const authContext = buildSystemAuthContext(workspaceId);
return this.globalWorkspaceOrmManager.executeInWorkspaceContext(
authContext,
async () => {
const workflowRunRepository =
await this.globalWorkspaceOrmManager.getRepository<WorkflowRunWorkspaceEntity>(
@@ -359,6 +358,7 @@ export class WorkflowRunWorkspaceService {
where: { id: workflowRunId },
});
},
authContext,
);
}
@@ -395,35 +395,32 @@ export class WorkflowRunWorkspaceService {
}) {
const authContext = buildSystemAuthContext(workspaceId);
await this.globalWorkspaceOrmManager.executeInWorkspaceContext(
authContext,
async () => {
const workflowRunRepository =
await this.globalWorkspaceOrmManager.getRepository<WorkflowRunWorkspaceEntity>(
workspaceId,
'workflowRun',
{ shouldBypassPermissionChecks: true },
);
const workflowRunToUpdate = await workflowRunRepository.findOneBy({
id: workflowRunId,
});
if (!workflowRunToUpdate) {
throw new WorkflowRunException(
`workflowRun ${workflowRunId} not found`,
WorkflowRunExceptionCode.WORKFLOW_RUN_NOT_FOUND,
);
}
await workflowRunRepository.update(
workflowRunToUpdate.id,
partialUpdate,
undefined,
['id'],
await this.globalWorkspaceOrmManager.executeInWorkspaceContext(async () => {
const workflowRunRepository =
await this.globalWorkspaceOrmManager.getRepository<WorkflowRunWorkspaceEntity>(
workspaceId,
'workflowRun',
{ shouldBypassPermissionChecks: true },
);
},
);
const workflowRunToUpdate = await workflowRunRepository.findOneBy({
id: workflowRunId,
});
if (!workflowRunToUpdate) {
throw new WorkflowRunException(
`workflowRun ${workflowRunId} not found`,
WorkflowRunExceptionCode.WORKFLOW_RUN_NOT_FOUND,
);
}
await workflowRunRepository.update(
workflowRunToUpdate.id,
partialUpdate,
undefined,
['id'],
);
}, authContext);
}
private getInitState(
packages/twenty-server/src/modules/workflow/workflow-status/jobs/__tests__/workflow-statuses-update.job.spec.ts
+1 -1
View File
@@ -49,7 +49,7 @@ describe('WorkflowStatusesUpdate', () => {
executeInWorkspaceContext: jest
.fn()
.mockImplementation((_authContext: any, fn: () => any) => fn()),
.mockImplementation((fn: () => any, _authContext?: any) => fn()),
};
const mockServerlessFunctionService = {
packages/twenty-server/src/modules/workflow/workflow-status/jobs/workflow-statuses-update.job.ts
+27 -30
View File
@@ -78,36 +78,33 @@ export class WorkflowStatusesUpdateJob {
async handle(event: WorkflowVersionBatchEvent): Promise<void> {
const authContext = buildSystemAuthContext(event.workspaceId);
await this.globalWorkspaceOrmManager.executeInWorkspaceContext(
authContext,
async () => {
switch (event.type) {
case WorkflowVersionEventType.CREATE:
case WorkflowVersionEventType.DELETE:
await Promise.all(
event.workflowIds.map((workflowId) =>
this.handleWorkflowVersionCreatedOrDeleted({
workflowId,
workspaceId: event.workspaceId,
}),
),
);
break;
case WorkflowVersionEventType.STATUS_UPDATE:
await Promise.all(
event.statusUpdates.map((statusUpdate) =>
this.handleWorkflowVersionStatusUpdated({
statusUpdate,
workspaceId: event.workspaceId,
}),
),
);
break;
default:
break;
}
},
);
await this.globalWorkspaceOrmManager.executeInWorkspaceContext(async () => {
switch (event.type) {
case WorkflowVersionEventType.CREATE:
case WorkflowVersionEventType.DELETE:
await Promise.all(
event.workflowIds.map((workflowId) =>
this.handleWorkflowVersionCreatedOrDeleted({
workflowId,
workspaceId: event.workspaceId,
}),
),
);
break;
case WorkflowVersionEventType.STATUS_UPDATE:
await Promise.all(
event.statusUpdates.map((statusUpdate) =>
this.handleWorkflowVersionStatusUpdated({
statusUpdate,
workspaceId: event.workspaceId,
}),
),
);
break;
default:
break;
}
}, authContext);
}
private async handleWorkflowVersionCreatedOrDeleted({
packages/twenty-server/src/modules/workflow/workflow-tools/tools/create-complete-workflow.tool.ts
+65 -74
View File
@@ -206,38 +206,35 @@ const createWorkflow = async ({
}): Promise<string> => {
const authContext = buildSystemAuthContext(context.workspaceId);
return deps.globalWorkspaceOrmManager.executeInWorkspaceContext(
authContext,
async () => {
const workflowRepository =
await deps.globalWorkspaceOrmManager.getRepository(
context.workspaceId,
'workflow',
context.rolePermissionConfig,
);
return deps.globalWorkspaceOrmManager.executeInWorkspaceContext(async () => {
const workflowRepository =
await deps.globalWorkspaceOrmManager.getRepository(
context.workspaceId,
'workflow',
context.rolePermissionConfig,
);
const workflowPosition =
await deps.recordPositionService.buildRecordPosition({
value: 'first',
objectMetadata: {
isCustom: false,
nameSingular: 'workflow',
},
workspaceId: context.workspaceId,
});
const workflowPosition =
await deps.recordPositionService.buildRecordPosition({
value: 'first',
objectMetadata: {
isCustom: false,
nameSingular: 'workflow',
},
workspaceId: context.workspaceId,
});
const workflow = {
id: uuidv4(),
name,
statuses: [WorkflowStatus.DRAFT],
position: workflowPosition,
};
const workflow = {
id: uuidv4(),
name,
statuses: [WorkflowStatus.DRAFT],
position: workflowPosition,
};
await workflowRepository.insert(workflow);
await workflowRepository.insert(workflow);
return workflow.id;
},
);
return workflow.id;
}, authContext);
};
const createWorkflowVersion = async ({
@@ -255,41 +252,38 @@ const createWorkflowVersion = async ({
}): Promise<string> => {
const authContext = buildSystemAuthContext(context.workspaceId);
return deps.globalWorkspaceOrmManager.executeInWorkspaceContext(
authContext,
async () => {
const workflowVersionRepository =
await deps.globalWorkspaceOrmManager.getRepository(
context.workspaceId,
'workflowVersion',
context.rolePermissionConfig,
);
return deps.globalWorkspaceOrmManager.executeInWorkspaceContext(async () => {
const workflowVersionRepository =
await deps.globalWorkspaceOrmManager.getRepository(
context.workspaceId,
'workflowVersion',
context.rolePermissionConfig,
);
const versionPosition =
await deps.recordPositionService.buildRecordPosition({
value: 'first',
objectMetadata: {
isCustom: false,
nameSingular: 'workflowVersion',
},
workspaceId: context.workspaceId,
});
const versionPosition =
await deps.recordPositionService.buildRecordPosition({
value: 'first',
objectMetadata: {
isCustom: false,
nameSingular: 'workflowVersion',
},
workspaceId: context.workspaceId,
});
const workflowVersion = {
id: uuidv4(),
workflowId,
name: 'v1',
status: WorkflowVersionStatus.DRAFT,
trigger,
steps,
position: versionPosition,
};
const workflowVersion = {
id: uuidv4(),
workflowId,
name: 'v1',
status: WorkflowVersionStatus.DRAFT,
trigger,
steps,
position: versionPosition,
};
await workflowVersionRepository.insert(workflowVersion);
await workflowVersionRepository.insert(workflowVersion);
return workflowVersion.id;
},
);
return workflowVersion.id;
}, authContext);
};
const updateWorkflowStatus = async ({
@@ -305,20 +299,17 @@ const updateWorkflowStatus = async ({
}) => {
const authContext = buildSystemAuthContext(context.workspaceId);
await deps.globalWorkspaceOrmManager.executeInWorkspaceContext(
authContext,
async () => {
const workflowRepository =
await deps.globalWorkspaceOrmManager.getRepository(
context.workspaceId,
'workflow',
context.rolePermissionConfig,
);
await deps.globalWorkspaceOrmManager.executeInWorkspaceContext(async () => {
const workflowRepository =
await deps.globalWorkspaceOrmManager.getRepository(
context.workspaceId,
'workflow',
context.rolePermissionConfig,
);
await workflowRepository.update(workflowId, {
statuses: [WorkflowStatus.ACTIVE],
lastPublishedVersionId: workflowVersionId,
});
},
);
await workflowRepository.update(workflowId, {
statuses: [WorkflowStatus.ACTIVE],
lastPublishedVersionId: workflowVersionId,
});
}, authContext);
};
packages/twenty-server/src/modules/workflow/workflow-tools/tools/get-workflow-current-version.tool.ts
+1 -1
View File
@@ -35,7 +35,6 @@ export const createGetWorkflowCurrentVersionTool = (
const authContext = buildSystemAuthContext(context.workspaceId);
return await deps.globalWorkspaceOrmManager.executeInWorkspaceContext(
authContext,
async () => {
const workflowRepository =
await deps.globalWorkspaceOrmManager.getRepository<WorkflowWorkspaceEntity>(
@@ -103,6 +102,7 @@ export const createGetWorkflowCurrentVersionTool = (
},
};
},
authContext,
);
} catch (error) {
return {
packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/automated-trigger.workspace-service.ts
+20 -26
View File
@@ -27,22 +27,19 @@ export class AutomatedTriggerWorkspaceService {
}) {
const authContext = buildSystemAuthContext(workspaceId);
await this.globalWorkspaceOrmManager.executeInWorkspaceContext(
authContext,
async () => {
const workflowAutomatedTriggerRepository =
await this.globalWorkspaceOrmManager.getRepository<WorkflowAutomatedTriggerWorkspaceEntity>(
workspaceId,
'workflowAutomatedTrigger',
);
await this.globalWorkspaceOrmManager.executeInWorkspaceContext(async () => {
const workflowAutomatedTriggerRepository =
await this.globalWorkspaceOrmManager.getRepository<WorkflowAutomatedTriggerWorkspaceEntity>(
workspaceId,
'workflowAutomatedTrigger',
);
await workflowAutomatedTriggerRepository.insert({
type,
settings,
workflowId,
});
},
);
await workflowAutomatedTriggerRepository.insert({
type,
settings,
workflowId,
});
}, authContext);
}
async deleteAutomatedTrigger({
@@ -54,17 +51,14 @@ export class AutomatedTriggerWorkspaceService {
}) {
const authContext = buildSystemAuthContext(workspaceId);
await this.globalWorkspaceOrmManager.executeInWorkspaceContext(
authContext,
async () => {
const workflowAutomatedTriggerRepository =
await this.globalWorkspaceOrmManager.getRepository<WorkflowAutomatedTriggerWorkspaceEntity>(
workspaceId,
'workflowAutomatedTrigger',
);
await this.globalWorkspaceOrmManager.executeInWorkspaceContext(async () => {
const workflowAutomatedTriggerRepository =
await this.globalWorkspaceOrmManager.getRepository<WorkflowAutomatedTriggerWorkspaceEntity>(
workspaceId,
'workflowAutomatedTrigger',
);
await workflowAutomatedTriggerRepository.delete({ workflowId });
},
);
await workflowAutomatedTriggerRepository.delete({ workflowId });
}, authContext);
}
}
packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/listeners/__tests__/workflow-database-event-trigger.listener.spec.ts
+1 -1
View File
@@ -52,7 +52,7 @@ describe('WorkflowDatabaseEventTriggerListener', () => {
getRepository: jest.fn().mockResolvedValue(mockRepository),
executeInWorkspaceContext: jest
.fn()
.mockImplementation((_authContext: any, fn: () => any) => fn()),
.mockImplementation((fn: () => any, _authContext?: any) => fn()),
} as any;
messageQueueService = {
packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/listeners/workflow-database-event-trigger.listener.ts
+86 -93
View File
@@ -247,64 +247,60 @@ export class WorkflowDatabaseEventTriggerListener {
}) {
const authContext = buildSystemAuthContext(workspaceId);
await this.globalWorkspaceOrmManager.executeInWorkspaceContext(
authContext,
async () => {
const { fieldIdByJoinColumnName } =
buildFieldMapsFromFlatObjectMetadata(
flatFieldMetadataMaps,
flatObjectMetadata,
await this.globalWorkspaceOrmManager.executeInWorkspaceContext(async () => {
const { fieldIdByJoinColumnName } = buildFieldMapsFromFlatObjectMetadata(
flatFieldMetadataMaps,
flatObjectMetadata,
);
for (const [joinColumnName, joinFieldId] of Object.entries(
fieldIdByJoinColumnName,
)) {
const joinField = findFlatEntityByIdInFlatEntityMapsOrThrow({
flatEntityMaps: flatFieldMetadataMaps,
flatEntityId: joinFieldId,
});
const joinRecordIds = records
.map((record) => record[joinColumnName])
.filter(isDefined);
if (joinRecordIds.length === 0) {
continue;
}
const relatedObjectMetadataId =
joinField.relationTargetObjectMetadataId;
if (!isDefined(relatedObjectMetadataId)) {
continue;
}
const relatedObjectMetadataNameSingular =
flatObjectMetadataMaps.byId[relatedObjectMetadataId]?.nameSingular;
if (!isDefined(relatedObjectMetadataNameSingular)) {
continue;
}
const relatedObjectRepository =
await this.globalWorkspaceOrmManager.getRepository(
workspaceId,
relatedObjectMetadataNameSingular,
{ shouldBypassPermissionChecks: true },
);
for (const [joinColumnName, joinFieldId] of Object.entries(
fieldIdByJoinColumnName,
)) {
const joinField = findFlatEntityByIdInFlatEntityMapsOrThrow({
flatEntityMaps: flatFieldMetadataMaps,
flatEntityId: joinFieldId,
});
const relatedRecords = await relatedObjectRepository.find({
where: { id: In(joinRecordIds) },
});
const joinRecordIds = records
.map((record) => record[joinColumnName])
.filter(isDefined);
if (joinRecordIds.length === 0) {
continue;
}
const relatedObjectMetadataId =
joinField.relationTargetObjectMetadataId;
if (!isDefined(relatedObjectMetadataId)) {
continue;
}
const relatedObjectMetadataNameSingular =
flatObjectMetadataMaps.byId[relatedObjectMetadataId]?.nameSingular;
if (!isDefined(relatedObjectMetadataNameSingular)) {
continue;
}
const relatedObjectRepository =
await this.globalWorkspaceOrmManager.getRepository(
workspaceId,
relatedObjectMetadataNameSingular,
{ shouldBypassPermissionChecks: true },
);
const relatedRecords = await relatedObjectRepository.find({
where: { id: In(joinRecordIds) },
});
for (const record of records) {
record[joinField.name] = relatedRecords.find(
(relatedRecord) => relatedRecord.id === record[joinColumnName],
);
}
for (const record of records) {
record[joinField.name] = relatedRecords.find(
(relatedRecord) => relatedRecord.id === record[joinColumnName],
);
}
},
);
}
}, authContext);
}
private async shouldIgnoreEvent(
@@ -339,50 +335,47 @@ export class WorkflowDatabaseEventTriggerListener {
const authContext = buildSystemAuthContext(workspaceId);
await this.globalWorkspaceOrmManager.executeInWorkspaceContext(
authContext,
async () => {
const workflowAutomatedTriggerRepository =
await this.globalWorkspaceOrmManager.getRepository<WorkflowAutomatedTriggerWorkspaceEntity>(
workspaceId,
automatedTriggerTableName,
{ shouldBypassPermissionChecks: true },
);
await this.globalWorkspaceOrmManager.executeInWorkspaceContext(async () => {
const workflowAutomatedTriggerRepository =
await this.globalWorkspaceOrmManager.getRepository<WorkflowAutomatedTriggerWorkspaceEntity>(
workspaceId,
automatedTriggerTableName,
{ shouldBypassPermissionChecks: true },
);
const eventListeners = await workflowAutomatedTriggerRepository.find({
where: {
type: AutomatedTriggerType.DATABASE_EVENT,
settings: Raw(
() =>
`"${automatedTriggerTableName}"."settings"->>'eventName' = :eventName`,
{ eventName: databaseEventName },
),
},
});
const eventListeners = await workflowAutomatedTriggerRepository.find({
where: {
type: AutomatedTriggerType.DATABASE_EVENT,
settings: Raw(
() =>
`"${automatedTriggerTableName}"."settings"->>'eventName' = :eventName`,
{ eventName: databaseEventName },
),
},
});
for (const eventListener of eventListeners) {
for (const eventPayload of payload.events) {
const shouldTriggerJob = this.shouldTriggerJob({
eventPayload,
eventListener,
action,
});
for (const eventListener of eventListeners) {
for (const eventPayload of payload.events) {
const shouldTriggerJob = this.shouldTriggerJob({
eventPayload,
eventListener,
action,
});
if (shouldTriggerJob) {
await this.messageQueueService.add<WorkflowTriggerJobData>(
WorkflowTriggerJob.name,
{
workspaceId,
workflowId: eventListener.workflowId,
payload: eventPayload,
},
{ retryLimit: 3 },
);
}
if (shouldTriggerJob) {
await this.messageQueueService.add<WorkflowTriggerJobData>(
WorkflowTriggerJob.name,
{
workspaceId,
workflowId: eventListener.workflowId,
payload: eventPayload,
},
{ retryLimit: 3 },
);
}
}
},
);
}
}, authContext);
}
private shouldTriggerJob({
packages/twenty-server/src/modules/workflow/workflow-trigger/jobs/workflow-trigger.job.ts
+70 -73
View File
@@ -44,81 +44,78 @@ export class WorkflowTriggerJob {
async handle(data: WorkflowTriggerJobData): Promise<void> {
const authContext = buildSystemAuthContext(data.workspaceId);
await this.globalWorkspaceOrmManager.executeInWorkspaceContext(
authContext,
async () => {
try {
const workflowRepository =
await this.globalWorkspaceOrmManager.getRepository<WorkflowWorkspaceEntity>(
data.workspaceId,
'workflow',
{ shouldBypassPermissionChecks: true },
);
await this.globalWorkspaceOrmManager.executeInWorkspaceContext(async () => {
try {
const workflowRepository =
await this.globalWorkspaceOrmManager.getRepository<WorkflowWorkspaceEntity>(
data.workspaceId,
'workflow',
{ shouldBypassPermissionChecks: true },
);
const workflow = await workflowRepository.findOneBy({
id: data.workflowId,
});
const workflow = await workflowRepository.findOneBy({
id: data.workflowId,
});
if (!workflow) {
throw new WorkflowTriggerException(
`Workflow ${data.workflowId} not found in workspace ${data.workspaceId}`,
WorkflowTriggerExceptionCode.NOT_FOUND,
);
}
if (!workflow.lastPublishedVersionId) {
throw new WorkflowTriggerException(
`Workflow ${data.workflowId} has no published version in workspace ${data.workspaceId}`,
WorkflowTriggerExceptionCode.INTERNAL_ERROR,
);
}
const workflowVersionRepository =
await this.globalWorkspaceOrmManager.getRepository<WorkflowVersionWorkspaceEntity>(
data.workspaceId,
'workflowVersion',
{ shouldBypassPermissionChecks: true },
);
const workflowVersion = await workflowVersionRepository.findOneBy({
id: workflow.lastPublishedVersionId,
});
if (!workflowVersion) {
throw new WorkflowTriggerException(
`Workflow version ${workflow.lastPublishedVersionId} not found in workspace ${data.workspaceId}`,
WorkflowTriggerExceptionCode.NOT_FOUND,
);
}
if (workflowVersion.status !== WorkflowVersionStatus.ACTIVE) {
throw new WorkflowTriggerException(
`Workflow version ${workflowVersion.id} is not active in workspace ${data.workspaceId}`,
WorkflowTriggerExceptionCode.INTERNAL_ERROR,
);
}
await this.workflowRunnerWorkspaceService.run({
workspaceId: data.workspaceId,
workflowVersionId: workflow.lastPublishedVersionId,
payload: data.payload,
source: {
source: FieldActorSource.WORKFLOW,
name:
isDefined(workflow.name) && !isEmpty(workflow.name)
? workflow.name
: DEFAULT_WORKFLOW_NAME,
context: {},
workspaceMemberId: null,
},
});
} catch (e) {
await this.messageQueueService.removeCron({
jobName: WorkflowTriggerJob.name,
jobId: data.workflowId,
});
handleWorkflowTriggerException(e);
if (!workflow) {
throw new WorkflowTriggerException(
`Workflow ${data.workflowId} not found in workspace ${data.workspaceId}`,
WorkflowTriggerExceptionCode.NOT_FOUND,
);
}
},
);
if (!workflow.lastPublishedVersionId) {
throw new WorkflowTriggerException(
`Workflow ${data.workflowId} has no published version in workspace ${data.workspaceId}`,
WorkflowTriggerExceptionCode.INTERNAL_ERROR,
);
}
const workflowVersionRepository =
await this.globalWorkspaceOrmManager.getRepository<WorkflowVersionWorkspaceEntity>(
data.workspaceId,
'workflowVersion',
{ shouldBypassPermissionChecks: true },
);
const workflowVersion = await workflowVersionRepository.findOneBy({
id: workflow.lastPublishedVersionId,
});
if (!workflowVersion) {
throw new WorkflowTriggerException(
`Workflow version ${workflow.lastPublishedVersionId} not found in workspace ${data.workspaceId}`,
WorkflowTriggerExceptionCode.NOT_FOUND,
);
}
if (workflowVersion.status !== WorkflowVersionStatus.ACTIVE) {
throw new WorkflowTriggerException(
`Workflow version ${workflowVersion.id} is not active in workspace ${data.workspaceId}`,
WorkflowTriggerExceptionCode.INTERNAL_ERROR,
);
}
await this.workflowRunnerWorkspaceService.run({
workspaceId: data.workspaceId,
workflowVersionId: workflow.lastPublishedVersionId,
payload: data.payload,
source: {
source: FieldActorSource.WORKFLOW,
name:
isDefined(workflow.name) && !isEmpty(workflow.name)
? workflow.name
: DEFAULT_WORKFLOW_NAME,
context: {},
workspaceMemberId: null,
},
});
} catch (e) {
await this.messageQueueService.removeCron({
jobName: WorkflowTriggerJob.name,
jobId: data.workflowId,
});
handleWorkflowTriggerException(e);
}
}, authContext);
}
}
packages/twenty-server/src/modules/workflow/workflow-trigger/workspace-services/workflow-trigger.workspace-service.ts
+2 -2
View File
@@ -82,7 +82,6 @@ export class WorkflowTriggerWorkspaceService {
const authContext = buildSystemAuthContext(workspaceId);
return this.globalWorkspaceOrmManager.executeInWorkspaceContext(
authContext,
async () => {
const workflowVersionRepository =
await this.globalWorkspaceOrmManager.getRepository<WorkflowVersionWorkspaceEntity>(
@@ -132,6 +131,7 @@ export class WorkflowTriggerWorkspaceService {
return true;
},
authContext,
);
}
@@ -142,7 +142,6 @@ export class WorkflowTriggerWorkspaceService {
const authContext = buildSystemAuthContext(workspaceId);
return this.globalWorkspaceOrmManager.executeInWorkspaceContext(
authContext,
async () => {
const workflowVersionRepository =
await this.globalWorkspaceOrmManager.getRepository<WorkflowVersionWorkspaceEntity>(
@@ -159,6 +158,7 @@ export class WorkflowTriggerWorkspaceService {
return true;
},
authContext,
);
}