From cda26b88225de7536acf0aeb65907c67e114bd9e Mon Sep 17 00:00:00 2001
From: Patrik Oldsberg <poldsberg@gmail.com>
Date: Wed, 18 Sep 2024 23:45:44 +0200
Subject: [PATCH] events-backend: add cleanup of old events for memory store

Signed-off-by: Patrik Oldsberg <poldsberg@gmail.com>
---
 .../service/hub/MemoryEventBusStore.test.ts   | 86 +++++++++++++++++++
 .../src/service/hub/MemoryEventBusStore.ts    | 19 +++-
 2 files changed, 104 insertions(+), 1 deletion(-)
 create mode 100644 plugins/events-backend/src/service/hub/MemoryEventBusStore.test.ts

diff --git a/plugins/events-backend/src/service/hub/MemoryEventBusStore.test.ts b/plugins/events-backend/src/service/hub/MemoryEventBusStore.test.ts
new file mode 100644
index 0000000000..6866f290b5
--- /dev/null
+++ b/plugins/events-backend/src/service/hub/MemoryEventBusStore.test.ts
@@ -0,0 +1,86 @@
+/*
+ * Copyright 2024 The Backstage Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+import { EventParams } from '@backstage/plugin-events-node';
+import { MemoryEventBusStore } from './MemoryEventBusStore';
+
+function mkEvent(message: string): EventParams {
+  return {
+    topic: 'test',
+    eventPayload: { message },
+  };
+}
+
+describe('MemoryEventBusStore', () => {
+  it('should publish to subscribers', async () => {
+    const store = new MemoryEventBusStore();
+
+    await expect(
+      store.publish({
+        event: mkEvent('hello'),
+        notifiedSubscribers: [],
+      }),
+    ).resolves.toEqual(undefined);
+
+    await expect(store.readSubscription('test')).rejects.toThrow(
+      'Subscription not found',
+    );
+
+    await store.upsertSubscription('tester', ['test']);
+
+    await expect(
+      store.publish({
+        event: mkEvent('hello'),
+        notifiedSubscribers: [],
+      }),
+    ).resolves.toEqual({ eventId: '1' });
+
+    await expect(
+      store.publish({
+        event: mkEvent('ignored'),
+        notifiedSubscribers: ['tester'],
+      }),
+    ).resolves.toEqual(undefined);
+
+    await expect(store.readSubscription('tester')).resolves.toEqual({
+      events: [mkEvent('hello')],
+    });
+  });
+
+  it('should clean up old events', async () => {
+    const store = new MemoryEventBusStore({ maxEvents: 5 });
+
+    await store.upsertSubscription('tester', ['test']);
+
+    for (let i = 0; i < 20; ++i) {
+      await expect(
+        store.publish({
+          event: mkEvent(`hello ${i}`),
+          notifiedSubscribers: [],
+        }),
+      ).resolves.toEqual({ eventId: String(i + 1) });
+    }
+
+    await expect(store.readSubscription('tester')).resolves.toEqual({
+      events: [
+        mkEvent('hello 15'),
+        mkEvent('hello 16'),
+        mkEvent('hello 17'),
+        mkEvent('hello 18'),
+        mkEvent('hello 19'),
+      ],
+    });
+  });
+});
diff --git a/plugins/events-backend/src/service/hub/MemoryEventBusStore.ts b/plugins/events-backend/src/service/hub/MemoryEventBusStore.ts
index b7b1d78052..67d384b1fa 100644
--- a/plugins/events-backend/src/service/hub/MemoryEventBusStore.ts
+++ b/plugins/events-backend/src/service/hub/MemoryEventBusStore.ts
@@ -18,8 +18,10 @@ import { EventBusStore } from './types';
 import { NotFoundError } from '@backstage/errors';
 
 const MAX_BATCH_SIZE = 10;
+const MAX_EVENTS_DEFAULT = 1_000;
 
 export class MemoryEventBusStore implements EventBusStore {
+  #maxEvents: number;
   #events = new Array<
     EventParams & { seq: number; notifiedSubscribers: Set<string> }
   >();
@@ -32,6 +34,10 @@ export class MemoryEventBusStore implements EventBusStore {
     resolve(result: { topic: string }): void;
   }>();
 
+  constructor(options: { maxEvents?: number } = {}) {
+    this.#maxEvents = options.maxEvents ?? MAX_EVENTS_DEFAULT;
+  }
+
   async publish(options: {
     event: EventParams;
     notifiedSubscribers: string[];
@@ -59,6 +65,12 @@ export class MemoryEventBusStore implements EventBusStore {
         this.#listeners.delete(listener);
       }
     }
+
+    // Trim old events
+    if (this.#events.length > this.#maxEvents) {
+      this.#events.shift();
+    }
+
     return { eventId: String(nextSeq) };
   }
 
@@ -96,7 +108,12 @@ export class MemoryEventBusStore implements EventBusStore {
 
     sub.seq = events[events.length - 1]?.seq ?? sub.seq;
 
-    return { events: events.map(event => ({ ...event, seq: undefined })) };
+    return {
+      events: events.map(({ topic, eventPayload }) => ({
+        topic,
+        eventPayload,
+      })),
+    };
   }
 
   async setupListener(