feat: Enable selection of all CI pipelines at once when the Environment filter is applied in Notifications

src/app.ts

@@ -0,0 +1,32 @@
+/*
+ * Copyright (c) 2024. Devtron Inc.
+ *
+ * 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 express from 'express';
+import bodyParser from 'body-parser';
+import { metricsMiddleware } from './middleware/metrics';
+
+export const createApp = () => {
+    const app = express();
+
+    // Middleware
+    app.use(bodyParser.json({ limit: '10mb' }));
+    app.use(express.json());
+    app.use(metricsMiddleware);
+
+    // Routes
+
+    return app;
+};

src/config/database.ts

@@ -0,0 +1,66 @@
+/*
+ * Copyright (c) 2024. Devtron Inc.
+ *
+ * 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 { ConnectionOptions, createConnection } from "typeorm";
+import { NotificationSettings } from "../entities/notificationSettings";
+import { NotifierEventLog } from "../entities/notifierEventLogs";
+import { Event } from "../notification/service/notificationService";
+import { NotificationTemplates } from "../entities/notificationTemplates";
+import { SlackConfig } from "../entities/slackConfig";
+import { SesConfig } from "../entities/sesConfig";
+import { SMTPConfig } from "../entities/smtpConfig";
+import { WebhookConfig } from "../entities/webhookconfig";
+import { Users } from "../entities/users";
+import { logger } from "./logger";
+import * as process from "process";
+
+export const connectToDatabase = async () => {
+    const dbHost: string = process.env.DB_HOST;
+    const dbPort: number = +process.env.DB_PORT;
+    const user: string = process.env.DB_USER;
+    const pwd: string = process.env.DB_PWD;
+    const db: string = process.env.DB;
+
+    const dbOptions: ConnectionOptions = {
+        type: "postgres",
+        host: dbHost,
+        port: dbPort,
+        username: user,
+        password: pwd,
+        database: db,
+        entities: [
+            NotificationSettings, 
+            NotifierEventLog, 
+            Event, 
+            NotificationTemplates, 
+            SlackConfig, 
+            SesConfig, 
+            SMTPConfig, 
+            WebhookConfig, 
+            Users
+        ]
+    };
+
+    try {
+        const connection = await createConnection(dbOptions);
+        logger.info("Connected to DB");
+        return connection;
+    } catch (error) {
+        logger.error("TypeORM connection error: ", error);
+        logger.error("shutting down notifier due to un-successful database connection...");
+        process.exit(1);
+    }
+};

src/config/logger.ts

@@ -0,0 +1,28 @@
+/*
+ * Copyright (c) 2024. Devtron Inc.
+ *
+ * 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 * as winston from 'winston';
+
+export const logger = winston.createLogger({
+    level: 'info',
+    format: winston.format.combine(
+        winston.format.timestamp(),
+        winston.format.printf(info => {
+            return `${info.timestamp} ${info.level}: ${info.message}`;
+        })
+    ),
+    transports: [new winston.transports.Console()]
+});

src/config/nats.ts

@@ -0,0 +1,80 @@
+/*
+ * Copyright (c) 2024. Devtron Inc.
+ *
+ * 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 { connect, NatsConnection } from "nats";
+import { logger } from "./logger";
+import { PubSubServiceImpl } from "../pubSub/pubSub";
+import { NOTIFICATION_EVENT_TOPIC } from "../pubSub/utils";
+import { Event } from "../notification/service/notificationService";
+import { NotificationService } from "../notification/service/notificationService";
+import { successNotificationMetricsCounter, failedNotificationMetricsCounter } from '../common/metrics';
+
+export const natsEventHandler = (notificationService: NotificationService) => async (msg: string) => {
+    const eventAsString = JSON.parse(msg);
+    const parsedData = JSON.parse(eventAsString);
+    let response;
+
+    try {
+        // First try to parse as V2 format (which includes both event and notificationSettings)
+        logger.info('Attempting to parse as V2 payload');
+        if (parsedData.event && parsedData.notificationSettings) {
+            // This is a V2 payload
+            const { event, notificationSettings } = parsedData;
+            logger.info({ natsEventBodyV2: { event, notificationSettingsCount: notificationSettings.length } });
+            response = await notificationService.sendNotificationV2(event, notificationSettings);
+        } else {
+            // Fall back to V1 format (which only includes the event)
+            logger.info('Falling back to V1 payload format');
+            const event = parsedData as Event;
+            logger.info({ natsEventBodyV1: event });
+            response = await notificationService.sendNotification(event);
+        }
+    } catch (error: any) {
+         {
+            logger.error(`Failed to process message in both V1 and V2 formats: ${error.message || 'Unknown error'}`);
+            failedNotificationMetricsCounter.inc();
+            throw error;
+        }
+    }
+
+    // Handle response metrics
+    if (response.status != 0) {
+        successNotificationMetricsCounter.inc();
+    } else {
+        failedNotificationMetricsCounter.inc();
+    }
+};
+
+export const connectToNats = async (notificationService: NotificationService) => {
+    const natsUrl = process.env.NATS_URL;
+
+    if (!natsUrl) {
+        logger.info("NATS_URL not provided, skipping NATS connection");
+        return;
+    }
+
+    try {
+        logger.info("Connecting to NATS server...");
+        const conn: NatsConnection = await connect({ servers: natsUrl });
+        const jsm = await conn.jetstreamManager();
+        const pubSubService = new PubSubServiceImpl(conn, jsm, logger);
+        await pubSubService.Subscribe(NOTIFICATION_EVENT_TOPIC, natsEventHandler(notificationService));
+        logger.info("Successfully connected to NATS");
+        return conn;
+    } catch (err) {
+        logger.error("Error connecting to NATS:", err);
+    }
+};

src/middleware/metrics.ts

@@ -0,0 +1,27 @@
+/*
+ * Copyright (c) 2024. Devtron Inc.
+ *
+ * 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 { Request, Response, NextFunction } from 'express';
+import { httpRequestMetricsCounter } from '../common/metrics';
+
+export const metricsMiddleware = (req: Request, res: Response, next: NextFunction) => {
+    httpRequestMetricsCounter.labels({
+        method: req.method, 
+        endpoint: req.url, 
+        statusCode: res.statusCode
+    }).inc();
+    next();
+};