set the refreshKeys in the updateProcessedEntity function

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

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

@@ -66,7 +66,7 @@ export const processingResult = Object.freeze({
     return { type: 'relation', relation: spec };
   },
 
-  refresh(entityRef: string, key: string): CatalogProcessorResult {
-    return { type: 'refresh', entityRef, key };
+  refresh(key: string): CatalogProcessorResult {
+    return { type: 'refresh', key };
   },
 } as const);

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

@@ -172,7 +172,6 @@ export type CatalogProcessorErrorResult = {
 /** @public */
 export type CatalogProcessorRefreshKeysResult = {
   type: 'refresh';
-  entityRef: string;
   key: string;
 };
 

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

@@ -103,11 +103,12 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
         `Conflicting write of processing result for ${id} with location key '${locationKey}'`,
       );
     }
+    const sourceEntityRef = stringifyEntityRef(processedEntity);
 
     // Schedule all deferred entities for future processing.
     await this.addUnprocessedEntities(tx, {
       entities: deferredEntities,
-      sourceEntityRef: stringifyEntityRef(processedEntity),
+      sourceEntityRef,
     });
 
     // Delete old relations
@@ -141,6 +142,19 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
       BATCH_SIZE,
     );
 
+    // Insert the refresh keys for the procssed entity
+    await Promise.all(
+      options.refreshKeys.map(k => {
+        return tx<DbRefreshKeysRow>('refresh_keys')
+          .insert({
+            entity_ref: sourceEntityRef,
+            key: k.key,
+          })
+          .onConflict(['entity_ref', 'key'])
+          .ignore();
+      }),
+    );
+
     return {
       previous: {
         relations: previousRelationRows,
@@ -526,41 +540,22 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
     const tx = txOpaque as Knex.Transaction;
     const { keys } = options;
 
-    const query = await tx<DbRefreshStateRow>('refresh_state')
+    const updateResult = await tx<DbRefreshStateRow>('refresh_state')
       .whereIn('entity_ref', function (tx2) {
         tx2
           .whereIn('key', keys)
           .select({
             entity_ref: 'refresh_keys.entity_ref',
           })
-          .from('refresh_keys')
-          .columns('entity_ref');
+          .from('refresh_keys');
       })
-      .update({ next_update_at: tx.fn.now() })
-      .toSQL()
-      .toNative();
+      .update({ next_update_at: tx.fn.now() });
 
-    console.log(query, '@@@@@@@!!!!!!!@@@@@');
-  }
-
-  async setRefreshKeys(
-    txOpaque: Transaction,
-    options: RefreshKeyOptions,
-  ): Promise<void> {
-    const tx = txOpaque as Knex.Transaction;
-    const { refreshKeys } = options;
-
-    await Promise.all(
-      refreshKeys.map(k => {
-        return tx<DbRefreshKeysRow>('refresh_keys')
-          .insert({
-            entity_ref: k.entityRef,
-            key: k.key,
-          })
-          .onConflict(['entity_ref', 'key'])
-          .ignore();
-      }),
-    );
+    if (updateResult === 0) {
+      throw new NotFoundError(
+        `Failed to schedule ${JSON.stringify(keys)} for keys`,
+      );
+    }
   }
 
   async transaction<T>(fn: (tx: Transaction) => Promise<T>): Promise<T> {

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

@@ -38,6 +38,7 @@ export type UpdateProcessedEntityOptions = {
   relations: EntityRelationSpec[];
   deferredEntities: DeferredEntity[];
   locationKey?: string;
+  refreshKeys: RefreshKeyData[];
 };
 
 export type UpdateEntityCacheOptions = {
@@ -157,14 +158,6 @@ export interface ProcessingDatabase {
    */
   refresh(txOpaque: Transaction, options: RefreshOptions): Promise<void>;
 
-  /**
-   * Schedules a refresh for all the entities that have the given refreshKey
-   */
-  setRefreshKeys(
-    txOpaque: Transaction,
-    options: RefreshKeyOptions,
-  ): Promise<void>;
-
   /**
    * Lists all ancestors of a given entityRef.
    *

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

@@ -29,6 +29,8 @@ import { stringifyEntityRef } from '@backstage/catalog-model';
 
 const glob = promisify(g);
 
+const LOCATION_TYPE = 'file';
+
 /** @public */
 export class FileReaderProcessor implements CatalogProcessor {
   getProcessorName(): string {
@@ -41,7 +43,7 @@ export class FileReaderProcessor implements CatalogProcessor {
     emit: CatalogProcessorEmit,
     parser: CatalogProcessorParser,
   ): Promise<boolean> {
-    if (location.type !== 'file') {
+    if (location.type !== LOCATION_TYPE) {
       return false;
     }
 
@@ -57,19 +59,16 @@ export class FileReaderProcessor implements CatalogProcessor {
           for await (const parseResult of parser({
             data: data,
             location: {
-              type: 'file',
+              type: LOCATION_TYPE,
               target: path.normalize(fileMatch),
             },
           })) {
             emit(parseResult);
-            if (parseResult.type === 'entity') {
-              emit(
-                processingResult.refresh(
-                  stringifyEntityRef(parseResult.entity),
-                  path.normalize(fileMatch),
-                ),
-              );
-            }
+            emit(
+              processingResult.refresh(
+                `${LOCATION_TYPE}:${path.normalize(fileMatch)}`,
+              ),
+            );
           }
         }
       } else if (!optional) {

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

@@ -134,7 +134,7 @@ export class PlaceholderProcessor implements CatalogProcessor {
           base,
         });
 
-      emit(processingResult.refresh(stringifyEntityRef(entity), resolverValue));
+      emit(processingResult.refresh(`url:${resolverValue}`));
 
       return [
         await resolver({

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

@@ -30,6 +30,7 @@ import {
   LocationSpec,
   processingResult,
 } from '../../api';
+import { locationSpecToLocationEntity } from '../../util';
 
 const CACHE_KEY = 'v1';
 
@@ -83,14 +84,6 @@ export class UrlReaderProcessor implements CatalogProcessor {
         })) {
           parseResults.push(parseResult);
           emit(parseResult);
-          if (parseResult.type === 'entity') {
-            emit(
-              processingResult.refresh(
-                stringifyEntityRef(parseResult.entity),
-                item.url,
-              ),
-            );
-          }
         }
       }
 
@@ -101,6 +94,8 @@ export class UrlReaderProcessor implements CatalogProcessor {
           value: parseResults as CatalogProcessorEntityResult[],
         });
       }
+
+      emit(processingResult.refresh(`${location.type}:${location.target}`));
     } catch (error) {
       assertError(error);
       const message = `Unable to read ${location.type}, ${error}`;

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

@@ -123,12 +123,6 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
 
           let hashBuilder = this.createHash().update(errorsString);
           if (result.ok) {
-            await this.processingDatabase.transaction(tx =>
-              this.processingDatabase.setRefreshKeys(tx, {
-                refreshKeys: result.refreshKeys,
-              }),
-            );
-
             const { entityRefs: parents } =
               await this.processingDatabase.transaction(tx =>
                 this.processingDatabase.listParents(tx, {
@@ -186,6 +180,7 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
                 relations: result.relations,
                 deferredEntities: result.deferredEntities,
                 locationKey,
+                refreshKeys: result.refreshKeys,
               });
             oldRelationSources = new Set(
               previous.relations.map(r => r.source_entity_ref),

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

@@ -51,7 +51,6 @@ export type EntityProcessingResult =
  */
 export type RefreshKeyData = {
   key: string;
-  entityRef: string;
 };
 
 /**