# Medics Event Publisher The Medics Event Publisher is a Node.js application designed to handle almost real-time event publishing for the medics. It integrates with Kafka for reliable event processing, MySQL for storing event data, and provides a scalable and robust solution for managing thousands of transactions per second. It's designed to handle high volumes of transactions and ensure data consistency, even when the Kafka service is temporarily unavailable. ## Table of Contents - [Overview](#overview) - [Features](#features) - [Project Structure](#project-structure) - [Installation](#installation) - [Configuration](#configuration) - [Usage](#usage) - [Dockerization](#dockerization) - [Error Handling](#error-handling) - [Concurrency and Performance](#concurrency-and-performance) - [Backpressure Handling](#backpressure-handling) - [Worker Threads](#worker-threads) - [Metrics and Monitoring](#metrics-and-monitoring) - [Dependencies](#dependencies) - [Further Development](#further-development) - [License](#license) ## Overview The Medics Event Publisher is a critical component of the hospital information management system, responsible for real-time event processing and publishing. It ensures the seamless flow of transactions, such as bills, receipts, advances, refunds, lab reports, and goods receipt notes, from the Medics application to downstream systems. ## Features - Real-time event processing and publishing with Kafka. - MySQL integration for storing event data. - Error handling and retry mechanisms for robustness. - Backpressure handling to manage high-throughput scenarios. - Scalable architecture to handle thousands of transactions per second. ## Project Structure The project follows a modular structure for better organization: - `medics-event-publisher.js`: Main entry point for the application. - `src/`: Source code directory. - `config/`: Configuration files. - `config.json`: Configuration settings for Kafka, MySQL, and other services. - `services/`: Reusable service modules. - `KafkaService.js`: Handles Kafka integration, message publishing, and retry logic. - `Logger.js`: Provides a logging service for the application. - `SQLConnectionManager.js`: Manages MySQL database connections. ## Installation 1. Clone the repository: ```bash git clone git@gitlab:medics/medics-event-publisher.git ``` 2. Install dependencies: ```bash cd medics-event-publisher npm install ``` ## Configuration - Update the `config/config.json` file with the appropriate configurations for Kafka, MySQL, and other services. ## Usage 1. Run the application: ```bash node medics-event-publisher.js ``` 2. Monitor health status: Open a web browser and visit `http://localhost:<PORT>/health` to check the health status of the application. ## Error Handling The application handles errors at various levels to ensure robustness. It logs errors, retries message publishing in case of failures, and provides detailed error messages in case of critical issues. ## Concurrency and Performance The use of worker threads enables parallel event processing, improving the application's performance. Adjust the `MAX_WORKERS` variable in the configuration to control the number of worker threads. ## Backpressure Handling Backpressure handling is implemented to manage scenarios of high throughput. The application pauses producing or buffers messages when necessary to alleviate backpressure. ## Metrics and Monitoring The application collects and monitors various metrics to assess its health and performance. Metrics include but are not limited to: - Kafka producer metrics (e.g., message throughput, latency). - Database connection pool metrics. ## Dependencies - `kafkajs`: Kafka library for Node.js. - `mysql2`: MySQL library for Node.js. - `node-locksmith`: Lock management library for handling application instances. - `winston`: Logging library for Node.js. ## Dockerization To run the application in a Docker container, follow these steps: 1. Build the Docker image: ```bash ./build.sh ``` 2. Run the Docker container: ```bash ./run.sh ``` ## Further Development While the Medics Event Publisher is currently a robust solution for real-time event processing, there are opportunities for further development: - **Enhanced Monitoring**: Integrate with a comprehensive monitoring solution for in-depth insights into application performance. - **Dynamic Configuration**: Implement dynamic configuration reloading to allow for on-the-fly adjustments without restarting the application. - **Additional Retry Strategies**: Explore and implement additional retry strategies for message publishing, considering exponential backoff and jitter. ## Deployment These instructions will guide the release team on how to deploy the Medics Event Publisher on a Linux server using Docker. ### Prerequisites - Ensure Docker and Docker Compose are installed on the Linux server. - The user deploying the application should have the necessary permissions to run Docker commands. - Access to the repository containing the Dockerfile or access to a registry where the image is stored. - Ensure that the configuration files (`config.json`) are updated with the production environment settings. ### Steps 1. **Obtain the Docker image:** The Docker image for Medics Event Publisher can either be built directly on the server or pulled from a Docker registry. - To build the image on the server, first clone the repository, then navigate to the directory containing the Dockerfile and run the build command: ```bash git clone git@gitlab:medics/medics-event-publisher.git cd medics-event-publisher docker build -t medics-event-publisher . ``` - If the image is stored in a registry, use the pull command: ```bash docker pull registry.gitlab.com/medics/medics-event-publisher:latest ``` 2. **Prepare the environment variables:** Create an `.env` file on the server and specify all necessary environment variables. 3. **Run the Docker container:** Use the following command to run the application in a Docker container: ```bash docker run -d --name medics-event-publisher --env-file .env -p <HOST_PORT>:<CONTAINER_PORT> medics-event-publisher ``` 

This is the start file 

const express = require("express"); const app = express(); require("dotenv").config(); const createLogger = require("./src/services/Logger"); const logger = createLogger("medics-notification-server"); const LockManager = require("node-locksmith"); const SQLConnectionManager = require("./src/services/SQLConnectionManager"); const KafkaService = require("./src/services/KafkaService"); let config; if (process.env.NODE_ENV !== "production") { config = require("./src/config/config.json"); } else { config = { sql: { host: process.env.SQL_HOST, port: process.env.SQL_PORT, user: process.env.SQL_USER, password: process.env.SQL_PASSWORD, database: process.env.SQL_DATABASE, table: process.env.SQL_TABLE, }, kafka: { brokers: process.env.KAFKA_BROKERS.split(","), clientId: process.env.KAFKA_CLIENT_ID, groupId: process.env.KAFKA_GROUP_ID, autoCommit: process.env.KAFKA_AUTO_COMMIT === "true", }, serverPort: process.env.SERVER_PORT, }; } const kafkaConfig = config.kafka; const kafkaService = new KafkaService(kafkaConfig, logger); if (kafkaService.isConnected()) { console.log("Connected to kafka"); } else { console.error("couldn't connect to kafka"); return; } const SQLService = new SQLConnectionManager( logger, require("./src/config/config.json") ); const POLL_INTERVAL = 2000; // Poll every 2 seconds const CONCURRENCY_LIMIT = 5; // Number of concurrent tasks you want to allow const lockManager = new LockManager({ fileName: "medics-event-publisher.lock", }); /** * Initialize termination event handlers for graceful application shutdown. */ lockManager.initializeTerminationHandlers(); let isProcessing = false; /** * Process a chunk of pending events and publish them to Kafka. * @param {Array} recordsChunk - Chunk of pending records. * @param {Object} connection - Database connection object. */ async function processChunk(recordsChunk, connection) { await Promise.all( recordsChunk.map(async (record) => { const { id, topic, message } = record; try { await kafkaService.publishMessageWithRetry( topic, JSON.stringify(message) ); await markEventAsDelivered(id, connection, true); // Mark as delivered } catch (error) { logger.error(`Error publishing message to Kafka: ${error.message}`); await markEventAsDelivered(id, connection, false, error.message); // Mark as failed } }) ); } /** * Process pending events and publish them to Kafka. * @param {Object} connection - Database connection object. */ async function processPendingEvents(connection) { if (isProcessing) { return; } isProcessing = true; try { // Fetch pending events from the database const pendingRecords = await fetchPendingEvents(connection); // Chunk the task const recordChunks = chunkArray( pendingRecords, Math.ceil(pendingRecords.length / CONCURRENCY_LIMIT) ); // Process chunks in parallel await Promise.all( recordChunks.map((chunk) => processChunk(chunk, connection)) ); } catch (error) { logger.error( "Error during processing pending events:", error.message ? error.message : error ); } finally { isProcessing = false; // Schedule the next execution after a specific interval setTimeout(() => processPendingEvents(connection), POLL_INTERVAL); } } /** * Fetch pending events from the database. * @param {Object} connection - Database connection object. * @returns {Promise<Array>} - Array of pending events. */ async function fetchPendingEvents(connection) { const sql = "SELECT * FROM notification_queue WHERE status = 'PENDING'"; return await SQLService.executeQuery(sql, connection); } /** * Mark an event as delivered or update retry information in the database. * @param {number} id - Event ID. * @param {Object} connection - Database connection object. * @param {boolean} success - Indicates whether the event was delivered successfully. * @param {string|null} errorMessage - Error message in case of failure. * @param {number} maxRetries - Maximum number of retry attempts allowed. * @returns {Promise<void>} - A promise indicating the completion of the operation. */ async function markEventAsDelivered( id, connection, success = true, errorMessage = null, maxRetries = 3 ) { let sql; if (success) { sql = `UPDATE notification_queue SET status = 'DELIVERED', delivered_at = NOW() WHERE id = ${id}`; } else { sql = ` UPDATE notification_queue SET retry_count = retry_count + 1, last_checked_at = NOW(), error_message = '${errorMessage}', status = CASE WHEN retry_count + 1 >= ${maxRetries} THEN 'FAILED' ELSE status END WHERE id = ${id}`; } await SQLService.executeQuery(sql, connection); } /** * Create a chunk of an array. * @param {Array} array - The array to be chunked. * @param {number} chunkSize - The size of each chunk. * @returns {Array} - An array containing chunks of the original array. */ function chunkArray(array, chunkSize) { const results = []; while (array.length) { results.push(array.splice(0, chunkSize)); } return results; } /** * Initialize the application by creating worker threads. */ async function initializeApp() { try { // Ensure the application is not already running await lockManager.checkLock(); // Create a lock to prevent multiple instances of the application await lockManager.createLock(Infinity, 3); // Get a connection from the database pool const connection = await SQLService.getConnection(); app.get(["/hc", "/healthcheck"], (req, res) => { res.status(200).send("OK"); }); app.listen(config.serverPort, () => { console.log(`Server running at http://localhost:${config.serverPort}`); }); processPendingEvents(connection); // Handle unhandled promise rejections and log them process.on("unhandledRejection", (error, promise) => { logger.error("Unhandled Promise Rejection:", error); }); // Graceful shutdown on SIGTERM process.on("SIGTERM", () => { logger.info("Received SIGTERM. Shutting down gracefully."); process.exit(0); }); } catch (error) { // Log initialization errors and exit the process with an error code logger.error("Initialization error:", error); process.exit(1); } } // Initialize the application initializeApp(); 
This is existing kafkaservice

This is my existing kafka service
const { Kafka } = require("kafkajs"); const { promisify } = require("util"); /** * KafkaService class for handling Kafka message publishing. */ class KafkaService { /** * Constructor for KafkaService. * @param {Object} kafkaConfig - Kafka configuration object. * @param {Object} logger - Logger instance. */ constructor(kafkaConfig, logger) { this.producer = new Kafka({ brokers: kafkaConfig.brokers }).producer(); this.logger = logger; this.isProducingPaused = false; // Flag to control producing pause/resume this.bufferSize = 1000; // Maximum size of the message buffer this.messageBuffer = []; // Array to store buffered messages try { // Attempt to establish Kafka connection this.producer = new Kafka({ brokers: kafkaConfig.brokers }).producer(); this.producer.connect().catch((error) => { this.handleKafkaError( error, "Failed to connect to Kafka brokers during initialization." ); }); } catch (error) { this.handleKafkaError(error, "Failed to initialize Kafka producer."); } } /** * Check if the Kafka producer is connected. * @returns {boolean} - True if connected, false otherwise. */ isConnected() { return this.producer && this.producer._isConnected === true; } /** * Publish a message to Kafka. * @param {string} topic - Kafka topic to publish the message to. * @param {string} payload - Message payload to be published. * @returns {Promise<void>} - A promise indicating the completion of the operation. */ async publishMessage(topic, payload) { await this.produceMessage(topic, payload); } /** * Publish a message to Kafka with retry logic. * @param {string} topic - Kafka topic to publish the message to. * @param {string} payload - Message payload to be published. * @param {number} retryCount - Number of retry attempts (default is 3). * @returns {Promise<void>} - A promise indicating the completion of the operation. */ async publishMessageWithRetry(topic, payload, retryCount = 3) { for (let attempt = 1; attempt <= retryCount; attempt++) { try { await this.produceMessage(topic, payload); // Break the loop only if the produceMessage is successful break; } catch (error) { this.handleKafkaError(error, payload); if (attempt === retryCount) { // If it's the last attempt, throw the error to propagate it further throw error; } // If there are more attempts, wait for a short duration before retrying await new Promise((resolve) => setTimeout(resolve, 1000 * attempt)); } } } /** * Handle Kafka errors and take appropriate actions. * @param {Object} error - Kafka error object. * @param {string} payload - Message payload associated with the error. */ handleKafkaError(error, payload) { if (error.type === "INVALID_CONFIG") { this.logger.error( "Invalid Kafka configuration. Check your Kafka broker configuration." ); // Additional logic like notifying administrators can be implemented } else if (error.type === "PRODUCER_QUEUE_SIZE_EXCEEDED") { this.logger.error( "Producer queue size exceeded. Applying backpressure strategy." ); this.pauseProducing(); } else { this.logger.error("Error producing message to Kafka:", { error, payload, }); } } /** * Pause producing messages to Kafka to alleviate backpressure. */ pauseProducing() { this.logger.warn("Pausing producer to alleviate backpressure."); // Additional backpressure handling logic can be added here this.isProducingPaused = true; // Set a timeout to resume producing after a certain time setTimeout(() => { this.logger.warn("Resuming producer after backpressure period."); this.isProducingPaused = false; }, 5000); // Resume after 5 seconds } /** * Produce a message to Kafka. * @param {string} topic - Kafka topic to publish the message to. * @param {string} payload - Message payload to be published. * @returns {Promise<void>} - A promise indicating the completion of the operation. */ async produceMessage(topic, payload) { // Check if producing is paused if (this.isProducingPaused) { this.logger.warn( "Producing is paused. Message will be buffered or skipped." ); this.bufferMessage({ topic, payload }); return; } try { await this.producer.send({ topic, messages: [{ value: payload }], }); this.logger.info("Message produced to Kafka successfully.", { payload }); } catch (error) { this.handleKafkaError(error, payload); throw error; // Throw the error to propagate it further } } /** * Buffer a message when producing is paused. * @param {Object} message - The message object containing topic and payload. */ bufferMessage(message) { // Implement logic to buffer messages for later processing // For example, you can store them in an array or a dedicated buffer // Ensure the buffer size does not exceed the maximum limit if (this.messageBuffer.length < this.bufferSize) { this.messageBuffer.push(message); } else { // If the buffer is full, discard the oldest message const discardedMessage = this.messageBuffer.shift(); this.logger.warn( `Buffer is full. Discarding oldest message: ${JSON.stringify( discardedMessage )}` ); // Add the new message to the buffer this.messageBuffer.push(message); } this.logger.info("Message buffered:", message); } /** * Log message send failure after multiple retries. * @param {string} payload - Message payload that failed to send. */ logMessageSendFailure(payload) { this.logger.error( `Message could not be sent after multiple retries: ${payload}` ); // Additional actions or notifications can be added here } } module.exports = KafkaService;

We need to make some changes in the above application, we want to make this generic we have some issues with the kafka in our environment. SO since this application is dependent on kafka we couldnt able to process any events. SO we want to make it generic so that not only kafka, other ways like even we can able to send that through some api calls to the list of consumers as an emergency situation like this