SMS Marketing Campaigns with Twilio & Fastify Node.js: Step-by-Step Tutorial (2025)

graph LR
    A[User/Admin via Frontend/API Client] -- Manages Campaigns/Subscribers --> B(Fastify API);
    B -- Sends SMS via Twilio SDK --> C(Twilio Messaging API);
    C -- Sends SMS --> D(End User Phone);
    C -- Sends Status Webhook --> B;
    B -- Stores/Retrieves Data --> E(PostgreSQL Database w/ Prisma);
    D -- Sends Opt-Out Reply (e.g., STOP) --> C;
    C -- Sends Opt-Out Webhook --> B;

    style B fill:#f9f,stroke:#333,stroke-width:2px
    style C fill:#ccf,stroke:#333,stroke-width:2px
    style E fill:#cfc,stroke:#333,stroke-width:2px

mkdir fastify-twilio-sms-campaigns
cd fastify-twilio-sms-campaigns
npm init -y
# Set package type to module for ES Module syntax
npm pkg set type=module

npm install fastify twilio prisma @prisma/client dotenv pino-pretty @fastify/formbody @fastify/rate-limit @fastify/helmet
npm install --save-dev prisma supertest jest # Or vitest

npx prisma init --datasource-provider postgresql

# .env

# Database Connection (replace with your actual connection string)
# Format: postgresql://USER:PASSWORD@HOST:PORT/DATABASE?schema=public
DATABASE_URL=""postgresql://user:password@localhost:5432/sms_campaigns?schema=public""

# Twilio Credentials
TWILIO_ACCOUNT_SID=""ACxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"" # Find on Twilio Console Dashboard
TWILIO_AUTH_TOKEN=""your_auth_token""             # Find on Twilio Console Dashboard
TWILIO_PHONE_NUMBER=""+15551234567""             # Your Twilio SMS-capable number
# OR use a Messaging Service SID for better scaling/features (Recommended)
# TWILIO_MESSAGING_SERVICE_SID=""MGxxxxxxxxxxxxxxxxxxxxxxxxxxxxx""

# Application Settings
API_BASE_URL=""http://localhost:3000"" # Base URL for webhook callbacks (use ngrok URL during dev)
PORT=3000
NODE_ENV=""development"" # or production

# Optional: For simple API Key Auth example (NOT production recommended for storing keys)
# API_KEYS=""key1,key2,key3""
# Optional: For Sentry integration
# SENTRY_DSN=""""

mkdir -p src/routes src/services src/controllers src/schemas src/db tests/api .github/workflows
touch src/config.js src/server.js src/app.js src/db/prisma.js Dockerfile fly.toml .github/workflows/deploy.yml .gitignore
echo "".env"" >> .gitignore
echo ""node_modules"" >> .gitignore
echo ""/dist"" >> .gitignore # If using a build step

// src/config.js
import dotenv from 'dotenv';

dotenv.config();

// Basic check for essential variables
if (!process.env.DATABASE_URL || !process.env.TWILIO_ACCOUNT_SID || !process.env.TWILIO_AUTH_TOKEN) {
    console.error(""FATAL ERROR: Required environment variables are missing (DATABASE_URL, TWILIO_ACCOUNT_SID, TWILIO_AUTH_TOKEN)."");
    process.exit(1);
}
if (!process.env.TWILIO_PHONE_NUMBER && !process.env.TWILIO_MESSAGING_SERVICE_SID) {
     console.error(""FATAL ERROR: Either TWILIO_PHONE_NUMBER or TWILIO_MESSAGING_SERVICE_SID must be set."");
    process.exit(1);
}

export default {
    port: parseInt(process.env.PORT || '3000', 10),
    host: process.env.HOST || '0.0.0.0',
    nodeEnv: process.env.NODE_ENV || 'development',
    databaseUrl: process.env.DATABASE_URL,
    twilio: {
        accountSid: process.env.TWILIO_ACCOUNT_SID,
        authToken: process.env.TWILIO_AUTH_TOKEN,
        phoneNumber: process.env.TWILIO_PHONE_NUMBER,
        messagingServiceSid: process.env.TWILIO_MESSAGING_SERVICE_SID,
    },
    apiBaseUrl: process.env.API_BASE_URL || `http://localhost:${process.env.PORT || 3000}`,
    // Simple API Key Auth Example (Insecure storage - use secrets manager in prod)
    apiKeys: new Set((process.env.API_KEYS || '').split(',').filter(Boolean)),
    // Sentry DSN
    sentryDsn: process.env.SENTRY_DSN,
    // Rate Limit Config
    rateLimit: {
        max: parseInt(process.env.RATE_LIMIT_MAX || '100', 10),
        timeWindow: process.env.RATE_LIMIT_WINDOW || '1 minute',
    }
};

// src/db/prisma.js
import { PrismaClient } from '@prisma/client';
import config from '../config.js'; // Import config to access NODE_ENV

const prisma = new PrismaClient({
    log: config.nodeEnv === 'development' ? ['query', 'info', 'warn', 'error'] : ['warn', 'error'],
});

export default prisma;

// src/server.js
import Fastify from 'fastify';
import formbody from '@fastify/formbody';
import rateLimit from '@fastify/rate-limit';
import helmet from '@fastify/helmet';
import * as Sentry from '@sentry/node';
import { ProfilingIntegration } from '@sentry/profiling-node';
import config from './config.js';
import prisma from './db/prisma.js'; // Import prisma for health check

// Import routes
import campaignRoutes from './routes/campaignRoutes.js';
import subscriberRoutes from './routes/subscriberRoutes.js';
import twilioWebhookRoutes from './routes/twilioWebhookRoutes.js';

// Initialize Sentry (if DSN is provided and in production)
if (config.sentryDsn && config.nodeEnv === 'production') {
    Sentry.init({
        dsn: config.sentryDsn,
        integrations: [new ProfilingIntegration()],
        tracesSampleRate: 1.0, // Adjust in production
        profilesSampleRate: 1.0, // Adjust in production
        environment: config.nodeEnv,
        // release: 'my-project-name@1.0.0', // Optional: Set release version
    });
    console.log('Sentry initialized for production.');
}

export function buildServer(options = {}) {
    const fastify = Fastify({
        logger: config.nodeEnv === 'development'
            ? {
                transport: {
                    target: 'pino-pretty',
                    options: {
                        translateTime: 'HH:MM:ss Z',
                        ignore: 'pid,hostname',
                    },
                },
            }
            : { level: 'info' }, // Use default JSON logger in production
        ...options,
    });

    // Register Sentry Fastify plugin (must be done early)
    if (config.sentryDsn && config.nodeEnv === 'production') {
        fastify.register(import('@sentry/fastify')).after(() => {
            // Custom error handler integrated with Sentry
            fastify.setErrorHandler(async (error, request, reply) => {
                // Log error locally regardless
                 request.log.error(error);
                // Send error to Sentry
                Sentry.captureException(error);
                 // Use Fastify's default handling or customize response
                if (!reply.sent) {
                    const statusCode = error.statusCode && error.statusCode >= 400 ? error.statusCode : 500;
                    const message = statusCode >= 500 ? 'Internal Server Error' : error.message;
                    reply.code(statusCode).send({ message });
                }
            });
        });
    }

    // Register essential plugins
    fastify.register(helmet); // Security headers
    fastify.register(formbody); // Parse form bodies (for Twilio webhooks)
    fastify.register(rateLimit, { // Rate limiting
        max: config.rateLimit.max,
        timeWindow: config.rateLimit.timeWindow,
    });

    // --- Authentication Hook (Simple API Key Example) ---
    // WARNING: Storing keys directly in env vars is NOT secure for production.
    // Use a secrets manager or hashed keys in a database.
    fastify.addHook('onRequest', async (request, reply) => {
        // Exclude webhooks and health check from this simple auth
        if (request.url.startsWith('/api/v1/webhooks') || request.url === '/health') {
            return;
        }
        // Skip auth if no keys are configured (allows open access during initial dev)
        if (config.apiKeys.size === 0 && config.nodeEnv !== 'production') {
             request.log.warn('API Key authentication skipped (no keys configured)');
            return;
        }
         if (config.apiKeys.size === 0 && config.nodeEnv === 'production') {
             request.log.error('CRITICAL: API Key authentication mandatory in production, but no keys configured!');
             reply.code(500).send({ message: 'Server configuration error' });
             return reply; // Prevent further processing by returning reply
         }

        const apiKey = request.headers['x-api-key'];
        if (!apiKey || !config.apiKeys.has(apiKey)) {
            request.log.warn(`Unauthorized API access attempt. Path: ${request.url}, Key Provided: ${!!apiKey}`);
            reply.code(401).send({ message: 'Unauthorized' });
            return reply; // Stop processing by returning reply
        }
        // Optional: Log successful validation (can be noisy)
        // request.log.info(`API Key validated for request: ${request.url}`);
    });
    // --- End Authentication Hook ---

    // Register API routes
    fastify.register(campaignRoutes, { prefix: '/api/v1/campaigns' });
    fastify.register(subscriberRoutes, { prefix: '/api/v1/subscribers' });
    fastify.register(twilioWebhookRoutes, { prefix: '/api/v1/webhooks/twilio' });

    // Health check endpoint
    fastify.get('/health', async (request, reply) => {
        try {
            await prisma.$queryRaw`SELECT 1`; // Check DB connection
            return { status: 'ok', timestamp: new Date().toISOString(), checks: { database: 'ok' } };
        } catch (dbError) {
            request.log.error({ err: dbError }, 'Health check failed: Database connection error.');
            reply.code(503); // Service Unavailable
            return { status: 'error', timestamp: new Date().toISOString(), checks: { database: 'error' }, error: 'Database connection failed' };
        }
    });

    return fastify;
}

// src/app.js
import { buildServer } from './server.js';
import config from './config.js';
import prisma from './db/prisma.js';

const server = buildServer();

const start = async () => {
    try {
        // Test DB connection on startup
        await prisma.$connect();
        server.log.info('Database connection successful.');

        await server.listen({ port: config.port, host: config.host });
        // Note: Fastify logs listening address automatically on successful listen

    } catch (err) {
        server.log.error(err, 'Failed to start server or connect to database');
        await prisma.$disconnect().catch(e => server.log.error(e, 'Failed to disconnect prisma'));
        process.exit(1);
    }
};

// Graceful shutdown
const shutdown = async (signal) => {
    server.log.info(`Received ${signal}. Shutting down gracefully...`);
    try {
        await server.close();
        server.log.info('HTTP server closed.');
        await prisma.$disconnect();
        server.log.info('Database connection closed.');
        process.exit(0);
    } catch (err) {
        server.log.error(err, 'Error during graceful shutdown');
        process.exit(1);
    }
};

// Handle termination signals
process.on('SIGINT', () => shutdown('SIGINT'));
process.on('SIGTERM', () => shutdown('SIGTERM'));

// Handle unhandled promise rejections
process.on('unhandledRejection', (reason, promise) => {
  server.log.error({ reason, promise }, 'Unhandled Rejection at Promise');
  // Consider whether to exit or let Sentry capture it
});

// Handle uncaught exceptions
process.on('uncaughtException', (error) => {
  server.log.fatal(error, 'Uncaught Exception');
  // It's generally recommended to exit after an uncaught exception
  shutdown('uncaughtException').finally(() => process.exit(1));
});

start();

// package.json (add within ""scripts"")
""scripts"": {
  ""start"": ""node src/app.js"",
  ""dev"": ""node --watch src/app.js"",
  ""test"": ""jest"",
  ""test:watch"": ""jest --watch"",
  ""test:coverage"": ""jest --coverage"",
  ""db:migrate:dev"": ""npx prisma migrate dev"",
  ""db:migrate:deploy"": ""npx prisma migrate deploy"",
  ""db:generate"": ""npx prisma generate"",
  ""db:studio"": ""npx prisma studio""
},
// Add Jest config if needed (or use vitest.config.js)
""jest"": {
  ""testEnvironment"": ""node"",
  ""coverageProvider"": ""v8""
}

// prisma/schema.prisma

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

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

// Note on IDs: Using CUIDs (default) which are collision-resistant unique IDs.
// UUIDs are another common alternative if preferred.

model Subscriber {
  id        String   @id @default(cuid())
  phone     String   @unique // E.164 format recommended (e.g., +15551234567)
  firstName String?
  lastName  String?
  status    String   @default(""active"") // e.g., active, unsubscribed, bounced
  createdAt DateTime @default(now())
  updatedAt DateTime @updatedAt

  // Many-to-many relationship: a subscriber can belong to multiple campaigns
  campaigns Campaign[] @relation(""CampaignSubscribers"")
  messageLogs MessageLog[] // One-to-many: a subscriber receives multiple messages
}

model Campaign {
  id          String   @id @default(cuid())
  name        String
  messageBody String   @db.Text // Use Text for potentially long messages
  status      String   @default(""draft"") // e.g., draft, scheduled, sending, sent, failed
  scheduledAt DateTime? // Optional: for scheduled campaigns
  sentAt      DateTime? // When the last message batch was initiated
  createdAt   DateTime @default(now())
  updatedAt   DateTime @updatedAt

  // Many-to-many relationship
  subscribers Subscriber[] @relation(""CampaignSubscribers"")
  messageLogs MessageLog[] // One-to-many: a campaign generates multiple messages
}

model MessageLog {
  id             String   @id @default(cuid())
  subscriberId   String
  campaignId     String?  // Can be null if message not part of campaign (e.g., direct send)
  twilioSid      String   @unique // Twilio Message SID (critical for status updates)
  status         String   // e.g., queued, sending, sent, delivered, undelivered, failed (matches Twilio statuses)
  errorCode      Int?     // Twilio error code if failed/undelivered
  errorMessage   String?  @db.Text
  sentAt         DateTime @default(now()) // When the API call was made
  lastStatusAt   DateTime @updatedAt // When the last status update was received from Twilio

  subscriber Subscriber @relation(fields: [subscriberId], references: [id], onDelete: Cascade) // If subscriber deleted, delete logs
  campaign   Campaign?  @relation(fields: [campaignId], references: [id], onDelete: SetNull) // If campaign deleted, keep logs but remove link

  // Indexes for common query patterns
  @@index([subscriberId])
  @@index([campaignId])
  @@index([status])
  @@index([sentAt])
}

# Create a new migration file based on schema changes
# Make sure your .env file has the correct DATABASE_URL
npx prisma migrate dev --name initial_schema

# If prompted, enter a name for the migration (e.g., ""initial_schema"")

// src/services/twilioService.js
import twilio from 'twilio';
import pRetry from 'p-retry';
import config from '../config.js';
import prisma from '../db/prisma.js';

const client = twilio(config.twilio.accountSid, config.twilio.authToken);
const sender = config.twilio.messagingServiceSid
    ? { messagingServiceSid: config.twilio.messagingServiceSid }
    : { from: config.twilio.phoneNumber };

if (!sender.messagingServiceSid && !sender.from) {
    // This check is also in config.js, but good to have redundancy here
    console.error('CRITICAL: TWILIO_PHONE_NUMBER or TWILIO_MESSAGING_SERVICE_SID must be configured.');
    // Consider throwing an error instead of exiting if this service is loaded conditionally
    process.exit(1);
}

/**
 * Sends a single SMS message using Twilio with retry logic.
 * @param {string} to - Recipient phone number (E.164 format).
 * @param {string} body - Message content.
 * @param {string} [campaignId] - Optional campaign ID for logging.
 * @param {string} subscriberId - Subscriber ID for logging.
 * @returns {Promise<object>} - Twilio message object (after successful send or final retry failure).
 */
export async function sendSms(to, body, campaignId, subscriberId) {
    const statusCallbackUrl = `${config.apiBaseUrl}/api/v1/webhooks/twilio/status`;

    const runSend = async () => {
        // This code runs potentially multiple times due to pRetry
        console.log(`Attempting to send SMS via Twilio to ${to}`);
        const message = await client.messages.create({
            ...sender, // Use 'from' or 'messagingServiceSid'
            to: to,
            body: body,
            statusCallback: statusCallbackUrl, // URL Twilio posts status updates to
        });
        console.log(`SMS successfully queued via Twilio for ${to}. SID: ${message.sid}`);
        return message; // Return the successful message object
    };

    try {
        const message = await pRetry(runSend, {
            retries: 3, // Total attempts = 4 (1 initial + 3 retries)
            minTimeout: 500, // Start with 500ms delay
            factor: 2, // Double delay each time
            randomize: true, // Add jitter to avoid thundering herd
            onFailedAttempt: error => {
                console.warn(`Twilio send attempt ${error.attemptNumber} failed for ${to}. Retries left: ${error.retriesLeft}. Error: ${error.message} (Status: ${error.status}, Code: ${error.code})`);
                // Optional: Decide whether to retry based on error code
                // e.g., don't retry on 400 Bad Request (like invalid number)
                if (error.status === 400) {
                     console.error(`Permanent error sending to ${to}, stopping retries. Code: ${error.code}`);
                     throw error; // Prevent further retries for non-retryable errors
                }
            }
        });

        // Log initial 'queued' status AFTER successful API call (potentially after retries)
        await prisma.messageLog.create({
            data: {
                subscriberId: subscriberId,
                campaignId: campaignId,
                twilioSid: message.sid,
                status: message.status, // Initial status from Twilio (e.g., 'queued', 'accepted')
                errorCode: message.errorCode,
                errorMessage: message.errorMessage,
            },
        });

        return message; // Return the successful Twilio message object

    } catch (error) {
        console.error(`Failed to send SMS via Twilio to ${to} after all retries:`, error.message);
        // Log failed attempt only after all retries are exhausted
        await prisma.messageLog.create({
             data: {
                 subscriberId: subscriberId,
                 campaignId: campaignId,
                 twilioSid: `failed-${Date.now()}-${subscriberId}`, // Create a unique placeholder SID
                 status: 'failed', // Mark as failed immediately in our log
                 errorCode: error.code || null, // Twilio error code if available from last attempt
                 errorMessage: `Failed after retries: ${error.message}`,
             },
         }).catch(logError => console.error(""Failed to log final send error:"", logError));

        throw error; // Re-throw the final error to be handled by the caller (e.g., batch processor)
    }
}

/**
 * Sends messages in batches with concurrency control.
 * @param {Array<{to: string, body: string, campaignId: string, subscriberId: string}>} messages
 * @param {number} [concurrency=5] - Number of messages to send in parallel.
 * @returns {Promise<{success: Array<{sid: string, to: string}>, errors: Array<{to: string, error: string}>}>}
 */
export async function sendBatchSms(messages, concurrency = 5) {
    const results = { success: [], errors: [] };
    const queue = [...messages]; // Clone the array to avoid modifying the original

    // Use Promise.allLimit style concurrency (simple implementation)
    const executing = [];
    while (queue.length > 0 || executing.length > 0) {
        while (executing.length < concurrency && queue.length > 0) {
            const msg = queue.shift();
            if (!msg) continue;

            const promise = sendSms(msg.to, msg.body, msg.campaignId, msg.subscriberId)
                .then(result => {
                    results.success.push({ sid: result.sid, to: msg.to });
                })
                .catch(error => {
                    // Error is already logged within sendSms after retries fail
                    results.errors.push({ to: msg.to, error: error.message });
                })
                .finally(() => {
                    // Remove the promise from the executing list when done
                    const index = executing.indexOf(promise);
                    if (index > -1) {
                        executing.splice(index, 1);
                    }
                });

            executing.push(promise);
        }
        // Wait for at least one promise to settle if the queue is empty but tasks are running
        if (queue.length === 0 && executing.length > 0) {
             await Promise.race(executing);
        } else if (executing.length === concurrency) {
            // Wait for one promise to settle before adding more if concurrency limit reached
            await Promise.race(executing);
        }
    }

    // Ensure all promises complete (though they remove themselves, this is a safeguard)
    await Promise.all(executing);

    console.log(`Batch send processing complete. Success attempts: ${results.success.length}, Failed attempts (after retries): ${results.errors.length}`);
    return results;
}

/**
 * Handles incoming status updates from Twilio webhook.
 * @param {object} statusData - Data from Twilio webhook request body.
 */
export async function handleStatusUpdate(statusData) {
    const { MessageSid, MessageStatus, ErrorCode, ErrorMessage } = statusData;

    if (!MessageSid) {
        console.warn('Received status update without MessageSid:', statusData);
        return; // Cannot process without SID
    }

    console.log(`Received status update for SID ${MessageSid}: ${MessageStatus}`);

    try {
        await prisma.messageLog.update({
            where: { twilioSid: MessageSid },
            data: {
                status: MessageStatus, // Update with the status from Twilio
                errorCode: ErrorCode ? parseInt(ErrorCode, 10) : null,
                errorMessage: ErrorMessage || null,
                // lastStatusAt is updated automatically by @updatedAt
            },
        });
        console.log(`Updated status for message ${MessageSid} to ${MessageStatus}`);

        // Optional: Update subscriber status based on final delivery failure codes
        if (MessageStatus === 'undelivered' || MessageStatus === 'failed') {
            const permanentFailureCodes = [
                21211, // Invalid 'To' Phone Number
                21610, // Attempt to send to unsubscribed recipient
                21614, // To number is not SMS capable
                30003, // Unreachable destination handset
                30005, // Unknown destination handset
                30006, // Landline or unreachable carrier
            ];
            if (ErrorCode && permanentFailureCodes.includes(parseInt(ErrorCode, 10))) {
                 console.warn(`Permanent failure code ${ErrorCode} for SID ${MessageSid}. Marking subscriber potentially.`);
                 // Find the subscriber associated with this message log
                 const messageLog = await prisma.messageLog.findUnique({
                     where: { twilioSid: MessageSid },
                     select: { subscriberId: true }
                 });
                 if (messageLog) {
                     await prisma.subscriber.update({
                         where: { id: messageLog.subscriberId },
                         data: { status: 'bounced' } // Or a more specific status
                     });
                     console.log(`Marked subscriber ${messageLog.subscriberId} as bounced due to error code ${ErrorCode}.`);
                 }
            }
        }

    } catch (error) {
        // Handle case where message log might not exist (e.g., race condition, data issue)
        if (error.code === 'P2025') { // Prisma code for record not found
             console.error(`MessageLog not found for Twilio SID: ${MessageSid}. Status update ignored.`);
        } else {
            console.error(`Error updating message log for SID ${MessageSid}:`, error);
            // Potentially re-queue or alert on persistent errors
        }
        // Do not throw here, as Twilio expects a 2xx response
    }
}

/**
 * Handles incoming opt-out messages (e.g., STOP) from Twilio webhook.
 * This relies on Twilio's Advanced Opt-Out feature configured on the Messaging Service.
 * @param {object} messageData - Data from Twilio webhook request body.
 */
export async function handleOptOut(messageData) {
    const { From, Body } = messageData;

    if (!From) {
        console.warn('Received opt-out message without From number:', messageData);
        return;
    }

    // Twilio's Advanced Opt-Out automatically handles STOP/START/UNSTOP keywords
    // We just need to update our subscriber status in the database
    const normalizedBody = Body?.trim().toUpperCase();
    const optOutKeywords = ['STOP', 'STOPALL', 'UNSUBSCRIBE', 'CANCEL', 'END', 'QUIT'];
    const optInKeywords = ['START', 'UNSTOP'];

    try {
        if (optOutKeywords.includes(normalizedBody)) {
            // Mark subscriber as unsubscribed
            await prisma.subscriber.update({
                where: { phone: From },
                data: { status: 'unsubscribed' },
            });
            console.log(`Subscriber ${From} opted out (keyword: ${normalizedBody})`);
        } else if (optInKeywords.includes(normalizedBody)) {
            // Reactivate subscriber
            await prisma.subscriber.update({
                where: { phone: From },
                data: { status: 'active' },
            });
            console.log(`Subscriber ${From} opted back in (keyword: ${normalizedBody})`);
        }
    } catch (error) {
        if (error.code === 'P2025') {
            console.warn(`Received opt-out/opt-in from unknown number: ${From}`);
        } else {
            console.error(`Error handling opt-out for ${From}:`, error);
        }
        // Do not throw here, as Twilio expects a 2xx response
    }
}