feat(core): Execution curation (#10342)

Co-authored-by: oleg <me@olegivaniv.com>

packages/cli/test/integration/execution.service.integration.test.ts

@@ -3,7 +3,7 @@ import { ExecutionService } from '@/executions/execution.service';
 import { mock } from 'jest-mock-extended';
 import Container from 'typedi';
 import { createWorkflow } from './shared/db/workflows';
-import { createExecution } from './shared/db/executions';
+import { annotateExecution, createAnnotationTags, createExecution } from './shared/db/executions';
 import * as testDb from './shared/test-db';
 import { WorkflowRepository } from '@/databases/repositories/workflow.repository';
 import type { ExecutionSummaries } from '@/executions/execution.types';
@@ -19,6 +19,8 @@ describe('ExecutionService', () => {
 		executionRepository = Container.get(ExecutionRepository);
 
 		executionService = new ExecutionService(
+			mock(),
+			mock(),
 			mock(),
 			mock(),
 			mock(),
@@ -70,6 +72,10 @@ describe('ExecutionService', () => {
 				waitTill: null,
 				retrySuccessId: null,
 				workflowName: expect.any(String),
+				annotation: {
+					tags: expect.arrayContaining([]),
+					vote: null,
+				},
 			};
 
 			expect(output.count).toBe(2);
@@ -462,4 +468,201 @@ describe('ExecutionService', () => {
 			expect(results[1].status).toBe('running');
 		});
 	});
+
+	describe('annotation', () => {
+		const summaryShape = {
+			id: expect.any(String),
+			workflowId: expect.any(String),
+			mode: expect.any(String),
+			retryOf: null,
+			status: expect.any(String),
+			startedAt: expect.any(String),
+			stoppedAt: expect.any(String),
+			waitTill: null,
+			retrySuccessId: null,
+			workflowName: expect.any(String),
+		};
+
+		afterEach(async () => {
+			await testDb.truncate(['AnnotationTag', 'ExecutionAnnotation']);
+		});
+
+		test('should add and retrieve annotation', async () => {
+			const workflow = await createWorkflow();
+
+			const execution1 = await createExecution({ status: 'success' }, workflow);
+			const execution2 = await createExecution({ status: 'success' }, workflow);
+
+			const annotationTags = await createAnnotationTags(['tag1', 'tag2', 'tag3']);
+
+			await annotateExecution(
+				execution1.id,
+				{ vote: 'up', tags: [annotationTags[0].id, annotationTags[1].id] },
+				[workflow.id],
+			);
+			await annotateExecution(execution2.id, { vote: 'down', tags: [annotationTags[2].id] }, [
+				workflow.id,
+			]);
+
+			const query: ExecutionSummaries.RangeQuery = {
+				kind: 'range',
+				status: ['success'],
+				range: { limit: 20 },
+				accessibleWorkflowIds: [workflow.id],
+			};
+
+			const output = await executionService.findRangeWithCount(query);
+
+			expect(output.count).toBe(2);
+			expect(output.estimated).toBe(false);
+			expect(output.results).toEqual(
+				expect.arrayContaining([
+					{
+						...summaryShape,
+						annotation: {
+							tags: [expect.objectContaining({ name: 'tag3' })],
+							vote: 'down',
+						},
+					},
+					{
+						...summaryShape,
+						annotation: {
+							tags: [
+								expect.objectContaining({ name: 'tag1' }),
+								expect.objectContaining({ name: 'tag2' }),
+							],
+							vote: 'up',
+						},
+					},
+				]),
+			);
+		});
+
+		test('should update annotation', async () => {
+			const workflow = await createWorkflow();
+
+			const execution = await createExecution({ status: 'success' }, workflow);
+
+			const annotationTags = await createAnnotationTags(['tag1', 'tag2', 'tag3']);
+
+			await annotateExecution(execution.id, { vote: 'up', tags: [annotationTags[0].id] }, [
+				workflow.id,
+			]);
+
+			await annotateExecution(execution.id, { vote: 'down', tags: [annotationTags[1].id] }, [
+				workflow.id,
+			]);
+
+			const query: ExecutionSummaries.RangeQuery = {
+				kind: 'range',
+				status: ['success'],
+				range: { limit: 20 },
+				accessibleWorkflowIds: [workflow.id],
+			};
+
+			const output = await executionService.findRangeWithCount(query);
+
+			expect(output.count).toBe(1);
+			expect(output.estimated).toBe(false);
+			expect(output.results).toEqual([
+				{
+					...summaryShape,
+					annotation: {
+						tags: [expect.objectContaining({ name: 'tag2' })],
+						vote: 'down',
+					},
+				},
+			]);
+		});
+
+		test('should filter by annotation tags', async () => {
+			const workflow = await createWorkflow();
+
+			const executions = await Promise.all([
+				createExecution({ status: 'success' }, workflow),
+				createExecution({ status: 'success' }, workflow),
+			]);
+
+			const annotationTags = await createAnnotationTags(['tag1', 'tag2', 'tag3']);
+
+			await annotateExecution(
+				executions[0].id,
+				{ vote: 'up', tags: [annotationTags[0].id, annotationTags[1].id] },
+				[workflow.id],
+			);
+			await annotateExecution(executions[1].id, { vote: 'down', tags: [annotationTags[2].id] }, [
+				workflow.id,
+			]);
+
+			const query: ExecutionSummaries.RangeQuery = {
+				kind: 'range',
+				status: ['success'],
+				range: { limit: 20 },
+				accessibleWorkflowIds: [workflow.id],
+				annotationTags: [annotationTags[0].id],
+			};
+
+			const output = await executionService.findRangeWithCount(query);
+
+			expect(output.count).toBe(1);
+			expect(output.estimated).toBe(false);
+			expect(output.results).toEqual([
+				{
+					...summaryShape,
+					annotation: {
+						tags: [
+							expect.objectContaining({ name: 'tag1' }),
+							expect.objectContaining({ name: 'tag2' }),
+						],
+						vote: 'up',
+					},
+				},
+			]);
+		});
+
+		test('should filter by annotation vote', async () => {
+			const workflow = await createWorkflow();
+
+			const executions = await Promise.all([
+				createExecution({ status: 'success' }, workflow),
+				createExecution({ status: 'success' }, workflow),
+			]);
+
+			const annotationTags = await createAnnotationTags(['tag1', 'tag2', 'tag3']);
+
+			await annotateExecution(
+				executions[0].id,
+				{ vote: 'up', tags: [annotationTags[0].id, annotationTags[1].id] },
+				[workflow.id],
+			);
+			await annotateExecution(executions[1].id, { vote: 'down', tags: [annotationTags[2].id] }, [
+				workflow.id,
+			]);
+
+			const query: ExecutionSummaries.RangeQuery = {
+				kind: 'range',
+				status: ['success'],
+				range: { limit: 20 },
+				accessibleWorkflowIds: [workflow.id],
+				vote: 'up',
+			};
+
+			const output = await executionService.findRangeWithCount(query);
+
+			expect(output.count).toBe(1);
+			expect(output.estimated).toBe(false);
+			expect(output.results).toEqual([
+				{
+					...summaryShape,
+					annotation: {
+						tags: [
+							expect.objectContaining({ name: 'tag1' }),
+							expect.objectContaining({ name: 'tag2' }),
+						],
+						vote: 'up',
+					},
+				},
+			]);
+		});
+	});
 });

packages/cli/test/integration/pruning.service.test.ts

@@ -13,7 +13,11 @@ import { Logger } from '@/logger';
 
 import { mockInstance } from '../shared/mocking';
 import { createWorkflow } from './shared/db/workflows';
-import { createExecution, createSuccessfulExecution } from './shared/db/executions';
+import {
+	annotateExecution,
+	createExecution,
+	createSuccessfulExecution,
+} from './shared/db/executions';
 import { mock } from 'jest-mock-extended';
 
 describe('softDeleteOnPruningCycle()', () => {
@@ -40,7 +44,7 @@ describe('softDeleteOnPruningCycle()', () => {
 	});
 
 	beforeEach(async () => {
-		await testDb.truncate(['Execution']);
+		await testDb.truncate(['Execution', 'ExecutionAnnotation']);
 	});
 
 	afterAll(async () => {
@@ -138,6 +142,25 @@ describe('softDeleteOnPruningCycle()', () => {
 				expect.objectContaining({ id: executions[1].id, deletedAt: null }),
 			]);
 		});
+
+		test('should not prune annotated executions', async () => {
+			const executions = [
+				await createSuccessfulExecution(workflow),
+				await createSuccessfulExecution(workflow),
+				await createSuccessfulExecution(workflow),
+			];
+
+			await annotateExecution(executions[0].id, { vote: 'up' }, [workflow.id]);
+
+			await pruningService.softDeleteOnPruningCycle();
+
+			const result = await findAllExecutions();
+			expect(result).toEqual([
+				expect.objectContaining({ id: executions[0].id, deletedAt: null }),
+				expect.objectContaining({ id: executions[1].id, deletedAt: expect.any(Date) }),
+				expect.objectContaining({ id: executions[2].id, deletedAt: null }),
+			]);
+		});
 	});
 
 	describe('when EXECUTIONS_DATA_MAX_AGE is set', () => {
@@ -226,5 +249,33 @@ describe('softDeleteOnPruningCycle()', () => {
 				expect.objectContaining({ id: executions[1].id, deletedAt: null }),
 			]);
 		});
+
+		test('should not prune annotated executions', async () => {
+			const executions = [
+				await createExecution(
+					{ finished: true, startedAt: yesterday, stoppedAt: yesterday, status: 'success' },
+					workflow,
+				),
+				await createExecution(
+					{ finished: true, startedAt: yesterday, stoppedAt: yesterday, status: 'success' },
+					workflow,
+				),
+				await createExecution(
+					{ finished: true, startedAt: now, stoppedAt: now, status: 'success' },
+					workflow,
+				),
+			];
+
+			await annotateExecution(executions[0].id, { vote: 'up' }, [workflow.id]);
+
+			await pruningService.softDeleteOnPruningCycle();
+
+			const result = await findAllExecutions();
+			expect(result).toEqual([
+				expect.objectContaining({ id: executions[0].id, deletedAt: null }),
+				expect.objectContaining({ id: executions[1].id, deletedAt: expect.any(Date) }),
+				expect.objectContaining({ id: executions[2].id, deletedAt: null }),
+			]);
+		});
 	});
 });

packages/cli/test/integration/shared/db/executions.ts

@@ -5,6 +5,13 @@ import type { WorkflowEntity } from '@/databases/entities/workflow-entity';
 import { ExecutionRepository } from '@/databases/repositories/execution.repository';
 import { ExecutionDataRepository } from '@/databases/repositories/execution-data.repository';
 import { ExecutionMetadataRepository } from '@/databases/repositories/execution-metadata.repository';
+import { ExecutionService } from '@/executions/execution.service';
+import type { AnnotationVote } from 'n8n-workflow';
+import { mockInstance } from '@test/mocking';
+import { Telemetry } from '@/telemetry';
+import { AnnotationTagRepository } from '@/databases/repositories/annotation-tag.repository';
+
+mockInstance(Telemetry);
 
 export async function createManyExecutions(
 	amount: number,
@@ -85,6 +92,19 @@ export async function createWaitingExecution(workflow: WorkflowEntity) {
 	);
 }
 
+export async function annotateExecution(
+	executionId: string,
+	annotation: { vote?: AnnotationVote | null; tags?: string[] },
+	sharedWorkflowIds: string[],
+) {
+	await Container.get(ExecutionService).annotate(executionId, annotation, sharedWorkflowIds);
+}
+
 export async function getAllExecutions() {
 	return await Container.get(ExecutionRepository).find();
 }
+
+export async function createAnnotationTags(annotationTags: string[]) {
+	const tagRepository = Container.get(AnnotationTagRepository);
+	return await tagRepository.save(annotationTags.map((name) => tagRepository.create({ name })));
+}

packages/cli/test/integration/shared/test-db.ts

@@ -48,11 +48,13 @@ export async function terminate() {
 
 // Can't use `Object.keys(entities)` here because some entities have a `Entity` suffix, while the repositories don't
 const repositories = [
+	'AnnotationTag',
 	'AuthIdentity',
 	'AuthProviderSyncHistory',
 	'Credentials',
 	'EventDestinations',
 	'Execution',
+	'ExecutionAnnotation',
 	'ExecutionData',
 	'ExecutionMetadata',
 	'InstalledNodes',

packages/cli/test/integration/shared/types.ts

@@ -26,6 +26,7 @@ type EndpointGroup =
 	| 'eventBus'
 	| 'license'
 	| 'variables'
+	| 'annotationTags'
 	| 'tags'
 	| 'externalSecrets'
 	| 'mfa'

packages/cli/test/integration/shared/utils/test-server.ts

@@ -122,6 +122,10 @@ export const setupTestServer = ({
 		if (endpointGroups.length) {
 			for (const group of endpointGroups) {
 				switch (group) {
+					case 'annotationTags':
+						await import('@/controllers/annotation-tags.controller');
+						break;
+
 					case 'credentials':
 						await import('@/credentials/credentials.controller');
 						break;