Developer Guide: Real-Time SMS Delivery Status with Node.js, Infobip & SSE

+-------------+       +-----------------+       +-----------------+       +---------------+
|  Frontend   |<----->|  Node.js/Express|<----->|     Infobip     |<----->| Mobile Device |
| (React/Vue) |       |     Backend     |       |     Platform    |       |  (Recipient)  |
+-------------+       +-----------------+       +-----------------+       +---------------+
      |  /|\               |        |                  |         |               |
      |   | SSE            | Send SMS Request         |         | Send SMS      |
      |   +----------------+                          |         +-------------->|
      |                    |                          |                         |
      |                    | Store Initial Status     | Receive SMS             |
      |                    | (e.g., PENDING)          |                         |
      |                    V                          |                         |
      |              +----------+                     | Status Update           |
      |              | Database |                     | (e.g., DELIVERED)       |
      |              | (Postgres)                     |<------------------------+
      |              +----------+                     |                         |
      |                    ^                          |                         |
      | Update Status      | Webhook Notification     |                         |
      | /|\                +--------------------------+                         |
      |  | Push Update     |                          |                         |
      +--+-----------------+                          |                         |
         SSE Connection                               |                         |
         (listens for status changes)                 |                         |

mkdir infobip-status-backend
cd infobip-status-backend

npm init -y

npm install express axios dotenv cors sse-channel uuid @prisma/client
npm install --save-dev nodemon prisma

npx prisma init --datasource-provider postgresql

# .env

# Prisma Database Connection
# Example for local PostgreSQL: postgresql://USER:PASSWORD@HOST:PORT/DATABASE
DATABASE_URL="postgresql://postgres:yourpassword@localhost:5432/infobip_status?schema=public"

# Infobip Credentials
INFOBIP_BASE_URL="YOUR_INFOBIP_BASE_URL" # e.g., xyz.api.infobip.com (Find this in your Infobip account)
INFOBIP_API_KEY="YOUR_INFOBIP_API_KEY"  # Find this in your Infobip account

# Server Configuration
PORT=3001
# This will be the public URL ngrok provides for your webhook during development
WEBHOOK_PUBLIC_URL="YOUR_NGROK_FORWARDING_URL" # e.g., https://abcd-1234.ngrok.io

# Optional: Secret for basic webhook validation
WEBHOOK_SECRET="a-very-secure-secret-string"

// package.json (add/modify the "scripts" section)
{
  "scripts": {
    "start": "node src/server.js",
    "dev": "nodemon src/server.js",
    "prisma:migrate": "prisma migrate dev --name init",
    "prisma:generate": "prisma generate"
  }
}

infobip-status-backend/
├── prisma/
│   └── schema.prisma
├── src/
│   ├── config/
│   │   └── index.js
│   ├── controllers/
│   │   └── messageController.js
│   ├── middleware/
│   │   └── webhookValidation.js
│   ├── routes/
│   │   ├── index.js
│   │   └── messageRoutes.js
│   ├── services/
│   │   ├── databaseService.js
│   │   ├── infobipService.js
│   │   └── sseService.js
│   ├── utils/
│   │   └── logger.js
│   └── server.js
├── .env
├── .gitignore
└── package.json

// src/server.js
require('dotenv').config(); // Load .env variables early
const express = require('express');
const cors = require('cors');
const config = require('./config');
const routes = require('./routes');
const { initializeSSE } = require('./services/sseService');
const logger = require('./utils/logger');

const app = express();

// --- Middleware ---
// Enable CORS for all origins (adjust for production)
app.use(cors());

// Use express.json() for webhook parsing, BEFORE custom raw body middleware
// Increase limit if large payloads are expected
app.use(express.json({ limit: '5mb' }));

// Raw body parsing specifically for webhook verification (if needed)
// We might need the raw body *before* express.json() parses it
// Example: app.use('/api/messages/webhook', express.raw({ type: 'application/json' })); // Ensure path matches route
// If using a simple secret header or basic auth handled elsewhere, express.json() is usually sufficient.

// Initialize SSE - must be done before routes that use it
initializeSSE(app);

// --- Routes ---
app.use('/api', routes);

// --- Basic Health Check ---
app.get('/health', (req, res) => {
    res.status(200).send('OK');
});

// --- Error Handling (Basic) ---
app.use((err, req, res, next) => {
    logger.error('Unhandled Error:', err);
    res.status(500).json({ message: 'Internal Server Error' });
});

// --- Start Server ---
app.listen(config.port, () => {
    logger.info(`Server running on port ${config.port}`);
    logger.info(`Infobip Webhook URL should be configured to point to: ${config.webhookPublicUrl}/api/messages/webhook`);
    // Remind user to run ngrok if URL is not set
    if (!config.webhookPublicUrl || config.webhookPublicUrl === "YOUR_NGROK_FORWARDING_URL") {
        logger.warn('WEBHOOK_PUBLIC_URL not set in .env. Run ngrok and update .env or Infobip configuration.');
    }
});

module.exports = app; // Export for potential testing

// src/config/index.js
require('dotenv').config();

const config = {
    port: process.env.PORT || 3001,
    infobip: {
        baseUrl: process.env.INFOBIP_BASE_URL,
        apiKey: process.env.INFOBIP_API_KEY,
    },
    databaseUrl: process.env.DATABASE_URL,
    webhookPublicUrl: process.env.WEBHOOK_PUBLIC_URL,
    webhookSecret: process.env.WEBHOOK_SECRET, // Optional secret
};

// Basic validation
if (!config.infobip.baseUrl || !config.infobip.apiKey) {
    console.error('FATAL ERROR: Infobip Base URL or API Key not configured in .env');
    process.exit(1);
}
if (!config.databaseUrl) {
    console.error('FATAL ERROR: DATABASE_URL not configured in .env');
    process.exit(1);
}

module.exports = config;

// src/utils/logger.js

// NOTE: This is a very basic logger. For production, use a robust library like Winston or Pino.
const logger = {
    info: (...args) => console.log('[INFO]', ...args),
    warn: (...args) => console.warn('[WARN]', ...args),
    error: (...args) => console.error('[ERROR]', ...args),
    debug: (...args) => console.debug('[DEBUG]', ...args),
};

module.exports = logger;

// prisma/schema.prisma

generator client {
  provider = ""prisma-client-js""
}

datasource db {
  provider = ""postgresql""
  url      = env(""DATABASE_URL"")
}

model MessageStatus {
  id               String    @id @default(uuid()) // Internal unique ID
  correlationId    String    @unique // Our own ID to link frontend request <-> Infobip message
  infobipMessageId String?   // The ID returned by Infobip when sending
  phoneNumber      String    // Recipient phone number
  messageText      String?   // Optional: store message text
  status           String    // e.g., PENDING, SENT, DELIVERED, FAILED, UNDELIVERABLE, REJECTED
  statusGroup      String?   // e.g., PENDING, SENT, DELIVERED, FAILED, UNDELIVERABLE
  statusDescription String?  // Detailed description from Infobip
  errorCode        Int?      // Infobip error code if applicable
  errorMessage     String?   // Infobip error message if applicable
  createdAt        DateTime  @default(now())
  updatedAt        DateTime  @updatedAt

  @@index([infobipMessageId]) // Index for quick lookup on webhook callback
  @@index([createdAt])
}

npx prisma migrate dev --name init

npx prisma generate

// src/services/databaseService.js
const { PrismaClient } = require('@prisma/client');
const logger = require('../utils/logger');

const prisma = new PrismaClient({
    // log: ['query', 'info', 'warn', 'error'], // Enable logging if needed
});

async function saveInitialMessageStatus(data) {
    try {
        const message = await prisma.messageStatus.create({
            data: {
                correlationId: data.correlationId,
                phoneNumber: data.phoneNumber,
                messageText: data.messageText, // Optional
                status: 'PENDING_SEND', // Initial status before hitting Infobip
                // infobipMessageId will be updated later
            },
        });
        logger.info(`Initial status saved for correlationId: ${data.correlationId}`);
        return message;
    } catch (error) {
        logger.error(`Error saving initial message status for ${data.correlationId}:`, error);
        throw error; // Re-throw to be handled by controller
    }
}

async function updateMessageStatusPostSend(correlationId, infobipResponse) {
     // Infobip response structure might vary slightly, adjust parsing as needed
    const messageInfo = infobipResponse?.messages?.[0];
    if (!messageInfo || !messageInfo.messageId) {
        logger.warn(`Could not extract messageId from Infobip response for ${correlationId}`);
        // Update status to reflect send failure or ambiguity
        try {
            return await prisma.messageStatus.update({
                 where: { correlationId },
                 data: {
                     status: 'SEND_RESPONSE_INVALID', // Indicate issue parsing Infobip's response
                     statusDescription: 'Infobip response was missing expected messageId.',
                     updatedAt: new Date(),
                 },
             });
        } catch (updateError) {
             logger.error(`Error updating status to SEND_RESPONSE_INVALID for ${correlationId}:`, updateError);
             return null; // Indicate update failure
        }
    }

    try {
        const updatedMessage = await prisma.messageStatus.update({
            where: { correlationId: correlationId },
            data: {
                infobipMessageId: messageInfo.messageId,
                status: messageInfo.status?.name || 'UNKNOWN_POST_SEND',
                statusGroup: messageInfo.status?.groupName || 'UNKNOWN',
                statusDescription: messageInfo.status?.description || 'Status after sending.',
                updatedAt: new Date(),
            },
        });
        logger.info(`Status updated post-send for correlationId ${correlationId} (Infobip ID: ${messageInfo.messageId})`);
        return updatedMessage;
    } catch (error) {
        logger.error(`Error updating message status post-send for ${correlationId}:`, error);
        // Don't crash the webhook receiver, but log the error
        // Consider adding specific error handling if the record wasn't found
        return null; // Indicate update failure
    }
}

async function updateMessageStatusFromWebhook(report) {
    const messageId = report.messageId;
    if (!messageId) {
        logger.warn('Webhook received without messageId:', report);
        return null;
    }

    try {
        // Find the message by Infobip's messageId
        const existingMessage = await prisma.messageStatus.findFirst({
             where: { infobipMessageId: messageId },
         });

         if (!existingMessage) {
             logger.warn(`Webhook received for unknown infobipMessageId: ${messageId}`);
             // Decide how to handle this: maybe store orphaned reports? For now, ignore.
             return null;
         }

        const updatedMessage = await prisma.messageStatus.update({
            where: { id: existingMessage.id }, // Update using the primary key found
            data: {
                status: report.status?.name || 'UNKNOWN_WEBHOOK',
                statusGroup: report.status?.groupName || 'UNKNOWN',
                statusDescription: report.status?.description || 'Status from webhook.',
                errorCode: report.error?.id,
                errorMessage: report.error?.name, // Or description/text if available
                updatedAt: new Date(report.doneAt || report.sentAt || Date.now()), // Use Infobip timestamp if available
            },
        });
        logger.info(`Webhook processed for infobipMessageId ${messageId}. Status: ${updatedMessage.status}`);
        return updatedMessage; // Return the updated record including correlationId
    } catch (error) {
        logger.error(`Error processing webhook for messageId ${messageId}:`, error);
         // Consider specific error handling (e.g., record not found after initial check - race condition?)
        return null; // Indicate failure
    }
}

module.exports = {
    prisma, // Export client for potential direct use (use with caution)
    saveInitialMessageStatus,
    updateMessageStatusPostSend,
    updateMessageStatusFromWebhook,
};

// src/services/sseService.js
const SSEChannel = require('sse-channel');
const logger = require('../utils/logger');

// Single channel instance to manage all client connections
const sseChannel = new SSEChannel({
    pingInterval: 15000, // Keep connections alive (milliseconds)
    historySize: 10,    // Store last 10 messages (optional)
    cors: { origins: ['*'], // Allow all origins for simplicity (restrict in production!)
            headers: ['Cache-Control', 'Connection', 'Content-Type', 'Last-Event-ID']}
});

function initializeSSE(app) {
    // Endpoint for clients to connect
    // The ':correlationId' allows us to associate a connection with a specific message request
    app.get('/api/messages/sse-status/:correlationId', (req, res) => {
        const { correlationId } = req.params;
        if (!correlationId) {
            return res.status(400).send('Missing correlationId');
        }

        // Add client to the channel, identifying them by correlationId
        // The second argument allows filtering publications later
        sseChannel.addClient(req, res, correlationId);
        logger.info(`SSE client connected for correlationId: ${correlationId}`);

        // Optional: Send initial status or confirmation message
        // publishStatusUpdate(correlationId, { status: "CONNECTED", message: "Waiting for updates..." });

        req.on('close', () => {
             logger.info(`SSE client disconnected for correlationId: ${correlationId}`);
             // sse-channel handles cleanup automatically, but you could add logic here if needed
        });
    });
    logger.info('SSE service initialized at /api/messages/sse-status/:correlationId');
}

// Function to publish updates to specific clients
function publishStatusUpdate(correlationId, data) {
    if (!correlationId) {
        logger.warn('Attempted to publish SSE update without correlationId:', data);
        return;
    }
    // Publish the event only to clients associated with this correlationId
    // The first argument is the event name (client-side listener uses this)
    // The second argument is the data payload (JSON)
    // The third argument is the filter criteria (client IDs)
    sseChannel.publish(
        'message_status_update', // Event name
        data,                   // Payload (must be serializable to JSON)
        { // Filter options: send only to clients with matching ID
            ids: [correlationId]
        }
    );
    logger.info(`Published SSE update for ${correlationId}: Status - ${data?.status}`);
}

// Optional: Get count of connected clients
function getClientCount() {
    return sseChannel.getConnectionCount();
}

module.exports = {
    initializeSSE,
    publishStatusUpdate,
    getClientCount,
    // sseChannel // Expose channel directly if needed, but prefer methods
};

// src/services/infobipService.js
const axios = require('axios');
const config = require('../config');
const logger = require('../utils/logger');

const infobipClient = axios.create({
    baseURL: `https://${config.infobip.baseUrl}`,
    headers: {
        'Authorization': `App ${config.infobip.apiKey}`,
        'Content-Type': 'application/json',
        'Accept': 'application/json',
    },
});

async function sendSms(phoneNumber, message, correlationId) {
    const url = '/sms/2/text/advanced';
    const requestBody = {
        messages: [
            {
                destinations: [{ to: phoneNumber }],
                text: message,
                // It's crucial Infobip knows where to send reports.
                // Set this globally in Infobip portal, or per message here:
                notifyUrl: `${config.webhookPublicUrl}/api/messages/webhook`,
                notifyContentType: 'application/json',
                // You can include correlationId for potential tracking/logging within Infobip if needed
                // customPayload: { correlationId: correlationId },
            },
        ],
        // Optional: Define tracking or other parameters globally
        // tracking: {
        //     track: "URL",
        //     processKey: correlationId // Example using correlationId for tracking
        // }
    };

    logger.info(`Sending SMS to ${phoneNumber} with correlationId: ${correlationId}`);
    // Log request body *without* sensitive info if needed for debugging
    // logger.debug('Infobip Request Body:', JSON.stringify({ messages: [{ destinations: [{/*omitted*/}], text: 'omitted' }] }));

    try {
        const response = await infobipClient.post(url, requestBody);
        logger.info(`Infobip API Response Status: ${response.status}`);
        // Log limited success response details
        const messageInfo = response.data?.messages?.[0];
        if (messageInfo) {
             logger.info(`Infobip Message ID: ${messageInfo.messageId}, Status: ${messageInfo.status?.groupName} (${messageInfo.status?.name})`);
        } else {
             logger.warn('Infobip response structure unexpected:', response.data);
        }

        return response.data; // Return the full response data
    } catch (error) {
        logger.error(`Error sending SMS via Infobip for ${correlationId}:`);
        if (error.response) {
            // The request was made and the server responded with a status code
            // that falls out of the range of 2xx
            logger.error(`Data: ${JSON.stringify(error.response.data)}`);
            logger.error(`Status: ${error.response.status}`);
            logger.error(`Headers: ${JSON.stringify(error.response.headers)}`);
            // Rethrow a more structured error or return a specific error object
            throw new Error(`Infobip API error: ${error.response.status} - ${JSON.stringify(error.response.data)}`);
        } else if (error.request) {
            // The request was made but no response was received
            logger.error('Request Error: No response received from Infobip.', error.request);
            throw new Error('Infobip API error: No response received.');
        } else {
            // Something happened in setting up the request that triggered an Error
            logger.error('Axios Error:', error.message);
            throw new Error(`Infobip API error: ${error.message}`);
        }
    }
}

module.exports = {
    sendSms,
};

// src/routes/index.js
const express = require('express');
const messageRoutes = require('./messageRoutes');

const router = express.Router();

router.use('/messages', messageRoutes);
// Add other resource routes here if needed (e.g., /users, /settings)

module.exports = router;

// src/routes/messageRoutes.js
const express = require('express');
const messageController = require('../controllers/messageController');
const { validateWebhook } = require('../middleware/webhookValidation'); // We'll create this middleware

const router = express.Router();

// POST /api/messages/send - Endpoint to initiate sending an SMS
router.post('/send', messageController.handleSendSms);

// POST /api/messages/webhook - Endpoint for Infobip to send delivery reports
// Apply validation middleware if using a secret or signature validation
router.post('/webhook', validateWebhook, messageController.handleInfobipWebhook);

// GET /api/messages/sse-status/:correlationId - SSE endpoint is handled directly in sseService/server.js setup
// No route definition needed here as it's attached directly to the app instance

module.exports = router;

// src/controllers/messageController.js
const { v4: uuidv4 } = require('uuid');
const infobipService = require('../services/infobipService');
const dbService = require('../services/databaseService');
const sseService = require('../services/sseService');
const logger = require('../utils/logger');

async function handleSendSms(req, res) {
    const { phoneNumber, message } = req.body;

    // Basic Input Validation (See Section 7 for more robust validation)
    if (!phoneNumber || !message) {
        return res.status(400).json({ message: 'Missing required fields: phoneNumber, message' });
    }
    // Add more robust validation (e.g., phone number format) needed for production
    // Example: if (!/^\+?\d{10,15}$/.test(phoneNumber)) { return res.status(400).json(...) }

    const correlationId = uuidv4(); // Generate unique ID for this request

    try {
        // 1. Save initial status to DB
        await dbService.saveInitialMessageStatus({
            correlationId,
            phoneNumber,
            messageText: message,
        });

        // 2. Send SMS via Infobip
        const infobipResponse = await infobipService.sendSms(phoneNumber, message, correlationId);

        // 3. Update DB with Infobip's response (messageId, initial status from Infobip)
         const updatedRecord = await dbService.updateMessageStatusPostSend(correlationId, infobipResponse);

        // 4. Respond to the client immediately (don't wait for webhook)
        res.status(202).json({ // 202 Accepted: Request accepted, processing will continue
            message: 'SMS submitted successfully. Waiting for delivery status.',
            correlationId: correlationId,
            infobipStatus: updatedRecord?.status || 'PENDING_INFOBIP_ID', // Send back current known status
            infobipMessageId: updatedRecord?.infobipMessageId // Send back the ID if available
        });

        // 5. Publish initial status update via SSE (optional, but good UX)
        if(updatedRecord) {
            sseService.publishStatusUpdate(correlationId, {
                correlationId: updatedRecord.correlationId,
                status: updatedRecord.status,
                statusGroup: updatedRecord.statusGroup,
                description: updatedRecord.statusDescription,
                timestamp: updatedRecord.updatedAt,
            });
        } else {
             // Handle case where updateMessageStatusPostSend failed (e.g., SEND_RESPONSE_INVALID)
             sseService.publishStatusUpdate(correlationId, {
                correlationId: correlationId,
                status: 'SEND_RESPONSE_INVALID',
                description: 'Failed to get valid confirmation from Infobip after sending.',
                timestamp: new Date(),
             });
        }

    } catch (error) {
        logger.error(`Failed to process send SMS request for ${correlationId}:`, error.message);
        // Update DB status to indicate failure before Infobip
         try {
             await dbService.prisma.messageStatus.update({
                 where: { correlationId },
                 data: { status: 'FAILED_PRE_SEND', errorMessage: error.message, updatedAt: new Date() }
             });
         } catch (dbError) {
             logger.error(`Failed to update DB status after send error for ${correlationId}:`, dbError);
         }

         // Publish failure via SSE
         sseService.publishStatusUpdate(correlationId, {
             correlationId: correlationId,
             status: 'FAILED_PRE_SEND',
             description: `Failed to initiate SMS send: ${error.message}`,
             timestamp: new Date(),
         });

        // Respond with server error
        res.status(500).json({
            message: 'Failed to initiate SMS send.',
            correlationId: correlationId,
            error: error.message // Provide limited error info
        });
    }
}

async function handleInfobipWebhook(req, res) {
    // Assuming validateWebhook middleware passed (or isn't used)
    const deliveryReports = req.body.results; // Infobip sends reports in a 'results' array

    if (!Array.isArray(deliveryReports)) {
        logger.warn('Webhook received with invalid format (expected ""results"" array):', req.body);
        return res.status(400).send('Bad Request: Invalid payload format.');
    }

    logger.info(`Received webhook with ${deliveryReports.length} report(s).`);

    // Process each report asynchronously
    // We don't wait for all promises here to respond quickly to Infobip
    deliveryReports.forEach(async (report) => {
        try {
            const updatedMessage = await dbService.updateMessageStatusFromWebhook(report);

            if (updatedMessage && updatedMessage.correlationId) {
                // Publish update via SSE using the correlationId found in the DB record
                sseService.publishStatusUpdate(updatedMessage.correlationId, {
                    correlationId: updatedMessage.correlationId,
                    status: updatedMessage.status,
                    statusGroup: updatedMessage.statusGroup,
                    description: updatedMessage.statusDescription,
                    errorCode: updatedMessage.errorCode,
                    errorMessage: updatedMessage.errorMessage,
                    timestamp: updatedMessage.updatedAt,
                });
            } else {
                logger.warn(`Webhook processed but no corresponding record or correlationId found for Infobip Message ID: ${report.messageId}`);
            }
        } catch (error) {
            // Log errors during individual report processing but don't crash
            logger.error(`Error processing individual webhook report for messageId ${report?.messageId}:`, error);
        }
    });

    // Respond to Infobip immediately that the webhook was received successfully
    // Infobip expects a 2xx response quickly.
    res.status(200).send('Webhook received.');
}

module.exports = {
    handleSendSms,
    handleInfobipWebhook,
};