diff --git a/server/src/config/services.shared.ts b/server/src/config/services.shared.ts index 49b73f52e..72fe3fcf4 100644 --- a/server/src/config/services.shared.ts +++ b/server/src/config/services.shared.ts @@ -21,6 +21,7 @@ import { GlobalPingService } from "@/service/globalPingService.js"; import { ILogger } from "@/utils/logger.js"; // Notification providers +import type { NotificationProviderRegistry } from "@/domain/notifications/notification.service.js"; import { WebhookProvider } from "@/domain/notifications/providers/webhook.js"; import { SlackProvider } from "@/domain/notifications/providers/slack.js"; import { EmailProvider } from "@/domain/notifications/providers/email.js"; @@ -167,24 +168,28 @@ export const buildShared = async ({ const twilioProvider = new TwilioProvider(logger); const ntfyProvider = new NtfyProvider(logger); - const notificationsService = new NotificationsService( + const notificationProviders: NotificationProviderRegistry = { + webhook: webhookProvider, + email: emailProvider, + slack: slackProvider, + discord: discordProvider, + pager_duty: pagerDutyProvider, + matrix: matrixProvider, + teams: teamsProvider, + telegram: telegramProvider, + pushover: pushoverProvider, + twilio: twilioProvider, + ntfy: ntfyProvider, + }; + + const notificationsService = new NotificationsService({ notificationsRepository, monitorsRepository, - webhookProvider, - emailProvider, - slackProvider, - discordProvider, - pagerDutyProvider, - matrixProvider, - teamsProvider, - telegramProvider, - pushoverProvider, - twilioProvider, - ntfyProvider, + providers: notificationProviders, settingsService, logger, - notificationMessageBuilder - ); + notificationMessageBuilder, + }); const sharedServices: SharedServices = { logger, diff --git a/server/src/config/services.worker.ts b/server/src/config/services.worker.ts index 44997af5f..b0bdff2d6 100644 --- a/server/src/config/services.worker.ts +++ b/server/src/config/services.worker.ts @@ -131,9 +131,9 @@ export const buildWorker = async (shared: SharedServices, envSettings: EnvConfig geoChecksRepository ); - const worker = await DBQueueWorker.create( + const worker = await DBQueueWorker.create({ logger, - () => mongoose.connection.readyState === 1, + isDbConnected: () => mongoose.connection.readyState === 1, jobsRepository, monitorsRepository, checksRepository, @@ -142,13 +142,13 @@ export const buildWorker = async (shared: SharedServices, envSettings: EnvConfig checkProducer, checkEvaluator, geoCheckPipeline, - reactorDispatcher, - workerHelper, + dispatcher: reactorDispatcher, + helper: workerHelper, queueWorkersRepository, - envSettings.queueMode, - envSettings.queuePrimaryProcesses, - workerId - ); + queueMode: envSettings.queueMode, + queuePrimaryProcesses: envSettings.queuePrimaryProcesses, + workerId, + }); return { worker, networkService, bufferService, statusService }; }; diff --git a/server/src/domain/notifications/notification.service.ts b/server/src/domain/notifications/notification.service.ts index 7fc6e3543..30141c3ea 100644 --- a/server/src/domain/notifications/notification.service.ts +++ b/server/src/domain/notifications/notification.service.ts @@ -9,6 +9,9 @@ import type { MonitorActionDecision } from "@/worker/worker.helper.js"; import type { ISettingsService } from "@/domain/app-settings/app-settings.service.js"; import { ILogger } from "@/utils/logger.js"; import type { INotificationMessageBuilder } from "@/domain/notifications/notification.message-builder.js"; +import type { NotificationChannel } from "@/domain/notifications/notification.type.js"; + +export type NotificationProviderRegistry = Record; export interface INotificationsService { createNotification: (notificationData: Partial, userId: string, teamId: string) => Promise; @@ -29,52 +32,29 @@ export class NotificationsService implements INotificationsService { private notificationsRepository: INotificationsRepository; private monitorsRepository: IMonitorsRepository; - private webhookProvider: INotificationProvider; - private emailProvider: INotificationProvider; - private slackProvider: INotificationProvider; - private discordProvider: INotificationProvider; - private pagerDutyProvider: INotificationProvider; - private matrixProvider: INotificationProvider; - private teamsProvider: INotificationProvider; - private telegramProvider: INotificationProvider; - private pushoverProvider: INotificationProvider; - private twilioProvider: INotificationProvider; - private ntfyProvider: INotificationProvider; - private logger: ILogger; + private providers: NotificationProviderRegistry; private settingsService: ISettingsService; + private logger: ILogger; private notificationMessageBuilder: INotificationMessageBuilder; - constructor( - notificationsRepository: INotificationsRepository, - monitorsRepository: IMonitorsRepository, - webhookProvider: INotificationProvider, - emailProvider: INotificationProvider, - slackProvider: INotificationProvider, - discordProvider: INotificationProvider, - pagerDutyProvider: INotificationProvider, - matrixProvider: INotificationProvider, - teamsProvider: INotificationProvider, - telegramProvider: INotificationProvider, - pushoverProvider: INotificationProvider, - twilioProvider: INotificationProvider, - ntfyProvider: INotificationProvider, - settingsService: ISettingsService, - logger: ILogger, - notificationMessageBuilder: INotificationMessageBuilder - ) { + constructor({ + notificationsRepository, + monitorsRepository, + providers, + settingsService, + logger, + notificationMessageBuilder, + }: { + notificationsRepository: INotificationsRepository; + monitorsRepository: IMonitorsRepository; + providers: NotificationProviderRegistry; + settingsService: ISettingsService; + logger: ILogger; + notificationMessageBuilder: INotificationMessageBuilder; + }) { this.notificationsRepository = notificationsRepository; this.monitorsRepository = monitorsRepository; - this.webhookProvider = webhookProvider; - this.emailProvider = emailProvider; - this.slackProvider = slackProvider; - this.discordProvider = discordProvider; - this.pagerDutyProvider = pagerDutyProvider; - this.matrixProvider = matrixProvider; - this.teamsProvider = teamsProvider; - this.telegramProvider = telegramProvider; - this.pushoverProvider = pushoverProvider; - this.twilioProvider = twilioProvider; - this.ntfyProvider = ntfyProvider; + this.providers = providers; this.settingsService = settingsService; this.logger = logger; this.notificationMessageBuilder = notificationMessageBuilder; @@ -97,37 +77,16 @@ export class NotificationsService implements INotificationsService { } // Route to provider based on notification type - switch (notification.type) { - case "webhook": - return await this.webhookProvider.sendMessage!(notification, notificationMessage); - case "slack": - return await this.slackProvider.sendMessage!(notification, notificationMessage); - case "matrix": - return await this.matrixProvider.sendMessage!(notification, notificationMessage); - case "pager_duty": - return await this.pagerDutyProvider.sendMessage!(notification, notificationMessage); - case "discord": - return await this.discordProvider.sendMessage!(notification, notificationMessage); - case "email": - return await this.emailProvider.sendMessage!(notification, notificationMessage); - case "teams": - return await this.teamsProvider.sendMessage!(notification, notificationMessage); - case "telegram": - return await this.telegramProvider.sendMessage!(notification, notificationMessage); - case "pushover": - return await this.pushoverProvider.sendMessage!(notification, notificationMessage); - case "twilio": - return await this.twilioProvider.sendMessage!(notification, notificationMessage); - case "ntfy": - return await this.ntfyProvider.sendMessage!(notification, notificationMessage); - default: - this.logger.warn({ - message: `Unknown notification type: ${notification.type}`, - service: SERVICE_NAME, - method: "send", - }); - return false; + const provider = this.providers[notification.type]; + if (!provider) { + this.logger.warn({ + message: `Unknown notification type: ${notification.type}`, + service: SERVICE_NAME, + method: "send", + }); + return false; } + return await provider.sendMessage(notification, notificationMessage); }; private sendNotifications = async (monitor: Monitor, monitorStatusResponse: MonitorStatusResponse, decision: MonitorActionDecision) => { @@ -165,32 +124,26 @@ export class NotificationsService implements INotificationsService { }; sendTestNotification = async (notification: Partial) => { - switch (notification.type) { - case "email": - return await this.emailProvider.sendTestAlert(notification); - case "slack": - return await this.slackProvider.sendTestAlert(notification); - case "discord": - return await this.discordProvider.sendTestAlert(notification); - case "pager_duty": - return await this.pagerDutyProvider.sendTestAlert(notification); - case "matrix": - return await this.matrixProvider.sendTestAlert(notification); - case "webhook": - return await this.webhookProvider.sendTestAlert(notification); - case "teams": - return await this.teamsProvider.sendTestAlert(notification); - case "telegram": - return await this.telegramProvider.sendTestAlert(notification); - case "pushover": - return await this.pushoverProvider.sendTestAlert(notification); - case "twilio": - return await this.twilioProvider.sendTestAlert(notification); - case "ntfy": - return await this.ntfyProvider.sendTestAlert(notification); - default: - return false; + const type = notification.type; + if (!type) { + this.logger.warn({ + message: "Notification type not provided", + service: SERVICE_NAME, + method: "sendTestNotification", + }); + return false; } + + const provider = this.providers[type]; + if (!provider) { + this.logger.warn({ + message: `Unknown notification type: ${notification.type}`, + service: SERVICE_NAME, + method: "sendTestNotification", + }); + return false; + } + return await provider.sendTestAlert(notification); }; testAllNotifications = async (notificationIds: string[]) => { diff --git a/server/src/worker/worker.db-queue.ts b/server/src/worker/worker.db-queue.ts index 3ea2fddb8..50e41a4f1 100644 --- a/server/src/worker/worker.db-queue.ts +++ b/server/src/worker/worker.db-queue.ts @@ -31,6 +31,25 @@ const CONCURRENCY: Record = { "cleanup-retention": 1, }; +export interface DBQueueWorkerDependencies { + logger: ILogger; // forwarded to super + isDbConnected: () => boolean; + jobsRepository: IJobsRepository; // forwarded + monitorsRepository: IMonitorsRepository; // forwarded + checksRepository: IChecksRepository; + checkService: ICheckService; + bufferService: IBufferService; + checkProducer: ICheckProducer; + checkEvaluator: ICheckEvaluator; + geoCheckPipeline: ICheckPipeline; + dispatcher: IReactorDispatcher; + helper: IWorkerHelper; + queueWorkersRepository: IQueueWorkersRepository; // forwarded + queueMode: QueueMode; // forwarded + queuePrimaryProcesses: boolean; // consumed immediately + workerId: string; +} + export class DBQueueWorker extends JobScheduler implements IQueueWorker { static SERVICE_NAME = SERVICE_NAME; @@ -45,68 +64,43 @@ export class DBQueueWorker extends JobScheduler implements IQueueWorker { "cleanup-retention": 0, }; - constructor( - logger: ILogger, // forwarded to super - private isDbConnected: () => boolean, - jobsRepository: IJobsRepository, // forwarded - monitorsRepository: IMonitorsRepository, // forwarded - private checksRepository: IChecksRepository, - private checkService: ICheckService, - private bufferService: IBufferService, - private checkProducer: ICheckProducer, - private checkEvaluator: ICheckEvaluator, - private geoCheckPipeline: ICheckPipeline, - private dispatcher: IReactorDispatcher, - private helper: IWorkerHelper, - queueWorkersRepository: IQueueWorkersRepository, // forwarded - queueMode: QueueMode, // forwarded - queuePrimaryProcesses: boolean, // consumed immediately - workerId: string - ) { - super(jobsRepository, queueWorkersRepository, monitorsRepository, queueMode, logger, workerId); - this.processesJobs = queuePrimaryProcesses === true || queueMode === "worker"; + private checksRepository: IChecksRepository; + private checkProducer: ICheckProducer; + private checkService: ICheckService; + private checkEvaluator: ICheckEvaluator; + private dispatcher: IReactorDispatcher; + private geoCheckPipeline: ICheckPipeline; + private helper: IWorkerHelper; + private bufferService: IBufferService; + private isDbConnected: () => boolean; + + constructor(dependencies: DBQueueWorkerDependencies) { + super( + dependencies.jobsRepository, + dependencies.queueWorkersRepository, + dependencies.monitorsRepository, + dependencies.queueMode, + dependencies.logger, + dependencies.workerId + ); + this.processesJobs = dependencies.queuePrimaryProcesses === true || dependencies.queueMode === "worker"; + this.checksRepository = dependencies.checksRepository; + this.checkProducer = dependencies.checkProducer; + this.checkService = dependencies.checkService; + this.checkEvaluator = dependencies.checkEvaluator; + this.dispatcher = dependencies.dispatcher; + this.geoCheckPipeline = dependencies.geoCheckPipeline; + this.helper = dependencies.helper; + this.bufferService = dependencies.bufferService; + this.isDbConnected = dependencies.isDbConnected; } get serviceName() { return DBQueueWorker.SERVICE_NAME; } - static async create( - logger: ILogger, - isDbConnected: () => boolean, - jobsRepository: IJobsRepository, - monitorsRepository: IMonitorsRepository, - checksRepository: IChecksRepository, - checkService: ICheckService, - bufferService: IBufferService, - checkProducer: ICheckProducer, - checkEvaluator: ICheckEvaluator, - geoCheckPipeline: ICheckPipeline, - dispatcher: IReactorDispatcher, - helper: IWorkerHelper, - queueWorkersRepository: IQueueWorkersRepository, - queueMode: QueueMode, - queuePrimaryProcesses: boolean, - workerId: string - ): Promise { - const instance = new DBQueueWorker( - logger, - isDbConnected, - jobsRepository, - monitorsRepository, - checksRepository, - checkService, - bufferService, - checkProducer, - checkEvaluator, - geoCheckPipeline, - dispatcher, - helper, - queueWorkersRepository, - queueMode, - queuePrimaryProcesses, - workerId - ); + static async create(dependencies: DBQueueWorkerDependencies): Promise { + const instance = new DBQueueWorker(dependencies); await instance.init(); return instance; } diff --git a/server/test/unit/services/dbQueueWorker.test.ts b/server/test/unit/services/dbQueueWorker.test.ts index 68df5f02a..4956e474c 100644 --- a/server/test/unit/services/dbQueueWorker.test.ts +++ b/server/test/unit/services/dbQueueWorker.test.ts @@ -139,24 +139,12 @@ const createWorker = (overrides?: { queueMode?: QueueMode; queuePrimaryProcesses ...overrides?.mocks, }; - const worker = new DBQueueWorker( - mocks.logger as any, - mocks.isDbConnected as any, - mocks.jobsRepository as any, - mocks.monitorsRepository as any, - mocks.checksRepository as any, - mocks.checkService as any, - mocks.bufferService as any, - mocks.checkProducer as any, - mocks.checkEvaluator as any, - mocks.geoCheckPipeline as any, - mocks.dispatcher as any, - mocks.helper as any, - mocks.queueWorkersRepository as any, - overrides?.queueMode ?? "worker", - overrides?.queuePrimaryProcesses ?? true, - "worker-1" - ); + const worker = new DBQueueWorker({ + ...mocks, + queueMode: overrides?.queueMode ?? "worker", + queuePrimaryProcesses: overrides?.queuePrimaryProcesses ?? true, + workerId: "worker-1", + } as any); return { worker, mocks }; }; @@ -243,24 +231,12 @@ describe("DBQueueWorker", () => { it("create() constructs the worker and runs init", async () => { const { mocks } = createWorker(); // reuse the mock set; the un-init'd instance is discarded - const worker = await DBQueueWorker.create( - mocks.logger as any, - mocks.isDbConnected as any, - mocks.jobsRepository as any, - mocks.monitorsRepository as any, - mocks.checksRepository as any, - mocks.checkService as any, - mocks.bufferService as any, - mocks.checkProducer as any, - mocks.checkEvaluator as any, - mocks.geoCheckPipeline as any, - mocks.dispatcher as any, - mocks.helper as any, - mocks.queueWorkersRepository as any, - "worker", - true, - "worker-1" - ); + const worker = await DBQueueWorker.create({ + ...mocks, + queueMode: "worker", + queuePrimaryProcesses: true, + workerId: "worker-1", + } as any); active = worker; await jest.advanceTimersByTimeAsync(1); diff --git a/server/test/unit/services/notificationsService.test.ts b/server/test/unit/services/notificationsService.test.ts index 8ba8b34aa..6fbbfa4a1 100644 --- a/server/test/unit/services/notificationsService.test.ts +++ b/server/test/unit/services/notificationsService.test.ts @@ -73,24 +73,26 @@ const createService = (overrides?: Record) => { ...overrides, }; - const service = new NotificationsService( - defaults.notificationsRepository as any, - defaults.monitorsRepository as any, - defaults.webhookProvider as any, - defaults.emailProvider as any, - defaults.slackProvider as any, - defaults.discordProvider as any, - defaults.pagerDutyProvider as any, - defaults.matrixProvider as any, - defaults.teamsProvider as any, - defaults.telegramProvider as any, - defaults.pushoverProvider as any, - defaults.twilioProvider as any, - defaults.ntfyProvider as any, - defaults.settingsService as any, - defaults.logger as any, - defaults.notificationMessageBuilder as any - ); + const service = new NotificationsService({ + notificationsRepository: defaults.notificationsRepository, + monitorsRepository: defaults.monitorsRepository, + providers: { + webhook: defaults.webhookProvider, + email: defaults.emailProvider, + slack: defaults.slackProvider, + discord: defaults.discordProvider, + pager_duty: defaults.pagerDutyProvider, + matrix: defaults.matrixProvider, + teams: defaults.teamsProvider, + telegram: defaults.telegramProvider, + pushover: defaults.pushoverProvider, + twilio: defaults.twilioProvider, + ntfy: defaults.ntfyProvider, + }, + settingsService: defaults.settingsService, + logger: defaults.logger, + notificationMessageBuilder: defaults.notificationMessageBuilder, + } as any); return { service, ...defaults }; };