import { randomUUID } from "node:crypto"; import { env } from "../config.js"; import { htmlToTextBody, sanitizeEmailHtml } from "../lib/html.js"; import { AppError } from "../lib/errors.js"; import { isoNow } from "../lib/utils.js"; import { supabaseAdmin } from "../lib/supabase.js"; import { logCampaignActivity } from "./activity-log-service.js"; import { incrementEmailUsage, enforceEmailQuota } from "./quota-service.js"; import { appendUnsubscribeFooter, buildListUnsubscribeHeaders, createContactUnsubscribeToken } from "./unsubscribe-service.js"; import type { EmailProvider } from "./email/provider.js"; import { ResendProvider } from "./email/resend-provider.js"; export type CampaignBlock = { id: string; type: "hero" | "text" | "image" | "button" | "divider" | "spacer" | "social" | "footer" | "header_logo" | "columns" | "quote" | "video" | "countdown" | "list"; content: Record; }; const uuidPattern = /^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i; const ensureUuid = (value?: string) => (value && uuidPattern.test(value) ? value : randomUUID()); const scheduledCampaignProvider = new ResendProvider(); const schedulerPollIntervalMs = 30_000; let schedulerStarted = false; const publicApiBaseUrl = (env.API_PUBLIC_BASE_URL || env.APP_BASE_URL).replace(/\/$/, ""); type SchedulerLogger = { info?: (...args: unknown[]) => void; warn?: (...args: unknown[]) => void; error?: (...args: unknown[]) => void; }; type CampaignSendSummary = { sentCount: number; failedCount: number; latestSentAt: string | null; }; function buildDeliveryEnvelope(campaign: Record) { const platformFrom = `${campaign.from_name || env.PLATFORM_FROM_NAME} <${env.PLATFORM_FROM_EMAIL}>`; const normalizedReplyTo = String(campaign.from_email ?? "").trim().toLowerCase(); const normalizedPlatformFrom = env.PLATFORM_FROM_EMAIL.trim().toLowerCase(); return { from: platformFrom, replyTo: normalizedReplyTo && normalizedReplyTo !== normalizedPlatformFrom ? normalizedReplyTo : undefined, }; } async function loadBrandKitForOrganization(organizationId: string) { const { data, error } = await supabaseAdmin .from("brand_kits") .select("logo_path, primary_color, secondary_color, accent_color") .eq("organization_id", organizationId) .maybeSingle(); if (error) { throw new AppError(400, "BRAND_KIT_LOOKUP_FAILED", error.message); } return { ...data, logo_url: data?.logo_path ? `${publicApiBaseUrl}/api/public/brand-kit/logo/${organizationId}` : null, }; } async function loadCampaignBlocks(campaignId: string) { const { data, error } = await supabaseAdmin .from("campaign_blocks") .select("id, block_type, content, position") .eq("campaign_id", campaignId) .order("position", { ascending: true }); if (error) { throw new AppError(400, "CAMPAIGN_BLOCKS_LOOKUP_FAILED", error.message); } return (data ?? []).map((row) => ({ id: String(row.id ?? randomUUID()), type: row.block_type as CampaignBlock["type"], content: (row.content as Record) ?? {}, })); } function isSuccessfulSendStatus(status: string) { const normalized = status.trim().toLowerCase(); return normalized === "sent" || normalized === "opened" || normalized === "clicked" || normalized === "complained"; } function deriveCompletedCampaignStatus(sentCount: number, failedCount: number) { if (sentCount > 0 && failedCount > 0) return "partial"; if (sentCount > 0) return "sent"; return "failed"; } function deriveCancelledCampaignStatus(summary: CampaignSendSummary) { return summary.sentCount > 0 ? "partial" : "cancelled"; } async function summarizeCampaignSends(campaignId: string): Promise { const { data, error } = await supabaseAdmin .from("campaign_sends") .select("status, sent_at") .eq("campaign_id", campaignId); if (error) { throw new AppError(400, "CAMPAIGN_SEND_SUMMARY_FAILED", error.message); } let sentCount = 0; let failedCount = 0; let latestSentAt: string | null = null; for (const row of data ?? []) { const status = String(row.status ?? ""); const sentAt = row.sent_at ? String(row.sent_at) : null; if (isSuccessfulSendStatus(status) || sentAt) { sentCount += 1; if (sentAt && (!latestSentAt || sentAt > latestSentAt)) { latestSentAt = sentAt; } continue; } if (status.trim().toLowerCase() === "failed") { failedCount += 1; } } return { sentCount, failedCount, latestSentAt, }; } async function prepareCampaignForDelivery(campaignId: string) { const campaign = await loadCampaign(campaignId); if (campaign.editor_mode !== "builder") { return campaign; } const [brandKit, blocks] = await Promise.all([ loadBrandKitForOrganization(String(campaign.organization_id)), loadCampaignBlocks(campaignId), ]); const html = renderBlocksToHtml(blocks, brandKit); const textBody = htmlToTextBody(html); const { error } = await supabaseAdmin .from("campaigns") .update({ html_body: html, text_body: textBody, last_activity_at: isoNow(), updated_at: isoNow(), }) .eq("id", campaignId); if (error) { throw new AppError(400, "CAMPAIGN_UPDATE_FAILED", error.message); } return { ...campaign, html_body: html, text_body: textBody, }; } async function persistSentSnapshot(campaign: Record) { const snapshotAt = isoNow(); const { error } = await supabaseAdmin .from("campaigns") .update({ sent_html_body: campaign.html_body ?? null, sent_text_body: campaign.text_body ?? null, sent_subject: campaign.subject ?? null, sent_from_name: campaign.from_name ?? null, sent_from_email: campaign.from_email ?? null, sent_editor_mode: campaign.editor_mode ?? null, sent_snapshot_at: snapshotAt, last_activity_at: snapshotAt, updated_at: snapshotAt, }) .eq("id", campaign.id); if (error) { throw new AppError(400, "CAMPAIGN_SNAPSHOT_FAILED", error.message); } return snapshotAt; } export function renderBlocksToHtml(blocks: CampaignBlock[], brandKit?: Record | null) { const accent = typeof brandKit?.primary_color === "string" ? brandKit.primary_color : "#0f766e"; const logoUrl = typeof brandKit?.logo_url === "string" ? brandKit.logo_url : ""; const rendered = blocks .map((block) => { switch (block.type) { case "hero": return `
${logoUrl ? `Logo` : ""}

${String(block.content.heading ?? "Your next campaign")}

${String(block.content.body ?? "")}

`; case "text": return `

${String(block.content.heading ?? "")}

${String(block.content.body ?? "")}

`; case "image": return `
${String(block.content.alt ??
`; case "button": return `
${String(block.content.label ?? "Learn more")}
`; case "divider": return `
`; case "spacer": return `
`; case "social": return `
${String(block.content.text ?? "Follow us on social media.")}
`; case "footer": return `
${String(block.content.body ?? "")}
`; case "header_logo": { const alignment = block.content.alignment === "left" ? "left" : "center"; const logoWidth = Number(block.content.logoWidth ?? 160); const explicitLogoSrc = String(block.content.logoSrc ?? "").trim(); const logoSrc = explicitLogoSrc || logoUrl || ""; const logoAlt = String(block.content.logoAlt ?? "Logo"); return logoSrc ? `
${logoAlt}
` : `
${logoAlt}
`; } case "columns": { const cols = Array.isArray(block.content.columns) ? block.content.columns as Array> : []; const colWidth = cols.length === 3 ? "33.33%" : "50%"; const tds = cols.map((col) => `
${String(col.iconEmoji ?? "")}
${String(col.heading ?? "")}
${String(col.body ?? "")}
` ).join(""); return `
${tds}
`; } case "quote": return `

"${String(block.content.quote ?? "")}"

${String(block.content.author ?? "")}${block.content.role ? `, ${String(block.content.role)}` : ""}
`; case "video": { const thumbSrc = String(block.content.thumbnailSrc ?? ""); const videoUrl = String(block.content.videoUrl ?? "#"); const alt = String(block.content.alt ?? "Video"); const caption = String(block.content.caption ?? ""); return `
${alt}${caption ? `

${caption}

` : ""}
`; } case "countdown": { const endDate = new Date(String(block.content.endDate ?? "")); const now = new Date(); const diff = Math.max(0, endDate.getTime() - now.getTime()); if (diff <= 0) { return `
${String(block.content.expiredText ?? "This offer has expired.")}
`; } const days = Math.floor(diff / 86400000); const hours = Math.floor((diff % 86400000) / 3600000); const minutes = Math.floor((diff % 3600000) / 60000); return `
${String(block.content.label ?? "")}
${days}d ${hours}h ${minutes}m
`; } case "list": { const items = Array.isArray(block.content.items) ? block.content.items as string[] : []; const style = String(block.content.style ?? "bullet"); const heading = String(block.content.heading ?? ""); const listRows = items.map((item, i) => { const marker = style === "checkmark" ? "✓" : style === "numbered" ? `${i + 1}.` : "•"; return `${marker}${String(item)}`; }).join(""); return `
${heading ? `

${heading}

` : ""}${listRows}
`; } default: return ""; } }) .join(""); return `
${rendered}
`; } export async function persistCampaignContent(input: { campaignId: string; editorMode: "builder" | "html"; htmlBody?: string | null; textBody?: string | null; blocks?: CampaignBlock[]; actorAuthUserId?: string | null; }) { if (input.editorMode === "html") { const sanitizedHtml = sanitizeEmailHtml(input.htmlBody ?? ""); const textBody = input.textBody?.trim() || htmlToTextBody(sanitizedHtml); const { error } = await supabaseAdmin .from("campaigns") .update({ editor_mode: "html", html_body: sanitizedHtml, text_body: textBody, updated_by_auth_user_id: input.actorAuthUserId ?? null, last_composed_at: isoNow(), last_activity_at: isoNow(), updated_at: isoNow() }) .eq("id", input.campaignId); if (error) { throw new AppError(400, "CAMPAIGN_UPDATE_FAILED", error.message); } await supabaseAdmin.from("campaign_blocks").delete().eq("campaign_id", input.campaignId); return; } const blocks = input.blocks ?? []; const campaign = await loadCampaign(input.campaignId); const brandKit = await loadBrandKitForOrganization(String(campaign.organization_id)); const html = renderBlocksToHtml(blocks, brandKit); const textBody = input.textBody?.trim() || htmlToTextBody(html); const { error: campaignError } = await supabaseAdmin .from("campaigns") .update({ editor_mode: "builder", html_body: html, text_body: textBody, updated_by_auth_user_id: input.actorAuthUserId ?? null, last_composed_at: isoNow(), last_activity_at: isoNow(), updated_at: isoNow() }) .eq("id", input.campaignId); if (campaignError) { throw new AppError(400, "CAMPAIGN_UPDATE_FAILED", campaignError.message); } await supabaseAdmin.from("campaign_blocks").delete().eq("campaign_id", input.campaignId); if (blocks.length) { const { error: blockError } = await supabaseAdmin.from("campaign_blocks").insert( blocks.map((block, index) => ({ id: ensureUuid(block.id), campaign_id: input.campaignId, position: index, block_type: block.type, content: block.content })) ); if (blockError) { throw new AppError(400, "CAMPAIGN_BLOCKS_FAILED", blockError.message); } } } export async function sendCampaignTest(campaignId: string, toEmails: string[], provider: EmailProvider) { const campaign = await prepareCampaignForDelivery(campaignId); const envelope = buildDeliveryEnvelope(campaign); const result = await provider.sendTestEmail({ to: toEmails, from: envelope.from, replyTo: envelope.replyTo, subject: campaign.subject, html: campaign.html_body, text: campaign.text_body, tags: [{ name: "campaign_id", value: campaign.id }] }); return result; } function normalizeRecipientIds(contactIds?: string[]) { const normalized = Array.from( new Set( (contactIds ?? []) .map((contactId) => String(contactId ?? "").trim()) .filter(Boolean) ) ); return normalized.length ? normalized : null; } function assertCampaignReadyForDelivery(campaign: Record) { const publishedAt = campaign.last_published_at ? new Date(String(campaign.last_published_at)) : null; const composedAt = campaign.last_composed_at ? new Date(String(campaign.last_composed_at)) : campaign.updated_at ? new Date(String(campaign.updated_at)) : null; if (!publishedAt || Number.isNaN(publishedAt.getTime())) { throw new AppError(400, "CAMPAIGN_PUBLISH_REQUIRED", "Publish the campaign before sending or scheduling it."); } if (composedAt && !Number.isNaN(composedAt.getTime()) && composedAt.getTime() - publishedAt.getTime() > 1000) { throw new AppError(400, "CAMPAIGN_REPUBLISH_REQUIRED", "Publish the latest changes before sending or scheduling this campaign."); } } function readRecipientIds(value: unknown) { if (!Array.isArray(value)) return null; const normalized = value.map((entry) => String(entry ?? "").trim()).filter(Boolean); return normalized.length ? normalized : null; } async function loadRecipientsForCampaign(campaign: Record, contactIds?: string[] | null) { let contactsQuery = supabaseAdmin .from("contacts") .select("id, email, full_name, tags") .eq("organization_id", campaign.organization_id) .eq("status", "subscribed"); if (contactIds?.length) { contactsQuery = contactsQuery.in("id", contactIds); } const { data: contacts, error } = await contactsQuery; if (error) { throw new AppError(400, "CAMPAIGN_RECIPIENT_LOOKUP_FAILED", error.message); } return contacts ?? []; } function triggerCampaignDelivery(campaignId: string, provider: EmailProvider, logger?: SchedulerLogger) { setImmediate(() => { void processQueuedCampaign(campaignId, provider, logger); }); } async function processQueuedCampaign(campaignId: string, provider: EmailProvider, logger?: SchedulerLogger) { const campaign = await loadCampaign(campaignId); if (campaign.is_reusable_template) { throw new AppError(400, "CAMPAIGN_TEMPLATE_SEND_BLOCKED", "Reusable campaigns cannot be sent directly."); } if (campaign.status !== "queued") { return; } const recipientIds = readRecipientIds(campaign.recipient_contact_ids); const recipients = await loadRecipientsForCampaign(campaign, recipientIds); if (!recipients.length) { await supabaseAdmin .from("campaigns") .update({ status: "failed", last_activity_at: isoNow(), updated_at: isoNow() }) .eq("id", campaignId); await logCampaignActivity({ organizationId: String(campaign.organization_id), campaignId, actorAuthUserId: campaign.last_delivery_requested_by_auth_user_id ?? null, action: "campaign.delivery_failed", metadata: { reason: "no_recipients", } }); throw new AppError(400, "CAMPAIGN_RECIPIENTS_REQUIRED", "Select at least one subscribed contact before sending."); } await enforceEmailQuota(campaign.organization_id, recipients.length); const { data: claimedCampaign, error: claimError } = await supabaseAdmin .from("campaigns") .update({ status: "sending", last_activity_at: isoNow(), updated_at: isoNow() }) .eq("id", campaignId) .eq("status", "queued") .select("*") .maybeSingle(); if (claimError) { throw new AppError(400, "CAMPAIGN_SEND_START_FAILED", claimError.message); } if (!claimedCampaign) { return; } try { await logCampaignActivity({ organizationId: String(claimedCampaign.organization_id), campaignId, actorAuthUserId: claimedCampaign.last_delivery_requested_by_auth_user_id ?? null, action: "campaign.delivery_started", metadata: { recipientCount: recipients.length, deliveryAction: claimedCampaign.last_delivery_action ?? "send", } }); const preparedCampaign = await prepareCampaignForDelivery(campaignId); const snapshotAt = await persistSentSnapshot(preparedCampaign); const envelope = buildDeliveryEnvelope(preparedCampaign); let sentCount = 0; let failedCount = 0; for (const contact of recipients) { const liveCampaign = await loadCampaign(campaignId); if (liveCampaign.status !== "queued" && liveCampaign.status !== "sending") { return; } let sendRowId: string | null = null; try { const { data: sendRow, error: sendRowError } = await supabaseAdmin .from("campaign_sends") .insert({ organization_id: preparedCampaign.organization_id, campaign_id: preparedCampaign.id, contact_id: contact.id, initiated_by_auth_user_id: preparedCampaign.last_delivery_requested_by_auth_user_id ?? null, contact_email: contact.email, contact_full_name: contact.full_name ?? null, contact_tags: Array.isArray(contact.tags) ? contact.tags : [], status: "queued", provider: env.EMAIL_PROVIDER, created_at: isoNow(), updated_at: isoNow(), }) .select("id") .single(); if (sendRowError || !sendRow) { throw new AppError(400, "CAMPAIGN_SEND_ROW_CREATE_FAILED", sendRowError?.message ?? "Unable to create campaign send row."); } sendRowId = String(sendRow.id); const { unsubscribeUrl } = await createContactUnsubscribeToken({ organizationId: String(preparedCampaign.organization_id), contactId: String(contact.id), campaignId: String(preparedCampaign.id), campaignSendId: sendRowId, recipientEmail: contact.email ? String(contact.email) : null, }); const renderedContent = appendUnsubscribeFooter( String(preparedCampaign.html_body ?? ""), String(preparedCampaign.text_body ?? ""), unsubscribeUrl ); const sendResult = await provider.sendCampaignBatch({ to: [contact.email as string], from: envelope.from, replyTo: envelope.replyTo, subject: preparedCampaign.subject, html: renderedContent.html, text: renderedContent.text, tags: [{ name: "campaign_id", value: preparedCampaign.id }], headers: buildListUnsubscribeHeaders(unsubscribeUrl), }); const externalId = sendResult.messageIds[0] ?? null; const { error: sendUpdateError } = await supabaseAdmin .from("campaign_sends") .update({ status: "sent", provider: sendResult.provider, external_message_id: externalId, sent_at: isoNow(), updated_at: isoNow(), }) .eq("id", sendRowId); if (sendUpdateError) { throw new AppError(400, "CAMPAIGN_SEND_ROW_UPDATE_FAILED", sendUpdateError.message); } await supabaseAdmin.from("email_provider_messages").insert({ organization_id: preparedCampaign.organization_id, campaign_id: preparedCampaign.id, campaign_send_id: sendRowId, provider: sendResult.provider, external_message_id: externalId, metadata: sendResult.raw }); sentCount += 1; } catch (error) { failedCount += 1; const sendErrorMessage = error instanceof Error ? error.message : "Email send failed."; if (sendRowId) { await supabaseAdmin .from("campaign_sends") .update({ status: "failed", provider: env.EMAIL_PROVIDER, error_message: sendErrorMessage, updated_at: isoNow(), }) .eq("id", sendRowId); } logger?.error?.(error); } } if (sentCount > 0) { await incrementEmailUsage(preparedCampaign.organization_id, sentCount); } const finalCampaign = await loadCampaign(campaignId); if (finalCampaign.status !== "queued" && finalCampaign.status !== "sending") { return; } await supabaseAdmin .from("campaigns") .update({ status: deriveCompletedCampaignStatus(sentCount, failedCount), sent_at: sentCount > 0 ? snapshotAt : null, scheduled_for: null, last_activity_at: isoNow(), updated_at: isoNow() }) .eq("id", campaignId); const completionStatus = deriveCompletedCampaignStatus(sentCount, failedCount); await logCampaignActivity({ organizationId: String(preparedCampaign.organization_id), campaignId, actorAuthUserId: preparedCampaign.last_delivery_requested_by_auth_user_id ?? null, action: completionStatus === "partial" ? "campaign.delivery_partial" : completionStatus === "sent" ? "campaign.delivery_completed" : "campaign.delivery_failed", metadata: { status: completionStatus, sentCount, failedCount, recipientCount: recipients.length, } }); if (failedCount > 0 && sentCount > 0) { logger?.warn?.(`Campaign ${campaignId} completed with partial failures: ${failedCount} failed, ${sentCount} sent.`); } } catch (error) { const sendSummary = await summarizeCampaignSends(campaignId); const failureStatus = deriveCompletedCampaignStatus(sendSummary.sentCount, sendSummary.failedCount); await supabaseAdmin .from("campaigns") .update({ status: failureStatus, sent_at: sendSummary.sentCount > 0 ? sendSummary.latestSentAt ?? isoNow() : null, scheduled_for: null, last_activity_at: isoNow(), updated_at: isoNow() }) .eq("id", campaignId); await logCampaignActivity({ organizationId: String(campaign.organization_id), campaignId, actorAuthUserId: campaign.last_delivery_requested_by_auth_user_id ?? null, action: failureStatus === "partial" ? "campaign.delivery_partial" : "campaign.delivery_failed", metadata: { status: failureStatus, sentCount: sendSummary.sentCount, failedCount: sendSummary.failedCount, reason: error instanceof Error ? error.message : "unknown_error", } }); logger?.error?.(error); } } export async function queueCampaignSend(campaignId: string, provider: EmailProvider, contactIds?: string[], actorAuthUserId?: string | null) { const campaign = await loadCampaign(campaignId); if (campaign.is_reusable_template) { throw new AppError(400, "CAMPAIGN_TEMPLATE_SEND_BLOCKED", "Reusable campaigns cannot be sent directly."); } assertCampaignReadyForDelivery(campaign); const { error } = await supabaseAdmin .from("campaigns") .update({ status: "queued", sent_at: null, scheduled_for: null, recipient_contact_ids: normalizeRecipientIds(contactIds), updated_by_auth_user_id: actorAuthUserId ?? null, last_delivery_requested_at: isoNow(), last_delivery_requested_by_auth_user_id: actorAuthUserId ?? null, last_delivery_action: "send", last_activity_at: isoNow(), updated_at: isoNow() }) .eq("id", campaignId); if (error) { throw new AppError(400, "CAMPAIGN_QUEUE_FAILED", error.message); } await logCampaignActivity({ organizationId: String(campaign.organization_id), campaignId, actorAuthUserId, action: "campaign.delivery_queued", metadata: { recipientSelectionCount: contactIds?.length ?? 0, } }); triggerCampaignDelivery(campaignId, provider); } export async function scheduleCampaignSend(campaignId: string, scheduledFor: string, contactIds?: string[], actorAuthUserId?: string | null) { const campaign = await loadCampaign(campaignId); if (campaign.is_reusable_template) { throw new AppError(400, "CAMPAIGN_TEMPLATE_SCHEDULE_BLOCKED", "Reusable campaigns cannot be scheduled directly."); } assertCampaignReadyForDelivery(campaign); const scheduleAt = new Date(scheduledFor); if (Number.isNaN(scheduleAt.getTime())) { throw new AppError(400, "CAMPAIGN_SCHEDULE_INVALID", "Provide a valid schedule time."); } if (scheduleAt.getTime() < Date.now() + 60_000) { throw new AppError(400, "CAMPAIGN_SCHEDULE_TOO_SOON", "Schedule the campaign at least one minute in the future."); } const { error } = await supabaseAdmin .from("campaigns") .update({ status: "scheduled", scheduled_for: scheduleAt.toISOString(), recipient_contact_ids: normalizeRecipientIds(contactIds), sent_at: null, updated_by_auth_user_id: actorAuthUserId ?? null, last_delivery_requested_at: isoNow(), last_delivery_requested_by_auth_user_id: actorAuthUserId ?? null, last_delivery_action: "schedule", last_activity_at: isoNow(), updated_at: isoNow() }) .eq("id", campaignId); if (error) { throw new AppError(400, "CAMPAIGN_SCHEDULE_FAILED", error.message); } await logCampaignActivity({ organizationId: String(campaign.organization_id), campaignId, actorAuthUserId, action: "campaign.delivery_scheduled", metadata: { scheduledFor: scheduleAt.toISOString(), recipientSelectionCount: contactIds?.length ?? 0, } }); } export async function cancelCampaignSend(campaignId: string, actorAuthUserId?: string | null) { const campaign = await loadCampaign(campaignId); if (campaign.status !== "scheduled" && campaign.status !== "queued" && campaign.status !== "sending") { throw new AppError(400, "CAMPAIGN_CANCEL_BLOCKED", "Only scheduled, queued, or in-progress campaigns can be cancelled."); } const sendSummary = await summarizeCampaignSends(campaignId); const nextStatus = deriveCancelledCampaignStatus(sendSummary); const { error } = await supabaseAdmin .from("campaigns") .update({ status: nextStatus, sent_at: sendSummary.sentCount > 0 ? sendSummary.latestSentAt ?? campaign.sent_at ?? isoNow() : null, scheduled_for: null, updated_by_auth_user_id: actorAuthUserId ?? null, last_activity_at: isoNow(), updated_at: isoNow() }) .eq("id", campaignId); if (error) { throw new AppError(400, "CAMPAIGN_CANCEL_FAILED", error.message); } await logCampaignActivity({ organizationId: String(campaign.organization_id), campaignId, actorAuthUserId, action: "campaign.delivery_cancelled", metadata: { previousStatus: campaign.status, status: nextStatus, sentCount: sendSummary.sentCount, failedCount: sendSummary.failedCount, } }); return nextStatus; } export async function loadCampaign(campaignId: string) { const { data, error } = await supabaseAdmin .from("campaigns") .select("*") .eq("id", campaignId) .single(); if (error || !data) { throw new AppError(404, "CAMPAIGN_NOT_FOUND", error?.message ?? "Campaign not found."); } return data as Record; } export async function cloneCampaign(campaignId: string, options?: { asReusableTemplate?: boolean; name?: string | null; actorAuthUserId?: string | null }) { const sourceCampaign = await loadCampaign(campaignId); const { data: sourceBlocks, error: blockError } = await supabaseAdmin .from("campaign_blocks") .select("position, block_type, content") .eq("campaign_id", campaignId) .order("position", { ascending: true }); if (blockError) { throw new AppError(400, "CAMPAIGN_CLONE_FAILED", blockError.message); } const nextName = options?.name?.trim() || (options?.asReusableTemplate ? `${sourceCampaign.name} reusable` : `${sourceCampaign.name} copy`); const { data: clonedCampaign, error } = await supabaseAdmin .from("campaigns") .insert({ organization_id: sourceCampaign.organization_id, name: nextName, subject: sourceCampaign.subject, from_name: sourceCampaign.from_name, from_email: sourceCampaign.from_email, editor_mode: sourceCampaign.editor_mode, html_body: sourceCampaign.html_body, text_body: sourceCampaign.text_body, status: "draft", scheduled_for: null, recipient_contact_ids: null, is_reusable_template: Boolean(options?.asReusableTemplate), source_campaign_id: sourceCampaign.id, created_by_auth_user_id: options?.actorAuthUserId ?? null, updated_by_auth_user_id: options?.actorAuthUserId ?? null, last_composed_at: isoNow(), last_activity_at: isoNow(), created_at: isoNow(), updated_at: isoNow() }) .select("*") .single(); if (error || !clonedCampaign) { throw new AppError(400, "CAMPAIGN_CLONE_FAILED", error?.message ?? "Unable to clone campaign."); } if ((sourceBlocks ?? []).length) { const { error: insertBlocksError } = await supabaseAdmin.from("campaign_blocks").insert( (sourceBlocks ?? []).map((block: any) => ({ campaign_id: clonedCampaign.id, position: block.position, block_type: block.block_type, content: block.content })) ); if (insertBlocksError) { throw new AppError(400, "CAMPAIGN_CLONE_FAILED", insertBlocksError.message); } } await logCampaignActivity({ organizationId: String(sourceCampaign.organization_id), campaignId: String(clonedCampaign.id), actorAuthUserId: options?.actorAuthUserId ?? null, action: options?.asReusableTemplate ? "campaign.reusable_created" : "campaign.cloned", metadata: { sourceCampaignId: sourceCampaign.id, } }); return clonedCampaign; } export function startCampaignScheduler(logger?: SchedulerLogger) { if (schedulerStarted) return; schedulerStarted = true; const run = async () => { try { const now = isoNow(); const { data: dueCampaigns, error } = await supabaseAdmin .from("campaigns") .select("id") .eq("status", "scheduled") .eq("is_reusable_template", false) .lte("scheduled_for", now) .order("scheduled_for", { ascending: true }) .limit(20); if (error) { logger?.error?.(error); return; } for (const campaign of dueCampaigns ?? []) { const { data: claimedCampaign, error: claimError } = await supabaseAdmin .from("campaigns") .update({ status: "queued", scheduled_for: null, updated_at: isoNow() }) .eq("id", campaign.id) .eq("status", "scheduled") .select("id") .maybeSingle(); if (claimError) { logger?.error?.(claimError); continue; } if (!claimedCampaign) { continue; } triggerCampaignDelivery(String(campaign.id), scheduledCampaignProvider, logger); } } catch (error) { logger?.error?.(error); } }; void run(); setInterval(() => { void run(); }, schedulerPollIntervalMs); }