Add relations and stitcher class

Signed-off-by: Johan Haals <johan.haals@gmail.com>

plugins/catalog-backend/package.json

@@ -47,6 +47,7 @@
     "cross-fetch": "^3.0.6",
     "express": "^4.17.1",
     "express-promise-router": "^4.1.0",
+    "fast-json-stable-stringify": "^2.1.0",
     "fs-extra": "^9.0.0",
     "git-url-parse": "^11.4.4",
     "glob": "^7.1.6",

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

@@ -23,8 +23,13 @@ import {
   CatalogProcessingOrchestrator,
 } from './types';
 
-import { EntitiesCatalog } from '../catalog/types';
 import { Logger } from 'winston';
+import {
+  Entity,
+  stringifyEntityRef,
+  EntityRelationSpec,
+} from '@backstage/catalog-model';
+import { Stitcher } from './Stitcher';
 
 export class CatalogProcessingEngineImpl implements CatalogProcessingEngine {
   private subscriptions: Subscription[] = [];
@@ -35,7 +40,7 @@ export class CatalogProcessingEngineImpl implements CatalogProcessingEngine {
     private readonly entityProviders: EntityProvider[],
     private readonly stateManager: ProcessingStateManager,
     private readonly orchestrator: CatalogProcessingOrchestrator,
-    private readonly entitiesCatalog: EntitiesCatalog,
+    private readonly stitcher: Stitcher,
   ) {}
 
   async start() {
@@ -77,6 +82,14 @@ export class CatalogProcessingEngineImpl implements CatalogProcessingEngine {
       await this.stateManager.addProcessingItems({
         entities: result.deferredEntites,
       });
+
+      const setOfThingsToStitch = new Set<string>([
+        stringifyEntityRef(result.completedEntity),
+        ...result.relations.map(relation =>
+          stringifyEntityRef(relation.source),
+        ),
+      ]);
+      await this.stitcher.stitch(setOfThingsToStitch);
     }
   }
 

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

@@ -46,18 +46,13 @@ import {
   FileReaderProcessor,
   GithubDiscoveryProcessor,
   GithubOrgReaderProcessor,
-  HigherOrderOperation,
-  HigherOrderOperations,
   LdapOrgReaderProcessor,
-  LocationEntityProcessor,
-  LocationReaders,
   MicrosoftGraphOrgReaderProcessor,
   PlaceholderProcessor,
   PlaceholderResolver,
   StaticLocationProcessor,
   UrlReaderProcessor,
 } from '../ingestion';
-import { CatalogRulesEnforcer } from '../ingestion/CatalogRules';
 import { RepoLocationAnalyzer } from '../ingestion/LocationAnalyzer';
 import {
   jsonPlaceholderResolver,
@@ -73,6 +68,7 @@ import { DatabaseLocationProvider } from '../next/DatabaseLocationProvider';
 import { LocationStoreImpl } from '../next/LocationStoreImpl';
 import { ProcessingStateManagerImpl } from '../next/ProcessingStateManagerImpl';
 import { CatalogProcessingEngine } from '../next/types';
+import { Stitcher } from './Stitcher';
 
 export type CatalogEnvironment = {
   logger: Logger;
@@ -255,16 +251,17 @@ export class NextCatalogBuilder {
       parser,
       policy,
     });
-    const entitiesCatalog = new DatabaseEntitiesCatalog(db, this.env.logger);
+    const entitiesCatalog = new DatabaseEntitiesCatalog(db, logger);
 
     const locationStore = new LocationStoreImpl(db);
     const dbLocationProvider = new DatabaseLocationProvider(locationStore);
+    const stitcher = new Stitcher(dbClient, logger);
     const processingEngine = new CatalogProcessingEngineImpl(
       logger,
       [dbLocationProvider], // entityproviders
       stateManager,
       orchestrator,
-      entitiesCatalog,
+      stitcher,
     );
 
     const locationsCatalog = new DatabaseLocationsCatalog(db);

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

@@ -0,0 +1,65 @@
+import { Transaction } from '../database';
+
+/*
+ * 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 { Knex } from 'knex';
+import { Logger } from 'winston';
+import { ConflictError } from '@backstage/errors';
+
+export class Stitcher {
+  constructor(
+    private readonly database: Knex,
+    private readonly logger: Logger,
+  ) {}
+
+  async stitch(entityRefs: Set<string>) {
+    console.log(entityRefs);
+  }
+
+  private async transaction<T>(
+    fn: (tx: Transaction) => Promise<T>,
+  ): Promise<T> {
+    try {
+      let result: T | undefined = undefined;
+
+      await this.database.transaction(
+        async tx => {
+          // We can't return here, as knex swallows the return type in case the transaction is rolled back:
+          // https://github.com/knex/knex/blob/e37aeaa31c8ef9c1b07d2e4d3ec6607e557d800d/lib/transaction.js#L136
+          result = await fn(tx);
+        },
+        {
+          // If we explicitly trigger a rollback, don't fail.
+          doNotRejectOnRollback: true,
+        },
+      );
+
+      return result!;
+    } catch (e) {
+      this.logger.debug(`Error during transaction, ${e}`);
+
+      if (
+        /SQLITE_CONSTRAINT: UNIQUE/.test(e.message) ||
+        /unique constraint/.test(e.message)
+      ) {
+        throw new ConflictError(`Rejected due to a conflicting entity`, e);
+      }
+
+      throw e;
+    }
+  }
+}

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

@@ -14,17 +14,21 @@
  * limitations under the License.
  */
 
-import { ConflictError, NotFoundError } from '@backstage/errors';
+import { ConflictError, NotFoundError, InputError } from '@backstage/errors';
 import { Knex } from 'knex';
 import { Transaction } from '../../database';
+import { DbEntitiesRow } from '../../database/types';
 import {
   ProcessingDatabase,
   AddUnprocessedEntitiesOptions,
   UpdateProcessedEntityOptions,
   GetProcessableEntitiesResult,
+  UpdateFinalEntityOptions,
 } from './types';
 import type { Logger } from 'winston';
 import { Entity, stringifyEntityRef } from '@backstage/catalog-model';
+import { createHash } from 'crypto';
+import stringify from 'fast-json-stable-stringify';
 import { v4 as uuid } from 'uuid';
 
 export type DbRefreshStateRequest = {
@@ -43,12 +47,52 @@ export type DbRefreshStateRow = {
   errors: string;
 };
 
+function generateEntityEtag(entity: Entity) {
+  return createHash('sha1')
+    .update(stringify({ ...entity }))
+    .digest('hex');
+}
+
 export class ProcessingDatabaseImpl implements ProcessingDatabase {
   constructor(
     private readonly database: Knex,
     private readonly logger: Logger,
   ) {}
 
+  async updateFinalEntity(
+    txOpaque: Transaction,
+    options: UpdateFinalEntityOptions,
+  ): Promise<void> {
+    const tx = txOpaque as Knex.Transaction;
+
+    const { finalEntity } = options;
+    const { relations } = finalEntity;
+    const { uid, etag, generation } = finalEntity.metadata;
+
+    if (uid === undefined || etag === undefined || generation === undefined) {
+      throw new InputError(
+        'One of the metadata fields "uid", "etag", or "generation" was missing',
+      );
+    } else if (relations === undefined) {
+      throw new InputError('The field "relations" was missing');
+    }
+    // TODO(freben): state/errors?
+
+    const fullName = stringifyEntityRef(finalEntity);
+
+    const result = await tx<DbEntitiesRow>('entities')
+      .insert({
+        id: uid,
+        location_id: null,
+        etag,
+        generation,
+        full_name: fullName,
+        data: JSON.stringify(finalEntity),
+      })
+      .onConflict('full_name')
+      .merge(['etag', 'generation', 'data']);
+  }
+
   async updateProcessedEntity(
     txOpaque: Transaction,
     options: UpdateProcessedEntityOptions,

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

@@ -46,8 +46,19 @@ export type GetProcessableEntitiesResult = {
   items: RefreshStateItem[];
 };
 
+export type UpdateFinalEntityOptions = {
+  finalEntity: Entity;
+  // TODO(freben): search
+};
+
 export interface ProcessingDatabase {
   transaction<T>(fn: (tx: Transaction) => Promise<T>): Promise<T>;
+
+  updateFinalEntity(
+    txOpaque: Transaction,
+    options: UpdateFinalEntityOptions,
+  ): Promise<void>;
+
   addUnprocessedEntities(
     tx: Transaction,
     options: AddUnprocessedEntitiesOptions,

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

@@ -18,6 +18,7 @@ import {
   EntityName,
   LocationSpec,
   Location,
+  EntityRelationSpec,
 } from '@backstage/catalog-model';
 import { JsonObject } from '@backstage/config';
 import { Observable } from '@backstage/core'; // << nooo
@@ -81,6 +82,7 @@ export type EntityProcessingResult =
       state: Map<string, JsonObject>;
       completedEntity: Entity;
       deferredEntites: Entity[];
+      relations: EntityRelationSpec[];
       errors: Error[];
     }
   | {