Merge pull request #16977 from sblausten/conflict-events

Allow event to be sent on conflicts

.changeset/sweet-fans-juggle.md

@@ -0,0 +1,5 @@
+---
+'@backstage/plugin-catalog-backend': minor
+---
+
+Adds an optional `EventBroker` that is used for sending an event when there are conflicts, with details of the conflict so that it can be handled elsewhere.

plugins/catalog-backend/api-report.md

@@ -33,6 +33,7 @@ import { EntityProvider as EntityProvider_2 } from '@backstage/plugin-catalog-no
 import { EntityProviderConnection as EntityProviderConnection_2 } from '@backstage/plugin-catalog-node';
 import { EntityProviderMutation as EntityProviderMutation_2 } from '@backstage/plugin-catalog-node';
 import { EntityRelationSpec as EntityRelationSpec_2 } from '@backstage/plugin-catalog-node';
+import { EventBroker } from '@backstage/plugin-events-node';
 import { GetEntitiesRequest } from '@backstage/catalog-client';
 import { JsonValue } from '@backstage/types';
 import { LocationSpec as LocationSpec_2 } from '@backstage/plugin-catalog-common';
@@ -113,6 +114,9 @@ export class BuiltinKindsEntityProcessor implements CatalogProcessor_2 {
   validateEntityKind(entity: Entity): Promise<boolean>;
 }
 
+// @public (undocumented)
+export const CATALOG_CONFLICTS_TOPIC = 'experimental.catalog.conflict';
+
 // @public
 export class CatalogBuilder {
   addEntityPolicy(
@@ -142,6 +146,7 @@ export class CatalogBuilder {
   replaceProcessors(processors: CatalogProcessor_2[]): CatalogBuilder;
   setAllowedLocationTypes(allowedLocationTypes: string[]): CatalogBuilder;
   setEntityDataParser(parser: CatalogProcessorParser_2): CatalogBuilder;
+  setEventBroker(broker: EventBroker): CatalogBuilder;
   setFieldFormatValidators(validators: Partial<Validators>): CatalogBuilder;
   setLocationAnalyzer(locationAnalyzer: LocationAnalyzer): CatalogBuilder;
   setPlaceholderResolver(

plugins/catalog-backend/package.json

@@ -56,6 +56,7 @@
     "@backstage/plugin-auth-node": "workspace:^",
     "@backstage/plugin-catalog-common": "workspace:^",
     "@backstage/plugin-catalog-node": "workspace:^",
+    "@backstage/plugin-events-node": "workspace:^",
     "@backstage/plugin-permission-common": "workspace:^",
     "@backstage/plugin-permission-node": "workspace:^",
     "@backstage/plugin-scaffolder-common": "workspace:^",

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

@@ -311,3 +311,29 @@ export type Cursor = {
    */
   totalItems?: number;
 };
+
+/**
+ * The details of the conflicting entity.
+ */
+export type CatalogConflictEventPayload = {
+  /**
+   * The new conflicting entity in raw, unvalidated form
+   */
+  unprocessedEntity: Entity;
+  /**
+   * The new conflicting entity's ref
+   */
+  entityRef: string;
+  /**
+   * The location key of this new conflicting entity
+   */
+  newLocationKey: string;
+  /**
+   * The location key of the entity already in the catalog that shares the same name and namespace as this entity
+   */
+  existingLocationKey: string;
+  /**
+   * The datetime that the conflict was processed at
+   */
+  lastConflictAt: string;
+};

plugins/catalog-backend/src/constants.ts

@@ -0,0 +1,18 @@
+/*
+ * Copyright 2020 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.
+ */
+
+/** @public */
+export const CATALOG_CONFLICTS_TOPIC = 'experimental.catalog.conflict';

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

@@ -44,6 +44,10 @@ import { insertUnprocessedEntity } from './operations/refreshState/insertUnproce
 import { updateUnprocessedEntity } from './operations/refreshState/updateUnprocessedEntity';
 import { deleteOrphanedEntities } from './operations/util/deleteOrphanedEntities';
 import { generateStableHash } from './util';
+import { EventBroker, EventParams } from '@backstage/plugin-events-node';
+import { DateTime } from 'luxon';
+import { CATALOG_CONFLICTS_TOPIC } from '../constants';
+import { CatalogConflictEventPayload } from '../catalog/types';
 
 // 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
@@ -57,6 +61,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
       database: Knex;
       logger: Logger;
       refreshInterval: ProcessingIntervalFunction;
+      eventBroker?: EventBroker;
     },
   ) {
     initDatabaseMetrics(options.database);
@@ -361,6 +366,19 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
         this.options.logger.warn(
           `Detected conflicting entityRef ${entityRef} already referenced by ${conflictingKey} and now also ${locationKey}`,
         );
+        if (this.options.eventBroker && locationKey) {
+          const eventParams: EventParams<CatalogConflictEventPayload> = {
+            topic: CATALOG_CONFLICTS_TOPIC,
+            eventPayload: {
+              unprocessedEntity: entity,
+              entityRef,
+              newLocationKey: locationKey,
+              existingLocationKey: conflictingKey,
+              lastConflictAt: DateTime.now().toISO(),
+            },
+          };
+          await this.options.eventBroker?.publish(eventParams);
+        }
       }
     }
 

plugins/catalog-backend/src/index.ts

@@ -27,6 +27,7 @@ export * from './processing';
 export * from './search';
 export * from './service';
 export * from './deprecated';
+export * from './constants';
 
 import {
   DefaultCatalogCollatorFactory as _DefaultCatalogCollatorFactory,

plugins/catalog-backend/src/service/CatalogBuilder.ts

@@ -99,6 +99,7 @@ import {
 import { AuthorizedLocationService } from './AuthorizedLocationService';
 import { DefaultProviderDatabase } from '../database/DefaultProviderDatabase';
 import { DefaultCatalogDatabase } from '../database/DefaultCatalogDatabase';
+import { EventBroker } from '@backstage/plugin-events-node';
 
 /**
  * This is a duplicate of the alpha `CatalogPermissionRule` type, for use in the stable API.
@@ -168,6 +169,7 @@ export class CatalogBuilder {
   private readonly permissionRules: CatalogPermissionRuleInput[];
   private allowedLocationType: string[];
   private legacySingleProcessorValidation = false;
+  private eventBroker?: EventBroker;
 
   /**
    * Creates a catalog builder.
@@ -419,6 +421,14 @@ export class CatalogBuilder {
     return this;
   }
 
+  /**
+   * Enables the publishing of events for cloflicts in the DefaultProcessingDatabase
+   */
+  setEventBroker(broker: EventBroker): CatalogBuilder {
+    this.eventBroker = broker;
+    return this;
+  }
+
   /**
    * Wires up and returns all of the component parts of the catalog
    */
@@ -442,6 +452,7 @@ export class CatalogBuilder {
       database: dbClient,
       logger,
       refreshInterval: this.processingInterval,
+      eventBroker: this.eventBroker,
     });
     const providerDatabase = new DefaultProviderDatabase({
       database: dbClient,

yarn.lock

@@ -5702,6 +5702,7 @@ __metadata:
     "@backstage/plugin-auth-node": "workspace:^"
     "@backstage/plugin-catalog-common": "workspace:^"
     "@backstage/plugin-catalog-node": "workspace:^"
+    "@backstage/plugin-events-node": "workspace:^"
     "@backstage/plugin-permission-common": "workspace:^"
     "@backstage/plugin-permission-node": "workspace:^"
     "@backstage/plugin-scaffolder-common": "workspace:^"