add dev app, more testing

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

plugins/catalog-backend-module-incremental-ingestion/package.json

@@ -54,7 +54,10 @@
     "winston": "^3.2.1"
   },
   "devDependencies": {
-    "@backstage/cli": "workspace:^"
+    "@backstage/backend-app-api": "workspace:^",
+    "@backstage/cli": "workspace:^",
+    "@backstage/plugin-catalog-backend": "workspace:^",
+    "get-port": "^6.1.2"
   },
   "files": [
     "alpha",

plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts

@@ -28,8 +28,8 @@ import { v4 } from 'uuid';
 import { stringifyError } from '@backstage/errors';
 
 export class IncrementalIngestionEngine implements IterationEngine {
-  restLength: Duration;
-  backoff: DurationObjectUnits[];
+  private readonly restLength: Duration;
+  private readonly backoff: DurationObjectUnits[];
 
   private manager: IncrementalIngestionDatabaseManager;
 

plugins/catalog-backend-module-incremental-ingestion/src/module/WrapperProviders.test.ts

@@ -53,17 +53,17 @@ describe('WrapperProviders', () => {
     async databaseId => {
       const client = await databases.init(databaseId);
 
-      const provider1: IncrementalEntityProvider<{}, number> = {
+      const provider1: IncrementalEntityProvider<number, {}> = {
         getProviderName: () => 'provider1',
         around: burst => burst(0),
         next: async (cursor, _context) => {
-          return cursor === 0
+          return !cursor
             ? { done: false, entities: [], cursor: 1 }
             : { done: true };
         },
       };
 
-      const provider2: IncrementalEntityProvider<{}, number> = {
+      const provider2: IncrementalEntityProvider<number, {}> = {
         getProviderName: () => 'provider2',
         around: burst => burst(0),
         next: async (cursor, _context) => {

plugins/catalog-backend-module-incremental-ingestion/src/module/WrapperProviders.ts

@@ -22,12 +22,14 @@ import {
   EntityProvider,
   EntityProviderConnection,
 } from '@backstage/plugin-catalog-node';
+import express from 'express';
 import { Knex } from 'knex';
 import once from 'lodash/once';
 import { Duration } from 'luxon';
 import { IncrementalIngestionDatabaseManager } from '../database/IncrementalIngestionDatabaseManager';
 import { applyDatabaseMigrations } from '../database/migrations';
 import { IncrementalIngestionEngine } from '../engine/IncrementalIngestionEngine';
+import { createIncrementalProviderRouter } from '../router/routes';
 import {
   IncrementalEntityProvider,
   IncrementalEntityProviderOptions,
@@ -70,6 +72,13 @@ export class WrapperProviders {
     };
   }
 
+  async adminRouter(): Promise<express.Router> {
+    return createIncrementalProviderRouter(
+      new IncrementalIngestionDatabaseManager({ client: this.options.client }),
+      loggerToWinstonLogger(this.options.logger),
+    );
+  }
+
   private async startProvider(
     provider: IncrementalEntityProvider<unknown, unknown>,
     providerOptions: IncrementalEntityProviderOptions,

plugins/catalog-backend-module-incremental-ingestion/src/module/incrementalIngestionEntityProviderCatalogModule.test.ts

@@ -18,6 +18,7 @@ import { getVoidLogger } from '@backstage/backend-common';
 import {
   configServiceRef,
   databaseServiceRef,
+  httpRouterServiceRef,
   loggerServiceRef,
   schedulerServiceRef,
 } from '@backstage/backend-plugin-api';
@@ -29,22 +30,26 @@ import { incrementalIngestionEntityProviderCatalogModule } from './incrementalIn
 
 describe('bitbucketServerEntityProviderCatalogModule', () => {
   it('should register provider at the catalog extension point', async () => {
-    const provider1: IncrementalEntityProvider<{}, number> = {
+    const provider1: IncrementalEntityProvider<number, {}> = {
       getProviderName: () => 'provider1',
       around: burst => burst(0),
       next: async (cursor, _context) => {
-        return cursor === 0
+        return !cursor
           ? { done: false, entities: [], cursor: 1 }
           : { done: true };
       },
     };
 
     const addEntityProvider = jest.fn();
+    const httpRouterUse = jest.fn();
 
     const scheduler = {};
     const database = {
       getClient: jest.fn(),
     };
+    const httpRouter = {
+      use: httpRouterUse,
+    };
 
     await startTestBackend({
       extensionPoints: [
@@ -52,9 +57,10 @@ describe('bitbucketServerEntityProviderCatalogModule', () => {
       ],
       services: [
         [configServiceRef, new ConfigReader({})],
+        [databaseServiceRef, database],
+        [httpRouterServiceRef, httpRouter],
         [loggerServiceRef, getVoidLogger()],
         [schedulerServiceRef, scheduler],
-        [databaseServiceRef, database],
       ],
       features: [
         incrementalIngestionEntityProviderCatalogModule({
@@ -76,5 +82,6 @@ describe('bitbucketServerEntityProviderCatalogModule', () => {
     expect(addEntityProvider.mock.calls[0][0].getProviderName()).toBe(
       'provider1',
     );
+    expect(httpRouterUse).toHaveBeenCalledTimes(1);
   });
 });

plugins/catalog-backend-module-incremental-ingestion/src/module/incrementalIngestionEntityProviderCatalogModule.ts

@@ -18,6 +18,7 @@ import {
   configServiceRef,
   createBackendModule,
   databaseServiceRef,
+  httpRouterServiceRef,
   loggerServiceRef,
   schedulerServiceRef,
 } from '@backstage/backend-plugin-api';
@@ -50,21 +51,34 @@ export const incrementalIngestionEntityProviderCatalogModule =
         deps: {
           catalog: catalogProcessingExtensionPoint,
           config: configServiceRef,
-          logger: loggerServiceRef,
           database: databaseServiceRef,
+          httpRouter: httpRouterServiceRef,
+          logger: loggerServiceRef,
           scheduler: schedulerServiceRef,
         },
-        async init({ catalog, config, logger, database, scheduler }) {
+        async init({
+          catalog,
+          config,
+          database,
+          httpRouter,
+          logger,
+          scheduler,
+        }) {
+          const client = await database.getClient();
+
           const providers = new WrapperProviders({
             config,
             logger,
-            client: await database.getClient(),
+            client,
             scheduler,
           });
+
           for (const entry of options.providers) {
             const wrapped = providers.wrap(entry.provider, entry.options);
             catalog.addEntityProvider(wrapped);
           }
+
+          httpRouter.use(await providers.adminRouter());
         },
       });
     },

plugins/catalog-backend-module-incremental-ingestion/src/run.ts

@@ -0,0 +1,99 @@
+/*
+ * Copyright 2022 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.
+ */
+
+// eslint-disable-next-line import/no-extraneous-dependencies
+import {
+  databaseFactory,
+  discoveryFactory,
+  httpRouterFactory,
+  lifecycleFactory,
+  loggerFactory,
+  permissionsFactory,
+  rootLoggerFactory,
+  schedulerFactory,
+  tokenManagerFactory,
+  urlReaderFactory,
+} from '@backstage/backend-app-api';
+import { configServiceRef } from '@backstage/backend-plugin-api';
+import { startTestBackend } from '@backstage/backend-test-utils';
+import { ConfigReader } from '@backstage/config';
+import { catalogPlugin } from '@backstage/plugin-catalog-backend';
+import {
+  IncrementalEntityProvider,
+  incrementalIngestionEntityProviderCatalogModule,
+} from '.';
+
+const provider: IncrementalEntityProvider<number, {}> = {
+  getProviderName: () => 'test-provider',
+  around: burst => burst(0),
+  next: async (_context, cursor) => {
+    await new Promise(resolve => setTimeout(resolve, 500));
+    if (cursor === undefined || cursor < 3) {
+      console.log(`### Returning batch #${cursor}`);
+      return { done: false, entities: [], cursor: (cursor ?? 0) + 1 };
+    }
+
+    console.log('### Last batch reached, stopping');
+    return { done: true };
+  },
+};
+
+async function main() {
+  const config = {
+    backend: {
+      baseUrl: 'http://localhost:7007',
+      listen: ':7007',
+      database: { client: 'better-sqlite3', connection: ':memory:' },
+    },
+  };
+
+  await startTestBackend({
+    services: [
+      [configServiceRef, new ConfigReader(config)],
+      databaseFactory(),
+      discoveryFactory(),
+      httpRouterFactory(),
+      lifecycleFactory(),
+      loggerFactory(),
+      permissionsFactory(),
+      rootLoggerFactory(),
+      schedulerFactory(),
+      tokenManagerFactory(),
+      urlReaderFactory(),
+    ],
+    extensionPoints: [],
+    features: [
+      catalogPlugin(),
+      incrementalIngestionEntityProviderCatalogModule({
+        providers: [
+          {
+            provider: provider,
+            options: {
+              burstInterval: { seconds: 1 },
+              burstLength: { seconds: 10 },
+              restLength: { seconds: 10 },
+            },
+          },
+        ],
+      }),
+    ],
+  });
+}
+
+main().catch(error => {
+  console.error(error.stack);
+  process.exit(1);
+});

yarn.lock

@@ -4623,6 +4623,7 @@ __metadata:
   version: 0.0.0-use.local
   resolution: "@backstage/plugin-catalog-backend-module-incremental-ingestion@workspace:plugins/catalog-backend-module-incremental-ingestion"
   dependencies:
+    "@backstage/backend-app-api": "workspace:^"
     "@backstage/backend-common": "workspace:^"
     "@backstage/backend-plugin-api": "workspace:^"
     "@backstage/backend-tasks": "workspace:^"
@@ -4638,6 +4639,7 @@ __metadata:
     "@types/luxon": ^3.0.0
     express: ^4.17.1
     express-promise-router: ^4.1.0
+    get-port: ^6.1.2
     knex: ^2.0.0
     lodash: ^4.17.21
     luxon: ^3.0.0