Plivo Next.js SMS Marketing Campaigns: Complete Guide with BullMQ & Prisma

graph LR
    A[User Browser] -- HTTP Request --> B(Next.js Frontend);
    B -- API Call / Form Submit --> C(Next.js API Route);
    C -- Add Job --> D(Redis / BullMQ);
    C -- Write Campaign/List Data --> E(PostgreSQL DB / Prisma);
    F[BullMQ Worker Process] -- Process Job --> D;
    F -- Send SMS/MMS --> G(Plivo API);
    F -- Update Message Status --> E;
    G -- Status Callback (Optional) --> H(Next.js Webhook API Route);
    H -- Update Message Status --> E;

    style B fill:#f9f,stroke:#333,stroke-width:2px;
    style C fill:#ccf,stroke:#333,stroke-width:2px;
    style H fill:#ccf,stroke:#333,stroke-width:2px;
    style D fill:#f69,stroke:#333,stroke-width:2px;
    style F fill:#9cf,stroke:#333,stroke-width:2px;
    style E fill:#ccc,stroke:#333,stroke-width:2px;
    style G fill:#f90,stroke:#333,stroke-width:2px;

npx create-next-app@latest plivo-marketing-app
cd plivo-marketing-app

# Core dependencies
npm install plivo-node prisma @prisma/client bullmq zod date-fns ioredis

# Dev dependencies for Prisma/scripts
npm install -D @types/node typescript ts-node @faker-js/faker

# If using Tailwind
npm install -D tailwindcss postcss autoprefixer

npx prisma init --datasource-provider postgresql

# .env or .env.local

# Plivo Credentials
# Get these from your Plivo Console: https://console.plivo.com/dashboard/
PLIVO_AUTH_ID="YOUR_PLIVO_AUTH_ID"
PLIVO_AUTH_TOKEN="YOUR_PLIVO_AUTH_TOKEN"
PLIVO_SENDER_ID="YOUR_PLIVO_PHONE_NUMBER_OR_SENDER_ID" # Your Plivo sending number (e.g., +14155551212) or Alphanumeric Sender ID

# Database Connection (Prisma)
# Example for local Docker setup (see step 1.5)
DATABASE_URL="postgresql://user:password@localhost:5432/plivo_marketing?schema=public"

# Redis Connection (BullMQ)
# Example for local Docker setup (see step 1.5)
REDIS_URL="redis://localhost:6379"

# Application Settings
NEXT_PUBLIC_APP_URL="http://localhost:3000" # Base URL for callbacks, etc. (Update for production)
API_SECRET_KEY="YOUR_STRONG_RANDOM_SECRET_KEY" # For simple internal API protection

# docker-compose.yml
version: '3.8'
services:
  postgres:
    image: postgres:16  # PostgreSQL 16 is current stable version as of 2025
    restart: always
    environment:
      POSTGRES_DB: plivo_marketing
      POSTGRES_USER: user
      POSTGRES_PASSWORD: password
    ports:
      - "5432:5432"
    volumes:
      - postgres_data:/var/lib/postgresql/data

  redis:
    image: redis:7.4  # Redis 7.4 is current stable version as of 2025
    restart: always
    ports:
      - "6379:6379"
    volumes:
      - redis_data:/data

volumes:
  postgres_data:
  redis_data:

docker-compose up -d # Start in detached mode

plivo-marketing-app/
├── components/       # Reusable React components (e.g., forms, UI elements)
├── jobs/             # Recommended location for BullMQ worker logic (e.g., worker.ts)
├── lib/              # Shared utility functions (Prisma client, Plivo client, queue setup, etc.)
├── pages/
│   ├── api/          # Next.js API routes (our backend)
│   │   ├── campaigns.ts
│   │   └── webhooks/ # (Optional) Plivo callbacks
│   ├── _app.tsx      # Global App component (assuming TS)
│   ├── _document.tsx # Custom Document (assuming TS)
│   └── index.tsx     # Main dashboard page (assuming TS)
├── prisma/
│   ├── migrations/   # Database migration files
│   └── schema.prisma # Prisma schema definition
├── public/           # Static assets
├── scripts/          # Standalone scripts (e.g., seeding)
├── styles/           # CSS files
├── .env.local        # Environment variables (DO NOT COMMIT)
├── .gitignore
├── docker-compose.yml
├── next.config.js
├── package.json
└── README.md

// prisma/schema.prisma

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

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

model Campaign {
  id          String    @id @default(cuid())
  name        String
  message     String
  mediaUrls   String[] // Array of URLs for MMS
  status      CampaignStatus @default(PENDING)
  scheduledAt DateTime? // Optional: for scheduled campaigns
  createdAt   DateTime  @default(now())
  updatedAt   DateTime  @updatedAt
  recipients  Recipient[]
  messageLogs MessageLog[]
}

model Recipient {
  id               String      @id @default(cuid())
  phoneNumber      String      // Consider adding @@unique constraint if numbers should be unique globally or per campaign import
  campaignId       String
  campaign         Campaign    @relation(fields: [campaignId], references: [id], onDelete: Cascade)
  status           MessageStatus @default(PENDING) // Status for this specific recipient in the campaign
  plivoMessageUuid String?     @unique // Store Plivo's message ID if available
  createdAt        DateTime    @default(now())
  updatedAt        DateTime    @updatedAt
  messageLog       MessageLog? // Link to the specific log entry for this recipient

  @@index([campaignId])
  @@index([phoneNumber]) // Index phone number for opt-out checks
  @@unique([campaignId, phoneNumber]) // Ensure a number appears only once per campaign
}

// Optional but recommended: Log individual message attempts/statuses
model MessageLog {
  id               String        @id @default(cuid())
  campaignId       String
  campaign         Campaign      @relation(fields: [campaignId], references: [id], onDelete: Cascade)
  recipientId      String        @unique // One log per recipient attempt
  recipient        Recipient     @relation(fields: [recipientId], references: [id], onDelete: Cascade)
  plivoMessageUuid String?       @unique
  status           MessageStatus
  errorCode        String?       // Plivo error code, if any
  errorMessage     String?       // Plivo error message
  sentAt           DateTime?
  deliveredAt      DateTime?     // If using delivery reports
  createdAt        DateTime      @default(now())
  updatedAt        DateTime      @updatedAt

  @@index([campaignId])
  @@index([plivoMessageUuid])
}

// Recommended for Opt-Out handling
model OptOut {
  id          String   @id @default(cuid())
  phoneNumber String   @unique // Store opted-out numbers in E.164 format
  source      String?  // Optional: 'SMS', 'USER_REQUEST', etc.
  createdAt   DateTime @default(now())
}

enum CampaignStatus {
  PENDING     // Campaign created, not yet processing
  PROCESSING  // Jobs are being added to the queue
  QUEUED      // All jobs added, waiting for workers
  SENDING     // Workers are actively sending messages
  COMPLETED   // All messages attempted (success or fail)
  FAILED      // Major failure during processing/setup
  SCHEDULED   // Waiting for scheduled time
}

enum MessageStatus {
  PENDING     // Not yet sent
  QUEUED      // In the BullMQ queue
  SENT        // Successfully handed off to Plivo
  DELIVERED   // Confirmed delivery (requires webhooks)
  UNDELIVERED // Failed delivery (requires webhooks)
  FAILED      // Failed to send (API error, etc.)
  OPTED_OUT   // Recipient had opted out before send attempt
}

npx prisma migrate dev --name init

// lib/prisma.ts
import { PrismaClient } from '@prisma/client';

declare global {
  // allow global `var` declarations
  // eslint-disable-next-line no-var
  var prisma: PrismaClient | undefined;
}

export const prisma =
  global.prisma ||
  new PrismaClient({
    log: process.env.NODE_ENV === 'development' ? ['query', 'error', 'warn'] : ['error'],
  });

if (process.env.NODE_ENV !== 'production') {
  global.prisma = prisma;
}

export default prisma;

// lib/plivoClient.ts
import * as plivo from 'plivo';

let client: plivo.Client | null = null;

export function getPlivoClient(): plivo.Client {
  if (!client) {
    const authId = process.env.PLIVO_AUTH_ID;
    const authToken = process.env.PLIVO_AUTH_TOKEN;

    if (!authId || !authToken) {
      throw new Error(""Plivo Auth ID or Auth Token not configured in environment variables."");
    }
    client = new plivo.Client(authId, authToken);
  }
  return client;
}

// lib/queue.ts
import { Queue, Worker, Job } from 'bullmq';
import Redis from 'ioredis'; // ioredis is recommended by BullMQ

const connection = new Redis(process.env.REDIS_URL!, {
  maxRetriesPerRequest: null // Prevent Redis commands from failing after max retries
});

// Define the type for our job data
interface SmsJobData {
  recipientId: string;
  campaignId: string;
  to: string;
  from: string;
  text: string;
  mediaUrls?: string[];
}

const SMS_QUEUE_NAME = 'sms-campaign-queue';

// Create the queue instance
const smsQueue = new Queue<SmsJobData&gt;(SMS_QUEUE_NAME, { connection });

// Export the queue and connection for use elsewhere
export { smsQueue, connection, SMS_QUEUE_NAME };
export type { SmsJobData };

// --- Worker Setup (Run this in a separate process or script) ---

// **Recommendation:** For better separation of concerns and easier scaling,
// it's highly recommended to move the worker logic below into a separate file,
// for example, `jobs/worker.ts` or `scripts/worker.ts`.
// You would then update the `start:worker` script in `package.json`
// (see Section 12.2) to execute that specific file (e.g., `ts-node jobs/worker.ts`).
// The code below demonstrates keeping it in the same file for simplicity in this guide.

if (require.main === module) { // Only run worker logic if this file is executed directly
  console.log(`Starting SMS worker for queue: ${SMS_QUEUE_NAME}...`);

  const startWorker = async () =&gt; {
    // Dynamically import dependencies needed only by the worker
    const { getPlivoClient } = await import('./plivoClient');
    const { default: prisma } = await import('./prisma');

    const worker = new Worker<SmsJobData&gt;(
      SMS_QUEUE_NAME,
      async (job: Job<SmsJobData&gt;) =&gt; {
        console.log(`Processing job ${job.id} for recipient ${job.data.to}`);
        const { recipientId, campaignId, to, from, text, mediaUrls } = job.data;

        try {
          const client = getPlivoClient();
          const messageParams: plivo.MessageCreateParams = {
            src: from,
            dst: to,
            text: text,
          };

          // Add media URLs for MMS
          if (mediaUrls && mediaUrls.length &gt; 0) {
              messageParams.type = 'mms';
              messageParams.media_urls = mediaUrls;
              console.log(`Sending MMS to ${to} with media: ${mediaUrls.join(', ')}`);
          } else {
               console.log(`Sending SMS to ${to}`);
          }

          // Optional: Add status callback URL (Requires a webhook endpoint)
          // messageParams.url = `${process.env.NEXT_PUBLIC_APP_URL}/api/webhooks/plivo-status`;
          // messageParams.method = ""POST"";

          const response = await client.messages.create(messageParams);

          const plivoUuid = response.messageUuid && response.messageUuid.length &gt; 0 ? response.messageUuid[0] : null;
          console.log(`Message sent to ${to}, Plivo UUID: ${plivoUuid}`);

          // Update Recipient and create MessageLog on success
          await prisma.recipient.update({
            where: { id: recipientId },
            data: {
              status: 'SENT',
              plivoMessageUuid: plivoUuid,
            },
          });
          await prisma.messageLog.create({
              data: {
                  campaignId: campaignId,
                  recipientId: recipientId,
                  plivoMessageUuid: plivoUuid,
                  status: 'SENT',
                  sentAt: new Date(),
              }
          });

        } catch (error: any) {
          console.error(`Failed to send message to ${to} (Job ${job.id}):`, error);
          const errorCode = error.statusCode || 'UNKNOWN';
          const errorMessage = error.message || 'Failed to process job';

          // Update Recipient and create MessageLog on failure
           try {
              await prisma.recipient.update({
                where: { id: recipientId },
                data: { status: 'FAILED' },
              });
              await prisma.messageLog.create({
                data: {
                  campaignId: campaignId,
                  recipientId: recipientId,
                  status: 'FAILED',
                  errorCode: errorCode.toString(),
                  errorMessage: errorMessage,
                }
              });
           } catch (dbError) {
               console.error(`Failed to update database after sending failure for job ${job.id}:`, dbError);
           }

          // Important: Throw the error again so BullMQ knows the job failed
          // This enables retry logic if configured.
          throw error;
        }
      },
      {
        connection,
        concurrency: 5, // Process up to 5 jobs concurrently (adjust based on resources/Plivo limits)
        limiter: { // Optional: Rate limit outgoing requests to Plivo
          max: 10, // Max 10 jobs
          duration: 1000, // per second
        },
        attempts: 3, // Retry failed jobs up to 3 times
        backoff: {   // Exponential backoff strategy
          type: 'exponential',
          delay: 5000, // Initial delay 5 seconds
        },
      }
    );

    worker.on('completed', (job: Job<SmsJobData&gt;) =&gt; {
      console.log(`Job ${job.id} completed for ${job.data.to}`);
    });

    worker.on('failed', (job: Job<SmsJobData&gt; | undefined, err: Error) =&gt; {
        if (job) {
          console.error(`Job ${job.id} failed for ${job.data.to} after ${job.attemptsMade} attempts:`, err.message);
        } else {
          console.error(`A job failed with error:`, err.message);
        }
    });

     console.log(""SMS Worker started successfully."");

     const gracefulShutdown = async (signal: string) =&gt; {
         console.log(`${signal} signal received: closing worker gracefully.`);
         await worker.close();
         console.log('Worker closed.');
         process.exit(0);
     }

     process.on('SIGTERM', () =&gt; gracefulShutdown('SIGTERM'));
     process.on('SIGINT', () =&gt; gracefulShutdown('SIGINT'));
  }

  startWorker().catch(err =&gt; {
      console.error(""Failed to start worker:"", err);
      process.exit(1);
  });

}

// pages/api/campaigns.ts
import type { NextApiRequest, NextApiResponse } from 'next';
import { z } from 'zod';
import { CampaignStatus, MessageStatus } from '@prisma/client';
// Consider adding 'csv-parse/sync' if handling CSV uploads: npm install csv-parse
import prisma from '../../lib/prisma';
import { smsQueue, SmsJobData } from '../../lib/queue';
import { differenceInSeconds } from 'date-fns';

// Input validation schema
// Note on API Key: Accepting in body for simplicity. Recommended practice is via headers (e.g., Authorization: Bearer <key&gt; or X-API-Key: <key&gt;).
const campaignSchema = z.object({
  name: z.string().min(3, ""Campaign name must be at least 3 characters""),
  message: z.string().min(1, ""Message cannot be empty""),
  recipients: z.array(z.string().regex(/^\+?[1-9]\d{1,14}$/, ""Invalid phone number format"")).min(1, ""At least one recipient is required""),
  mediaUrls: z.array(z.string().url(""Invalid media URL format"")).optional(),
  scheduledAt: z.string().datetime({ offset: true, message: ""Invalid schedule date format"" }).optional(),
  apiKey: z.string().refine(key =&gt; key === process.env.API_SECRET_KEY, { message: ""Invalid API Key"" }),
});

// Helper to validate E.164 format strictly
function isValidE164(phoneNumber: string): boolean {
    return /^\+[1-9]\d{1,14}$/.test(phoneNumber);
}

export default async function handler(req: NextApiRequest, res: NextApiResponse) {
  if (req.method !== 'POST') {
    res.setHeader('Allow', ['POST']);
    return res.status(405).json({ message: `Method ${req.method} Not Allowed` });
  }

  // --- 1. Input Validation ---
  const validationResult = campaignSchema.safeParse(req.body);
  if (!validationResult.success) {
    return res.status(400).json({ message: ""Invalid input"", errors: validationResult.error.format() });
  }

  const { name, message, recipients: rawRecipients, mediaUrls, scheduledAt, apiKey } = validationResult.data;

  // --- 2. Sanitize, Deduplicate, and Validate Recipients ---
  const uniqueRecipients = [...new Set(rawRecipients)]
      .map(num =&gt; num.trim())
      .filter(isValidE164); // Ensure E.164 format

  if (uniqueRecipients.length === 0) {
      return res.status(400).json({ message: ""No valid E.164 recipient phone numbers provided."" });
  }

  // --- 2.5 Check Opt-Out List (CRITICAL FOR COMPLIANCE) ---
  let optedOutNumbers: Set<string&gt;;
  try {
      const optOuts = await prisma.optOut.findMany({
          where: { phoneNumber: { in: uniqueRecipients } },
          select: { phoneNumber: true }
      });
      optedOutNumbers = new Set(optOuts.map(o =&gt; o.phoneNumber));
      console.log(`Found ${optedOutNumbers.size} opted-out numbers in the provided list.`);
  } catch (error) {
      console.error(""Error checking opt-out list:"", error);
      return res.status(500).json({ message: ""Failed to verify recipient opt-out status."" });
  }

  const finalRecipients = uniqueRecipients.filter(num =&gt; !optedOutNumbers.has(num));

  if (finalRecipients.length === 0) {
      return res.status(400).json({ message: ""All provided recipients have opted out or are invalid."" });
  }
  const skippedCount = uniqueRecipients.length - finalRecipients.length;

  // --- 3. Check Schedule Time ---
  let delay = 0;
  let campaignInitialStatus = CampaignStatus.PROCESSING;
  let jobOptions = {};

  if (scheduledAt) {
    const scheduleDate = new Date(scheduledAt);
    const now = new Date();
    if (scheduleDate &lt;= now) {
      return res.status(400).json({ message: ""Scheduled time must be in the future."" });
    }
    delay = differenceInSeconds(scheduleDate_ now) * 1000; // Delay in milliseconds
    campaignInitialStatus = CampaignStatus.SCHEDULED;
    console.log(`Campaign scheduled for ${scheduleDate}. Delay: ${delay}ms`);
    jobOptions = { delay };
  }

  // --- 4. Create Campaign in Database ---
  let campaign;
  try {
    campaign = await prisma.campaign.create({
      data: {
        name_
        message_
        mediaUrls: mediaUrls ?? []_
        status: campaignInitialStatus_
        scheduledAt: scheduledAt ? new Date(scheduledAt) : null_
        recipients: {
          create: finalRecipients.map(phone =&gt; ({
            phoneNumber: phone,
            // Mark as PENDING if scheduled, QUEUED otherwise. Worker picks up QUEUED.
            status: scheduledAt ? MessageStatus.PENDING : MessageStatus.QUEUED
          })),
        },
      },
      include: {
        recipients: true, // Include recipients to get their IDs for jobs
      },
    });
  } catch (error) {
    console.error(""Error creating campaign in DB:"", error);
    // Handle potential unique constraint violation if campaign name needs to be unique etc.
    return res.status(500).json({ message: ""Failed to save campaign data."" });
  }

  // --- 5. Add Jobs to Queue ---
  const senderId = process.env.PLIVO_SENDER_ID;
  if (!senderId) {
    // Optionally update campaign status to FAILED here
    console.error(""PLIVO_SENDER_ID is not set."");
    return res.status(500).json({ message: ""Server configuration error: Sender ID missing."" });
  }

  try {
    const jobs: { name: string; data: SmsJobData; opts?: any }[] = campaign.recipients.map(recipient =&gt; ({
      name: `send-${campaign.id}-${recipient.id}`, // Unique job name helpful for tracking
      data: {
        recipientId: recipient.id,
        campaignId: campaign.id,
        to: recipient.phoneNumber,
        from: senderId,
        text: message,
        mediaUrls: mediaUrls,
      },
      opts: jobOptions // Apply delay if scheduled
    }));

    if (jobs.length &gt; 0) {
        await smsQueue.addBulk(jobs); // Efficiently add multiple jobs
    } else {
        console.log(`No jobs to queue for campaign ${campaign.id} (all recipients might have opted out or were invalid).`);
        // If no jobs were created (e.g., all opted out), mark campaign as completed/failed?
        await prisma.campaign.update({
            where: { id: campaign.id },
            data: { status: CampaignStatus.COMPLETED } // Or FAILED? Depends on desired logic
        });
    }


    // Update campaign status if not scheduled and jobs were added
    if (!scheduledAt && jobs.length &gt; 0) {
        await prisma.campaign.update({
            where: { id: campaign.id },
            data: { status: CampaignStatus.QUEUED }
        });
    }

    console.log(`Successfully added ${jobs.length} jobs to the queue for campaign ${campaign.id}. Skipped ${skippedCount} opted-out/invalid numbers.`);
    return res.status(201).json({
      message: scheduledAt ? ""Campaign scheduled successfully!"" : ""Campaign queued successfully!"",
      campaignId: campaign.id,
      jobCount: jobs.length,
      skippedCount: skippedCount,
    });

  } catch (error) {
    console.error(""Error adding jobs to queue:"", error);
    // Attempt to mark campaign as FAILED
     try {
       await prisma.campaign.update({
         where: { id: campaign.id },
         data: { status: CampaignStatus.FAILED },
       });
     } catch (dbError) {
         console.error(`Failed to mark campaign ${campaign.id} as FAILED after queue error:`, dbError);
     }
    return res.status(500).json({ message: ""Failed to queue messages for sending."" });
  }
}

## API Endpoints

### POST /api/campaigns

Creates and queues (or schedules) a new SMS/MMS campaign after validating recipients against the opt-out list.

**Request Body:** (`application/json`)

```json
{
  "name": "Summer Promo Blast",
  "message": "Hot deals for summer! \n Up to 40% off selected items. Visit example.com/summer",
  "recipients": [
    "+14155551212",
    "+12125559876",
    "+442071234567"
  ],
  "mediaUrls": [
      "https://example.com/images/summer_deal.png"
  ],
  "scheduledAt": "2024-08-01T14:00:00.000Z",
  "apiKey": "YOUR_STRONG_RANDOM_SECRET_KEY"
}

{
  "message": "Campaign scheduled successfully!" / "Campaign queued successfully!",
  "campaignId": "clerk123abc...",
  "jobCount": 2, // Number of messages actually queued/scheduled
  "skippedCount": 1 // Number of recipients skipped (opted-out or invalid)
}