chore: pass through an events service optionally

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

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,10 @@ import {
   coreServices,
   createBackendPlugin,
 } from '@backstage/backend-plugin-api';
+import {
+  DefaultEventsService,
+  eventsServiceRef,
+} from '@backstage/plugin-events-node';
 import { Entity, Validators } from '@backstage/catalog-model';
 import { CatalogBuilder, CatalogPermissionRuleInput } from './CatalogBuilder';
 import {
@@ -39,6 +43,7 @@ import {
 import { merge } from 'lodash';
 import { Permission } from '@backstage/plugin-permission-common';
 import { ForwardedError } from '@backstage/errors';
+import { eventsExtensionPoint } from '@backstage/plugin-events-node/alpha';
 
 class CatalogProcessingExtensionPointImpl
   implements CatalogProcessingExtensionPoint
@@ -211,6 +216,7 @@ export const catalogPlugin = createBackendPlugin({
         discovery: coreServices.discovery,
         auth: coreServices.auth,
         httpAuth: coreServices.httpAuth,
+        events: eventsServiceRef,
       },
       async init({
         logger,
@@ -224,6 +230,7 @@ export const catalogPlugin = createBackendPlugin({
         discovery,
         auth,
         httpAuth,
+        events,
       }) {
         const builder = await CatalogBuilder.create({
           config,
@@ -236,6 +243,9 @@ export const catalogPlugin = createBackendPlugin({
           auth,
           httpAuth,
         });
+
+        builder.setEventBroker(events);
+
         if (processingExtensions.onProcessingErrorHandler) {
           builder.subscribe({
             onProcessingError: processingExtensions.onProcessingErrorHandler,