compute hash and update on each round

Signed-off-by: Fredrik Adelöw <freben@gmail.com>

plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts

@@ -20,6 +20,8 @@ import {
   stringifyEntityRef,
 } from '@backstage/catalog-model';
 import { serializeError } from '@backstage/errors';
+import { createHash } from 'crypto';
+import stableStringify from 'fast-json-stable-stringify';
 import { Logger } from 'winston';
 import { ProcessingDatabase, RefreshStateItem } from './database/types';
 import { CatalogProcessingOrchestrator } from './processing/types';
@@ -125,7 +127,14 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
       },
       processTask: async item => {
         try {
-          const { id, state, unprocessedEntity, entityRef, locationKey } = item;
+          const {
+            id,
+            state,
+            unprocessedEntity,
+            entityRef,
+            locationKey,
+            hash: previousHash,
+          } = item;
           const result = await this.orchestrator.process({
             entity: unprocessedEntity,
             state,
@@ -142,6 +151,22 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
             result.errors.map(e => serializeError(e)),
           );
 
+          let hashBuilder = createHash('sha1').update(errorsString);
+          if (result.ok) {
+            hashBuilder = hashBuilder
+              .update(stableStringify({ ...result.completedEntity }))
+              .update(stableStringify([...result.deferredEntities]))
+              .update(stableStringify([...result.relations]))
+              .update(stableStringify({ ...result.state }));
+          }
+
+          const hash = hashBuilder.digest('hex');
+          if (hash === previousHash) {
+            console.log('skipping ', entityRef);
+            return;
+          }
+          console.log('going ahead with ', entityRef);
+
           // If the result was marked as not OK, it signals that some part of the
           // processing pipeline threw an exception. This can happen both as part of
           // non-catastrophic things such as due to validation errors, as well as if
@@ -154,6 +179,7 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
               await this.processingDatabase.updateProcessedEntityErrors(tx, {
                 id,
                 errors: errorsString,
+                hash,
               });
             });
             await this.stitcher.stitch(
@@ -167,6 +193,7 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
             await this.processingDatabase.updateProcessedEntity(tx, {
               id,
               processedEntity: result.completedEntity,
+              hash,
               state: result.state,
               errors: errorsString,
               relations: result.relations,

plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts

@@ -206,6 +206,7 @@ describe('Default Processing Database', () => {
             db.updateProcessedEntity(tx, {
               id,
               processedEntity,
+              hash: '',
               state: new Map<string, JsonObject>(),
               relations: [],
               deferredEntities: [],
@@ -224,6 +225,7 @@ describe('Default Processing Database', () => {
         const options = {
           id,
           processedEntity,
+          hash: '',
           state: new Map<string, JsonObject>(),
           relations: [],
           deferredEntities: [],
@@ -250,7 +252,11 @@ describe('Default Processing Database', () => {
 
         await db.transaction(tx =>
           expect(
-            db.updateProcessedEntity(tx, { ...options, locationKey: 'fail' }),
+            db.updateProcessedEntity(tx, {
+              ...options,
+              hash: '',
+              locationKey: 'fail',
+            }),
           ).rejects.toThrow(
             `Conflicting write of processing result for ${id} with location key 'fail'`,
           ),
@@ -280,6 +286,7 @@ describe('Default Processing Database', () => {
           db.updateProcessedEntity(tx, {
             id,
             processedEntity,
+            hash: '',
             state,
             relations: [],
             deferredEntities: [],
@@ -336,6 +343,7 @@ describe('Default Processing Database', () => {
           db.updateProcessedEntity(tx, {
             id,
             processedEntity,
+            hash: '',
             state: new Map<string, JsonObject>(),
             relations: relations,
             deferredEntities: [],
@@ -387,6 +395,7 @@ describe('Default Processing Database', () => {
           db.updateProcessedEntity(tx, {
             id,
             processedEntity,
+            hash: '',
             state: new Map<string, JsonObject>(),
             relations: [],
             deferredEntities,

plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts

@@ -60,6 +60,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
     const {
       id,
       processedEntity,
+      hash,
       state,
       errors,
       relations,
@@ -69,6 +70,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
     const refreshResult = await tx<DbRefreshStateRow>('refresh_state')
       .update({
         processed_entity: JSON.stringify(processedEntity),
+        hash,
         cache: JSON.stringify(state),
         errors,
         location_key: locationKey,
@@ -492,6 +494,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
             processedEntity: i.processed_entity
               ? (JSON.parse(i.processed_entity) as Entity)
               : undefined,
+            hash: i.hash,
             nextUpdateAt: i.next_update_at,
             lastDiscoveryAt: i.last_discovery_at,
             state: i.cache

plugins/catalog-backend/src/next/database/tables.ts

@@ -25,6 +25,7 @@ export type DbRefreshStateRow = {
   entity_ref: string;
   unprocessed_entity: string;
   processed_entity?: string;
+  hash?: string;
   cache?: string;
   next_update_at: string;
   last_discovery_at: string; // remove?

plugins/catalog-backend/src/next/database/types.ts

@@ -34,6 +34,7 @@ export type AddUnprocessedEntitiesResult = {};
 export type UpdateProcessedEntityOptions = {
   id: string;
   processedEntity: Entity;
+  hash: string;
   state?: Map<string, JsonObject>;
   errors?: string;
   relations: EntityRelationSpec[];
@@ -44,6 +45,7 @@ export type UpdateProcessedEntityOptions = {
 export type UpdateProcessedEntityErrorsOptions = {
   id: string;
   errors?: string;
+  hash: string;
 };
 
 export type RefreshStateItem = {
@@ -51,6 +53,7 @@ export type RefreshStateItem = {
   entityRef: string;
   unprocessedEntity: Entity;
   processedEntity?: Entity;
+  hash: string;
   nextUpdateAt: string;
   lastDiscoveryAt: string; // remove?
   state: Map<string, JsonObject>;