events-backend: use consumedBy everywhere instead of subscriptionIds

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',
                   },
-                  subscriptionIds: {
+                  consumedBy: {
                     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'
-                subscriptionIds:
+                consumedBy:
                   type: array
                   description: The IDs of subscriptions that have already received this event
                   items:

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

@@ -323,9 +323,10 @@ export class DatabaseEventBusStore implements EventBusStore {
 
   async publish(options: {
     params: EventParams;
-    subscriberIds: string[];
+    consumedBy?: string[];
   }): Promise<{ id: string } | undefined> {
     const topic = options.params.topic;
+    const consumedBy = options.consumedBy ?? [];
     // 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
@@ -351,12 +352,12 @@ export class DatabaseEventBusStore implements EventBusStore {
                   metadata: options.params.metadata,
                 }),
               ]),
-              this.#db.raw('?', [options.subscriberIds]),
+              this.#db.raw('?', [consumedBy]),
             )
             // The rest of this query is to check whether there are any
             // subscribers that have not been notified yet
             .from(TABLE_SUBSCRIPTIONS)
-            .whereNotIn('id', options.subscriberIds) // Skip notified subscribers
+            .whereNotIn('id', consumedBy) // 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

@@ -19,9 +19,7 @@ import { EventBusStore } from './types';
 const MAX_BATCH_SIZE = 5;
 
 export class MemoryEventBusStore implements EventBusStore {
-  #events = new Array<
-    EventParams & { seq: number; subscriberIds: Set<string> }
-  >();
+  #events = new Array<EventParams & { seq: number; consumedBy: Set<string> }>();
   #subscribers = new Map<
     string,
     { id: string; seq: number; topics: Set<string> }
@@ -33,14 +31,14 @@ export class MemoryEventBusStore implements EventBusStore {
 
   async publish(options: {
     params: EventParams;
-    subscriberIds: string[];
+    consumedBy: string[];
   }): Promise<{ id: string } | undefined> {
     const topic = options.params.topic;
-    const subscriberIds = new Set(options.subscriberIds);
+    const consumedBy = new Set(options.consumedBy);
 
     let hasOtherSubscribers = false;
     for (const sub of this.#subscribers.values()) {
-      if (sub.topics.has(topic) && !subscriberIds.has(sub.id)) {
+      if (sub.topics.has(topic) && !consumedBy.has(sub.id)) {
         hasOtherSubscribers = true;
         break;
       }
@@ -50,7 +48,7 @@ export class MemoryEventBusStore implements EventBusStore {
     }
 
     const nextSeq = this.#getMaxSeq() + 1;
-    this.#events.push({ ...options.params, subscriberIds, seq: nextSeq });
+    this.#events.push({ ...options.params, consumedBy, seq: nextSeq });
 
     for (const listener of this.#listeners) {
       if (listener.topics.has(topic)) {
@@ -89,7 +87,7 @@ export class MemoryEventBusStore implements EventBusStore {
         event =>
           event.seq > sub.seq &&
           sub.topics.has(event.topic) &&
-          !event.subscriberIds.has(id),
+          !event.consumedBy.has(id),
       )
       .slice(0, MAX_BATCH_SIZE);
 

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

@@ -160,13 +160,13 @@ export async function createEventBusRouter(options: {
       allow: ['service'],
     });
     const topic = req.body.event.topic;
-    const subscriberIds = req.body.subscriptionIds ?? [];
+    const consumedBy = req.body.consumedBy;
     const result = await store.publish({
       params: {
         topic,
         eventPayload: req.body.event.payload,
       } as EventParams,
-      subscriberIds,
+      consumedBy: req.body.consumedBy,
     });
     if (result) {
       logger.info(`Published event to '${topic}' with ID '${result.id}'`, {
@@ -174,14 +174,18 @@ export async function createEventBusRouter(options: {
       });
       res.status(201).end();
     } else {
-      logger.info(
-        `Skipped publishing of event to '${topic}', subscribers have already been notified: '${subscriberIds.join(
-          "', '",
-        )}'`,
-        {
-          subject: credentials.principal.subject,
-        },
-      );
+      if (consumedBy) {
+        const notified = `'${consumedBy.join("', '")}'`;
+        logger.info(
+          `Skipped publishing of event to '${topic}', subscribers have already been notified: ${notified}`,
+          { subject: credentials.principal.subject },
+        );
+      } else {
+        logger.info(
+          `Skipped publishing of event to '${topic}', no subscribers present`,
+          { subject: credentials.principal.subject },
+        );
+      }
       res.status(204).end();
     }
   });
@@ -267,7 +271,7 @@ export async function createEventBusRouter(options: {
     await store.upsertSubscription(id, req.body.topics);
 
     logger.info(
-      `New subscription '${id}' topics='${req.body.topics.join("', '")}'`,
+      `New subscription '${id}' for topics '${req.body.topics.join("', '")}'`,
       { 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;
-    subscriberIds: string[];
+    consumedBy?: string[];
   }): Promise<{ id: string } | undefined>;
 
   upsertSubscription(id: 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 },
-            subscriptionIds: notifiedSubscribers,
+            consumedBy: 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
    */
-  subscriptionIds?: Array<string>;
+  consumedBy?: Array<string>;
 }