diff --git a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.test.ts b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.test.ts index a358f1d07e..3002218d89 100644 --- a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.test.ts +++ b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.test.ts @@ -452,4 +452,247 @@ describe('DefaultCatalogProcessingEngine', () => { ); await engine.stop(); }); + + it('should not stitch sources entities when relations are the same', async () => { + const engine = new DefaultCatalogProcessingEngine( + getVoidLogger(), + db, + orchestrator, + stitcher, + () => hash, + 100, + ); + + db.transaction.mockImplementation(cb => cb((() => {}) as any)); + + const entity = { + apiVersion: '1', + kind: 'k', + metadata: { name: 'me', namespace: 'ns' }, + }; + const processableEntity = { + entityRef: 'foo', + id: '1', + unprocessedEntity: entity, + resultHash: '', + state: [] as any, + nextUpdateAt: DateTime.now(), + lastDiscoveryAt: DateTime.now(), + }; + + db.listParents.mockResolvedValue({ entityRefs: [] }); + db.getProcessableEntities.mockResolvedValueOnce({ + items: [processableEntity], + }); + db.updateProcessedEntity.mockImplementationOnce(async () => ({ + previous: { + relations: [ + { + originating_entity_id: '', + type: 't', + source_entity_ref: 'k:ns/other1', + target_entity_ref: 'k:ns/me', + }, + { + originating_entity_id: '', + type: 't', + source_entity_ref: 'k:ns/other2', + target_entity_ref: 'k:ns/me', + }, + ], + }, + })); + + orchestrator.process.mockResolvedValueOnce({ + ok: true, + completedEntity: entity, + relations: [ + { + type: 't', + source: { kind: 'k', namespace: 'ns', name: 'other1' }, + target: { kind: 'k', namespace: 'ns', name: 'me' }, + }, + { + type: 't', + source: { kind: 'k', namespace: 'ns', name: 'other2' }, + target: { kind: 'k', namespace: 'ns', name: 'me' }, + }, + ], + errors: [], + deferredEntities: [], + state: {}, + refreshKeys: [], + }); + + await engine.start(); + await waitForExpect(() => { + expect(stitcher.stitch).toHaveBeenCalledTimes(1); + }); + expect([...stitcher.stitch.mock.calls[0][0]]).toEqual( + expect.arrayContaining(['k:ns/me']), + ); + await engine.stop(); + }); + + it('should stitch sources entities when new relation of different type added', async () => { + const engine = new DefaultCatalogProcessingEngine( + getVoidLogger(), + db, + orchestrator, + stitcher, + () => hash, + 100, + ); + + db.transaction.mockImplementation(cb => cb((() => {}) as any)); + + const entity = { + apiVersion: '1', + kind: 'k', + metadata: { name: 'me', namespace: 'ns' }, + }; + const processableEntity = { + entityRef: 'foo', + id: '1', + unprocessedEntity: entity, + resultHash: '', + state: [] as any, + nextUpdateAt: DateTime.now(), + lastDiscoveryAt: DateTime.now(), + }; + + db.listParents.mockResolvedValue({ entityRefs: [] }); + db.getProcessableEntities.mockResolvedValueOnce({ + items: [processableEntity], + }); + db.updateProcessedEntity.mockImplementationOnce(async () => ({ + previous: { + relations: [ + { + originating_entity_id: '', + type: 't', + source_entity_ref: 'k:ns/other1', + target_entity_ref: 'k:ns/me', + }, + { + originating_entity_id: '', + type: 't', + source_entity_ref: 'k:ns/other2', + target_entity_ref: 'k:ns/me', + }, + ], + }, + })); + + orchestrator.process.mockResolvedValueOnce({ + ok: true, + completedEntity: entity, + relations: [ + { + type: 't', + source: { kind: 'k', namespace: 'ns', name: 'other1' }, + target: { kind: 'k', namespace: 'ns', name: 'me' }, + }, + { + type: 't', + source: { kind: 'k', namespace: 'ns', name: 'other2' }, + target: { kind: 'k', namespace: 'ns', name: 'me' }, + }, + { + type: 'u', + source: { kind: 'k', namespace: 'ns', name: 'other2' }, + target: { kind: 'k', namespace: 'ns', name: 'me' }, + }, + ], + errors: [], + deferredEntities: [], + state: {}, + refreshKeys: [], + }); + + await engine.start(); + await waitForExpect(() => { + expect(stitcher.stitch).toHaveBeenCalledTimes(1); + }); + expect([...stitcher.stitch.mock.calls[0][0]]).toEqual( + expect.arrayContaining(['k:ns/me', 'k:ns/other2']), + ); + await engine.stop(); + }); + + it('should stitch sources entities when relation is removed', async () => { + const engine = new DefaultCatalogProcessingEngine( + getVoidLogger(), + db, + orchestrator, + stitcher, + () => hash, + 100, + ); + + db.transaction.mockImplementation(cb => cb((() => {}) as any)); + + const entity = { + apiVersion: '1', + kind: 'k', + metadata: { name: 'me', namespace: 'ns' }, + }; + const processableEntity = { + entityRef: 'foo', + id: '1', + unprocessedEntity: entity, + resultHash: '', + state: [] as any, + nextUpdateAt: DateTime.now(), + lastDiscoveryAt: DateTime.now(), + }; + + db.listParents.mockResolvedValue({ entityRefs: [] }); + db.getProcessableEntities.mockResolvedValueOnce({ + items: [processableEntity], + }); + db.updateProcessedEntity.mockImplementationOnce(async () => ({ + previous: { + relations: [ + { + originating_entity_id: '', + type: 't', + source_entity_ref: 'k:ns/other1', + target_entity_ref: 'k:ns/me', + }, + { + originating_entity_id: '', + type: 't', + source_entity_ref: 'k:ns/other2', + target_entity_ref: 'k:ns/me', + }, + ], + }, + })); + + orchestrator.process.mockResolvedValueOnce({ + ok: true, + completedEntity: entity, + relations: [ + { + type: 't', + source: { kind: 'k', namespace: 'ns', name: 'other1' }, + target: { kind: 'k', namespace: 'ns', name: 'me' }, + }, + ], + errors: [], + deferredEntities: [], + state: {}, + refreshKeys: [], + }); + + await engine.start(); + await waitForExpect(() => { + expect(stitcher.stitch).toHaveBeenCalledTimes(1); + }); + expect([...stitcher.stitch.mock.calls[0][0]]).toEqual( + expect.arrayContaining(['k:ns/me', 'k:ns/other2']), + ); + await engine.stop(); + }); }); diff --git a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts index c8f159b195..20dd325a71 100644 --- a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts +++ b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts @@ -199,7 +199,7 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { } result.completedEntity.metadata.uid = id; - let oldRelationSources: Set; + let oldRelationSources: Map; await this.processingDatabase.transaction(async tx => { const { previous } = await this.processingDatabase.updateProcessedEntity(tx, { @@ -212,28 +212,37 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { locationKey, refreshKeys: result.refreshKeys, }); - oldRelationSources = new Set( - previous.relations.map(r => r.source_entity_ref), + oldRelationSources = new Map( + previous.relations.map(r => [ + `${r.source_entity_ref}:${r.target_entity_ref}:${r.type}`, + r.source_entity_ref, + ]), ); }); - const newRelationSources = new Set( - result.relations.map(relation => - stringifyEntityRef(relation.source), - ), + const newRelationSources = new Map( + result.relations.map(relation => { + const sourceEntityRef = stringifyEntityRef(relation.source); + return [ + `${sourceEntityRef}:${stringifyEntityRef(relation.target)}:${ + relation.type + }`, + sourceEntityRef, + ]; + }), ); const setOfThingsToStitch = new Set([ stringifyEntityRef(result.completedEntity), ]); - newRelationSources.forEach(r => { - if (!oldRelationSources.has(r)) { - setOfThingsToStitch.add(r); + newRelationSources.forEach((sourceEntityRef, uniqueKey) => { + if (!oldRelationSources.has(uniqueKey)) { + setOfThingsToStitch.add(sourceEntityRef); } }); - oldRelationSources!.forEach(r => { - if (!newRelationSources.has(r)) { - setOfThingsToStitch.add(r); + oldRelationSources!.forEach((sourceEntityRef, uniqueKey) => { + if (!newRelationSources.has(uniqueKey)) { + setOfThingsToStitch.add(sourceEntityRef); } });