From d3c8eee07e76edaa1a744d92ba5fcf9dab417f87 Mon Sep 17 00:00:00 2001 From: Charlie Kolb Date: Wed, 16 Sep 2026 07:34:49 +0000 Subject: [PATCH] fix(core): Include sub-workflow tool calls in the workflow dependency index (#38062) Co-authored-by: Claude Fable 5 --- packages/@n8n/db/src/entities/index.ts | 6 +- .../entities/workflow-dependency-entity.ts | 7 ++ .../workflow-dependency.repository.ts | 44 +++++-- .../src/repositories/workflow.repository.ts | 28 +++-- .../__tests__/workflow-index.service.test.ts | 108 ++++++++++++++++++ .../workflow-index/workflow-index.service.ts | 17 ++- .../workflow-dependency.repository.test.ts | 101 +++++++++++++++- .../repositories/workflow.repository.test.ts | 108 +++++++++++++++++- .../workflows/workflow-index.test.ts | 2 +- 9 files changed, 392 insertions(+), 29 deletions(-) diff --git a/packages/@n8n/db/src/entities/index.ts b/packages/@n8n/db/src/entities/index.ts index 1d3c0bcfe5a..efdd72a5073 100644 --- a/packages/@n8n/db/src/entities/index.ts +++ b/packages/@n8n/db/src/entities/index.ts @@ -70,7 +70,10 @@ import { TestRun } from './test-run.ee'; import { User } from './user'; import { Variables } from './variables'; import { WebhookEntity } from './webhook-entity'; -import { WorkflowDependency } from './workflow-dependency-entity'; +import { + WorkflowDependency, + WORKFLOW_DEPENDENCY_INDEX_VERSION, +} from './workflow-dependency-entity'; import { WorkflowEntity } from './workflow-entity'; import { WorkflowHistory } from './workflow-history'; import { @@ -160,6 +163,7 @@ export { TagEntity, User, WorkflowDependency, + WORKFLOW_DEPENDENCY_INDEX_VERSION, WorkflowEntity, WorkflowStatistics, WorkflowTagMapping, diff --git a/packages/@n8n/db/src/entities/workflow-dependency-entity.ts b/packages/@n8n/db/src/entities/workflow-dependency-entity.ts index 66bdb05251c..7ca001a6b56 100644 --- a/packages/@n8n/db/src/entities/workflow-dependency-entity.ts +++ b/packages/@n8n/db/src/entities/workflow-dependency-entity.ts @@ -11,6 +11,13 @@ import { import { WithCreatedAt } from './abstract-entity'; import type { WorkflowEntity } from './workflow-entity'; +// Version of the index structure. When a change to the indexer alters what it +// extracts, bump this version. The staleness scan behind the rebuild on server +// start then reindexes every workflow whose rows carry an older version. +// Version 2: the indexer records `workflowCall` rows for sub-workflow tool and +// retriever nodes, not only for the Execute Sub-workflow node. +export const WORKFLOW_DEPENDENCY_INDEX_VERSION = 2; + export type DependencyType = | 'credentialId' | 'dataTableId' diff --git a/packages/@n8n/db/src/repositories/workflow-dependency.repository.ts b/packages/@n8n/db/src/repositories/workflow-dependency.repository.ts index 64a5d61acb0..bdc93bf2d0d 100644 --- a/packages/@n8n/db/src/repositories/workflow-dependency.repository.ts +++ b/packages/@n8n/db/src/repositories/workflow-dependency.repository.ts @@ -1,12 +1,18 @@ import { DatabaseConfig } from '@n8n/config'; import { Service } from '@n8n/di'; -import { DataSource, EntityManager, IsNull, LessThan, Repository, Not } from '@n8n/typeorm'; +import { + DataSource, + EntityManager, + IsNull, + LessThan, + LessThanOrEqual, + Repository, + Not, +} from '@n8n/typeorm'; import { SharedWorkflowRepository } from './shared-workflow.repository'; import type { User } from '../entities'; -import { WorkflowDependency, WorkflowEntity } from '../entities'; - -const INDEX_VERSION_ID = 1; +import { WorkflowDependency, WorkflowEntity, WORKFLOW_DEPENDENCY_INDEX_VERSION } from '../entities'; /** * Which workflows an aggregate over the index may span. The roles come from the caller @@ -45,7 +51,7 @@ export class WorkflowDependencies { workflowId: this.workflowId, workflowVersionId: this.workflowVersionId, publishedVersionId: this.publishedVersionId, - indexVersionId: INDEX_VERSION_ID, + indexVersionId: WORKFLOW_DEPENDENCY_INDEX_VERSION, }); this.dependencies.push(dep); } @@ -203,17 +209,31 @@ export class WorkflowDependencyRepository extends Repository dependencies: WorkflowDependencies, tx: EntityManager, ): Promise { - const deleteResult = await tx.delete(WorkflowDependency, { + // NOTE: this relies on the fact that we only want to track the latest published version or draft dependencies. + // If we're updating published dependencies, checking for Not Null works because we don't actually + // care about the specific previous published version id. + const publishedVersionCondition = dependencies.publishedVersionId ? Not(IsNull()) : IsNull(); + + // Replace rows from an older workflow version, and rows an older indexer + // version wrote for the current workflow version. + const outdatedVersionResult = await tx.delete(WorkflowDependency, { workflowId, workflowVersionId: LessThan(dependencies.workflowVersionId), - // NOTE: this relies on the fact that we only want to track the latest published version or draft dependencies. - // If we're updating published dependencies, checking for Not Null works because we don't actually - // care about the specific previous published version id. - publishedVersionId: dependencies.publishedVersionId ? Not(IsNull()) : IsNull(), + publishedVersionId: publishedVersionCondition, }); + // The workflow version bound stops a stale request from replacing rows a + // newer workflow version wrote before this indexer version deployed. + const outdatedIndexResult = await tx.delete(WorkflowDependency, { + workflowId, + workflowVersionId: LessThanOrEqual(dependencies.workflowVersionId), + indexVersionId: LessThan(WORKFLOW_DEPENDENCY_INDEX_VERSION), + publishedVersionId: publishedVersionCondition, + }); + const deletedCount = + (outdatedVersionResult.affected ?? 0) + (outdatedIndexResult.affected ?? 0); - // If we deleted something, the incoming version is newer - proceed with insert - if (deleteResult.affected && deleteResult.affected > 0) { + // If we deleted something, the incoming rows are newer - proceed with insert + if (deletedCount > 0) { // NOTE: we cast to any[] because TypeORM doesn't like the JSON column. // eslint-disable-next-line @typescript-eslint/no-explicit-any await tx.insert(WorkflowDependency, dependencies.dependencies as any[]); diff --git a/packages/@n8n/db/src/repositories/workflow.repository.ts b/packages/@n8n/db/src/repositories/workflow.repository.ts index b96e21a44ad..7bc86c6f542 100644 --- a/packages/@n8n/db/src/repositories/workflow.repository.ts +++ b/packages/@n8n/db/src/repositories/workflow.repository.ts @@ -25,6 +25,7 @@ import { WorkflowEntity, WorkflowTagMapping, WorkflowDependency, + WORKFLOW_DEPENDENCY_INDEX_VERSION, User, } from '../entities'; import { SharedWorkflow } from '../entities/shared-workflow'; @@ -1750,8 +1751,9 @@ export class WorkflowRepository extends BaseRepository { } /** - * Find workflows that need draft indexing - either unindexed (no draft entries in workflow_dependency) - * or outdated (versionCounter > workflowVersionId in workflow_dependency for drafts). + * Find workflows that need draft indexing - unindexed (no draft entries in workflow_dependency), + * outdated (versionCounter > workflowVersionId in workflow_dependency for drafts), or indexed + * by an older indexer version (indexVersionId < WORKFLOW_DEPENDENCY_INDEX_VERSION). * * NOTE: we use a simple batch limit instead of proper pagination because we use this * method to retrieve workflows and then index them immediately - so they won't be returned @@ -1762,6 +1764,7 @@ export class WorkflowRepository extends BaseRepository { const qb = this.createQueryBuilder('workflow'); const workflowIdAlias = 'workflowId'; const maxVersionIdAlias = 'maxVersionId'; + const minIndexVersionAlias = 'minIndexVersion'; const depAlias = 'dep'; // Only select columns needed for indexing to avoid loading large unused @@ -1779,6 +1782,7 @@ export class WorkflowRepository extends BaseRepository { subQuery .select('wd.workflowId', workflowIdAlias) .addSelect('MAX(wd.workflowVersionId)', maxVersionIdAlias) + .addSelect('MIN(wd.indexVersionId)', minIndexVersionAlias) .from(WorkflowDependency, 'wd') // Only consider draft dependencies (publishedVersionId IS NULL) .where('wd.publishedVersionId IS NULL') @@ -1792,9 +1796,12 @@ export class WorkflowRepository extends BaseRepository { // Include workflows that are either: // 1. Unindexed (no draft dependency entries exist) // 2. Outdated (workflow version is newer than indexed version) - qb.where(`${qb.escape(depAlias)}.${qb.escape(workflowIdAlias)} IS NULL`).orWhere( - `workflow.versionCounter > ${qb.escape(depAlias)}.${qb.escape(maxVersionIdAlias)}`, - ); + // 3. Indexed by an older indexer version + qb.where(`${qb.escape(depAlias)}.${qb.escape(workflowIdAlias)} IS NULL`) + .orWhere(`workflow.versionCounter > ${qb.escape(depAlias)}.${qb.escape(maxVersionIdAlias)}`) + .orWhere(`${qb.escape(depAlias)}.${qb.escape(minIndexVersionAlias)} < :indexVersion`, { + indexVersion: WORKFLOW_DEPENDENCY_INDEX_VERSION, + }); if (batchSize) { qb.limit(batchSize); } @@ -1806,7 +1813,8 @@ export class WorkflowRepository extends BaseRepository { * Find active workflows that need published version indexing. * These are workflows where: * - activeVersionId IS NOT NULL (workflow is active/published) - * - No dependency rows exist with matching publishedVersionId = activeVersionId + * - No dependency rows exist with matching publishedVersionId = activeVersionId, + * or the matching rows were written by an older indexer version * * This includes the activeVersion relation for efficiency. */ @@ -1816,6 +1824,7 @@ export class WorkflowRepository extends BaseRepository { const qb = this.createQueryBuilder('workflow'); const depAlias = 'dep'; const publishedVersionIdAlias = 'publishedVersionId'; + const minIndexVersionAlias = 'minIndexVersion'; // Only select columns needed for indexing to avoid loading large unused // JSON columns (connections, staticData, pinData) that can cause OOM. @@ -1829,6 +1838,7 @@ export class WorkflowRepository extends BaseRepository { return subQuery .select('wd.workflowId', 'workflowId') .addSelect('wd.publishedVersionId', publishedVersionIdAlias) + .addSelect('MIN(wd.indexVersionId)', minIndexVersionAlias) .from(WorkflowDependency, 'wd') .where('wd.publishedVersionId IS NOT NULL') .groupBy('wd.workflowId') @@ -1838,9 +1848,11 @@ export class WorkflowRepository extends BaseRepository { `workflow.id = ${qb.escape(depAlias)}.${qb.escape('workflowId')} AND workflow.activeVersionId = ${qb.escape(depAlias)}.${qb.escape(publishedVersionIdAlias)}`, ); - // Only include active workflows with no matching published version dependency + // Only include active workflows whose published version dependency is + // missing or was written by an older indexer version qb.where('workflow.activeVersionId IS NOT NULL').andWhere( - `${qb.escape(depAlias)}.${qb.escape(publishedVersionIdAlias)} IS NULL`, + `(${qb.escape(depAlias)}.${qb.escape(publishedVersionIdAlias)} IS NULL OR ${qb.escape(depAlias)}.${qb.escape(minIndexVersionAlias)} < :indexVersion)`, + { indexVersion: WORKFLOW_DEPENDENCY_INDEX_VERSION }, ); // Include the published version's nodes for indexing (skip connections to save memory). diff --git a/packages/cli/src/modules/workflow-index/__tests__/workflow-index.service.test.ts b/packages/cli/src/modules/workflow-index/__tests__/workflow-index.service.test.ts index 1ab0f5f5823..7f33d1b3b34 100644 --- a/packages/cli/src/modules/workflow-index/__tests__/workflow-index.service.test.ts +++ b/packages/cli/src/modules/workflow-index/__tests__/workflow-index.service.test.ts @@ -239,6 +239,114 @@ describe('WorkflowIndexService', () => { ); }); + it('should create workflowCall dependencies for sub-workflow tool and retriever nodes', async () => { + mockWorkflowDependencyRepository.updateDependenciesForWorkflow.mockResolvedValue(true); + + const workflow = createWorkflow([ + createNode({ + id: 'node-1', + type: '@n8n/n8n-nodes-langchain.toolWorkflow', + typeVersion: 2.2, + parameters: { source: 'database', workflowId: { mode: 'list', value: 'sub-workflow-1' } }, + }), + // toolWorkflow <= 1.1 stores the workflowId as a plain string. + createNode({ + id: 'node-2', + type: '@n8n/n8n-nodes-langchain.toolWorkflow', + typeVersion: 1, + parameters: { source: 'database', workflowId: 'sub-workflow-2' }, + }), + createNode({ + id: 'node-3', + type: '@n8n/n8n-nodes-langchain.retrieverWorkflow', + parameters: { source: 'database', workflowId: 'sub-workflow-3' }, + }), + ]); + + await service.updateIndexForDraft(workflow); + + expect(mockWorkflowDependencyRepository.updateDependenciesForWorkflow).toHaveBeenCalledWith( + 'workflow-123', + expect.objectContaining({ + dependencies: expect.arrayContaining([ + expect.objectContaining({ + dependencyType: 'workflowCall', + dependencyKey: 'sub-workflow-1', + dependencyInfo: { nodeId: 'node-1', nodeVersion: 2.2 }, + }), + expect.objectContaining({ + dependencyType: 'workflowCall', + dependencyKey: 'sub-workflow-2', + dependencyInfo: { nodeId: 'node-2', nodeVersion: 1 }, + }), + expect.objectContaining({ + dependencyType: 'workflowCall', + dependencyKey: 'sub-workflow-3', + dependencyInfo: { nodeId: 'node-3', nodeVersion: 1 }, + }), + ]), + }), + ); + }); + + it('should not create workflowCall dependencies for sub-workflow tool nodes with an inline workflow', async () => { + mockWorkflowDependencyRepository.updateDependenciesForWorkflow.mockResolvedValue(true); + + const workflow = createWorkflow([ + createNode({ + id: 'node-1', + type: '@n8n/n8n-nodes-langchain.toolWorkflow', + parameters: { source: 'parameter', workflowJson: '{}' }, + }), + ]); + + await service.updateIndexForDraft(workflow); + + expect(mockWorkflowDependencyRepository.updateDependenciesForWorkflow).toHaveBeenCalledWith( + 'workflow-123', + expect.objectContaining({ + dependencies: expect.not.arrayContaining([ + expect.objectContaining({ + dependencyType: 'workflowCall', + }), + ]), + }), + ); + }); + + it('should not create workflowCall dependencies for expression-based workflow IDs', async () => { + mockWorkflowDependencyRepository.updateDependenciesForWorkflow.mockResolvedValue(true); + + const workflow = createWorkflow([ + createNode({ + id: 'node-1', + type: 'n8n-nodes-base.executeWorkflow', + parameters: { workflowId: '={{ $json.workflowId }}' }, + }), + createNode({ + id: 'node-2', + type: '@n8n/n8n-nodes-langchain.toolWorkflow', + parameters: { + source: 'database', + workflowId: { mode: 'id', value: '={{ $json.workflowId }}' }, + }, + }), + ]); + + await service.updateIndexForDraft(workflow); + + expect(mockWorkflowDependencyRepository.updateDependenciesForWorkflow).toHaveBeenCalledWith( + 'workflow-123', + expect.objectContaining({ + dependencies: expect.not.arrayContaining([ + expect.objectContaining({ + dependencyType: 'workflowCall', + }), + ]), + }), + ); + }); + it('should handle multiple credentials on a single node', async () => { mockWorkflowDependencyRepository.updateDependenciesForWorkflow.mockResolvedValue(true); diff --git a/packages/cli/src/modules/workflow-index/workflow-index.service.ts b/packages/cli/src/modules/workflow-index/workflow-index.service.ts index 46e0ba8a601..493bdb3fb5b 100644 --- a/packages/cli/src/modules/workflow-index/workflow-index.service.ts +++ b/packages/cli/src/modules/workflow-index/workflow-index.service.ts @@ -5,7 +5,13 @@ import { WorkflowDependencies, WorkflowDependencyRepository, WorkflowRepository import { Service } from '@n8n/di'; import { ErrorReporter, SpanStatus, Tracing } from 'n8n-core'; import { ensureError } from '@n8n/utils/errors/ensure-error'; -import { DATA_TABLE_NODE_TYPES, INode, IWorkflowBase, IWorkflowSettings } from 'n8n-workflow'; +import { + DATA_TABLE_NODE_TYPES, + INode, + IWorkflowBase, + IWorkflowSettings, + isNodeWithWorkflowSelector, +} from 'n8n-workflow'; import { EventService } from '@/events/event.service'; @@ -307,13 +313,18 @@ export class WorkflowIndexService { } private addWorkflowCallDependencies(node: INode, dependencyUpdates: WorkflowDependencies): void { - if (node.type !== 'n8n-nodes-base.executeWorkflow') { + // Covers the Execute Sub-workflow node, the sub-workflow tool, and the workflow retriever. + if (!isNodeWithWorkflowSelector(node)) { return; } const calledWorkflowId: string | undefined = this.getCalledWorkflowIdFrom(node); if (!calledWorkflowId) { return; } + // Expressions resolve at runtime, so they are not static dependencies. + if (calledWorkflowId.trim().startsWith('=')) { + return; + } dependencyUpdates.add({ dependencyType: 'workflowCall', dependencyKey: calledWorkflowId, @@ -378,7 +389,7 @@ export class WorkflowIndexService { return node.parameters['workflowId']['value']; } this.errorReporter.warn( - `While indexing, could not determine called workflow ID from executeWorkflow node ${node.id}`, + `While indexing, could not determine called workflow ID from ${node.type} node ${node.id}`, { extra: node.parameters }, ); return undefined; diff --git a/packages/cli/test/integration/database/repositories/workflow-dependency.repository.test.ts b/packages/cli/test/integration/database/repositories/workflow-dependency.repository.test.ts index a81aba49605..9373c9ac36f 100644 --- a/packages/cli/test/integration/database/repositories/workflow-dependency.repository.test.ts +++ b/packages/cli/test/integration/database/repositories/workflow-dependency.repository.test.ts @@ -10,6 +10,7 @@ import { WorkflowDependencyRepository, WorkflowDependencies, WorkflowRepository, + WORKFLOW_DEPENDENCY_INDEX_VERSION, } from '@n8n/db'; import { Container } from '@n8n/di'; @@ -73,7 +74,7 @@ describe('WorkflowDependencyRepository', () => { dependencyType: 'credentialId', dependencyKey: 'cred-123', dependencyInfo: { name: 'Test Credential' }, - indexVersionId: 1, + indexVersionId: WORKFLOW_DEPENDENCY_INDEX_VERSION, }); expect(savedDependencies[1]).toMatchObject({ workflowId: workflow.id, @@ -81,7 +82,7 @@ describe('WorkflowDependencyRepository', () => { dependencyType: 'nodeType', dependencyKey: 'n8n-nodes-base.httpRequest', dependencyInfo: null, - indexVersionId: 1, + indexVersionId: WORKFLOW_DEPENDENCY_INDEX_VERSION, }); }); @@ -179,6 +180,100 @@ describe('WorkflowDependencyRepository', () => { expect(savedDependencies[0].workflowVersionId).toBe(2); }); + it('should replace rows an older indexer version wrote for the same workflow version', async () => { + // + // ARRANGE + // + const workflow = await createWorkflow({ versionId: 'v1', nodes: [] }); + + // Seed rows directly, as an older indexer version would have written them. + await workflowDependencyRepository.insert({ + workflowId: workflow.id, + workflowVersionId: 1, + publishedVersionId: null, + dependencyType: 'credentialId', + dependencyKey: 'cred-old', + dependencyInfo: null, + indexVersionId: WORKFLOW_DEPENDENCY_INDEX_VERSION - 1, + }); + + const sameVersionDeps = new WorkflowDependencies(workflow.id, 1); + sameVersionDeps.add({ + dependencyType: 'credentialId', + dependencyKey: 'cred-new', + dependencyInfo: null, + }); + + // + // ACT + // + const result = await workflowDependencyRepository.updateDependenciesForWorkflow( + workflow.id, + sameVersionDeps, + ); + + // + // ASSERT + // + expect(result).toBe(true); + const savedDependencies = await workflowDependencyRepository.find({ + where: { workflowId: workflow.id }, + }); + expect(savedDependencies).toHaveLength(1); + expect(savedDependencies[0]).toMatchObject({ + dependencyKey: 'cred-new', + workflowVersionId: 1, + indexVersionId: WORKFLOW_DEPENDENCY_INDEX_VERSION, + }); + }); + + it('should not let a stale request replace rows a newer workflow version wrote', async () => { + // + // ARRANGE + // + const workflow = await createWorkflow({ versionId: 'v1', nodes: [] }); + + // Rows an older indexer version wrote for a newer workflow version. + await workflowDependencyRepository.insert({ + workflowId: workflow.id, + workflowVersionId: 5, + publishedVersionId: null, + dependencyType: 'credentialId', + dependencyKey: 'cred-newer', + dependencyInfo: null, + indexVersionId: WORKFLOW_DEPENDENCY_INDEX_VERSION - 1, + }); + + const staleDeps = new WorkflowDependencies(workflow.id, 3); + staleDeps.add({ + dependencyType: 'credentialId', + dependencyKey: 'cred-stale', + dependencyInfo: null, + }); + + // + // ACT + // + const result = await workflowDependencyRepository.updateDependenciesForWorkflow( + workflow.id, + staleDeps, + ); + + // + // ASSERT + // + expect(result).toBe(false); + const savedDependencies = await workflowDependencyRepository.find({ + where: { workflowId: workflow.id }, + }); + expect(savedDependencies).toHaveLength(1); + expect(savedDependencies[0]).toMatchObject({ + dependencyKey: 'cred-newer', + workflowVersionId: 5, + indexVersionId: WORKFLOW_DEPENDENCY_INDEX_VERSION - 1, + }); + }); + it('should prevent races between concurrent updates', async () => { // // ARRANGE @@ -594,7 +689,7 @@ describe('WorkflowDependencyRepository node usage', () => { dependencyType: 'nodeType' as const, dependencyKey: index < anthropicCount ? ANTHROPIC : OPENAI, dependencyInfo: null, - indexVersionId: 1, + indexVersionId: WORKFLOW_DEPENDENCY_INDEX_VERSION, })), async (chunk) => await repository.insert(chunk), ); diff --git a/packages/cli/test/integration/database/repositories/workflow.repository.test.ts b/packages/cli/test/integration/database/repositories/workflow.repository.test.ts index 98659b25cb2..5b6421536db 100644 --- a/packages/cli/test/integration/database/repositories/workflow.repository.test.ts +++ b/packages/cli/test/integration/database/repositories/workflow.repository.test.ts @@ -8,7 +8,12 @@ import { getWorkflowById, setActiveVersion, } from '@n8n/backend-test-utils'; -import { WorkflowRepository, WorkflowDependencyRepository, WorkflowDependencies } from '@n8n/db'; +import { + WorkflowRepository, + WorkflowDependencyRepository, + WorkflowDependencies, + WORKFLOW_DEPENDENCY_INDEX_VERSION, +} from '@n8n/db'; import { Container } from '@n8n/di'; import type { Scope } from '@n8n/permissions'; @@ -484,6 +489,51 @@ describe('WorkflowRepository', () => { expect(workflowIds).not.toContain(workflow3.id); }); + it('should return workflows whose rows an older indexer version wrote', async () => { + // + // ARRANGE + // + const workflowRepository = Container.get(WorkflowRepository); + const workflowDependencyRepository = Container.get(WorkflowDependencyRepository); + + // Indexed at the current version counter, but by an older indexer version. + const outdatedWorkflow = await createWorkflow({ versionCounter: 3, nodes: [] }); + await workflowDependencyRepository.insert({ + workflowId: outdatedWorkflow.id, + workflowVersionId: 3, + publishedVersionId: null, + dependencyType: 'nodeType', + dependencyKey: 'n8n-nodes-base.httpRequest', + dependencyInfo: null, + indexVersionId: WORKFLOW_DEPENDENCY_INDEX_VERSION - 1, + }); + + // Indexed by the current indexer version. + const currentWorkflow = await createWorkflow({ versionCounter: 3, nodes: [] }); + const dependencies = new WorkflowDependencies(currentWorkflow.id, 3); + dependencies.add({ + dependencyType: 'nodeType', + dependencyKey: 'n8n-nodes-base.httpRequest', + dependencyInfo: null, + }); + await workflowDependencyRepository.updateDependenciesForWorkflow( + currentWorkflow.id, + dependencies, + ); + + // + // ACT + // + const workflowsNeedingIndexing = await workflowRepository.findWorkflowsNeedingIndexing(); + + // + // ASSERT + // + const workflowIds = workflowsNeedingIndexing.map((w) => w.id); + expect(workflowIds).toContain(outdatedWorkflow.id); + expect(workflowIds).not.toContain(currentWorkflow.id); + }); + it('should respect the batch size limit', async () => { // // ARRANGE @@ -509,6 +559,62 @@ describe('WorkflowRepository', () => { }); }); + describe('findWorkflowsNeedingPublishedVersionIndexing', () => { + it('should return active workflows with missing or outdated published rows', async () => { + // + // ARRANGE + // + const workflowRepository = Container.get(WorkflowRepository); + const workflowDependencyRepository = Container.get(WorkflowDependencyRepository); + + // Active, but no published dependency rows exist. + const unindexedWorkflow = await createActiveWorkflow(); + + // Published rows exist, but an older indexer version wrote them. + const outdatedWorkflow = await createActiveWorkflow(); + await workflowDependencyRepository.insert({ + workflowId: outdatedWorkflow.id, + workflowVersionId: 1, + publishedVersionId: outdatedWorkflow.activeVersionId, + dependencyType: 'nodeType', + dependencyKey: 'n8n-nodes-base.httpRequest', + dependencyInfo: null, + indexVersionId: WORKFLOW_DEPENDENCY_INDEX_VERSION - 1, + }); + + // Published rows exist at the current indexer version. + const currentWorkflow = await createActiveWorkflow(); + const dependencies = new WorkflowDependencies( + currentWorkflow.id, + 1, + currentWorkflow.activeVersionId, + ); + dependencies.add({ + dependencyType: 'nodeType', + dependencyKey: 'n8n-nodes-base.httpRequest', + dependencyInfo: null, + }); + await workflowDependencyRepository.updateDependenciesForWorkflow( + currentWorkflow.id, + dependencies, + ); + + // + // ACT + // + const workflowsNeedingIndexing = + await workflowRepository.findWorkflowsNeedingPublishedVersionIndexing(); + + // + // ASSERT + // + const workflowIds = workflowsNeedingIndexing.map((w) => w.id); + expect(workflowIds).toContain(unindexedWorkflow.id); + expect(workflowIds).toContain(outdatedWorkflow.id); + expect(workflowIds).not.toContain(currentWorkflow.id); + }); + }); + describe('findWorkflowsWithNodeType', () => { it('should return workflows that have the specified node type in workflow_dependency', async () => { // diff --git a/packages/cli/test/integration/workflows/workflow-index.test.ts b/packages/cli/test/integration/workflows/workflow-index.test.ts index a8725064d8b..068d7bb4a94 100644 --- a/packages/cli/test/integration/workflows/workflow-index.test.ts +++ b/packages/cli/test/integration/workflows/workflow-index.test.ts @@ -208,7 +208,7 @@ describe('WorkflowIndexService Integration', () => { nodeId: 'node-1', nodeVersion: 1, }, - indexVersionId: 1, + indexVersionId: 2, }); }); });