From 217149ae9893edee3f93b311041a905c753e4d23 Mon Sep 17 00:00:00 2001
From: Patrik Oldsberg <poldsberg@gmail.com>
Date: Mon, 2 Jan 2023 14:17:25 +0100
Subject: [PATCH] events-backend: await subscribers and log errors

Signed-off-by: Patrik Oldsberg <poldsberg@gmail.com>
---
 .changeset/nasty-planets-end.md               |  5 ++
 .../src/service/InMemoryEventBroker.test.ts   | 48 +++++++++++++++++++
 .../src/service/InMemoryEventBroker.ts        | 13 ++++-
 3 files changed, 65 insertions(+), 1 deletion(-)
 create mode 100644 .changeset/nasty-planets-end.md

diff --git a/.changeset/nasty-planets-end.md b/.changeset/nasty-planets-end.md
new file mode 100644
index 0000000000..b361b28199
--- /dev/null
+++ b/.changeset/nasty-planets-end.md
@@ -0,0 +1,5 @@
+---
+'@backstage/plugin-events-backend': patch
+---
+
+The default event broker will now catch and log errors thrown by the `onEvent` method of subscribers. The returned promise from `publish` method will also not resolve until all subscribers have handled the event.
diff --git a/plugins/events-backend/src/service/InMemoryEventBroker.test.ts b/plugins/events-backend/src/service/InMemoryEventBroker.test.ts
index 68a6f63a72..861cdda940 100644
--- a/plugins/events-backend/src/service/InMemoryEventBroker.test.ts
+++ b/plugins/events-backend/src/service/InMemoryEventBroker.test.ts
@@ -16,6 +16,7 @@
 
 import { getVoidLogger } from '@backstage/backend-common';
 import { TestEventSubscriber } from '@backstage/plugin-events-backend-test-utils';
+import { EventParams, EventSubscriber } from '@backstage/plugin-events-node';
 import { InMemoryEventBroker } from './InMemoryEventBroker';
 
 const logger = getVoidLogger();
@@ -63,4 +64,51 @@ describe('InMemoryEventBroker', () => {
       eventPayload: { test: 'topicC' },
     });
   });
+
+  it('logs errors from subscribers', async () => {
+    const topic = 'testTopic';
+
+    const subscriber1 = new (class Subscriber1 implements EventSubscriber {
+      supportsEventTopics() {
+        return [topic];
+      }
+      async onEvent(event: EventParams) {
+        throw new Error(`NOPE ${event.eventPayload}`);
+      }
+    })();
+    const subscriber2 = new (class Subscriber2 implements EventSubscriber {
+      supportsEventTopics() {
+        return [topic];
+      }
+      async onEvent(event: EventParams) {
+        throw new Error(`NOPE ${event.eventPayload}`);
+      }
+    })();
+
+    const errorSpy = jest.spyOn(logger, 'error');
+    const eventBroker = new InMemoryEventBroker(logger);
+
+    eventBroker.subscribe(subscriber1);
+    await eventBroker.publish({ topic, eventPayload: '1' });
+
+    expect(errorSpy).toHaveBeenCalledTimes(1);
+    expect(errorSpy).toHaveBeenCalledWith(
+      'Subscriber "Subscriber1" failed to process event',
+      new Error('NOPE 1'),
+    );
+
+    eventBroker.subscribe(subscriber2);
+    await eventBroker.publish({ topic, eventPayload: '2' });
+
+    // With two subscribers we should not halt on the first error but call all subscribers
+    expect(errorSpy).toHaveBeenCalledTimes(3);
+    expect(errorSpy).toHaveBeenCalledWith(
+      'Subscriber "Subscriber1" failed to process event',
+      new Error('NOPE 2'),
+    );
+    expect(errorSpy).toHaveBeenCalledWith(
+      'Subscriber "Subscriber2" failed to process event',
+      new Error('NOPE 2'),
+    );
+  });
 });
diff --git a/plugins/events-backend/src/service/InMemoryEventBroker.ts b/plugins/events-backend/src/service/InMemoryEventBroker.ts
index 90c7d912fe..04be01f658 100644
--- a/plugins/events-backend/src/service/InMemoryEventBroker.ts
+++ b/plugins/events-backend/src/service/InMemoryEventBroker.ts
@@ -42,7 +42,18 @@ export class InMemoryEventBroker implements EventBroker {
     );
 
     const subscribed = this.subscribers[params.topic] ?? [];
-    subscribed.forEach(subscriber => subscriber.onEvent(params));
+    await Promise.all(
+      subscribed.map(async subscriber => {
+        try {
+          await subscriber.onEvent(params);
+        } catch (error) {
+          this.logger.error(
+            `Subscriber "${subscriber.constructor.name}" failed to process event`,
+            error,
+          );
+        }
+      }),
+    );
   }
 
   subscribe(