events-backend: drop optimization for sparse events

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

plugins/events-backend/src/service/hub/DatabaseEventBusStore.test.ts

@@ -210,7 +210,7 @@ describe('DatabaseEventBusStore', () => {
       // in a real-world scenario given our count window size is 10,000.
       await db.raw(`
         INSERT INTO event_bus_events (id, created_by, topic, data_json, notified_subscribers)
-        SELECT id, 'abc', CONCAT('test-', id), '{}', '{"${String(
+        SELECT id, 'abc', CONCAT('test-', MOD(id, 10)), CONCAT('{"payload":{"id":"', id, '"}}'), '{"${String(
           Math.random(),
         ).slice(2, 6)}"}'
         FROM generate_series(1, ${COUNT}) AS id
@@ -219,18 +219,26 @@ describe('DatabaseEventBusStore', () => {
         id: 'tester',
         created_by: 'abc',
         read_until: 0,
-        topics: ['test-1000'],
+        topics: ['test-5'],
       });
 
       const start = Date.now();
       const { events } = await store.readSubscription('tester');
       const duration = Date.now() - start;
 
-      expect(events).toEqual([{ topic: 'test-1000', eventPayload: undefined }]);
+      expect(events).toEqual([
+        { topic: 'test-5', eventPayload: { id: '5' } },
+        { topic: 'test-5', eventPayload: { id: '15' } },
+        { topic: 'test-5', eventPayload: { id: '25' } },
+        { topic: 'test-5', eventPayload: { id: '35' } },
+        { topic: 'test-5', eventPayload: { id: '45' } },
+        { topic: 'test-5', eventPayload: { id: '55' } },
+        { topic: 'test-5', eventPayload: { id: '65' } },
+        { topic: 'test-5', eventPayload: { id: '75' } },
+        { topic: 'test-5', eventPayload: { id: '85' } },
+        { topic: 'test-5', eventPayload: { id: '95' } },
+      ]);
 
-      // When not run in an optimized way this takes anywhere from 10ms to
-      // 100ms. When optimized it's closed to 1ms, but we leave a bit of room
-      // for CI load spikes.
       expect(duration).toBeLessThan(20);
     },
   );

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

@@ -472,15 +472,6 @@ export class DatabaseEventBusStore implements EventBusStore {
     // an update, reads events for the subscription up to the limit, and then
     // updates the pointer to the last read event.
     //
-    // The query for selected_events is written in this particular way to force
-    // evaluation of the notified subscribers last. Without this, the query
-    // planner loves to first do a sequential scan of the events table to filter
-    // out the events that have already been consumed, but that is often a small
-    // subset and extremely wasteful. We instead want to make sure that the
-    // query is executed such that we select the events that are relevant to the
-    // subscription first, and then filter out any events that have already been
-    // consumed last.
-    //
     // This is written as a plain SQL query to spare us all the horrors of
     // expressing this in knex.
 
@@ -499,13 +490,8 @@ export class DatabaseEventBusStore implements EventBusStore {
         FROM event_bus_events
         INNER JOIN subscription
         ON event_bus_events.topic = ANY(subscription.topics)
-        WHERE (
-          CASE WHEN event_bus_events.id > subscription.read_until THEN
-            CASE WHEN NOT :id = ANY(event_bus_events.notified_subscribers)
-              THEN 1
-            END
-          END
-        ) = 1
+        WHERE event_bus_events.id > subscription.read_until
+        AND NOT :id = ANY(event_bus_events.notified_subscribers)
         ORDER BY event_bus_events.id ASC LIMIT :limit
       ),
       last_event_id AS (