Merge pull request #23503 from drodil/broadcast_notifications

feat: support for broadcast notifications

.changeset/beige-pets-design.md

@@ -0,0 +1,7 @@
+---
+'@backstage/plugin-notifications-backend': patch
+'@backstage/plugin-notifications-common': patch
+'@backstage/plugin-notifications-node': patch
+---
+
+Support for broadcast notifications

plugins/notifications-backend/migrations/20240313_broadcast.js

@@ -0,0 +1,55 @@
+/*
+ * 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.
+ */
+
+exports.up = async function up(knex) {
+  await knex.schema.createTable('broadcast', table => {
+    table.uuid('id').primary();
+    table.string('title').notNullable();
+    table.text('description').nullable();
+    table.string('severity', 8).notNullable();
+    table.text('link').nullable();
+    table.string('origin', 255).notNullable();
+    table.string('scope', 255).nullable();
+    table.string('topic', 255).nullable();
+    table.datetime('created').defaultTo(knex.fn.now()).notNullable();
+    table.datetime('updated').nullable();
+    table.index(['scope', 'origin'], 'broadcast_cope_origin_idx');
+  });
+
+  await knex.schema.createTable('broadcast_user_status', table => {
+    table.uuid('broadcast_id').notNullable();
+    table.string('user', 255).notNullable();
+    table.datetime('read').nullable();
+    table.datetime('saved').nullable();
+
+    table
+      .foreign('broadcast_id')
+      .references(['id'])
+      .inTable('broadcast')
+      .onDelete('CASCADE');
+    table.unique(['broadcast_id', 'user'], {
+      indexName: 'broadcast_user_idx',
+    });
+  });
+};
+
+/**
+ * @param {import('knex').Knex} knex
+ */
+exports.down = async function down(knex) {
+  await knex.schema.dropTable('broadcast_user_status');
+  await knex.schema.dropTable('broadcast');
+};

plugins/notifications-backend/src/database/DatabaseNotificationsStore.test.ts

@@ -163,6 +163,7 @@ describe.each(databases.eachSupportedId())(
     afterEach(async () => {
       jest.resetAllMocks();
       await knex('notification').del();
+      await knex('broadcast').del();
     });
 
     describe('saveNotification', () => {
@@ -188,19 +189,21 @@ describe.each(databases.eachSupportedId())(
       it('should return all notifications for user', async () => {
         await storage.saveNotification(testNotification1);
         await storage.saveNotification(testNotification2);
+        await storage.saveBroadcast(testNotification3);
         await storage.saveNotification(otherUserNotification);
 
         const notifications = await storage.getNotifications({ user });
         expect(notifications.map(idOnly)).toEqual([
           /* default sorting from new to old */
           id2,
+          id3,
           id1,
         ]);
       });
 
       it('should return read notifications for user', async () => {
         await storage.saveNotification(testNotification1);
-        await storage.saveNotification(testNotification2);
+        await storage.saveBroadcast(testNotification2);
         await storage.saveNotification(testNotification3);
         await storage.saveNotification(otherUserNotification);
 
@@ -215,7 +218,7 @@ describe.each(databases.eachSupportedId())(
 
       it('should return unread notifications for user', async () => {
         await storage.saveNotification(testNotification1);
-        await storage.saveNotification(testNotification2);
+        await storage.saveBroadcast(testNotification2);
         await storage.saveNotification(testNotification3);
         await storage.saveNotification(otherUserNotification);
 
@@ -231,7 +234,7 @@ describe.each(databases.eachSupportedId())(
 
       it('should return both read and unread notifications for user', async () => {
         await storage.saveNotification(testNotification1);
-        await storage.saveNotification(testNotification2);
+        await storage.saveBroadcast(testNotification2);
         await storage.saveNotification(testNotification3);
 
         await storage.markRead({ ids: [id1, id3], user });
@@ -245,7 +248,7 @@ describe.each(databases.eachSupportedId())(
 
       it('should allow searching for notifications', async () => {
         await storage.saveNotification(testNotification2);
-        await storage.saveNotification(testNotification1);
+        await storage.saveBroadcast(testNotification1);
         await storage.saveNotification(otherUserNotification);
 
         const notifications = await storage.getNotifications({
@@ -258,7 +261,7 @@ describe.each(databases.eachSupportedId())(
 
       it('should filter notifications based on created date', async () => {
         await storage.saveNotification(testNotification1);
-        await storage.saveNotification(testNotification2);
+        await storage.saveBroadcast(testNotification2);
         await storage.saveNotification(otherUserNotification);
 
         const notifications = await storage.getNotifications({
@@ -335,7 +338,7 @@ describe.each(databases.eachSupportedId())(
         await storage.saveNotification(testNotification3);
         await storage.saveNotification(testNotification4);
         await storage.saveNotification(testNotification5);
-        await storage.saveNotification(testNotification6);
+        await storage.saveBroadcast(testNotification6);
         await storage.saveNotification(testNotification7);
         await storage.saveNotification(otherUserNotification);
       });
@@ -395,7 +398,7 @@ describe.each(databases.eachSupportedId())(
     describe('getNotifications sorting', () => {
       beforeEach(async () => {
         await storage.saveNotification(testNotification1);
-        await storage.saveNotification(testNotification2);
+        await storage.saveBroadcast(testNotification2);
         await storage.saveNotification(testNotification3);
       });
 
@@ -466,11 +469,13 @@ describe.each(databases.eachSupportedId())(
           read: new Date(),
         });
         await storage.saveNotification(testNotification2);
+        await storage.saveBroadcast({ ...testNotification3, read: new Date() });
+        await storage.saveBroadcast(testNotification4);
         await storage.saveNotification(otherUserNotification);
 
         const status = await storage.getStatus({ user });
-        expect(status.read).toEqual(1);
-        expect(status.unread).toEqual(1);
+        expect(status.read).toEqual(2);
+        expect(status.unread).toEqual(2);
       });
     });
 
@@ -487,6 +492,17 @@ describe.each(databases.eachSupportedId())(
         expect(existing).not.toBeNull();
         expect(existing?.id).toEqual(id2);
       });
+
+      it('should return existing scope broadcast', async () => {
+        await storage.saveBroadcast(testNotification1);
+        await storage.saveBroadcast(testNotification2);
+        const existing = await storage.getExistingScopeBroadcast({
+          origin: 'cd-origin',
+          scope: 'scaffolder-1234',
+        });
+        expect(existing).not.toBeNull();
+        expect(existing?.id).toEqual(id2);
+      });
     });
 
     describe('restoreExistingNotification', () => {
@@ -511,6 +527,27 @@ describe.each(databases.eachSupportedId())(
         expect(existing?.payload.severity).toEqual('low');
         expect(existing?.read).toBeNull();
       });
+
+      it('should return restore existing scope broadcast', async () => {
+        await storage.saveBroadcast(testNotification1);
+        await storage.saveBroadcast(testNotification2);
+
+        const existing = await storage.restoreExistingNotification({
+          id: id2,
+          notification: {
+            user: testNotification2.user,
+            payload: {
+              title: 'New notification',
+              link: '/scaffolder/task/1234',
+              severity: 'normal',
+            },
+          } as any,
+        });
+        expect(existing).not.toBeNull();
+        expect(existing?.id).toEqual(id2);
+        expect(existing?.payload.title).toEqual('New notification');
+        expect(existing?.read).toBeNull();
+      });
     });
 
     describe('getNotification', () => {
@@ -519,6 +556,13 @@ describe.each(databases.eachSupportedId())(
         const notification = await storage.getNotification({ id: id1 });
         expect(notification?.id).toEqual(id1);
       });
+
+      it('should return broadcast by id', async () => {
+        await storage.saveNotification(testNotification1);
+        await storage.saveBroadcast(testNotification2);
+        const notification = await storage.getNotification({ id: id2 });
+        expect(notification?.id).toEqual(id2);
+      });
     });
 
     describe('markRead', () => {
@@ -530,6 +574,15 @@ describe.each(databases.eachSupportedId())(
         const notification = await storage.getNotification({ id: id1 });
         expect(notification?.read).not.toBeNull();
       });
+
+      it('should mark broadcast read', async () => {
+        await storage.saveBroadcast(testNotification1);
+        const notificationBefore = await storage.getNotification({ id: id1 });
+        expect(notificationBefore?.read).toBeNull();
+        await storage.markRead({ ids: [id1], user });
+        const notification = await storage.getNotification({ id: id1 });
+        expect(notification?.read).not.toBeNull();
+      });
     });
 
     describe('markUnread', () => {
@@ -544,6 +597,18 @@ describe.each(databases.eachSupportedId())(
         const notification = await storage.getNotification({ id: id1 });
         expect(notification?.read).toBeNull();
       });
+
+      it('should mark broadcast unread', async () => {
+        await storage.saveBroadcast({
+          ...testNotification1,
+          read: new Date(),
+        });
+        const notificationBefore = await storage.getNotification({ id: id1 });
+        expect(notificationBefore?.read).not.toBeNull();
+        await storage.markUnread({ ids: [id1], user });
+        const notification = await storage.getNotification({ id: id1 });
+        expect(notification?.read).toBeNull();
+      });
     });
 
     describe('markSaved', () => {
@@ -555,6 +620,15 @@ describe.each(databases.eachSupportedId())(
         const notification = await storage.getNotification({ id: id1 });
         expect(notification?.saved).not.toBeNull();
       });
+
+      it('should mark broadcast saved', async () => {
+        await storage.saveBroadcast(testNotification1);
+        const notificationBefore = await storage.getNotification({ id: id1 });
+        expect(notificationBefore?.saved).toBeNull();
+        await storage.markSaved({ ids: [id1], user });
+        const notification = await storage.getNotification({ id: id1 });
+        expect(notification?.saved).not.toBeNull();
+      });
     });
 
     describe('markUnsaved', () => {
@@ -569,6 +643,18 @@ describe.each(databases.eachSupportedId())(
         const notification = await storage.getNotification({ id: id1 });
         expect(notification?.saved).toBeNull();
       });
+
+      it('should mark broadcast not saved', async () => {
+        await storage.saveBroadcast({
+          ...testNotification1,
+          saved: new Date(),
+        });
+        const notificationBefore = await storage.getNotification({ id: id1 });
+        expect(notificationBefore?.saved).not.toBeNull();
+        await storage.markUnsaved({ ids: [id1], user });
+        const notification = await storage.getNotification({ id: id1 });
+        expect(notification?.saved).toBeNull();
+      });
     });
   },
 );

plugins/notifications-backend/src/database/DatabaseNotificationsStore.ts

@@ -33,6 +33,22 @@ const migrationsDir = resolvePackagePath(
   'migrations',
 );
 
+const NOTIFICATION_COLUMNS = [
+  'id',
+  'title',
+  'description',
+  'severity',
+  'link',
+  'origin',
+  'scope',
+  'topic',
+  'created',
+  'updated',
+  'user',
+  'read',
+  'saved',
+];
+
 const severities: NotificationSeverity[] = [
   'critical',
   'high',
@@ -50,7 +66,11 @@ export const normalizeSeverity = (input?: string): NotificationSeverity => {
 
 /** @internal */
 export class DatabaseNotificationsStore implements NotificationsStore {
-  private constructor(private readonly db: Knex) {}
+  private readonly isSQLite = false;
+
+  private constructor(private readonly db: Knex) {
+    this.isSQLite = this.db.client.config.client.includes('sqlite3');
+  }
 
   static async create({
     database,
@@ -112,13 +132,44 @@ export class DatabaseNotificationsStore implements NotificationsStore {
     };
   };
 
+  private mapBroadcastToDbRow = (notification: Notification) => {
+    return {
+      id: notification.id,
+      origin: notification.origin,
+      created: notification.created,
+      topic: notification.payload?.topic,
+      link: notification.payload?.link,
+      title: notification.payload?.title,
+      description: notification.payload?.description,
+      severity: normalizeSeverity(notification.payload?.severity),
+      scope: notification.payload?.scope,
+    };
+  };
+
+  private getBroadcastUnion = () => {
+    return this.db('broadcast')
+      .leftJoin(
+        'broadcast_user_status',
+        'id',
+        '=',
+        'broadcast_user_status.broadcast_id',
+      )
+      .select(NOTIFICATION_COLUMNS);
+  };
+
   private getNotificationsBaseQuery = (
     options: NotificationGetOptions | NotificationModifyOptions,
   ) => {
     const { user } = options;
-    const isSQLite = this.db.client.config.client.includes('sqlite3');
 
-    const query = this.db('notification').where('user', user);
+    const subQuery = this.db('notification')
+      .select(NOTIFICATION_COLUMNS)
+      .unionAll([this.getBroadcastUnion()])
+      .as('notifications');
+
+    const query = this.db.from(subQuery).where(q => {
+      q.where('user', user).orWhereNull('user');
+    });
 
     if (options.sort !== undefined && options.sort !== null) {
       query.orderBy(options.sort, options.sortOrder ?? 'desc');
@@ -127,18 +178,10 @@ export class DatabaseNotificationsStore implements NotificationsStore {
     }
 
     if (options.createdAfter) {
-      if (isSQLite) {
-        query.where(
-          'notification.created',
-          '>=',
-          options.createdAfter.valueOf(),
-        );
+      if (this.isSQLite) {
+        query.where('created', '>=', options.createdAfter.valueOf());
       } else {
-        query.where(
-          'notification.created',
-          '>=',
-          options.createdAfter.toISOString(),
-        );
+        query.where('created', '>=', options.createdAfter.toISOString());
       }
     }
 
@@ -152,31 +195,31 @@ export class DatabaseNotificationsStore implements NotificationsStore {
 
     if (options.search) {
       query.whereRaw(
-        `(LOWER(notification.title) LIKE LOWER(?) OR LOWER(notification.description) LIKE LOWER(?))`,
+        `(LOWER(title) LIKE LOWER(?) OR LOWER(description) LIKE LOWER(?))`,
         [`%${options.search}%`, `%${options.search}%`],
       );
     }
 
     if (options.ids) {
-      query.whereIn('notification.id', options.ids);
+      query.whereIn('id', options.ids);
     }
 
     if (options.read) {
-      query.whereNotNull('notification.read');
+      query.whereNotNull('read');
     } else if (options.read === false) {
-      query.whereNull('notification.read');
+      query.whereNull('read');
     } // or match both if undefined
 
     if (options.saved) {
-      query.whereNotNull('notification.saved');
+      query.whereNotNull('saved');
     } else if (options.saved === false) {
-      query.whereNull('notification.saved');
+      query.whereNull('saved');
     } // or match both if undefined
 
     if (options.minimumSeverity !== undefined) {
       const idx = severities.indexOf(options.minimumSeverity);
       const equalOrHigher = severities.slice(0, idx + 1);
-      query.whereIn('notification.severity', equalOrHigher);
+      query.whereIn('severity', equalOrHigher);
     }
 
     return query;
@@ -184,7 +227,7 @@ export class DatabaseNotificationsStore implements NotificationsStore {
 
   async getNotifications(options: NotificationGetOptions) {
     const notificationQuery = this.getNotificationsBaseQuery(options);
-    const notifications = await notificationQuery.select();
+    const notifications = await notificationQuery.select(NOTIFICATION_COLUMNS);
     return this.mapToNotifications(notifications);
   }
 
@@ -194,7 +237,7 @@ export class DatabaseNotificationsStore implements NotificationsStore {
     countOptions.offset = undefined;
     countOptions.sort = null;
     const notificationQuery = this.getNotificationsBaseQuery(countOptions);
-    const response = await notificationQuery.count('* as CNT');
+    const response = await notificationQuery.count('id as CNT');
     return Number(response[0].CNT);
   }
 
@@ -204,6 +247,22 @@ export class DatabaseNotificationsStore implements NotificationsStore {
       .into('notification');
   }
 
+  async saveBroadcast(notification: Notification) {
+    await this.db
+      .insert(this.mapBroadcastToDbRow(notification))
+      .into('broadcast');
+    if (notification.saved || notification.read) {
+      await this.db
+        .insert({
+          user: notification.user,
+          broadcast_id: notification.id,
+          saved: notification.saved,
+          read: notification.read,
+        })
+        .into('broadcast_user_status');
+    }
+  }
+
   async getStatus(options: NotificationGetOptions) {
     const notificationQuery = this.getNotificationsBaseQuery({
       ...options,
@@ -239,7 +298,19 @@ export class DatabaseNotificationsStore implements NotificationsStore {
       .where('user', options.user)
       .where('scope', options.scope)
       .where('origin', options.origin)
-      .select()
+      .limit(1);
+
+    const rows = await query;
+    if (!rows || rows.length === 0) {
+      return null;
+    }
+    return rows[0] as Notification;
+  }
+
+  async getExistingScopeBroadcast(options: { scope: string; origin: string }) {
+    const query = this.db('broadcast')
+      .where('scope', options.scope)
+      .where('origin', options.origin)
       .limit(1);
 
     const rows = await query;
@@ -256,11 +327,7 @@ export class DatabaseNotificationsStore implements NotificationsStore {
     id: string;
     notification: Notification;
   }) {
-    const query = this.db('notification')
-      .where('id', id)
-      .where('user', notification.user);
-
-    await query.update({
+    const updateColumns = {
       title: notification.payload.title,
       description: notification.payload.description,
       link: notification.payload.link,
@@ -268,15 +335,31 @@ export class DatabaseNotificationsStore implements NotificationsStore {
       updated: new Date(),
       severity: normalizeSeverity(notification.payload?.severity),
       read: null,
-    });
+    };
+
+    const notificationQuery = this.db('notification')
+      .where('id', id)
+      .where('user', notification.user);
+    const broadcastQuery = this.db('broadcast').where('id', id);
+
+    await Promise.all([
+      notificationQuery.update(updateColumns),
+      broadcastQuery.update({ ...updateColumns, read: undefined }),
+    ]);
 
     return await this.getNotification({ id });
   }
 
   async getNotification(options: { id: string }): Promise<Notification | null> {
-    const rows = await this.db('notification')
+    const rows = await this.db
+      .select('*')
+      .from(
+        this.db('notification')
+          .select(NOTIFICATION_COLUMNS)
+          .unionAll([this.getBroadcastUnion()])
+          .as('notifications'),
+      )
       .where('id', options.id)
-      .select()
       .limit(1);
     if (!rows || rows.length === 0) {
       return null;
@@ -284,23 +367,65 @@ export class DatabaseNotificationsStore implements NotificationsStore {
     return this.mapToNotifications(rows)[0];
   }
 
+  private markReadSaved = async (
+    ids: string[],
+    user: string,
+    read?: Date | null,
+    saved?: Date | null,
+  ) => {
+    await this.db('notification')
+      .whereIn('id', ids)
+      .where('user', user)
+      .update({ read, saved });
+
+    const broadcasts = this.mapToNotifications(
+      await this.db('broadcast').whereIn('id', ids).select(),
+    );
+
+    if (broadcasts.length > 0)
+      if (!this.isSQLite) {
+        await this.db('broadcast_user_status')
+          .insert(
+            broadcasts.map(b => ({
+              broadcast_id: b.id,
+              user,
+              read,
+              saved,
+            })),
+          )
+          .onConflict(['broadcast_id', 'user'])
+          .merge(['read', 'saved']);
+      } else {
+        // SQLite does not support upsert so fall back to this (mostly for tests and local dev)
+        for (const b of broadcasts) {
+          const baseQuery = this.db('broadcast_user_status')
+            .where('broadcast_id', b.id)
+            .where('user', user);
+          const exists = await baseQuery.clone().limit(1).select().first();
+          if (exists) {
+            await baseQuery.clone().update({ read, saved });
+          } else {
+            await baseQuery
+              .clone()
+              .insert({ broadcast_id: b.id, user, read, saved });
+          }
+        }
+      }
+  };
+
   async markRead(options: NotificationModifyOptions): Promise<void> {
-    const notificationQuery = this.getNotificationsBaseQuery(options);
-    await notificationQuery.update({ read: new Date() });
+    await this.markReadSaved(options.ids, options.user, new Date(), undefined);
   }
 
   async markUnread(options: NotificationModifyOptions): Promise<void> {
-    const notificationQuery = this.getNotificationsBaseQuery(options);
-    await notificationQuery.update({ read: null });
+    await this.markReadSaved(options.ids, options.user, null, undefined);
   }
 
   async markSaved(options: NotificationModifyOptions): Promise<void> {
-    const notificationQuery = this.getNotificationsBaseQuery(options);
-    await notificationQuery.update({ saved: new Date() });
+    await this.markReadSaved(options.ids, options.user, undefined, new Date());
   }
 
   async markUnsaved(options: NotificationModifyOptions): Promise<void> {
-    const notificationQuery = this.getNotificationsBaseQuery(options);
-    await notificationQuery.update({ saved: null });
+    await this.markReadSaved(options.ids, options.user, undefined, null);
   }
 }

plugins/notifications-backend/src/database/NotificationsStore.ts

@@ -48,12 +48,19 @@ export interface NotificationsStore {
 
   saveNotification(notification: Notification): Promise<void>;
 
+  saveBroadcast(notification: Notification): Promise<void>;
+
   getExistingScopeNotification(options: {
     user: string;
     scope: string;
     origin: string;
   }): Promise<Notification | null>;
 
+  getExistingScopeBroadcast(options: {
+    scope: string;
+    origin: string;
+  }): Promise<Notification | null>;
+
   restoreExistingNotification(options: {
     id: string;
     notification: Notification;

plugins/notifications-backend/src/service/router.ts

@@ -115,7 +115,6 @@ export async function createRouter(
       targetPluginId: 'catalog',
     });
 
-    // TODO: Support for broadcast
     if (entityRef === null) {
       return [];
     }
@@ -175,7 +174,7 @@ export async function createRouter(
   };
 
   const decorateNotification = async (notification: Notification) => {
-    let ret: Notification = notification;
+    let ret = notification;
     for (const processor of processors ?? []) {
       ret = processor.decorate ? await processor.decorate(ret) : ret;
     }
@@ -314,43 +313,59 @@ export async function createRouter(
     res.status(200).send(notifications);
   });
 
-  // Add new notification
-  router.post('/', async (req, res) => {
-    const { recipients, payload } = req.body;
-    const notifications = [];
-    let users = [];
-
-    const credentials = await httpAuth.credentials(req, { allow: ['service'] });
-
-    const { title, scope } = payload;
-
-    if (!recipients || !title) {
-      logger.error(`Invalid notification request received`);
-      throw new InputError();
-    }
-
-    let entityRef = null;
-    // TODO: Support for broadcast notifications
-    if (recipients.entityRef && recipients.type === 'entity') {
-      entityRef = recipients.entityRef;
-    }
-
-    try {
-      users = await getUsersForEntityRef(entityRef);
-    } catch (e) {
-      throw new InputError();
-    }
-
-    const origin = credentials.principal.subject;
-    const baseNotification: Omit<Notification, 'id' | 'user'> = {
-      payload: {
-        ...payload,
-        severity: payload.severity ?? 'normal',
-      },
-      origin,
-      created: new Date(),
+  const sendBroadcastNotification = async (
+    baseNotification: Omit<Notification, 'user' | 'id'>,
+    opts: { scope?: string; origin: string },
+  ) => {
+    const { scope, origin } = opts;
+    const broadcastNotification = {
+      ...baseNotification,
+      id: uuid(),
     };
+    const notification = await decorateNotification({
+      ...broadcastNotification,
+      user: '',
+    });
+    let existingNotification;
+    if (scope) {
+      existingNotification = await store.getExistingScopeBroadcast({
+        scope,
+        origin,
+      });
+    }
 
+    let ret = notification;
+    if (existingNotification) {
+      const restored = await store.restoreExistingNotification({
+        id: existingNotification.id,
+        notification: { ...notification, user: '' },
+      });
+      ret = restored ?? notification;
+    } else {
+      await store.saveBroadcast(notification);
+    }
+    processorSendNotification(ret);
+
+    if (signals) {
+      await signals.publish<NewNotificationSignal>({
+        recipients: null,
+        message: {
+          action: 'new_notification',
+          notification_id: ret.id,
+        },
+        channel: 'notifications',
+      });
+    }
+    return notification;
+  };
+
+  const sendUserNotifications = async (
+    baseNotification: Omit<Notification, 'user' | 'id'>,
+    users: string[],
+    opts: { scope?: string; origin: string },
+  ) => {
+    const notifications = [];
+    const { scope, origin } = opts;
     const uniqueUsers = [...new Set(users)];
     for (const user of uniqueUsers) {
       const userNotification = {
@@ -394,6 +409,55 @@ export async function createRouter(
         });
       }
     }
+    return notifications;
+  };
+
+  // Add new notification
+  router.post('/', async (req, res) => {
+    const { recipients, payload } = req.body;
+    const notifications = [];
+    let users = [];
+
+    const credentials = await httpAuth.credentials(req, { allow: ['service'] });
+
+    const { title, scope } = payload;
+
+    if (!recipients || !title) {
+      logger.error(`Invalid notification request received`);
+      throw new InputError();
+    }
+
+    const origin = credentials.principal.subject;
+    const baseNotification = {
+      payload: {
+        ...payload,
+        severity: payload.severity ?? 'normal',
+      },
+      origin,
+      created: new Date(),
+    };
+
+    if (recipients.type === 'broadcast') {
+      const broadcast = await sendBroadcastNotification(baseNotification, {
+        scope,
+        origin,
+      });
+      notifications.push(broadcast);
+    } else {
+      const entityRef = recipients.entityRef;
+
+      try {
+        users = await getUsersForEntityRef(entityRef);
+      } catch (e) {
+        throw new InputError();
+      }
+      const userNotifications = await sendUserNotifications(
+        baseNotification,
+        users,
+        { scope, origin },
+      );
+      notifications.push(...userNotifications);
+    }
 
     res.json(notifications);
   });

plugins/notifications-node/api-report.md

@@ -27,10 +27,14 @@ export interface NotificationProcessor {
 }
 
 // @public (undocumented)
-export type NotificationRecipients = {
-  type: 'entity';
-  entityRef: string | string[];
-};
+export type NotificationRecipients =
+  | {
+      type: 'entity';
+      entityRef: string | string[];
+    }
+  | {
+      type: 'broadcast';
+    };
 
 // @public (undocumented)
 export type NotificationSendOptions = {

plugins/notifications-node/src/service/DefaultNotificationService.ts

@@ -25,13 +25,12 @@ export type NotificationServiceOptions = {
 };
 
 /** @public */
-export type NotificationRecipients = {
-  type: 'entity';
-  entityRef: string | string[];
-};
-
-// TODO: Support for broadcast messages
-//  | { type: 'broadcast' };
+export type NotificationRecipients =
+  | {
+      type: 'entity';
+      entityRef: string | string[];
+    }
+  | { type: 'broadcast' };
 
 /** @public */
 export type NotificationSendOptions = {