From 96a24563674313a3071d359bfccaf33d5e130ab8 Mon Sep 17 00:00:00 2001 From: Thomas Trompette Date: Thu, 23 Jul 2026 14:30:12 +0200 Subject: [PATCH] feat(workflow): periodic core-consistency check for workflows, versions and triggers (#23103) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- .../commands/cron-register-all.command.ts | 6 + .../commands/database-command.module.ts | 2 + .../metrics/types/metrics-keys.type.ts | 3 + .../workflow-core-consistency-cron.command.ts | 35 +++ .../workflow-core-consistency-cron.job.ts | 31 +++ .../workflow-core-consistency.service.spec.ts | 223 ++++++++++++++++ .../workflow-core-consistency.service.ts | 251 ++++++++++++++++++ .../workflow-core-consistency.module.ts | 18 ++ .../workflow-trigger.module.ts | 2 + 9 files changed, 571 insertions(+) create mode 100644 packages/twenty-server/src/modules/workflow/workflow-core-consistency/crons/commands/workflow-core-consistency-cron.command.ts create mode 100644 packages/twenty-server/src/modules/workflow/workflow-core-consistency/crons/jobs/workflow-core-consistency-cron.job.ts create mode 100644 packages/twenty-server/src/modules/workflow/workflow-core-consistency/services/__tests__/workflow-core-consistency.service.spec.ts create mode 100644 packages/twenty-server/src/modules/workflow/workflow-core-consistency/services/workflow-core-consistency.service.ts create mode 100644 packages/twenty-server/src/modules/workflow/workflow-core-consistency/workflow-core-consistency.module.ts diff --git a/packages/twenty-server/src/database/commands/cron-register-all.command.ts b/packages/twenty-server/src/database/commands/cron-register-all.command.ts index cb03764c5a..ef2d777d0c 100644 --- a/packages/twenty-server/src/database/commands/cron-register-all.command.ts +++ b/packages/twenty-server/src/database/commands/cron-register-all.command.ts @@ -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, diff --git a/packages/twenty-server/src/database/commands/database-command.module.ts b/packages/twenty-server/src/database/commands/database-command.module.ts index 86301c4666..e66104a6af 100644 --- a/packages/twenty-server/src/database/commands/database-command.module.ts +++ b/packages/twenty-server/src/database/commands/database-command.module.ts @@ -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, diff --git a/packages/twenty-server/src/engine/core-modules/metrics/types/metrics-keys.type.ts b/packages/twenty-server/src/engine/core-modules/metrics/types/metrics-keys.type.ts index 2cc220098c..c9e766a5f7 100644 --- a/packages/twenty-server/src/engine/core-modules/metrics/types/metrics-keys.type.ts +++ b/packages/twenty-server/src/engine/core-modules/metrics/types/metrics-keys.type.ts @@ -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', diff --git a/packages/twenty-server/src/modules/workflow/workflow-core-consistency/crons/commands/workflow-core-consistency-cron.command.ts b/packages/twenty-server/src/modules/workflow/workflow-core-consistency/crons/commands/workflow-core-consistency-cron.command.ts new file mode 100644 index 0000000000..b8fb9cbcbb --- /dev/null +++ b/packages/twenty-server/src/modules/workflow/workflow-core-consistency/crons/commands/workflow-core-consistency-cron.command.ts @@ -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 { + await this.messageQueueService.addCron({ + jobName: WorkflowCoreConsistencyCronJob.name, + data: undefined, + options: { + repeat: { + pattern: WORKFLOW_CORE_CONSISTENCY_CRON_PATTERN, + }, + }, + }); + } +} diff --git a/packages/twenty-server/src/modules/workflow/workflow-core-consistency/crons/jobs/workflow-core-consistency-cron.job.ts b/packages/twenty-server/src/modules/workflow/workflow-core-consistency/crons/jobs/workflow-core-consistency-cron.job.ts new file mode 100644 index 0000000000..ef0ee78bb2 --- /dev/null +++ b/packages/twenty-server/src/modules/workflow/workflow-core-consistency/crons/jobs/workflow-core-consistency-cron.job.ts @@ -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 { + this.logger.log('WorkflowCoreConsistencyCronJob started'); + + await this.workflowCoreConsistencyService.runConsistencyCheck(); + + this.logger.log('WorkflowCoreConsistencyCronJob completed'); + } +} diff --git a/packages/twenty-server/src/modules/workflow/workflow-core-consistency/services/__tests__/workflow-core-consistency.service.spec.ts b/packages/twenty-server/src/modules/workflow/workflow-core-consistency/services/__tests__/workflow-core-consistency.service.spec.ts new file mode 100644 index 0000000000..70f69d3005 --- /dev/null +++ b/packages/twenty-server/src/modules/workflow/workflow-core-consistency/services/__tests__/workflow-core-consistency.service.spec.ts @@ -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; + + 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 } }, + ); + }); +}); diff --git a/packages/twenty-server/src/modules/workflow/workflow-core-consistency/services/workflow-core-consistency.service.ts b/packages/twenty-server/src/modules/workflow/workflow-core-consistency/services/workflow-core-consistency.service.ts new file mode 100644 index 0000000000..3a2e572d18 --- /dev/null +++ b/packages/twenty-server/src/modules/workflow/workflow-core-consistency/services/workflow-core-consistency.service.ts @@ -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; + +// 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 { + // 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 { + 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 { + 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 { + 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 { + 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}`, + ); + } + } + } +} diff --git a/packages/twenty-server/src/modules/workflow/workflow-core-consistency/workflow-core-consistency.module.ts b/packages/twenty-server/src/modules/workflow/workflow-core-consistency/workflow-core-consistency.module.ts new file mode 100644 index 0000000000..9430c04018 --- /dev/null +++ b/packages/twenty-server/src/modules/workflow/workflow-core-consistency/workflow-core-consistency.module.ts @@ -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 {} diff --git a/packages/twenty-server/src/modules/workflow/workflow-trigger/workflow-trigger.module.ts b/packages/twenty-server/src/modules/workflow/workflow-trigger/workflow-trigger.module.ts index 2527555198..7a517f395b 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-trigger/workflow-trigger.module.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-trigger/workflow-trigger.module.ts @@ -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,