Merge pull request #18216 from thefrontside/incremental-ingestion-async-event-hook

✨Allow incremental ingestion event handler to be async

.changeset/rotten-colts-decide.md

@@ -0,0 +1,38 @@
+---
+'@backstage/plugin-catalog-backend-module-incremental-ingestion': minor
+---
+
+**BREAKING** Allow incremental event handlers to be async; Force event handler
+to indicate if it made a change. Instead of returning `null` or `undefined` from an event
+handler to indicate no-oop, instead return the value { type: "ignored" }.
+
+**before**
+
+```javascript
+import { createDelta, shouldIgnore } from "./my-delta-creater";
+
+eventHandler: {
+  onEvent(params) {
+    if (shouldIgnore(params)) {
+      return;
+    }
+    return createDelta(params);
+  }
+}
+```
+
+**after**
+
+```javascript
+import { createDelta, shouldIgnore } from "./my-delta-creater";
+
+eventHandler: {
+  async onEvent(params) {
+    if (shouldIgnore(params) {
+      return { type: "ignored" };
+    }
+    // code to create delta can now be async if needed
+    return await createDelta(params);
+  }
+}
+```

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.