✨Make incremental event handler to be both explicity and async

The point of the event hook is to map incoming changes into catalog
deltas. However, this is not always a synchronous process, and might
require accessing other resources to get the actual delta.

This just converts the hook into an async function and awaits it on
the other end. This also cleans up the API a little bit by not relying
on `undefined` to indicate that the event was ignored, but by
explicitly forcing the event handler to indicate that the event is to
be ignore by returning { type: "ignored" }.

Since ignoring events is a perfectly acceptable outcome, the warning
if the event was dropped is just ignored. If there is an error, it
should be propagated as an exception.

This is purely an aesthetic change designed to communicate via usage
how to use the API.

Signed-off-by: Charles Lowell <cowboyd@frontside.com>

.changeset/rotten-colts-decide.md

@@ -0,0 +1,5 @@
+---
+'@backstage/plugin-catalog-backend-module-incremental-ingestion': minor
+---
+
+Allow incremental event handlers to be async; Force event handler to indicate if it made a change

plugins/catalog-backend-module-incremental-ingestion/api-report.md

@@ -48,18 +48,24 @@ export class IncrementalCatalogBuilder {
   ): Promise<IncrementalCatalogBuilder>;
 }
 
+// @public
+export type IncrementalEntityEventResult =
+  | {
+      type: 'ignored';
+    }
+  | {
+      type: 'delta';
+      added: DeferredEntity[];
+      removed: {
+        entityRef: string;
+      }[];
+    };
+
 // @public
 export interface IncrementalEntityProvider<TCursor, TContext> {
   around(burst: (context: TContext) => Promise<void>): Promise<void>;
   eventHandler?: {
-    onEvent: (params: EventParams) =>
-      | undefined
-      | {
-          added: DeferredEntity[];
-          removed: {
-            entityRef: string;
-          }[];
-        };
+    onEvent: (params: EventParams) => Promise<IncrementalEntityEventResult>;
     supportsEventTopics: () => string[];
   };
   getProviderName(): string;

plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts

@@ -347,10 +347,10 @@ export class IncrementalIngestionEngine
       return;
     }
 
-    const delta = provider.eventHandler.onEvent(params);
+    const result = await provider.eventHandler.onEvent(params);
 
-    if (delta) {
-      if (delta.added.length > 0) {
+    if (result.type === 'delta') {
+      if (result.added.length > 0) {
         const ingestionRecord = await this.manager.getCurrentIngestionRecord(
           providerName,
         );
@@ -370,24 +370,21 @@ export class IncrementalIngestionEngine
               `Cannot apply delta, page records are missing! Please re-run incremental ingestion for ${providerName}.`,
             );
           }
-          await this.manager.createMarkEntities(mark.id, delta.added);
+          await this.manager.createMarkEntities(mark.id, result.added);
         }
       }
 
-      if (delta.removed.length > 0) {
-        await this.manager.deleteEntityRecordsByRef(delta.removed);
+      if (result.removed.length > 0) {
+        await this.manager.deleteEntityRecordsByRef(result.removed);
       }
 
-      await connection.applyMutation({
-        type: 'delta',
-        ...delta,
-      });
+      await connection.applyMutation(result);
       logger.debug(
         `incremental-engine: ${providerName} processed delta from '${topic}' event`,
       );
     } else {
-      logger.warn(
-        `incremental-engine: Rejected delta from '${topic}' event - empty or invalid`,
+      logger.debug(
+        `incremental-engine: ${providerName} ignored event from topic '${topic}'`,
       );
     }
   }

plugins/catalog-backend-module-incremental-ingestion/src/index.ts

@@ -23,6 +23,7 @@
 export * from './service';
 export {
   type EntityIteratorResult,
+  type IncrementalEntityEventResult,
   type IncrementalEntityProvider,
   type IncrementalEntityProviderOptions,
   type PluginEnvironment,

plugins/catalog-backend-module-incremental-ingestion/src/types.ts

@@ -91,15 +91,12 @@ export interface IncrementalEntityProvider<TCursor, TContext> {
      * optionally maps the payload to an object containing a delta
      * mutation.
      *
-     * If a valid delta is returned by this method, it will be ingested
-     * automatically by the provider.
+     * If a delta result is returned by this method, it will be ingested
+     * automatically by the provider. Alternatively, if an "ignored" result is
+     * returned, then it is understood that this event should not cause anything
+     * to be ingested.
      */
-    onEvent: (params: EventParams) =>
-      | undefined
-      | {
-          added: DeferredEntity[];
-          removed: { entityRef: string }[];
-        };
+    onEvent: (params: EventParams) => Promise<IncrementalEntityEventResult>;
 
     /**
      * This method returns an array of topics for the IncrementalEntityProvider
@@ -109,6 +106,22 @@ export interface IncrementalEntityProvider<TCursor, TContext> {
   };
 }
 
+/**
+ * An object returned by event handler to indicate whether to ignore the event
+ * or to apply a delta in response to the event.
+ *
+ * @public
+ */
+export type IncrementalEntityEventResult =
+  | {
+      type: 'ignored';
+    }
+  | {
+      type: 'delta';
+      added: DeferredEntity[];
+      removed: { entityRef: string }[];
+    };
+
 /**
  * Value returned by an {@link IncrementalEntityProvider} to provide a
  * single page of entities to ingest.