diff --git a/apps/web/app/(ee)/api/campaigns/[campaignId]/route.ts b/apps/web/app/(ee)/api/campaigns/[campaignId]/route.ts index a5ded7839a0..288cf9d3d6b 100644 --- a/apps/web/app/(ee)/api/campaigns/[campaignId]/route.ts +++ b/apps/web/app/(ee)/api/campaigns/[campaignId]/route.ts @@ -1,8 +1,5 @@ import { getCampaignOrThrow } from "@/lib/api/campaigns/get-campaign-or-throw"; -import { - deleteCampaignSchedule, - scheduleCampaign, -} from "@/lib/api/campaigns/schedule-campaigns"; +import { shouldEnqueueDueMarketingBroadcast } from "@/lib/api/campaigns/marketing-campaign-broadcast"; import { campaignEligibilityIncludes, transformCampaign, @@ -14,12 +11,13 @@ import { getDefaultProgramIdOrThrow } from "@/lib/api/programs/get-default-progr import { parseRequestBody } from "@/lib/api/utils"; import { validateWorkflowConditions } from "@/lib/api/workflows/validate-workflow-conditions"; import { withWorkspace } from "@/lib/auth"; +import { qstash } from "@/lib/cron"; import { prisma } from "@/lib/prisma"; import { CampaignSchema, updateCampaignSchema, } from "@/lib/zod/schemas/campaigns"; -import { arrayEqual, pluck } from "@dub/utils"; +import { APP_DOMAIN_WITH_NGROK, arrayEqual, pluck } from "@dub/utils"; import { PartnerGroup } from "@prisma/client"; import { waitUntil } from "@vercel/functions"; import { NextResponse } from "next/server"; @@ -191,12 +189,25 @@ export const PATCH = withWorkspace( }); }); - waitUntil( - scheduleCampaign({ - campaign, - updatedCampaign, - }), - ); + if ( + shouldEnqueueDueMarketingBroadcast({ + previous: campaign, + next: updatedCampaign, + }) + ) { + waitUntil( + qstash.publishJSON({ + url: `${APP_DOMAIN_WITH_NGROK}/api/cron/campaigns/broadcast`, + flowControl: { + key: `broadcast-marketing-campaign-${campaignId}`, + parallelism: 1, + }, + body: { + campaignId, + }, + }), + ); + } return NextResponse.json( CampaignSchema.parse(transformCampaign(updatedCampaign)), @@ -217,15 +228,6 @@ export const DELETE = withWorkspace( const campaign = await getCampaignOrThrow({ programId, campaignId, - include: { - workflow: { - select: { - id: true, - actions: true, - triggerConditions: true, - }, - }, - }, }); await prisma.$transaction(async (tx) => { @@ -244,8 +246,6 @@ export const DELETE = withWorkspace( } }); - waitUntil(deleteCampaignSchedule(campaign)); - return NextResponse.json({ id: campaignId }); }, { diff --git a/apps/web/app/(ee)/api/cron/campaigns/broadcast/route.ts b/apps/web/app/(ee)/api/cron/campaigns/broadcast/route.ts index ced6b7dd481..9cae722a3b5 100644 --- a/apps/web/app/(ee)/api/cron/campaigns/broadcast/route.ts +++ b/apps/web/app/(ee)/api/cron/campaigns/broadcast/route.ts @@ -3,7 +3,7 @@ import { renderCampaignEmailHTML } from "@/lib/api/campaigns/render-campaign-ema import { campaignEligibilityIncludes } from "@/lib/api/campaigns/transform-campaign"; import { validateCampaignFromAddress } from "@/lib/api/campaigns/validate-campaign"; import { createId } from "@/lib/api/create-id"; -import { handleAndReturnErrorResponse } from "@/lib/api/errors"; +import { DubApiError, handleAndReturnErrorResponse } from "@/lib/api/errors"; import { qstash } from "@/lib/cron"; import { verifyQstashSignature } from "@/lib/cron/verify-qstash"; import { resolveCampaignFromAddress } from "@/lib/email/parse-campaign-from-address"; @@ -13,9 +13,13 @@ import { ACTIVE_ENROLLMENT_STATUSES } from "@/lib/zod/schemas/partners"; import { sendBatchEmail } from "@dub/email"; import CampaignEmail from "@dub/email/templates/campaign-email"; import { APP_DOMAIN_WITH_NGROK, chunk, log, pluck } from "@dub/utils"; -import { NotificationEmailType } from "@prisma/client"; +import { + Campaign, + CampaignStatus, + EmailDomain, + NotificationEmailType, +} from "@prisma/client"; import { differenceInMinutes } from "date-fns"; -import { headers } from "next/headers"; import * as z from "zod/v4"; import { logAndRespond } from "../../utils"; @@ -102,46 +106,44 @@ export async function POST(req: Request) { } } - // This is a safety check to ensure the campaign broadcast is not "initiated" multiple times - const headersList = await headers(); - const upstashMessageId = headersList.get("Upstash-Message-Id"); - - if ( - !startingAfter && // First run - campaign.qstashMessageId && - upstashMessageId !== campaign.qstashMessageId - ) { - return logAndRespond( - `Campaign ${campaignId} broadcast was skipped because it is not the current message being processed.`, - ); - } - const program = campaign.program; - // TODO: We should make the from address required. There are existing campaign without from address - if (campaign.from) { - validateCampaignFromAddress({ - campaign, - emailDomains: program.emailDomains, + // Claim the first run so leftover delayed messages / scanner retries + // cannot start a second broadcast. A QStash retry of the claiming + // message (matching qstashMessageId) is allowed to continue. + if (!startingAfter) { + const messageId = req.headers.get("Upstash-Message-Id"); + + const claimed = await prisma.campaign.updateMany({ + where: { + id: campaignId, + status: CampaignStatus.scheduled, + OR: [{ scheduledAt: null }, { scheduledAt: { lte: new Date() } }], + }, + data: { + status: CampaignStatus.sending, + qstashMessageId: messageId, + }, }); - } - // Mark the campaign as sending (if it's in scheduled status) - if (campaign.status === "scheduled") { - try { - await prisma.campaign.update({ - where: { - id: campaignId, - }, - data: { - status: "sending", - }, - }); - } catch (error) { - // + if (claimed.count === 0) { + if (!messageId || campaign.qstashMessageId !== messageId) { + return logAndRespond( + `Campaign ${campaignId} broadcast already initiated. Skipping...`, + ); + } } } + const invalidFromResponse = await cancelCampaignIfInvalidFromAddress({ + campaign, + emailDomains: program.emailDomains, + }); + + if (invalidFromResponse) { + return invalidFromResponse; + } + const campaignGroupIds = pluck(campaign.groups, "groupId"); const campaignPartnerTagIds = pluck(campaign.partnerTags, "partnerTagId"); @@ -376,3 +378,47 @@ export async function POST(req: Request) { return handleAndReturnErrorResponse(error); } } + +async function cancelCampaignIfInvalidFromAddress({ + campaign, + emailDomains, +}: { + campaign: Pick; + emailDomains: Pick[]; +}) { + if (!campaign.from) { + return; + } + + try { + validateCampaignFromAddress({ + campaign, + emailDomains, + }); + } catch (error) { + if (!(error instanceof DubApiError)) { + throw error; + } + + await prisma.campaign.updateMany({ + where: { + id: campaign.id, + status: { + in: [CampaignStatus.scheduled, CampaignStatus.sending], + }, + }, + data: { + status: CampaignStatus.canceled, + }, + }); + + await log({ + type: "errors", + message: `Campaign ${campaign.id} canceled: ${error.message}`, + }); + + return logAndRespond( + `Campaign ${campaign.id} canceled: invalid from address.`, + ); + } +} diff --git a/apps/web/app/(ee)/api/cron/campaigns/queue-scheduled/route.ts b/apps/web/app/(ee)/api/cron/campaigns/queue-scheduled/route.ts new file mode 100644 index 00000000000..f9f93be5b09 --- /dev/null +++ b/apps/web/app/(ee)/api/cron/campaigns/queue-scheduled/route.ts @@ -0,0 +1,178 @@ +import { isScheduledWorkflow } from "@/lib/api/workflows/utils"; +import { CRON_BATCH_SIZE } from "@/lib/cron"; +import { enqueueBatchJobs } from "@/lib/cron/enqueue-batch-jobs"; +import { withCron } from "@/lib/cron/with-cron"; +import { prisma } from "@/lib/prisma"; +import { + APP_DOMAIN_WITH_NGROK, + isRejected, + log, + serializeError, +} from "@dub/utils"; +import { CampaignStatus, CampaignType } from "@prisma/client"; +import { logAndRespond } from "../../utils"; + +export const dynamic = "force-dynamic"; +export const maxDuration = 600; + +// GET /api/cron/campaigns/queue-scheduled +// Fans out due marketing broadcasts and (on the 12h tick) scheduled transactional workflows. +export const GET = withCron(async () => { + const now = new Date(); + + const [transactional, marketing] = await Promise.allSettled([ + queueTransactionalCampaigns(now), + queueMarketingCampaigns(now), + ]); + + const failures: string[] = []; + + if (isRejected(transactional)) { + failures.push(`transactional: ${serializeError(transactional.reason)}`); + } + + if (isRejected(marketing)) { + failures.push(`marketing: ${serializeError(marketing.reason)}`); + } + + if (failures.length > 0) { + const message = `Campaign queueing partially failed: ${failures.join("; ")}`; + await log({ type: "errors", message }); + return logAndRespond(message, { logLevel: "error" }); + } + + const transactionalQueued = + transactional.status === "fulfilled" ? transactional.value : 0; + const marketingQueued = + marketing.status === "fulfilled" ? marketing.value : 0; + + if (transactionalQueued + marketingQueued === 0) { + return logAndRespond("No campaigns to queue."); + } + + return logAndRespond( + `Queued ${marketingQueued} marketing and ${transactionalQueued} transactional campaigns.`, + ); +}); + +// First 5 minutes of 00:00/12:00 UTC. QStash dedup lasts 10 minutes, so this +// absorbs Vercel cron jitter without leaking a second publish after expiry. +function isTransactionalTick(now: Date) { + return now.getUTCHours() % 12 === 0 && now.getUTCMinutes() < 5; +} + +async function queueTransactionalCampaigns(now: Date) { + // Matches the 12h enrollment window in executeSendCampaignWorkflow. + // 5-minute window absorbs Vercel cron jitter; QStash dedup (10 min) collapses extra publishes. + if (!isTransactionalTick(now)) { + return 0; + } + + let queued = 0; + let lastCampaignId: string | undefined; + + while (true) { + const campaigns = await prisma.campaign.findMany({ + where: { + type: CampaignType.transactional, + status: CampaignStatus.active, + workflow: { + disabledAt: null, + }, + ...(lastCampaignId && { id: { gt: lastCampaignId } }), + }, + select: { + id: true, + workflow: { + select: { + id: true, + triggerConditions: true, + actions: true, + }, + }, + }, + take: CRON_BATCH_SIZE, + orderBy: { + id: "asc", + }, + }); + + if (campaigns.length === 0) { + break; + } + + const scheduledWorkflows = campaigns.flatMap((campaign) => + campaign.workflow && isScheduledWorkflow(campaign.workflow) + ? [campaign.workflow] + : [], + ); + + if (scheduledWorkflows.length > 0) { + await enqueueBatchJobs( + scheduledWorkflows.map((workflow) => ({ + url: `${APP_DOMAIN_WITH_NGROK}/api/cron/workflows/${workflow.id}`, + deduplicationId: workflow.id, + label: "execute-scheduled-workflow", + flowControl: { + key: "execute-scheduled-workflow", + parallelism: 10, + }, + body: {}, + })), + ); + + queued += scheduledWorkflows.length; + } + + lastCampaignId = campaigns[campaigns.length - 1].id; + } + + return queued; +} + +async function queueMarketingCampaigns(now: Date) { + let queued = 0; + let lastCampaignId: string | undefined; + + while (true) { + const campaigns = await prisma.campaign.findMany({ + where: { + type: CampaignType.marketing, + // Do not reclaim `sending` campaigns; failures Slack-alert and we resume them manually. + status: CampaignStatus.scheduled, + OR: [{ scheduledAt: null }, { scheduledAt: { lte: now } }], + ...(lastCampaignId && { id: { gt: lastCampaignId } }), + }, + select: { + id: true, + }, + take: CRON_BATCH_SIZE, + orderBy: { + id: "asc", + }, + }); + + if (campaigns.length === 0) { + break; + } + + await enqueueBatchJobs( + campaigns.map((campaign) => ({ + url: `${APP_DOMAIN_WITH_NGROK}/api/cron/campaigns/broadcast`, + label: "broadcast-marketing-campaign", + flowControl: { + key: `broadcast-marketing-campaign-${campaign.id}`, + parallelism: 1, + }, + body: { + campaignId: campaign.id, + }, + })), + ); + + queued += campaigns.length; + lastCampaignId = campaigns[campaigns.length - 1].id; + } + + return queued; +} diff --git a/apps/web/app/(ee)/api/embed/referrals/tremendous/send-otp/route.ts b/apps/web/app/(ee)/api/embed/referrals/tremendous/send-otp/route.ts index 2364e85afbd..d7fa8907835 100644 --- a/apps/web/app/(ee)/api/embed/referrals/tremendous/send-otp/route.ts +++ b/apps/web/app/(ee)/api/embed/referrals/tremendous/send-otp/route.ts @@ -5,10 +5,7 @@ import { EMAIL_OTP_EXPIRY_IN } from "@/lib/auth/constants"; import { extractEmailDomain } from "@/lib/email/extract-email-domain"; import { withReferralsEmbedToken } from "@/lib/embed/referrals/auth"; import { prisma } from "@/lib/prisma"; -import { - TREMENDOUS_ENABLED_PROGRAM_IDS, - TREMENDOUS_PROHIBITED_TOP_LEVEL_DOMAINS, -} from "@/lib/tremendous/constants"; +import { TREMENDOUS_PROHIBITED_TOP_LEVEL_DOMAINS } from "@/lib/tremendous/constants"; import { ratelimit, redis } from "@/lib/upstash"; import { emailSchema } from "@/lib/zod/schemas/auth"; import { ACTIVE_ENROLLMENT_STATUSES } from "@/lib/zod/schemas/partners"; @@ -33,13 +30,6 @@ export const POST = withReferralsEmbedToken( }); } - if (!TREMENDOUS_ENABLED_PROGRAM_IDS.includes(programEnrollment.programId)) { - throw new DubApiError({ - code: "forbidden", - message: "Gift card payouts are not available for this program.", - }); - } - const { email } = sendOtpSchema.parse(await parseRequestBody(req)); const { partnerId } = programEnrollment; diff --git a/apps/web/app/(ee)/api/embed/referrals/tremendous/verify-otp/route.ts b/apps/web/app/(ee)/api/embed/referrals/tremendous/verify-otp/route.ts index 0591fa3088d..ec1e06af65d 100644 --- a/apps/web/app/(ee)/api/embed/referrals/tremendous/verify-otp/route.ts +++ b/apps/web/app/(ee)/api/embed/referrals/tremendous/verify-otp/route.ts @@ -3,10 +3,7 @@ import { parseRequestBody } from "@/lib/api/utils"; import { extractEmailDomain } from "@/lib/email/extract-email-domain"; import { withReferralsEmbedToken } from "@/lib/embed/referrals/auth"; import { prisma } from "@/lib/prisma"; -import { - TREMENDOUS_ENABLED_PROGRAM_IDS, - TREMENDOUS_PROHIBITED_TOP_LEVEL_DOMAINS, -} from "@/lib/tremendous/constants"; +import { TREMENDOUS_PROHIBITED_TOP_LEVEL_DOMAINS } from "@/lib/tremendous/constants"; import { ratelimit, redis } from "@/lib/upstash"; import { emailSchema } from "@/lib/zod/schemas/auth"; import { ACTIVE_ENROLLMENT_STATUSES } from "@/lib/zod/schemas/partners"; @@ -31,13 +28,6 @@ export const POST = withReferralsEmbedToken( }); } - if (!TREMENDOUS_ENABLED_PROGRAM_IDS.includes(programEnrollment.programId)) { - throw new DubApiError({ - code: "forbidden", - message: "Gift card payouts are not available for this program.", - }); - } - const { email, code } = verifyOtpSchema.parse(await parseRequestBody(req)); const { partnerId } = programEnrollment; diff --git a/apps/web/app/(ee)/app.dub.co/embed/referrals/page-client.tsx b/apps/web/app/(ee)/app.dub.co/embed/referrals/page-client.tsx index adbc82f108b..a9707b0ac81 100644 --- a/apps/web/app/(ee)/app.dub.co/embed/referrals/page-client.tsx +++ b/apps/web/app/(ee)/app.dub.co/embed/referrals/page-client.tsx @@ -3,7 +3,6 @@ import { constructPartnerReferralLink } from "@/lib/partner-referrals/utils"; import { constructPartnerLink } from "@/lib/partners/construct-partner-link"; import { QueryLinkStructureHelpText } from "@/lib/partners/query-link-structure-help-text"; -import { TREMENDOUS_ENABLED_PROGRAM_IDS } from "@/lib/tremendous/constants"; import { DiscountProps, PartnerBountyProps, @@ -206,12 +205,11 @@ export function ReferralsEmbedPageClient({ TREMENDOUS_SUPPORTED_COUNTRIES.includes(partner.country), ); - // Show Tremendous payout settings if the partner already uses Tremendous, + // Show Tremendous payout settings if the partner already uses Tremendous for payouts, // or hasn't selected a payout method yet and is eligible based on country. const showSettingsTab = - TREMENDOUS_ENABLED_PROGRAM_IDS.includes(program.id) && - (partner.defaultPayoutMethod === "tremendous" || - (!partner.defaultPayoutMethod && isTremendousCountrySupported)); + partner.defaultPayoutMethod === "tremendous" || + (!partner.defaultPayoutMethod && isTremendousCountrySupported); const customerRewards = useMemo( () => rewards.filter((reward) => reward.event !== "referral"), diff --git a/apps/web/app/(ee)/app.dub.co/embed/referrals/quickstart.tsx b/apps/web/app/(ee)/app.dub.co/embed/referrals/quickstart.tsx index e3841abe92b..230e4bb1b20 100644 --- a/apps/web/app/(ee)/app.dub.co/embed/referrals/quickstart.tsx +++ b/apps/web/app/(ee)/app.dub.co/embed/referrals/quickstart.tsx @@ -1,5 +1,4 @@ import { constructPartnerLink } from "@/lib/partners/construct-partner-link"; -import { TREMENDOUS_ENABLED_PROGRAM_IDS } from "@/lib/tremendous/constants"; import { programEmbedSchema } from "@/lib/zod/schemas/program-embed"; import { Button, @@ -140,17 +139,12 @@ export function ReferralsEmbedQuickstart({ !partner.country || TREMENDOUS_SUPPORTED_COUNTRIES.includes(partner.country), ); - - const usesTremendous = - partner.defaultPayoutMethod === "tremendous"; - - // Show Tremendous payout settings if the partner already uses Tremendous, + // Show Tremendous payout settings if the partner already uses Tremendous for payouts, // or hasn't selected a payout method yet and is eligible based on country. const showTremendousSettings = - TREMENDOUS_ENABLED_PROGRAM_IDS.includes(program.id) && - (usesTremendous || - (!partner.defaultPayoutMethod && - isTremendousCountrySupported)); + partner.defaultPayoutMethod === "tremendous" || + (!partner.defaultPayoutMethod && + isTremendousCountrySupported); if (showTremendousSettings) { setSelectedTab("Settings"); diff --git a/apps/web/app/(ee)/partners.dub.co/(dashboard)/profile/about-you-form.tsx b/apps/web/app/(ee)/partners.dub.co/(dashboard)/profile/about-you-form.tsx index 8111a245215..5b58dff685e 100644 --- a/apps/web/app/(ee)/partners.dub.co/(dashboard)/profile/about-you-form.tsx +++ b/apps/web/app/(ee)/partners.dub.co/(dashboard)/profile/about-you-form.tsx @@ -9,6 +9,7 @@ import { PartnerProps } from "@/lib/types"; import { MAX_PARTNER_DESCRIPTION_LENGTH } from "@/lib/zod/schemas/partners"; import { MaxCharactersCounter } from "@/ui/shared/max-characters-counter"; import { Button, RadioGroup, RadioGroupItem, useEnterSubmit } from "@dub/ui"; +import { Plus } from "@dub/ui/icons"; import { cn } from "@dub/utils"; import { IndustryInterest, MonthlyTraffic } from "@prisma/client"; import { useAction } from "next-safe-action/hooks"; @@ -151,12 +152,40 @@ export function AboutYouForm({ partner }: { partner?: PartnerProps }) { )) : [...Array(3)].map((_, idx) => ( -
setShowIndustryInterestsModal(true)} className={cn( - "border-border-subtle h-11 w-32 rounded-full border border-dashed bg-white", + "relative flex h-11 w-32 items-center justify-center rounded-full bg-white", + !disabled && + "transition-colors hover:bg-neutral-50/60", + disabled && "cursor-not-allowed", )} - /> + > + + + ))}