events-backend: use memory store if not using postgres + memory updates

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

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();
   };
 

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);
+      },
     };
   }
 }

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>;