Build a Bulk SMS System with MessageBird, Node.js, and Fastify

+-----------------+      +-----------------+      +-----------------+      +-------------------+      +------------------+
| Client (e.g. UI)| ---> | Fastify API     | ---> | Redis (BullMQ)  | ---> | Background Worker | ---> | MessageBird API  |
|                 |      | (Accepts Request)|      | (Job Queue)     |      | (Processes Jobs)  |      | (Sends SMS)      |
+-----------------+      +-----------------+      +-----------------+      +-------------------+      +------------------+
        |                      |                                                 |         ^
        |                      | Optional: Write Job Status                      |         | Optional: Read Job Status
        |                      V                                                 V         |
        +----------------------|-------------------------------------------------|---------+
                               |                  +-----------------+
                               +----------------> | Database (PG)   |
                                                  | (Tracks Messages)|
                                                  +-----------------+

mkdir fastify-messagebird-bulk
cd fastify-messagebird-bulk

npm init -y

# Core dependencies
npm install fastify messagebird bullmq ioredis dotenv

# Optional: Prisma for database tracking
npm install prisma @prisma/client pg --save-dev

# Development dependencies (optional but recommended)
npm install -D nodemon concurrently

fastify-messagebird-bulk/
├── prisma/               # (Optional) Prisma schema and migrations
├── src/
│   ├── config/           # Configuration files (e.g., MessageBird, Redis)
│   ├── jobs/             # BullMQ job definitions and processors
│   ├── routes/           # Fastify API route definitions
│   ├── services/         # Business logic (e.g., MessageBird service wrapper)
│   ├── utils/            # Utility functions
│   ├── api.js            # Fastify server entry point
│   └── worker.js         # BullMQ worker entry point
├── .env                  # Environment variables (DO NOT COMMIT)
├── .env.example          # Example environment variables
├── .gitignore
└── package.json

# MessageBird Configuration
MESSAGEBIRD_API_KEY=YOUR_MESSAGEBIRD_LIVE_ACCESS_KEY
MESSAGEBIRD_ORIGINATOR=YOUR_SENDER_ID_OR_NUMBER # e.g., +12025550147 or "MyCompany"

# Redis Configuration (for BullMQ)
REDIS_HOST=127.0.0.1
REDIS_PORT=6379
# REDIS_PASSWORD=your_redis_password # Uncomment if your Redis requires auth
# REDIS_URL=redis://user:password@host:port # Alternative connection string

# Database Configuration (Optional – for Prisma)
# Example for PostgreSQL
DATABASE_URL="postgresql://user:password@host:port/database_name?schema=public"

# Application Configuration
API_PORT=3000
API_HOST=0.0.0.0 # Listen on all available network interfaces
QUEUE_NAME=sms_sending_queue

cp .env.example .env

# Dependencies
/node_modules

# Environment Variables
.env

# Prisma
# Prisma migration files (*.sql within migration folders) define schema changes and should be committed.
# Ignore SQLite files if used during development
*.db
*.db-journal

# Logs
*.log

# Build Output (if any)
/dist

# OS generated files
.DS_Store
Thumbs.db

npx prisma init --datasource-provider postgresql

{
  // ... other package.json content
  "scripts": {
    "start:api": "node src/api.js",
    "start:worker": "node src/worker.js",
    "dev:api": "nodemon src/api.js",
    "dev:worker": "nodemon src/worker.js",
    "dev": "concurrently \"npm:dev:api\" \"npm:dev:worker\"",
    "db:migrate": "npx prisma migrate dev --name init",
    "db:studio": "npx prisma studio"
  }
  // ... rest of package.json
}

const messagebird = require('messagebird');
const dotenv = require('dotenv');

dotenv.config(); // Load .env variables

const apiKey = process.env.MESSAGEBIRD_API_KEY;
const originator = process.env.MESSAGEBIRD_ORIGINATOR;

if (!apiKey) {
    console.error('ERROR: MESSAGEBIRD_API_KEY environment variable not set.');
    process.exit(1); // Exit if key is missing
}
if (!originator) {
    console.warn('WARN: MESSAGEBIRD_ORIGINATOR environment variable not set. Using default or MessageBird assigned number might occur.');
    // Depending on MessageBird's behavior, this might still work if you have a default number,
    // but explicitly setting it is highly recommended.
}

let messagebirdClient;
try {
    // Initialize the MessageBird client with your API key
    messagebirdClient = messagebird(apiKey);
    console.log('MessageBird SDK initialized successfully.');
} catch (error) {
    console.error('ERROR: Failed to initialize MessageBird SDK:', error);
    process.exit(1);
}

/**
 * Sends a single SMS message using the MessageBird API.
 * 
 * @param {string} recipient - The recipient's phone number (E.164 format, e.g., +12025550147).
 * @param {string} body - The text message content.
 * @returns {Promise<object&gt;} - A promise that resolves with the MessageBird API response.
 * @throws {Error} - Throws an error if the API call fails.
 */
const sendSms = async (recipient, body) =&gt; {
    if (!messagebirdClient) {
        throw new Error('MessageBird SDK is not initialized.');
    }
    if (!recipient || !body) {
        throw new Error('Recipient and body are required for sending SMS.');
    }

    const params = {
        originator: originator, // Use the configured originator
        recipients: [recipient], // API expects an array
        body: body,
        // Add more parameters if needed, e.g., scheduledAt, reportUrl
        // See: https://developers.messagebird.com/api/sms-messaging/#send-outbound-sms
    };

    console.log(`Attempting to send SMS to ${recipient} via MessageBird…`);

    // Using Promise wrapper for the callback-based API
    return new Promise((resolve, reject) =&gt; {
        messagebirdClient.messages.create(params, (err, response) =&gt; {
            if (err) {
                // Log detailed error information from MessageBird
                console.error(`ERROR sending SMS to ${recipient}:`, JSON.stringify(err, null, 2));
                // Reject the promise with a structured error
                reject({
                    message: `Failed to send SMS via MessageBird to ${recipient}`,
                    details: err.errors || [err] // Normalize error structure
                });
            } else {
                console.log(`SUCCESS: MessageBird API response for ${recipient}:`, JSON.stringify(response, null, 2));
                // Potentially check response status here if needed
                resolve(response); // Resolve with the successful response
            }
        });
    });
};

module.exports = {
    sendSms,
    messagebirdClient // Export client if needed elsewhere (e.g., for balance checks)
};

const Redis = require('ioredis');
const dotenv = require('dotenv');

dotenv.config();

const redisConfig = {
    host: process.env.REDIS_HOST || '127.0.0.1',
    port: process.env.REDIS_PORT || 6379,
    // password: process.env.REDIS_PASSWORD, // Uncomment if needed
    maxRetriesPerRequest: null, // Recommended by BullMQ docs
};

// Support for Redis URL connection string
const redisUrl = process.env.REDIS_URL;
let connection;

if (redisUrl) {
    connection = new Redis(redisUrl, { maxRetriesPerRequest: null });
     console.log(`Connecting to Redis using URL: ${redisUrl.split('@')[1] || redisUrl}`); // Avoid logging password
} else {
    connection = new Redis(redisConfig);
    console.log(`Connecting to Redis at ${redisConfig.host}:${redisConfig.port}`);
}

connection.on('connect', () => {
    console.log('Successfully connected to Redis.');
});

connection.on('error', (err) => {
    console.error('Redis connection error:', err);
    // Consider exiting the process if Redis connection is critical and fails initially
    // process.exit(1);
});

module.exports = connection;

const { Queue } = require('bullmq');
const redisConnection = require('../config/redis');
const dotenv = require('dotenv');

dotenv.config();

const QUEUE_NAME = process.env.QUEUE_NAME || 'sms_sending_queue';

// Create a new Queue instance
// Reuse the connection object
const smsQueue = new Queue(QUEUE_NAME, {
    connection: redisConnection,
    defaultJobOptions: {
        attempts: 3, // Number of times to retry a failed job
        backoff: {
            type: 'exponential', // Exponential backoff strategy
            delay: 1000, // Initial delay in ms (1 second)
        },
        removeOnComplete: true, // Automatically remove job after completion
        removeOnFail: false,     // Keep failed jobs for inspection (set to true or a number of jobs to keep if preferred)
    },
});

console.log(`BullMQ queue "${QUEUE_NAME}" initialized.`);

// Optional: Listen for queue events globally if needed (e.g., for monitoring)
smsQueue.on('error', (error) => {
  console.error(`Queue "${QUEUE_NAME}" error:`, error);
});

smsQueue.on('waiting', (jobId) => {
  // console.log(`Job ${jobId} is waiting in queue "${QUEUE_NAME}".`);
});

smsQueue.on('active', (job) => {
  // console.log(`Job ${job.id} has started processing in queue "${QUEUE_NAME}".`);
});

smsQueue.on('completed', (job, result) => {
 // console.log(`Job ${job.id} completed successfully in queue "${QUEUE_NAME}". Result:`, result);
});

smsQueue.on('failed', (job, err) => {
  console.error(`Job ${job.id} failed in queue "${QUEUE_NAME}". Attempt ${job.attemptsMade}/${job.opts.attempts}. Error:`, err);
});

/**
 * Adds a job to send a single SMS message to the queue.
 * @param {string} recipient - The recipient's phone number.
 * @param {string} body - The message content.
 * @returns {Promise<import('bullmq').Job&gt;} - A promise that resolves with the added job.
 */
const addSmsJob = async (recipient, body) =&gt; {
    if (!recipient || !body) {
        throw new Error('Recipient and body are required to add SMS job.');
    }
    console.log(`Adding SMS job to queue for recipient: ${recipient}`);
    // Job data should be serializable (plain objects, strings, numbers)
    const jobData = { recipient, body };
    // Job name is optional but useful for categorizing/identifying jobs
    const jobName = 'send-single-sms';

    try {
        const job = await smsQueue.add(jobName, jobData);
        console.log(`Job ${job.id} added to queue "${QUEUE_NAME}" for recipient ${recipient}.`);
        return job;
    } catch (error) {
        console.error(`Error adding job to queue "${QUEUE_NAME}":`, error);
        throw error; // Re-throw to be handled by the caller (e.g., API route)
    }
};

module.exports = {
    smsQueue,
    addSmsJob,
    QUEUE_NAME
};

const { Worker } = require('bullmq');
const redisConnection = require('./config/redis');
const { QUEUE_NAME } = require('./jobs/smsQueue');
const { sendSms } = require('./services/messagebirdService');
// Optional: Prisma client for database updates
// const { PrismaClient } = require('@prisma/client');
// const prisma = new PrismaClient();

console.log(`Starting worker for queue "${QUEUE_NAME}"…`);

// Define the processing function for jobs
const processSmsJob = async (job) => {
    const { recipient, body } = job.data; // Extract data added by addSmsJob
    console.log(`Processing job ${job.id} (Attempt ${job.attemptsMade + 1}/${job.opts.attempts}): Send SMS to ${recipient}`);

    try {
        // Optional: Update database status to 'processing' before sending
        // await prisma.messageLog.update({ where: { jobId: job.id }, data: { status: 'processing' } });

        const result = await sendSms(recipient, body);

        // Optional: Update database status to 'sent' on success
        // await prisma.messageLog.update({ where: { jobId: job.id }, data: { status: 'sent', messageBirdId: result.id /* Store MB ID if available */ } });

        console.log(`Job ${job.id} completed successfully for ${recipient}. MessageBird Response ID: ${result?.id || 'N/A'}`);
        return { messageBirdId: result?.id, status: 'sent' }; // Return value stored if job completes

    } catch (error) {
        console.error(`Job ${job.id} failed for recipient ${recipient}:`, error.message || error);

        // Optional: Update database status to 'failed'
        // await prisma.messageLog.update({
        //     where: { jobId: job.id },
        //     data: { status: 'failed', failureReason: JSON.stringify(error.details || error.message) }
        // });

        // IMPORTANT: Throw the error again to signal BullMQ that the job failed
        // This allows BullMQ to handle retries based on the queue's configuration.
        throw error;
    }
};

// Create a new Worker instance
const worker = new Worker(QUEUE_NAME, processSmsJob, {
    connection: redisConnection,
    concurrency: 5, // Process up to 5 jobs concurrently (adjust based on resources/API limits)
    limiter: {      // Optional: Rate limiting to avoid hitting MessageBird limits
        max: 10,    // Max 10 jobs
        duration: 1000 // per 1000 ms (1 second) → Adjust based on MessageBird's rate limits for your account/channel
    }
});

// --- Worker Event Listeners (Similar to Queue, but specific to this worker instance) ---
worker.on('completed', (job, returnValue) => {
    // console.log(`Worker completed job ${job.id} for queue "${QUEUE_NAME}". Return value:`, returnValue);
});

worker.on('failed', (job, err) => {
    console.error(`Worker failed job ${job.id} for queue "${QUEUE_NAME}" after ${job.attemptsMade} attempts. Error:`, err.message || err);
});

worker.on('error', (err) => {
    console.error(`Worker error for queue "${QUEUE_NAME}":`, err);
});

worker.on('active', (job) => {
    // console.log(`Worker started processing job ${job.id} for queue "${QUEUE_NAME}".`);
});

worker.on('progress', (job, progress) => {
     // If your job reports progress
     // console.log(`Worker progress for job ${job.id}:`, progress);
});

console.log(`Worker listening for jobs on queue "${QUEUE_NAME}" with concurrency ${worker.opts.concurrency}.`);

// Graceful shutdown
const gracefulShutdown = async (signal) => {
    console.log(`\nReceived ${signal}. Closing worker and Redis connection…`);
    await worker.close();
    await redisConnection.quit();
    // Optional: Close database connection
    // await prisma.$disconnect();
    console.log('Worker and connections closed gracefully.');
    process.exit(0);
};

process.on('SIGTERM', () => gracefulShutdown('SIGTERM'));
process.on('SIGINT', () => gracefulShutdown('SIGINT')); // Catches Ctrl+C

const Fastify = require('fastify');
const dotenv = require('dotenv');
const { addSmsJob, QUEUE_NAME } = require('./jobs/smsQueue');
// Optional: Prisma client for database interaction
// const { PrismaClient } = require('@prisma/client');
// const prisma = new PrismaClient();

dotenv.config();

const server = Fastify({
    logger: true // Enable built-in Pino logger (good for production)
});

// --- Request Validation Schema ---
const bulkSmsSchema = {
    body: {
        type: 'object',
        required: ['recipients', 'body'],
        properties: {
            recipients: {
                type: 'array',
                minItems: 1,
                items: {
                    type: 'string',
                    // Basic E.164 format regex (ensures +country code and digits, matches entire string)
                    pattern: '^\\+[1-9]\\d{1,14}$'
                },
                description: 'An array of recipient phone numbers in E.164 format (e.g., +12025550147).'
            },
            body: {
                type: 'string',
                minLength: 1,
                maxLength: 1600, // Allow longer messages (MessageBird handles splitting)
                description: 'The text content of the SMS message.'
            },
            // Optional: Add a correlation ID for tracking batches
            correlationId: {
                type: 'string',
                description: 'Optional identifier to correlate this batch request.'
            }
        },
        additionalProperties: false
    },
    response: { // Define expected success response structure
        202: { // Use 202 Accepted as processing is asynchronous
            type: 'object',
            properties: {
                message: { type: 'string' },
                jobCount: { type: 'integer' },
                correlationId: { type: 'string', nullable: true }, // Reflect back correlationId if provided
                queueName: { type: 'string' }
            }
        }
    }
};

// --- API Route ---
server.post('/send-bulk', { schema: bulkSmsSchema }, async (request, reply) => {
    const { recipients, body, correlationId } = request.body;
    const addedJobs = [];
    const failedAdds = [];

    request.log.info(`Received bulk SMS request. CorrelationID: ${correlationId || 'N/A'}, Recipients: ${recipients.length}, Body: "${body.substring(0, 50)}…"`);

    try {
        // (Optional) Record the initial batch request in the database if needed

        // Add a job to the queue for each recipient
        for (const recipient of recipients) {
            try {
                // Optional: Create initial DB record before adding job
                // const dbRecord = await prisma.messageLog.create({
                //    data: {
                //        recipient: recipient,
                //        body: body,
                //        status: 'pending', // Initial status
                //        correlationId: correlationId,
                //        // jobId will be added after job creation if needed
                //    }
                // });

                const job = await addSmsJob(recipient, body);
                addedJobs.push({ recipient: recipient, jobId: job.id });

                // Optional: Update DB record with the jobId
                // await prisma.messageLog.update({
                //    where: { id: dbRecord.id },
                //    data: { jobId: job.id }
                // });

            } catch (queueError) {
                request.log.error(`Failed to add job for recipient ${recipient}:`, queueError);
                failedAdds.push({ recipient: recipient, error: queueError.message });
                // Decide if you want to stop the whole batch or continue
                // For robustness, we continue adding other jobs here.
            }
        }

        if (failedAdds.length > 0) {
            // Partial success – some jobs added, some failed to queue
            // Log this clearly, maybe return a specific status or detailed error
             request.log.warn(`Partial success: ${addedJobs.length} jobs added, ${failedAdds.length} failed to queue.`);
             // Decide on the response – here we still acknowledge the queued ones
        }

        // Use 202 Accepted: Request received, processing will happen asynchronously.
        reply.code(202).send({
            message: `Accepted ${addedJobs.length} SMS messages for background processing.`,
            jobCount: addedJobs.length,
            correlationId: correlationId || null, // Echo back ID or null
            queueName: QUEUE_NAME
        });

    } catch (error) {
        request.log.error('Error processing bulk SMS request:', error);
        // General server error if something unexpected happens before/during looping
        reply.code(500).send({ error: 'Internal Server Error', message: 'Failed to process the bulk SMS request.' });
    }
});

// --- Health Check Route ---
server.get('/health', async (request, reply) => {
    // Basic health check – can be expanded (see Section 10.1)
    return { status: 'ok', timestamp: new Date().toISOString() };
});

// --- Start Server ---
const start = async () => {
    try {
        const port = process.env.API_PORT || 3000;
        const host = process.env.API_HOST || '0.0.0.0';
        await server.listen({ port: port, host: host });
        server.log.info(`API server listening on ${server.server.address().address}:${server.server.address().port}`);
    } catch (err) {
        server.log.error(err);
        // Optional: Disconnect Prisma on error
        // await prisma.$disconnect();
        process.exit(1);
    }
};

start();