From 9a6080e6771e2a97837480683da877d6f07f4037 Mon Sep 17 00:00:00 2001 From: Hellgren Heikki Date: Tue, 18 Mar 2025 12:16:14 +0200 Subject: [PATCH] feat(notifications): support notification send throttling Signed-off-by: Hellgren Heikki --- .changeset/rotten-bobcats-notice.md | 5 + plugins/notifications-backend/config.d.ts | 32 ++++ plugins/notifications-backend/package.json | 6 +- .../src/service/router.ts | 138 +++++++++++------- yarn.lock | 2 + 5 files changed, 127 insertions(+), 56 deletions(-) create mode 100644 .changeset/rotten-bobcats-notice.md create mode 100644 plugins/notifications-backend/config.d.ts diff --git a/.changeset/rotten-bobcats-notice.md b/.changeset/rotten-bobcats-notice.md new file mode 100644 index 0000000000..0f6d1c74f9 --- /dev/null +++ b/.changeset/rotten-bobcats-notice.md @@ -0,0 +1,5 @@ +--- +'@backstage/plugin-notifications-backend': patch +--- + +Allow throttling notification sending not to block the system with huge amount of receiving users diff --git a/plugins/notifications-backend/config.d.ts b/plugins/notifications-backend/config.d.ts new file mode 100644 index 0000000000..5dd795f8d3 --- /dev/null +++ b/plugins/notifications-backend/config.d.ts @@ -0,0 +1,32 @@ +/* + * 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. + */ +import { HumanDuration } from '@backstage/types'; + +export interface Config { + /** + * Configuration options for notifications-backend + */ + notifications?: { + /** + * Concurrency limit for notification sending, defaults to 10 + */ + concurrencyLimit?: number; + /** + * Throttle duration between notification sending, defaults to 50ms + */ + throttleInterval?: HumanDuration | string; + }; +} diff --git a/plugins/notifications-backend/package.json b/plugins/notifications-backend/package.json index c96539caf8..6e9de108db 100644 --- a/plugins/notifications-backend/package.json +++ b/plugins/notifications-backend/package.json @@ -26,6 +26,7 @@ "types": "src/index.ts", "files": [ "dist", + "config.d.ts", "migrations/**/*.{js,d.ts}" ], "scripts": { @@ -49,9 +50,11 @@ "@backstage/plugin-notifications-common": "workspace:^", "@backstage/plugin-notifications-node": "workspace:^", "@backstage/plugin-signals-node": "workspace:^", + "@backstage/types": "workspace:^", "express": "^4.17.1", "express-promise-router": "^4.1.0", "knex": "^3.0.0", + "p-throttle": "^4.1.1", "uuid": "^11.0.0", "winston": "^3.2.1", "yn": "^4.0.0" @@ -68,5 +71,6 @@ "@types/supertest": "^2.0.8", "msw": "^1.0.0", "supertest": "^7.0.0" - } + }, + "configSchema": "config.d.ts" } diff --git a/plugins/notifications-backend/src/service/router.ts b/plugins/notifications-backend/src/service/router.ts index 368974b6ca..7f0279b437 100644 --- a/plugins/notifications-backend/src/service/router.ts +++ b/plugins/notifications-backend/src/service/router.ts @@ -48,7 +48,9 @@ import { } from '@backstage/plugin-notifications-common'; import { parseEntityOrderFieldParams } from './parseEntityOrderFieldParams'; import { getUsersForEntityRef } from './getUsersForEntityRef'; -import { Config } from '@backstage/config'; +import { Config, readDurationFromConfig } from '@backstage/config'; +import { durationToMilliseconds } from '@backstage/types'; +import pThrottle from 'p-throttle'; /** @internal */ export interface RouterOptions { @@ -82,6 +84,19 @@ export async function createRouter( const WEB_NOTIFICATION_CHANNEL = 'Web'; const store = await DatabaseNotificationsStore.create({ database }); const frontendBaseUrl = config.getString('app.baseUrl'); + const concurrencyLimit = + config.getOptionalNumber('notifications.concurrencyLimit') ?? 10; + const throttleInterval = config.has('notifications.throttleInterval') + ? durationToMilliseconds( + readDurationFromConfig(config, { + key: 'notifications.throttleInterval', + }), + ) + : 50; + const throttle = pThrottle({ + limit: concurrencyLimit, + interval: throttleInterval, + }); const getUser = async (req: Request) => { const credentials = await httpAuth.credentials(req, { allow: ['user'] }); @@ -470,71 +485,84 @@ export async function createRouter( }, channel: 'notifications', }); - postProcessNotification(ret, opts); } + postProcessNotification(ret, opts); return notification; }; + const sendUserNotification = async ( + baseNotification: Omit, + user: string, + opts: NotificationSendOptions, + origin: string, + scope?: string, + ): Promise => { + const userNotification = { + ...baseNotification, + id: uuid(), + user, + }; + const notification = await preProcessNotification(userNotification, opts); + + const enabled = await isNotificationsEnabled({ + user, + channel: WEB_NOTIFICATION_CHANNEL, + origin: userNotification.origin, + }); + + let ret = notification; + + if (!enabled) { + postProcessNotification(ret, opts); + return undefined; + } + + let existingNotification; + if (scope) { + existingNotification = await store.getExistingScopeNotification({ + user, + scope, + origin, + }); + } + + if (existingNotification) { + const restored = await store.restoreExistingNotification({ + id: existingNotification.id, + notification, + }); + ret = restored ?? notification; + } else { + await store.saveNotification(notification); + } + + if (signals) { + await signals.publish({ + recipients: { type: 'user', entityRef: [user] }, + message: { + action: 'new_notification', + notification_id: ret.id, + }, + channel: 'notifications', + }); + } + postProcessNotification(ret, opts); + return ret; + }; + const sendUserNotifications = async ( baseNotification: Omit, users: string[], opts: NotificationSendOptions, origin: string, - ) => { - const notifications = []; + ): Promise => { const { scope } = opts.payload; const uniqueUsers = [...new Set(users)]; - for (const user of uniqueUsers) { - const userNotification = { - ...baseNotification, - id: uuid(), - user, - }; - const notification = await preProcessNotification(userNotification, opts); - - const enabled = await isNotificationsEnabled({ - user, - channel: WEB_NOTIFICATION_CHANNEL, - origin: userNotification.origin, - }); - - let ret = notification; - if (enabled) { - let existingNotification; - if (scope) { - existingNotification = await store.getExistingScopeNotification({ - user, - scope, - origin, - }); - } - - if (existingNotification) { - const restored = await store.restoreExistingNotification({ - id: existingNotification.id, - notification, - }); - ret = restored ?? notification; - } else { - await store.saveNotification(notification); - } - - notifications.push(ret); - - if (signals) { - await signals.publish({ - recipients: { type: 'user', entityRef: [user] }, - message: { - action: 'new_notification', - notification_id: ret.id, - }, - channel: 'notifications', - }); - } - } - postProcessNotification(ret, opts); - } - return notifications; + const throttled = throttle((user: string) => + sendUserNotification(baseNotification, user, opts, origin, scope), + ); + const sent = await Promise.all(uniqueUsers.map(user => throttled(user))); + return sent.filter(n => n !== undefined); }; const createNotificationHandler = async ( diff --git a/yarn.lock b/yarn.lock index 0ea358c567..1f38a972bc 100644 --- a/yarn.lock +++ b/yarn.lock @@ -7053,12 +7053,14 @@ __metadata: "@backstage/plugin-notifications-node": "workspace:^" "@backstage/plugin-signals-backend": "workspace:^" "@backstage/plugin-signals-node": "workspace:^" + "@backstage/types": "workspace:^" "@types/express": ^4.17.6 "@types/supertest": ^2.0.8 express: ^4.17.1 express-promise-router: ^4.1.0 knex: ^3.0.0 msw: ^1.0.0 + p-throttle: ^4.1.1 supertest: ^7.0.0 uuid: ^11.0.0 winston: ^3.2.1