At TextDrip I was the sole backend engineer responsible for a system that sent millions of SMS messages a month for thousands of businesses. Drip campaigns sequences like "welcome message now, follow-up in 2 days, offer in 5 days" sound simple until you actually have to build them reliably at scale, across multiple tenants, with carrier rate limits, retry logic, and zero message loss.
This is the architecture that actually worked. Not the first version, and not the "clean" version I'd design on a whiteboard. The one that survived production.
Why Not Just Use Cron Jobs
The naive approach is a cron job that runs every minute, queries the database for messages due to be sent, and fires them. I see this in codebases all the time. Here's why it breaks:
- Race conditions: if your cron runs on two servers simultaneously, both pick up the same pending messages and you send duplicates
- No retry logic: if the SMS carrier API returns a 429, the message is just lost unless you build retry state into your DB schema
- Blocking: a slow carrier API call blocks the next batch from starting on time
- No observability: you can't see the queue depth, failure rates, or processing speed without custom instrumentation
- Server-bound: if your server restarts mid-batch, messages in progress have no state to recover from
A message queue solves all of these. I chose Redis with BullMQ because the team was already running Redis for session storage, BullMQ has excellent TypeScript types, and its delayed job support is exactly what drip campaigns need.
The Data Model
Before the queue, the database. A drip campaign has a sequence of steps. Each step has a delay relative to when the contact enrolled. When a contact enrolls in a campaign, we create a campaign_enrollment record and immediately enqueue all messages with their calculated send times.
schema.sql
-- A campaign belongs to a tenant (business)
CREATE TABLE campaigns (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
tenant_id UUID NOT NULL,
name TEXT NOT NULL,
status TEXT NOT NULL DEFAULT 'draft', -- draft | active | paused
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
-- Each step in the drip sequence
CREATE TABLE campaign_steps (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
campaign_id UUID NOT NULL REFERENCES campaigns(id),
delay_hours INT NOT NULL, -- hours after enrollment
message TEXT NOT NULL,
step_order INT NOT NULL
);
-- When a contact enrolls
CREATE TABLE campaign_enrollments (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
campaign_id UUID NOT NULL REFERENCES campaigns(id),
contact_id UUID NOT NULL,
enrolled_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
status TEXT NOT NULL DEFAULT 'active'
);
-- Individual message sends (one per step per enrollment)
CREATE TABLE campaign_messages (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
enrollment_id UUID NOT NULL REFERENCES campaign_enrollments(id),
step_id UUID NOT NULL REFERENCES campaign_steps(id),
tenant_id UUID NOT NULL,
phone TEXT NOT NULL,
body TEXT NOT NULL,
scheduled_at TIMESTAMPTZ NOT NULL,
sent_at TIMESTAMPTZ,
status TEXT NOT NULL DEFAULT 'queued', -- queued | sent | failed | skipped
attempts INT NOT NULL DEFAULT 0,
bull_job_id TEXT -- BullMQ job ID for deduplication
);
Queue Setup with BullMQ
src/queues/sms.queue.ts
import { Queue, Worker, QueueEvents } from "bullmq";
import { redis } from "../lib/redis";
export interface SmsJobData {
messageId: string;
tenantId: string;
phone: string;
body: string;
attempt: number;
}
// One queue per logical domain. Don't share queues across unrelated jobs.
export const smsQueue = new Queue<SmsJobData>("sms", {
connection: redis,
defaultJobOptions: {
attempts: 3,
backoff: {
type: "exponential",
delay: 60_000, // first retry after 1 min, then 2 min, then 4 min
},
removeOnComplete: { count: 1000 }, // keep last 1000 completed jobs for auditing
removeOnFail: { count: 5000 }, // keep failed jobs longer for debugging
},
});
Enrolling a Contact: Scheduling All Messages Upfront
When a contact enrolls, we calculate every future send time immediately and add all jobs to the queue with delay set in milliseconds. BullMQ handles the timing it won't process the job until the delay has elapsed.
src/services/enrollment.service.ts
import { smsQueue } from "../queues/sms.queue";
import { db } from "../lib/db";
export async function enrollContact(
campaignId: string,
contactId: string,
tenantId: string,
phone: string
): Promise<void> {
// Create the enrollment record
const [enrollment] = await db
.insertInto("campaign_enrollments")
.values({ campaign_id: campaignId, contact_id: contactId, status: "active" })
.returning(["id"])
.execute();
// Load all steps for this campaign
const steps = await db
.selectFrom("campaign_steps")
.where("campaign_id", "=", campaignId)
.orderBy("step_order", "asc")
.selectAll()
.execute();
const now = Date.now();
for (const step of steps) {
const scheduledAt = new Date(now + step.delay_hours * 60 * 60 * 1000);
const delayMs = scheduledAt.getTime() - now;
// Create the DB record first so we have an ID
const [msg] = await db
.insertInto("campaign_messages")
.values({
enrollment_id: enrollment.id,
step_id: step.id,
tenant_id: tenantId,
phone,
body: step.message,
scheduled_at: scheduledAt,
status: "queued",
})
.returning(["id"])
.execute();
// Enqueue with the BullMQ job ID = our DB message ID for deduplication
const job = await smsQueue.add(
"send-sms",
{
messageId: msg.id,
tenantId,
phone,
body: step.message,
attempt: 1,
},
{
delay: delayMs,
jobId: msg.id, // idempotency key — BullMQ won't add a second job with same ID
}
);
// Record the Bull job ID for tracking
await db
.updateTable("campaign_messages")
.set({ bull_job_id: job.id })
.where("id", "=", msg.id)
.execute();
}
}
TipUsing jobId: msg.id is the most important line in this function. If your server crashes mid-enrollment and you replay the operation, BullMQ will silently skip duplicate job IDs. No double-sends.
The Worker: Processing Messages
src/workers/sms.worker.ts
import { Worker } from "bullmq";
import { redis } from "../lib/redis";
import { db } from "../lib/db";
import { sendViaTwilio } from "../lib/twilio";
import type { SmsJobData } from "../queues/sms.queue";
export const smsWorker = new Worker<SmsJobData>(
"sms",
async (job) => {
const { messageId, tenantId, phone, body } = job.data;
// Check if this message was cancelled (contact unsubscribed, campaign paused)
const msg = await db
.selectFrom("campaign_messages")
.where("id", "=", messageId)
.select(["status"])
.executeTakeFirst();
if (!msg || msg.status === "skipped") {
return; // Silently skip — don't throw, that would count as a failure
}
// Increment attempt count
await db
.updateTable("campaign_messages")
.set({ attempts: db.raw("attempts + 1") })
.where("id", "=", messageId)
.execute();
// Send via carrier
await sendViaTwilio({ to: phone, body, tenantId });
// Mark as sent
await db
.updateTable("campaign_messages")
.set({ status: "sent", sent_at: new Date() })
.where("id", "=", messageId)
.execute();
},
{
connection: redis,
concurrency: 10, // process 10 jobs simultaneously per worker process
limiter: {
max: 100, // max 100 jobs per...
duration: 1000 // ...per second (Twilio free tier rate limit)
},
}
);
smsWorker.on("failed", async (job, err) => {
if (!job) return;
const isLastAttempt = job.attemptsMade >= (job.opts.attempts ?? 3);
if (isLastAttempt) {
await db
.updateTable("campaign_messages")
.set({ status: "failed" })
.where("id", "=", job.data.messageId)
.execute();
}
});
Handling Unsubscribes and Campaign Pauses
This is the tricky part. When a contact unsubscribes or you pause a campaign, you have future jobs already sitting in the Redis queue. You have two options: remove them from the queue, or let the worker skip them.
I use the skip pattern: mark affected campaign_messages rows as skipped in the database, and let the worker bail out early when it sees that status. Removing jobs from BullMQ requires iterating the delayed queue, which is O(n) and slow. A DB status check is a single indexed lookup.
src/services/unsubscribe.service.ts
export async function unsubscribeContact(contactId: string, tenantId: string) {
// Mark all future queued messages as skipped — worker will bail when it sees this
await db
.updateTable("campaign_messages as cm")
.set({ status: "skipped" })
.where("cm.tenant_id", "=", tenantId)
.where("cm.status", "=", "queued")
.where((eb) =>
eb(
"cm.enrollment_id",
"in",
eb
.selectFrom("campaign_enrollments")
.where("contact_id", "=", contactId)
.select("id")
)
)
.execute();
}
What I Learned the Hard Way
- 1.Don't share a Redis instance between BullMQ and your app cache. BullMQ uses blocking Redis commands (
BRPOPLPUSH) that can starve your cache reads under load. Use a separate Redis connection or a second Redis instance. - 2.Rate limit per tenant, not globally. Some carriers throttle at the sender level. If tenant A sends a burst, they should eat the 429s, not tenant B. BullMQ's
limiter option is global per queue; implement per-tenant rate limiting in the worker itself using a sliding window counter in Redis. - 3.Log the carrier's message SID, not just your internal ID. When Twilio reports a delivery failure, you need to correlate it back to your
campaign_messages row. Store their SID. - 4.Test with `runInBackground: false` in BullMQ dev mode. In local dev, it's painful to wait 2 hours to see if your 2-hour-delay job fires. Set
BULL_DEV_MODE=true and reduce all delays to seconds for testing.
“The queue system was the hardest thing I've ever built. Not because the code was complex — but because the failure modes were. Messages you never sent. Messages you sent twice. Messages delivered to someone who unsubscribed. Each one was a real business consequence.”
Margin note from my TextDrip postmortem doc