diff --git a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts index 1911f8f90e..0e77e6e9b4 100644 --- a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts +++ b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts @@ -73,11 +73,12 @@ class Connection implements EntityProviderConnection { }); }); } else if (mutation.type === 'refresh') { - await db.transaction(async tx => { - await db.refreshUnprocessedEntities(tx, { - match: mutation.match, - }); - }); + // await db.transaction(async tx => { + // await db.refreshUnprocessedEntities(tx, { + // match: mutation.match, + // }); + // }); + console.log('wopoop'); } } @@ -245,17 +246,20 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { } async refresh(options: EntityRefreshOptions) { - await Promise.all( - this.entityProviders.map(async provider => { - try { - await provider.refresh?.(options); - } catch (e) { - throw new Error( - `Provider ${provider.getProviderName()} failed refresh, ${e}`, - ); - } - }), - ); + await this.processingDatabase.transaction(async tx => { + await this.processingDatabase.refreshUnprocessedEntities(tx, options); + }); + // await Promise.all( + // this.entityProviders.map(async provider => { + // try { + // await provider.refresh?.(options); + // } catch (e) { + // throw new Error( + // `Provider ${provider.getProviderName()} failed refresh, ${e}`, + // ); + // } + // }), + // ); } } diff --git a/plugins/catalog-backend/src/next/DefaultLocationStore.ts b/plugins/catalog-backend/src/next/DefaultLocationStore.ts index 1e77d60418..09eacc3955 100644 --- a/plugins/catalog-backend/src/next/DefaultLocationStore.ts +++ b/plugins/catalog-backend/src/next/DefaultLocationStore.ts @@ -137,23 +137,20 @@ export class DefaultLocationStore implements LocationStore, EntityProvider { }); } - async refresh(options: EntityRefreshOptions) { - const match: RefreshStateMatch = {}; - - // locationKey?: string; - // entityRef?: string; - - if (options.entityRef) { - match.entityRef = options.entityRef; - } - if (options.locationRef) { - // TODO - } - - await this.connection.applyMutation({ - type: 'refresh', - match, - }); + async refresh(_options: EntityRefreshOptions) { + // const match: RefreshStateMatch = {}; + // // locationKey?: string; + // // entityRef?: string; + // if (options.entityRef) { + // match.entityRef = options.entityRef; + // } + // if (options.locationRef) { + // // TODO + // } + // await this.connection.applyMutation({ + // type: 'refresh', + // match, + // }); } private async locations(dbOrTx: Knex.Transaction | Knex = this.db) { diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts index 23829bd33f..e027830601 100644 --- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts +++ b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts @@ -1017,4 +1017,87 @@ describe('Default Processing Database', () => { 60_000, ); }); + + describe('refreshEntities', () => { + it.each(databases.eachSupportedId())( + 'should refresh the location further up the tree, %p', + async databaseId => { + const { knex, db } = await createDatabase(databaseId); + + await knex('refresh_state').insert({ + entity_id: '7', + entity_ref: 'location:default/myloc', + unprocessed_entity: JSON.stringify({ + kind: 'Location', + apiVersion: '1.0.0', + metadata: { + name: 'xyz', + }, + } as Entity), + errors: '[]', + next_update_at: '2031-01-01 23:00:00', + last_discovery_at: '2021-04-01 13:37:00', + }); + + await knex('refresh_state').insert({ + entity_id: '8', + entity_ref: 'component:default/mycomp', + unprocessed_entity: JSON.stringify({ + kind: 'Component', + apiVersion: '1.0.0', + metadata: { + name: 'xyz', + }, + } as Entity), + errors: '[]', + next_update_at: '2031-01-01 23:00:00', + last_discovery_at: '2021-04-01 13:37:00', + }); + + await knex('refresh_state').insert({ + entity_id: '9', + entity_ref: 'api:default/myapi', + unprocessed_entity: JSON.stringify({ + kind: 'Api', + apiVersion: '1.0.0', + metadata: { + name: 'xyz', + }, + } as Entity), + errors: '[]', + next_update_at: '2031-01-01 23:00:00', + last_discovery_at: '2021-04-01 13:37:00', + }); + + await insertRefRow(knex, { + source_entity_ref: 'component:default/mycomp', + target_entity_ref: 'api:default/myapi', + }); + await insertRefRow(knex, { + source_entity_ref: 'location:default/myloc', + target_entity_ref: 'component:default/mycomp', + }); + + await insertRefRow(knex, { + source_key: 'ConfigLocationProvider', + target_entity_ref: 'location:default/myloc', + }); + + await db.transaction(async tx => { + await db.refreshUnprocessedEntities(tx, { + entityRef: 'api:default/myapi', + }); + }); + + const [result] = await knex('refresh_state') + .where('entity_ref', 'api:default/myapi') + .select(); + + // TODO: This is going to break after 2031 + expect(parseDate(result.next_update_at).year).toEqual( + DateTime.local().year, + ); + }, + ); + }); }); diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts index d2c17f7ad1..a19f88188d 100644 --- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts +++ b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts @@ -14,9 +14,13 @@ * limitations under the License. */ -import { Entity, stringifyEntityRef } from '@backstage/catalog-model'; +import { + Entity, + LOCATION_ANNOTATION, + stringifyEntityRef, +} from '@backstage/catalog-model'; import { JsonObject } from '@backstage/config'; -import { ConflictError } from '@backstage/errors'; +import { ConflictError, NotFoundError } from '@backstage/errors'; import { Knex } from 'knex'; import lodash from 'lodash'; import { v4 as uuid } from 'uuid'; @@ -351,6 +355,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { unprocessed_entity: serializedEntity, location_key: locationKey, last_discovery_at: tx.fn.now(), + next_update_at: tx.fn.now(), }) .where('entity_ref', entityRef) .andWhere(inner => { @@ -509,11 +514,62 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { }; } + // TODO(jhaals): Rename this to refreshEntity? async refreshUnprocessedEntities( txOpaque: Transaction, options: RefreshUnprocessedEntitiesOptions, ): Promise { const tx = txOpaque as Knex.Transaction; + if ('entityRef' in options) { + const { entityRef } = options; + const [result] = await tx('refresh_state') + .where({ entity_ref: entityRef }) + .select(); + + if (!result) { + throw new NotFoundError(`EntityRef ${entityRef} not found`); + } + + const refs = await tx( + '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('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`, + ); + } + } + } + /* + For a given entityRef: + - Fetch entity + - recursively select from refresh_state_references where target is our entityRef. + Continue until we find a location. + - Process and run addUnprocessedEntities with a flag telling it to bump the timestamp for all deferred entities. + */ + /* + For a given URL + - Fetch entity based on managed by location URL? + - repeat for entityRef + */ + } + + if ('locationRef' in options) { + // TODO(jhaals): managed by or origin? + // const entity: Entity = JSON.parse(result.processed_entity); + // // TODO: ONly required for URLz + // const managedBy = entity.metadata?.annotations?.[LOCATION_ANNOTATION] + // console.log(managedBy); + } } async transaction(fn: (tx: Transaction) => Promise): Promise { diff --git a/plugins/catalog-backend/src/next/database/types.ts b/plugins/catalog-backend/src/next/database/types.ts index 3a65cc7361..e8ced9d250 100644 --- a/plugins/catalog-backend/src/next/database/types.ts +++ b/plugins/catalog-backend/src/next/database/types.ts @@ -85,9 +85,12 @@ export type RefreshStateMatch = { parentOfEntityRef: string; }; -export type RefreshUnprocessedEntitiesOptions = { - match: RefreshStateMatch; -}; +export type RefreshUnprocessedEntitiesOptions = + | { + // match: RefreshStateMatch; + entityRef: string; + } + | { locationRef: string }; export interface ProcessingDatabase { transaction(fn: (tx: Transaction) => Promise): Promise;