From ff60692772640bae271ad1dec0d445f55b38764a Mon Sep 17 00:00:00 2001
From: Patrik Oldsberg <poldsberg@gmail.com>
Date: Fri, 24 May 2024 12:14:51 +0200
Subject: [PATCH] events-backend: use memory store if not using postgres +
 memory updates

Signed-off-by: Patrik Oldsberg <poldsberg@gmail.com>
---
 .../src/service/hub/EventHub.ts               | 33 ++++++++++++-------
 .../src/service/hub/MemoryEventHubStore.ts    | 20 +++++++----
 .../events-backend/src/service/hub/types.ts   |  2 +-
 3 files changed, 36 insertions(+), 19 deletions(-)

diff --git a/plugins/events-backend/src/service/hub/EventHub.ts b/plugins/events-backend/src/service/hub/EventHub.ts
index 97d641e107..b54e9fdf08 100644
--- a/plugins/events-backend/src/service/hub/EventHub.ts
+++ b/plugins/events-backend/src/service/hub/EventHub.ts
@@ -37,10 +37,19 @@ export class EventHub {
     const { database, httpAuth } = options;
     const logger = options.logger.child({ type: 'EventHub' });
     const router = Router();
-    const store = await DatabaseEventHubStore.create({
-      database,
-      logger,
-    });
+
+    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);
 
@@ -92,19 +101,21 @@ export class EventHub {
     const credentials = await this.#httpAuth.credentials(req, {
       allow: ['service'],
     });
-    const { id } = await this.#store.publish({
+    const result = await this.#store.publish({
       params: {
         topic: req.body.event.topic,
         eventPayload: req.body.event.payload,
       } as EventParams,
       subscriberIds: req.body.subscriptionIds ?? [],
     });
-    this.#logger.info(
-      `Published event to '${req.body.event.topic}' with ID '${id}'`,
-      {
-        subject: credentials.principal.subject,
-      },
-    );
+    if (result) {
+      this.#logger.info(
+        `Published event to '${req.body.event.topic}' with ID '${result.id}'`,
+        {
+          subject: credentials.principal.subject,
+        },
+      );
+    }
     res.status(201).end();
   };
 
diff --git a/plugins/events-backend/src/service/hub/MemoryEventHubStore.ts b/plugins/events-backend/src/service/hub/MemoryEventHubStore.ts
index 7ab5b98f82..9b72c18770 100644
--- a/plugins/events-backend/src/service/hub/MemoryEventHubStore.ts
+++ b/plugins/events-backend/src/service/hub/MemoryEventHubStore.ts
@@ -34,7 +34,7 @@ export class MemoryEventHubStore implements EventHubStore {
   async publish(options: {
     params: EventParams;
     subscriberIds: string[];
-  }): Promise<void> {
+  }): Promise<{ id: string } | undefined> {
     const topicId = options.params.topic;
     const subscriberIds = new Set(options.subscriberIds);
 
@@ -46,7 +46,7 @@ export class MemoryEventHubStore implements EventHubStore {
       }
     }
     if (!hasOtherSubscribers) {
-      return;
+      return undefined;
     }
 
     const nextSeq = this.#getMaxSeq() + 1;
@@ -57,6 +57,7 @@ export class MemoryEventHubStore implements EventHubStore {
         listener.notify(topicId);
       }
     }
+    return { id: String(nextSeq) };
   }
 
   #getMaxSeq() {
@@ -98,16 +99,21 @@ export class MemoryEventHubStore implements EventHubStore {
 
   async listen(
     subscriptionId: string,
-    onNotify: (topicId: string) => void,
-  ): Promise<() => void> {
+    listeners: {
+      onNotify(topicId: string): void;
+      onError(): void;
+    },
+  ): Promise<{ cancel(): void }> {
     const sub = this.#subscribers.get(subscriptionId);
     if (!sub) {
       throw new Error(`Subscription not found`);
     }
-    const listener = { topics: sub.topics, notify: onNotify };
+    const listener = { topics: sub.topics, notify: listeners.onNotify };
     this.#listeners.add(listener);
-    return () => {
-      this.#listeners.delete(listener);
+    return {
+      cancel: () => {
+        this.#listeners.delete(listener);
+      },
     };
   }
 }
diff --git a/plugins/events-backend/src/service/hub/types.ts b/plugins/events-backend/src/service/hub/types.ts
index 0a0a27b1c1..6cc1cc58e5 100644
--- a/plugins/events-backend/src/service/hub/types.ts
+++ b/plugins/events-backend/src/service/hub/types.ts
@@ -20,7 +20,7 @@ export type EventHubStore = {
   publish(options: {
     params: EventParams;
     subscriberIds: string[];
-  }): Promise<{ id: string }>;
+  }): Promise<{ id: string } | undefined>;
 
   upsertSubscription(id: string, topics: string[]): Promise<void>;