+-----------------+      +-----------------+      +-----------------+      +------------------+      +--------+
|   Client (API   | ---> |  Node.js/Express| ---> |   Redis (BullMQ)| ---> | Node.js Worker   | ---> | Vonage |
|   Requester)    |      |    API Server   |      |      Queue      |      |  (SMS Sender)    |      |   API  |
+-----------------+      +-----------------+      +-----------------+      +------------------+      +--------+
         |                      ^      |                      ^                     |
         |                      |      |                      |                     |
         |      +---------------+      +----------------------+                     |
         |      |                                                                   |
         v      v                                                                   v
+-----------------+                                                          +-----------------+
|   PostgreSQL    | <--------------------------------------------------------- |   Vonage Status |
|   (Recipients)  |                                                          |    Webhooks     |
+-----------------+                                                          +-----------------+

mkdir vonage-bulk-sms
cd vonage-bulk-sms
npm init -y

# Core Framework & Vonage SDK
npm install express @vonage/server-sdk dotenv

# Queueing System
npm install bullmq ioredis

# Database ORM & Client
npm install @prisma/client
npm install prisma --save-dev

# Input Validation
npm install express-validator

# Replace postgresql with your chosen DB provider if different (e.g._ mysql_ sqlite)
npx prisma init --datasource-provider postgresql

# .env

# Vonage API Credentials (Get from Vonage Dashboard > API Settings)
VONAGE_API_KEY=YOUR_VONAGE_API_KEY
VONAGE_API_SECRET=YOUR_VONAGE_API_SECRET

# Vonage Application Credentials (Create an Application in Vonage Dashboard > Your Applications)
# Application ID is shown after creation.
VONAGE_APPLICATION_ID=YOUR_VONAGE_APPLICATION_ID
# Path relative to project root where you save the generated private key file.
VONAGE_PRIVATE_KEY_PATH=./private.key

# Vonage Number (Purchase/find in Vonage Dashboard > Numbers > Your Numbers)
# Use E.164 format (e.g., 12015550123)
VONAGE_NUMBER=YOUR_VONAGE_VIRTUAL_NUMBER

# Database Connection String (Specific to your DB setup)
# Example for PostgreSQL: postgresql://USER:PASSWORD@HOST:PORT/DATABASE
DATABASE_URL="postgresql://user:password@localhost:5432/bulk_sms?schema=public"

# Redis Connection URL (Specific to your Redis setup)
# Example: redis://HOST:PORT or redis://:PASSWORD@HOST:PORT
REDIS_URL="redis://localhost:6379"

# Server Port
PORT=3000

# API Key for simple authentication (Generate a strong random string)
API_SECRET_KEY=YOUR_STRONG_RANDOM_API_KEY

vonage-bulk-sms/
├── prisma/
│   ├── schema.prisma  # Database schema definition
│   └── migrations/    # Generated by Prisma Migrate
├── node_modules/
├── .env               # Environment variables (DO NOT COMMIT)
├── .gitignore         # Git ignore file
├── index.js           # Main application entry point (API Server)
├── worker.js          # Queue worker process
├── config.js          # Centralized configuration loader
├── queue.js           # Queue definition and setup
├── services/
│   ├── vonageClient.js # Vonage SDK initialization
│   └── prismaClient.js # Prisma client initialization
├── routes/
│   ├── broadcast.js    # API routes for broadcasting
│   └── webhooks.js     # Routes for Vonage webhooks
├── middleware/
│   └── auth.js         # Authentication middleware
├── private.key        # Your Vonage private key (DO NOT COMMIT)
├── package.json
└── package-lock.json

# .gitignore
node_modules
.env
private.key
npm-debug.log*
yarn-debug.log*
yarn-error.log*
*.log
dist

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

const config = {
    vonage: {
        apiKey: process.env.VONAGE_API_KEY,
        apiSecret: process.env.VONAGE_API_SECRET,
        applicationId: process.env.VONAGE_APPLICATION_ID,
        privateKeyPath: process.env.VONAGE_PRIVATE_KEY_PATH,
        senderNumber: process.env.VONAGE_NUMBER,
    },
    database: {
        url: process.env.DATABASE_URL,
    },
    redis: {
        url: process.env.REDIS_URL,
    },
    server: {
        port: process.env.PORT || 3000,
    },
    auth: {
        apiKey: process.env.API_SECRET_KEY,
    },
    queue: {
        name: 'smsQueue',
    }
};

// Basic validation
if (!config.vonage.apiKey || !config.vonage.apiSecret || !config.vonage.applicationId || !config.vonage.privateKeyPath || !config.vonage.senderNumber) {
    console.error(""Vonage configuration missing in .env file"");
    process.exit(1);
}
if (!config.database.url) {
    console.error(""DATABASE_URL missing in .env file"");
    process.exit(1);
}
if (!config.redis.url) {
    console.error(""REDIS_URL missing in .env file"");
    process.exit(1);
}
if (!config.auth.apiKey) {
    console.warn(""API_SECRET_KEY not set in .env file. API will be unprotected."");
}

module.exports = config;

// services/vonageClient.js
const { Vonage } = require('@vonage/server-sdk');
const { Messages } = require('@vonage/messages');
const config = require('../config');
const fs = require('fs');

let privateKey;
try {
    privateKey = fs.readFileSync(config.vonage.privateKeyPath);
} catch (err) {
    console.error(`Error reading private key from ${config.vonage.privateKeyPath}:`, err.message);
    process.exit(1); // Exit if key is essential and missing
}

const vonage = new Vonage({
    apiKey: config.vonage.apiKey,
    apiSecret: config.vonage.apiSecret,
    applicationId: config.vonage.applicationId,
    privateKey: privateKey
}, {
    // Optional: Add custom headers or configure timeouts
    // appendUserAgent: ""my-bulk-sms-app/1.0.0""
});

const messages = new Messages(vonage.credentials);

// Export the Messages client instance directly for sending SMS
module.exports = { messages };

// queue.js
const { Queue, Worker, QueueEvents } = require('bullmq');
const IORedis = require('ioredis');
const config = require('./config');

const connection = new IORedis(config.redis.url, {
    maxRetriesPerRequest: null // BullMQ handles retries
});

connection.on('error', err => {
    console.error('Redis Connection Error:', err);
    // Optional: Implement more robust error handling/alerting
});

const smsQueue = new Queue(config.queue.name, { connection });

// Optional: Listen to queue events for logging/monitoring
const queueEvents = new QueueEvents(config.queue.name, { connection });

queueEvents.on('completed', ({ jobId }) => {
    console.log(`Job ${jobId} completed successfully.`);
});

queueEvents.on('failed', ({ jobId, failedReason }) => {
    console.error(`Job ${jobId} failed: ${failedReason}`);
});

// Export necessary components for API server and worker
module.exports = { smsQueue, connection, Worker, QueueEvents };

// worker.js
const { Worker } = require('./queue'); // Import Worker from queue setup
const config = require('./config');
const { messages } = require('./services/vonageClient'); // Use the Messages client
const IORedis = require('ioredis');

const connection = new IORedis(config.redis.url, {
    maxRetriesPerRequest: null
});

connection.on('error', err => {
    console.error('Worker Redis Connection Error:', err);
});

console.log(`Worker connecting to Redis at ${config.redis.url} for queue ${config.queue.name}`);

const worker = new Worker(config.queue.name, async job => {
    const { to, message } = job.data;
    console.log(`Processing job ${job.id}: Sending SMS to ${to}`);

    try {
        const response = await messages.send({
            message_type: ""text"",
            to: to, // Recipient number
            from: config.vonage.senderNumber, // Your Vonage number
            channel: ""sms"",
            text: message, // The message content
            // Optional: Client reference for tracking
            // client_ref: `job-${job.id}`
        });

        console.log(`Job ${job.id}: SMS sent to ${to}. Message UUID: ${response.message_uuid}`);
        return { message_uuid: response.message_uuid }; // Return data on success

    } catch (error) {
        console.error(`Job ${job.id}: Failed to send SMS to ${to}. Error: ${error.message}`, error);

        // Check for specific Vonage/network errors if needed
        // Example: Check for rate limit errors (429 status code in error response)
        if (error.response && error.response.status === 429) {
            console.warn(`Job ${job.id}: Rate limit hit. Re-throwing error for BullMQ to handle retry with backoff.`);
            // No manual delay needed here - BullMQ's backoff strategy will handle the delay before the next attempt.
        }
        // Throw error to signal failure to BullMQ, triggering its retry mechanism based on configured attempts/backoff
        throw error;
    }
}, {
    connection,
    concurrency: 5, // Process up to 5 jobs concurrently (Adjust based on Vonage/Carrier limits)
    limiter: {      // Rate limiting
        max: 25,    // Max 25 jobs
        duration: 1000 // per 1 second (Adjust based on Vonage/Carrier limits - 30/sec is Vonage API max)
    },
    attempts: 3,    // Retry failed jobs up to 3 times
    backoff: {      // Exponential backoff strategy
        type: 'exponential',
        delay: 5000, // Initial delay 5 seconds
    }
});

worker.on('error', err => {
    // Log worker errors
    console.error('Worker encountered an error:', err);
});

console.log(""SMS Worker started. Waiting for jobs..."");

// services/prismaClient.js
const { PrismaClient } = require('@prisma/client');

const prisma = new PrismaClient();

module.exports = prisma;

// prisma/schema.prisma
generator client {
  provider = "prisma-client-js"
}

datasource db {
  provider = "postgresql" // Or your chosen DB
  url      = env("DATABASE_URL")
}

model Recipient {
  id          String   @id @default(cuid())
  phoneNumber String   @unique // E.164 format recommended
  firstName   String?
  lastName    String?
  createdAt   DateTime @default(now())
  updatedAt   DateTime @updatedAt

  // Add other relevant fields like tags, lists, optOut status, etc.
  // Example:
  // isOptedOut  Boolean  @default(false)
  // tags        String[] @default([])
}

// Consider adding models for Broadcasts/Campaigns to track jobs
// model Broadcast {
//   id        String   @id @default(cuid())
//   message   String
//   status    String   @default("PENDING") // PENDING, PROCESSING, COMPLETED, FAILED
//   createdAt DateTime @default(now())
//   updatedAt DateTime @updatedAt
//   // Relation to jobs or recipients if needed
// }

// Consider adding a model for tracking individual message statuses via webhooks
// model MessageLog {
//   id              String    @id @default(cuid())
//   messageUuid     String    @unique // From Vonage API response/webhook
//   recipientNumber String
//   broadcastId     String?   // Optional link to a Broadcast
//   status          String    // e.g., submitted, delivered, failed, rejected
//   statusTimestamp DateTime?
//   createdAt       DateTime  @default(now())
//   updatedAt       DateTime  @updatedAt
// }

# Generate SQL migration files based on schema changes
npx prisma migrate dev --name init_recipients

# Optional: Populate with sample data (Create a seed script)
# npx prisma db seed (Requires setup in package.json and a seed file)

// middleware/auth.js
const config = require('../config');

const authenticateApiKey = (req, res, next) => {
    const apiKey = req.headers['x-api-key'];

    if (!config.auth.apiKey) {
        console.warn("API Key auth middleware active, but no API_SECRET_KEY set in config. Allowing request.");
        return next(); // Allow if no key is configured (for local dev maybe)
    }

    if (!apiKey || apiKey !== config.auth.apiKey) {
        return res.status(401).json({ message: 'Unauthorized: Invalid or missing API key' });
    }

    next();
};

module.exports = { authenticateApiKey };

// routes/broadcast.js
const express = require('express');
const { body, validationResult } = require('express-validator');
const { smsQueue } = require('../queue');
const prisma = require('../services/prismaClient');

const router = express.Router();

router.post(
    '/',
    [ // Input validation
        body('message').isString().notEmpty().withMessage('Message text is required'),
        body('recipientIds')
            .optional({ nullable: true })
            .isArray({ min: 1 }).withMessage('recipientIds must be an array with at least one ID if provided'),
        body('recipientIds.*').isString().withMessage('Each recipientId must be a string'),

        // --- Placeholder for additional recipient selection methods ---
        // Validation for alternative ways (e.g., tags, all) would go here.
        // body('recipientTags').optional().isArray()...
        // body('sendToAll').optional().isBoolean()...

        // --- Mutual Exclusivity Check Example ---
        // Ensure only one method of specifying recipients is used per request.
        // This example assumes recipientIds is the only method implemented for now.
        body().custom((value, { req }) => {
            const { recipientIds /*, recipientTags, sendToAll */ } = req.body;
            // Extend this logic if recipientTags or sendToAll are implemented
            const providedMethods = [recipientIds].filter(v => v !== undefined).length;
            if (providedMethods !== 1) {
                 // Adjust error message if other methods are added
                throw new Error('Please provide exactly one method for specifying recipients (e.g., recipientIds).');
            }
            return true;
        })
    ],
    async (req, res) => {
        const errors = validationResult(req);
        if (!errors.isEmpty()) {
            return res.status(400).json({ errors: errors.array() });
        }

        const { message, recipientIds /*, recipientTags, sendToAll */ } = req.body;
        let recipients = [];

        try {
            // --- Logic to fetch recipients based on input ---
            if (recipientIds) {
                recipients = await prisma.recipient.findMany({
                    where: {
                        id: { in: recipientIds },
                        // Add other conditions like NOT opted out
                        // isOptedOut: false
                    },
                    select: { phoneNumber: true } // Only select needed field
                });

                // Check if all provided IDs were found
                if (recipients.length !== recipientIds.length) {
                   console.warn(`Some recipient IDs not found or match filtering criteria (e.g., opted out). Proceeding with ${recipients.length} found recipients.`);
                   // Decide if this is an error or just proceed with found ones
                }
            }
            // --- Add logic for fetching by tags or all recipients ---
            // else if (recipientTags) {
            //   console.log('Fetching recipients by tags - Implementation needed.');
            //   // Implementation left as an exercise: query Prisma based on tags.
            // }
            // else if (sendToAll) {
            //   console.log('Fetching all recipients - Implementation needed.');
            //   // Implementation left as an exercise: query Prisma for all non-opted-out recipients.
            // }

            if (!recipients || recipients.length === 0) {
                return res.status(404).json({ message: 'No valid recipients found for the given criteria.' });
            }

            // --- Enqueue jobs ---
            const jobPromises = recipients.map(recipient => {
                return smsQueue.add(
                    'send-sms', // Job name (can be more descriptive)
                    {
                        to: recipient.phoneNumber,
                        message: message
                    },
                    {
                        // Optional: Job specific settings like unique job IDs
                        // jobId: `sms-${recipient.id}-${Date.now()}` // Example unique ID
                        removeOnComplete: true, // Keep queue clean
                        removeOnFail: 50 // Keep failed jobs for inspection
                    }
                );
            });

            await Promise.all(jobPromises);

            const messageCount = recipients.length;
            console.log(`Successfully enqueued ${messageCount} SMS jobs for broadcast.`);

            res.status(202).json({
                message: `Accepted: Queued ${messageCount} SMS messages for delivery.`,
                // Optional: Return a broadcast ID for tracking
                // broadcastId: "some-unique-id"
            });

        } catch (error) {
            console.error('Error processing broadcast request:', error);
            res.status(500).json({ message: 'Internal Server Error while queuing messages.' });
        }
    }
);

module.exports = router;

// routes/webhooks.js
const express = require('express');
const prisma = require('../services/prismaClient'); // Import prisma if updating DB

const router = express.Router();

// Endpoint for Vonage Message Status updates
router.post('/status', async (req, res) => { // Make async if doing DB operations
    const statusData = req.body;
    console.log('Received Vonage Status Webhook:', JSON.stringify(statusData, null, 2));

    // --- Process the status update ---
    // Examples:
    // - Log the status (already done above)
    // - Update a database record tracking the message status using `message_uuid` or `client_ref`
    // - Trigger alerts on failure statuses ('failed', 'rejected')

    // --- Example: Update DB (pseudo-code) ---
    // Note: This requires a 'MessageLog' model in schema.prisma (see schema example)
    // and corresponding database migration.
    /*
    if (statusData.message_uuid && statusData.status) {
        try {
            // Find the message log entry using the UUID provided by Vonage
            await prisma.messageLog.updateMany({ // Use updateMany in case client_ref maps to multiple messages
                where: { messageUuid: statusData.message_uuid },
                data: {
                    status: statusData.status,
                    statusTimestamp: statusData.timestamp ? new Date(statusData.timestamp) : new Date(),
                    // You might also store error codes if provided
                    // errorCode: statusData.error?.code,
                    // networkCode: statusData.network_code
                }
            });
             console.log(`Updated status for message ${statusData.message_uuid} to ${statusData.status}`);
        } catch (dbError) {
            console.error(`Failed to update message status in DB for UUID ${statusData.message_uuid}:`, dbError);
            // Decide how to handle DB errors - potentially retry later or log for investigation
        }
    } else {
        console.warn('Received status webhook without message_uuid or status.');
    }
    */

    // Vonage expects a 200 OK response to acknowledge receipt
    res.status(200).send('OK');
});

// Endpoint for Vonage Inbound Messages (if needed for replies)
router.post('/inbound', async (req, res) => { // Make async if doing DB operations
    const inboundData = req.body;
    console.log('Received Vonage Inbound SMS:', JSON.stringify(inboundData, null, 2));

    // --- Process inbound message ---
    // Examples:
    // - Handle STOP/OPT-OUT keywords (update recipient status in DB)
    // - Route message to support system
    // - Trigger automated replies

    // Example: Basic STOP keyword handling
    /*
    if (inboundData.text && inboundData.msisdn) { // msisdn is the sender's number
        const messageText = inboundData.text.trim().toUpperCase();
        if (messageText === 'STOP' || messageText === 'UNSUBSCRIBE') {
            try {
                await prisma.recipient.updateMany({
                    where: { phoneNumber: inboundData.msisdn }, // Use E.164 format match
                    data: { isOptedOut: true } // Assuming an isOptedOut field
                });
                console.log(`Recipient ${inboundData.msisdn} opted out.`);
            } catch (dbError) {
                console.error(`Failed to process opt-out for ${inboundData.msisdn}:`, dbError);
            }
        }
    }
    */

    res.status(200).send('OK');
});

module.exports = router;

// index.js
const express = require('express');
const config = require('./config');
const { authenticateApiKey } = require('./middleware/auth');
const broadcastRoutes = require('./routes/broadcast');
const webhookRoutes = require('./routes/webhooks');

const app = express();

// --- Middleware ---
console.log('Setting up middleware...');
app.use(express.json()); // Parse JSON request bodies
app.use(express.urlencoded({ extended: true })); // Parse URL-encoded bodies (needed for Vonage webhooks sometimes)

// --- Routes ---
console.log('Registering routes...');
// Apply API key auth ONLY to the broadcast route
app.use('/api/broadcast', authenticateApiKey, broadcastRoutes);
// Webhooks should generally be open but could have signature validation (see Vonage docs - recommended for production)
app.use('/webhooks', webhookRoutes);

// --- Basic Health Check ---
app.get('/health', (req, res) => {
    res.status(200).json({ status: 'UP', timestamp: new Date().toISOString() });
});

// --- Error Handling (Basic) ---
// Catch-all for unhandled routes
app.use((req, res, next) => {
    res.status(404).json({ message: 'Not Found' });
});

// Global error handler
app.use((err, req, res, next) => {
    console.error('Unhandled Application Error:', err);
    res.status(err.status || 500).json({
        message: err.message || 'Internal Server Error',
        // Optionally include stack trace in development
        // stack: process.env.NODE_ENV === 'development' ? err.stack : undefined
    });
});

// --- Start Server ---
const PORT = config.server.port;
app.listen(PORT, () => {
    console.log(`Server listening on port ${PORT}`);
    console.log(`Redis configured at ${config.redis.url}`);
    console.log(`API Key Authentication ${config.auth.apiKey ? 'Enabled' : 'Disabled'}`);
    console.log(`Vonage Sender Number: ${config.vonage.senderNumber}`);
});