From 754f07d04a660b2973c95aebae2a2b33be32e4e1 Mon Sep 17 00:00:00 2001 From: Pankaj Beniwal Date: Tue, 21 Jul 2026 14:32:47 +0530 Subject: [PATCH 1/6] feat: add gradual delivery data model --- .../migration.sql | 34 +++++++ apps/web/prisma/schema.prisma | 97 ++++++++++++------- apps/web/src/lib/campaign-delivery.ts | 58 +++++++++++ .../src/lib/campaign-delivery.unit.test.ts | 61 ++++++++++++ 4 files changed, 215 insertions(+), 35 deletions(-) create mode 100644 apps/web/prisma/migrations/20260721120000_add_gradual_campaign_sending/migration.sql create mode 100644 apps/web/src/lib/campaign-delivery.ts create mode 100644 apps/web/src/lib/campaign-delivery.unit.test.ts diff --git a/apps/web/prisma/migrations/20260721120000_add_gradual_campaign_sending/migration.sql b/apps/web/prisma/migrations/20260721120000_add_gradual_campaign_sending/migration.sql new file mode 100644 index 00000000..f226bde9 --- /dev/null +++ b/apps/web/prisma/migrations/20260721120000_add_gradual_campaign_sending/migration.sql @@ -0,0 +1,34 @@ +-- CreateEnum +CREATE TYPE "CampaignDeliveryMode" AS ENUM ('ALL_AT_ONCE', 'GRADUAL'); + +-- CreateEnum +CREATE TYPE "CampaignRecipientStatus" AS ENUM ('PENDING', 'QUEUED', 'SUPPRESSED', 'SKIPPED', 'FAILED'); + +-- AlterTable +ALTER TABLE "CampaignEmail" ALTER COLUMN "emailId" DROP NOT NULL; +ALTER TABLE "CampaignEmail" ADD COLUMN "status" "CampaignRecipientStatus"; +ALTER TABLE "CampaignEmail" ADD COLUMN "processedAt" TIMESTAMP(3); + +UPDATE "CampaignEmail" +SET "status" = 'QUEUED', "processedAt" = "createdAt"; + +ALTER TABLE "CampaignEmail" ALTER COLUMN "status" SET NOT NULL; +ALTER TABLE "CampaignEmail" ALTER COLUMN "status" SET DEFAULT 'PENDING'; + +-- AlterTable +ALTER TABLE "Campaign" ADD COLUMN "deliveryMode" "CampaignDeliveryMode" NOT NULL DEFAULT 'ALL_AT_ONCE'; +ALTER TABLE "Campaign" ADD COLUMN "deliveryBatchPercentage" INTEGER; +ALTER TABLE "Campaign" ADD COLUMN "deliveryIntervalMinutes" INTEGER; +ALTER TABLE "Campaign" ADD COLUMN "deliveryBatchSize" INTEGER; +ALTER TABLE "Campaign" ADD COLUMN "currentDeliveryBatch" INTEGER NOT NULL DEFAULT 0; +ALTER TABLE "Campaign" ADD COLUMN "deliveryBatchProcessed" INTEGER NOT NULL DEFAULT 0; +ALTER TABLE "Campaign" ADD COLUMN "nextDeliveryAt" TIMESTAMP(3); +ALTER TABLE "Campaign" ADD COLUMN "audienceCapturedAt" TIMESTAMP(3); +ALTER TABLE "Campaign" ADD COLUMN "audiencePreparedAt" TIMESTAMP(3); +ALTER TABLE "Campaign" ADD COLUMN "pausedAt" TIMESTAMP(3); + +-- CreateIndex +CREATE INDEX "CampaignEmail_campaignId_status_contactId_idx" ON "CampaignEmail"("campaignId", "status", "contactId"); + +-- CreateIndex +CREATE INDEX "Campaign_status_nextDeliveryAt_idx" ON "Campaign"("status", "nextDeliveryAt"); diff --git a/apps/web/prisma/schema.prisma b/apps/web/prisma/schema.prisma index 1492adf8..ff446848 100644 --- a/apps/web/prisma/schema.prisma +++ b/apps/web/prisma/schema.prisma @@ -271,12 +271,15 @@ model Email { } model CampaignEmail { - campaignId String - contactId String - emailId String - createdAt DateTime @default(now()) + campaignId String + contactId String + emailId String? + status CampaignRecipientStatus @default(PENDING) + processedAt DateTime? + createdAt DateTime @default(now()) @@id([campaignId, contactId]) + @@index([campaignId, status, contactId]) } model EmailEvent { @@ -349,43 +352,67 @@ enum CampaignStatus { SENT } +enum CampaignDeliveryMode { + ALL_AT_ONCE + GRADUAL +} + +enum CampaignRecipientStatus { + PENDING + QUEUED + SUPPRESSED + SKIPPED + FAILED +} + model Campaign { - id String @id @default(cuid()) - name String - teamId Int - from String - cc String[] - bcc String[] - replyTo String[] - domainId Int - subject String - previewText String? - html String? - content String? - contactBookId String? - scheduledAt DateTime? - total Int @default(0) - sent Int @default(0) - delivered Int @default(0) - opened Int @default(0) - clicked Int @default(0) - unsubscribed Int @default(0) - bounced Int @default(0) - hardBounced Int @default(0) - complained Int @default(0) - isApi Boolean @default(false) - status CampaignStatus @default(DRAFT) - batchSize Int @default(500) - batchWindowMinutes Int @default(0) - lastCursor String? - lastSentAt DateTime? - createdAt DateTime @default(now()) - updatedAt DateTime @updatedAt + id String @id @default(cuid()) + name String + teamId Int + from String + cc String[] + bcc String[] + replyTo String[] + domainId Int + subject String + previewText String? + html String? + content String? + contactBookId String? + scheduledAt DateTime? + total Int @default(0) + sent Int @default(0) + delivered Int @default(0) + opened Int @default(0) + clicked Int @default(0) + unsubscribed Int @default(0) + bounced Int @default(0) + hardBounced Int @default(0) + complained Int @default(0) + isApi Boolean @default(false) + status CampaignStatus @default(DRAFT) + batchSize Int @default(500) + batchWindowMinutes Int @default(0) + lastCursor String? + lastSentAt DateTime? + deliveryMode CampaignDeliveryMode @default(ALL_AT_ONCE) + deliveryBatchPercentage Int? + deliveryIntervalMinutes Int? + deliveryBatchSize Int? + currentDeliveryBatch Int @default(0) + deliveryBatchProcessed Int @default(0) + nextDeliveryAt DateTime? + audienceCapturedAt DateTime? + audiencePreparedAt DateTime? + pausedAt DateTime? + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt team Team @relation(fields: [teamId], references: [id], onDelete: Cascade) @@index([createdAt(sort: Desc)]) @@index([status, scheduledAt]) + @@index([status, nextDeliveryAt]) } model Template { diff --git a/apps/web/src/lib/campaign-delivery.ts b/apps/web/src/lib/campaign-delivery.ts new file mode 100644 index 00000000..4e36fd24 --- /dev/null +++ b/apps/web/src/lib/campaign-delivery.ts @@ -0,0 +1,58 @@ +export const GRADUAL_DELIVERY_MIN_PERCENTAGE = 1; +export const GRADUAL_DELIVERY_MAX_PERCENTAGE = 50; + +export const GRADUAL_DELIVERY_INTERVAL_MINUTES = { + minute: 1, + hour: 60, +} as const; + +export type GradualDeliveryInterval = + keyof typeof GRADUAL_DELIVERY_INTERVAL_MINUTES; + +export function calculateGradualDelivery({ + audienceSize, + batchPercentage, + intervalMinutes, + startsAt, +}: { + audienceSize: number; + batchPercentage: number; + intervalMinutes: number; + startsAt: Date; +}) { + if (!Number.isInteger(audienceSize) || audienceSize < 0) { + throw new Error("Audience size must be a non-negative integer"); + } + + if ( + !Number.isInteger(batchPercentage) || + batchPercentage < GRADUAL_DELIVERY_MIN_PERCENTAGE || + batchPercentage > GRADUAL_DELIVERY_MAX_PERCENTAGE + ) { + throw new Error( + `Batch percentage must be between ${GRADUAL_DELIVERY_MIN_PERCENTAGE} and ${GRADUAL_DELIVERY_MAX_PERCENTAGE}`, + ); + } + + if (!Number.isInteger(intervalMinutes) || intervalMinutes <= 0) { + throw new Error("Delivery interval must be a positive number of minutes"); + } + + const batchSize = + audienceSize === 0 + ? 0 + : Math.max(1, Math.ceil((audienceSize * batchPercentage) / 100)); + const totalBatches = + batchSize === 0 ? 0 : Math.ceil(audienceSize / batchSize); + const durationMinutes = Math.max(0, totalBatches - 1) * intervalMinutes; + const completesAt = new Date( + startsAt.getTime() + durationMinutes * 60 * 1000, + ); + + return { + batchSize, + totalBatches, + durationMinutes, + completesAt, + }; +} diff --git a/apps/web/src/lib/campaign-delivery.unit.test.ts b/apps/web/src/lib/campaign-delivery.unit.test.ts new file mode 100644 index 00000000..7867a930 --- /dev/null +++ b/apps/web/src/lib/campaign-delivery.unit.test.ts @@ -0,0 +1,61 @@ +import { describe, expect, it } from "vitest"; +import { calculateGradualDelivery } from "~/lib/campaign-delivery"; + +describe("calculateGradualDelivery", () => { + const startsAt = new Date("2026-07-21T09:00:00.000Z"); + + it("calculates percentage-based batches and completion from the first wave", () => { + const result = calculateGradualDelivery({ + audienceSize: 50_000, + batchPercentage: 10, + intervalMinutes: 60, + startsAt, + }); + + expect(result).toEqual({ + batchSize: 5_000, + totalBatches: 10, + durationMinutes: 540, + completesAt: new Date("2026-07-21T18:00:00.000Z"), + }); + }); + + it("rounds the batch size up and leaves the remainder for the final batch", () => { + const result = calculateGradualDelivery({ + audienceSize: 3, + batchPercentage: 50, + intervalMinutes: 1, + startsAt, + }); + + expect(result.batchSize).toBe(2); + expect(result.totalBatches).toBe(2); + expect(result.completesAt).toEqual( + new Date("2026-07-21T09:01:00.000Z"), + ); + }); + + it("returns an empty schedule for an empty audience", () => { + const result = calculateGradualDelivery({ + audienceSize: 0, + batchPercentage: 10, + intervalMinutes: 60, + startsAt, + }); + + expect(result.batchSize).toBe(0); + expect(result.totalBatches).toBe(0); + expect(result.completesAt).toEqual(startsAt); + }); + + it.each([0, 51, 10.5])("rejects an invalid percentage of %s", (value) => { + expect(() => + calculateGradualDelivery({ + audienceSize: 100, + batchPercentage: value, + intervalMinutes: 60, + startsAt, + }), + ).toThrow("Batch percentage must be between 1 and 50"); + }); +}); From 861fd467b49c6a8959015e16acc481e76acb959d Mon Sep 17 00:00:00 2001 From: Pankaj Beniwal Date: Tue, 21 Jul 2026 14:37:09 +0530 Subject: [PATCH 2/6] feat: process campaigns in delivery waves --- .../src/server/service/campaign-service.ts | 530 ++++++++++++++++-- .../service/campaign-service.unit.test.ts | 110 +++- 2 files changed, 572 insertions(+), 68 deletions(-) diff --git a/apps/web/src/server/service/campaign-service.ts b/apps/web/src/server/service/campaign-service.ts index 5a109993..7a853478 100644 --- a/apps/web/src/server/service/campaign-service.ts +++ b/apps/web/src/server/service/campaign-service.ts @@ -31,6 +31,22 @@ import { } from "../utils/contact-variable-replacement"; import { updateContactSubscription } from "./contact-service"; import { getCampaignUnsubscribeVariableValues } from "~/lib/constants/campaign"; +import { + calculateGradualDelivery, + GRADUAL_DELIVERY_INTERVAL_MINUTES, +} from "~/lib/campaign-delivery"; +import type { GradualDeliveryInterval } from "~/lib/campaign-delivery"; + +const CAMPAIGN_AUDIENCE_SNAPSHOT_CHUNK_SIZE = 5_000; +const GRADUAL_DELIVERY_INTERNAL_BATCH_SIZE = 500; + +export type CampaignDeliveryInput = + | { strategy: "ALL_AT_ONCE" } + | { + strategy: "GRADUAL"; + batchPercentage: number; + interval: GradualDeliveryInterval; + }; const CAMPAIGN_UNSUB_PLACEHOLDER_TOKENS = [ "{{unsend_unsubscribe_url}}", @@ -69,6 +85,40 @@ function sanitizeAddressList(addresses?: string | string[]) { .filter((address) => address.length > 0); } +function getCampaignDeliveryData({ + delivery, + audienceSize, + startsAt, +}: { + delivery?: CampaignDeliveryInput; + audienceSize: number; + startsAt: Date; +}) { + if (!delivery || delivery.strategy === "ALL_AT_ONCE") { + return { + deliveryMode: "ALL_AT_ONCE" as const, + deliveryBatchPercentage: null, + deliveryIntervalMinutes: null, + deliveryBatchSize: null, + }; + } + + const intervalMinutes = GRADUAL_DELIVERY_INTERVAL_MINUTES[delivery.interval]; + const estimate = calculateGradualDelivery({ + audienceSize, + batchPercentage: delivery.batchPercentage, + intervalMinutes, + startsAt, + }); + + return { + deliveryMode: "GRADUAL" as const, + deliveryBatchPercentage: delivery.batchPercentage, + deliveryIntervalMinutes: intervalMinutes, + deliveryBatchSize: estimate.batchSize, + }; +} + async function prepareCampaignHtml( campaign: Campaign, ): Promise<{ campaign: Campaign; html: string }> { @@ -305,6 +355,13 @@ export async function getCampaignForTeam({ scheduledAt: true, batchSize: true, batchWindowMinutes: true, + deliveryMode: true, + deliveryBatchPercentage: true, + deliveryIntervalMinutes: true, + deliveryBatchSize: true, + currentDeliveryBatch: true, + deliveryBatchProcessed: true, + nextDeliveryAt: true, total: true, sent: true, delivered: true, @@ -390,11 +447,13 @@ export async function scheduleCampaign({ teamId, scheduledAt: scheduledAtInput, batchSize, + delivery, }: { campaignId: string; teamId: number; scheduledAt?: Date | string; batchSize?: number; + delivery?: CampaignDeliveryInput; }) { let campaign = await db.campaign.findUnique({ where: { id: campaignId, teamId }, @@ -406,6 +465,14 @@ export async function scheduleCampaign({ }); } + if (campaign.status !== "DRAFT" && campaign.status !== "SCHEDULED") { + throw new UnsendApiError({ + code: "BAD_REQUEST", + message: + "Delivery settings cannot be changed after a campaign has started", + }); + } + let html: string; try { const prepared = await prepareCampaignHtml(campaign); @@ -461,18 +528,36 @@ export async function scheduleCampaign({ : new Date(scheduledAtInput) : new Date(); - const shouldResetCursor = - campaign.status === "DRAFT" || campaign.status === "SENT"; + const shouldResetCursor = campaign.status === "DRAFT"; - await db.campaign.update({ - where: { id: campaign.id }, - data: { - status: "SCHEDULED", - scheduledAt, - total, - ...(batchSize ? { batchSize } : {}), - ...(shouldResetCursor ? { lastCursor: null } : {}), - }, + const deliveryData = getCampaignDeliveryData({ + delivery, + audienceSize: total, + startsAt: scheduledAt, + }); + + await db.$transaction(async (tx) => { + await tx.campaignEmail.deleteMany({ + where: { campaignId: campaign.id }, + }); + + await tx.campaign.update({ + where: { id: campaign.id }, + data: { + status: "SCHEDULED", + scheduledAt, + total, + ...deliveryData, + currentDeliveryBatch: 0, + deliveryBatchProcessed: 0, + nextDeliveryAt: null, + audienceCapturedAt: null, + audiencePreparedAt: null, + pausedAt: null, + ...(batchSize ? { batchSize } : {}), + ...(shouldResetCursor ? { lastCursor: null } : {}), + }, + }); }); return { ok: true }; @@ -498,7 +583,7 @@ export async function pauseCampaign({ await db.campaign.update({ where: { id: campaignId }, - data: { status: "PAUSED" }, + data: { status: "PAUSED", pausedAt: new Date() }, }); return { ok: true }; @@ -522,15 +607,31 @@ export async function resumeCampaign({ }); } - if (campaign.scheduledAt && campaign.scheduledAt.getTime() > Date.now()) { + const now = new Date(); + const pausedDurationMs = campaign.pausedAt + ? Math.max(0, now.getTime() - campaign.pausedAt.getTime()) + : 0; + const shiftedNextDeliveryAt = campaign.nextDeliveryAt + ? new Date(campaign.nextDeliveryAt.getTime() + pausedDurationMs) + : null; + + if (campaign.scheduledAt && campaign.scheduledAt.getTime() > now.getTime()) { await db.campaign.update({ where: { id: campaignId }, - data: { status: "SCHEDULED" }, + data: { + status: "SCHEDULED", + pausedAt: null, + nextDeliveryAt: shiftedNextDeliveryAt, + }, }); } else { await db.campaign.update({ where: { id: campaignId }, - data: { status: "RUNNING" }, + data: { + status: "RUNNING", + pausedAt: null, + nextDeliveryAt: shiftedNextDeliveryAt, + }, }); } @@ -818,12 +919,37 @@ export async function recordCampaignContactFailure({ emailId = failedEmail.id; } - await tx.campaignEmail.create({ - data: { + await tx.campaignEmail.upsert({ + where: { + campaignId_contactId: { + campaignId: campaign.id, + contactId: contact.id, + }, + }, + create: { campaignId: campaign.id, contactId: contact.id, emailId, + status: "FAILED", + processedAt: new Date(), + }, + update: { + emailId, + status: "FAILED", + processedAt: new Date(), + }, + }); + } + + if (existingCampaignEmail?.emailId) { + await tx.campaignEmail.update({ + where: { + campaignId_contactId: { + campaignId: campaign.id, + contactId: contact.id, + }, }, + data: { status: "FAILED", processedAt: new Date() }, }); } @@ -933,11 +1059,24 @@ async function processContactEmail(jobData: CampaignEmailJob) { }, }); - await db.campaignEmail.create({ - data: { + await db.campaignEmail.upsert({ + where: { + campaignId_contactId: { + campaignId: emailConfig.campaignId, + contactId: contact.id, + }, + }, + create: { campaignId: emailConfig.campaignId, contactId: contact.id, emailId: email.id, + status: "SUPPRESSED", + processedAt: new Date(), + }, + update: { + emailId: email.id, + status: "SUPPRESSED", + processedAt: new Date(), }, }); @@ -987,11 +1126,24 @@ async function processContactEmail(jobData: CampaignEmailJob) { }, }); - await db.campaignEmail.create({ - data: { + await db.campaignEmail.upsert({ + where: { + campaignId_contactId: { + campaignId: emailConfig.campaignId, + contactId: contact.id, + }, + }, + create: { campaignId: emailConfig.campaignId, contactId: contact.id, emailId: email.id, + status: "QUEUED", + processedAt: new Date(), + }, + update: { + emailId: email.id, + status: "QUEUED", + processedAt: new Date(), }, }); @@ -1056,6 +1208,169 @@ export async function updateCampaignAnalytics( // Simple campaign batch queue // --------------------------- +async function prepareCampaignAudience(campaign: Campaign) { + if (campaign.audiencePreparedAt) { + return campaign; + } + + if (!campaign.contactBookId) { + throw new Error("No contact book found for campaign"); + } + + const capturedAt = campaign.audienceCapturedAt ?? new Date(); + + if (!campaign.audienceCapturedAt) { + await db.campaign.update({ + where: { id: campaign.id }, + data: { audienceCapturedAt: capturedAt }, + }); + } + + let cursor: string | undefined; + + while (true) { + const contacts = await db.contact.findMany({ + where: { + contactBookId: campaign.contactBookId, + subscribed: true, + createdAt: { lte: capturedAt }, + }, + select: { id: true }, + orderBy: { id: "asc" }, + take: CAMPAIGN_AUDIENCE_SNAPSHOT_CHUNK_SIZE, + ...(cursor ? { cursor: { id: cursor }, skip: 1 } : {}), + }); + + if (contacts.length === 0) { + break; + } + + await db.campaignEmail.createMany({ + data: contacts.map((contact) => ({ + campaignId: campaign.id, + contactId: contact.id, + status: "PENDING" as const, + })), + skipDuplicates: true, + }); + + cursor = contacts[contacts.length - 1]?.id; + + if (contacts.length < CAMPAIGN_AUDIENCE_SNAPSHOT_CHUNK_SIZE) { + break; + } + } + + const total = await db.campaignEmail.count({ + where: { campaignId: campaign.id }, + }); + + const deliveryBatchSize = + campaign.deliveryMode === "GRADUAL" + ? calculateGradualDelivery({ + audienceSize: total, + batchPercentage: campaign.deliveryBatchPercentage ?? 0, + intervalMinutes: campaign.deliveryIntervalMinutes ?? 0, + startsAt: campaign.scheduledAt ?? new Date(), + }).batchSize + : null; + + return db.campaign.update({ + where: { id: campaign.id }, + data: { + total, + audienceCapturedAt: capturedAt, + audiencePreparedAt: new Date(), + deliveryBatchSize, + }, + }); +} + +function getGradualDeliveryWaveSize(campaign: Campaign) { + if ( + campaign.deliveryMode !== "GRADUAL" || + !campaign.deliveryBatchSize || + campaign.currentDeliveryBatch <= 0 + ) { + return 0; + } + + const previouslyReleased = + (campaign.currentDeliveryBatch - 1) * campaign.deliveryBatchSize; + + return Math.max( + 0, + Math.min(campaign.deliveryBatchSize, campaign.total - previouslyReleased), + ); +} + +async function getCampaignProcessingLimit(campaign: Campaign) { + if (campaign.deliveryMode !== "GRADUAL") { + return { campaign, take: campaign.batchSize ?? 500 }; + } + + if (!campaign.deliveryBatchSize || !campaign.deliveryIntervalMinutes) { + throw new Error("Gradual delivery configuration is incomplete"); + } + + const deliveryBatchSize = campaign.deliveryBatchSize; + const deliveryIntervalMinutes = campaign.deliveryIntervalMinutes; + + let currentCampaign = campaign; + let currentWaveSize = getGradualDeliveryWaveSize(currentCampaign); + + if (currentCampaign.currentDeliveryBatch === 0) { + currentCampaign = await db.campaign.update({ + where: { id: currentCampaign.id }, + data: { + currentDeliveryBatch: 1, + deliveryBatchProcessed: 0, + nextDeliveryAt: new Date( + Date.now() + deliveryIntervalMinutes * 60 * 1000, + ), + }, + }); + currentWaveSize = getGradualDeliveryWaveSize(currentCampaign); + } else if (currentCampaign.deliveryBatchProcessed >= currentWaveSize) { + const allWavesReleased = + currentCampaign.currentDeliveryBatch * deliveryBatchSize >= + currentCampaign.total; + + if (allWavesReleased) { + return { campaign: currentCampaign, take: 0 }; + } + + if ( + currentCampaign.nextDeliveryAt && + currentCampaign.nextDeliveryAt.getTime() > Date.now() + ) { + return { campaign: currentCampaign, take: 0 }; + } + + currentCampaign = await db.campaign.update({ + where: { id: currentCampaign.id }, + data: { + currentDeliveryBatch: { increment: 1 }, + deliveryBatchProcessed: 0, + nextDeliveryAt: new Date( + Date.now() + deliveryIntervalMinutes * 60 * 1000, + ), + }, + }); + currentWaveSize = getGradualDeliveryWaveSize(currentCampaign); + } + + const remainingInWave = Math.max( + 0, + currentWaveSize - currentCampaign.deliveryBatchProcessed, + ); + + return { + campaign: currentCampaign, + take: Math.min(GRADUAL_DELIVERY_INTERNAL_BATCH_SIZE, remainingInWave), + }; +} + type CampaignBatchJob = TeamJob<{ campaignId: string }>; export class CampaignBatchService { @@ -1073,11 +1388,12 @@ export class CampaignBatchService { createWorkerHandler(async (job: CampaignBatchJob) => { const { campaignId } = job.data; - const campaign = await db.campaign.findUnique({ + let campaign = await db.campaign.findUnique({ where: { id: campaignId }, }); if (!campaign) return; if (!campaign.contactBookId) return; + const contactBookId = campaign.contactBookId; // Skip paused campaigns if (campaign.status === "PAUSED") return; @@ -1088,31 +1404,65 @@ export class CampaignBatchService { // First touch moves SCHEDULED -> RUNNING if (campaign.status === "SCHEDULED") { - await db.campaign.update({ + campaign = await db.campaign.update({ where: { id: campaignId }, data: { status: "RUNNING" }, }); } - const batchSize = campaign.batchSize ?? 500; + campaign = await prepareCampaignAudience(campaign); - const where = { - contactBookId: campaign.contactBookId, - subscribed: true, - } as const; - const pagination: any = { - take: batchSize, - orderBy: { id: "asc" as const }, - }; - if (campaign.lastCursor) { - pagination.cursor = { id: campaign.lastCursor }; - pagination.skip = 1; // do not include the cursor row + if (campaign.total === 0) { + await db.campaign.update({ + where: { id: campaignId }, + data: { status: "SENT" }, + }); + return; + } + + const processingLimit = await getCampaignProcessingLimit(campaign); + campaign = processingLimit.campaign; + + if (processingLimit.take === 0) { + const pendingRecipients = await db.campaignEmail.count({ + where: { campaignId, status: "PENDING" }, + }); + + if (pendingRecipients === 0) { + await db.campaign.update({ + where: { id: campaignId }, + data: { status: "SENT", nextDeliveryAt: null }, + }); + } + return; } - const contacts = await db.contact.findMany({ where, ...pagination }); + const recipients = await db.campaignEmail.findMany({ + where: { campaignId, status: "PENDING" }, + select: { contactId: true }, + orderBy: { contactId: "asc" }, + take: processingLimit.take, + }); + + if (recipients.length === 0) { + await db.campaign.update({ + where: { id: campaignId }, + data: { status: "SENT", nextDeliveryAt: null }, + }); + return; + } + + const contacts = await db.contact.findMany({ + where: { + id: { in: recipients.map((recipient) => recipient.contactId) }, + }, + }); + const contactsById = new Map( + contacts.map((contact) => [contact.id, contact]), + ); const contactBook = await db.contactBook.findUnique({ - where: { id: campaign.contactBookId }, + where: { id: contactBookId }, select: { variables: true }, }); @@ -1121,34 +1471,31 @@ export class CampaignBatchService { ...(contactBook?.variables ?? []), ]; - if (contacts.length === 0) { - // No more contacts -> mark SENT - await db.campaign.update({ - where: { id: campaignId }, - data: { status: "SENT" }, - }); - return; - } - // Fetch domain for region and id const domain = await db.domain.findUnique({ where: { id: campaign.domainId }, }); if (!domain) return; - // Bulk existence check to avoid duplicates while unique is not enforced - const existing = await db.campaignEmail.findMany({ - where: { - campaignId: campaign.id, - contactId: { in: contacts.map((c) => c.id) }, - }, - select: { contactId: true }, - }); - const existingSet = new Set(existing.map((e) => e.contactId)); - // Process each contact in this batch - for (const contact of contacts) { - if (existingSet.has(contact.id)) continue; + let processedRecipientCount = 0; + + for (const recipient of recipients) { + const contact = contactsById.get(recipient.contactId); + + if (!contact || !contact.subscribed) { + await db.campaignEmail.update({ + where: { + campaignId_contactId: { + campaignId, + contactId: recipient.contactId, + }, + }, + data: { status: "SKIPPED", processedAt: new Date() }, + }); + processedRecipientCount += 1; + continue; + } try { await processContactEmail({ @@ -1188,21 +1535,43 @@ export class CampaignBatchService { }, error: err, }); + processedRecipientCount += 1; } catch (recordErr) { logger.error( { err: recordErr, contactId: contact.id, campaignId }, "Failed to record campaign contact failure; skipping to next", ); } + continue; } + + processedRecipientCount += 1; } - // Advance cursor and timestamp - const newCursor = contacts[contacts.length - 1]?.id; await db.campaign.update({ where: { id: campaignId }, - data: { lastCursor: newCursor, lastSentAt: new Date() }, + data: { + lastSentAt: new Date(), + ...(campaign.deliveryMode === "GRADUAL" + ? { + deliveryBatchProcessed: { + increment: processedRecipientCount, + }, + } + : {}), + }, }); + + const pendingRecipients = await db.campaignEmail.count({ + where: { campaignId, status: "PENDING" }, + }); + + if (pendingRecipients === 0) { + await db.campaign.update({ + where: { id: campaignId }, + data: { status: "SENT", nextDeliveryAt: null }, + }); + } }), { connection: getRedis(), @@ -1223,10 +1592,47 @@ export class CampaignBatchService { try { const campaign = await db.campaign.findUnique({ where: { id: campaignId }, - select: { lastSentAt: true, batchWindowMinutes: true, status: true }, + select: { + lastSentAt: true, + batchWindowMinutes: true, + status: true, + total: true, + deliveryMode: true, + deliveryBatchSize: true, + currentDeliveryBatch: true, + deliveryBatchProcessed: true, + nextDeliveryAt: true, + }, }); if (!campaign) return; if (campaign.status === "PAUSED" || campaign.status === "SENT") return; + + if ( + campaign.deliveryMode === "GRADUAL" && + campaign.deliveryBatchSize && + campaign.currentDeliveryBatch > 0 + ) { + const previouslyReleased = + (campaign.currentDeliveryBatch - 1) * campaign.deliveryBatchSize; + const currentWaveSize = Math.max( + 0, + Math.min( + campaign.deliveryBatchSize, + campaign.total - previouslyReleased, + ), + ); + const currentWaveComplete = + campaign.deliveryBatchProcessed >= currentWaveSize; + + if ( + currentWaveComplete && + campaign.nextDeliveryAt && + campaign.nextDeliveryAt.getTime() > Date.now() + ) { + return; + } + } + const windowMin = campaign.batchWindowMinutes ?? 0; if (windowMin > 0 && campaign.lastSentAt) { const elapsedMs = Date.now() - new Date(campaign.lastSentAt).getTime(); diff --git a/apps/web/src/server/service/campaign-service.unit.test.ts b/apps/web/src/server/service/campaign-service.unit.test.ts index 2fe049ec..2ba0a112 100644 --- a/apps/web/src/server/service/campaign-service.unit.test.ts +++ b/apps/web/src/server/service/campaign-service.unit.test.ts @@ -1,4 +1,4 @@ -import { beforeEach, describe, expect, it, vi } from "vitest"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { createHash } from "crypto"; import { UnsubscribeReason } from "@prisma/client"; @@ -6,7 +6,8 @@ const { mockDb, mockTx, mockUpdateContactSubscription } = vi.hoisted(() => { const mockTx = { campaignEmail: { findUnique: vi.fn(), - create: vi.fn(), + update: vi.fn(), + upsert: vi.fn(), }, email: { findFirst: vi.fn(), @@ -28,6 +29,7 @@ const { mockDb, mockTx, mockUpdateContactSubscription } = vi.hoisted(() => { findUnique: vi.fn(), }, campaign: { + findUnique: vi.fn(), update: vi.fn(), }, }, @@ -93,6 +95,8 @@ vi.mock("~/server/logger/log", () => ({ import { recordCampaignContactFailure, + resumeCampaign, + scheduleCampaign, subscribeContact, unsubscribeContact, } from "~/server/service/campaign-service"; @@ -142,6 +146,15 @@ describe("recordCampaignContactFailure", () => { teamId: 7, }, }); + expect(mockTx.campaignEmail.update).toHaveBeenCalledWith({ + where: { + campaignId_contactId: { + campaignId: "campaign_1", + contactId: "contact_1", + }, + }, + data: { status: "FAILED", processedAt: expect.any(Date) }, + }); }); it("creates a failed email and campaign link when processing failed before persistence", async () => { @@ -160,11 +173,24 @@ describe("recordCampaignContactFailure", () => { }), select: { id: true }, }); - expect(mockTx.campaignEmail.create).toHaveBeenCalledWith({ - data: { + expect(mockTx.campaignEmail.upsert).toHaveBeenCalledWith({ + where: { + campaignId_contactId: { + campaignId: "campaign_1", + contactId: "contact_1", + }, + }, + create: { campaignId: "campaign_1", contactId: "contact_1", emailId: "email_2", + status: "FAILED", + processedAt: expect.any(Date), + }, + update: { + emailId: "email_2", + status: "FAILED", + processedAt: expect.any(Date), }, }); expect(mockTx.emailEvent.create).toHaveBeenCalledWith({ @@ -182,11 +208,24 @@ describe("recordCampaignContactFailure", () => { await recordCampaignContactFailure(input); expect(mockTx.email.create).not.toHaveBeenCalled(); - expect(mockTx.campaignEmail.create).toHaveBeenCalledWith({ - data: { + expect(mockTx.campaignEmail.upsert).toHaveBeenCalledWith({ + where: { + campaignId_contactId: { + campaignId: "campaign_1", + contactId: "contact_1", + }, + }, + create: { campaignId: "campaign_1", contactId: "contact_1", emailId: "email_3", + status: "FAILED", + processedAt: expect.any(Date), + }, + update: { + emailId: "email_3", + status: "FAILED", + processedAt: expect.any(Date), }, }); expect(mockTx.email.update).toHaveBeenCalledWith({ @@ -196,6 +235,65 @@ describe("recordCampaignContactFailure", () => { }); }); +describe("campaign delivery lifecycle", () => { + beforeEach(() => { + vi.clearAllMocks(); + vi.useRealTimers(); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + it("does not allow delivery settings to change after sending starts", async () => { + mockDb.campaign.findUnique.mockResolvedValue({ + id: "campaign_1", + status: "RUNNING", + }); + + await expect( + scheduleCampaign({ + campaignId: "campaign_1", + teamId: 7, + delivery: { + strategy: "GRADUAL", + batchPercentage: 10, + interval: "hour", + }, + }), + ).rejects.toMatchObject({ + code: "BAD_REQUEST", + message: + "Delivery settings cannot be changed after a campaign has started", + }); + expect(mockDb.$transaction).not.toHaveBeenCalled(); + }); + + it("moves the next wave by the paused duration when resuming", async () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-07-21T12:00:00.000Z")); + mockDb.campaign.findUnique.mockResolvedValue({ + id: "campaign_1", + teamId: 7, + status: "PAUSED", + scheduledAt: new Date("2026-07-21T09:00:00.000Z"), + pausedAt: new Date("2026-07-21T10:00:00.000Z"), + nextDeliveryAt: new Date("2026-07-21T10:30:00.000Z"), + }); + + await resumeCampaign({ campaignId: "campaign_1", teamId: 7 }); + + expect(mockDb.campaign.update).toHaveBeenCalledWith({ + where: { id: "campaign_1" }, + data: { + status: "RUNNING", + pausedAt: null, + nextDeliveryAt: new Date("2026-07-21T12:30:00.000Z"), + }, + }); + }); +}); + describe("campaign contact subscription changes", () => { beforeEach(() => { vi.clearAllMocks(); From 8a2dacbcaeb78036b770c6778585aee01e091df4 Mon Sep 17 00:00:00 2001 From: Pankaj Beniwal Date: Tue, 21 Jul 2026 14:40:48 +0530 Subject: [PATCH 3/6] feat: expose gradual campaign delivery --- .../routers/campaign-delivery.trpc.test.ts | 131 ++++++++++++++++++ apps/web/src/server/api/routers/campaign.ts | 73 +++++++++- .../api/campaigns/create-campaign.ts | 18 +++ .../api/campaigns/schedule-campaign.ts | 13 +- .../public-api/schemas/campaign-schema.ts | 24 +++- .../schemas/campaign-schema.unit.test.ts | 47 +++++++ .../src/server/service/campaign-service.ts | 47 ++++++- .../service/campaign-service.unit.test.ts | 44 +++++- 8 files changed, 385 insertions(+), 12 deletions(-) create mode 100644 apps/web/src/server/api/routers/campaign-delivery.trpc.test.ts create mode 100644 apps/web/src/server/public-api/schemas/campaign-schema.unit.test.ts diff --git a/apps/web/src/server/api/routers/campaign-delivery.trpc.test.ts b/apps/web/src/server/api/routers/campaign-delivery.trpc.test.ts new file mode 100644 index 00000000..b4d0e6df --- /dev/null +++ b/apps/web/src/server/api/routers/campaign-delivery.trpc.test.ts @@ -0,0 +1,131 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const { mockDb, mockScheduleCampaign } = vi.hoisted(() => ({ + mockDb: { + teamUser: { findFirst: vi.fn() }, + campaign: { findUnique: vi.fn() }, + contact: { count: vi.fn() }, + campaignEmail: { groupBy: vi.fn() }, + email: { groupBy: vi.fn() }, + }, + mockScheduleCampaign: vi.fn(), +})); + +vi.mock("~/server/db", () => ({ db: mockDb })); +vi.mock("~/server/auth", () => ({ getServerAuthSession: vi.fn() })); +vi.mock("~/server/service/campaign-service", () => ({ + scheduleCampaign: mockScheduleCampaign, +})); +vi.mock("~/server/service/webhook-service", () => ({})); +vi.mock("~/server/service/domain-service", () => ({ + validateDomainFromEmail: vi.fn(), +})); +vi.mock("~/server/service/storage-service", () => ({ + getDocumentUploadUrl: vi.fn(), + isStorageConfigured: vi.fn(() => false), +})); + +import { campaignRouter } from "~/server/api/routers/campaign"; +import { createCallerFactory } from "~/server/api/trpc"; + +const createCaller = createCallerFactory(campaignRouter); + +function getContext() { + return { + db: mockDb, + headers: new Headers(), + session: { + user: { + id: 1, + email: "owner@example.com", + isWaitlisted: false, + isAdmin: false, + isBetaUser: true, + }, + }, + } as any; +} + +describe("campaign delivery procedures", () => { + beforeEach(() => { + vi.clearAllMocks(); + mockDb.teamUser.findFirst.mockResolvedValue({ + teamId: 10, + userId: 1, + role: "ADMIN", + team: { id: 10, name: "Acme" }, + }); + mockDb.campaign.findUnique.mockResolvedValue({ + id: "camp_1", + teamId: 10, + contactBookId: "book_1", + sent: 15, + }); + }); + + it("counts the current subscribed audience for the schedule preview", async () => { + mockDb.contact.count.mockResolvedValue(50_000); + + const result = await createCaller(getContext()).getAudienceCount({ + campaignId: "camp_1", + }); + + expect(result).toEqual({ total: 50_000 }); + expect(mockDb.contact.count).toHaveBeenCalledWith({ + where: { contactBookId: "book_1", subscribed: true }, + }); + }); + + it("returns delivery progress across recipient and email states", async () => { + mockDb.campaignEmail.groupBy.mockResolvedValue([ + { status: "PENDING", _count: { _all: 40 } }, + { status: "QUEUED", _count: { _all: 30 } }, + { status: "SUPPRESSED", _count: { _all: 2 } }, + { status: "SKIPPED", _count: { _all: 3 } }, + { status: "FAILED", _count: { _all: 1 } }, + ]); + mockDb.email.groupBy.mockResolvedValue([ + { latestStatus: "SCHEDULED", _count: { _all: 10 } }, + { latestStatus: "QUEUED", _count: { _all: 20 } }, + { latestStatus: "FAILED", _count: { _all: 1 } }, + ]); + + const result = await createCaller(getContext()).getDeliveryProgress({ + campaignId: "camp_1", + }); + + expect(result).toEqual({ + pending: 40, + processed: 36, + queued: 30, + sent: 15, + failed: 1, + suppressed: 5, + }); + }); + + it("validates and forwards gradual scheduling settings", async () => { + mockScheduleCampaign.mockResolvedValue({ ok: true }); + + await createCaller(getContext()).scheduleCampaign({ + campaignId: "camp_1", + delivery: { + strategy: "GRADUAL", + batchPercentage: 10, + interval: "hour", + }, + }); + + expect(mockScheduleCampaign).toHaveBeenCalledWith({ + campaignId: "camp_1", + teamId: 10, + scheduledAt: undefined, + batchSize: undefined, + delivery: { + strategy: "GRADUAL", + batchPercentage: 10, + interval: "hour", + }, + }); + }); +}); diff --git a/apps/web/src/server/api/routers/campaign.ts b/apps/web/src/server/api/routers/campaign.ts index a321ab0e..4611fcdb 100644 --- a/apps/web/src/server/api/routers/campaign.ts +++ b/apps/web/src/server/api/routers/campaign.ts @@ -19,6 +19,14 @@ import { } from "~/server/service/storage-service"; const statuses = Object.values(CampaignStatus) as [CampaignStatus]; +const campaignDeliverySchema = z.discriminatedUnion("strategy", [ + z.object({ strategy: z.literal("ALL_AT_ONCE") }), + z.object({ + strategy: z.literal("GRADUAL"), + batchPercentage: z.number().int().min(1).max(50), + interval: z.enum(["minute", "hour"]), + }), +]); export const campaignRouter = createTRPCRouter({ getCampaigns: teamProcedure @@ -171,9 +179,11 @@ export const campaignRouter = createTRPCRouter({ return campaign; }), - deleteCampaign: campaignProcedure.mutation(async ({ ctx: { team }, input }) => { - return await campaignService.deleteCampaign(input.campaignId, team.id); - }), + deleteCampaign: campaignProcedure.mutation( + async ({ ctx: { team }, input }) => { + return await campaignService.deleteCampaign(input.campaignId, team.id); + }, + ), getCampaign: campaignProcedure.query(async ({ ctx: { db, team }, input }) => { const campaign = await db.campaign.findUnique({ @@ -202,6 +212,61 @@ export const campaignRouter = createTRPCRouter({ }; }), + getAudienceCount: campaignProcedure.query( + async ({ ctx: { db, campaign } }) => { + if (!campaign.contactBookId) { + return { total: 0 }; + } + + const total = await db.contact.count({ + where: { + contactBookId: campaign.contactBookId, + subscribed: true, + }, + }); + + return { total }; + }, + ), + + getDeliveryProgress: campaignProcedure.query( + async ({ ctx: { db, campaign } }) => { + const [recipientStatuses, emailStatuses] = await Promise.all([ + db.campaignEmail.groupBy({ + by: ["status"], + where: { campaignId: campaign.id }, + _count: { _all: true }, + }), + db.email.groupBy({ + by: ["latestStatus"], + where: { campaignId: campaign.id }, + _count: { _all: true }, + }), + ]); + + const recipientCount = new Map( + recipientStatuses.map((row) => [row.status, row._count._all]), + ); + const emailCount = new Map( + emailStatuses.map((row) => [row.latestStatus, row._count._all]), + ); + + return { + pending: recipientCount.get("PENDING") ?? 0, + processed: + recipientStatuses.reduce((total, row) => total + row._count._all, 0) - + (recipientCount.get("PENDING") ?? 0), + queued: + (emailCount.get("QUEUED") ?? 0) + (emailCount.get("SCHEDULED") ?? 0), + sent: campaign.sent, + failed: emailCount.get("FAILED") ?? 0, + suppressed: + (recipientCount.get("SUPPRESSED") ?? 0) + + (recipientCount.get("SKIPPED") ?? 0), + }; + }, + ), + latestEmails: campaignProcedure.query( async ({ ctx: { db, team, campaign } }) => { const emails = await db.email.findMany({ @@ -266,6 +331,7 @@ export const campaignRouter = createTRPCRouter({ campaignId: z.string(), scheduledAt: z.union([z.string().datetime(), z.date()]).optional(), batchSize: z.number().min(1).max(100_000).optional(), + delivery: campaignDeliverySchema.optional(), }), ) .mutation(async ({ ctx: { team }, input }) => { @@ -274,6 +340,7 @@ export const campaignRouter = createTRPCRouter({ teamId: team.id, scheduledAt: input.scheduledAt, batchSize: input.batchSize, + delivery: input.delivery, }); return { ok: true }; }), diff --git a/apps/web/src/server/public-api/api/campaigns/create-campaign.ts b/apps/web/src/server/public-api/api/campaigns/create-campaign.ts index 6b056d31..4fab0835 100644 --- a/apps/web/src/server/public-api/api/campaigns/create-campaign.ts +++ b/apps/web/src/server/public-api/api/campaigns/create-campaign.ts @@ -55,6 +55,15 @@ function createCampaign(app: PublicAPIApp) { cc: body.cc, bcc: body.bcc, batchSize: body.batchSize, + delivery: body.delivery + ? body.delivery.strategy === "gradual" + ? { + strategy: "GRADUAL", + batchPercentage: body.delivery.batchPercentage, + interval: body.delivery.interval, + } + : { strategy: "ALL_AT_ONCE" } + : undefined, }); if (body.sendNow || body.scheduledAt) { @@ -67,6 +76,15 @@ function createCampaign(app: PublicAPIApp) { teamId: team.id, scheduledAt: scheduledAtInput, batchSize: body.batchSize, + delivery: body.delivery + ? body.delivery.strategy === "gradual" + ? { + strategy: "GRADUAL", + batchPercentage: body.delivery.batchPercentage, + interval: body.delivery.interval, + } + : { strategy: "ALL_AT_ONCE" } + : undefined, }); } diff --git a/apps/web/src/server/public-api/api/campaigns/schedule-campaign.ts b/apps/web/src/server/public-api/api/campaigns/schedule-campaign.ts index a626b4aa..336a1a25 100644 --- a/apps/web/src/server/public-api/api/campaigns/schedule-campaign.ts +++ b/apps/web/src/server/public-api/api/campaigns/schedule-campaign.ts @@ -5,9 +5,7 @@ import { CampaignScheduleInput, parseScheduledAt, } from "~/server/public-api/schemas/campaign-schema"; -import { - scheduleCampaign as scheduleCampaignService, -} from "~/server/service/campaign-service"; +import { scheduleCampaign as scheduleCampaignService } from "~/server/service/campaign-service"; const route = createRoute({ method: "post", path: "/v1/campaigns/{campaignId}/schedule", @@ -58,6 +56,15 @@ function scheduleCampaign(app: PublicAPIApp) { teamId: team.id, scheduledAt: parseScheduledAt(body.scheduledAt), batchSize: body.batchSize, + delivery: body.delivery + ? body.delivery.strategy === "gradual" + ? { + strategy: "GRADUAL", + batchPercentage: body.delivery.batchPercentage, + interval: body.delivery.interval, + } + : { strategy: "ALL_AT_ONCE" } + : undefined, }); return c.json({ success: true }); diff --git a/apps/web/src/server/public-api/schemas/campaign-schema.ts b/apps/web/src/server/public-api/schemas/campaign-schema.ts index 5ba10000..531f0b7a 100644 --- a/apps/web/src/server/public-api/schemas/campaign-schema.ts +++ b/apps/web/src/server/public-api/schemas/campaign-schema.ts @@ -7,6 +7,15 @@ const stringOrStringArray = z.union([ z.array(z.string().min(1)), ]); +export const campaignDeliverySchema = z.discriminatedUnion("strategy", [ + z.object({ strategy: z.literal("all_at_once") }), + z.object({ + strategy: z.literal("gradual"), + batchPercentage: z.number().int().min(1).max(50), + interval: z.enum(["minute", "hour"]), + }), +]); + export const parseScheduledAt = (scheduledAt?: string): Date | undefined => { if (!scheduledAt) return undefined; @@ -45,13 +54,14 @@ export const campaignCreateSchema = z .string() .optional() .describe( - "Timestamp in ISO 8601 format or natural language (e.g., 'tomorrow 9am', 'next monday 10:30')" + "Timestamp in ISO 8601 format or natural language (e.g., 'tomorrow 9am', 'next monday 10:30')", ), batchSize: z.number().int().min(1).max(100_000).optional(), + delivery: campaignDeliverySchema.optional(), }) .refine( (data) => !!data.content || !!data.html, - "Either content or html must be provided." + "Either content or html must be provided.", ); export const campaignScheduleSchema = z.object({ @@ -59,9 +69,10 @@ export const campaignScheduleSchema = z.object({ .string() .optional() .describe( - "Timestamp in ISO 8601 format or natural language (e.g., 'tomorrow 9am', 'next monday 10:30')" + "Timestamp in ISO 8601 format or natural language (e.g., 'tomorrow 9am', 'next monday 10:30')", ), batchSize: z.number().int().min(1).max(100_000).optional(), + delivery: campaignDeliverySchema.optional(), }); export type CampaignCreateInput = z.infer; @@ -80,6 +91,13 @@ export const campaignResponseSchema = z.object({ scheduledAt: z.string().datetime().nullable(), batchSize: z.number().int(), batchWindowMinutes: z.number().int(), + deliveryMode: z.enum(["ALL_AT_ONCE", "GRADUAL"]), + deliveryBatchPercentage: z.number().int().nullable(), + deliveryIntervalMinutes: z.number().int().nullable(), + deliveryBatchSize: z.number().int().nullable(), + currentDeliveryBatch: z.number().int(), + deliveryBatchProcessed: z.number().int(), + nextDeliveryAt: z.string().datetime().nullable(), total: z.number().int(), sent: z.number().int(), delivered: z.number().int(), diff --git a/apps/web/src/server/public-api/schemas/campaign-schema.unit.test.ts b/apps/web/src/server/public-api/schemas/campaign-schema.unit.test.ts new file mode 100644 index 00000000..eaf4409f --- /dev/null +++ b/apps/web/src/server/public-api/schemas/campaign-schema.unit.test.ts @@ -0,0 +1,47 @@ +import { describe, expect, it } from "vitest"; +import { + campaignCreateSchema, + campaignScheduleSchema, +} from "~/server/public-api/schemas/campaign-schema"; + +describe("campaign delivery API schemas", () => { + it("accepts gradual delivery when creating a campaign", () => { + const result = campaignCreateSchema.safeParse({ + name: "Product launch", + from: "hello@example.com", + subject: "We are live", + contactBookId: "book_1", + html: "

Hello

", + delivery: { + strategy: "gradual", + batchPercentage: 10, + interval: "hour", + }, + }); + + expect(result.success).toBe(true); + }); + + it.each([0, 51, 10.5])( + "rejects an invalid gradual percentage of %s", + (batchPercentage) => { + const result = campaignScheduleSchema.safeParse({ + delivery: { + strategy: "gradual", + batchPercentage, + interval: "minute", + }, + }); + + expect(result.success).toBe(false); + }, + ); + + it("accepts the all-at-once delivery strategy", () => { + const result = campaignScheduleSchema.safeParse({ + delivery: { strategy: "all_at_once" }, + }); + + expect(result.success).toBe(true); + }); +}); diff --git a/apps/web/src/server/service/campaign-service.ts b/apps/web/src/server/service/campaign-service.ts index 7a853478..2aca4aa3 100644 --- a/apps/web/src/server/service/campaign-service.ts +++ b/apps/web/src/server/service/campaign-service.ts @@ -119,6 +119,48 @@ function getCampaignDeliveryData({ }; } +function getCampaignDraftDeliveryData(delivery?: CampaignDeliveryInput) { + if (!delivery || delivery.strategy === "ALL_AT_ONCE") { + return { + deliveryMode: "ALL_AT_ONCE" as const, + deliveryBatchPercentage: null, + deliveryIntervalMinutes: null, + deliveryBatchSize: null, + }; + } + + return { + deliveryMode: "GRADUAL" as const, + deliveryBatchPercentage: delivery.batchPercentage, + deliveryIntervalMinutes: + GRADUAL_DELIVERY_INTERVAL_MINUTES[delivery.interval], + deliveryBatchSize: null, + }; +} + +function getStoredCampaignDelivery(campaign: Campaign): CampaignDeliveryInput { + if (campaign.deliveryMode !== "GRADUAL") { + return { strategy: "ALL_AT_ONCE" }; + } + + const interval = Object.entries(GRADUAL_DELIVERY_INTERVAL_MINUTES).find( + ([, minutes]) => minutes === campaign.deliveryIntervalMinutes, + )?.[0] as GradualDeliveryInterval | undefined; + + if (!campaign.deliveryBatchPercentage || !interval) { + throw new UnsendApiError({ + code: "BAD_REQUEST", + message: "Gradual delivery configuration is incomplete", + }); + } + + return { + strategy: "GRADUAL", + batchPercentage: campaign.deliveryBatchPercentage, + interval, + }; +} + async function prepareCampaignHtml( campaign: Campaign, ): Promise<{ campaign: Campaign; html: string }> { @@ -228,6 +270,7 @@ export async function createCampaignFromApi({ cc, bcc, batchSize, + delivery, }: { teamId: number; apiKeyId?: number; @@ -242,6 +285,7 @@ export async function createCampaignFromApi({ cc?: string | string[]; bcc?: string | string[]; batchSize?: number; + delivery?: CampaignDeliveryInput; }) { if (!content && !html) { throw new UnsendApiError({ @@ -326,6 +370,7 @@ export async function createCampaignFromApi({ bcc: sanitizeAddressList(bcc), teamId, domainId: domain.id, + ...getCampaignDraftDeliveryData(delivery), ...(typeof batchSize === "number" ? { batchSize } : {}), }, }); @@ -531,7 +576,7 @@ export async function scheduleCampaign({ const shouldResetCursor = campaign.status === "DRAFT"; const deliveryData = getCampaignDeliveryData({ - delivery, + delivery: delivery ?? getStoredCampaignDelivery(campaign), audienceSize: total, startsAt: scheduledAt, }); diff --git a/apps/web/src/server/service/campaign-service.unit.test.ts b/apps/web/src/server/service/campaign-service.unit.test.ts index 2ba0a112..48f68ad9 100644 --- a/apps/web/src/server/service/campaign-service.unit.test.ts +++ b/apps/web/src/server/service/campaign-service.unit.test.ts @@ -2,7 +2,12 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { createHash } from "crypto"; import { UnsubscribeReason } from "@prisma/client"; -const { mockDb, mockTx, mockUpdateContactSubscription } = vi.hoisted(() => { +const { + mockDb, + mockTx, + mockUpdateContactSubscription, + mockValidateDomainFromEmail, +} = vi.hoisted(() => { const mockTx = { campaignEmail: { findUnique: vi.fn(), @@ -28,12 +33,17 @@ const { mockDb, mockTx, mockUpdateContactSubscription } = vi.hoisted(() => { contact: { findUnique: vi.fn(), }, + contactBook: { + findUnique: vi.fn(), + }, campaign: { + create: vi.fn(), findUnique: vi.fn(), update: vi.fn(), }, }, mockUpdateContactSubscription: vi.fn(), + mockValidateDomainFromEmail: vi.fn(), }; }); @@ -81,7 +91,7 @@ vi.mock("~/server/service/suppression-service", () => ({ vi.mock("~/server/service/domain-service", () => ({ validateApiKeyDomainAccess: vi.fn(), - validateDomainFromEmail: vi.fn(), + validateDomainFromEmail: mockValidateDomainFromEmail, })); vi.mock("~/server/logger/log", () => ({ @@ -94,6 +104,7 @@ vi.mock("~/server/logger/log", () => ({ })); import { + createCampaignFromApi, recordCampaignContactFailure, resumeCampaign, scheduleCampaign, @@ -245,6 +256,35 @@ describe("campaign delivery lifecycle", () => { vi.useRealTimers(); }); + it("stores gradual settings on API-created drafts", async () => { + mockDb.contactBook.findUnique.mockResolvedValue({ id: "book_1" }); + mockValidateDomainFromEmail.mockResolvedValue({ id: 11 }); + mockDb.campaign.create.mockResolvedValue({ id: "campaign_1" }); + + await createCampaignFromApi({ + teamId: 7, + name: "Launch", + from: "sender@example.com", + subject: "Hello", + html: 'Unsubscribe', + contactBookId: "book_1", + delivery: { + strategy: "GRADUAL", + batchPercentage: 10, + interval: "hour", + }, + }); + + expect(mockDb.campaign.create).toHaveBeenCalledWith({ + data: expect.objectContaining({ + deliveryMode: "GRADUAL", + deliveryBatchPercentage: 10, + deliveryIntervalMinutes: 60, + deliveryBatchSize: null, + }), + }); + }); + it("does not allow delivery settings to change after sending starts", async () => { mockDb.campaign.findUnique.mockResolvedValue({ id: "campaign_1", From 5ee95c71cd2fe8f558aaf8031bc0c6146b4e6b6e Mon Sep 17 00:00:00 2001 From: Pankaj Beniwal Date: Tue, 21 Jul 2026 14:47:31 +0530 Subject: [PATCH 4/6] feat: add gradual delivery campaign controls --- .../campaigns/[campaignId]/page.tsx | 127 ++++++++- .../campaigns/schedule-campaign.tsx | 257 +++++++++++++++++- 2 files changed, 373 insertions(+), 11 deletions(-) diff --git a/apps/web/src/app/(dashboard)/campaigns/[campaignId]/page.tsx b/apps/web/src/app/(dashboard)/campaigns/[campaignId]/page.tsx index 93aeaf66..f360dac5 100644 --- a/apps/web/src/app/(dashboard)/campaigns/[campaignId]/page.tsx +++ b/apps/web/src/app/(dashboard)/campaigns/[campaignId]/page.tsx @@ -22,6 +22,7 @@ import { Button } from "@usesend/ui/src/button"; import { Card, CardContent, CardHeader, CardTitle } from "@usesend/ui/src/card"; import { EmailStatusBadge } from "../../emails/email-status-badge"; import { AnimatePresence, motion } from "framer-motion"; +import { Clock3, Send } from "lucide-react"; export default function CampaignDetailsPage({ params, @@ -45,7 +46,7 @@ export default function CampaignDetailsPage({ } return false; }, - } + }, ); const { data: latestEmails, isLoading: latestEmailsLoading } = @@ -53,9 +54,20 @@ export default function CampaignDetailsPage({ { campaignId: campaignId }, { refetchInterval: 5000, - } + }, ); + const { data: deliveryProgress } = api.campaign.getDeliveryProgress.useQuery( + { campaignId }, + { + refetchInterval: + campaign?.status === CampaignStatus.RUNNING || + campaign?.status === CampaignStatus.PAUSED + ? 5000 + : false, + }, + ); + if (isLoading) { return (
@@ -100,7 +112,31 @@ export default function CampaignDetailsPage({ ]; const total = campaign.total ?? 0; - const processed = campaign.sent ?? 0; + const processed = deliveryProgress?.processed ?? campaign.sent ?? 0; + const awaiting = Math.max(deliveryProgress?.pending ?? 0, total - processed); + const completionPercentage = + total > 0 ? Math.min(100, (processed / total) * 100) : 0; + const deliveryBatchSize = campaign.deliveryBatchSize ?? 0; + const totalDeliveryBatches = + deliveryBatchSize > 0 ? Math.ceil(total / deliveryBatchSize) : 0; + const currentDeliveryBatch = campaign.currentDeliveryBatch ?? 0; + const isGradualDelivery = campaign.deliveryMode === "GRADUAL"; + const nextDeliveryAt = campaign.nextDeliveryAt + ? new Date(campaign.nextDeliveryAt) + : null; + const nextDeliveryLabel = nextDeliveryAt + ? nextDeliveryAt.getTime() <= Date.now() + ? "Due now" + : formatDistanceToNow(nextDeliveryAt, { addSuffix: true }) + : null; + + const deliveryMetrics = [ + { label: "Awaiting", value: awaiting }, + { label: "Queued", value: deliveryProgress?.queued ?? 0 }, + { label: "Sent", value: deliveryProgress?.sent ?? campaign.sent ?? 0 }, + { label: "Failed", value: deliveryProgress?.failed ?? 0 }, + { label: "Suppressed", value: deliveryProgress?.suppressed ?? 0 }, + ]; return (
@@ -134,7 +170,90 @@ export default function CampaignDetailsPage({ )}
-
+
+ + +
+ Delivery +
+ {isGradualDelivery ? ( + + ) : ( + + )} + + {isGradualDelivery + ? (campaign.deliveryBatchPercentage ?? 0) + + "% every " + + (campaign.deliveryIntervalMinutes === 1 + ? "minute" + : "hour") + : "All at once"} + +
+
+ + {isGradualDelivery && totalDeliveryBatches > 0 ? ( +
+
+ {currentDeliveryBatch > 0 ? ( + <> + Wave {currentDeliveryBatch} of {totalDeliveryBatches} + + ) : ( + <>{totalDeliveryBatches} waves planned + )} +
+ {nextDeliveryLabel && campaign.status === "RUNNING" ? ( +
+ Next wave {nextDeliveryLabel} +
+ ) : null} +
+ ) : null} + + + +
+
+ + {processed.toLocaleString()} of {total.toLocaleString()}{" "} + processed + + + {completionPercentage.toFixed(0)}% + +
+
+
+
+
+ +
+ {deliveryMetrics.map((metric) => ( +
+
+ {metric.label} +
+
+ {metric.value.toLocaleString()} +
+
+ ))} +
+ + +
diff --git a/apps/web/src/app/(dashboard)/campaigns/schedule-campaign.tsx b/apps/web/src/app/(dashboard)/campaigns/schedule-campaign.tsx index 3d34980d..f19fbb01 100644 --- a/apps/web/src/app/(dashboard)/campaigns/schedule-campaign.tsx +++ b/apps/web/src/app/(dashboard)/campaigns/schedule-campaign.tsx @@ -18,11 +18,25 @@ import * as chrono from "chrono-node"; import { api } from "~/trpc/react"; import React, { useEffect, useMemo, useRef, useState } from "react"; import { toast } from "@usesend/ui/src/toaster"; -import { Calendar as CalendarIcon } from "lucide-react"; +import { Calendar as CalendarIcon, Clock3, Send } from "lucide-react"; import { Calendar } from "@usesend/ui/src/calendar"; -import { Campaign } from "@prisma/client"; +import type { Campaign } from "@prisma/client"; import { format } from "date-fns"; import { Spinner } from "@usesend/ui/src/spinner"; +import { + Select, + SelectContent, + SelectItem, + SelectTrigger, + SelectValue, +} from "@usesend/ui/src/select"; +import { + calculateGradualDelivery, + GRADUAL_DELIVERY_INTERVAL_MINUTES, +} from "~/lib/campaign-delivery"; +import type { GradualDeliveryInterval } from "~/lib/campaign-delivery"; + +type DeliveryMode = "ALL_AT_ONCE" | "GRADUAL"; export const ScheduleCampaign: React.FC<{ campaign: Partial & { id: string }; @@ -42,7 +56,21 @@ export const ScheduleCampaign: React.FC<{ ); const [isConfirmNow, setIsConfirmNow] = useState(false); const [error, setError] = useState(null); + const [deliveryMode, setDeliveryMode] = useState( + campaign.deliveryMode ?? "ALL_AT_ONCE", + ); + const [batchPercentage, setBatchPercentage] = useState( + String(campaign.deliveryBatchPercentage ?? 10), + ); + const [deliveryInterval, setDeliveryInterval] = + useState( + campaign.deliveryIntervalMinutes === 1 ? "minute" : "hour", + ); const scheduleMutation = api.campaign.scheduleCampaign.useMutation(); + const audienceQuery = api.campaign.getAudienceCount.useQuery( + { campaignId: campaign.id }, + { enabled: open }, + ); const utils = api.useUtils(); const scheduledAtTimestamp = campaign.scheduledAt ? new Date(campaign.scheduledAt).getTime() @@ -51,6 +79,12 @@ export const ScheduleCampaign: React.FC<{ useEffect(() => { if (!open) return; + setDeliveryMode(campaign.deliveryMode ?? "ALL_AT_ONCE"); + setBatchPercentage(String(campaign.deliveryBatchPercentage ?? 10)); + setDeliveryInterval( + campaign.deliveryIntervalMinutes === 1 ? "minute" : "hour", + ); + if (scheduledAtTimestamp != null) { const scheduledDate = new Date(scheduledAtTimestamp); setSelectedDate(scheduledDate); @@ -61,15 +95,66 @@ export const ScheduleCampaign: React.FC<{ const now = new Date(); setSelectedDate(now); setScheduleInput(""); - }, [open, scheduledAtTimestamp]); + }, [ + campaign.deliveryBatchPercentage, + campaign.deliveryIntervalMinutes, + campaign.deliveryMode, + open, + scheduledAtTimestamp, + ]); + + const parsedBatchPercentage = Number(batchPercentage); + const isBatchPercentageValid = + batchPercentage.trim().length > 0 && + Number.isInteger(parsedBatchPercentage) && + parsedBatchPercentage >= 1 && + parsedBatchPercentage <= 50; + + const deliveryEstimate = useMemo(() => { + if ( + deliveryMode !== "GRADUAL" || + !isBatchPercentageValid || + audienceQuery.data == null + ) { + return null; + } + + return calculateGradualDelivery({ + audienceSize: audienceQuery.data.total, + batchPercentage: parsedBatchPercentage, + intervalMinutes: GRADUAL_DELIVERY_INTERVAL_MINUTES[deliveryInterval], + startsAt: selectedDate ?? new Date(), + }); + }, [ + audienceQuery.data, + batchPercentage, + deliveryInterval, + deliveryMode, + isBatchPercentageValid, + selectedDate, + ]); const onSchedule = (scheduledAt?: Date) => { if (error) setError(null); + + if (deliveryMode === "GRADUAL" && !isBatchPercentageValid) { + setError("Batch percentage must be a whole number between 1 and 50"); + return; + } + scheduleMutation.mutate( { campaignId: campaign.id, // Never send free text to backend; only a Date scheduledAt: scheduledAt ? scheduledAt : undefined, + delivery: + deliveryMode === "GRADUAL" + ? { + strategy: "GRADUAL", + batchPercentage: parsedBatchPercentage, + interval: deliveryInterval, + } + : { strategy: "ALL_AT_ONCE" }, }, { onSuccess: () => { @@ -162,11 +247,14 @@ export const ScheduleCampaign: React.FC<{ - + Schedule Campaign -
+
+
+
+
Delivery
+
+ Choose how quickly recipients enter the sending queue. +
+
+ +
+ + + +
+ + {deliveryMode === "GRADUAL" ? ( +
+
+
+ +
+ + setBatchPercentage(event.target.value) + } + aria-invalid={!isBatchPercentageValid} + className="pr-8 font-mono" + /> + + % + +
+ {!isBatchPercentageValid ? ( +

+ Enter a whole number from 1 to 50. +

+ ) : null} +
+ +
+ + +
+
+ +
+ {audienceQuery.isLoading ? ( +
+ + Calculating delivery plan +
+ ) : deliveryEstimate && deliveryEstimate.batchSize > 0 ? ( +
+
+
+
+ Per wave +
+
+ {deliveryEstimate.batchSize.toLocaleString()} +
+
+
+
+ Waves +
+
+ {deliveryEstimate.totalBatches.toLocaleString()} +
+
+
+
+ Estimated end +
+
+ {format(deliveryEstimate.completesAt, "MMM d, p")} +
+
+
+

+ Based on the current subscribed audience. The final + audience is captured when sending starts. +

+
+ ) : ( +

+ No subscribed recipients to preview yet. +

+ )} +
+
+ ) : null} +
+ {error && (
{error} @@ -261,14 +498,17 @@ export const ScheduleCampaign: React.FC<{ {isConfirmNow ? (
- Are you sure you want to send this campaign now? + Are you sure you want to start this campaign now? @@ -298,6 +538,9 @@ export const ScheduleCampaign: React.FC<{ }} isLoading={scheduleMutation.isPending} showSpinner={true} + disabled={ + deliveryMode === "GRADUAL" && !isBatchPercentageValid + } > {scheduleMutation.isPending ? "Scheduling" : "Schedule"} From 5dc089a30c1b663733506f1d141a74967bf74c0e Mon Sep 17 00:00:00 2001 From: Pankaj Beniwal Date: Tue, 21 Jul 2026 21:34:24 +0530 Subject: [PATCH 5/6] fix: harden gradual campaign scheduling and delivery --- apps/docs/api-reference/openapi.json | 156 +++ .../migration.sql | 12 +- .../migration.sql | 4 + apps/web/prisma/schema.prisma | 2 + .../campaigns/[campaignId]/page.tsx | 53 +- .../campaigns/schedule-campaign.tsx | 42 +- .../campaigns/toggle-pause-campaign.tsx | 10 +- .../src/lib/campaign-delivery.unit.test.ts | 4 +- .../routers/campaign-delivery.trpc.test.ts | 12 +- apps/web/src/server/api/routers/campaign.ts | 14 +- .../src/server/jobs/campaign-scheduler-job.ts | 143 +-- .../jobs/campaign-scheduler-job.unit.test.ts | 106 ++ .../api/campaigns/create-campaign.ts | 2 +- .../src/server/service/campaign-service.ts | 966 ++++++++++++------ .../service/campaign-service.unit.test.ts | 485 ++++++++- packages/python-sdk/README.md | 5 + packages/python-sdk/tests/test_resources.py | 38 +- packages/python-sdk/usesend/types.py | 33 + packages/sdk/types/schema.d.ts | 47 + 19 files changed, 1737 insertions(+), 397 deletions(-) create mode 100644 apps/web/prisma/migrations/20260721190000_add_campaign_email_status_index/migration.sql create mode 100644 apps/web/src/server/jobs/campaign-scheduler-job.unit.test.ts diff --git a/apps/docs/api-reference/openapi.json b/apps/docs/api-reference/openapi.json index ce39eb17..c2d886dc 100644 --- a/apps/docs/api-reference/openapi.json +++ b/apps/docs/api-reference/openapi.json @@ -2085,6 +2085,39 @@ "type": "integer", "minimum": 1, "maximum": 100000 + }, + "delivery": { + "oneOf": [ + { + "type": "object", + "properties": { + "strategy": { + "type": "string", + "enum": ["all_at_once"] + } + }, + "required": ["strategy"] + }, + { + "type": "object", + "properties": { + "strategy": { + "type": "string", + "enum": ["gradual"] + }, + "batchPercentage": { + "type": "integer", + "minimum": 1, + "maximum": 50 + }, + "interval": { + "type": "string", + "enum": ["minute", "hour"] + } + }, + "required": ["strategy", "batchPercentage", "interval"] + } + ] } }, "required": ["name", "from", "subject", "contactBookId"] @@ -2116,6 +2149,29 @@ }, "batchSize": { "type": "integer" }, "batchWindowMinutes": { "type": "integer" }, + "deliveryMode": { + "type": "string", + "enum": ["ALL_AT_ONCE", "GRADUAL"] + }, + "deliveryBatchPercentage": { + "type": "integer", + "nullable": true + }, + "deliveryIntervalMinutes": { + "type": "integer", + "nullable": true + }, + "deliveryBatchSize": { + "type": "integer", + "nullable": true + }, + "currentDeliveryBatch": { "type": "integer" }, + "deliveryBatchProcessed": { "type": "integer" }, + "nextDeliveryAt": { + "type": "string", + "nullable": true, + "format": "date-time" + }, "total": { "type": "integer" }, "sent": { "type": "integer" }, "delivered": { "type": "integer" }, @@ -2147,6 +2203,13 @@ "scheduledAt", "batchSize", "batchWindowMinutes", + "deliveryMode", + "deliveryBatchPercentage", + "deliveryIntervalMinutes", + "deliveryBatchSize", + "currentDeliveryBatch", + "deliveryBatchProcessed", + "nextDeliveryAt", "total", "sent", "delivered", @@ -2302,6 +2365,29 @@ }, "batchSize": { "type": "integer" }, "batchWindowMinutes": { "type": "integer" }, + "deliveryMode": { + "type": "string", + "enum": ["ALL_AT_ONCE", "GRADUAL"] + }, + "deliveryBatchPercentage": { + "type": "integer", + "nullable": true + }, + "deliveryIntervalMinutes": { + "type": "integer", + "nullable": true + }, + "deliveryBatchSize": { + "type": "integer", + "nullable": true + }, + "currentDeliveryBatch": { "type": "integer" }, + "deliveryBatchProcessed": { "type": "integer" }, + "nextDeliveryAt": { + "type": "string", + "nullable": true, + "format": "date-time" + }, "total": { "type": "integer" }, "sent": { "type": "integer" }, "delivered": { "type": "integer" }, @@ -2333,6 +2419,13 @@ "scheduledAt", "batchSize", "batchWindowMinutes", + "deliveryMode", + "deliveryBatchPercentage", + "deliveryIntervalMinutes", + "deliveryBatchSize", + "currentDeliveryBatch", + "deliveryBatchProcessed", + "nextDeliveryAt", "total", "sent", "delivered", @@ -2391,6 +2484,29 @@ }, "batchSize": { "type": "integer" }, "batchWindowMinutes": { "type": "integer" }, + "deliveryMode": { + "type": "string", + "enum": ["ALL_AT_ONCE", "GRADUAL"] + }, + "deliveryBatchPercentage": { + "type": "integer", + "nullable": true + }, + "deliveryIntervalMinutes": { + "type": "integer", + "nullable": true + }, + "deliveryBatchSize": { + "type": "integer", + "nullable": true + }, + "currentDeliveryBatch": { "type": "integer" }, + "deliveryBatchProcessed": { "type": "integer" }, + "nextDeliveryAt": { + "type": "string", + "nullable": true, + "format": "date-time" + }, "total": { "type": "integer" }, "sent": { "type": "integer" }, "delivered": { "type": "integer" }, @@ -2422,6 +2538,13 @@ "scheduledAt", "batchSize", "batchWindowMinutes", + "deliveryMode", + "deliveryBatchPercentage", + "deliveryIntervalMinutes", + "deliveryBatchSize", + "currentDeliveryBatch", + "deliveryBatchProcessed", + "nextDeliveryAt", "total", "sent", "delivered", @@ -2473,6 +2596,39 @@ "type": "integer", "minimum": 1, "maximum": 100000 + }, + "delivery": { + "oneOf": [ + { + "type": "object", + "properties": { + "strategy": { + "type": "string", + "enum": ["all_at_once"] + } + }, + "required": ["strategy"] + }, + { + "type": "object", + "properties": { + "strategy": { + "type": "string", + "enum": ["gradual"] + }, + "batchPercentage": { + "type": "integer", + "minimum": 1, + "maximum": 50 + }, + "interval": { + "type": "string", + "enum": ["minute", "hour"] + } + }, + "required": ["strategy", "batchPercentage", "interval"] + } + ] } } } diff --git a/apps/web/prisma/migrations/20260721120000_add_gradual_campaign_sending/migration.sql b/apps/web/prisma/migrations/20260721120000_add_gradual_campaign_sending/migration.sql index f226bde9..a35ddf27 100644 --- a/apps/web/prisma/migrations/20260721120000_add_gradual_campaign_sending/migration.sql +++ b/apps/web/prisma/migrations/20260721120000_add_gradual_campaign_sending/migration.sql @@ -2,17 +2,15 @@ CREATE TYPE "CampaignDeliveryMode" AS ENUM ('ALL_AT_ONCE', 'GRADUAL'); -- CreateEnum -CREATE TYPE "CampaignRecipientStatus" AS ENUM ('PENDING', 'QUEUED', 'SUPPRESSED', 'SKIPPED', 'FAILED'); +CREATE TYPE "CampaignRecipientStatus" AS ENUM ('PENDING', 'PROCESSING', 'QUEUED', 'SUPPRESSED', 'SKIPPED', 'FAILED'); -- AlterTable ALTER TABLE "CampaignEmail" ALTER COLUMN "emailId" DROP NOT NULL; -ALTER TABLE "CampaignEmail" ADD COLUMN "status" "CampaignRecipientStatus"; +-- PostgreSQL 11+ stores a constant default as metadata instead of rewriting +-- every existing row. Existing recipients are already queued, while new rows +-- created by the gradual delivery worker should default to pending. +ALTER TABLE "CampaignEmail" ADD COLUMN "status" "CampaignRecipientStatus" NOT NULL DEFAULT 'QUEUED'; ALTER TABLE "CampaignEmail" ADD COLUMN "processedAt" TIMESTAMP(3); - -UPDATE "CampaignEmail" -SET "status" = 'QUEUED', "processedAt" = "createdAt"; - -ALTER TABLE "CampaignEmail" ALTER COLUMN "status" SET NOT NULL; ALTER TABLE "CampaignEmail" ALTER COLUMN "status" SET DEFAULT 'PENDING'; -- AlterTable diff --git a/apps/web/prisma/migrations/20260721190000_add_campaign_email_status_index/migration.sql b/apps/web/prisma/migrations/20260721190000_add_campaign_email_status_index/migration.sql new file mode 100644 index 00000000..f1bc1458 --- /dev/null +++ b/apps/web/prisma/migrations/20260721190000_add_campaign_email_status_index/migration.sql @@ -0,0 +1,4 @@ +-- CreateIndex +-- Keep this as the migration's only SQL statement so Prisma runs it outside a +-- transaction and PostgreSQL can build it without blocking inserts or updates. +CREATE INDEX CONCURRENTLY "Email_campaignId_latestStatus_idx" ON "Email"("campaignId", "latestStatus"); diff --git a/apps/web/prisma/schema.prisma b/apps/web/prisma/schema.prisma index ff446848..b6283407 100644 --- a/apps/web/prisma/schema.prisma +++ b/apps/web/prisma/schema.prisma @@ -267,6 +267,7 @@ model Email { emailEvents EmailEvent[] @@index([campaignId, contactId]) + @@index([campaignId, latestStatus]) @@index([createdAt(sort: Desc)]) } @@ -359,6 +360,7 @@ enum CampaignDeliveryMode { enum CampaignRecipientStatus { PENDING + PROCESSING QUEUED SUPPRESSED SKIPPED diff --git a/apps/web/src/app/(dashboard)/campaigns/[campaignId]/page.tsx b/apps/web/src/app/(dashboard)/campaigns/[campaignId]/page.tsx index f360dac5..b8768ee1 100644 --- a/apps/web/src/app/(dashboard)/campaigns/[campaignId]/page.tsx +++ b/apps/web/src/app/(dashboard)/campaigns/[campaignId]/page.tsx @@ -39,6 +39,7 @@ export default function CampaignDetailsPage({ if (!c) return false; if ( + c.status === CampaignStatus.SCHEDULED || c.status === CampaignStatus.RUNNING || c.status === CampaignStatus.PAUSED ) { @@ -60,11 +61,29 @@ export default function CampaignDetailsPage({ const { data: deliveryProgress } = api.campaign.getDeliveryProgress.useQuery( { campaignId }, { - refetchInterval: - campaign?.status === CampaignStatus.RUNNING || - campaign?.status === CampaignStatus.PAUSED - ? 5000 - : false, + refetchInterval: (query) => { + if ( + campaign?.status === CampaignStatus.SCHEDULED || + campaign?.status === CampaignStatus.RUNNING + ) { + return 5000; + } + + const progress = query.state.data; + if ( + campaign?.status === CampaignStatus.PAUSED && + progress?.processing + ) { + return 5000; + } + + const needsFinalSnapshot = + campaign?.status === CampaignStatus.SENT && + progress != null && + progress.pending > 0; + + return needsFinalSnapshot ? 5000 : false; + }, }, ); @@ -121,6 +140,10 @@ export default function CampaignDetailsPage({ deliveryBatchSize > 0 ? Math.ceil(total / deliveryBatchSize) : 0; const currentDeliveryBatch = campaign.currentDeliveryBatch ?? 0; const isGradualDelivery = campaign.deliveryMode === "GRADUAL"; + const showDeliveryCard = + campaign.status === CampaignStatus.SCHEDULED || + campaign.status === CampaignStatus.RUNNING || + campaign.status === CampaignStatus.PAUSED; const nextDeliveryAt = campaign.nextDeliveryAt ? new Date(campaign.nextDeliveryAt) : null; @@ -138,6 +161,18 @@ export default function CampaignDetailsPage({ { label: "Suppressed", value: deliveryProgress?.suppressed ?? 0 }, ]; + const completedDeliverySummary = [ + isGradualDelivery && totalDeliveryBatches > 0 + ? `Sent in ${totalDeliveryBatches.toLocaleString()} ${ + totalDeliveryBatches === 1 ? "wave" : "waves" + }` + : "Sent all at once", + `${processed.toLocaleString()} processed`, + ...(deliveryProgress?.failed + ? [`${deliveryProgress.failed.toLocaleString()} failed`] + : []), + ].join(" · "); + return (
@@ -171,7 +206,7 @@ export default function CampaignDetailsPage({
- +