feat(workflow): Pick Record load balanced strategy (3/3) (#21902)

## Overview Final PR in the Pick Record stack. Adds the **Load Balanced** strategy: pick the candidate that currently has the *fewest related records*. This is the "fair assignment" mode — e.g. assign a new company to the account owner who currently owns the fewest companies, or route a lead to the rep with the fewest open opportunities. **Stacked on #21900** (which is stacked on #21899) — merge in order. This PR's diff against `main` includes PRs 1 & 2 until they merge. ## What changed - Widened the `strategy` enum to add `LOAD_BALANCED`, and added an optional `loadBalance: { objectNameSingular, fieldName }` to the action input. - Editor: selecting **Load balanced** reveals a **Balance by** object picker and a **Count by** field picker (the related object's many-to-one relation fields). - Executor: for each candidate, counts records of the chosen related object whose chosen relation points at that candidate, then selects the least-loaded one. ## How it works Given pool = workspace members and config `{ objectNameSingular: "opportunity", fieldName: "pointOfContact" }`, the executor counts, per member, the opportunities whose `pointOfContact` is that member, and picks the member with the lowest count. ## Design decisions & tradeoffs 1. **No persistent state — computed live each run.** Unlike round robin, load balancing reads current data, so there's no cursor to store. Correct by construction even under concurrency (each run recomputes counts); the only caveat is two simultaneous runs can both see the same "least loaded" candidate before either assignment lands (a small, self-correcting skew), which is inherent to load-balancing and acceptable. 2. **Count via per-candidate queries.** One filtered count per candidate (`{ [relationField]: { id: { eq: candidateId } } }`), run in parallel. For the realistic pool sizes this targets (a team), this is simple and clear. A single `group_by` aggregate would scale better for very large pools — noted as a future optimization, deliberately not done to keep the logic obvious. 3. **Deterministic tie-break.** Candidates are pre-sorted by id (shared with round robin), and the first minimum wins — so equal-load ties resolve deterministically rather than arbitrarily. 4. **`Count by` lists all many-to-one relations of the chosen object** (not filtered to those targeting the pool object). Keeps the editor simple; picking an unrelated field just yields zero counts, which is visibly wrong. Filtering options to relations that target the pool object is a nice follow-up. 5. **Filter on the counted set** (e.g. only *open* opportunities) is intentionally out of scope for this first cut — documented as a follow-up. ## Testing Added `pick-record-load-balanced-workflow.integration-spec.ts`: creates two fresh companies (0 related opportunities each), attaches one opportunity to the second, configures `LOAD_BALANCED` counting opportunities by `company`, and asserts the step picks the **first** company (0 < 1). Passes locally alongside the random and round-robin tests (3 suites / 4 tests). `typecheck` + `lint:diff-with-main` green for shared/server/front. ## The full stack 1. #21899 — Random (the action + the whole scaffold) 2. #21900 — Round robin (atomic Redis cursor) 3. this — Load balanced Together these enable round-robin / load-balanced / random **assignment workflows** in Twenty, composed via the standard variable picker (assign the chosen record downstream with `{{step.<id>.id}}`). https://claude.ai/code/session_01MuPWZsqf2bmQSevk6fRbX8 --- _Generated by [Claude Code](https://claude.ai/code/session_01MuPWZsqf2bmQSevk6fRbX8)_  <a href="https://cubic.dev/pr/twentyhq/twenty/pull/21902?utm_source=github" target="_blank" rel="noopener noreferrer" data-no-image-dialog="true"><picture><source media="(prefers-color-scheme: dark)" srcset="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"><source media="(prefers-color-scheme: light)" srcset="https://www.cubic.dev/buttons/review-in-cubic-light.svg"><img alt="Review in cubic" src="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"></picture></a>
2026-06-22 07:09:15 +02:00
parent 82f6597dc3
commit 2abf9c2930
11 changed files with 824 additions and 10 deletions
@@ -0,0 +1,69 @@
+import { isNonEmptyString } from '@sniptt/guards';
+import { FieldMetadataType, RelationType } from 'twenty-shared/types';
+import { isDefined } from 'twenty-shared/utils';
+
+import { type FlatEntityMaps } from 'src/engine/metadata-modules/flat-entity/types/flat-entity-maps.type';
+import { type FlatFieldMetadata } from 'src/engine/metadata-modules/flat-field-metadata/types/flat-field-metadata.type';
+import { isFlatFieldMetadataOfType } from 'src/engine/metadata-modules/flat-field-metadata/utils/is-flat-field-metadata-of-type.util';
+import { type WorkflowPickRecordAction } from 'src/modules/workflow/workflow-executor/workflow-actions/types/workflow-action.type';
+
+export const getPickRecordLoadBalanceConfigError = ({
+  step,
+  objectIdByNameSingular,
+  flatFieldMetadataMaps,
+}: {
+  step: WorkflowPickRecordAction;
+  objectIdByNameSingular: Record<string, string>;
+  flatFieldMetadataMaps: FlatEntityMaps<FlatFieldMetadata>;
+}): string | undefined => {
+  const input = step.settings?.input;
+
+  if (!isDefined(input) || input.strategy !== 'LOAD_BALANCED') {
+    return undefined;
+  }
+
+  const stepName = step.name ?? step.id;
+  const loadBalance = input.loadBalance;
+
+  if (
+    !isNonEmptyString(loadBalance?.objectNameSingular) ||
+    !isNonEmptyString(loadBalance?.fieldName)
+  ) {
+    return `Step "${stepName}" uses load balancing but is missing the object and field to count by.`;
+  }
+
+  const poolObjectId = objectIdByNameSingular[input.objectName];
+
+  if (!isDefined(poolObjectId)) {
+    return `Step "${stepName}" picks from object "${input.objectName}" which does not exist in this workspace.`;
+  }
+
+  const loadBalanceObjectId =
+    objectIdByNameSingular[loadBalance.objectNameSingular];
+
+  if (!isDefined(loadBalanceObjectId)) {
+    return `Step "${stepName}" balances load by object "${loadBalance.objectNameSingular}" which does not exist in this workspace.`;
+  }
+
+  const countByField = Object.values(
+    flatFieldMetadataMaps.byUniversalIdentifier,
+  )
+    .filter(isDefined)
+    .find(
+      (field) =>
+        field.objectMetadataId === loadBalanceObjectId &&
+        field.name === loadBalance.fieldName,
+    );
+
+  const countsRelatedRecordsForPool =
+    isDefined(countByField) &&
+    isFlatFieldMetadataOfType(countByField, FieldMetadataType.RELATION) &&
+    countByField.settings?.relationType === RelationType.MANY_TO_ONE &&
+    countByField.relationTargetObjectMetadataId === poolObjectId;
+
+  if (!countsRelatedRecordsForPool) {
+    return `Step "${stepName}" must balance load by a many-to-one relation on "${loadBalance.objectNameSingular}" that points to "${input.objectName}".`;
+  }
+
+  return undefined;
+};
@@ -27,6 +27,7 @@ import {

 import { WorkflowCommonWorkspaceService } from 'src/modules/workflow/common/workspace-services/workflow-common.workspace-service';
 import { WorkflowSchemaWorkspaceService } from 'src/modules/workflow/workflow-builder/workflow-schema/workflow-schema.workspace-service';
+import { getPickRecordLoadBalanceConfigError } from 'src/modules/workflow/workflow-builder/workflow-validation/utils/get-pick-record-load-balance-config-error.util';
 import {
  type WorkflowAction,
  type WorkflowAiAgentAction,
@@ -590,7 +591,7 @@ export class WorkflowValidationWorkspaceService {
      return [];
    }

-    const { objectIdByNameSingular } =
+    const { objectIdByNameSingular, flatFieldMetadataMaps } =
      await this.workflowCommonWorkspaceService.getFlatEntityMaps(workspaceId);

    const issues: WorkflowValidationIssue[] = [];
@@ -618,6 +619,25 @@ export class WorkflowValidationWorkspaceService {
          message: `Step "${step.name ?? step.id}" targets object "${objectName}" which does not exist in this workspace.`,
          stepId: step.id,
        });
+
+        continue;
+      }
+
+      if (step.type === WorkflowActionType.PICK_RECORD) {
+        const loadBalanceError = getPickRecordLoadBalanceConfigError({
+          step,
+          objectIdByNameSingular,
+          flatFieldMetadataMaps,
+        });
+
+        if (isDefined(loadBalanceError)) {
+          issues.push({
+            severity: 'error',
+            code: 'INVALID_STEP_PARAMS',
+            message: loadBalanceError,
+            stepId: step.id,
+          });
+        }
      }
    }

@@ -2,7 +2,7 @@ import { Injectable } from '@nestjs/common';

 import { isNonEmptyString } from '@sniptt/guards';
 import { type ObjectRecord } from 'twenty-shared/types';
-import { resolveInput } from 'twenty-shared/utils';
+import { isDefined, resolveInput } from 'twenty-shared/utils';

 import { type WorkflowAction } from 'src/modules/workflow/workflow-executor/interfaces/workflow-action.interface';

@@ -10,6 +10,7 @@ import { InjectCacheStorage } from 'src/engine/core-modules/cache-storage/decora
 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 { FindRecordsService } from 'src/engine/core-modules/record-crud/services/find-records.service';
+import { GroupByRecordsService } from 'src/engine/core-modules/record-crud/services/group-by-records.service';
 import {
  WorkflowStepExecutorException,
  WorkflowStepExecutorExceptionCode,
@@ -21,15 +22,21 @@ import { findStepOrThrow } from 'src/modules/workflow/workflow-executor/utils/fi
 import { isWorkflowPickRecordAction } from 'src/modules/workflow/workflow-executor/workflow-actions/record-crud/guards/is-workflow-pick-record-action.guard';
 import {
  type WorkflowPickRecordActionInput,
+  type WorkflowPickRecordLoadBalance,
  type WorkflowPickRecordStrategy,
 } from 'src/modules/workflow/workflow-executor/workflow-actions/record-crud/types/workflow-pick-record-action-input.type';

 const ROUND_ROBIN_CURSOR_TTL_MS = 1000 * 60 * 60 * 24 * 90;

+type PickRecordExecutionContext = Awaited<
+  ReturnType<WorkflowExecutionContextService['getExecutionContext']>
+>;
+
@Injectable()
 export class PickRecordWorkflowAction implements WorkflowAction {
  constructor(
    private readonly findRecordsService: FindRecordsService,
+    private readonly groupByRecordsService: GroupByRecordsService,
    private readonly workflowExecutionContextService: WorkflowExecutionContextService,
    @InjectCacheStorage(CacheStorageNamespace.ModuleWorkflow)
    private readonly cacheStorageService: CacheStorageService,
@@ -53,7 +60,7 @@ export class PickRecordWorkflowAction implements WorkflowAction {
      );
    }

-    const { objectName, strategy, recordIds } = resolveInput(
+    const { objectName, strategy, recordIds, loadBalance } = resolveInput(
      step.settings.input,
      context,
    ) as WorkflowPickRecordActionInput;
@@ -94,6 +101,27 @@ export class PickRecordWorkflowAction implements WorkflowAction {
      String(recordA.id).localeCompare(String(recordB.id)),
    );

+    if (strategy === 'LOAD_BALANCED') {
+      if (!isDefined(loadBalance)) {
+        return {
+          error:
+            'Pick record action is missing its load balancing configuration',
+        };
+      }
+
+      const leastLoadedResult = await this.getLeastLoadedIndex({
+        orderedRecords,
+        loadBalance,
+        executionContext,
+      });
+
+      if ('error' in leastLoadedResult) {
+        return { error: leastLoadedResult.error };
+      }
+
+      return { result: orderedRecords[leastLoadedResult.index] };
+    }
+
    const pickedIndex = await this.getPickedIndex({
      strategy,
      candidateCount: orderedRecords.length,
@@ -104,6 +132,59 @@ export class PickRecordWorkflowAction implements WorkflowAction {
    return { result: orderedRecords[pickedIndex] };
  }

+  private async getLeastLoadedIndex({
+    orderedRecords,
+    loadBalance,
+    executionContext,
+  }: {
+    orderedRecords: ObjectRecord[];
+    loadBalance: WorkflowPickRecordLoadBalance;
+    executionContext: PickRecordExecutionContext;
+  }): Promise<{ index: number } | { error: string }> {
+    const candidateIds = orderedRecords.map((record) => String(record.id));
+
+    const groupByOutput = await this.groupByRecordsService.execute({
+      objectName: loadBalance.objectNameSingular,
+      groupBy: [{ [loadBalance.fieldName]: { id: true } }],
+      filter: { [loadBalance.fieldName]: { id: { in: candidateIds } } },
+      authContext: executionContext.authContext,
+      rolePermissionConfig: executionContext.rolePermissionConfig,
+    });
+
+    if (!groupByOutput.success) {
+      return {
+        error:
+          groupByOutput.error ||
+          groupByOutput.message ||
+          'Pick record action failed to count related records',
+      };
+    }
+
+    const countByCandidateId = new Map<string, number>();
+
+    for (const group of groupByOutput.result?.groups ?? []) {
+      const candidateId = group.dimensions[0];
+
+      if (isDefined(candidateId)) {
+        countByCandidateId.set(String(candidateId), Number(group.value ?? 0));
+      }
+    }
+
+    let leastLoadedIndex = 0;
+    let leastLoadedCount = countByCandidateId.get(candidateIds[0]) ?? 0;
+
+    for (let index = 1; index < candidateIds.length; index++) {
+      const candidateCount = countByCandidateId.get(candidateIds[index]) ?? 0;
+
+      if (candidateCount < leastLoadedCount) {
+        leastLoadedIndex = index;
+        leastLoadedCount = candidateCount;
+      }
+    }
+
+    return { index: leastLoadedIndex };
+  }
+
  private async getPickedIndex({
    strategy,
    candidateCount,
@@ -1,7 +1,16 @@
-export type WorkflowPickRecordStrategy = 'RANDOM' | 'ROUND_ROBIN';
+export type WorkflowPickRecordStrategy =
+  | 'RANDOM'
+  | 'ROUND_ROBIN'
+  | 'LOAD_BALANCED';
+
+export type WorkflowPickRecordLoadBalance = {
+  objectNameSingular: string;
+  fieldName: string;
+};

 export type WorkflowPickRecordActionInput = {
  objectName: string;
  strategy: WorkflowPickRecordStrategy;
  recordIds: string[];
+  loadBalance?: WorkflowPickRecordLoadBalance;
 };
@@ -1,6 +1,8 @@
 import { msg } from '@lingui/core/macro';
 import { Injectable, Logger } from '@nestjs/common';
 import { type ActorMetadata } from 'twenty-shared/types';
+import { isDefined } from 'twenty-shared/utils';
+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';
@@ -21,7 +23,12 @@ import {
 import { type WorkflowWorkspaceEntity } from 'src/modules/workflow/common/standard-objects/workflow.workspace-entity';
 import { assertWorkflowVersionTriggerIsDefined } from 'src/modules/workflow/common/utils/assert-workflow-version-trigger-is-defined.util';
 import { WorkflowCommonWorkspaceService } from 'src/modules/workflow/common/workspace-services/workflow-common.workspace-service';
+import { getPickRecordLoadBalanceConfigError } from 'src/modules/workflow/workflow-builder/workflow-validation/utils/get-pick-record-load-balance-config-error.util';
 import { CodeStepBuildService } from 'src/modules/workflow/workflow-builder/workflow-version-step/code-step/services/code-step-build.service';
+import {
+  type WorkflowAction,
+  type WorkflowPickRecordAction,
+} from 'src/modules/workflow/workflow-executor/workflow-actions/types/workflow-action.type';
 import { WorkflowRunnerWorkspaceService } from 'src/modules/workflow/workflow-runner/workspace-services/workflow-runner.workspace-service';
 import { WORKFLOW_VERSION_STATUS_UPDATED } from 'src/modules/workflow/workflow-status/constants/workflow-version-status-updated.constants';
 import { type WorkflowVersionStatusUpdate } from 'src/modules/workflow/workflow-status/jobs/workflow-statuses-update.job';
@@ -131,6 +138,11 @@ export class WorkflowTriggerWorkspaceService {

        assertVersionCanBeActivated(workflowVersion, workflow);

+        await this.assertPickRecordLoadBalanceConfigIsValid({
+          steps: workflowVersion.steps ?? [],
+          workspaceId,
+        });
+
        await this.codeStepBuildService.buildCodeStepsFromSourceForSteps({
          workspaceId,
          steps: workflowVersion.steps ?? [],
@@ -155,6 +167,41 @@ export class WorkflowTriggerWorkspaceService {
    );
  }

+  private async assertPickRecordLoadBalanceConfigIsValid({
+    steps,
+    workspaceId,
+  }: {
+    steps: WorkflowAction[];
+    workspaceId: string;
+  }) {
+    const pickRecordSteps = steps.filter(
+      (step): step is WorkflowPickRecordAction =>
+        step.type === WorkflowActionType.PICK_RECORD,
+    );
+
+    if (pickRecordSteps.length === 0) {
+      return;
+    }
+
+    const { objectIdByNameSingular, flatFieldMetadataMaps } =
+      await this.workflowCommonWorkspaceService.getFlatEntityMaps(workspaceId);
+
+    for (const step of pickRecordSteps) {
+      const loadBalanceError = getPickRecordLoadBalanceConfigError({
+        step,
+        objectIdByNameSingular,
+        flatFieldMetadataMaps,
+      });
+
+      if (isDefined(loadBalanceError)) {
+        throw new WorkflowTriggerException(
+          loadBalanceError,
+          WorkflowTriggerExceptionCode.INVALID_WORKFLOW_VERSION,
+        );
+      }
+    }
+  }
+
  async deactivateWorkflowVersion(
    workflowVersionId: string,
    workspaceId: string,