diff --git a/plugins/catalog-backend/src/api/processingResult.ts b/plugins/catalog-backend/src/api/processingResult.ts index 4eefca196c..84f1f4b70d 100644 --- a/plugins/catalog-backend/src/api/processingResult.ts +++ b/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); diff --git a/plugins/catalog-backend/src/api/processor.ts b/plugins/catalog-backend/src/api/processor.ts index 196391cf50..44e8b298b0 100644 --- a/plugins/catalog-backend/src/api/processor.ts +++ b/plugins/catalog-backend/src/api/processor.ts @@ -172,7 +172,6 @@ export type CatalogProcessorErrorResult = { /** @public */ export type CatalogProcessorRefreshKeysResult = { type: 'refresh'; - entityRef: string; key: string; }; diff --git a/plugins/catalog-backend/src/database/DefaultProcessingDatabase.ts b/plugins/catalog-backend/src/database/DefaultProcessingDatabase.ts index 3b93e9a222..b8d7493360 100644 --- a/plugins/catalog-backend/src/database/DefaultProcessingDatabase.ts +++ b/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('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('refresh_state') + const updateResult = await tx('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 { - const tx = txOpaque as Knex.Transaction; - const { refreshKeys } = options; - - await Promise.all( - refreshKeys.map(k => { - return tx('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(fn: (tx: Transaction) => Promise): Promise { diff --git a/plugins/catalog-backend/src/database/types.ts b/plugins/catalog-backend/src/database/types.ts index 4d2b64a721..da7aa65714 100644 --- a/plugins/catalog-backend/src/database/types.ts +++ b/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; - /** - * Schedules a refresh for all the entities that have the given refreshKey - */ - setRefreshKeys( - txOpaque: Transaction, - options: RefreshKeyOptions, - ): Promise; - /** * Lists all ancestors of a given entityRef. * diff --git a/plugins/catalog-backend/src/modules/core/FileReaderProcessor.ts b/plugins/catalog-backend/src/modules/core/FileReaderProcessor.ts index 20eabf2d03..1763a17acb 100644 --- a/plugins/catalog-backend/src/modules/core/FileReaderProcessor.ts +++ b/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 { - 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) { diff --git a/plugins/catalog-backend/src/modules/core/PlaceholderProcessor.ts b/plugins/catalog-backend/src/modules/core/PlaceholderProcessor.ts index f1ad5c66f7..5b6ed31f4d 100644 --- a/plugins/catalog-backend/src/modules/core/PlaceholderProcessor.ts +++ b/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({ diff --git a/plugins/catalog-backend/src/modules/core/UrlReaderProcessor.ts b/plugins/catalog-backend/src/modules/core/UrlReaderProcessor.ts index ec8f7c11ec..61bb8fed99 100644 --- a/plugins/catalog-backend/src/modules/core/UrlReaderProcessor.ts +++ b/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}`; diff --git a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts index f1c7ba6943..10d6fc6975 100644 --- a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts +++ b/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), diff --git a/plugins/catalog-backend/src/processing/types.ts b/plugins/catalog-backend/src/processing/types.ts index 64b1dac16f..e94125b6f5 100644 --- a/plugins/catalog-backend/src/processing/types.ts +++ b/plugins/catalog-backend/src/processing/types.ts @@ -51,7 +51,6 @@ export type EntityProcessingResult = */ export type RefreshKeyData = { key: string; - entityRef: string; }; /**