From 105d68cd09d8a8e8953072b3d6b66f751b6f8c62 Mon Sep 17 00:00:00 2001
From: Patrik Oldsberg <poldsberg@gmail.com>
Date: Sun, 26 May 2024 21:12:55 +0200
Subject: [PATCH] eventa-backend: fix for pg listener not actually waiting for
 initial setup

Signed-off-by: Patrik Oldsberg <poldsberg@gmail.com>
---
 .../src/service/hub/DatabaseEventBusStore.ts  | 19 ++++++++++++-------
 1 file changed, 12 insertions(+), 7 deletions(-)

diff --git a/plugins/events-backend/src/service/hub/DatabaseEventBusStore.ts b/plugins/events-backend/src/service/hub/DatabaseEventBusStore.ts
index dfcf10ece4..53f55ce67f 100644
--- a/plugins/events-backend/src/service/hub/DatabaseEventBusStore.ts
+++ b/plugins/events-backend/src/service/hub/DatabaseEventBusStore.ts
@@ -97,10 +97,10 @@ class DatabaseEventBusListener {
     this.#logger = logger.child({ type: 'DatabaseEventBusListener' });
   }
 
-  async waitForUpdate(
+  async setupListener(
     topics: Set<string>,
     signal: AbortSignal,
-  ): Promise<{ topic: string }> {
+  ): Promise<{ waitForUpdate(): Promise<{ topic: string }> }> {
     if (this.#connTimeout) {
       clearTimeout(this.#connTimeout);
       this.#connTimeout = undefined;
@@ -108,7 +108,7 @@ class DatabaseEventBusListener {
 
     await this.#ensureConnection();
 
-    return new Promise<{ topic: string }>((resolve, reject) => {
+    const updatePromise = new Promise<{ topic: string }>((resolve, reject) => {
       const listener = { topics, resolve, reject };
       this.#listeners.add(listener);
 
@@ -118,6 +118,11 @@ class DatabaseEventBusListener {
         reject(signal.reason);
       });
     });
+
+    // Ignore unhandled rejections
+    updatePromise.catch(() => {});
+
+    return { waitForUpdate: () => updatePromise };
   }
 
   async shutdown() {
@@ -492,10 +497,10 @@ export class DatabaseEventBusStore implements EventBusStore {
 
     options.signal.throwIfAborted();
 
-    const topics = new Set(result.topics ?? []);
-    return {
-      waitForUpdate: () => this.#listener.waitForUpdate(topics, options.signal),
-    };
+    return this.#listener.setupListener(
+      new Set(result.topics ?? []),
+      options.signal,
+    );
   }
 
   #cleanup = async () => {