feat(workflow): periodic core-consistency check for workflows, versions and triggers (#23103)

Monitoring for the soft-ref migration. The `workflow` / `workflowVersion` dual-write into core is **best-effort** (async, not transactional; failures only go to Sentry), so `core.workflow` / `core.workflowVersion` can silently drift from the workspace source of truth. This adds a periodic job that detects that drift across **all three workflow entities** and emits it as metrics. Supersedes the earlier inline shadow-parity approach that lived on this branch — that only covered trigger dispatch and added a cache read + diff to every cron tick and every DB-event batch (too much hot-path overhead). This is broader and fully off the dispatch path. ## What A cron (`cron:workflow:core-consistency-check`, every 3 hours, wired into `cron:register:all`). Per run: - **Bounded**: `SELECT DISTINCT "workspaceId" FROM core."workflow"` — only workspaces that actually use workflows (skips the large majority). - Per such workspace, emit a drift metric per `(entity, driftType)` — detect-only, plus a triage log: - **workflow** and **workflowVersion** — `unlinked` / `missingCore` / `orphanCore` / `fieldMismatch`, via cross-schema `COUNT` aggregates (core + workspace are the same DB, so indexed joins — no rows pulled into JS). - **automated triggers** — the `workflowAutomatedTrigger` table vs the `workflowAutomatedTriggerMaps` cache: `inTableNotCache` / `inCacheNotTable` / settings `mismatch`. - Per-workspace failures are isolated (caught → Sentry) so one bad workspace does not stop the sweep. ## Why it is efficient Two central `core.*` queries + a few `COUNT` queries per *workflow-using* workspace, on a relaxed cadence, off the dispatch path. Shardable across ticks later if needed. ## Metrics `workflow-core-consistency/{workflow,version,automated-trigger}/drift` counters, attribute `driftType`. Dashboards: twentyhq/twenty-infra#800. ## Not in scope Detect-only — no auto-heal (the existing backfill/rebuild command can heal). No dispatch or flag changes. ## Test - The consistency SQL (workflow/version sync counts, orphan counts, trigger read) validated against a live workspace (it surfaced real drift there — unlinked versions + an orphan core version). The whole cron→service→SQL→metric pipeline is proven live: the cron is already emitting real drift counters on a running server. - Unit specs for the service (clean → no metric; per-entity drift per dimension; per-workspace error isolation). - Command boots and registers via `cron:register:all` (verified). Typecheck + lint clean.
2026-07-23 14:30:12 +02:00
parent 66df0ac47c
commit 96a2456367
9 changed files with 571 additions and 0 deletions
@@ -27,6 +27,7 @@ import { MessagingMessageListFetchCronCommand } from 'src/modules/messaging/mess
 import { MessagingMessagesImportCronCommand } from 'src/modules/messaging/message-import-manager/crons/commands/messaging-messages-import.cron.command';
 import { MessagingOngoingStaleCronCommand } from 'src/modules/messaging/message-import-manager/crons/commands/messaging-ongoing-stale.cron.command';
 import { MessagingRelaunchFailedMessageChannelsCronCommand } from 'src/modules/messaging/message-import-manager/crons/commands/messaging-relaunch-failed-message-channels.cron.command';
+import { WorkflowCoreConsistencyCronCommand } from 'src/modules/workflow/workflow-core-consistency/crons/commands/workflow-core-consistency-cron.command';
 import { WorkflowCleanWorkflowRunsCronCommand } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/workflow-clean-workflow-runs.cron.command';
 import { WorkflowHandleStaledRunsCronCommand } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/workflow-handle-staled-runs.cron.command';
 import { WorkflowRunEnqueueCronCommand } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/workflow-run-enqueue.cron.command';
@@ -56,6 +57,7 @@ export class CronRegisterAllCommand extends CommandRunner {
    private readonly workflowRunEnqueueCronCommand: WorkflowRunEnqueueCronCommand,
    private readonly workflowHandleStaledRunsCronCommand: WorkflowHandleStaledRunsCronCommand,
    private readonly workflowCleanWorkflowRunsCronCommand: WorkflowCleanWorkflowRunsCronCommand,
+    private readonly workflowCoreConsistencyCronCommand: WorkflowCoreConsistencyCronCommand,

    private readonly checkCustomDomainValidRecordsCronCommand: CheckCustomDomainValidRecordsCronCommand,
    private readonly checkPublicDomainsValidRecordsCronCommand: CheckPublicDomainsValidRecordsCronCommand,
@@ -150,6 +152,10 @@ export class CronRegisterAllCommand extends CommandRunner {
        name: 'WorkflowCleanWorkflowRuns',
        command: this.workflowCleanWorkflowRunsCronCommand,
      },
+      {
+        name: 'WorkflowCoreConsistency',
+        command: this.workflowCoreConsistencyCronCommand,
+      },
      {
        name: 'CronTrigger',
        command: this.cronTriggerCronCommand,
@@ -53,6 +53,7 @@ import { CalendarEventImportManagerModule } from 'src/modules/calendar/calendar-
 import { MessagingImportManagerModule } from 'src/modules/messaging/message-import-manager/messaging-import-manager.module';
 import { WorkflowRunQueueModule } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workflow-run-queue.module';
 import { AutomatedTriggerModule } from 'src/modules/workflow/workflow-trigger/automated-trigger/automated-trigger.module';
+import { WorkflowCoreConsistencyModule } from 'src/modules/workflow/workflow-core-consistency/workflow-core-consistency.module';

@Module({
  imports: [
@@ -64,6 +65,7 @@ import { AutomatedTriggerModule } from 'src/modules/workflow/workflow-trigger/au
    CalendarEventImportManagerModule,
    WebhookSubscriptionModule,
    AutomatedTriggerModule,
+    WorkflowCoreConsistencyModule,
    FileModule,
    WorkspaceModule,
    WorkflowRunQueueModule,
@@ -25,6 +25,9 @@ export enum MetricsKeys {
  WorkflowRunSystemError = 'workflow-run/system-error',
  WorkflowRunStuckRunningDetected = 'workflow-run/stuck-running/detected',
  WorkflowRunStuckRunningFalsePositive = 'workflow-run/stuck-running/false-positive',
+  WorkflowCoreConsistencyWorkflowDrift = 'workflow-core-consistency/workflow/drift',
+  WorkflowCoreConsistencyVersionDrift = 'workflow-core-consistency/version/drift',
+  WorkflowCoreConsistencyAutomatedTriggerDrift = 'workflow-core-consistency/automated-trigger/drift',
  AiChatToolExecutionSucceeded = 'ai-chat/tool-execution-succeeded',
  AiChatToolExecutionFailed = 'ai-chat/tool-execution-failed',
  AiChatToolExecutionDurationMs = 'ai-chat/tool-execution-duration-ms',
@@ -0,0 +1,35 @@
+import { Command, CommandRunner } from 'nest-commander';
+
+import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator';
+import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
+import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service';
+import {
+  WORKFLOW_CORE_CONSISTENCY_CRON_PATTERN,
+  WorkflowCoreConsistencyCronJob,
+} from 'src/modules/workflow/workflow-core-consistency/crons/jobs/workflow-core-consistency-cron.job';
+
+@Command({
+  name: 'cron:workflow:core-consistency-check',
+  description:
+    'Starts a cron job that checks workflow, version and automated-trigger consistency between the workspace and core',
+})
+export class WorkflowCoreConsistencyCronCommand extends CommandRunner {
+  constructor(
+    @InjectMessageQueue(MessageQueue.cronQueue)
+    private readonly messageQueueService: MessageQueueService,
+  ) {
+    super();
+  }
+
+  async run(): Promise<void> {
+    await this.messageQueueService.addCron<undefined>({
+      jobName: WorkflowCoreConsistencyCronJob.name,
+      data: undefined,
+      options: {
+        repeat: {
+          pattern: WORKFLOW_CORE_CONSISTENCY_CRON_PATTERN,
+        },
+      },
+    });
+  }
+}
@@ -0,0 +1,31 @@
+import { Logger } from '@nestjs/common';
+
+import { SentryCronMonitor } from 'src/engine/core-modules/cron/sentry-cron-monitor.decorator';
+import { Process } from 'src/engine/core-modules/message-queue/decorators/process.decorator';
+import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.decorator';
+import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
+import { WorkflowCoreConsistencyService } from 'src/modules/workflow/workflow-core-consistency/services/workflow-core-consistency.service';
+
+export const WORKFLOW_CORE_CONSISTENCY_CRON_PATTERN = '0 */3 * * *';
+
+@Processor(MessageQueue.cronQueue)
+export class WorkflowCoreConsistencyCronJob {
+  private readonly logger = new Logger(WorkflowCoreConsistencyCronJob.name);
+
+  constructor(
+    private readonly workflowCoreConsistencyService: WorkflowCoreConsistencyService,
+  ) {}
+
+  @Process(WorkflowCoreConsistencyCronJob.name)
+  @SentryCronMonitor(
+    WorkflowCoreConsistencyCronJob.name,
+    WORKFLOW_CORE_CONSISTENCY_CRON_PATTERN,
+  )
+  async handle(): Promise<void> {
+    this.logger.log('WorkflowCoreConsistencyCronJob started');
+
+    await this.workflowCoreConsistencyService.runConsistencyCheck();
+
+    this.logger.log('WorkflowCoreConsistencyCronJob completed');
+  }
+}
@@ -0,0 +1,223 @@
+import { type DataSource } from 'typeorm';
+
+import { type ExceptionHandlerService } from 'src/engine/core-modules/exception-handler/exception-handler.service';
+import { type MetricsService } from 'src/engine/core-modules/metrics/metrics.service';
+import { MetricsKeys } from 'src/engine/core-modules/metrics/types/metrics-keys.type';
+import { type WorkspaceCacheService } from 'src/engine/workspace-cache/services/workspace-cache.service';
+import { AutomatedTriggerType } from 'src/modules/workflow/common/standard-objects/workflow-automated-trigger.workspace-entity';
+import { WorkflowCoreConsistencyService } from 'src/modules/workflow/workflow-core-consistency/services/workflow-core-consistency.service';
+
+const workspaceId = '20202020-1c25-4d02-bf25-6aeccf7ea419';
+const schema = 'workspace_test';
+
+type Counts = { unlinked: number; missingCore: number; fieldMismatch: number };
+
+describe('WorkflowCoreConsistencyService', () => {
+  let service: WorkflowCoreConsistencyService;
+  let coreDataSource: { query: jest.Mock };
+  let workspaceCacheService: { getOrRecompute: jest.Mock };
+  let metricsService: { incrementCounterBy: jest.Mock };
+  let exceptionHandlerService: { captureExceptions: jest.Mock };
+
+  let workflowCounts: Counts;
+  let workflowOrphan: number;
+  let versionCounts: Counts;
+  let versionOrphan: number;
+  let triggerRows: Array<{
+    workflowId: string;
+    type: string;
+    settings: unknown;
+  }>;
+  let cacheByWorkflowId: Record<string, unknown>;
+
+  const zero = (): Counts => ({
+    unlinked: 0,
+    missingCore: 0,
+    fieldMismatch: 0,
+  });
+
+  beforeEach(() => {
+    workflowCounts = zero();
+    workflowOrphan = 0;
+    versionCounts = zero();
+    versionOrphan = 0;
+    triggerRows = [];
+    cacheByWorkflowId = {};
+
+    coreDataSource = {
+      query: jest.fn().mockImplementation((sql: string) => {
+        if (sql.includes('DISTINCT "workspaceId"')) {
+          return Promise.resolve([{ workspaceId }]);
+        }
+        if (sql.includes('"databaseSchema"')) {
+          return Promise.resolve([{ databaseSchema: schema }]);
+        }
+        if (sql.includes('AS "orphanCore"')) {
+          return Promise.resolve([
+            {
+              orphanCore: sql.includes('workflowVersion')
+                ? versionOrphan
+                : workflowOrphan,
+            },
+          ]);
+        }
+        if (sql.includes('AS unlinked')) {
+          return Promise.resolve([
+            sql.includes('workflowVersion') ? versionCounts : workflowCounts,
+          ]);
+        }
+        if (sql.includes('workflowAutomatedTrigger')) {
+          return Promise.resolve(triggerRows);
+        }
+
+        return Promise.resolve([]);
+      }),
+    };
+    workspaceCacheService = {
+      getOrRecompute: jest.fn().mockImplementation(() =>
+        Promise.resolve({
+          workflowAutomatedTriggerMaps: { byWorkflowId: cacheByWorkflowId },
+        }),
+      ),
+    };
+    metricsService = { incrementCounterBy: jest.fn() };
+    exceptionHandlerService = { captureExceptions: jest.fn() };
+
+    service = new WorkflowCoreConsistencyService(
+      coreDataSource as unknown as DataSource,
+      workspaceCacheService as unknown as WorkspaceCacheService,
+      metricsService as unknown as MetricsService,
+      exceptionHandlerService as unknown as ExceptionHandlerService,
+    );
+
+    jest.spyOn(service['logger'], 'warn').mockImplementation();
+  });
+
+  afterEach(() => {
+    jest.clearAllMocks();
+  });
+
+  it('emits nothing for a fully consistent workspace', async () => {
+    await service.runConsistencyCheck();
+
+    expect(metricsService.incrementCounterBy).not.toHaveBeenCalled();
+    expect(exceptionHandlerService.captureExceptions).not.toHaveBeenCalled();
+  });
+
+  it('emits a drift count per dimension for workflow drift', async () => {
+    workflowCounts = { unlinked: 2, missingCore: 0, fieldMismatch: 1 };
+
+    await service.runConsistencyCheck();
+
+    expect(metricsService.incrementCounterBy).toHaveBeenCalledWith({
+      key: MetricsKeys.WorkflowCoreConsistencyWorkflowDrift,
+      amount: 2,
+      attributes: { driftType: 'unlinked' },
+    });
+    expect(metricsService.incrementCounterBy).toHaveBeenCalledWith({
+      key: MetricsKeys.WorkflowCoreConsistencyWorkflowDrift,
+      amount: 1,
+      attributes: { driftType: 'fieldMismatch' },
+    });
+  });
+
+  it('emits version drift including orphan core rows', async () => {
+    versionCounts = { unlinked: 0, missingCore: 3, fieldMismatch: 0 };
+    versionOrphan = 1;
+
+    await service.runConsistencyCheck();
+
+    expect(metricsService.incrementCounterBy).toHaveBeenCalledWith({
+      key: MetricsKeys.WorkflowCoreConsistencyVersionDrift,
+      amount: 3,
+      attributes: { driftType: 'missingCore' },
+    });
+    expect(metricsService.incrementCounterBy).toHaveBeenCalledWith({
+      key: MetricsKeys.WorkflowCoreConsistencyVersionDrift,
+      amount: 1,
+      attributes: { driftType: 'orphanCore' },
+    });
+  });
+
+  it('flags an automated trigger present in the cache but not the table', async () => {
+    cacheByWorkflowId = {
+      wf1: {
+        workflowId: 'wf1',
+        workflowVersionId: 'v1',
+        type: AutomatedTriggerType.CRON,
+        settings: { pattern: '* * * * *' },
+      },
+    };
+    triggerRows = [];
+
+    await service.runConsistencyCheck();
+
+    expect(metricsService.incrementCounterBy).toHaveBeenCalledWith({
+      key: MetricsKeys.WorkflowCoreConsistencyAutomatedTriggerDrift,
+      amount: 1,
+      attributes: { driftType: 'inCacheNotTable' },
+    });
+  });
+
+  it('flags an automated trigger present in the table but not the cache', async () => {
+    triggerRows = [
+      {
+        workflowId: 'wf1',
+        type: AutomatedTriggerType.DATABASE_EVENT,
+        settings: { eventName: 'person.created' },
+      },
+    ];
+    cacheByWorkflowId = {};
+
+    await service.runConsistencyCheck();
+
+    expect(metricsService.incrementCounterBy).toHaveBeenCalledWith({
+      key: MetricsKeys.WorkflowCoreConsistencyAutomatedTriggerDrift,
+      amount: 1,
+      attributes: { driftType: 'inTableNotCache' },
+    });
+  });
+
+  it('flags an automated trigger whose settings differ between table and cache', async () => {
+    cacheByWorkflowId = {
+      wf1: {
+        workflowId: 'wf1',
+        workflowVersionId: 'v1',
+        type: AutomatedTriggerType.DATABASE_EVENT,
+        settings: { eventName: 'person.created' },
+      },
+    };
+    triggerRows = [
+      {
+        workflowId: 'wf1',
+        type: AutomatedTriggerType.DATABASE_EVENT,
+        settings: { eventName: 'company.created' },
+      },
+    ];
+
+    await service.runConsistencyCheck();
+
+    expect(metricsService.incrementCounterBy).toHaveBeenCalledWith({
+      key: MetricsKeys.WorkflowCoreConsistencyAutomatedTriggerDrift,
+      amount: 1,
+      attributes: { driftType: 'mismatch' },
+    });
+  });
+
+  it('isolates a per-workspace failure and reports it to Sentry', async () => {
+    coreDataSource.query.mockImplementation((sql: string) => {
+      if (sql.includes('DISTINCT "workspaceId"')) {
+        return Promise.resolve([{ workspaceId }]);
+      }
+
+      return Promise.reject(new Error('boom'));
+    });
+
+    await expect(service.runConsistencyCheck()).resolves.toBeUndefined();
+
+    expect(exceptionHandlerService.captureExceptions).toHaveBeenCalledWith(
+      [expect.any(Error)],
+      { workspace: { id: workspaceId } },
+    );
+  });
+});
@@ -0,0 +1,251 @@
+import { Injectable, Logger } from '@nestjs/common';
+import { InjectDataSource } from '@nestjs/typeorm';
+
+import { isDefined } from 'twenty-shared/utils';
+import { DataSource } from 'typeorm';
+
+import { ExceptionHandlerService } from 'src/engine/core-modules/exception-handler/exception-handler.service';
+import { MetricsService } from 'src/engine/core-modules/metrics/metrics.service';
+import { MetricsKeys } from 'src/engine/core-modules/metrics/types/metrics-keys.type';
+import { WorkspaceCacheService } from 'src/engine/workspace-cache/services/workspace-cache.service';
+import { AutomatedTriggerType } from 'src/modules/workflow/common/standard-objects/workflow-automated-trigger.workspace-entity';
+import {
+  type BaseDatabaseEventTriggerSettings,
+  type CronTriggerSettings,
+} from 'src/modules/workflow/workflow-trigger/automated-trigger/constants/automated-trigger-settings';
+
+type DriftCounts = Record<string, number>;
+
+// Detect drift between the workspace source-of-truth records and their core
+// mirror. The dual-write is best-effort (async, not transactional), so core can
+// silently fall out of sync; this quantifies that per workspace as metrics.
+@Injectable()
+export class WorkflowCoreConsistencyService {
+  private readonly logger = new Logger(WorkflowCoreConsistencyService.name);
+
+  constructor(
+    @InjectDataSource()
+    private readonly coreDataSource: DataSource,
+    private readonly workspaceCacheService: WorkspaceCacheService,
+    private readonly metricsService: MetricsService,
+    private readonly exceptionHandlerService: ExceptionHandlerService,
+  ) {}
+
+  async runConsistencyCheck(): Promise<void> {
+    // Only workspaces that actually use workflows; the soft-ref makes this a
+    // cheap central lookup that skips the vast majority of workspaces.
+    const workspaces: Array<{ workspaceId: string }> =
+      await this.coreDataSource.query(
+        `SELECT DISTINCT "workspaceId" FROM core."workflow"`,
+      );
+
+    for (const { workspaceId } of workspaces) {
+      try {
+        await this.checkWorkspace(workspaceId);
+      } catch (error) {
+        this.exceptionHandlerService.captureExceptions([error], {
+          workspace: { id: workspaceId },
+        });
+      }
+    }
+  }
+
+  private async checkWorkspace(workspaceId: string): Promise<void> {
+    const [workspace] = await this.coreDataSource.query(
+      `SELECT "databaseSchema" FROM core."workspace" WHERE id = $1`,
+      [workspaceId],
+    );
+
+    if (!isDefined(workspace?.databaseSchema)) {
+      return;
+    }
+
+    const schema: string = workspace.databaseSchema;
+
+    await this.checkWorkflowSync(workspaceId, schema);
+    await this.checkWorkflowVersionSync(workspaceId, schema);
+    await this.checkAutomatedTriggerSync(workspaceId, schema);
+  }
+
+  private async checkWorkflowSync(
+    workspaceId: string,
+    schema: string,
+  ): Promise<void> {
+    const [counts] = await this.coreDataSource.query(
+      `SELECT
+         count(*) FILTER (WHERE wf."coreWorkflowId" IS NULL)::int AS unlinked,
+         count(*) FILTER (WHERE wf."coreWorkflowId" IS NOT NULL AND c.id IS NULL)::int AS "missingCore",
+         count(*) FILTER (WHERE c.id IS NOT NULL AND (
+           wf.name IS DISTINCT FROM c.name
+           OR NULLIF(wf."lastPublishedVersionId", '') IS DISTINCT FROM c."lastPublishedVersionId"::text
+         ))::int AS "fieldMismatch"
+       FROM "${schema}"."workflow" wf
+       LEFT JOIN core."workflow" c
+         ON c.id = wf."coreWorkflowId" AND c."workspaceId" = $1
+       WHERE wf."deletedAt" IS NULL`,
+      [workspaceId],
+    );
+
+    const [{ orphanCore }] = await this.coreDataSource.query(
+      `SELECT count(*)::int AS "orphanCore"
+       FROM core."workflow" c
+       WHERE c."workspaceId" = $1
+         AND NOT EXISTS (
+           SELECT 1 FROM "${schema}"."workflow" wf WHERE wf."coreWorkflowId" = c.id
+         )`,
+      [workspaceId],
+    );
+
+    this.emitDrift(
+      MetricsKeys.WorkflowCoreConsistencyWorkflowDrift,
+      workspaceId,
+      'workflow',
+      {
+        unlinked: counts.unlinked,
+        missingCore: counts.missingCore,
+        fieldMismatch: counts.fieldMismatch,
+        orphanCore,
+      },
+    );
+  }
+
+  private async checkWorkflowVersionSync(
+    workspaceId: string,
+    schema: string,
+  ): Promise<void> {
+    const [counts] = await this.coreDataSource.query(
+      `SELECT
+         count(*) FILTER (WHERE wf."coreWorkflowVersionId" IS NULL)::int AS unlinked,
+         count(*) FILTER (WHERE wf."coreWorkflowVersionId" IS NOT NULL AND c.id IS NULL)::int AS "missingCore",
+         count(*) FILTER (WHERE c.id IS NOT NULL AND (
+           c.status::text IS DISTINCT FROM wf.status::text
+           OR c."workflowId" IS DISTINCT FROM wf."workflowId"
+           OR c.steps IS DISTINCT FROM wf.steps
+           OR c.triggers IS DISTINCT FROM (
+             CASE WHEN wf.trigger IS NULL THEN NULL ELSE jsonb_build_array(wf.trigger) END
+           )
+         ))::int AS "fieldMismatch"
+       FROM "${schema}"."workflowVersion" wf
+       LEFT JOIN core."workflowVersion" c
+         ON c.id = wf."coreWorkflowVersionId" AND c."workspaceId" = $1
+       WHERE wf."deletedAt" IS NULL`,
+      [workspaceId],
+    );
+
+    const [{ orphanCore }] = await this.coreDataSource.query(
+      `SELECT count(*)::int AS "orphanCore"
+       FROM core."workflowVersion" c
+       WHERE c."workspaceId" = $1
+         AND NOT EXISTS (
+           SELECT 1 FROM "${schema}"."workflowVersion" wf WHERE wf."coreWorkflowVersionId" = c.id
+         )`,
+      [workspaceId],
+    );
+
+    this.emitDrift(
+      MetricsKeys.WorkflowCoreConsistencyVersionDrift,
+      workspaceId,
+      'workflowVersion',
+      {
+        unlinked: counts.unlinked,
+        missingCore: counts.missingCore,
+        fieldMismatch: counts.fieldMismatch,
+        orphanCore,
+      },
+    );
+  }
+
+  private async checkAutomatedTriggerSync(
+    workspaceId: string,
+    schema: string,
+  ): Promise<void> {
+    const { workflowAutomatedTriggerMaps } =
+      await this.workspaceCacheService.getOrRecompute(workspaceId, [
+        'workflowAutomatedTriggerMaps',
+      ]);
+    const cacheByWorkflowId = workflowAutomatedTriggerMaps.byWorkflowId;
+
+    const tableRows: Array<{
+      workflowId: string;
+      type: AutomatedTriggerType;
+      settings: BaseDatabaseEventTriggerSettings | CronTriggerSettings | null;
+    }> = await this.coreDataSource.query(
+      `SELECT "workflowId", type, settings FROM "${schema}"."workflowAutomatedTrigger"`,
+    );
+
+    const tableByWorkflowId = new Map(
+      tableRows.map((row) => [row.workflowId, row]),
+    );
+
+    let inTableNotCache = 0;
+    let inCacheNotTable = 0;
+    let mismatch = 0;
+
+    for (const [workflowId, row] of tableByWorkflowId) {
+      const cached = cacheByWorkflowId[workflowId];
+
+      if (!isDefined(cached)) {
+        inTableNotCache++;
+        continue;
+      }
+
+      if (
+        cached.type !== row.type ||
+        this.triggerIdentity(row.type, row.settings) !==
+          this.triggerIdentity(cached.type, cached.settings)
+      ) {
+        mismatch++;
+      }
+    }
+
+    for (const workflowId of Object.keys(cacheByWorkflowId)) {
+      if (!tableByWorkflowId.has(workflowId)) {
+        inCacheNotTable++;
+      }
+    }
+
+    this.emitDrift(
+      MetricsKeys.WorkflowCoreConsistencyAutomatedTriggerDrift,
+      workspaceId,
+      'automatedTrigger',
+      { inTableNotCache, inCacheNotTable, mismatch },
+    );
+  }
+
+  private triggerIdentity(
+    type: AutomatedTriggerType,
+    settings: BaseDatabaseEventTriggerSettings | CronTriggerSettings | null,
+  ): string {
+    if (type === AutomatedTriggerType.CRON) {
+      return (settings as CronTriggerSettings | null)?.pattern ?? '';
+    }
+
+    if (type === AutomatedTriggerType.DATABASE_EVENT) {
+      return (
+        (settings as BaseDatabaseEventTriggerSettings | null)?.eventName ?? ''
+      );
+    }
+
+    return '';
+  }
+
+  private emitDrift(
+    key: MetricsKeys,
+    workspaceId: string,
+    entity: string,
+    counts: DriftCounts,
+  ): void {
+    for (const [driftType, count] of Object.entries(counts)) {
+      if (count > 0) {
+        this.metricsService.incrementCounterBy({
+          key,
+          amount: count,
+          attributes: { driftType },
+        });
+        this.logger.warn(
+          `Workflow core consistency drift: workspace=${workspaceId} entity=${entity} ${driftType}=${count}`,
+        );
+      }
+    }
+  }
+}
@@ -0,0 +1,18 @@
+import { Module } from '@nestjs/common';
+
+import { MetricsModule } from 'src/engine/core-modules/metrics/metrics.module';
+import { WorkspaceCacheModule } from 'src/engine/workspace-cache/workspace-cache.module';
+import { WorkflowCoreConsistencyCronCommand } from 'src/modules/workflow/workflow-core-consistency/crons/commands/workflow-core-consistency-cron.command';
+import { WorkflowCoreConsistencyCronJob } from 'src/modules/workflow/workflow-core-consistency/crons/jobs/workflow-core-consistency-cron.job';
+import { WorkflowCoreConsistencyService } from 'src/modules/workflow/workflow-core-consistency/services/workflow-core-consistency.service';
+
+@Module({
+  imports: [MetricsModule, WorkspaceCacheModule],
+  providers: [
+    WorkflowCoreConsistencyService,
+    WorkflowCoreConsistencyCronJob,
+    WorkflowCoreConsistencyCronCommand,
+  ],
+  exports: [WorkflowCoreConsistencyCronCommand],
+})
+export class WorkflowCoreConsistencyModule {}
@@ -6,6 +6,7 @@ import { CommandMenuItemModule } from 'src/engine/metadata-modules/command-menu-
 import { LogicFunctionModule } from 'src/engine/metadata-modules/logic-function/logic-function.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';
 import { WorkflowRunnerModule } from 'src/modules/workflow/workflow-runner/workflow-runner.module';
 import { AutomatedTriggerModule } from 'src/modules/workflow/workflow-trigger/automated-trigger/automated-trigger.module';
 import { WorkflowTriggerJob } from 'src/modules/workflow/workflow-trigger/jobs/workflow-trigger.job';
@@ -17,6 +18,7 @@ import { WorkflowTriggerWorkspaceService } from 'src/modules/workflow/workflow-t
    CodeStepBuildModule,
    WorkflowRunnerModule,
    AutomatedTriggerModule,
+    WorkflowCoreConsistencyModule,
    CacheStorageModule,
    CommandMenuItemModule,
    FeatureFlagModule,