Merge pull request #24847 from backstage/blam/events

catalog: support events service in new backend system

.changeset/mean-laws-lay.md

@@ -0,0 +1,5 @@
+---
+'@backstage/plugin-catalog-backend': minor
+---
+
+Pass through `EventsService` too in the new backend system

plugins/catalog-backend/api-report.md

@@ -39,6 +39,7 @@ import { EntityProviderConnection as EntityProviderConnection_2 } from '@backsta
 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 { EventsService } from '@backstage/plugin-events-node';
 import { GetEntitiesRequest } from '@backstage/catalog-client';
 import { HttpAuthService } from '@backstage/backend-plugin-api';
 import { LocationAnalyzer as LocationAnalyzer_2 } from '@backstage/plugin-catalog-node';
@@ -166,7 +167,7 @@ export class CatalogBuilder {
   replaceProcessors(processors: CatalogProcessor_2[]): CatalogBuilder;
   setAllowedLocationTypes(allowedLocationTypes: string[]): CatalogBuilder;
   setEntityDataParser(parser: CatalogProcessorParser_2): CatalogBuilder;
-  setEventBroker(broker: EventBroker): CatalogBuilder;
+  setEventBroker(broker: EventBroker | EventsService): CatalogBuilder;
   setFieldFormatValidators(validators: Partial<Validators>): CatalogBuilder;
   setLocationAnalyzer(locationAnalyzer: LocationAnalyzer_2): CatalogBuilder;
   setPlaceholderResolver(

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

@@ -42,7 +42,11 @@ import { checkLocationKeyConflict } from './operations/refreshState/checkLocatio
 import { insertUnprocessedEntity } from './operations/refreshState/insertUnprocessedEntity';
 import { updateUnprocessedEntity } from './operations/refreshState/updateUnprocessedEntity';
 import { generateStableHash } from './util';
-import { EventBroker, EventParams } from '@backstage/plugin-events-node';
+import {
+  EventBroker,
+  EventParams,
+  EventsService,
+} from '@backstage/plugin-events-node';
 import { DateTime } from 'luxon';
 import { CATALOG_CONFLICTS_TOPIC } from '../constants';
 import { CatalogConflictEventPayload } from '../catalog/types';
@@ -60,7 +64,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
       database: Knex;
       logger: LoggerService;
       refreshInterval: ProcessingIntervalFunction;
-      eventBroker?: EventBroker;
+      eventBroker?: EventBroker | EventsService;
     },
   ) {
     initDatabaseMetrics(options.database);

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

@@ -37,7 +37,7 @@ import {
   withActiveSpan,
 } from '../util/opentelemetry';
 import { deleteOrphanedEntities } from '../database/operations/util/deleteOrphanedEntities';
-import { EventBroker } from '@backstage/plugin-events-node';
+import { EventBroker, EventsService } from '@backstage/plugin-events-node';
 import { CATALOG_ERRORS_TOPIC } from '../constants';
 import { LoggerService } from '@backstage/backend-plugin-api';
 
@@ -69,7 +69,7 @@ export class DefaultCatalogProcessingEngine {
     errors: Error[];
   }) => Promise<void> | void;
   private readonly tracker: ProgressTracker;
-  private readonly eventBroker?: EventBroker;
+  private readonly eventBroker?: EventBroker | EventsService;
 
   private stopFunc?: () => void;
 
@@ -89,7 +89,7 @@ export class DefaultCatalogProcessingEngine {
       errors: Error[];
     }) => Promise<void> | void;
     tracker?: ProgressTracker;
-    eventBroker?: EventBroker;
+    eventBroker?: EventBroker | EventsService;
   }) {
     this.config = options.config;
     this.scheduler = options.scheduler;

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

@@ -104,7 +104,7 @@ import {
 import { AuthorizedLocationService } from './AuthorizedLocationService';
 import { DefaultProviderDatabase } from '../database/DefaultProviderDatabase';
 import { DefaultCatalogDatabase } from '../database/DefaultCatalogDatabase';
-import { EventBroker } from '@backstage/plugin-events-node';
+import { EventBroker, EventsService } from '@backstage/plugin-events-node';
 import { durationToMilliseconds } from '@backstage/types';
 import {
   AuthService,
@@ -182,7 +182,7 @@ export class CatalogBuilder {
   private readonly permissionRules: CatalogPermissionRuleInput[];
   private allowedLocationType: string[];
   private legacySingleProcessorValidation = false;
-  private eventBroker?: EventBroker;
+  private eventBroker?: EventBroker | EventsService;
 
   /**
    * Creates a catalog builder.
@@ -453,7 +453,7 @@ export class CatalogBuilder {
   /**
    * Enables the publishing of events for conflicts in the DefaultProcessingDatabase
    */
-  setEventBroker(broker: EventBroker): CatalogBuilder {
+  setEventBroker(broker: EventBroker | EventsService): CatalogBuilder {
     this.eventBroker = broker;
     return this;
   }

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

@@ -17,6 +17,7 @@ import {
   coreServices,
   createBackendPlugin,
 } from '@backstage/backend-plugin-api';
+import { eventsServiceRef } from '@backstage/plugin-events-node';
 import { Entity, Validators } from '@backstage/catalog-model';
 import { CatalogBuilder, CatalogPermissionRuleInput } from './CatalogBuilder';
 import {
@@ -211,6 +212,7 @@ export const catalogPlugin = createBackendPlugin({
         discovery: coreServices.discovery,
         auth: coreServices.auth,
         httpAuth: coreServices.httpAuth,
+        events: eventsServiceRef,
       },
       async init({
         logger,
@@ -224,6 +226,7 @@ export const catalogPlugin = createBackendPlugin({
         discovery,
         auth,
         httpAuth,
+        events,
       }) {
         const builder = await CatalogBuilder.create({
           config,
@@ -236,6 +239,9 @@ export const catalogPlugin = createBackendPlugin({
           auth,
           httpAuth,
         });
+
+        builder.setEventBroker(events);
+
         if (processingExtensions.onProcessingErrorHandler) {
           builder.subscribe({
             onProcessingError: processingExtensions.onProcessingErrorHandler,