events-backend: update listening logic to be more robust

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

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

@@ -348,8 +348,11 @@ export class DatabaseEventHubStore implements EventHubStore {
 
   async listen(
     subscriptionId: string,
-    onNotify: (topicId: string) => void,
-  ): Promise<() => void> {
+    listeners: {
+      onNotify: (topicId: string) => void;
+      onError: () => void;
+    },
+  ): Promise<{ cancel(): void }> {
     const result = await this.#db<SubscriptionsRow>(TABLE_SUBSCRIPTIONS)
       .select('topics')
       .where({ id: subscriptionId })
@@ -361,6 +364,11 @@ export class DatabaseEventHubStore implements EventHubStore {
       );
     }
     const topics = new Set(result.topics ?? []);
-    return this.#listener.listen(topics, onNotify, () => {});
+    const cancel = await this.#listener.listen(
+      topics,
+      listeners.onNotify,
+      listeners.onError,
+    );
+    return { cancel };
   }
 }

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

@@ -118,21 +118,46 @@ export class EventHub {
     });
     const id = req.params.subscriptionId;
 
-    const { events } = await this.#store.readSubscription(id);
-
-    this.#logger.info(
-      `Reading subscription '${id}' resulted in ${events.length} events`,
-      { subject: credentials.principal.subject },
-    );
-
-    if (events.length > 0) {
-      res.json({ events });
-      return;
-    }
-
-    this.#store.listen(id, () => {
-      res.status(204).end();
+    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);
+
+    try {
+      const { events } = await this.#store.readSubscription(id);
+
+      this.#logger.info(
+        `Reading subscription '${id}' resulted in ${events.length} events`,
+        { subject: credentials.principal.subject },
+      );
+
+      if (events.length > 0) {
+        res.json({ events });
+        resolveShouldNotify!(false);
+      } else {
+        resolveShouldNotify!(true);
+      }
+    } finally {
+      resolveShouldNotify!(false);
+    }
   };
 
   #handlePutSubscription: internal.DocRequestHandler<

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

@@ -28,6 +28,9 @@ export type EventHubStore = {
 
   listen(
     subscriptionId: string,
-    onNotify: (topicId: string) => void,
-  ): Promise<() => void>;
+    listeners: {
+      onNotify(topicId: string): void;
+      onError(): void;
+    },
+  ): Promise<{ cancel(): void }>;
 };