campaign-fanout.queue.ts

campaign-fanout.queue.ts

  import { env } from '@/config/env';
  import {
    PLUTO_REJECT_REASON,
    PLUTO_REJECT_STATUSES,
    getPlutoClient,
  } from '@/services/pluto/pluto.client';
  import { addToSuppressionList } from '@/services/suppression.service';
  import { prisma } from '@/services/util/prisma.service';
  import {
    CampaignContactStatus,
    CampaignEmailType,
    CampaignStatus,
    ContactStatus,
  } from '@/types/marketing.types';
  import { LogType, log } from '@/utils/logger.util';
  import axios from 'axios';
  import { Queue, Worker } from 'bullmq';
  import type { Job } from 'bullmq';
  import { uuidv7 } from 'uuidv7';
  import { campaignBatchQueue } from './campaign-batch.queue';

  const DEBOUNCE_BLOCK_STATUSES = new Set(['invalid', 'spamtrap', 'abuse', 'do_not_mail']);

  const redisConnection = {
    host: env.REDIS_HOST,
    port: env.REDIS_PORT,
    password: env.REDIS_PASSWORD || undefined,
  };

  export const campaignFanoutQueue = new Queue('campaign-fanout', { connection: redisConnection });

  export async function runFanout(job: Job) {
    const {
      campaignId,
      workspaceId,
      emailType,
      provider,
      fromEmail,
      fromName,
      subject,
      templateId,
      variables,
    } = job.data;

    // Distributed lock — prevents duplicate fanout if job is retried or picked up by a second worker
    const lockResult = await prisma.$executeRaw`
      UPDATE campaigns
      SET dispatch_started_at = NOW(), status = 'SENDING', updated_at = NOW()
      WHERE id = ${campaignId}::uuid AND dispatch_started_at IS NULL
    `;
    if (lockResult === 0) return;

    const contactRows: { id: string; email: string }[] = await prisma.$queryRaw`
      SELECT DISTINCT c.id, c.email
      FROM contacts c
      JOIN audience_contacts ac ON ac.contact_id = c.id
      JOIN campaign_audiences ca ON ca.audience_id = ac.audience_id
      WHERE ca.campaign_id = ${campaignId}::uuid
        AND c.status = ${ContactStatus.SUBSCRIBED}
        AND ac.status = 'SUBSCRIBED'
        AND c.workspace_id = ${workspaceId}::uuid
      ORDER BY c.id ASC
    `;

    if (contactRows.length === 0) {
      await prisma.campaign.update({
        where: { id: campaignId },
        data: {
          status: CampaignStatus.COMPLETED,
          sentAt: new Date(),
          totalContacts: 0,
          updatedAt: new Date(),
        },
      });
      return;
    }

    // Check suppression list before any verification passes
    const emails = contactRows.map((c) => c.email.toLowerCase());
    const suppressedRows: { email_address: string }[] = await prisma.$queryRaw`
      SELECT email_address
      FROM email_suppression_list
      WHERE email_address = ANY(${emails}::text[])
        AND is_active = true
    `;
    const suppressedSet = new Set(suppressedRows.map((r) => r.email_address));

    const suppressedContacts = contactRows.filter((c) => suppressedSet.has(c.email.toLowerCase()));
    let sendableContacts = contactRows.filter((c) => !suppressedSet.has(c.email.toLowerCase()));

    log(
      `Campaign ${campaignId} suppression: ${contactRows.length} total, ${suppressedContacts.length} suppressed,
  ${sendableContacts.length} valid`,
      LogType.INFO,
    );

    type PreFilterFailed = {
      id: string;
      campaignId: string;
      contactId: string;
      status: CampaignContactStatus;
      failureReason: string;
      createdAt: Date;
    };

    const now = new Date();
    const plutoFailedRecords: PreFilterFailed[] = [];
    const debounceFailedRecords: PreFilterFailed[] = [];

    // Pluto: internal email pre-validator (disposable, invalid format, no MX, etc.)
    // Runs before DB writes so rejections are never written as PENDING
    const pluto = getPlutoClient();
    if (pluto && sendableContacts.length > 0) {
      try {
        const plutoEmails = sendableContacts.map((c) => c.email);
        log(
          `Campaign ${campaignId} Pluto pre-filter: checking ${plutoEmails.length} emails`,
          LogType.INFO,
        );
        const plutoResults = await pluto.validateBatch(plutoEmails);
        const plutoRejectedEmails = new Set(
          plutoResults
            .filter((r) => PLUTO_REJECT_STATUSES.has(r.status))
            .map((r) => r.email.toLowerCase()),
        );

        if (plutoRejectedEmails.size > 0) {
          const plutoStatusMap = new Map(plutoResults.map((r) => [r.email.toLowerCase(), r.status]));
          for (const c of sendableContacts) {
            if (plutoRejectedEmails.has(c.email.toLowerCase())) {
              plutoFailedRecords.push({
                id: uuidv7(),
                campaignId,
                contactId: c.id,
                status: CampaignContactStatus.FAILED,
                failureReason: `Invalid email: ${PLUTO_REJECT_REASON[plutoStatusMap.get(c.email.toLowerCase()) ?? ''] ?? 'failed
  pre-send validation'}`,
                createdAt: now,
              });
            }
          }
          log(
            `Campaign ${campaignId} Pluto rejected ${plutoFailedRecords.length} contacts`,
            LogType.WARNING,
          );
          sendableContacts = sendableContacts.filter(
            (c) => !plutoRejectedEmails.has(c.email.toLowerCase()),
          );

          // Feed Pluto rejections back into the suppression list — future campaigns skip them at cheapest stage
          const suppressionPayload = [...plutoRejectedEmails].map((email) => ({
            email,
            reason: `Pluto: ${PLUTO_REJECT_REASON[plutoStatusMap.get(email) ?? ''] ?? 'invalid email'}`,
          }));
          addToSuppressionList(suppressionPayload, workspaceId, { fireAndForget: true });
        }
      } catch (err) {
        // Degrade gracefully — Pluto unavailability should not block the campaign
        log(
          `Campaign ${campaignId} Pluto pre-filter failed — proceeding without it: ${err instanceof Error ? err.message :
  String(err)}`,
          LogType.WARNING,
        );
      }
    }

    // Debounce: external email verification API (domain-type campaigns only)
    if (emailType === CampaignEmailType.DOMAIN && sendableContacts.length > 0) {
      try {
        const emailEngineBase = env.EMAIL_ENGINE_URL;
        const emailsToVerify = sendableContacts.map((c) => c.email);
        log(
          `Campaign ${campaignId} Debounce verification: checking ${emailsToVerify.length} emails`,
          LogType.INFO,
        );

        const verifyRes = await axios.post<{
          success: boolean;
          results: Array<{ address: string; status: string; sub_status?: string }>;
        }>(
          `${emailEngineBase}/verify-batch`,
          { emails: emailsToVerify },
          { headers: { 'x-secret': env.SOMETHING_SECRET_ETC }, timeout: 300_000 },
        );

        if (!verifyRes.data.success) {
          // API returned but signalled failure — skip rather than block
          log(
            `Campaign ${campaignId} Debounce verify-batch returned success=false — skipping`,
            LogType.WARNING,
          );
        } else {
          const debounceBlockedEmails = new Set(
            verifyRes.data.results
              .filter((r) => DEBOUNCE_BLOCK_STATUSES.has(r.status))
              .map((r) => r.address.toLowerCase()),
          );

          if (debounceBlockedEmails.size > 0) {
            for (const c of sendableContacts) {
              if (debounceBlockedEmails.has(c.email.toLowerCase())) {
                debounceFailedRecords.push({
                  id: uuidv7(),
                  campaignId,
                  contactId: c.id,
                  status: CampaignContactStatus.FAILED,
                  failureReason: 'Invalid recipient email address',
                  createdAt: now,
                });
              }
            }
            log(
              `Campaign ${campaignId} Debounce blocked ${debounceFailedRecords.length} contacts`,
              LogType.WARNING,
            );
            sendableContacts = sendableContacts.filter(
              (c) => !debounceBlockedEmails.has(c.email.toLowerCase()),
            );
          }
        }
      } catch (err) {
        log(
          `Campaign ${campaignId} Debounce verification failed — proceeding without it: ${err instanceof Error ? err.message :
  String(err)}`,
          LogType.WARNING,
        );
      }
    }

    // All verification done — write every contact's final status in one atomic createMany.
    // This is intentional: writing after all checks means no contact is ever inserted as PENDING
    // and then "upgraded" to FAILED. The unique constraint on (campaignId, contactId) would cause
    // skipDuplicates to silently drop the second write, leaving the wrong status in place.
    const allContactRecords = [
      ...suppressedContacts.map((c) => ({
        id: uuidv7(),
        campaignId,
        contactId: c.id,
        status: CampaignContactStatus.FAILED,
        failureReason: 'Recipient is suppressed',
        createdAt: now,
      })),
      ...plutoFailedRecords,
      ...debounceFailedRecords,
      ...sendableContacts.map((c) => ({
        id: uuidv7(),
        campaignId,
        contactId: c.id,
        status: CampaignContactStatus.PENDING,
        createdAt: now,
      })),
    ];

    await prisma.campaignContact.createMany({ data: allContactRecords, skipDuplicates: true });

    const preFilterFailedCount =
      suppressedContacts.length + plutoFailedRecords.length + debounceFailedRecords.length;

    await prisma.$executeRaw`
      UPDATE campaigns
      SET
        total_contacts = ${contactRows.length},
        failed_count = failed_count + ${preFilterFailedCount},
        updated_at = NOW()
      WHERE id = ${campaignId}::uuid
    `;

    if (sendableContacts.length === 0) {
      await prisma.$executeRaw`
        UPDATE campaigns
        SET status = 'COMPLETED', sent_at = NOW(), updated_at = NOW()
        WHERE id = ${campaignId}::uuid AND status = 'SENDING'
      `;
      log(`Campaign ${campaignId} completed — all contacts filtered before send`, LogType.WARNING);
      return;
    }

    const BATCH_SIZE = 500;
    const batchJobs = [];
    for (let i = 0; i < sendableContacts.length; i += BATCH_SIZE) {
      const chunk = sendableContacts.slice(i, i + BATCH_SIZE);
      batchJobs.push({
        name: 'batch',
        data: {
          campaignId,
          workspaceId,
          subject,
          templateId,
          emailType,
          provider,
          fromEmail,
          fromName,
          variables: variables ?? {},
          minContactId: chunk[0].id,
          maxContactId: chunk[chunk.length - 1].id,
        },
        opts: {
          attempts: 3,
          backoff: { type: 'exponential' as const, delay: 5000 },
          jobId: `${campaignId}-batch-${Math.floor(i / BATCH_SIZE)}`,
          removeOnComplete: true,
          removeOnFail: false,
        },
      });
    }

    await campaignBatchQueue.addBulk(batchJobs);
  }

  export function initializeCampaignFanoutWorker() {
    const worker = new Worker('campaign-fanout', runFanout, {
      connection: redisConnection,
      concurrency: 2,
    });
    worker.on('failed', (job, err) => {
      log(`Fanout job ${job?.id} failed: ${err.message}`, LogType.ERROR);
      if (job && job.attemptsMade >= (job.opts.attempts ?? 1)) {
        const campaignId = job.data.campaignId;
        prisma.campaign
          .update({
            where: { id: campaignId },
            data: { status: CampaignStatus.FAILED, updatedAt: new Date() },
          })
          .catch((e) =>
            log(`Failed to mark campaign ${campaignId} as failed: ${e.message}`, LogType.ERROR),
          );
      }
    });
    return worker;
  }

  export async function shutdownCampaignFanoutQueue() {
    await campaignFanoutQueue.close();
  }

suppression.service.ts

  import { env } from '@/config/env';
  import { httpFetch } from '@/utils/fetch.util';
  import { LogType, log } from '@/utils/logger.util';

  export interface SuppressionEntry {
    email: string;
    reason: string;
  }

  export async function addToSuppressionList(
    entries: SuppressionEntry[],
    workspaceId?: string,
    { fireAndForget = false }: { fireAndForget?: boolean } = {},
  ): Promise<void> {
    if (entries.length === 0) return;
    const emailEngineBase = env.EMAIL_ENGINE_URL;
    const request = httpFetch(`${emailEngineBase}/suppression/add-bulk`, {
      method: 'POST',
      headers: { 'Content-Type': 'application/json', 'x-secret': env.SOMETHING_SECRET_ETC },
      body: JSON.stringify({ emails: entries, workspace_id: workspaceId }),
    }).catch((err: unknown) => {
      log(
        `Suppression sync failed: ${err instanceof Error ? err.message : String(err)}`,
        LogType.WARNING,
      );
    });
    if (!fireAndForget) await request;
  }