feat: clean up old notifications

a new scheduled task that will delete old notifications. the default is
that over 1 year old notifications will be deleted. the scheduled task
is run every 24 hours. this can be disabled by setting the
retention period to false in the notifications config.

Signed-off-by: Hellgren Heikki <heikki.hellgren@op.fi>

plugins/notifications-backend/config.d.ts

@@ -28,5 +28,10 @@ export interface Config {
      * Throttle duration between notification sending, defaults to 50ms
      */
     throttleInterval?: HumanDuration | string;
+    /**
+     * Time to keep the notifications in the database, defaults to 365 days.
+     * Can be disabled by setting to false.
+     */
+    retention?: HumanDuration | string | false;
   };
 }

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

@@ -805,5 +805,24 @@ describe.each(databases.eachSupportedId())(
         });
       });
     });
+
+    describe('clearNotifications', () => {
+      it('should clear notifications older than specified days', async () => {
+        const oldDate = new Date();
+        oldDate.setDate(oldDate.getDate() - 10); // 10 days ago
+        await storage.saveNotification({
+          ...testNotification1,
+          created: oldDate,
+        });
+        await storage.saveNotification(testNotification2);
+
+        const result = await storage.clearNotifications({
+          maxAge: { days: 5 },
+        }); // Clear notifications older than 5 days
+        expect(result.deletedCount).toBe(1); // Only the first notification should be cleared
+        const remainingNotifications = await storage.getNotifications({ user });
+        expect(remainingNotifications.map(idOnly)).toEqual([id2]);
+      });
+    });
   },
 );

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

@@ -31,6 +31,7 @@ import {
 } from '@backstage/plugin-notifications-common';
 import { Knex } from 'knex';
 import crypto from 'crypto';
+import { durationToMilliseconds, HumanDuration } from '@backstage/types';
 
 const migrationsDir = resolvePackagePath(
   '@backstage/plugin-notifications-backend',
@@ -656,4 +657,24 @@ export class DatabaseNotificationsStore implements NotificationsStore {
       .distinct(['topic']);
     return { topics: topics.map(row => row.topic) };
   }
+
+  async clearNotifications(options: {
+    maxAge: HumanDuration;
+  }): Promise<{ deletedCount: number }> {
+    const ms = durationToMilliseconds(options.maxAge);
+    const now = new Date(new Date().getTime() - ms);
+    const notificationsCount = await this.db('notification')
+      .where(builder => {
+        builder.where('created', '<=', now).whereNull('updated');
+      })
+      .orWhere('updated', '<=', now)
+      .delete();
+    const broadcastsCount = await this.db('broadcast')
+      .where(builder => {
+        builder.where('created', '<=', now).whereNull('updated');
+      })
+      .orWhere('updated', '<=', now)
+      .delete();
+    return { deletedCount: notificationsCount + broadcastsCount };
+  }
 }

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

@@ -20,6 +20,7 @@ import {
   NotificationSeverity,
   NotificationStatus,
 } from '@backstage/plugin-notifications-common';
+import { HumanDuration } from '@backstage/types';
 
 /** @internal */
 export type EntityOrder = {
@@ -99,6 +100,10 @@ export interface NotificationsStore {
     user: string;
   }): Promise<{ origins: string[] }>;
 
+  getUserNotificationTopics(options: {
+    user: string;
+  }): Promise<{ topics: { origin: string; topic: string }[] }>;
+
   getNotificationSettings(options: {
     user: string;
   }): Promise<NotificationSettings>;
@@ -109,4 +114,8 @@ export interface NotificationsStore {
   }): Promise<void>;
 
   getTopics(options: TopicGetOptions): Promise<{ topics: string[] }>;
+
+  clearNotifications(options: {
+    maxAge: HumanDuration;
+  }): Promise<{ deletedCount: number }>;
 }

plugins/notifications-backend/src/plugin.ts

@@ -26,6 +26,8 @@ import {
   NotificationsProcessingExtensionPoint,
 } from '@backstage/plugin-notifications-node';
 import { catalogServiceRef } from '@backstage/plugin-catalog-node';
+import { DatabaseNotificationsStore } from './database';
+import { NotificationCleaner } from './service/NotificationCleaner.ts';
 
 class NotificationsProcessingExtensionPointImpl
   implements NotificationsProcessingExtensionPoint
@@ -69,6 +71,7 @@ export const notificationsPlugin = createBackendPlugin({
         signals: signalsServiceRef,
         config: coreServices.rootConfig,
         catalog: catalogServiceRef,
+        scheduler: coreServices.scheduler,
       },
       async init({
         auth,
@@ -80,7 +83,10 @@ export const notificationsPlugin = createBackendPlugin({
         signals,
         config,
         catalog,
+        scheduler,
       }) {
+        const store = await DatabaseNotificationsStore.create({ database });
+
         httpRouter.use(
           await createRouter({
             auth,
@@ -88,7 +94,7 @@ export const notificationsPlugin = createBackendPlugin({
             userInfo,
             logger,
             config,
-            database,
+            store,
             catalog,
             signals,
             processors: processingExtensions.processors,
@@ -98,6 +104,14 @@ export const notificationsPlugin = createBackendPlugin({
           path: '/health',
           allow: 'unauthenticated',
         });
+
+        const cleaner = new NotificationCleaner(
+          config,
+          scheduler,
+          logger,
+          store,
+        );
+        await cleaner.initTaskRunner();
       },
     });
   },

plugins/notifications-backend/src/service/NotificationCleaner.test.ts

@@ -0,0 +1,118 @@
+/*
+ * Copyright 2025 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 { LoggerService, SchedulerService } from '@backstage/backend-plugin-api';
+import { Config } from '@backstage/config';
+import { mockServices } from '@backstage/backend-test-utils';
+import { NotificationsStore } from '../database';
+import { NotificationCleaner } from './NotificationCleaner.ts';
+
+describe('NotificationCleaner', () => {
+  let mockConfig: Config;
+  let mockScheduler: SchedulerService;
+  let mockLogger: LoggerService;
+  let mockDatabase: NotificationsStore;
+
+  beforeEach(() => {
+    mockConfig = mockServices.rootConfig();
+    mockScheduler = mockServices.scheduler.mock();
+    mockLogger = mockServices.logger.mock();
+    mockDatabase = {
+      clearNotifications: jest.fn(),
+    } as unknown as NotificationsStore;
+  });
+
+  afterEach(() => {
+    jest.clearAllMocks();
+  });
+
+  describe('initNotificationCleaner', () => {
+    it('should initialize the notification cleaner with the correct schedule', async () => {
+      const mockTaskRunner = {
+        run: jest.fn(),
+      };
+      mockScheduler.createScheduledTaskRunner = jest
+        .fn()
+        .mockReturnValue(mockTaskRunner);
+
+      const cleaner = new NotificationCleaner(
+        mockConfig,
+        mockScheduler,
+        mockLogger,
+        mockDatabase,
+      );
+      expect(cleaner).toBeInstanceOf(NotificationCleaner);
+      await cleaner.initTaskRunner();
+
+      expect(mockScheduler.createScheduledTaskRunner).toHaveBeenCalled();
+      expect(mockTaskRunner.run).toHaveBeenCalledWith(
+        expect.objectContaining({
+          id: 'notification-cleaner',
+          fn: expect.any(Function),
+        }),
+      );
+    });
+
+    it('should not create a task runner if retention is disabled', async () => {
+      mockConfig = mockServices.rootConfig({
+        data: { notifications: { retention: false } },
+      });
+      const cleaner = new NotificationCleaner(
+        mockConfig,
+        mockScheduler,
+        mockLogger,
+        mockDatabase,
+      );
+      await cleaner.initTaskRunner();
+
+      expect(mockScheduler.createScheduledTaskRunner).not.toHaveBeenCalled();
+      expect(mockLogger.info).toHaveBeenCalledWith(
+        'Notification retention is disabled, skipping notification cleaner task',
+      );
+    });
+  });
+
+  describe('clearNotifications', () => {
+    it('should clear notifications', async () => {
+      mockDatabase.clearNotifications = jest
+        .fn()
+        .mockResolvedValue({ deletedCount: 1 });
+      const mockTaskRunner = {
+        run: jest.fn().mockImplementation(({ fn }) => fn()),
+      };
+      mockScheduler.createScheduledTaskRunner = jest
+        .fn()
+        .mockReturnValue(mockTaskRunner);
+
+      const cleaner = new NotificationCleaner(
+        mockConfig,
+        mockScheduler,
+        mockLogger,
+        mockDatabase,
+      );
+      await cleaner.initTaskRunner();
+
+      expect(mockLogger.info).toHaveBeenCalledWith(
+        'Starting notification cleaner task',
+      );
+      expect(mockLogger.info).toHaveBeenCalledWith(
+        'Notification cleaner task completed successfully, deleted 1 notifications',
+      );
+      expect(mockDatabase.clearNotifications).toHaveBeenCalledWith({
+        maxAge: { years: 1 },
+      });
+    });
+  });
+});

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

@@ -0,0 +1,91 @@
+/*
+ * Copyright 2025 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 {
+  LoggerService,
+  SchedulerService,
+  SchedulerServiceTaskScheduleDefinition,
+} from '@backstage/backend-plugin-api';
+import { Config, readDurationFromConfig } from '@backstage/config';
+import { NotificationsStore } from '../database';
+import { HumanDuration } from '@backstage/types';
+import { ForwardedError } from '@backstage/errors';
+
+export class NotificationCleaner {
+  private readonly retention: HumanDuration = { years: 1 };
+  private readonly enabled: boolean = true;
+
+  constructor(
+    config: Config,
+    private readonly scheduler: SchedulerService,
+    private readonly logger: LoggerService,
+    private readonly database: NotificationsStore,
+  ) {
+    if (config.has('notifications.retention')) {
+      const retentionConfig = config.get('notifications.retention');
+      if (typeof retentionConfig === 'boolean' && !retentionConfig) {
+        logger.info(
+          'Notification retention is disabled, skipping notification cleaner task',
+        );
+        this.enabled = false;
+        return;
+      }
+      this.retention = readDurationFromConfig(config, {
+        key: 'notifications.retention',
+      });
+    }
+  }
+
+  async initTaskRunner() {
+    if (!this.enabled) {
+      return;
+    }
+
+    const schedule: SchedulerServiceTaskScheduleDefinition = {
+      frequency: { cron: '0 0 * * *' },
+      timeout: { hours: 1 },
+      initialDelay: { hours: 1 },
+      scope: 'global',
+    };
+
+    const taskRunner = this.scheduler.createScheduledTaskRunner(schedule);
+    await taskRunner.run({
+      id: 'notification-cleaner',
+      fn: async () => {
+        await this.clearNotifications(
+          this.logger,
+          this.database,
+          this.retention,
+        );
+      },
+    });
+  }
+
+  private async clearNotifications(
+    logger: LoggerService,
+    database: NotificationsStore,
+    retention: HumanDuration,
+  ) {
+    logger.info('Starting notification cleaner task');
+    try {
+      const result = await database.clearNotifications({ maxAge: retention });
+      logger.info(
+        `Notification cleaner task completed successfully, deleted ${result.deletedCount} notifications`,
+      );
+    } catch (error) {
+      throw new ForwardedError('Notification cleaner task failed', error);
+    }
+  }
+}

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

@@ -29,8 +29,10 @@ import { NotificationSendOptions } from '@backstage/plugin-notifications-node';
 import { catalogServiceMock } from '@backstage/plugin-catalog-node/testUtils';
 import { DatabaseService } from '@backstage/backend-plugin-api';
 import { v4 as uuid } from 'uuid';
+import { DatabaseNotificationsStore } from '../database';
 
 const databases = TestDatabases.create();
+let store: DatabaseNotificationsStore;
 
 async function createDatabase(
   databaseId: TestDatabaseId,
@@ -83,6 +85,9 @@ describe.each(databases.eachSupportedId())('createRouter (%s)', databaseId => {
 
   beforeAll(async () => {
     database = await createDatabase(databaseId);
+    store = await DatabaseNotificationsStore.create({
+      database,
+    });
   });
 
   describe('POST /notifications', () => {
@@ -93,7 +98,7 @@ describe.each(databases.eachSupportedId())('createRouter (%s)', databaseId => {
     beforeAll(async () => {
       const router = await createRouter({
         logger: mockServices.logger.mock(),
-        database,
+        store,
         signals: signalService,
         userInfo,
         config,
@@ -460,7 +465,7 @@ describe.each(databases.eachSupportedId())('createRouter (%s)', databaseId => {
     beforeAll(async () => {
       const router = await createRouter({
         logger: mockServices.logger.mock(),
-        database,
+        store,
         signals: signalService,
         userInfo,
         config,
@@ -550,7 +555,7 @@ describe.each(databases.eachSupportedId())('createRouter (%s)', databaseId => {
     beforeAll(async () => {
       const router = await createRouter({
         logger: mockServices.logger.mock(),
-        database,
+        store,
         signals: signalService,
         userInfo,
         config,
@@ -600,7 +605,7 @@ describe.each(databases.eachSupportedId())('createRouter (%s)', databaseId => {
     beforeAll(async () => {
       const router = await createRouter({
         logger: mockServices.logger.mock(),
-        database,
+        store,
         signals: signalService,
         userInfo,
         config,

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

@@ -17,9 +17,9 @@
 import express, { Request, Response } from 'express';
 import Router from 'express-promise-router';
 import {
-  DatabaseNotificationsStore,
   normalizeSeverity,
   NotificationGetOptions,
+  NotificationsStore,
   TopicGetOptions,
 } from '../database';
 import { v4 as uuid } from 'uuid';
@@ -31,7 +31,6 @@ import {
 import { InputError, NotFoundError } from '@backstage/errors';
 import {
   AuthService,
-  DatabaseService,
   HttpAuthService,
   LoggerService,
   UserInfoService,
@@ -58,7 +57,7 @@ import pThrottle from 'p-throttle';
 export interface RouterOptions {
   logger: LoggerService;
   config: Config;
-  database: DatabaseService;
+  store: NotificationsStore;
   auth: AuthService;
   httpAuth: HttpAuthService;
   userInfo: UserInfoService;
@@ -74,7 +73,7 @@ export async function createRouter(
   const {
     config,
     logger,
-    database,
+    store,
     auth,
     httpAuth,
     userInfo,
@@ -84,7 +83,6 @@ export async function createRouter(
   } = options;
 
   const WEB_NOTIFICATION_CHANNEL = 'Web';
-  const store = await DatabaseNotificationsStore.create({ database });
   const frontendBaseUrl = config.getString('app.baseUrl');
   const concurrencyLimit =
     config.getOptionalNumber('notifications.concurrencyLimit') ?? 10;