WIP

Signed-off-by: Fredrik Adelöw <freben@gmail.com>

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/src/next/DefaultCatalogProcessingEngine.ts

@@ -36,6 +36,7 @@ import {
   EntityProvider,
   EntityProviderConnection,
   EntityProviderMutation,
+  EntityRefreshOptions,
 } from './types';
 
 class Connection implements EntityProviderConnection {
@@ -43,8 +44,8 @@ class Connection implements EntityProviderConnection {
 
   constructor(
     private readonly config: {
-      processingDatabase: ProcessingDatabase;
       id: string;
+      processingDatabase: ProcessingDatabase;
     },
   ) {}
 
@@ -60,19 +61,24 @@ 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,
+        });
       });
-    });
+    } else if (mutation.type === 'refresh') {
+      await db.transaction(async tx => {
+        await db.refreshUnprocessedEntities(tx, {
+          match: mutation.match,
+        });
+      });
+    }
   }
 
   private check(entities: Entity[]) {
@@ -107,8 +113,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,
         }),
       );
     }
@@ -237,6 +243,20 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
       this.stopFunc = undefined;
     }
   }
+
+  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}`,
+          );
+        }
+      }),
+    );
+  }
 }
 
 // Helps wrap the timing and logging behaviors

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

@@ -19,10 +19,12 @@ import { ConflictError, NotFoundError } from '@backstage/errors';
 import { Knex } from 'knex';
 import { v4 as uuid } from 'uuid';
 import { DbLocationsRow } from './database/tables';
+import { RefreshStateMatch } from './database/types';
 import { getEntityLocationRef } from './processing/util';
 import {
   EntityProvider,
   EntityProviderConnection,
+  EntityRefreshOptions,
   LocationStore,
 } from './types';
 import { locationSpecToLocationEntity } from './util';
@@ -135,6 +137,25 @@ 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,
+    });
+  }
+
   private async locations(dbOrTx: Knex.Transaction | Knex = this.db) {
     const locations = await dbOrTx<DbLocationsRow>('locations').select();
     return (

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,7 @@ import {
   RefreshIntervalFunction,
 } from './refresh';
 import { CatalogEnvironment } from '../service/CatalogBuilder';
+import { createNextRouter } from './NextRouter';
 
 /**
  * A builder that helps wire up all of the component parts of the catalog.
@@ -277,6 +279,7 @@ export class NextCatalogBuilder {
     locationAnalyzer: LocationAnalyzer;
     processingEngine: CatalogProcessingEngine;
     locationService: LocationService;
+    router: Router;
   }> {
     const { config, database, logger } = this.env;
 
@@ -333,12 +336,22 @@ export class NextCatalogBuilder {
       orchestrator,
     );
 
+    const router = await createNextRouter({
+      entitiesCatalog,
+      locationAnalyzer,
+      locationService,
+      processingEngine,
+      logger,
+      config,
+    });
+
     return {
       entitiesCatalog,
       locationsCatalog,
       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 { CatalogProcessingEngine, LocationService, EntityRefreshOptions } from './types';
 
 export interface NextRouterOptions {
   entitiesCatalog?: EntitiesCatalog;
   locationAnalyzer?: LocationAnalyzer;
   locationService: LocationService;
+  processingEngine?: CatalogProcessingEngine;
   logger: Logger;
   config: Config;
 }
@@ -47,7 +48,7 @@ export interface NextRouterOptions {
 export async function createNextRouter(
   options: NextRouterOptions,
 ): Promise<express.Router> {
-  const { entitiesCatalog, locationAnalyzer, locationService, config, logger } =
+  const { entitiesCatalog, locationAnalyzer, locationService, processingEngine, config, logger } =
     options;
 
   const router = Router();
@@ -59,6 +60,14 @@ export async function createNextRouter(
     logger.info('Catalog is running in readonly mode');
   }
 
+  if (processingEngine) {
+    router.post('/refresh'), async (req, res) => {
+      const options: EntityRefreshOptions = req.body;
+      await processingEngine.refresh(options);
+      res.status(200);
+    });
+  }
+
   if (entitiesCatalog) {
     router
       .get('/entities', async (req, res) => {

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

@@ -36,6 +36,7 @@ import {
   GetProcessableEntitiesResult,
   ProcessingDatabase,
   RefreshStateItem,
+  RefreshUnprocessedEntitiesOptions,
   ReplaceUnprocessedEntitiesOptions,
   UpdateProcessedEntityOptions,
 } from './types';
@@ -508,6 +509,13 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
     };
   }
 
+  async refreshUnprocessedEntities(
+    txOpaque: Transaction,
+    options: RefreshUnprocessedEntitiesOptions,
+  ): Promise<void> {
+    const tx = txOpaque as Knex.Transaction;
+  }
+
   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,6 +79,16 @@ export type ReplaceUnprocessedEntitiesOptions =
       type: 'delta';
     };
 
+export type RefreshStateMatch = {
+  locationKey?: string;
+  entityRef?: string;
+  parentOfEntityRef: string;
+};
+
+export type RefreshUnprocessedEntitiesOptions = {
+  match: RefreshStateMatch;
+};
+
 export interface ProcessingDatabase {
   transaction<T>(fn: (tx: Transaction) => Promise<T>): Promise<T>;
 
@@ -86,6 +96,7 @@ export interface ProcessingDatabase {
     txOpaque: Transaction,
     options: ReplaceUnprocessedEntitiesOptions,
   ): Promise<void>;
+
   getProcessableEntities(
     txOpaque: Transaction,
     request: { processBatchSize: number },
@@ -106,4 +117,9 @@ export interface ProcessingDatabase {
     txOpaque: Transaction,
     options: UpdateProcessedEntityErrorsOptions,
   ): Promise<void>;
+
+  refreshUnprocessedEntities(
+    txOpaque: Transaction,
+    options: RefreshUnprocessedEntitiesOptions,
+  ): Promise<void>;
 }

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

@@ -15,6 +15,7 @@
  */
 
 import { Entity, Location, LocationSpec } from '@backstage/catalog-model';
+import { RefreshStateMatch } from './database/types';
 import { DeferredEntity } from './processing/types';
 
 export interface LocationService {
@@ -37,9 +38,15 @@ export interface LocationStore {
 export interface CatalogProcessingEngine {
   start(): Promise<void>;
   stop(): Promise<void>;
+  refresh(options: EntityRefreshOptions): Promise<void>;
 }
 
+export type EntityRefreshOptions =
+  | { entityRef: string } // example: component:default/backstage
+  | { locationRef: string }; // example: url:https://github.com/backstage/backstage/blob/master/catalog-info.yaml
+
 export type EntityProviderMutation =
+  | { type: 'refresh'; match: RefreshStateMatch } // TODO(jhaals): Should this really use a type from the db?
   | { type: 'full'; entities: DeferredEntity[] }
   | { type: 'delta'; added: DeferredEntity[]; removed: DeferredEntity[] };
 
@@ -50,4 +57,5 @@ export interface EntityProviderConnection {
 export interface EntityProvider {
   getProviderName(): string;
   connect(connection: EntityProviderConnection): Promise<void>;
+  refresh?(options: EntityRefreshOptions): Promise<void>;
 }