add test for storing refreshKeys

Signed-off-by: Kiss Miklos <miklos@roadie.io>

plugins/catalog-backend/migrations/20220616202842_refresh_keys.js

@@ -25,7 +25,7 @@ exports.up = async function up(knex) {
     table
       .text('entity_ref')
       .notNullable()
-      .references('entity_id')
+      .references('entity_ref')
       .inTable('refresh_state')
       .onDelete('CASCADE')
       .comment('A reference to the entity that the refresh key is tied to');
@@ -35,6 +35,7 @@ exports.up = async function up(knex) {
       .comment(
         'A reference to a key which should be used to trigger a refresh on this entity',
       );
+    table.unique(['entity_ref', 'key']);
     table.index('entity_ref', 'refresh_keys_entity_ref_idx');
     table.index('key', 'refresh_keys_key_idx');
   });

plugins/catalog-backend/src/database/DefaultProcessingDatabase.test.ts

@@ -24,6 +24,7 @@ import { DateTime } from 'luxon';
 import { applyDatabaseMigrations } from './migrations';
 import { DefaultProcessingDatabase } from './DefaultProcessingDatabase';
 import {
+  DbRefreshKeysRow,
   DbRefreshStateReferencesRow,
   DbRefreshStateRow,
   DbRelationsRow,
@@ -473,6 +474,63 @@ describe('Default Processing Database', () => {
       },
       60_000,
     );
+
+    it.each(databases.eachSupportedId())(
+      'stores the refresh keys for the entity',
+      async databaseId => {
+        const mockLogger = {
+          debug: jest.fn(),
+          error: jest.fn(),
+          warn: jest.fn(),
+        };
+        const { knex, db } = await createDatabase(
+          databaseId,
+          mockLogger as unknown as Logger,
+        );
+        await insertRefreshStateRow(knex, {
+          entity_id: id,
+          entity_ref: 'location:default/fakelocation',
+          unprocessed_entity: '{}',
+          processed_entity: '{}',
+          errors: '[]',
+          next_update_at: '2021-04-01 13:37:00',
+          last_discovery_at: '2021-04-01 13:37:00',
+        });
+
+        const deferredEntities = [
+          {
+            entity: {
+              apiVersion: '1',
+              kind: 'Location',
+              metadata: {
+                name: 'next',
+              },
+            },
+            locationKey: 'mock',
+          },
+        ];
+
+        await db.transaction(tx =>
+          db.updateProcessedEntity(tx, {
+            id,
+            processedEntity,
+            resultHash: '',
+            relations: [],
+            deferredEntities,
+            refreshKeys: [{ key: 'protocol:foo-bar.com' }],
+          }),
+        );
+
+        const refreshKeys = await knex<DbRefreshKeysRow>('refresh_keys')
+          .where({ entity_ref: stringifyEntityRef(processedEntity) })
+          .select();
+
+        expect(refreshKeys[0]).toEqual({
+          entity_ref: 'location:default/fakelocation',
+          key: 'protocol:foo-bar.com',
+        });
+      },
+    );
   });
 
   describe('updateEntityCache', () => {

plugins/catalog-backend/src/database/DefaultProcessingDatabase.ts

@@ -141,12 +141,24 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
       BATCH_SIZE,
     );
 
-    // Insert the refresh keys for the procssed entity
+    // Find the top-level location entity that manages the processedEntity
+    let entityRefToRefresh = sourceEntityRef;
+    const { entityRefs } = await this.listAncestors(tx, {
+      entityRef: sourceEntityRef,
+    });
+    const locationAncestor = entityRefs.find(ref =>
+      ref.startsWith('location:'),
+    );
+    if (locationAncestor) {
+      entityRefToRefresh = locationAncestor;
+    }
+
+    // Insert the refresh keys for the processed entity
     await Promise.all(
       options.refreshKeys.map(k => {
         return tx<DbRefreshKeysRow>('refresh_keys')
           .insert({
-            entity_ref: sourceEntityRef,
+            entity_ref: entityRefToRefresh,
             key: k.key,
           })
           .onConflict(['entity_ref', 'key'])