events-backend: rename column consumed_by -> notified_subscribers

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

plugins/events-backend/migrations/20240523100528_init.js

@@ -44,7 +44,7 @@ exports.up = async function up(knex) {
         .notNullable()
         .comment('The payload data of this event');
       table
-        .specificType('consumed_by', 'text ARRAY')
+        .specificType('notified_subscribers', 'text ARRAY')
         .comment(
           'The IDs of the subscribers that have already consumed this event',
         );

plugins/events-backend/src/migrations.test.ts

@@ -58,7 +58,7 @@ describe('migrations', () => {
         topic: 'test',
         created_by: 'abc',
         data_json: JSON.stringify({ message: 'hello' }),
-        consumed_by: ['tester'],
+        notified_subscribers: ['tester'],
       });
       await knex('event_bus_subscriptions').insert({
         id: 'tester',
@@ -74,7 +74,7 @@ describe('migrations', () => {
           topic: 'test',
           data_json: JSON.stringify({ message: 'hello' }),
           created_at: expect.anything(),
-          consumed_by: ['tester'],
+          notified_subscribers: ['tester'],
         },
       ]);
       await expect(knex('event_bus_subscriptions')).resolves.toEqual([

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

@@ -209,7 +209,7 @@ describe('DatabaseEventBusStore', () => {
       // Insert 100,000 events, a lot more than we'd expect to ever have
       // 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, consumed_by)
+        INSERT INTO event_bus_events (id, created_by, topic, data_json, notified_subscribers)
         SELECT id, 'abc', CONCAT('test-', id), '{}', '{"${String(
           Math.random(),
         ).slice(2, 6)}"}'

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

@@ -46,7 +46,7 @@ type EventsRow = {
   created_at: Date;
   topic: string;
   data_json: string;
-  consumed_by: string[];
+  notified_subscribers: string[];
 };
 
 type SubscriptionsRow = {
@@ -391,7 +391,7 @@ export class DatabaseEventBusStore implements EventBusStore {
           'created_by',
           'topic',
           'data_json',
-          'consumed_by',
+          'notified_subscribers',
         ]),
       )
       .insert<EventsRow>(
@@ -501,7 +501,7 @@ export class DatabaseEventBusStore implements EventBusStore {
         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.consumed_by)
+            CASE WHEN NOT :id = ANY(event_bus_events.notified_subscribers)
               THEN 1
             END
           END