events-backend: refactor EventHub into a router creator

Signed-off-by: Patrik Oldsberg <poldsberg@gmail.com>

plugins/events-backend/src/service/EventsPlugin.ts

@@ -28,7 +28,7 @@ import {
 } from '@backstage/plugin-events-node';
 import Router from 'express-promise-router';
 import { HttpPostIngressEventPublisher } from './http';
-import { EventHub } from './hub';
+import { createEventBusRouter } from './hub';
 
 class EventsExtensionPointImpl implements EventsExtensionPoint {
   #httpPostIngresses: HttpPostIngressOptions[] = [];
@@ -97,8 +97,7 @@ export const eventsPlugin = createBackendPlugin({
         const eventsRouter = Router();
         http.bind(eventsRouter);
 
-        const hub = await EventHub.create({ database, logger, httpAuth });
-        router.use(hub.handler());
+        router.use(await createEventBusRouter({ database, logger, httpAuth }));
 
         router.use(eventsRouter);
         router.addAuthPolicy({

plugins/events-backend/src/service/hub/EventHub.ts

@@ -21,89 +21,45 @@ import {
 } from '@backstage/backend-plugin-api';
 import { Handler } from 'express';
 import Router from 'express-promise-router';
-import { spec, createOpenApiRouter } from '../../schema/openapi.generated';
-import { internal } from '@backstage/backend-openapi-utils';
+import { createOpenApiRouter } from '../../schema/openapi.generated';
 import { MemoryEventHubStore } from './MemoryEventHubStore';
 import { DatabaseEventHubStore } from './DatabaseEventHubStore';
 import { EventHubStore } from './types';
 import { EventParams } from '@backstage/plugin-events-node';
 
-export class EventHub {
-  static async create(options: {
-    logger: LoggerService;
-    database: DatabaseService;
-    httpAuth: HttpAuthService;
-  }) {
-    const { database, httpAuth } = options;
-    const logger = options.logger.child({ type: 'EventHub' });
-    const router = Router();
+export async function createEventBusRouter(options: {
+  logger: LoggerService;
+  database: DatabaseService;
+  httpAuth: HttpAuthService;
+}): Promise<Handler> {
+  const { database, httpAuth } = options;
+  const logger = options.logger.child({ type: 'EventHub' });
+  const router = Router();
 
-    let store: EventHubStore;
-    const db = await database.getClient();
-    if (db.client.config.client === 'pg') {
-      logger.info('Database is PostgreSQL, using database store');
-      store = await DatabaseEventHubStore.create({
-        database,
-        logger,
-      });
-    } else {
-      logger.info('Database is not PostgreSQL, using memory store');
-      store = new MemoryEventHubStore();
-    }
-
-    const hub = new EventHub(router, logger, httpAuth, store);
-
-    const apiRouter = await createOpenApiRouter();
-
-    router.use(apiRouter);
-
-    apiRouter.post('/hub/events', hub.#handlePostEvents);
-
-    // Long-polling
-    apiRouter.get(
-      '/hub/subscriptions/:subscriptionId/events',
-      hub.#handleGetSubscription,
-    );
-    apiRouter.put(
-      '/hub/subscriptions/:subscriptionId',
-      hub.#handlePutSubscription,
-    );
-
-    return hub;
+  let store: EventHubStore;
+  const db = await database.getClient();
+  if (db.client.config.client === 'pg') {
+    logger.info('Database is PostgreSQL, using database store');
+    store = await DatabaseEventHubStore.create({
+      database,
+      logger,
+    });
+  } else {
+    logger.info('Database is not PostgreSQL, using memory store');
+    store = new MemoryEventHubStore();
   }
 
-  readonly #handler: Handler;
-  readonly #logger: LoggerService;
-  readonly #httpAuth: HttpAuthService;
-  readonly #store: EventHubStore;
+  const apiRouter = await createOpenApiRouter();
 
-  private constructor(
-    handler: Handler,
-    logger: LoggerService,
-    httpAuth: HttpAuthService,
-    store: EventHubStore,
-  ) {
-    this.#handler = handler;
-    this.#logger = logger;
-    this.#httpAuth = httpAuth;
-    this.#store = store;
-  }
+  router.use(apiRouter);
 
-  handler(): Handler {
-    return this.#handler;
-  }
-
-  #handlePostEvents: internal.DocRequestHandler<
-    typeof spec,
-    '/hub/events',
-    'post'
-  > = async (req, res) => {
-    const credentials = await this.#httpAuth.credentials(req, {
+  apiRouter.post('/hub/events', async (req, res) => {
+    const credentials = await httpAuth.credentials(req, {
       allow: ['service'],
     });
     const topic = req.body.event.topic;
     const subscriberIds = req.body.subscriptionIds ?? [];
-    const result = await this.#store.publish({
+    const result = await store.publish({
       params: {
         topic,
         eventPayload: req.body.event.payload,
@@ -111,15 +67,12 @@ export class EventHub {
       subscriberIds,
     });
     if (result) {
-      this.#logger.info(
-        `Published event to '${topic}' with ID '${result.id}'`,
-        {
-          subject: credentials.principal.subject,
-        },
-      );
+      logger.info(`Published event to '${topic}' with ID '${result.id}'`, {
+        subject: credentials.principal.subject,
+      });
       res.status(201).end();
     } else {
-      this.#logger.info(
+      logger.info(
         `Skipped publishing of event to '${topic}', subscribers have already been notified: '${subscriberIds.join(
           "', '",
         )}'`,
@@ -129,77 +82,74 @@ export class EventHub {
       );
       res.status(204).end();
     }
-  };
+  });
 
-  #handleGetSubscription: internal.DocRequestHandler<
-    typeof spec,
-    '/hub/subscriptions/{subscriptionId}/events',
-    'get'
-  > = async (req, res) => {
-    const credentials = await this.#httpAuth.credentials(req, {
-      allow: ['service'],
-    });
-    const id = req.params.subscriptionId;
+  apiRouter.get(
+    '/hub/subscriptions/:subscriptionId/events',
+    async (req, res) => {
+      const credentials = await httpAuth.credentials(req, {
+        allow: ['service'],
+      });
+      const id = req.params.subscriptionId;
 
-    let resolveShouldNotify: (shouldNotify: boolean) => void;
-    const shouldNotifyPromise = new Promise<boolean>(resolve => {
-      resolveShouldNotify = resolve;
-    });
+      let resolveShouldNotify: (shouldNotify: boolean) => void;
+      const shouldNotifyPromise = new Promise<boolean>(resolve => {
+        resolveShouldNotify = resolve;
+      });
 
-    const { cancel } = await this.#store.listen(id, {
-      onNotify() {
-        shouldNotifyPromise.then(shouldNotify => {
-          if (shouldNotify) {
-            res.status(204).end();
-          }
-        });
-      },
-      onError() {
-        shouldNotifyPromise.then(shouldNotify => {
-          if (shouldNotify) {
-            res.status(500).end();
-          }
-        });
-      },
-    });
-    req.on('end', cancel);
+      const { cancel } = await store.listen(id, {
+        onNotify() {
+          shouldNotifyPromise.then(shouldNotify => {
+            if (shouldNotify) {
+              res.status(204).end();
+            }
+          });
+        },
+        onError() {
+          shouldNotifyPromise.then(shouldNotify => {
+            if (shouldNotify) {
+              res.status(500).end();
+            }
+          });
+        },
+      });
+      req.on('end', cancel);
 
-    try {
-      const { events } = await this.#store.readSubscription(id);
+      try {
+        const { events } = await store.readSubscription(id);
 
-      this.#logger.info(
-        `Reading subscription '${id}' resulted in ${events.length} events`,
-        { subject: credentials.principal.subject },
-      );
+        logger.info(
+          `Reading subscription '${id}' resulted in ${events.length} events`,
+          { subject: credentials.principal.subject },
+        );
 
-      if (events.length > 0) {
-        res.json({ events });
+        if (events.length > 0) {
+          res.json({ events });
+          resolveShouldNotify!(false);
+        } else {
+          resolveShouldNotify!(true);
+        }
+      } finally {
         resolveShouldNotify!(false);
-      } else {
-        resolveShouldNotify!(true);
       }
-    } finally {
-      resolveShouldNotify!(false);
-    }
-  };
+    },
+  );
 
-  #handlePutSubscription: internal.DocRequestHandler<
-    typeof spec,
-    '/hub/subscriptions/{subscriptionId}',
-    'put'
-  > = async (req, res) => {
-    const credentials = await this.#httpAuth.credentials(req, {
+  apiRouter.put('/hub/subscriptions/:subscriptionId', async (req, res) => {
+    const credentials = await httpAuth.credentials(req, {
       allow: ['service'],
     });
     const id = req.params.subscriptionId;
 
-    await this.#store.upsertSubscription(id, req.body.topics);
+    await store.upsertSubscription(id, req.body.topics);
 
-    this.#logger.info(
+    logger.info(
       `New subscription '${id}' topics='${req.body.topics.join("', '")}'`,
       { subject: credentials.principal.subject },
     );
 
     res.status(201).end();
-  };
+  });
+
+  return apiRouter;
 }

plugins/events-backend/src/service/hub/index.ts

@@ -14,4 +14,4 @@
  * limitations under the License.
  */
 
-export { EventHub } from './EventHub';
+export { createEventBusRouter } from './EventHub';