feat: Add notification support

server/package.json

@@ -36,9 +36,12 @@
     "axios": "^1.7.9",
     "chalk": "^4.1.2",
     "cron-validator": "^1.3.1",
+    "email-templates": "9.0.0",
     "lodash": "^4.17.21",
     "nest-winston": "^1.10.0",
     "node-cache": "^5.1.2",
+    "nodemailer": "6.9.1",
+    "openpgp": "^5.7.0",
     "plex-api": "^5.3.2",
     "reflect-metadata": "^0.2.2",
     "rxjs": "^7.8.1",
@@ -57,6 +60,7 @@
     "@nestjs/cli": "^10.4.9",
     "@nestjs/schematics": "^10.2.3",
     "@nestjs/testing": "^10.4.15",
+    "@types/email-templates": "^10.0.4",
     "@types/jest": "^29.5.14",
     "@types/lodash": "^4.17.14",
     "@types/node": "^22",

server/src/app/app.module.ts

@@ -16,6 +16,8 @@ import { OverseerrApiService } from '../modules/api/overseerr-api/overseerr-api.
 import ormConfig from './config/typeOrmConfig';
 import { TautulliApiModule } from '../modules/api/tautulli-api/tautulli-api.module';
 import { TautulliApiService } from '../modules/api/tautulli-api/tautulli-api.service';
+import { NotificationsModule } from '../modules/notifications/notifications.module';
+import { NotificationService } from '../modules/notifications/notifications.service';
 
 @Module({
   imports: [
@@ -29,6 +31,7 @@ import { TautulliApiService } from '../modules/api/tautulli-api/tautulli-api.ser
     TautulliApiModule,
     RulesModule,
     CollectionsModule,
+    NotificationsModule,
   ],
   controllers: [AppController],
   providers: [AppService],
@@ -39,12 +42,16 @@ export class AppModule implements OnModuleInit {
     private readonly plexApi: PlexApiService,
     private readonly overseerApi: OverseerrApiService,
     private readonly tautulliApi: TautulliApiService,
+    private readonly notificationService: NotificationService,
   ) {}
   async onModuleInit() {
-    // Initialize stuff needing settings here.. Otherwise problems
+    // Initialize modules requiring settings
     await this.settings.init();
     await this.plexApi.initialize({});
     await this.overseerApi.init();
     await this.tautulliApi.init();
+
+    // intialize notification agents
+    await this.notificationService.registerConfiguredAgents();
   }
 }

server/src/database/migrations/1727693832830-Add_Notification_settings.ts

@@ -0,0 +1,38 @@
+import { MigrationInterface, QueryRunner } from 'typeorm';
+
+export class AddNotificationSettings1727693832830
+  implements MigrationInterface
+{
+  name = 'AddNotificationSettings1727693832830';
+
+  public async up(queryRunner: QueryRunner): Promise<void> {
+    // Create notification table
+    await queryRunner.query(`
+      CREATE TABLE "notification" (
+        "id" INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,
+        "name" VARCHAR NOT NULL,
+        "agent" VARCHAR NOT NULL,
+        "enabled" BOOLEAN DEFAULT false,
+        "types" TEXT,
+        "options" TEXT NOT NULL
+      );
+    `);
+
+    // Create notification_rulegroup table with foreign key constraints directly
+    await queryRunner.query(`
+      CREATE TABLE "notification_rulegroup" (
+        "notificationId" INTEGER NOT NULL,
+        "rulegroupId" INTEGER NOT NULL,
+        PRIMARY KEY ("notificationId", "rulegroupId"),
+        FOREIGN KEY ("notificationId") REFERENCES "notification"("id") ON DELETE CASCADE,
+        FOREIGN KEY ("rulegroupId") REFERENCES "rule_group"("id") ON DELETE CASCADE
+      );
+    `);
+  }
+
+  public async down(queryRunner: QueryRunner): Promise<void> {
+    // Drop the tables in reverse order
+    await queryRunner.query(`DROP TABLE "notification_rulegroup"`);
+    await queryRunner.query(`DROP TABLE "notification"`);
+  }
+}

server/src/database/migrations/1732008945000-Notification_settings_aboutScale.ts

@@ -0,0 +1,19 @@
+import { MigrationInterface, QueryRunner } from 'typeorm';
+
+export class NotificationSettingsAboutScale1732008945000
+  implements MigrationInterface
+{
+  name = 'NotificationSettingsAboutScale1732008945000';
+
+  public async up(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query(`
+      ALTER TABLE "notification" ADD COLUMN "aboutScale" INTEGER NOT NULL DEFAULT 3;
+    `);
+  }
+
+  public async down(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query(`
+      ALTER TABLE "notification" DROP COLUMN "aboutScale";
+    `);
+  }
+}

server/src/modules/collections/collection-worker.service.ts

@@ -15,6 +15,8 @@ import { PlexMetadata } from '../api/plex-api/interfaces/media.interface';
 import { EPlexDataType } from '../api/plex-api/enums/plex-data-type-enum';
 import { TmdbIdService } from '../api/tmdb-api/tmdb-id.service';
 import { TaskBase } from '../tasks/task.base';
+import { NotificationService } from '../notifications/notifications.service';
+import { NotificationType } from '../notifications/notifications-interfaces';
 
 @Injectable()
 export class CollectionWorkerService extends TaskBase {
@@ -37,6 +39,7 @@ export class CollectionWorkerService extends TaskBase {
     private readonly settings: SettingsService,
     private readonly tmdbIdService: TmdbIdService,
     private readonly tmdbIdHelper: TmdbIdService,
+    private readonly notificationService: NotificationService,
   ) {
     super(taskService);
   }
@@ -46,76 +49,105 @@ export class CollectionWorkerService extends TaskBase {
   }
 
   public async execute() {
-    // check if another instance of this task is already running
-    if (await this.isRunning()) {
-      this.logger.log(
-        `Another instance of the ${this.name} task is currently running. Skipping this execution`,
-      );
-      return;
-    }
-
-    await super.execute();
+    try {
+      // check if another instance of this task is already running
+      if (await this.isRunning()) {
+        this.logger.log(
+          `Another instance of the ${this.name} task is currently running. Skipping this execution`,
+        );
+        return;
+      }
 
-    // wait 5 seconds to make sure we're not executing together with the rule handler
-    await new Promise((resolve) => setTimeout(resolve, 5000));
+      await this.notificationService.registerConfiguredAgents(true); // re-register notification agents, to avoid flukes
+      await super.execute();
 
-    // if we are, then wait..
-    await this.taskService.waitUntilTaskIsFinished('Rule Handler', this.name);
+      // wait 5 seconds to make sure we're not executing together with the rule handler
+      await new Promise((resolve) => setTimeout(resolve, 5000));
 
-    // Start actual task
-    const appStatus = await this.settings.testConnections();
+      // if we are, then wait..
+      await this.taskService.waitUntilTaskIsFinished('Rule Handler', this.name);
 
-    this.logger.log('Start handling all collections');
-    let handledCollections = 0;
-    if (appStatus) {
-      // loop over all active collections
-      const collections = await this.collectionRepo.find({
-        where: { isActive: true },
-      });
-      for (const collection of collections) {
-        this.infoLogger(`Handling collection '${collection.title}'`);
+      // Start actual task
+      const appStatus = await this.settings.testConnections();
 
-        const collectionMedia = await this.collectionMediaRepo.find({
-          where: {
-            collectionId: collection.id,
-          },
+      this.logger.log('Start handling all collections');
+      let handledCollections = 0;
+      if (appStatus) {
+        // loop over all active collections
+        const collections = await this.collectionRepo.find({
+          where: { isActive: true },
         });
+        for (const collection of collections) {
+          this.infoLogger(`Handling collection '${collection.title}'`);
 
-        const dangerDate = new Date(
-          new Date().getTime() - +collection.deleteAfterDays * 86400000,
-        );
+          const collectionMedia = await this.collectionMediaRepo.find({
+            where: {
+              collectionId: collection.id,
+            },
+          });
 
-        for (const media of collectionMedia) {
-          // handle media addate <= due date
-          if (new Date(media.addDate) <= dangerDate) {
-            await this.handleMedia(collection, media);
-            handledCollections++;
+          const dangerDate = new Date(
+            new Date().getTime() - +collection.deleteAfterDays * 86400000,
+          );
+
+          const handledMediaForNotification = [];
+          for (const media of collectionMedia) {
+            // handle media addate <= due date
+            if (new Date(media.addDate) <= dangerDate) {
+              await this.handleMedia(collection, media);
+              handledCollections++;
+              handledMediaForNotification.push({ plexId: media.plexId });
+            }
           }
-        }
 
-        this.infoLogger(`Handling collection '${collection.title}' finished`);
-      }
-      if (handledCollections > 0) {
-        if (this.settings.overseerrConfigured()) {
-          setTimeout(() => {
-            this.overseerrApi.api
-              .post('/settings/jobs/availability-sync/run')
-              .then(() => {
-                this.infoLogger(
-                  `All collections handled. Triggered Overseerr's availability-sync because media was altered`,
-                );
-              });
-          }, 7000);
+          // handle notification
+          if (handledMediaForNotification.length > 0) {
+            await this.notificationService.handleNotification(
+              NotificationType.MEDIA_HANDLED,
+              handledMediaForNotification,
+              collection.title,
+            );
+          }
+
+          this.infoLogger(`Handling collection '${collection.title}' finished`);
+        }
+        if (handledCollections > 0) {
+          if (this.settings.overseerrConfigured()) {
+            setTimeout(() => {
+              this.overseerrApi.api
+                .post('/settings/jobs/availability-sync/run')
+                .then(() => {
+                  this.infoLogger(
+                    `All collections handled. Triggered Overseerr's availability-sync because media was altered`,
+                  );
+                });
+            }, 7000);
+          }
+        } else {
+          this.infoLogger(`All collections handled. No data was altered`);
         }
       } else {
-        this.infoLogger(`All collections handled. No data was altered`);
+        this.infoLogger(
+          'Not all applications are reachable.. Skipping collection handling',
+        );
+
+        // notify
+        await this.notificationService.handleNotification(
+          NotificationType.COLLECTION_HANDLING_FAILED,
+          undefined,
+        );
       }
-    } else {
-      this.infoLogger(
-        'Not all applications are reachable.. Skipping collection handling',
+      await this.finish();
+    } catch (e) {
+      this.logger.error('An error occurred where handling collections');
+      this.logger.debug(e);
+
+      // notify
+      await this.notificationService.handleNotification(
+        NotificationType.COLLECTION_HANDLING_FAILED,
+        undefined,
       );
     }
-    await this.finish();
   }
 
   private async handleMedia(collection: Collection, media: CollectionMedia) {

server/src/modules/collections/collections.module.ts

@@ -15,6 +15,8 @@ import { Exclusion } from '../rules/entities/exclusion.entities';
 import { CollectionLog } from '../collections/entities/collection_log.entities';
 import { CollectionLogCleanerService } from '../collections/tasks/collection-log-cleaner.service';
 import { TautulliApiModule } from '../api/tautulli-api/tautulli-api.module';
+import { Notification } from '../notifications/entities/notification.entities';
+import { NotificationService } from '../notifications/notifications.service';
 
 @Module({
   imports: [
@@ -25,6 +27,7 @@ import { TautulliApiModule } from '../api/tautulli-api/tautulli-api.module';
       CollectionLog,
       RuleGroup,
       Exclusion,
+      Notification,
     ]),
     OverseerrApiModule,
     TautulliApiModule,
@@ -36,6 +39,7 @@ import { TautulliApiModule } from '../api/tautulli-api/tautulli-api.module';
     CollectionsService,
     CollectionWorkerService,
     CollectionLogCleanerService,
+    NotificationService,
   ],
   controllers: [CollectionsController],
   exports: [CollectionsService],

server/src/modules/collections/collections.service.ts

@@ -25,6 +25,7 @@ import { ICollection } from './interfaces/collection.interface';
 import { Exclusion } from '../rules/entities/exclusion.entities';
 import { CollectionLog } from '../../modules/collections/entities/collection_log.entities';
 import { ECollectionLogType } from '../../modules/collections/entities/collection_log.entities';
+import { NotificationService } from '../notifications/notifications.service';
 
 interface addCollectionDbResponse {
   id: number;
@@ -54,6 +55,7 @@ export class CollectionsService {
     private readonly plexApi: PlexApiService,
     private readonly tmdbApi: TmdbApiService,
     private readonly tmdbIdHelper: TmdbIdService,
+    private readonly notificationService: NotificationService,
   ) {}
 
   async getCollection(id?: number, title?: string) {

server/src/modules/notifications/agents/agent.ts

@@ -0,0 +1,24 @@
+import { Notification } from '../entities/notification.entities';
+import {
+  NotificationAgentConfig,
+  NotificationAgentKey,
+  NotificationType,
+} from '../notifications-interfaces';
+
+export interface NotificationPayload {
+  event?: string;
+  subject: string;
+  notifySystem: boolean;
+  image?: string;
+  message?: string;
+  extra?: { name: string; value: string }[];
+}
+
+export interface NotificationAgent {
+  notification: Notification;
+  shouldSend(): boolean;
+  send(type: NotificationType, payload: NotificationPayload): Promise<boolean>;
+  getIdentifier(): NotificationAgentKey;
+  getSettings(): NotificationAgentConfig;
+  getNotification(): Notification;
+}