chore: worked on some more of the new catalog migration with fixing deps

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

Signed-off-by: blam <ben@blam.sh>

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

@@ -59,13 +59,14 @@ exports.up = async function up(knex) {
       .notNullable()
       .comment('JSON array containing all errors related to entity');
     table
-      .dateTime('next_update_at')
+      .dateTime('next_update_at') // TOOD: timezone or change to epoch-millis or similar
       .notNullable()
       .comment('Timestamp of when entity should be updated');
     table
-      .dateTime('last_discovery_at')
+      .dateTime('last_discovery_at') // TOOD: timezone or change to epoch-millis or similar
       .notNullable()
       .comment('The last timestamp of which this entity was discovered');
+    table.index('entity_id', 'refresh_state_entity_id_idx');
     table.index('entity_ref', 'refresh_state_entity_ref_idx');
     table.index('next_update_at', 'refresh_state_next_update_at_idx');
   });
@@ -94,38 +95,35 @@ exports.up = async function up(knex) {
       '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.',
     );
     table
-      .text('source_special_key')
+      .text('source_key')
       .nullable()
       .comment(
         'When the reference source is not an entity, this is an opaque identifier for that source.',
       );
     table
-      .text('source_entity_id')
+      .text('source_entity_ref')
       .nullable()
-      .references('entity_id')
+      .references('entity_ref')
       .inTable('refresh_state')
       .onDelete('CASCADE')
       .comment(
-        'When the reference source is an entity, this is the ID of the source entity.',
+        'When the reference source is an entity, this is the EntityRef of the source entity.',
       );
     table
-      .text('target_entity_id')
+      .text('target_entity_ref')
       .notNullable()
-      .references('entity_id')
+      .references('entity_ref')
       .inTable('refresh_state')
       .onDelete('CASCADE')
-      .comment('The ID of the target entity.');
+      .comment('The EntityRef of the target entity.');
+    table.index('source_key', 'refresh_state_references_source_key_idx');
     table.index(
-      'source_special_key',
-      'refresh_state_references_source_special_key_idx',
+      'source_entity_ref',
+      'refresh_state_references_source_entity_ref_idx',
     );
     table.index(
-      'source_entity_id',
-      'refresh_state_references_source_entity_id_idx',
-    );
-    table.index(
-      'target_entity_id',
-      'refresh_state_references_target_entity_id_idx',
+      'target_entity_ref',
+      'refresh_state_references_target_entity_ref_idx',
     );
   });
 
@@ -165,6 +163,7 @@ exports.down = async function down(knex) {
     table.dropIndex([], 'refresh_state_references_target_entity_id_idx');
   });
   await knex.schema.alterTable('refresh_state', table => {
+    table.dropIndex([], 'refresh_state_entity_id_idx');
     table.dropIndex([], 'refresh_state_entity_ref_idx');
     table.dropIndex([], 'refresh_state_next_update_at_idx');
   });

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

@@ -30,9 +30,31 @@ import { stringifyEntityRef } from '@backstage/catalog-model';
 import { Stitcher } from './Stitcher';
 
 class Connection implements EntityProviderConnection {
-  constructor(private readonly stateManager: ProcessingStateManager) {}
+  constructor(
+    private readonly config: {
+      stateManager: ProcessingStateManager;
+      id: string;
+    },
+  ) {}
 
-  async applyMutation(mutation: EntityProviderMutation): Promise<void> {}
+  async applyMutation(mutation: EntityProviderMutation): Promise<void> {
+    if (mutation.type === 'full') {
+      await this.config.stateManager.replaceProcessingItems({
+        id: this.config.id,
+        type: 'full',
+        items: mutation.entities,
+      });
+
+      return;
+    }
+
+    await this.config.stateManager.replaceProcessingItems({
+      id: this.config.id,
+      type: 'delta',
+      added: mutation.added,
+      removed: mutation.removed,
+    });
+  }
 }
 
 export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
@@ -48,7 +70,9 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
 
   async start() {
     for (const provider of this.entityProviders) {
-      provider.connect(new Connection(this.stateManager));
+      // TODO: this ID should be some form of identifier for the EntityProvider
+      const id = 'databaseProvider';
+      provider.connect(new Connection({ stateManager: this.stateManager, id }));
     }
 
     this.running = true;
@@ -57,12 +81,12 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
       const {
         id,
         entity,
-        state: intialState,
+        state: initialState,
       } = await this.stateManager.getNextProcessingItem();
 
       const result = await this.orchestrator.process({
         entity,
-        state: intialState,
+        state: initialState,
       });
 
       for (const error of result.errors) {
@@ -95,10 +119,6 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
 
   async stop() {
     this.running = false;
-
-    for (const subscription of this.subscriptions) {
-      subscription.unsubscribe();
-    }
   }
 
   private async onNext(id: string, message: EntityMessage) {

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

@@ -35,10 +35,6 @@ export class DefaultLocationStore implements LocationStore, EntityProvider {
   constructor(private readonly db: Database) {}
 
   createLocation(spec: LocationSpec): Promise<Location> {
-    if (!this.connection) {
-      throw new Error('location store is not initialized');
-    }
-
     return this.db.transaction(async tx => {
       // TODO: id should really be type and target combined and not a uuid.
 

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

@@ -17,14 +17,23 @@
 import { ProcessingDatabase, RefreshStateItem } from './database/types';
 import {
   AddProcessingItemRequest,
-  ProccessingItem,
+  ProcessingItem,
   ProcessingItemResult,
   ProcessingStateManager,
+  ReplaceProcessingItemsRequest,
 } from './types';
 
 export class DefaultProcessingStateManager implements ProcessingStateManager {
   constructor(private readonly db: ProcessingDatabase) {}
 
+  replaceProcessingItems(
+    request: ReplaceProcessingItemsRequest,
+  ): Promise<void> {
+    return this.db.transaction(async tx => {
+      await this.db.replaceUnprocessedEntities(tx, request);
+    });
+  }
+
   async setProcessingItemResult(result: ProcessingItemResult) {
     return this.db.transaction(async tx => {
       await this.db.updateProcessedEntity(tx, {
@@ -44,7 +53,7 @@ export class DefaultProcessingStateManager implements ProcessingStateManager {
     });
   }
 
-  async getNextProcessingItem(): Promise<ProccessingItem> {
+  async getNextProcessingItem(): Promise<ProcessingItem> {
     const entities = await new Promise<RefreshStateItem[]>(resolve =>
       this.popFromQueue(resolve),
     );

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

@@ -0,0 +1,114 @@
+/*
+ * 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 { getVoidLogger, resolvePackagePath } from '@backstage/backend-common';
+import knexFactory, { Knex } from 'knex';
+import { v4 as uuidv4 } from 'uuid';
+import { Logger } from 'winston';
+import { CommonDatabase } from '../../database/CommonDatabase';
+import { Database } from '../../database/types';
+import fs from 'fs-extra';
+
+export type CreateDatabaseOptions = {
+  logger: Logger;
+};
+
+const defaultOptions: CreateDatabaseOptions = {
+  logger: getVoidLogger(),
+};
+
+export class DatabaseManager {
+  public static async createDatabase(
+    knex: Knex,
+    options: Partial<CreateDatabaseOptions> = {},
+  ): Promise<Database> {
+    const allMigrations = resolvePackagePath(
+      '@backstage/plugin-catalog-backend',
+      'migrations',
+    );
+
+    const migrationsDir = resolvePackagePath(
+      '@backstage/plugin-catalog-backend',
+      'migrationsv2',
+    );
+    await fs.copy(allMigrations, migrationsDir);
+
+    await knex.migrate.latest({
+      directory: migrationsDir,
+    });
+    const { logger } = { ...defaultOptions, ...options };
+    return new CommonDatabase(knex, logger);
+  }
+
+  public static async createInMemoryDatabase(): Promise<Database> {
+    const knex = await this.createInMemoryDatabaseConnection();
+    return await this.createDatabase(knex);
+  }
+
+  public static async createInMemoryDatabaseConnection(): Promise<Knex> {
+    const knex = knexFactory({
+      client: 'sqlite3',
+      connection: ':memory:',
+      useNullAsDefault: true,
+    });
+
+    knex.client.pool.on('createSuccess', (_eventId: any, resource: any) => {
+      resource.run('PRAGMA foreign_keys = ON', () => {});
+    });
+
+    return knex;
+  }
+
+  public static async createTestDatabase(): Promise<Database> {
+    const knex = await this.createTestDatabaseConnection();
+    return await this.createDatabase(knex);
+  }
+
+  public static async createTestDatabaseConnection(): Promise<Knex> {
+    const config: Knex.Config<any> = {
+      /*
+      client: 'pg',
+      connection: {
+        host: 'localhost',
+        user: 'postgres',
+        password: 'postgres',
+      },
+      */
+      client: 'sqlite3',
+      connection: ':memory:',
+      useNullAsDefault: true,
+    };
+
+    let knex = knexFactory(config);
+    if (typeof config.connection !== 'string') {
+      const tempDbName = `d${uuidv4().replace(/-/g, '')}`;
+      await knex.raw(`CREATE DATABASE ${tempDbName};`);
+      knex = knexFactory({
+        ...config,
+        connection: {
+          ...config.connection,
+          database: tempDbName,
+        },
+      });
+    }
+
+    knex.client.pool.on('createSuccess', (_eventId: any, resource: any) => {
+      resource.run('PRAGMA foreign_keys = ON', () => {});
+    });
+
+    return knex;
+  }
+}

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

@@ -0,0 +1,31 @@
+/*
+ * Copyright 2021 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 { DefaultProcessingDatabase } from './DefaultProcessingDatabase';
+import { DatabaseManager } from './DatabaseManager';
+import { Knex } from 'knex';
+
+describe('Default Processing Database', () => {
+  let db: Knex | undefined;
+
+  beforeEach(async () => {
+    db = await DatabaseManager.createTestDatabaseConnection();
+    await DatabaseManager.createDatabase(db);
+  });
+
+  it('should write some stuff', async () => {
+    await db('refresh_state_referencess').select();
+  });
+});

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

@@ -15,6 +15,7 @@
  */
 
 import { ConflictError, NotFoundError } from '@backstage/errors';
+import { stringifyEntityRef, Entity } from '@backstage/catalog-model';
 import { Knex } from 'knex';
 import { Transaction } from '../../database';
 import lodash from 'lodash';
@@ -24,20 +25,21 @@ import {
   AddUnprocessedEntitiesOptions,
   UpdateProcessedEntityOptions,
   GetProcessableEntitiesResult,
+  ReplaceUnprocessedEntitiesOptions,
 } from './types';
 import type { Logger } from 'winston';
-import { Entity, stringifyEntityRef } from '@backstage/catalog-model';
+
 import { v4 as uuid } from 'uuid';
 
 export type DbRefreshStateRow = {
   entity_id: string;
   entity_ref: string;
   unprocessed_entity: string;
-  processed_entity: string;
-  cache: string;
+  processed_entity?: string;
+  cache?: string;
   next_update_at: string;
   last_discovery_at: string; // remove?
-  errors: string;
+  errors?: string;
 };
 
 export type DbRelationsRow = {
@@ -47,10 +49,10 @@ export type DbRelationsRow = {
   type: string;
 };
 
-export type DbRefreshStateReferences = {
-  source_special_key?: string;
-  source_entity_id?: string;
-  target_entity_id: string;
+export type DbRefreshStateReferencesRow = {
+  source_key?: string;
+  source_entity_ref?: string;
+  target_entity_ref: string;
 };
 
 // The number of items that are sent per batch to the database layer, when
@@ -128,6 +130,164 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
     );
   }
 
+  async replaceUnprocessedEntities(
+    txOpaque: Transaction,
+    options: ReplaceUnprocessedEntitiesOptions,
+  ): Promise<void> {
+    const tx = txOpaque as Knex.Transaction;
+    const entityIds = new Array<string>();
+
+    if (options.type === 'full') {
+      const oldRefs = await tx<DbRefreshStateReferencesRow>(
+        'refresh_state_references',
+      )
+        .where({ source_key: options.sourceKey })
+        .select('target_entity_ref')
+        .then(rows => rows.map(r => r.target_entity_ref));
+
+      const items = options.items.map(entity => ({
+        entity,
+        ref: stringifyEntityRef(entity),
+        id: uuid(),
+      }));
+
+      const oldRefsSet = new Set(oldRefs);
+      const newRefsSet = new Set(items.map(item => item.ref));
+      const toAdd = items.filter(item => !oldRefsSet.has(item.ref));
+      const toRemove = oldRefs.filter(ref => !newRefsSet.has(ref));
+
+      if (toRemove.length) {
+        // TODO(freben): Batch split, to not hit variable limits?
+
+        // get all refs where source = any of the toRemove
+        // delete all state rows matching any of the toRemove
+        // revisit the targets of all of those refs
+        // verify if they have references, otherwise delete from refresh_state
+
+        const current = [...toRemove];
+        while (true) {
+
+          tx.withRecursive('r', function refs() {
+            return tx.select({  })
+              .from({ r1: 'refresh_state_references' })
+              .where({ source_key: options.sourceKey })
+              .whereIn('target_entity_ref', toRemove)
+              .unionAll(function recurse() {
+                return tx.select({  })
+                  .from({ r2: 'refresh_state_references' })
+                  .whereNotExists()
+              })
+          })
+          .select().from('refs');
+
+          const nextLayerToRemove = await tx<DbRefreshStateReferencesRow>(
+            'refresh_state_references',
+          )
+            .whereIn('source_entity_ref', toRemove)
+            .leftOuterJoin('refresh_state_references AS b', function f() {
+
+            })
+
+            .whereNotNull('b.source_target_ref')
+            .select('target_entity_ref');
+
+          await tx<DbRefreshStateRow>('refresh_state')
+            .whereIn('entity_ref', toRemove)
+            .delete();
+
+          const refsThatStillHaveASourcePointingAtThe = await tx<DbRefreshStateReferencesRow>(
+            'refresh_state_references',
+          )
+            .whereIn('target_entity_ref', nextLayerTargetRefs)
+            .groupBy('target_entity_ref')
+            .select('target_entity_ref');
+
+      }
+
+
+
+
+      if (toAdd.length) {
+        const state: Knex.DbRecord<DbRefreshStateRow>[] = toAdd.map(item => ({
+          entity_id: item.id,
+          entity_ref: item.ref,
+          unprocessed_entity: JSON.stringify(item.entity),
+          errors: '',
+          next_update_at: tx.fn.now(),
+          last_discovery_at: tx.fn.now(),
+        }));
+        const stateReferences: DbRefreshStateReferencesRow[] = toAdd.map(
+          item => ({
+            source_key: options.sourceKey,
+            target_entity_ref: item.ref,
+          }),
+        );
+        // TODO(freben): Concurrency? If we did these one by one, a .onConflict().merge would have made sense
+        await tx.batchInsert('refresh_state', state, BATCH_SIZE);
+        await tx.batchInsert('refresh_state_references', stateReferences, BATCH_SIZE);
+      }
+
+
+      for (const entity of options.items) {
+        const entityRef = stringifyEntityRef(entity);
+        await tx<DbRefreshStateRow>('refresh_state')
+          .insert({
+            entity_id: uuid(),
+            entity_ref: entityRef,
+            unprocessed_entity: JSON.stringify(entity),
+            errors: '',
+            next_update_at: tx.fn.now(),
+            last_discovery_at: tx.fn.now(),
+          })
+          .onConflict('entity_ref')
+          .merge(['unprocessed_entity', 'last_discovery_at']);
+
+        const [{ entity_id: entityId }] = await tx<DbRefreshStateRow>(
+          'refresh_state',
+        ).where({ entity_ref: entityRef });
+        entityIds.push(entityId);
+      }
+      const referenceRows: DbRefreshStateReferencesRow[] = entityIds.map(
+        entityId => ({
+          source_key: options.sourceKey,
+          target_entityRef: entityRef,
+        }),
+      );
+      await tx.batchInsert(
+        'refresh_state_references',
+        referenceRows,
+        BATCH_SIZE,
+      );
+      return;
+    }
+
+    for (const entity of options.removed) {
+      const entityRef = stringifyEntityRef(entity);
+      const [result] = await tx<DbRefreshStateRow>('refresh_state')
+        .where({ entity_ref: entityRef })
+        .select('entity_id');
+
+      if (!result) {
+        this.logger.info(
+          `Unable to delete entity '${entityRef}', entity does not exist in refresh state`,
+        );
+        continue;
+      }
+
+      const referenceResults = await tx<DbRefreshStateReferencesRow>(
+        'refresh_state_references',
+      )
+        // todo correct key?
+        .where({ source_entity_id: result.entity_id })
+        .select();
+
+      await tx<DbRefreshStateReferencesRow>('refresh_state_references')
+        // todo correct key?
+        .where({ source_entity_id: result.entity_id })
+        .delete();
+    }
+  }
+
   async addUnprocessedEntities(
     txOpaque: Transaction,
     options: AddUnprocessedEntitiesOptions,
@@ -160,11 +320,11 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
         ? { source_special_key: options.id }
         : { source_entity_id: options.id };
     // copied from update refs
-    await tx<DbRefreshStateReferences>('refresh_state_references')
+    await tx<DbRefreshStateReferencesRow>('refresh_state_references')
       .where(key)
       .delete();
 
-    const referenceRows: DbRefreshStateReferences[] = entityIds.map(
+    const referenceRows: DbRefreshStateReferencesRow[] = entityIds.map(
       entityId => ({
         ...key,
         target_entity_id: entityId,

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

@@ -50,6 +50,19 @@ export type GetProcessableEntitiesResult = {
   items: RefreshStateItem[];
 };
 
+export type ReplaceUnprocessedEntitiesOptions =
+  | {
+      sourceKey: string;
+      items: Entity[];
+      type: 'full';
+    }
+  | {
+      sourceKey: string;
+      added: Entity[];
+      removed: Entity[];
+      type: 'delta';
+    };
+
 export interface ProcessingDatabase {
   transaction<T>(fn: (tx: Transaction) => Promise<T>): Promise<T>;
 
@@ -58,6 +71,10 @@ export interface ProcessingDatabase {
     options: AddUnprocessedEntitiesOptions,
   ): Promise<void>;
 
+  replaceUnprocessedEntities(
+    txOpaque: Transaction,
+    options: ReplaceUnprocessedEntitiesOptions,
+  ): Promise<void>;
   getProcessableEntities(
     txOpaque: Transaction,
     request: { processBatchSize: number },

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

@@ -58,8 +58,8 @@ export interface CatalogProcessingEngine {
 }
 
 export type EntityProviderMutation =
-  | { type: 'full'; entities: Iterable<Entity> }
-  | { type: 'delta'; added: Iterable<Entity>; removed: Iterable<Entity> };
+  | { type: 'full'; entities: Entity[] }
+  | { type: 'delta'; added: Entity[]; removed: Entity[] };
 
 export interface EntityProviderConnection {
   applyMutation(mutation: EntityProviderMutation): Promise<void>;
@@ -108,14 +108,27 @@ export type AddProcessingItemRequest = {
   entities: Entity[];
 };
 
-export type ProccessingItem = {
+export type ProcessingItem = {
   id: string;
   entity: Entity;
   state: Map<string, JsonObject>;
 };
 
+export type ReplaceProcessingItemsRequest =
+  | {
+      id: string;
+      items: Entity[];
+      type: 'full';
+    }
+  | {
+      id: string;
+      added: Entity[];
+      removed: Entity[];
+      type: 'delta';
+    };
 export interface ProcessingStateManager {
   setProcessingItemResult(result: ProcessingItemResult): Promise<void>;
-  getNextProcessingItem(): Promise<ProccessingItem>;
+  getNextProcessingItem(): Promise<ProcessingItem>;
   addProcessingItems(request: AddProcessingItemRequest): Promise<void>;
+  replaceProcessingItems(request: ReplaceProcessingItemsRequest): Promise<void>;
 }