From 5f2debd2204825feb219120a6c6fced142839627 Mon Sep 17 00:00:00 2001
From: blam <ben@blam.sh>
Date: Thu, 21 Nov 2024 12:41:37 +0100
Subject: [PATCH] chore: wire up subscriptions for new frontend system

Signed-off-by: blam <ben@blam.sh>
---
 .../src/module/WrapperProviders.ts              | 17 ++++++++++++++++-
 ...gModuleIncrementalIngestionEntityProvider.ts |  3 +++
 2 files changed, 19 insertions(+), 1 deletion(-)

diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/module/WrapperProviders.ts b/plugins/catalog-backend-module-incremental-ingestion/src/module/WrapperProviders.ts
index c322d9adda..b510eb2ead 100644
--- a/plugins/catalog-backend-module-incremental-ingestion/src/module/WrapperProviders.ts
+++ b/plugins/catalog-backend-module-incremental-ingestion/src/module/WrapperProviders.ts
@@ -36,6 +36,7 @@ import {
   IncrementalEntityProvider,
   IncrementalEntityProviderOptions,
 } from '../types';
+import { EventsService } from '@backstage/plugin-events-node';
 
 /**
  * Helps in the creation of the catalog entity providers that wrap the
@@ -53,6 +54,7 @@ export class WrapperProviders {
       client: Knex;
       scheduler: SchedulerService;
       applyDatabaseMigrations?: typeof applyDatabaseMigrations;
+      events: EventsService;
     },
   ) {}
 
@@ -64,7 +66,7 @@ export class WrapperProviders {
     return {
       getProviderName: () => provider.getProviderName(),
       connect: async connection => {
-        await this.startProvider(provider, options, connection);
+        await this.startProvider(provider, options, connection, events);
         this.numberOfProvidersToConnect -= 1;
         if (this.numberOfProvidersToConnect === 0) {
           this.readySignal.resolve();
@@ -84,6 +86,7 @@ export class WrapperProviders {
     provider: IncrementalEntityProvider<unknown, unknown>,
     providerOptions: IncrementalEntityProviderOptions,
     connection: EntityProviderConnection,
+    events: EventsService,
   ) {
     const logger = this.options.logger.child({
       entityProvider: provider.getProviderName(),
@@ -130,6 +133,18 @@ export class WrapperProviders {
         frequency,
         timeout: length,
       });
+
+      const topics = engine.supportsEventTopics();
+      if (topics.length > 0) {
+        logger.info(
+          `Provider ${provider.getProviderName()} subscribing to events for topics: ${topics.join()}`,
+        );
+        await events.subscribe({
+          topics,
+          id: provider.getProviderName(),
+          onEvent: evt => engine.onEvent(evt),
+        });
+      }
     } catch (error) {
       logger.warn(
         `Failed to initialize incremental ingestion provider ${provider.getProviderName()}, ${stringifyError(
diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/module/catalogModuleIncrementalIngestionEntityProvider.ts b/plugins/catalog-backend-module-incremental-ingestion/src/module/catalogModuleIncrementalIngestionEntityProvider.ts
index 8752bd3137..74244402a6 100644
--- a/plugins/catalog-backend-module-incremental-ingestion/src/module/catalogModuleIncrementalIngestionEntityProvider.ts
+++ b/plugins/catalog-backend-module-incremental-ingestion/src/module/catalogModuleIncrementalIngestionEntityProvider.ts
@@ -25,6 +25,7 @@ import {
   IncrementalEntityProviderOptions,
 } from '@backstage/plugin-catalog-backend-module-incremental-ingestion';
 import { WrapperProviders } from './WrapperProviders';
+import { eventsServiceRef } from '@backstage/plugin-events-node';
 
 /**
  * @public
@@ -106,6 +107,7 @@ export const catalogModuleIncrementalIngestionEntityProvider =
           httpRouter: coreServices.httpRouter,
           logger: coreServices.logger,
           scheduler: coreServices.scheduler,
+          events: eventsServiceRef,
         },
         async init({
           catalog,
@@ -122,6 +124,7 @@ export const catalogModuleIncrementalIngestionEntityProvider =
             logger,
             client,
             scheduler,
+            events,
           });
 
           for (const entry of addedProviders) {