From 40a40c2fb2d02d0ead326d48f2811a5471fec58c Mon Sep 17 00:00:00 2001
From: Johan Haals <johan.haals@gmail.com>
Date: Mon, 13 Sep 2021 13:08:55 +0200
Subject: [PATCH] Implement refresh for nested entities

Co-authored-by: Patrik Oldsberg <poldsberg@gmail.com>
Signed-off-by: Johan Haals <johan.haals@gmail.com>
---
 .../database/DefaultProcessingDatabase.ts     | 69 ++++++++++++-------
 1 file changed, 46 insertions(+), 23 deletions(-)

diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts
index 369deb36a4..58a384457d 100644
--- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts
+++ b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts
@@ -50,6 +50,7 @@ import {
 // errors in the underlying engine due to exceeding query limits, but large
 // enough to get the speed benefits.
 const BATCH_SIZE = 50;
+const MAX_REFRESH_DEPTH = 10;
 
 export class DefaultProcessingDatabase implements ProcessingDatabase {
   constructor(
@@ -522,33 +523,55 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
     const tx = txOpaque as Knex.Transaction;
     if ('entityRef' in options) {
       const { entityRef } = options;
-      const [result] = await tx<DbRefreshStateRow>('refresh_state')
-        .where({ entity_ref: entityRef })
-        .select();
 
-      if (!result) {
-        throw new NotFoundError(`EntityRef ${entityRef} not found`);
-      }
+      let refreshTarget = entityRef;
 
-      const refs = await tx<DbRefreshStateReferencesRow>(
-        'refresh_state_references',
-      )
-        .where({ target_entity_ref: entityRef })
-        .select();
-
-      for (const ref of refs) {
-        if (ref.source_entity_ref?.startsWith('location:')) {
-          const updateResult = await tx<DbRefreshStateRow>('refresh_state')
-            .where({ entity_ref: ref.source_entity_ref })
-            .update({ next_update_at: tx.fn.now() });
-
-          if (updateResult === 0) {
-            throw new ConflictError(
-              `Failed to schedule ${ref.source_entity_ref} for refresh`,
-            );
-          }
+      let currentRef = entityRef;
+      let depth = 0;
+      for (;;) {
+        if (depth++ > MAX_REFRESH_DEPTH) {
+          throw new Error(
+            `Unable to refresh Entity ${entityRef}, maximum refresh depth of ${MAX_REFRESH_DEPTH} reached`,
+          );
         }
+
+        const rows = await tx<DbRefreshStateReferencesRow>(
+          'refresh_state_references',
+        )
+          .where({ target_entity_ref: currentRef })
+          .select();
+
+        if (rows.length === 0) {
+          if (depth === 1) {
+            throw new NotFoundError(`Entity ${currentRef} not found`);
+          }
+          throw new NotFoundError(
+            `Entity ${entityRef} has a broken parent reference chain at ${currentRef}`,
+          );
+        }
+
+        const parentRef = rows[0].source_entity_ref;
+        if (!parentRef) {
+          // We've reached the top of the tree which is the entityProvider.
+          // In this case we refresh the entity itself.
+          break;
+        }
+        if (parentRef.startsWith('location:')) {
+          refreshTarget = parentRef;
+          break;
+        }
+        currentRef = parentRef;
       }
+
+      const updateResult = await tx<DbRefreshStateRow>('refresh_state')
+        .where({ entity_ref: refreshTarget })
+        .update({ next_update_at: tx.fn.now() });
+      if (updateResult === 0) {
+        throw new ConflictError(
+          `Failed to schedule ${refreshTarget} for refresh`,
+        );
+      }
+
       /*
         For a given entityRef:
         - Fetch entity