Better event handling

Signed-off-by: Damon Kaswell <damon.kaswell1@hp.com>

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

@@ -10,6 +10,7 @@ import { CatalogBuilder } from '@backstage/plugin-catalog-backend';
 import type { Config } from '@backstage/config';
 import type { DeferredEntity } from '@backstage/plugin-catalog-backend';
 import type { DurationObjectUnits } from 'luxon';
+import { EventParams } from '@backstage/plugin-events-node';
 import type { Logger } from 'winston';
 import type { PermissionEvaluator } from '@backstage/plugin-permission-common';
 import type { PluginDatabaseManager } from '@backstage/backend-common';
@@ -55,16 +56,18 @@ export interface IncrementalEntityProvider<TCursor, TContext> {
     context: TContext,
     cursor?: TCursor,
   ): Promise<EntityIteratorResult<TCursor>>;
-  onEvent?: (payload: unknown) => {
-    delta:
-      | {
-          added: DeferredEntity[];
-          removed: {
-            entityRef: string;
-          }[];
-        }
-      | undefined;
-  };
+  onEvent?: (params: EventParams) =>
+    | {
+        delta:
+          | {
+              added: DeferredEntity[];
+              removed: {
+                entityRef: string;
+              }[];
+            }
+          | undefined;
+      }
+    | undefined;
 }
 
 // @public (undocumented)

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

@@ -336,14 +336,14 @@ export class IncrementalIngestionEngine
   }
 
   async onEvent(params: EventParams): Promise<void> {
-    const { topic, eventPayload } = params;
+    const { topic } = params;
     if (topic !== this.providerEventTopic) {
       return;
     }
 
     const { logger, provider, connection } = this.options;
     const providerName = provider.getProviderName();
-    logger.info(
+    logger.debug(
       `incremental-engine: Received ${this.providerEventTopic} event`,
     );
 
@@ -351,48 +351,50 @@ export class IncrementalIngestionEngine
       return;
     }
 
-    const update = provider.onEvent(eventPayload);
+    const update = provider.onEvent(params);
 
-    if (update.delta) {
-      if (update.delta.added.length > 0) {
-        const ingestionRecord = await this.manager.getCurrentIngestionRecord(
-          providerName,
-        );
-
-        if (!ingestionRecord) {
-          logger.debug(
-            `incremental-engine: Skipping delta addition because incremental ingestion is restarting.`,
+    if (update) {
+      if (update.delta) {
+        if (update.delta.added.length > 0) {
+          const ingestionRecord = await this.manager.getCurrentIngestionRecord(
+            providerName,
           );
-        } else {
-          const mark =
-            ingestionRecord.status === 'resting'
-              ? await this.manager.getLastMark(ingestionRecord.id)
-              : await this.manager.getFirstMark(ingestionRecord.id);
 
-          if (!mark) {
-            throw new Error(
-              `Cannot apply delta, page records are missing! Please re-run incremental ingestion for ${providerName}.`,
+          if (!ingestionRecord) {
+            logger.debug(
+              `incremental-engine: Skipping delta addition because incremental ingestion is restarting.`,
             );
+          } else {
+            const mark =
+              ingestionRecord.status === 'resting'
+                ? await this.manager.getLastMark(ingestionRecord.id)
+                : await this.manager.getFirstMark(ingestionRecord.id);
+
+            if (!mark) {
+              throw new Error(
+                `Cannot apply delta, page records are missing! Please re-run incremental ingestion for ${providerName}.`,
+              );
+            }
+            await this.manager.createMarkEntities(mark.id, update.delta.added);
           }
-          await this.manager.createMarkEntities(mark.id, update.delta.added);
         }
-      }
 
-      if (update.delta.removed.length > 0) {
-        await this.manager.deleteEntityRecordsByRef(update.delta.removed);
-      }
+        if (update.delta.removed.length > 0) {
+          await this.manager.deleteEntityRecordsByRef(update.delta.removed);
+        }
 
-      await connection.applyMutation({
-        type: 'delta',
-        ...update.delta,
-      });
-      logger.info(
-        `incremental-engine: Processed ${this.providerEventTopic} event`,
-      );
-    } else {
-      logger.warn(
-        `incremental-engine: Rejected ${this.providerEventTopic} event - empty or invalid`,
-      );
+        await connection.applyMutation({
+          type: 'delta',
+          ...update.delta,
+        });
+        logger.debug(
+          `incremental-engine: Processed ${this.providerEventTopic} event`,
+        );
+      } else {
+        logger.warn(
+          `incremental-engine: Rejected ${this.providerEventTopic} event - empty or invalid`,
+        );
+      }
     }
   }
 

plugins/catalog-backend-module-incremental-ingestion/src/router/routes.ts

@@ -239,7 +239,7 @@ export class IncrementalProviderRouter implements EventPublisher {
       });
     });
 
-    router.post(`${PROVIDER_BASE_PATH}/delta`, async (req, res) => {
+    router.post(`${PROVIDER_BASE_PATH}/event`, async (req, res) => {
       const { provider } = req.params;
 
       const topic = `${provider}-push`;

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

@@ -27,6 +27,7 @@ import type {
   DeferredEntity,
   EntityProviderConnection,
 } from '@backstage/plugin-catalog-backend';
+import { EventParams } from '@backstage/plugin-events-node';
 import type { PermissionEvaluator } from '@backstage/plugin-permission-common';
 import type { DurationObjectUnits } from 'luxon';
 import type { Logger } from 'winston';
@@ -77,20 +78,23 @@ export interface IncrementalEntityProvider<TCursor, TContext> {
   around(burst: (context: TContext) => Promise<void>): Promise<void>;
 
   /**
-   * If present, this method accepts an incoming event, and maps the
-   * payload to an object containing a delta mutation.
+   * This method accepts an incoming event for the provider, and
+   * (optionally) maps the payload to an object containing a delta
+   * mutation.
    *
-   * If a delta is present, the incremental entity provider will apply
-   * it automatically.
+   * If a valid delta is returned by this method, it will be ingested
+   * automatically by the provider.
    */
-  onEvent?: (payload: unknown) => {
-    delta:
-      | {
-          added: DeferredEntity[];
-          removed: { entityRef: string }[];
-        }
-      | undefined;
-  };
+  onEvent?: (params: EventParams) =>
+    | {
+        delta:
+          | {
+              added: DeferredEntity[];
+              removed: { entityRef: string }[];
+            }
+          | undefined;
+      }
+    | undefined;
 }
 
 /**