add a potential implementation for refreshKeys

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

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

@@ -0,0 +1,41 @@
+/*
+ * Copyright 2022 The Backstage Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+exports.up = async function up(knex) {
+  await knex.schema.createTable('refresh_keys', table => {
+    table.comment(
+      'This table contains relations between entities and keys to trigger refreshes with',
+    );
+    table
+      .text('entity_ref')
+      .notNullable()
+      .comment('A reference to the entity that the refresh key is tied to');
+    table
+      .text('key')
+      .notNullable()
+      .comment(
+        'A reference to a key which should be used to trigger a refresh on this entity',
+      );
+    table.unique(['entity_ref', 'key']);
+  });
+};
+
+/**
+ * @param { import("knex").Knex } knex
+ */
+exports.down = async function down(knex) {
+  await knex.schema.dropTable('refresh_keys');
+};

plugins/catalog-backend/src/api/processor.ts

@@ -169,9 +169,17 @@ export type CatalogProcessorErrorResult = {
   location: LocationSpec;
 };
 
+/** @public */
+export type CatalogProcessorRefreshKeysResult = {
+  type: 'refresh';
+  entity: Entity;
+  key: String;
+};
+
 /** @public */
 export type CatalogProcessorResult =
   | CatalogProcessorLocationResult
   | CatalogProcessorEntityResult
   | CatalogProcessorRelationResult
-  | CatalogProcessorErrorResult;
+  | CatalogProcessorErrorResult
+  | CatalogProcessorRefreshKeysResult;

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

@@ -33,12 +33,15 @@ import {
   UpdateEntityCacheOptions,
   ListParentsOptions,
   ListParentsResult,
+  RefreshKeyOptions,
+  RefreshByKeyOptions,
 } from './types';
 import { DeferredEntity } from '../processing/types';
 import { ProcessingIntervalFunction } from '../processing/refresh';
 import { rethrowError, timestampToDateTime } from './conversion';
 import { initDatabaseMetrics } from './metrics';
 import {
+  DbRefreshKeysRow,
   DbRefreshStateReferencesRow,
   DbRefreshStateRow,
   DbRelationsRow,
@@ -515,7 +518,40 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
       throw new NotFoundError(`Failed to schedule ${entityRef} for refresh`);
     }
   }
+  async refreshByRefreshKey(
+    txOpaque: Transaction,
+    options: RefreshByKeyOptions,
+  ) {
+    const tx = txOpaque as Knex.Transaction;
+    const { key } = options;
 
+    const rows = await tx<DbRefreshKeysRow>('refresh_keys')
+      .where({ key })
+      .select({
+        entity_ref: 'refresh_keys.entity_ref',
+      });
+
+    await Promise.all(rows.map(r => this.refresh(tx, r.entity_ref)));
+  }
+  async addRefreshKeys(
+    txOpaque: Transaction,
+    options: RefreshKeyOptions,
+  ): Promise<void> {
+    const tx = txOpaque as Knex.Transaction;
+    const { keys } = options;
+
+    await Promise.all(
+      keys.map(k => {
+        return tx<DbRefreshKeysRow>('refresh_keys')
+          .insert({
+            entity_ref: stringifyEntityRef(k.entity),
+            key: k.key,
+          })
+          .onConflict(['entity_ref', 'key'])
+          .ignore();
+      }),
+    );
+  }
   async transaction<T>(fn: (tx: Transaction) => Promise<T>): Promise<T> {
     try {
       let result: T | undefined = undefined;

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

@@ -43,6 +43,11 @@ export type DbRefreshStateRow = {
   location_key?: string;
 };
 
+export type DbRefreshKeysRow = {
+  entity_ref: string;
+  key: string;
+};
+
 export type DbRefreshStateReferencesRow = {
   source_key?: string;
   source_entity_ref?: string;

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

@@ -81,6 +81,14 @@ export type ReplaceUnprocessedEntitiesOptions =
       type: 'delta';
     };
 
+export type RefreshKeyOptions = {
+  keys: { key: String; entity: Entity }[];
+};
+
+export type RefreshByKeyOptions = {
+  key: string;
+};
+
 export type RefreshOptions = {
   entityRef: string;
 };
@@ -149,6 +157,11 @@ export interface ProcessingDatabase {
    */
   refresh(txOpaque: Transaction, options: RefreshOptions): Promise<void>;
 
+  addRefreshKeys(
+    txOpaque: Transaction,
+    options: RefreshKeyOptions,
+  ): Promise<void>;
+
   /**
    * Lists all ancestors of a given entityRef.
    *

plugins/catalog-backend/src/modules/core/FileReaderProcessor.ts

@@ -61,6 +61,13 @@ export class FileReaderProcessor implements CatalogProcessor {
             },
           })) {
             emit(parseResult);
+            if (parseResult.type === 'entity') {
+              emit({
+                type: 'refresh',
+                key: path.normalize(fileMatch),
+                entity: parseResult.entity,
+              });
+            }
           }
         }
       } else if (!optional) {

plugins/catalog-backend/src/modules/core/PlaceholderProcessor.ts

@@ -19,7 +19,11 @@ import { Entity } from '@backstage/catalog-model';
 import { JsonValue } from '@backstage/types';
 import { ScmIntegrationRegistry } from '@backstage/integration';
 import yaml from 'yaml';
-import { CatalogProcessor, LocationSpec } from '../../api';
+import {
+  CatalogProcessor,
+  CatalogProcessorEmit,
+  LocationSpec,
+} from '../../api';
 
 /** @public */
 export type PlaceholderResolverRead = (url: string) => Promise<Buffer>;
@@ -66,6 +70,7 @@ export class PlaceholderProcessor implements CatalogProcessor {
   async preProcessEntity(
     entity: Entity,
     location: LocationSpec,
+    emit: CatalogProcessorEmit,
   ): Promise<Entity> {
     const process = async (data: any): Promise<[any, boolean]> => {
       if (!data || !(data instanceof Object)) {
@@ -102,6 +107,7 @@ export class PlaceholderProcessor implements CatalogProcessor {
 
       const resolverKey = keys[0].substr(1);
       const resolverValue = data[keys[0]];
+      emit({ type: 'refresh', key: resolverValue, entity });
       const resolver = this.options.resolvers[resolverKey];
       if (!resolver || typeof resolverValue !== 'string') {
         // If there was no such placeholder resolver or if the value was not a

plugins/catalog-backend/src/modules/core/UrlReaderProcessor.ts

@@ -83,6 +83,13 @@ export class UrlReaderProcessor implements CatalogProcessor {
         })) {
           parseResults.push(parseResult);
           emit(parseResult);
+          if (parseResult.type === 'entity') {
+            emit({
+              type: 'refresh',
+              key: item.url,
+              entity: parseResult.entity,
+            });
+          }
         }
       }
 

plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts

@@ -123,6 +123,12 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
 
           let hashBuilder = this.createHash().update(errorsString);
           if (result.ok) {
+            await this.processingDatabase.transaction(tx =>
+              this.processingDatabase.addRefreshKeys(tx, {
+                keys: result.refreshKeys,
+              }),
+            );
+
             const { entityRefs: parents } =
               await this.processingDatabase.transaction(tx =>
                 this.processingDatabase.listParents(tx, {

plugins/catalog-backend/src/processing/ProcessorOutputCollector.ts

@@ -38,6 +38,10 @@ export class ProcessorOutputCollector {
   private readonly errors = new Array<Error>();
   private readonly relations = new Array<EntityRelationSpec>();
   private readonly deferredEntities = new Array<DeferredEntity>();
+  private readonly refreshKeys = new Array<{
+    key: String;
+    entity: Entity;
+  }>();
   private done = false;
 
   constructor(
@@ -54,6 +58,7 @@ export class ProcessorOutputCollector {
     return {
       errors: this.errors,
       relations: this.relations,
+      refreshKeys: this.refreshKeys,
       deferredEntities: this.deferredEntities,
     };
   }
@@ -116,6 +121,8 @@ export class ProcessorOutputCollector {
       this.relations.push(i.relation);
     } else if (i.type === 'error') {
       this.errors.push(i.error);
+    } else if (i.type === 'refresh') {
+      this.refreshKeys.push({ key: i.key, entity: i.entity });
     }
   }
 }

plugins/catalog-backend/src/processing/types.ts

@@ -26,7 +26,6 @@ export type EntityProcessingRequest = {
   entity: Entity;
   state?: JsonObject; // Versions for multiple deployments etc
 };
-
 /**
  * The result of processing an entity.
  * @public
@@ -38,6 +37,7 @@ export type EntityProcessingResult =
       completedEntity: Entity;
       deferredEntities: DeferredEntity[];
       relations: EntityRelationSpec[];
+      refreshKeys: { key: String; entity: Entity }[];
       errors: Error[];
     }
   | {