feat(workflow): condition filter on database-event triggers (#21868)

## Problem Connecting a mailbox bulk-creates contacts via the email/calendar sync, and each `person.upserted` fires the seeded **"Create company when adding a new person"** workflow. The trigger enqueues one run per record (no batching) and each run bills several `WORKFLOW_NODE_RUN` events — so a single mailbox connect can rack up tens of thousands of runs and exhaust credits on a brand-new workspace. The workflow is also redundant on that path: the sync already creates the company from the email domain and links the person to it. ## What this does Adds an optional, user-defined **filter** to database-event (listener) triggers, evaluated in the listener **before a run is enqueued**. Non-matching events never create a run, so they consume zero execution credits. This is the Filter node's capability, lifted to the trigger level, and available for all event types (created / updated / upserted / deleted). The seeded "Create company when adding a new person" workflow now carries a visible trigger filter — `Created by → Source is not Email` **and** `is not Calendar` — so it no longer runs for sync-created contacts, while still running for manually / API / CSV-added people. ## How (reuse) - **Backend:** extracted `evaluateStepFilters()`, shared by the Filter action and the trigger listener's new `eventMatchesRecordFilter` gate. The record is exposed under the `trigger` key so filters reference it exactly like steps do (`{{trigger.properties.after.…}}`). - **Shared:** one optional `filter` added to the database-event trigger zod schema; the front-end type derives from it (settings stay JSON — no codegen). - **Frontend:** extracted `WorkflowStepFilterBuilder` from the Filter action's body; both the Filter action and the trigger editor render it. The field picker needed no changes — at the trigger it already resolves to the record's own fields via `TRIGGER_STEP_ID`. ## Scope / decisions - **No migration for existing workspaces** (by request) — only newly created workspaces get the filtered default; already-created workspaces keep the always-on workflow. - Deliberately did **not** add relation-enrichment to the upsert path (it would add a DB lookup to the very bulk-sync path we're relieving). Trigger filters work on the record's own scalar/composite fields (e.g. `createdBy.source`); relation-based filters work on created/updated where enrichment already runs. ## Verification - Typecheck: `twenty-shared`, `twenty-server`, `twenty-front` all green. - Lint (diff, autofix): 0 warnings / 0 errors across all three. - Unit tests: a new `evaluate-step-filters` spec exercising the exact `createdBy.source IS_NOT` seed mechanism, plus new listener specs proving non-matching events are not enqueued. All backend filter/listener suites pass. - Not run here: integration tests (need a DB) and Storybook. https://claude.ai/code/session_013k36vfekDppwRCgM6Ha7De --- _Generated by [Claude Code](https://claude.ai/code/session_013k36vfekDppwRCgM6Ha7De)_  <a href="https://cubic.dev/pr/twentyhq/twenty/pull/21868?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>  --------- Co-authored-by: Claude <noreply@anthropic.com>
2026-06-21 17:47:06 +02:00
parent f8db73598c
commit a0689d1577
19 changed files with 844 additions and 264 deletions
@@ -1,7 +1,5 @@
 import { Injectable } from '@nestjs/common';

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

 import {
@@ -12,7 +10,7 @@ import { type WorkflowActionInput } from 'src/modules/workflow/workflow-executor
 import { type WorkflowActionOutput } from 'src/modules/workflow/workflow-executor/types/workflow-action-output.type';
 import { findStepOrThrow } from 'src/modules/workflow/workflow-executor/utils/find-step-or-throw.util';
 import { isWorkflowFilterAction } from 'src/modules/workflow/workflow-executor/workflow-actions/filter/guards/is-workflow-filter-action.guard';
-import { evaluateFilterConditions } from 'src/modules/workflow/workflow-executor/workflow-actions/filter/utils/evaluate-filter-conditions.util';
+import { evaluateStepFilters } from 'src/modules/workflow/workflow-executor/workflow-actions/filter/utils/evaluate-step-filters.util';

@Injectable()
 export class FilterWorkflowAction implements WorkflowAction {
@@ -41,15 +39,10 @@ export class FilterWorkflowAction implements WorkflowAction {
      };
    }

-    const resolvedFilters = stepFilters.map((filter) => ({
-      ...filter,
-      rightOperand: resolveInput(filter.value, context),
-      leftOperand: resolveInput(filter.stepOutputKey, context),
-    }));
-
-    const matchesFilter = evaluateFilterConditions({
-      filterGroups: stepFilterGroups,
-      filters: resolvedFilters,
+    const matchesFilter = evaluateStepFilters({
+      stepFilters,
+      stepFilterGroups,
+      context,
    });

    return {
@@ -0,0 +1,143 @@
+import {
+  type StepFilter,
+  type StepFilterGroup,
+  StepLogicalOperator,
+  ViewFilterOperand,
+} from 'twenty-shared/types';
+
+import { evaluateStepFilters } from 'src/modules/workflow/workflow-executor/workflow-actions/filter/utils/evaluate-step-filters.util';
+
+describe('evaluateStepFilters', () => {
+  const context = {
+    trigger: {
+      properties: {
+        after: {
+          createdBy: { source: 'EMAIL' },
+          name: 'Acme',
+        },
+      },
+    },
+  };
+
+  const group: StepFilterGroup = {
+    id: 'group-1',
+    logicalOperator: StepLogicalOperator.AND,
+  };
+
+  const sourceFilter = (operand: ViewFilterOperand): StepFilter => ({
+    id: 'filter-1',
+    type: 'ACTOR',
+    operand,
+    value: JSON.stringify(['EMAIL']),
+    stepOutputKey: '{{trigger.properties.after.createdBy.source}}',
+    stepFilterGroupId: group.id,
+    compositeFieldSubFieldName: 'source',
+  });
+
+  it('returns true when there are no filters', () => {
+    expect(
+      evaluateStepFilters({
+        stepFilters: [],
+        stepFilterGroups: [],
+        context,
+      }),
+    ).toBe(true);
+  });
+
+  it('resolves operands from the context and matches the record', () => {
+    expect(
+      evaluateStepFilters({
+        stepFilterGroups: [group],
+        stepFilters: [sourceFilter(ViewFilterOperand.IS)],
+        context,
+      }),
+    ).toBe(true);
+  });
+
+  it('returns false when the record source is excluded (IS_NOT)', () => {
+    expect(
+      evaluateStepFilters({
+        stepFilterGroups: [group],
+        stepFilters: [sourceFilter(ViewFilterOperand.IS_NOT)],
+        context,
+      }),
+    ).toBe(false);
+  });
+
+  it('returns true when a different source is excluded (IS_NOT)', () => {
+    const calendarFilter: StepFilter = {
+      ...sourceFilter(ViewFilterOperand.IS_NOT),
+      value: JSON.stringify(['CALENDAR']),
+    };
+
+    expect(
+      evaluateStepFilters({
+        stepFilterGroups: [group],
+        stepFilters: [calendarFilter],
+        context,
+      }),
+    ).toBe(true);
+  });
+
+  it('evaluates IS_NOT_EMPTY against a present field when no value is set', () => {
+    const filter: StepFilter = {
+      id: 'filter-present',
+      type: 'TEXT',
+      operand: ViewFilterOperand.IS_NOT_EMPTY,
+      value: '',
+      stepOutputKey: '{{trigger.properties.after.name}}',
+      stepFilterGroupId: group.id,
+    };
+
+    expect(
+      evaluateStepFilters({
+        stepFilterGroups: [group],
+        stepFilters: [filter],
+        context,
+      }),
+    ).toBe(true);
+  });
+
+  it('resolves a missing field path to empty (IS_EMPTY is true)', () => {
+    const filter: StepFilter = {
+      id: 'filter-missing',
+      type: 'TEXT',
+      operand: ViewFilterOperand.IS_EMPTY,
+      value: '',
+      stepOutputKey: '{{trigger.properties.after.missingField}}',
+      stepFilterGroupId: group.id,
+    };
+
+    expect(
+      evaluateStepFilters({
+        stepFilterGroups: [group],
+        stepFilters: [filter],
+        context,
+      }),
+    ).toBe(true);
+  });
+
+  it('applies implicit AND across flat filters without groups', () => {
+    const nameContains: StepFilter = {
+      id: 'name-contains',
+      type: 'TEXT',
+      operand: ViewFilterOperand.CONTAINS,
+      value: 'Acme',
+      stepOutputKey: '{{trigger.properties.after.name}}',
+      stepFilterGroupId: 'unused',
+    };
+    const sourceIsCalendar: StepFilter = {
+      ...sourceFilter(ViewFilterOperand.IS),
+      value: JSON.stringify(['CALENDAR']),
+      stepFilterGroupId: 'unused',
+    };
+
+    expect(
+      evaluateStepFilters({
+        stepFilterGroups: [],
+        stepFilters: [nameContains, sourceIsCalendar],
+        context,
+      }),
+    ).toBe(false);
+  });
+});
@@ -0,0 +1,25 @@
+import { type StepFilter, type StepFilterGroup } from 'twenty-shared/types';
+import { resolveInput } from 'twenty-shared/utils';
+
+import { evaluateFilterConditions } from 'src/modules/workflow/workflow-executor/workflow-actions/filter/utils/evaluate-filter-conditions.util';
+
+export const evaluateStepFilters = ({
+  stepFilters,
+  stepFilterGroups,
+  context,
+}: {
+  stepFilters: StepFilter[];
+  stepFilterGroups: StepFilterGroup[];
+  context: Record<string, unknown>;
+}): boolean => {
+  const resolvedFilters = stepFilters.map((filter) => ({
+    ...filter,
+    rightOperand: resolveInput(filter.value, context),
+    leftOperand: resolveInput(filter.stepOutputKey, context),
+  }));
+
+  return evaluateFilterConditions({
+    filterGroups: stepFilterGroups,
+    filters: resolvedFilters,
+  });
+};
@@ -1,5 +1,13 @@
+import { type StepFilter, type StepFilterGroup } from 'twenty-shared/types';
+
+export type DatabaseEventTriggerFilterSettings = {
+  stepFilters: StepFilter[];
+  stepFilterGroups: StepFilterGroup[];
+};
+
 export type BaseDatabaseEventTriggerSettings = {
  eventName: string;
+  filter?: DatabaseEventTriggerFilterSettings;
 };

 export type DatabaseEventTriggerSettings =
@@ -9,7 +9,8 @@ import {
  type ObjectRecordUpsertEvent,
 } from 'twenty-shared/database-events';
 import { type ObjectRecord } from 'twenty-shared/types';
-import { isDefined } from 'twenty-shared/utils';
+import { isDefined, isNonEmptyArray } from 'twenty-shared/utils';
+import { TRIGGER_STEP_ID } from 'twenty-shared/workflow';
 import { In, Raw } from 'typeorm';

 import { OnDatabaseBatchEvent } from 'src/engine/api/graphql/graphql-query-runner/decorators/on-database-batch-event.decorator';
@@ -31,15 +32,22 @@ import {
  type WorkflowAutomatedTriggerWorkspaceEntity,
 } from 'src/modules/workflow/common/standard-objects/workflow-automated-trigger.workspace-entity';
 import { WorkflowCommonWorkspaceService } from 'src/modules/workflow/common/workspace-services/workflow-common.workspace-service';
+import { evaluateStepFilters } from 'src/modules/workflow/workflow-executor/workflow-actions/filter/utils/evaluate-step-filters.util';
 import {
+  type BaseDatabaseEventTriggerSettings,
  type UpdateEventTriggerSettings,
-  type UpsertEventTriggerSettings,
 } from 'src/modules/workflow/workflow-trigger/automated-trigger/constants/automated-trigger-settings';
 import {
  WorkflowTriggerJob,
  type WorkflowTriggerJobData,
 } from 'src/modules/workflow/workflow-trigger/jobs/workflow-trigger.job';

+type TriggerEvaluationArgs = {
+  eventPayload: ObjectRecordEvent;
+  eventListener: WorkflowAutomatedTriggerWorkspaceEntity;
+  action: DatabaseEventAction;
+};
+
@Injectable()
 export class WorkflowDatabaseEventTriggerListener {
  private readonly logger = new Logger(
@@ -386,27 +394,26 @@ export class WorkflowDatabaseEventTriggerListener {
    eventPayload,
    eventListener,
    action,
-  }: {
-    eventPayload: ObjectRecordEvent;
-    eventListener: WorkflowAutomatedTriggerWorkspaceEntity;
-    action: DatabaseEventAction;
-  }) {
-    if (action === DatabaseEventAction.UPDATED) {
+  }: TriggerEvaluationArgs) {
+    return (
+      this.eventMatchesWatchedFields({ eventPayload, eventListener, action }) &&
+      this.eventMatchesRecordFilter({ eventPayload, eventListener })
+    );
+  }
+
+  private eventMatchesWatchedFields({
+    eventPayload,
+    eventListener,
+    action,
+  }: TriggerEvaluationArgs) {
+    if (
+      action === DatabaseEventAction.UPDATED ||
+      action === DatabaseEventAction.UPSERTED
+    ) {
      const settings = eventListener.settings as UpdateEventTriggerSettings;
-      const updateEventPayload = eventPayload as ObjectRecordUpdateEvent;
-      const updatedFields = updateEventPayload?.properties?.updatedFields ?? [];
-
-      return (
-        !settings.fields ||
-        settings.fields.length === 0 ||
-        settings.fields.some((field) => updatedFields.includes(field))
-      );
-    }
-
-    if (action === DatabaseEventAction.UPSERTED) {
-      const settings = eventListener.settings as UpsertEventTriggerSettings;
-      const upsertEventPayload = eventPayload as ObjectRecordUpsertEvent;
-      const updatedFields = upsertEventPayload?.properties?.updatedFields ?? [];
+      const updatedFields =
+        (eventPayload as ObjectRecordUpdateEvent)?.properties?.updatedFields ??
+        [];

      return (
        !settings.fields ||
@@ -417,4 +424,32 @@ export class WorkflowDatabaseEventTriggerListener {

    return true;
  }
+
+  private eventMatchesRecordFilter({
+    eventPayload,
+    eventListener,
+  }: Pick<TriggerEvaluationArgs, 'eventPayload' | 'eventListener'>) {
+    const { filter } =
+      eventListener.settings as BaseDatabaseEventTriggerSettings;
+
+    if (!isDefined(filter) || !isNonEmptyArray(filter.stepFilters)) {
+      return true;
+    }
+
+    try {
+      return evaluateStepFilters({
+        stepFilters: filter.stepFilters,
+        stepFilterGroups: filter.stepFilterGroups,
+        context: { [TRIGGER_STEP_ID]: eventPayload },
+      });
+    } catch (error) {
+      this.logger.error(
+        `Failed to evaluate database-event trigger filter for workflow ${eventListener.workflowId}: ${
+          error instanceof Error ? error.message : String(error)
+        }`,
+      );
+
+      return false;
+    }
+  }
 }