diff --git a/.changeset/neat-coats-sell.md b/.changeset/neat-coats-sell.md new file mode 100644 index 0000000000..80c9ba111c --- /dev/null +++ b/.changeset/neat-coats-sell.md @@ -0,0 +1,37 @@ +--- +'@backstage/create-app': patch +--- + +This change adds an API endpoint for requesting a catalog refresh at `/refresh`, which is activated if a `RefreshService` is passed to `createRouter`. +The creation of the router has been abstracted behind the `CatalogBuilder` to simplify usage and future changes. The following **changes are required** to your `catalog.ts` for the refresh endpoint to function. + +```diff +- import { +- CatalogBuilder, +- createRouter, +- } from '@backstage/plugin-catalog-backend'; ++ import { CatalogBuilder } from '@backstage/plugin-catalog-backend'; + + export default async function createPlugin( + env: PluginEnvironment, + ): Promise { + const builder = await CatalogBuilder.create(env); +- const { +- entitiesCatalog, +- locationAnalyzer, +- processingEngine, +- locationService, +- } = await builder.build(); ++ const { processingEngine, router } = await builder.build(); + await processingEngine.start(); + +- return await createRouter({ +- entitiesCatalog, +- locationAnalyzer, +- locationService, +- logger: env.logger, +- config: env.config, +- }); ++ return router; + } +``` diff --git a/.changeset/red-lizards-accept.md b/.changeset/red-lizards-accept.md new file mode 100644 index 0000000000..74c3a76db5 --- /dev/null +++ b/.changeset/red-lizards-accept.md @@ -0,0 +1,7 @@ +--- +'@backstage/plugin-catalog-backend': minor +--- + +Add API endpoint for requesting a catalog refresh at `/refresh`, which is activated if a `RefreshService` is passed to `createRouter`. + +The new method is used to trigger a refresh of an entity in an as localized was as possible, usually by refreshing the parent location. diff --git a/packages/backend/src/plugins/catalog.ts b/packages/backend/src/plugins/catalog.ts index 055595dcb5..6a903d0df8 100644 --- a/packages/backend/src/plugins/catalog.ts +++ b/packages/backend/src/plugins/catalog.ts @@ -14,10 +14,7 @@ * limitations under the License. */ -import { - CatalogBuilder, - createRouter, -} from '@backstage/plugin-catalog-backend'; +import { CatalogBuilder } from '@backstage/plugin-catalog-backend'; import { Router } from 'express'; import { PluginEnvironment } from '../types'; @@ -25,20 +22,7 @@ export default async function createPlugin( env: PluginEnvironment, ): Promise { const builder = await CatalogBuilder.create(env); - const { - entitiesCatalog, - locationAnalyzer, - processingEngine, - locationService, - } = await builder.build(); - + const { processingEngine, router } = await builder.build(); await processingEngine.start(); - - return await createRouter({ - entitiesCatalog, - locationAnalyzer, - locationService, - logger: env.logger, - config: env.config, - }); + return router; } diff --git a/packages/create-app/templates/default-app/packages/backend/src/plugins/catalog.ts b/packages/create-app/templates/default-app/packages/backend/src/plugins/catalog.ts index 3145e588cc..d1ded511da 100644 --- a/packages/create-app/templates/default-app/packages/backend/src/plugins/catalog.ts +++ b/packages/create-app/templates/default-app/packages/backend/src/plugins/catalog.ts @@ -1,7 +1,4 @@ -import { - CatalogBuilder, - createRouter, -} from '@backstage/plugin-catalog-backend'; +import { CatalogBuilder } from '@backstage/plugin-catalog-backend'; import { Router } from 'express'; import { PluginEnvironment } from '../types'; @@ -9,22 +6,7 @@ export default async function createPlugin( env: PluginEnvironment, ): Promise { const builder = await CatalogBuilder.create(env); - const { - entitiesCatalog, - locationsCatalog, - locationService, - processingEngine, - locationAnalyzer, - } = await builder.build(); - + const { processingEngine, router } = await builder.build(); await processingEngine.start(); - - return await createRouter({ - entitiesCatalog, - locationsCatalog, - locationService, - locationAnalyzer, - logger: env.logger, - config: env.config, - }); + return router; } diff --git a/plugins/catalog-backend/api-report.md b/plugins/catalog-backend/api-report.md index ecc7c38b80..72216ad151 100644 --- a/plugins/catalog-backend/api-report.md +++ b/plugins/catalog-backend/api-report.md @@ -27,6 +27,7 @@ import { Organizations } from 'aws-sdk'; import { PluginDatabaseManager } from '@backstage/backend-common'; import { PluginEndpointDiscovery } from '@backstage/backend-common'; import { ResourceEntityV1alpha1 } from '@backstage/catalog-model'; +import { Router } from 'express'; import { ScmIntegrationRegistry } from '@backstage/integration'; import { ScmIntegrations } from '@backstage/integration'; import { UrlReader } from '@backstage/backend-common'; @@ -1208,6 +1209,7 @@ export class NextCatalogBuilder { locationAnalyzer: LocationAnalyzer; processingEngine: CatalogProcessingEngine; locationService: LocationService; + router: Router; }>; // Warning: (tsdoc-param-tag-missing-hyphen) The @param block should be followed by a parameter name and then a hyphen replaceEntityPolicies(policies: EntityPolicy[]): NextCatalogBuilder; @@ -1243,6 +1245,8 @@ export interface NextRouterOptions { locationService: LocationService; // (undocumented) logger: Logger_2; + // (undocumented) + refreshService?: RefreshService; } // Warning: (ae-missing-release-tag) "notFoundError" is exported by the package, but it is missing a release tag (@alpha, @beta, @public, or @internal) @@ -1363,6 +1367,16 @@ export type RecursivePartial = { // @public export type RefreshIntervalFunction = () => number; +// @public +export type RefreshOptions = { + entityRef: string; +}; + +// @public +export interface RefreshService { + refresh(options: RefreshOptions): Promise; +} + // Warning: (ae-missing-release-tag) "relation" is exported by the package, but it is missing a release tag (@alpha, @beta, @public, or @internal) // // @public (undocumented) @@ -1398,6 +1412,8 @@ export interface RouterOptions { locationService?: LocationService; // (undocumented) logger: Logger_2; + // (undocumented) + refreshService?: RefreshService; } // Warning: (tsdoc-param-tag-missing-hyphen) The @param block should be followed by a parameter name and then a hyphen diff --git a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts index d33094ec6d..4d2d74f05b 100644 --- a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts +++ b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts @@ -43,8 +43,8 @@ class Connection implements EntityProviderConnection { constructor( private readonly config: { - processingDatabase: ProcessingDatabase; id: string; + processingDatabase: ProcessingDatabase; }, ) {} @@ -60,19 +60,18 @@ class Connection implements EntityProviderConnection { items: mutation.entities, }); }); - return; - } - - this.check(mutation.added.map(e => e.entity)); - this.check(mutation.removed.map(e => e.entity)); - await db.transaction(async tx => { - await db.replaceUnprocessedEntities(tx, { - sourceKey: this.config.id, - type: 'delta', - added: mutation.added, - removed: mutation.removed, + } else if (mutation.type === 'delta') { + this.check(mutation.added.map(e => e.entity)); + this.check(mutation.removed.map(e => e.entity)); + await db.transaction(async tx => { + await db.replaceUnprocessedEntities(tx, { + sourceKey: this.config.id, + type: 'delta', + added: mutation.added, + removed: mutation.removed, + }); }); - }); + } } private check(entities: Entity[]) { @@ -97,6 +96,7 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { private readonly orchestrator: CatalogProcessingOrchestrator, private readonly stitcher: Stitcher, private readonly createHash: () => Hash, + private readonly pollingIntervalMs: number = 1000, ) {} async start() { @@ -107,8 +107,8 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { for (const provider of this.entityProviders) { await provider.connect( new Connection({ - processingDatabase: this.processingDatabase, id: provider.getProviderName(), + processingDatabase: this.processingDatabase, }), ); } @@ -116,6 +116,7 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { this.stopFunc = startTaskPipeline({ lowWatermark: 5, highWatermark: 10, + pollingIntervalMs: this.pollingIntervalMs, loadTasks: async count => { try { const { items } = await this.processingDatabase.transaction( diff --git a/plugins/catalog-backend/src/next/DefaultRefreshService.test.ts b/plugins/catalog-backend/src/next/DefaultRefreshService.test.ts new file mode 100644 index 0000000000..5df57c6897 --- /dev/null +++ b/plugins/catalog-backend/src/next/DefaultRefreshService.test.ts @@ -0,0 +1,328 @@ +/* + * Copyright 2021 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. + */ + +import { getVoidLogger } from '@backstage/backend-common'; +import { TestDatabaseId, TestDatabases } from '@backstage/backend-test-utils'; +import { createHash } from 'crypto'; +import { Knex } from 'knex'; +import { Logger } from 'winston'; +import { DatabaseManager } from './database/DatabaseManager'; +import { DefaultProcessingDatabase } from './database/DefaultProcessingDatabase'; +import { + DbRefreshStateReferencesRow, + DbRefreshStateRow, +} from './database/tables'; +import { ProcessingDatabase } from './database/types'; +import { DefaultCatalogProcessingEngine } from './DefaultCatalogProcessingEngine'; +import { EntityProcessingRequest } from './processing/types'; +import { Stitcher } from './stitching/Stitcher'; +import { Entity, stringifyEntityRef } from '@backstage/catalog-model'; +import { v4 as uuid } from 'uuid'; +import { DefaultRefreshService } from './DefaultRefreshService'; + +describe('Refresh integration', () => { + const defaultLogger = getVoidLogger(); + const databases = TestDatabases.create({ + ids: ['POSTGRES_13', 'POSTGRES_9', 'SQLITE_3'], + }); + + async function createDatabase( + databaseId: TestDatabaseId, + logger: Logger = defaultLogger, + ) { + const knex = await databases.init(databaseId); + await DatabaseManager.createDatabase(knex); + return { + knex, + db: new DefaultProcessingDatabase({ + database: knex, + logger, + refreshInterval: () => 100, + }), + }; + } + + const createPopulatedEngine = async (options: { + db: ProcessingDatabase; + knex: Knex; + entities: Entity[]; + references: { [source: string]: string[] }; + entityProcessor?: (entity: Entity) => void; + }) => { + const { db, knex, entities, references, entityProcessor } = options; + + const entityMap = new Map( + entities.map(entity => [stringifyEntityRef(entity), entity]), + ); + + for (const entity of entities) { + await knex('refresh_state').insert({ + entity_id: uuid(), + entity_ref: stringifyEntityRef(entity), + unprocessed_entity: JSON.stringify(entity), + errors: '[]', + next_update_at: '2031-01-01 23:00:00', + last_discovery_at: '2021-04-01 13:37:00', + }); + } + + const entitiesWithParent = new Set(Object.values(references).flat()); + for (const entityRef of entityMap.keys()) { + if (!entitiesWithParent.has(entityRef)) { + await knex( + 'refresh_state_references', + ).insert({ + source_key: 'ConfigLocationProvider', + target_entity_ref: entityRef, + }); + } + } + for (const [sourceRef, targetRefs] of Object.entries(references)) { + for (const targetRef of targetRefs) { + await knex( + 'refresh_state_references', + ).insert({ + source_entity_ref: sourceRef, + target_entity_ref: targetRef, + }); + } + } + + const engine = new DefaultCatalogProcessingEngine( + defaultLogger, + [], + db, + { + async process(request: EntityProcessingRequest) { + const entityRef = stringifyEntityRef(request.entity); + const entity = entityMap.get(entityRef); + if (!entity) { + throw new Error(`Unexpected entity: ${entityRef}`); + } + const deferredEntities = + references[entityRef]?.map(ref => { + const e = entityMap.get(ref); + if (!e) { + throw new Error(`Target entity not found: ${ref}`); + } + return { entity: e, locationKey: ref }; + }) || []; + + entityProcessor?.(entity); + + return { + ok: true, + completedEntity: { + ...entity, + metadata: { + ...entity.metadata, + annotations: { + ...entity.metadata.annotations, + 'refresh-completed': 'true', + }, + }, + }, + relations: [], + errors: [], + deferredEntities, + state: new Map(), + }; + }, + }, + new Stitcher(knex, defaultLogger), + () => createHash('sha1'), + 50, + ); + + return engine; + }; + + const waitForRefresh = async (knex: Knex, entityRef: string) => { + for (;;) { + const [result] = await knex('refresh_state') + .where('entity_ref', entityRef) + .select(); + + const entity = result.processed_entity + ? (JSON.parse(result.processed_entity) as Entity) + : undefined; + if (entity?.metadata?.annotations?.['refresh-completed']) { + // Reset the annotation so that we can run another verification + delete entity.metadata.annotations['refresh-completed']; + await knex('refresh_state') + .update({ + processed_entity: JSON.stringify(entity), + }) + .where('entity_ref', entityRef); + return true; + } + await new Promise(resolve => setTimeout(resolve, 500)); + } + }; + + it.each(databases.eachSupportedId())( + 'should refresh the parent location, %p', + async databaseId => { + const { knex, db } = await createDatabase(databaseId); + const refreshService = new DefaultRefreshService({ database: db }); + const engine = await createPopulatedEngine({ + db, + knex, + entities: [ + { + kind: 'Location', + apiVersion: '1.0.0', + metadata: { + name: 'myloc', + }, + }, + { + kind: 'Component', + apiVersion: '1.0.0', + metadata: { + name: 'mycomp', + }, + }, + ], + references: { + 'location:default/myloc': ['component:default/mycomp'], + }, + }); + + await engine.start(); + + await refreshService.refresh({ + entityRef: 'component:default/mycomp', + }); + + await expect( + waitForRefresh(knex, 'location:default/myloc'), + ).resolves.toBe(true); + + await engine.stop(); + }, + ); + + it.each(databases.eachSupportedId())( + 'should refresh the location further up the tree, %p', + async databaseId => { + const { knex, db } = await createDatabase(databaseId); + const refreshService = new DefaultRefreshService({ database: db }); + const engine = await createPopulatedEngine({ + db, + knex, + entities: [ + { + kind: 'Location', + apiVersion: '1.0.0', + metadata: { + name: 'myloc', + }, + }, + { + kind: 'Component', + apiVersion: '1.0.0', + metadata: { + name: 'mycomp', + }, + }, + { + kind: 'Api', + apiVersion: '1.0.0', + metadata: { + name: 'myapi', + }, + }, + ], + references: { + 'location:default/myloc': ['component:default/mycomp'], + 'component:default/mycomp': ['api:default/myapi'], + }, + }); + + await engine.start(); + + await refreshService.refresh({ + entityRef: 'api:default/myapi', + }); + + await expect(waitForRefresh(knex, 'api:default/myapi')).resolves.toBe( + true, + ); + + await engine.stop(); + }, + ); + + it.each(databases.eachSupportedId())( + 'should refresh even when parent has no changes', + async databaseId => { + let secondRound = false; + const { knex, db } = await createDatabase(databaseId); + const refreshService = new DefaultRefreshService({ database: db }); + const engine = await createPopulatedEngine({ + db, + knex, + entities: [ + { + kind: 'Location', + apiVersion: '1.0.0', + metadata: { + name: 'myloc', + }, + }, + { + kind: 'Component', + apiVersion: '1.0.0', + metadata: { + name: 'mycomp', + }, + }, + ], + references: { + 'location:default/myloc': ['component:default/mycomp'], + }, + entityProcessor: entity => { + if (entity.metadata.name === 'mycomp' && secondRound) { + entity.apiVersion = '2.0.0'; + } + }, + }); + + await engine.start(); + + await refreshService.refresh({ + entityRef: 'component:default/mycomp', + }); + + await expect( + waitForRefresh(knex, 'component:default/mycomp'), + ).resolves.toBe(true); + + secondRound = true; + + await refreshService.refresh({ + entityRef: 'component:default/mycomp', + }); + + await expect( + waitForRefresh(knex, 'component:default/mycomp'), + ).resolves.toBe(true); + + await engine.stop(); + }, + ); +}); diff --git a/plugins/catalog-backend/src/next/DefaultRefreshService.ts b/plugins/catalog-backend/src/next/DefaultRefreshService.ts new file mode 100644 index 0000000000..cfc7fa220b --- /dev/null +++ b/plugins/catalog-backend/src/next/DefaultRefreshService.ts @@ -0,0 +1,48 @@ +/* + * Copyright 2021 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. + */ + +import { DefaultProcessingDatabase } from './database/DefaultProcessingDatabase'; +import { RefreshOptions, RefreshService } from './types'; + +export class DefaultRefreshService implements RefreshService { + private database: DefaultProcessingDatabase; + + constructor(options: { database: DefaultProcessingDatabase }) { + this.database = options.database; + } + + async refresh(options: RefreshOptions) { + await this.database.transaction(async tx => { + const { entityRefs } = await this.database.listAncestors(tx, { + entityRef: options.entityRef, + }); + const locationAncestor = entityRefs.find(ref => + ref.startsWith('location:'), + ); + + // TODO: Refreshes are currently scheduled(as soon as possible) for execution and will therefore happen in the future. + // There's room for improvements here where the refresh could potentially hang or return an ID so that the user can check progress. + if (locationAncestor) { + await this.database.refresh(tx, { + entityRef: locationAncestor, + }); + } + await this.database.refresh(tx, { + entityRef: options.entityRef, + }); + }); + } +} diff --git a/plugins/catalog-backend/src/next/NextCatalogBuilder.ts b/plugins/catalog-backend/src/next/NextCatalogBuilder.ts index fa43286777..a34be2ca6a 100644 --- a/plugins/catalog-backend/src/next/NextCatalogBuilder.ts +++ b/plugins/catalog-backend/src/next/NextCatalogBuilder.ts @@ -27,6 +27,7 @@ import { } from '@backstage/catalog-model'; import { ScmIntegrations } from '@backstage/integration'; import { createHash } from 'crypto'; +import { Router } from 'express'; import lodash from 'lodash'; import { DatabaseLocationsCatalog, @@ -75,6 +76,8 @@ import { RefreshIntervalFunction, } from './refresh'; import { CatalogEnvironment } from '../service/CatalogBuilder'; +import { createNextRouter } from './NextRouter'; +import { DefaultRefreshService } from './DefaultRefreshService'; /** * A builder that helps wire up all of the component parts of the catalog. @@ -277,6 +280,7 @@ export class NextCatalogBuilder { locationAnalyzer: LocationAnalyzer; processingEngine: CatalogProcessingEngine; locationService: LocationService; + router: Router; }> { const { config, database, logger } = this.env; @@ -332,6 +336,17 @@ export class NextCatalogBuilder { locationStore, orchestrator, ); + const refreshService = new DefaultRefreshService({ + database: processingDatabase, + }); + const router = await createNextRouter({ + entitiesCatalog, + locationAnalyzer, + locationService, + refreshService, + logger, + config, + }); return { entitiesCatalog, @@ -339,6 +354,7 @@ export class NextCatalogBuilder { locationAnalyzer, processingEngine, locationService, + router, }; } diff --git a/plugins/catalog-backend/src/next/NextRouter.ts b/plugins/catalog-backend/src/next/NextRouter.ts index 99073be421..b9da91b30a 100644 --- a/plugins/catalog-backend/src/next/NextRouter.ts +++ b/plugins/catalog-backend/src/next/NextRouter.ts @@ -34,12 +34,13 @@ import { parseEntityTransformParams, } from '../service/request'; import { disallowReadonlyMode, validateRequestBody } from '../service/util'; -import { LocationService } from './types'; +import { RefreshService, RefreshOptions, LocationService } from './types'; export interface NextRouterOptions { entitiesCatalog?: EntitiesCatalog; locationAnalyzer?: LocationAnalyzer; locationService: LocationService; + refreshService?: RefreshService; logger: Logger; config: Config; } @@ -47,8 +48,14 @@ export interface NextRouterOptions { export async function createNextRouter( options: NextRouterOptions, ): Promise { - const { entitiesCatalog, locationAnalyzer, locationService, config, logger } = - options; + const { + entitiesCatalog, + locationAnalyzer, + locationService, + refreshService, + config, + logger, + } = options; const router = Router(); router.use(express.json()); @@ -59,6 +66,14 @@ export async function createNextRouter( logger.info('Catalog is running in readonly mode'); } + if (refreshService) { + router.post('/refresh', async (req, res) => { + const refreshOptions: RefreshOptions = req.body; + await refreshService.refresh(refreshOptions); + res.status(200).send(); + }); + } + if (entitiesCatalog) { router .get('/entities', async (req, res) => { diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts index 23829bd33f..afc3806d36 100644 --- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts +++ b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts @@ -30,6 +30,7 @@ import { DbRelationsRow, } from './tables'; import { createRandomRefreshInterval } from '../refresh'; +import { timestampToDateTime } from './conversion'; describe('Default Processing Database', () => { const defaultLogger = getVoidLogger(); @@ -66,21 +67,6 @@ describe('Default Processing Database', () => { await db('refresh_state').insert(ref); }; - const parseDate = (date: string | Date): DateTime => { - const parsedDate = - typeof date === 'string' - ? DateTime.fromSQL(date, { zone: 'UTC' }) - : DateTime.fromJSDate(date); - - if (!parsedDate.isValid) { - throw new Error( - `Failed to parse date, reason: ${parsedDate.invalidReason}, explanation: ${parsedDate.invalidExplanation}`, - ); - } - - return parsedDate; - }; - describe('addUprocessedEntities', () => { function mockEntity(name: string, type: string): Entity { return { @@ -1010,7 +996,7 @@ describe('Default Processing Database', () => { const result = await knex('refresh_state') .where('entity_ref', 'location:default/new-root') .select(); - const nextUpdate = parseDate(result[0].next_update_at); + const nextUpdate = timestampToDateTime(result[0].next_update_at); const nextUpdateDiff = nextUpdate.diff(now, 'seconds'); expect(nextUpdateDiff.seconds).toBeGreaterThanOrEqual(90); }, diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts index 30457f2cbd..8ba9d54b8f 100644 --- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts +++ b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts @@ -16,7 +16,7 @@ import { Entity, 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'; @@ -36,8 +36,11 @@ import { GetProcessableEntitiesResult, ProcessingDatabase, RefreshStateItem, + RefreshOptions, ReplaceUnprocessedEntitiesOptions, UpdateProcessedEntityOptions, + ListAncestorsOptions, + ListAncestorsResult, } from './types'; // The number of items that are sent per batch to the database layer, when @@ -45,6 +48,7 @@ import { // errors in the underlying engine due to exceeding query limits, but large // enough to get the speed benefits. const BATCH_SIZE = 50; +const MAX_ANCESTOR_DEPTH = 32; export class DefaultProcessingDatabase implements ProcessingDatabase { constructor( @@ -101,8 +105,6 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { sourceEntityRef: stringifyEntityRef(processedEntity), }); - // Update fragments - // Delete old relations await tx('relations') .where({ originating_entity_id: id }) @@ -327,6 +329,10 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { } } + /** + * Add a set of deferred entities for processing. + * The entities will be added at the front of the processing queue. + */ async addUnprocessedEntities( txOpaque: Transaction, options: AddUnprocessedEntitiesOptions, @@ -350,6 +356,10 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { unprocessed_entity: serializedEntity, location_key: locationKey, last_discovery_at: tx.fn.now(), + // We only get to this point if a processed entity actually had any changes, or + // if an entity provider requested this mutation, meaning that we can safely + // bump the deferred entities to the front of the queue for immediate processing. + next_update_at: tx.fn.now(), }) .where('entity_ref', entityRef) .andWhere(inner => { @@ -508,6 +518,57 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { }; } + async listAncestors( + txOpaque: Transaction, + options: ListAncestorsOptions, + ): Promise { + const tx = txOpaque as Knex.Transaction; + const { entityRef } = options; + const entityRefs = new Array(); + + let currentRef = entityRef.toLocaleLowerCase('en-US'); + for (let depth = 1; depth <= MAX_ANCESTOR_DEPTH; depth += 1) { + const rows = await tx( + 'refresh_state_references', + ) + .where({ target_entity_ref: currentRef }) + .select(); + + if (rows.length === 0) { + if (depth === 1) { + throw new NotFoundError(`Entity ${currentRef} not found`); + } + throw new NotFoundError( + `Entity ${entityRef} has a broken parent reference chain at ${currentRef}`, + ); + } + + const parentRef = rows.find(r => r.source_entity_ref)?.source_entity_ref; + if (!parentRef) { + // We've reached the top of the tree which is the entityProvider. + // In this case we refresh the entity itself. + return { entityRefs }; + } + entityRefs.push(parentRef); + currentRef = parentRef; + } + throw new Error( + `Unable receive ancestors for ${entityRef}, reached maximum depth of ${MAX_ANCESTOR_DEPTH}`, + ); + } + + async refresh(txOpaque: Transaction, options: RefreshOptions): Promise { + const tx = txOpaque as Knex.Transaction; + const { entityRef } = options; + + const updateResult = await tx('refresh_state') + .where({ entity_ref: entityRef.toLocaleLowerCase('en-US') }) + .update({ next_update_at: tx.fn.now() }); + if (updateResult === 0) { + throw new NotFoundError(`Failed to schedule ${entityRef} for refresh`); + } + } + async transaction(fn: (tx: Transaction) => Promise): Promise { try { let result: T | undefined = undefined; diff --git a/plugins/catalog-backend/src/next/database/types.ts b/plugins/catalog-backend/src/next/database/types.ts index 38ca59a431..3dfe8e5592 100644 --- a/plugins/catalog-backend/src/next/database/types.ts +++ b/plugins/catalog-backend/src/next/database/types.ts @@ -79,20 +79,39 @@ export type ReplaceUnprocessedEntitiesOptions = type: 'delta'; }; +export type RefreshOptions = { + entityRef: string; +}; + +export type ListAncestorsOptions = { + entityRef: string; +}; + +export type ListAncestorsResult = { + entityRefs: string[]; +}; + export interface ProcessingDatabase { transaction(fn: (tx: Transaction) => Promise): Promise; + /** + * Add unprocessed entities to the front of the processing queue using a mutation. + */ replaceUnprocessedEntities( txOpaque: Transaction, options: ReplaceUnprocessedEntitiesOptions, ): Promise; + getProcessableEntities( txOpaque: Transaction, request: { processBatchSize: number }, ): Promise; /** - * Updates a processed entity + * Updates a processed entity. + * + * Any deferred entities are added at the front of the processing queue for + * immediate processing, meaning this should only be called when the entity has changes. */ updateProcessedEntity( txOpaque: Transaction, @@ -106,4 +125,19 @@ export interface ProcessingDatabase { txOpaque: Transaction, options: UpdateProcessedEntityErrorsOptions, ): Promise; + + /** + * Schedules a refresh of a given entityRef. + */ + refresh(txOpaque: Transaction, options: RefreshOptions): Promise; + + /** + * Lists all ancestors of a given entityRef. + * + * The returned list is ordered from the most immediate ancestor to the most distant one. + */ + listAncestors( + txOpaque: Transaction, + options: ListAncestorsOptions, + ): Promise; } diff --git a/plugins/catalog-backend/src/next/index.ts b/plugins/catalog-backend/src/next/index.ts index cdd81d9f4d..5f70dd2125 100644 --- a/plugins/catalog-backend/src/next/index.ts +++ b/plugins/catalog-backend/src/next/index.ts @@ -28,4 +28,6 @@ export type { CatalogProcessingEngine, LocationService, LocationStore, + RefreshOptions, + RefreshService, } from './types'; diff --git a/plugins/catalog-backend/src/next/types.ts b/plugins/catalog-backend/src/next/types.ts index 3b27dd7d10..fa058c4138 100644 --- a/plugins/catalog-backend/src/next/types.ts +++ b/plugins/catalog-backend/src/next/types.ts @@ -39,6 +39,28 @@ export interface CatalogProcessingEngine { stop(): Promise; } +/** + * Options for requesting a refresh of entities in the catalog. + * + * @public + */ +export type RefreshOptions = { + /** The reference to a single entity that should be refreshed */ + entityRef: string; +}; + +/** + * A service that manages refreshes of entities in the catalog. + * + * @public + */ +export interface RefreshService { + /** + * Request a refresh of entities in the catalog. + */ + refresh(options: RefreshOptions): Promise; +} + export type EntityProviderMutation = | { type: 'full'; entities: DeferredEntity[] } | { type: 'delta'; added: DeferredEntity[]; removed: DeferredEntity[] }; diff --git a/plugins/catalog-backend/src/service/router.ts b/plugins/catalog-backend/src/service/router.ts index 71b8614a99..f1bb18c825 100644 --- a/plugins/catalog-backend/src/service/router.ts +++ b/plugins/catalog-backend/src/service/router.ts @@ -28,7 +28,7 @@ import { Logger } from 'winston'; import yn from 'yn'; import { EntitiesCatalog, LocationsCatalog } from '../catalog'; import { HigherOrderOperation, LocationAnalyzer } from '../ingestion/types'; -import { LocationService } from '../next/types'; +import { RefreshService, LocationService, RefreshOptions } from '../next/types'; import { basicEntityFilter, parseEntityFilterParams, @@ -47,6 +47,7 @@ export interface RouterOptions { higherOrderOperation?: HigherOrderOperation; locationAnalyzer?: LocationAnalyzer; locationService?: LocationService; + refreshService?: RefreshService; logger: Logger; config: Config; } @@ -60,6 +61,7 @@ export async function createRouter( higherOrderOperation, locationAnalyzer, locationService, + refreshService, config, logger, } = options; @@ -73,6 +75,14 @@ export async function createRouter( logger.info('Catalog is running in readonly mode'); } + if (refreshService) { + router.post('/refresh', async (req, res) => { + const refreshOptions: RefreshOptions = req.body; + await refreshService.refresh(refreshOptions); + res.status(200).send(); + }); + } + if (entitiesCatalog) { router .get('/entities', async (req, res) => {