events-backend: rename consumedBy -> notifiedSubscribers

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

plugins/events-backend/src/schema/openapi.generated.ts

@@ -159,7 +159,7 @@ export const spec = {
                   event: {
                     $ref: '#/components/schemas/Event',
                   },
-                  consumedBy: {
+                  notifiedSubscribers: {
                     type: 'array',
                     description:
                       'The IDs of subscriptions that have already received this event',

plugins/events-backend/src/schema/openapi.yaml

@@ -106,7 +106,7 @@ paths:
               properties:
                 event:
                   $ref: '#/components/schemas/Event'
-                consumedBy:
+                notifiedSubscribers:
                   type: array
                   description: The IDs of subscriptions that have already received this event
                   items:

plugins/events-backend/src/service/EventsPlugin.test.ts

@@ -119,12 +119,15 @@ describe('eventsPlugin', () => {
       publish(
         topic: string,
         payload: unknown,
-        options?: { consumedBy?: string[] },
+        options?: { notifiedSubscribers?: string[] },
       ) {
         return request(this.backend.server)
           .post('/api/events/bus/v1/events')
           .set('authorization', mockCredentials.service.header())
-          .send({ event: { topic, payload }, consumedBy: options?.consumedBy });
+          .send({
+            event: { topic, payload },
+            notifiedSubscribers: options?.notifiedSubscribers,
+          });
       }
 
       readEvents(id: string) {
@@ -278,7 +281,7 @@ describe('eventsPlugin', () => {
             'test',
             { for: 'tester-2' },
             {
-              consumedBy: ['tester-1'],
+              notifiedSubscribers: ['tester-1'],
             },
           )
           .expect(201);
@@ -287,7 +290,7 @@ describe('eventsPlugin', () => {
             'test',
             { for: 'tester-1' },
             {
-              consumedBy: ['tester-2'],
+              notifiedSubscribers: ['tester-2'],
             },
           )
           .expect(201);
@@ -362,7 +365,11 @@ describe('eventsPlugin', () => {
         await helper.subscribe('tester', ['test']).expect(201);
 
         await helper
-          .publish('test', { for: 'tester-2' }, { consumedBy: ['tester'] })
+          .publish(
+            'test',
+            { for: 'tester-2' },
+            { notifiedSubscribers: ['tester'] },
+          )
           .expect(204);
 
         await backend.stop();

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

@@ -349,10 +349,10 @@ export class DatabaseEventBusStore implements EventBusStore {
 
   async publish(options: {
     params: EventParams;
-    consumedBy?: string[];
+    notifiedSubscribers?: string[];
   }): Promise<{ eventId: string } | undefined> {
     const topic = options.params.topic;
-    const consumedBy = options.consumedBy ?? [];
+    const notifiedSubscribers = options.notifiedSubscribers ?? [];
     // This query inserts a new event into the database, but only if there are
     // subscribers to the topic that have not already been notified
     const result = await this.#db
@@ -378,12 +378,12 @@ export class DatabaseEventBusStore implements EventBusStore {
                   metadata: options.params.metadata,
                 }),
               ]),
-              this.#db.raw('?', [consumedBy]),
+              this.#db.raw('?', [notifiedSubscribers]),
             )
             // The rest of this query is to check whether there are any
             // subscribers that have not been notified yet
             .from(TABLE_SUBSCRIPTIONS)
-            .whereNotIn('id', consumedBy) // Skip notified subscribers
+            .whereNotIn('id', notifiedSubscribers) // Skip notified subscribers
             .andWhere(this.#db.raw('? = ANY(topics)', [topic])) // Match topic
             .having(this.#db.raw('count(*)'), '>', 0), // Check if there are any results
       )

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

@@ -20,7 +20,9 @@ import { NotFoundError } from '@backstage/errors';
 const MAX_BATCH_SIZE = 10;
 
 export class MemoryEventBusStore implements EventBusStore {
-  #events = new Array<EventParams & { seq: number; consumedBy: Set<string> }>();
+  #events = new Array<
+    EventParams & { seq: number; notifiedSubscribers: Set<string> }
+  >();
   #subscribers = new Map<
     string,
     { id: string; seq: number; topics: Set<string> }
@@ -32,14 +34,14 @@ export class MemoryEventBusStore implements EventBusStore {
 
   async publish(options: {
     params: EventParams;
-    consumedBy: string[];
+    notifiedSubscribers: string[];
   }): Promise<{ eventId: string } | undefined> {
     const topic = options.params.topic;
-    const consumedBy = new Set(options.consumedBy);
+    const notifiedSubscribers = new Set(options.notifiedSubscribers);
 
     let hasOtherSubscribers = false;
     for (const sub of this.#subscribers.values()) {
-      if (sub.topics.has(topic) && !consumedBy.has(sub.id)) {
+      if (sub.topics.has(topic) && !notifiedSubscribers.has(sub.id)) {
         hasOtherSubscribers = true;
         break;
       }
@@ -49,7 +51,7 @@ export class MemoryEventBusStore implements EventBusStore {
     }
 
     const nextSeq = this.#getMaxSeq() + 1;
-    this.#events.push({ ...options.params, consumedBy, seq: nextSeq });
+    this.#events.push({ ...options.params, notifiedSubscribers, seq: nextSeq });
 
     for (const listener of this.#listeners) {
       if (listener.topics.has(topic)) {
@@ -88,7 +90,7 @@ export class MemoryEventBusStore implements EventBusStore {
         event =>
           event.seq > sub.seq &&
           sub.topics.has(event.topic) &&
-          !event.consumedBy.has(id),
+          !event.notifiedSubscribers.has(id),
       )
       .slice(0, MAX_BATCH_SIZE);
 

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

@@ -158,13 +158,13 @@ export async function createEventBusRouter(options: {
       allow: ['service'],
     });
     const topic = req.body.event.topic;
-    const consumedBy = req.body.consumedBy;
+    const notifiedSubscribers = req.body.notifiedSubscribers;
     const result = await store.publish({
       params: {
         topic,
         eventPayload: req.body.event.payload,
       } as EventParams,
-      consumedBy: req.body.consumedBy,
+      notifiedSubscribers: req.body.notifiedSubscribers,
     });
     if (result) {
       logger.info(`Published event to '${topic}' with ID '${result.eventId}'`, {
@@ -172,8 +172,8 @@ export async function createEventBusRouter(options: {
       });
       res.status(201).end();
     } else {
-      if (consumedBy) {
-        const notified = `'${consumedBy.join("', '")}'`;
+      if (notifiedSubscribers) {
+        const notified = `'${notifiedSubscribers.join("', '")}'`;
         logger.info(
           `Skipped publishing of event to '${topic}', subscribers have already been notified: ${notified}`,
           { subject: credentials.principal.subject },

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

@@ -19,7 +19,7 @@ import { EventParams } from '@backstage/plugin-events-node';
 export type EventBusStore = {
   publish(options: {
     params: EventParams;
-    consumedBy?: string[];
+    notifiedSubscribers?: string[];
   }): Promise<{ eventId: string } | undefined>;
 
   upsertSubscription(subscriptionId: string, topics: string[]): Promise<void>;

plugins/events-node/src/api/DefaultEventsService.ts

@@ -127,7 +127,7 @@ class PluginEventsService implements EventsService {
         {
           body: {
             event: { payload: params.eventPayload, topic: params.topic },
-            consumedBy: notifiedSubscribers,
+            notifiedSubscribers,
           },
         },
         { token },

plugins/events-node/src/generated/models/PostEventRequest.model.ts

@@ -24,5 +24,5 @@ export interface PostEventRequest {
   /**
    * The IDs of subscriptions that have already received this event
    */
-  consumedBy?: Array<string>;
+  notifiedSubscribers?: Array<string>;
 }