feature: refactor processing engine, remove typing and provide an option to subscribe to errors while processing the entities.

Signed-off-by: Hasan Oezdemir <21654050+nodify-at@users.noreply.github.com>

plugins/catalog-backend/api-report.md

@@ -145,10 +145,13 @@ export class CatalogBuilder {
     processingInterval: ProcessingIntervalFunction,
   ): CatalogBuilder;
   setProcessingIntervalSeconds(seconds: number): CatalogBuilder;
-  // @alpha (undocumented)
-  subscribe(
-    catalogProcessingErrorListeners: CatalogProcessingErrorListener[],
-  ): void;
+  // (undocumented)
+  subscribe(options: {
+    onProcessingError: (event: {
+      unprocessedEntity: Entity;
+      errors: Error[];
+    }) => Promise<void> | void;
+  }): void;
 }
 
 // @alpha
@@ -212,16 +215,6 @@ export interface CatalogProcessingEngine {
   stop(): Promise<void>;
 }
 
-// @alpha
-export interface CatalogProcessingErrorListener {
-  // (undocumented)
-  onError(
-    unprocessedEntity: Entity,
-    result: EntityProcessingResult,
-    resultHash: String,
-  ): Promise<void>;
-}
-
 // @public (undocumented)
 export type CatalogProcessor = {
   getProcessorName(): string;

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

@@ -14,8 +14,8 @@
  * limitations under the License.
  */
 
-import { stringifyEntityRef } from '@backstage/catalog-model';
-import { assertError, serializeError } from '@backstage/errors';
+import { Entity, stringifyEntityRef } from '@backstage/catalog-model';
+import { assertError, serializeError, stringifyError } from '@backstage/errors';
 import { Hash } from 'crypto';
 import stableStringify from 'fast-json-stable-stringify';
 import { Logger } from 'winston';
@@ -23,7 +23,6 @@ import { ProcessingDatabase, RefreshStateItem } from '../database/types';
 import { createCounterMetric, createSummaryMetric } from '../util/metrics';
 import {
   CatalogProcessingEngine,
-  CatalogProcessingErrorListener,
   CatalogProcessingOrchestrator,
   EntityProcessingResult,
 } from './types';
@@ -43,7 +42,10 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
     private readonly stitcher: Stitcher,
     private readonly createHash: () => Hash,
     private readonly pollingIntervalMs: number = 1000,
-    private readonly catalogProcessingErrorListener?: CatalogProcessingErrorListener,
+    private readonly onProcessingError?: (event: {
+      unprocessedEntity: Entity;
+      errors: Error[];
+    }) => Promise<void> | void,
   ) {}
 
   async start() {
@@ -158,11 +160,20 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
           // the outside.
           if (!result.ok) {
             // notify the error listener if the entity can not be processed.
-            this.catalogProcessingErrorListener?.onError(
-              unprocessedEntity,
-              result,
-              resultHash,
-            );
+            Promise.resolve(undefined)
+              .then(() =>
+                this.onProcessingError?.({
+                  unprocessedEntity,
+                  errors: result.errors,
+                }),
+              )
+              .catch(error => {
+                this.logger.debug(
+                  `Processing error listener threw an exception, ${stringifyError(
+                    error,
+                  )}`,
+                );
+              });
 
             await this.processingDatabase.transaction(async tx => {
               await this.processingDatabase.updateProcessedEntityErrors(tx, {

plugins/catalog-backend/src/processing/index.ts

@@ -18,7 +18,6 @@ export type {
   CatalogProcessingEngine,
   EntityProcessingResult,
   DeferredEntity,
-  CatalogProcessingErrorListener,
 } from './types';
 
 export { createRandomProcessingInterval } from './refresh';

plugins/catalog-backend/src/processing/types.ts

@@ -67,16 +67,3 @@ export interface CatalogProcessingEngine {
   start(): Promise<void>;
   stop(): Promise<void>;
 }
-
-/**
- * An error listener for catalog processing engine. It can be used to listen and track entity errors.
- *
- * @alpha
- */
-export interface CatalogProcessingErrorListener {
-  onError(
-    unprocessedEntity: Entity,
-    result: EntityProcessingResult,
-    resultHash: String,
-  ): Promise<void>;
-}

plugins/catalog-backend/src/service/CatalogBuilder.ts

@@ -17,6 +17,7 @@
 import { PluginDatabaseManager, UrlReader } from '@backstage/backend-common';
 import {
   DefaultNamespaceEntityPolicy,
+  Entity,
   EntityPolicies,
   EntityPolicy,
   FieldFormatEntityPolicy,
@@ -56,10 +57,7 @@ import {
 } from '../modules/core/PlaceholderProcessor';
 import { defaultEntityDataParser } from '../modules/util/parse';
 import { LocationAnalyzer } from '../ingestion/types';
-import {
-  CatalogProcessingEngine,
-  CatalogProcessingErrorListener,
-} from '../processing';
+import { CatalogProcessingEngine } from '../processing';
 import { DefaultProcessingDatabase } from '../database/DefaultProcessingDatabase';
 import { applyDatabaseMigrations } from '../database/migrations';
 import { DefaultCatalogProcessingEngine } from '../processing/DefaultCatalogProcessingEngine';
@@ -136,7 +134,10 @@ export class CatalogBuilder {
   private processors: CatalogProcessor[];
   private processorsReplace: boolean;
   private parser: CatalogProcessorParser | undefined;
-  private catalogProcessingErrorListeners?: CatalogProcessingErrorListener[];
+  private onProcessingError?: (event: {
+    unprocessedEntity: Entity;
+    errors: Error[];
+  }) => Promise<void> | void;
   private processingInterval: ProcessingIntervalFunction =
     createRandomProcessingInterval({
       minSeconds: 100,
@@ -452,12 +453,8 @@ export class CatalogBuilder {
       stitcher,
       () => createHash('sha1'),
       1000,
-      {
-        onError: async (unprocessedEntity, result, resultHash) => {
-          this.catalogProcessingErrorListeners?.forEach(listener =>
-            listener.onError(unprocessedEntity, result, resultHash),
-          );
-        },
+      event => {
+        this.onProcessingError?.(event);
       },
     );
 
@@ -490,12 +487,13 @@ export class CatalogBuilder {
     };
   }
 
-  /**
-   * @alpha
-   * @param catalogProcessingErrorListeners - a list of listeners to get notified if an error occurs while processing an entity
-   */
-  subscribe(catalogProcessingErrorListeners: CatalogProcessingErrorListener[]) {
-    this.catalogProcessingErrorListeners = catalogProcessingErrorListeners;
+  subscribe(options: {
+    onProcessingError: (event: {
+      unprocessedEntity: Entity;
+      errors: Error[];
+    }) => Promise<void> | void;
+  }) {
+    this.onProcessingError = options.onProcessingError;
   }
 
   private buildEntityPolicy(): EntityPolicy {