Merge pull request #7171 from backstage/mob/refresh-catalog

Catalog: Add API endpoint for refreshing entities

.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<Router> {
+    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;
+  }
+```

.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.

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<Router> {
   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;
 }

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<Router> {
   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;
 }

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<T> = {
 // @public
 export type RefreshIntervalFunction = () => number;
 
+// @public
+export type RefreshOptions = {
+  entityRef: string;
+};
+
+// @public
+export interface RefreshService {
+  refresh(options: RefreshOptions): Promise<void>;
+}
+
 // 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

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<RefreshStateItem>({
       lowWatermark: 5,
       highWatermark: 10,
+      pollingIntervalMs: this.pollingIntervalMs,
       loadTasks: async count => {
         try {
           const { items } = await this.processingDatabase.transaction(

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<DbRefreshStateRow>('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<DbRefreshStateReferencesRow>(
+          '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<DbRefreshStateReferencesRow>(
+          '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<DbRefreshStateRow>('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<DbRefreshStateRow>('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();
+    },
+  );
+});

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,
+      });
+    });
+  }
+}

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,
     };
   }
 

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<express.Router> {
-  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) => {

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<DbRefreshStateRow>('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<DbRefreshStateRow>('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);
       },

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<DbRelationsRow>('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<ListAncestorsResult> {
+    const tx = txOpaque as Knex.Transaction;
+    const { entityRef } = options;
+    const entityRefs = new Array<string>();
+
+    let currentRef = entityRef.toLocaleLowerCase('en-US');
+    for (let depth = 1; depth <= MAX_ANCESTOR_DEPTH; depth += 1) {
+      const rows = await tx<DbRefreshStateReferencesRow>(
+        '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<void> {
+    const tx = txOpaque as Knex.Transaction;
+    const { entityRef } = options;
+
+    const updateResult = await tx<DbRefreshStateRow>('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<T>(fn: (tx: Transaction) => Promise<T>): Promise<T> {
     try {
       let result: T | undefined = undefined;

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<T>(fn: (tx: Transaction) => Promise<T>): Promise<T>;
 
+  /**
+   * Add unprocessed entities to the front of the processing queue using a mutation.
+   */
   replaceUnprocessedEntities(
     txOpaque: Transaction,
     options: ReplaceUnprocessedEntitiesOptions,
   ): Promise<void>;
+
   getProcessableEntities(
     txOpaque: Transaction,
     request: { processBatchSize: number },
   ): Promise<GetProcessableEntitiesResult>;
 
   /**
-   * 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<void>;
+
+  /**
+   * Schedules a refresh of a given entityRef.
+   */
+  refresh(txOpaque: Transaction, options: RefreshOptions): Promise<void>;
+
+  /**
+   * 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<ListAncestorsResult>;
 }

plugins/catalog-backend/src/next/index.ts

@@ -28,4 +28,6 @@ export type {
   CatalogProcessingEngine,
   LocationService,
   LocationStore,
+  RefreshOptions,
+  RefreshService,
 } from './types';

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

@@ -39,6 +39,28 @@ export interface CatalogProcessingEngine {
   stop(): Promise<void>;
 }
 
+/**
+ * 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<void>;
+}
+
 export type EntityProviderMutation =
   | { type: 'full'; entities: DeferredEntity[] }
   | { type: 'delta'; added: DeferredEntity[]; removed: DeferredEntity[] };

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) => {