diff --git a/packages/cli/src/databases/repositories/workflowStatistics.repository.ts b/packages/cli/src/databases/repositories/workflowStatistics.repository.ts index 3ce4ea6d8348b..5d6a9261da477 100644 --- a/packages/cli/src/databases/repositories/workflowStatistics.repository.ts +++ b/packages/cli/src/databases/repositories/workflowStatistics.repository.ts @@ -4,7 +4,7 @@ import config from '@/config'; import type { StatisticsNames } from '../entities/WorkflowStatistics'; import { WorkflowStatistics } from '../entities/WorkflowStatistics'; -type StatisticsInsertResult = 'insert' | 'failed'; +type StatisticsInsertResult = 'insert' | 'failed' | 'alreadyExists'; type StatisticsUpsertResult = StatisticsInsertResult | 'update'; @Service() @@ -21,6 +21,13 @@ export class WorkflowStatisticsRepository extends Repository ): Promise { // Try to insert the data loaded statistic try { + const exists = await this.findOne({ + where: { + workflowId, + name: eventName, + }, + }); + if (exists) return 'alreadyExists'; await this.insert({ workflowId, name: eventName, diff --git a/packages/cli/src/services/events.service.ts b/packages/cli/src/services/events.service.ts index 3f2504366ebeb..64c8a0e499ef7 100644 --- a/packages/cli/src/services/events.service.ts +++ b/packages/cli/src/services/events.service.ts @@ -73,7 +73,7 @@ export class EventsService extends EventEmitter { StatisticsNames.dataLoaded, workflowId, ); - if (insertResult === 'failed') return; + if (insertResult === 'failed' || insertResult === 'alreadyExists') return; // Compile the metrics since this was a new data loaded event const owner = await this.ownershipService.getWorkflowOwnerCached(workflowId); diff --git a/packages/cli/test/unit/repositories/workflowStatistics.test.ts b/packages/cli/test/unit/repositories/workflowStatistics.test.ts new file mode 100644 index 0000000000000..7aa0280600aab --- /dev/null +++ b/packages/cli/test/unit/repositories/workflowStatistics.test.ts @@ -0,0 +1,53 @@ +import { WorkflowStatisticsRepository } from '@db/repositories/workflowStatistics.repository'; +import { DataSource, EntityManager, InsertResult, QueryFailedError } from 'typeorm'; +import { mockInstance } from '../../shared/mocking'; +import { mock, mockClear } from 'jest-mock-extended'; +import { StatisticsNames, WorkflowStatistics } from '@/databases/entities/WorkflowStatistics'; + +const entityManager = mockInstance(EntityManager); +const dataSource = mockInstance(DataSource, { manager: entityManager }); +dataSource.getMetadata.mockReturnValue(mock()); +Object.assign(entityManager, { connection: dataSource }); +const workflowStatisticsRepository = new WorkflowStatisticsRepository(dataSource); + +describe('insertWorkflowStatistics', () => { + beforeEach(() => { + mockClear(entityManager.insert); + }); + it('Successfully inserts data when it is not yet present', async () => { + entityManager.findOne.mockResolvedValueOnce(null); + entityManager.insert.mockResolvedValueOnce(mockInstance(InsertResult)); + + const insertionResult = await workflowStatisticsRepository.insertWorkflowStatistics( + StatisticsNames.dataLoaded, + 'workflowId', + ); + + expect(insertionResult).toBe('insert'); + }); + + it('Does not insert when data is present', async () => { + entityManager.findOne.mockResolvedValueOnce(mockInstance(WorkflowStatistics)); + const insertionResult = await workflowStatisticsRepository.insertWorkflowStatistics( + StatisticsNames.dataLoaded, + 'workflowId', + ); + + expect(insertionResult).toBe('alreadyExists'); + expect(entityManager.insert).not.toHaveBeenCalled(); + }); + + it('throws an error when insertion fails', async () => { + entityManager.findOne.mockResolvedValueOnce(null); + entityManager.insert.mockImplementation(async () => { + throw new QueryFailedError('Query', [], 'driver error'); + }); + + const insertionResult = await workflowStatisticsRepository.insertWorkflowStatistics( + StatisticsNames.dataLoaded, + 'workflowId', + ); + + expect(insertionResult).toBe('failed'); + }); +});