Copy migrations to migrationsv2 before migration

Co-authored-by: Fredrik Adelöw <freben@users.noreply.github.com>
Co-authored-by: Patrik Oldsberg <poldsberg@gmail.com>
Signed-off-by: Johan Haals <johan.haals@gmail.com>

packages/backend/src/plugins/catalog.ts

@@ -27,7 +27,11 @@ import { PluginEnvironment } from '../types';
 export default async function createPlugin(
   env: PluginEnvironment,
 ): Promise<Router> {
-  // HIGHLY experimental rework of the software catalog
+  /*
+   * ** WARNING **
+   * DO NOT enable the experimental catalog, it will brick your database migrations.
+   * This is solely for internal backstage development.
+   */
   if (process.env.EXPERIMENTAL_CATALOG === '1') {
     const builder = new NextCatalogBuilder(env);
     const {

plugins/catalog-backend/migrationsv2/20210302150147_refresh_state.js

@@ -20,52 +20,6 @@
  * @param {import('knex').Knex} knex
  */
 exports.up = async function up(knex) {
-  await knex.schema.createTable('relations', table => {
-    table.comment('All relations between entities in the catalog');
-    table
-      .text('originating_entity_id')
-      .references('entity_id')
-      .inTable('refresh_state')
-      .onDelete('CASCADE')
-      .notNullable()
-      .comment('The entity that provided the relation');
-    table
-      .text('source_entity_ref')
-      .notNullable()
-      .comment('The entity reference of the source entity of the relation');
-    table
-      .text('type')
-      .notNullable()
-      .comment('The type of the relation between the entities');
-    table
-      .text('target_entity_ref')
-      .notNullable()
-      .comment('The entity reference of the target entity of the relation');
-
-    table.index(
-      ['source_entity_ref', 'type', 'target_entity_ref'],
-      'relations_full_ref_idx',
-    );
-  });
-
-  await knex.schema.createTable('final_entities', table => {
-    table.comment(
-      'This table contains the final entity result after processing and stitching',
-    );
-    table
-      .text('entity_id')
-      .primary()
-      .notNullable()
-      .references('entity_id')
-      .inTable('refresh_state')
-      .onDelete('CASCADE')
-      .comment(
-        'Entity ID which correspond to the ID in the refresh_state table',
-      );
-
-    table.text('finalized_entity').notNullable().comment('The final entity');
-  });
-
   await knex.schema.createTable('refresh_state', table => {
     table.comment(
       'Location refresh states. Every individual location (that was ever directly or indirectly discovered) and entity has an entry in this table. It therefore represents the entire live set of things that the refresh loop considers.',
@@ -116,6 +70,24 @@ exports.up = async function up(knex) {
     table.index('next_update_at', 'refresh_state_next_update_at_idx');
   });
 
+  await knex.schema.createTable('final_entities', table => {
+    table.comment(
+      'This table contains the final entity result after processing and stitching',
+    );
+    table
+      .text('entity_id')
+      .primary()
+      .notNullable()
+      .references('entity_id')
+      .inTable('refresh_state')
+      .onDelete('CASCADE')
+      .comment(
+        'Entity ID which correspond to the ID in the refresh_state table',
+      );
+    table.text('etag').notNullable().comment('Etag to be used for caching');
+    table.text('finalized_entity').notNullable().comment('The final entity');
+  });
+
   await knex.schema.createTable('refresh_state_references', table => {
     table.comment(
       'Holds edges between refresh state rows. Every time when an entity is processed and emits another entity, an edge will be stored to represent that fact. This is used to detect orphans and ultimately deletions.',
@@ -155,6 +127,34 @@ exports.up = async function up(knex) {
       'refresh_state_references_target_entity_id_idx',
     );
   });
+
+  await knex.schema.createTable('relations', table => {
+    table.comment('All relations between entities in the catalog');
+    table
+      .text('originating_entity_id')
+      .references('entity_id')
+      .inTable('refresh_state')
+      .onDelete('CASCADE')
+      .notNullable()
+      .comment('The entity that provided the relation');
+    table
+      .text('source_entity_ref')
+      .notNullable()
+      .comment('The entity reference of the source entity of the relation');
+    table
+      .text('type')
+      .notNullable()
+      .comment('The type of the relation between the entities');
+    table
+      .text('target_entity_ref')
+      .notNullable()
+      .comment('The entity reference of the target entity of the relation');
+
+    table.index(
+      ['source_entity_ref', 'type', 'target_entity_ref'],
+      'relations_full_ref_idx',
+    );
+  });
 };
 
 /**

plugins/catalog-backend/package.json

@@ -80,6 +80,7 @@
   "files": [
     "dist",
     "migrations/**/*.{js,d.ts}",
+    "migrationsv2/**/*.{js,d.ts}",
     "config.d.ts"
   ],
   "configSchema": "config.d.ts"

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

@@ -14,7 +14,12 @@
  * limitations under the License.
  */
 
-import { PluginDatabaseManager, UrlReader } from '@backstage/backend-common';
+import {
+  PluginDatabaseManager,
+  resolvePackagePath,
+  UrlReader,
+} from '@backstage/backend-common';
+import fs from 'fs-extra';
 import {
   DefaultNamespaceEntityPolicy,
   EntityPolicies,
@@ -30,12 +35,10 @@ import { ScmIntegrations } from '@backstage/integration';
 import lodash from 'lodash';
 import { Logger } from 'winston';
 import {
-  DatabaseEntitiesCatalog,
   DatabaseLocationsCatalog,
   EntitiesCatalog,
   LocationsCatalog,
 } from '../catalog';
-import { DatabaseManager } from '../database';
 import {
   AnnotateLocationEntityProcessor,
   BitbucketDiscoveryProcessor,
@@ -68,7 +71,9 @@ import { DatabaseLocationProvider } from '../next/DatabaseLocationProvider';
 import { LocationStoreImpl } from '../next/LocationStoreImpl';
 import { ProcessingStateManagerImpl } from '../next/ProcessingStateManagerImpl';
 import { CatalogProcessingEngine } from '../next/types';
+import { NextEntitiesCatalog } from './NextEntitiesCatalog';
 import { Stitcher } from './Stitcher';
+import { CommonDatabase } from '../database/CommonDatabase';
 
 export type CatalogEnvironment = {
   logger: Logger;
@@ -239,7 +244,20 @@ export class NextCatalogBuilder {
     const parser = this.parser || defaultEntityDataParser;
 
     const dbClient = await database.getClient();
-    const db = await DatabaseManager.createDatabase(dbClient, { logger });
+    const allMigrations = resolvePackagePath(
+      '@backstage/plugin-catalog-backend',
+      'migrations',
+    );
+
+    const migrationsDir = resolvePackagePath(
+      '@backstage/plugin-catalog-backend',
+      'migrationsv2',
+    );
+    await fs.copy(allMigrations, migrationsDir);
+    await dbClient.migrate.latest({
+      directory: migrationsDir,
+    });
+    const db = new CommonDatabase(dbClient, logger);
 
     const processingDatabase = new ProcessingDatabaseImpl(dbClient, logger);
     const stateManager = new ProcessingStateManagerImpl(processingDatabase);
@@ -251,7 +269,7 @@ export class NextCatalogBuilder {
       parser,
       policy,
     });
-    const entitiesCatalog = new DatabaseEntitiesCatalog(db, logger);
+    const entitiesCatalog = new NextEntitiesCatalog(dbClient, logger);
 
     const locationStore = new LocationStoreImpl(db);
     const dbLocationProvider = new DatabaseLocationProvider(locationStore);

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

@@ -0,0 +1,61 @@
+/*
+ * Copyright 2020 Spotify AB
+ *
+ * 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 { Logger } from 'winston';
+import { Knex } from 'knex';
+import { DbFinalEntitiesRow } from './Stitcher';
+import { EntitiesCatalog } from '../catalog';
+import { EntitiesRequest, EntitiesResponse } from '../catalog/types';
+
+export class NextEntitiesCatalog implements EntitiesCatalog {
+  constructor(
+    private readonly database: Knex,
+    private readonly logger: Logger,
+  ) {}
+
+  async entities(request?: EntitiesRequest): Promise<EntitiesResponse> {
+    if (request?.fields) {
+      throw new Error('Fields extraction is not implemented');
+    }
+    if (request?.pagination) {
+      throw new Error('Pagination is not implemented');
+    }
+    if (request?.filter) {
+      throw new Error('Filters are not implemented');
+    }
+
+    const dbResponse = await this.database<DbFinalEntitiesRow>(
+      'final_entities',
+    ).select();
+
+    const entities = dbResponse.map(e => JSON.parse(e.finalized_entity));
+
+    return {
+      entities,
+      pageInfo: {
+        hasNextPage: false,
+      },
+    };
+  }
+
+  async removeEntityByUid(_uid: string): Promise<void> {
+    throw new Error('Not implemented');
+  }
+
+  async batchAddOrUpdateEntities(): Promise<never> {
+    throw new Error('Not implemented');
+  }
+}

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

@@ -1,5 +1,3 @@
-import { Transaction } from '../database';
-
 /*
  * Copyright 2021 Spotify AB
  *
@@ -18,7 +16,28 @@ import { Transaction } from '../database';
 
 import { Knex } from 'knex';
 import { Logger } from 'winston';
+import { Transaction } from '../database';
 import { ConflictError } from '@backstage/errors';
+import {
+  DbRefreshStateReferences,
+  DbRefreshStateRow,
+  DbRelationsRow,
+} from './database/ProcessingDatabaseImpl';
+import { Entity, parseEntityRef } from '@backstage/catalog-model';
+import { createHash } from 'crypto';
+import stableStringify from 'fast-json-stable-stringify';
+
+export type DbFinalEntitiesRow = {
+  entity_id: string;
+  etag: string;
+  finalized_entity: string;
+};
+
+function generateEntityEtag(entity: Entity) {
+  return createHash('sha1')
+    .update(stableStringify({ ...entity }))
+    .digest('hex');
+}
 
 export class Stitcher {
   constructor(
@@ -27,7 +46,74 @@ export class Stitcher {
   ) {}
 
   async stitch(entityRefs: Set<string>) {
-    console.log(entityRefs);
+    for (const entityRef of entityRefs) {
+      await this.transaction(async txOpaque => {
+        const tx = txOpaque as Knex.Transaction;
+        const [result] = await tx<DbRefreshStateRow>('refresh_state')
+          .select('entity_id', 'processed_entity')
+          .where({ entity_ref: entityRef });
+
+        if (!result) {
+          this.logger.debug(
+            `Unable to stitch ${entityRef}, item does not exist in refresh state table`,
+          );
+          return;
+        } else if (!result.processed_entity) {
+          this.logger.debug(
+            `Unable to stitch ${entityRef}, the entity has not yet been processed`,
+          );
+          return;
+        }
+
+        const entity: Entity = JSON.parse(result.processed_entity);
+
+        const entityId = entity?.metadata?.uid;
+        if (!entityId) {
+          this.logger.error(`missing ID in entity ${JSON.stringify(entity)}`);
+          return;
+        }
+
+        const [reference_count_result] = await tx<DbRefreshStateReferences>(
+          'refresh_state_references',
+        )
+          .where({ target_entity_id: entity.metadata.uid })
+          .count({ reference_count: 'target_entity_id' });
+        console.log(
+          'DEBUG: reference_count_result =',
+          reference_count_result,
+          entityId,
+        );
+        if (Number(reference_count_result.reference_count) === 0) {
+          this.logger.debug(`${entityRef} is orphan`);
+          entity.metadata.annotations = {
+            ...entity.metadata.annotations,
+            ['backstage.io/orphan']: 'true',
+          };
+        }
+
+        const relationResults = await tx<DbRelationsRow>('relations')
+          .where({ originating_entity_id: entityId })
+          .select();
+
+        // TODO: entityRef is lower case and should be uppercase in the final result.
+        entity.relations = relationResults.map(relation => ({
+          type: relation.type,
+          target: parseEntityRef(relation.target_entity_ref),
+        }));
+        entity.metadata.generation = 1;
+        const etag = generateEntityEtag(entity);
+        entity.metadata.etag = etag;
+        console.log(JSON.stringify(entity, null, 2));
+        await tx<DbFinalEntitiesRow>('final_entities')
+          .insert({
+            finalized_entity: JSON.stringify(entity),
+            entity_id: entityId,
+            etag,
+          })
+          .onConflict('entity_id')
+          .merge(['finalized_entity', 'etag']);
+      });
+    }
   }
 
   private async transaction<T>(
@@ -45,6 +131,10 @@ export class Stitcher {
         {
           // If we explicitly trigger a rollback, don't fail.
           doNotRejectOnRollback: true,
+          isolationLevel:
+            this.database.client.config.client === 'sqlite3'
+              ? undefined // sqlite3 only supports serializable transactions, ignoring the isolation level param
+              : 'serializable',
         },
       );
 

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

@@ -27,8 +27,6 @@ import {
 } from './types';
 import type { Logger } from 'winston';
 import { Entity, stringifyEntityRef } from '@backstage/catalog-model';
-import { createHash } from 'crypto';
-import stableStringify from 'fast-json-stable-stringify';
 import { v4 as uuid } from 'uuid';
 
 export type DbRefreshStateRow = {
@@ -55,12 +53,6 @@ export type DbRefreshStateReferences = {
   target_entity_id: string;
 };
 
-function generateEntityEtag(entity: Entity) {
-  return createHash('sha1')
-    .update(stableStringify({ ...entity }))
-    .digest('hex');
-}
-
 // The number of items that are sent per batch to the database layer, when
 // doing .batchInsert calls to knex. This needs to be low enough to not cause
 // errors in the underlying engine due to exceeding query limits, but large
@@ -108,9 +100,6 @@ export class ProcessingDatabaseImpl implements ProcessingDatabase {
 
     // Update fragments
 
-    // Update relations
-    const entityRef = stringifyEntityRef(processedEntity);
-
     // Delete old relations
     await tx<DbRelationsRow>('relations')
       .where({ originating_entity_id: id })

tsconfig.json

@@ -4,7 +4,8 @@
     "packages/*/src",
     "plugins/*/src",
     "plugins/*/dev",
-    "plugins/*/migrations"
+    "plugins/*/migrations",
+    "plugins/catalog-backend/migrationsv2"
   ],
   "compilerOptions": {
     "outDir": "dist-types",