Synchronize ingestion_mark_entities table with deltas

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

plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.ts

@@ -269,6 +269,17 @@ export class IncrementalIngestionDatabaseManager {
     });
   }
 
+  /**
+   * This method is used to remove entity records from the ingestion_mark_entities
+   * table by their entity reference.
+   */
+  async deleteEntityRecordsByRef(entities: { entityRef: string }[]) {
+    const refs = entities.map(e => e.entityRef);
+    await this.client.transaction(async tx => {
+      await tx('ingestion_mark_entities').delete().whereIn('ref', refs);
+    });
+  }
+
   /**
    * Creates a new ingestion record.
    * @param provider - string
@@ -535,6 +546,21 @@ export class IncrementalIngestionDatabaseManager {
     });
   }
 
+  /**
+   * Returns the first record from `ingestion_marks` for the supplied ingestionId.
+   * @param ingestionId - string
+   * @returns MarkRecord | undefined
+   */
+  async getFirstMark(ingestionId: string) {
+    return await this.client.transaction(async tx => {
+      const mark = await tx<MarkRecord>('ingestion_marks')
+        .where('ingestion_id', ingestionId)
+        .orderBy('sequence', 'asc')
+        .first();
+      return mark;
+    });
+  }
+
   async getAllMarks(ingestionId: string) {
     return await this.client.transaction(async tx => {
       const marks = await tx<MarkRecord>('ingestion_marks')

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

@@ -339,6 +339,7 @@ export class IncrementalIngestionEngine<TInput>
     }
 
     const { logger, provider, connection } = this.options;
+    const providerName = provider.getProviderName();
     logger.info(
       `incremental-engine: Received ${this.providerEventTopic} event`,
     );
@@ -352,6 +353,34 @@ export class IncrementalIngestionEngine<TInput>
     const update = provider.deltaMapper(payload);
 
     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.`,
+          );
+        } 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);
+        }
+      }
+
+      if (update.delta.removed.length > 0) {
+        await this.manager.deleteEntityRecordsByRef(update.delta.removed);
+      }
+
       await connection.applyMutation({
         type: 'delta',
         ...update.delta,

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

@@ -81,7 +81,12 @@ export interface IncrementalEntityProvider<TCursor, TContext, TInput = null> {
    * outside of the incremental ingestion schedule.
    */
   deltaMapper?: (payload: TInput) => {
-    delta: { added: DeferredEntity[]; removed: DeferredEntity[] } | undefined;
+    delta:
+      | {
+          added: DeferredEntity[];
+          removed: { entityRef: string }[];
+        }
+      | undefined;
   };
 }