Fixes for migrations and routes

Signed-off-by: Damon Kaswell <damon.kaswell1@hp.com>

plugins/incremental-ingestion-backend/migrations/20221116073152_init.js

@@ -18,15 +18,6 @@
  * @param { import("knex").Knex } knex
  */
 exports.up = async function up(knex) {
-  /**
-   * Create an index on the final_entities table for the provider name
-   * annotation
-   */
-
-  await knex.raw(
-    `CREATE INDEX IF NOT EXISTS increment_ingestion_provider_name_idx ON public.final_entities ((final_entity::json #>> '{metadata, annotations, backstage.io/incremental-provider-name}'));`,
-  );
-
   /**
    * Sets up the ingestions table
    */
@@ -57,6 +48,7 @@ exports.up = async function up(knex) {
 
     table
       .timestamp('next_action_at')
+      .notNullable()
       .defaultTo(knex.fn.now())
       .comment('the moment in time at which point ingestion can begin again');
 
@@ -66,11 +58,13 @@ exports.up = async function up(knex) {
 
     table
       .integer('attempts')
+      .notNullable()
       .defaultTo(0)
       .comment('how many attempts have been made to burst without success');
 
     table
       .timestamp('created_at')
+      .notNullable()
       .defaultTo(knex.fn.now())
       .comment('when did this ingestion actually begin');
 
@@ -124,10 +118,14 @@ exports.up = async function up(knex) {
 
     table
       .integer('sequence')
+      .notNullable()
       .defaultTo(0)
       .comment('what is the order of this mark');
 
-    table.timestamp('created_at').defaultTo(knex.fn.now());
+    table
+      .timestamp('created_at')
+      .notNullable()
+      .defaultTo(knex.fn.now());
   });
 
   await knex.schema.alterTable('ingestion_marks', t => {
@@ -177,5 +175,4 @@ exports.down = async function down(knex) {
   await knex.schema.dropTable('ingestion_mark_entities');
   await knex.schema.dropTable('ingestion_marks');
   await knex.schema.dropTable('ingestions');
-  await knex.raw(`DROP INDEX increment_ingestion_provider_name_idx;`);
 };

plugins/incremental-ingestion-backend/src/database/IncrementalIngestionDatabaseManager.ts

@@ -300,10 +300,7 @@ export class IncrementalIngestionDatabaseManager {
           'refresh_state.entity_id',
           'final_entities.entity_id',
         )
-        .whereRaw(
-          `((final_entity::json #>> '{metadata, annotations, ${INCREMENTAL_ENTITY_PROVIDER_ANNOTATION}}')) = ?`,
-          [provider],
-        )
+        .join('search', 'search.entity_id', 'final_entities.entity_id')
         .whereNotIn(
           'entity_ref',
           tx('ingestion_marks')
@@ -314,7 +311,12 @@ export class IncrementalIngestionDatabaseManager {
             )
             .select('ingestion_mark_entities.ref')
             .where('ingestion_marks.ingestion_id', ingestionId),
-        );
+        )
+        .andWhere(
+          'search.key',
+          `metadata.annotations.${INCREMENTAL_ENTITY_PROVIDER_ANNOTATION}`,
+        )
+        .andWhere('search.value', provider);
       return removed.map(entity => {
         return { entity: JSON.parse(entity.entity) };
       });

plugins/incremental-ingestion-backend/src/database/migrations.ts

@@ -16,6 +16,7 @@
 
 import { resolvePackagePath } from '@backstage/backend-common';
 import { Knex } from 'knex';
+import { DB_MIGRATIONS_TABLE } from './tables';
 
 export async function applyDatabaseMigrations(knex: Knex): Promise<void> {
   const migrationsDir = resolvePackagePath(
@@ -25,5 +26,6 @@ export async function applyDatabaseMigrations(knex: Knex): Promise<void> {
 
   await knex.migrate.latest({
     directory: migrationsDir,
+    tableName: DB_MIGRATIONS_TABLE,
   });
 }

plugins/incremental-ingestion-backend/src/database/tables.ts

@@ -0,0 +1,17 @@
+/*
+ * 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.
+ */
+
+export const DB_MIGRATIONS_TABLE = 'plugin_incremental_ingestion_backend__knex_migrations';

plugins/incremental-ingestion-backend/src/router/paths.ts

@@ -0,0 +1,19 @@
+/*
+ * 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.
+ */
+
+export const PROVIDER_CLEANUP = '/providers/cleanup';
+export const PROVIDER_HEALTH = '/providers/health';
+export const PROVIDER_BASE_PATH = '/providers/:provider';

plugins/incremental-ingestion-backend/src/router/routes.ts

@@ -17,7 +17,8 @@ import { errorHandler } from '@backstage/backend-common';
 import express from 'express';
 import Router from 'express-promise-router';
 import { Logger } from 'winston';
-import { IncrementalIngestionDatabaseManager } from './';
+import { IncrementalIngestionDatabaseManager } from '..';
+import { PROVIDER_BASE_PATH, PROVIDER_CLEANUP, PROVIDER_HEALTH } from './paths';
 
 /** @public */
 export const createIncrementalProviderRouter = async (
@@ -28,7 +29,7 @@ export const createIncrementalProviderRouter = async (
   router.use(express.json());
 
   // Get the overall health of all incremental providers
-  router.get('/health', async (_, res) => {
+  router.get(PROVIDER_HEALTH, async (_, res) => {
     const records = await manager.healthcheck();
     const providers = records.map(record => record.provider_name);
     const duplicates = [
@@ -43,13 +44,13 @@ export const createIncrementalProviderRouter = async (
   });
 
   // Clean up and pause all providers
-  router.delete('/cleanup', async (_, res) => {
+  router.post(PROVIDER_CLEANUP, async (_, res) => {
     const result = await manager.cleanupProviders();
     res.json(result);
   });
 
   // Get basic status of the provider
-  router.get('/:provider', async (req, res) => {
+  router.get(PROVIDER_BASE_PATH, async (req, res) => {
     const { provider } = req.params;
     const record = await manager.getCurrentIngestionRecord(provider);
     if (record) {
@@ -84,7 +85,7 @@ export const createIncrementalProviderRouter = async (
   });
 
   // Trigger the provider's next action
-  router.put('/:provider', async (req, res) => {
+  router.put(PROVIDER_BASE_PATH, async (req, res) => {
     const { provider } = req.params;
     const record = await manager.getCurrentIngestionRecord(provider);
     if (record) {
@@ -111,7 +112,7 @@ export const createIncrementalProviderRouter = async (
   });
 
   // Start a brand-new ingestion cycle for the provider
-  router.post('/:provider', async (req, res) => {
+  router.post(PROVIDER_BASE_PATH, async (req, res) => {
     const { provider } = req.params;
 
     const record = await manager.getCurrentIngestionRecord(provider);
@@ -144,7 +145,7 @@ export const createIncrementalProviderRouter = async (
   });
 
   // Stop the provider and pause it for 24 hours
-  router.post('/:provider/cancel', async (req, res) => {
+  router.post(`${PROVIDER_BASE_PATH}/cancel`, async (req, res) => {
     const { provider } = req.params;
     const record = await manager.getCurrentIngestionRecord(provider);
     if (record) {
@@ -178,14 +179,14 @@ export const createIncrementalProviderRouter = async (
   });
 
   // Wipe out all ingestion records for the provider and pause for 24 hours
-  router.delete('/:provider', async (req, res) => {
+  router.delete(PROVIDER_BASE_PATH, async (req, res) => {
     const { provider } = req.params;
     const result = await manager.purgeAndResetProvider(provider);
     res.json(result);
   });
 
   // Get the ingestion marks for the current cycle
-  router.get('/:provider/marks', async (req, res) => {
+  router.get(`${PROVIDER_BASE_PATH}/marks`, async (req, res) => {
     const { provider } = req.params;
     const record = await manager.getCurrentIngestionRecord(provider);
     if (record) {
@@ -213,7 +214,7 @@ export const createIncrementalProviderRouter = async (
     }
   });
 
-  router.delete('/:provider/marks', async (req, res) => {
+  router.delete(`${PROVIDER_BASE_PATH}/marks`, async (req, res) => {
     const { provider } = req.params;
     const deletions = await manager.clearFinishedIngestions(provider);
 

plugins/incremental-ingestion-backend/src/service/IncrementalCatalogBuilder.ts

@@ -24,7 +24,7 @@ import { Knex } from 'knex';
 import { IncrementalIngestionEngine } from '../engine/IncrementalIngestionEngine';
 import { applyDatabaseMigrations } from '../database/migrations';
 import { IncrementalIngestionDatabaseManager } from '../database/IncrementalIngestionDatabaseManager';
-import { createIncrementalProviderRouter } from '../routes';
+import { createIncrementalProviderRouter } from '../router/routes';
 
 class Deferred<T> implements Promise<T> {
   #resolve?: (value: T) => void;