warmbox/apps/api/src/jobs/campaign-dispatcher.job.ts

import type { CampaignSatellite, Inbox } from '../generated/prisma/client';
import { CampaignStatus, InboxStatus, WarmupStatus, prisma } from '../lib/prisma';
import { withInboxGuard } from '../lib/inboxGuard';
import { logger } from '../lib/logger';
import { isAllowedWarmupSendPair } from '../lib/mailbox-topology';
import { isWithinSendWindow } from '../lib/send-window';
import * as aiService from '../services/ai.service';
import { syncCampaignDay } from '../services/campaign.service';
import * as emailService from '../services/email.service';
import {
  claimDueSendJob,
  completeSendJob,
  enqueueSendJob,
} from '../services/job-queue.service';

const MIN_JITTER_MS = 45 * 60 * 1000;

type SatelliteWithMailbox = CampaignSatellite & { satelliteMailbox: Inbox };

function pickWeightedSatellite(satellites: SatelliteWithMailbox[]): Inbox {
  const totalWeight = satellites.reduce((sum, sat) => sum + sat.weight, 0);
  let remaining = Math.random() * totalWeight;

  for (const satellite of satellites) {
    remaining -= satellite.weight;
    if (remaining <= 0) {
      return satellite.satelliteMailbox;
    }
  }

  return satellites[satellites.length - 1]!.satelliteMailbox;
}

function hasElapsedMinJitter(lastSendAt: Date | null): boolean {
  if (!lastSendAt) return true;
  return Date.now() - lastSendAt.getTime() >= MIN_JITTER_MS;
}

async function executeCampaignSend(
  campaignId: string,
  jobId: string | null,
): Promise<void> {
  const campaign = await prisma.warmupCampaign.findUnique({
    where: { id: campaignId },
    include: {
      primaryMailbox: true,
      satellites: {
        where: { active: true },
        include: { satelliteMailbox: true },
      },
    },
  });

  if (!campaign) return;

  const activeSatellites = campaign.satellites.filter(
    (sat) => sat.satelliteMailbox.status === InboxStatus.active,
  );
  if (activeSatellites.length === 0) {
    logger.warn({ campaignId }, 'Campaign has no active satellites');
    if (jobId) await completeSendJob(jobId, false, 'No active satellites');
    return;
  }

  const schedule = await prisma.dailySchedule.findUnique({
    where: {
      campaignId_dayIndex: {
        campaignId: campaign.id,
        dayIndex: campaign.currentDay,
      },
    },
  });

  if (!schedule || schedule.actualSends >= schedule.plannedSends) {
    if (jobId) await completeSendJob(jobId, true, 'Daily send quota reached');
    return;
  }

  const satellite = pickWeightedSatellite(activeSatellites);
  const primary = campaign.primaryMailbox;

  if (primary.status !== InboxStatus.active) {
    logger.warn({ campaignId }, 'Primary mailbox is not active');
    if (jobId) await completeSendJob(jobId, false, 'Primary mailbox inactive');
    return;
  }

  if (!isAllowedWarmupSendPair(satellite, primary)) {
    logger.warn(
      { campaignId, satellite: satellite.email, primary: primary.email },
      'Campaign dispatcher skipped invalid warmup pair',
    );
    if (jobId) await completeSendJob(jobId, false, 'Invalid mailbox roles for warmup send');
    return;
  }

  await withInboxGuard(satellite, async () => {
    const { subject, body, validationResult } =
      await aiService.generateInitialEmailWithValidation(primary.industry, {
        senderName: satellite.senderName,
        companyName: satellite.companyName,
      });

    const { messageId } = await emailService.sendMail(satellite, {
      to: primary.email,
      subject,
      body,
    });

    await prisma.$transaction(async (tx) => {
      const current = await tx.dailySchedule.findUnique({
        where: { id: schedule.id },
      });
      if (!current || current.actualSends >= current.plannedSends) {
        return;
      }

      await tx.warmupLog.create({
        data: {
          campaignId: campaign.id,
          senderId: satellite.id,
          receiverId: primary.id,
          messageId,
          subject,
          body,
          status: WarmupStatus.sent,
          validationResult: JSON.stringify(validationResult),
        },
      });

      await tx.dailySchedule.update({
        where: { id: schedule.id },
        data: { actualSends: { increment: 1 } },
      });

      await tx.warmupCampaign.update({
        where: { id: campaign.id },
        data: { lastSendAt: new Date() },
      });
    });

    if (jobId) await completeSendJob(jobId, true);

    logger.info(
      {
        campaignId: campaign.id,
        satellite: satellite.email,
        primary: primary.email,
        messageId,
      },
      'Campaign warmup email dispatched',
    );
  });
}

export async function runCampaignDispatcherJob(): Promise<string> {
  logger.info('Campaign dispatcher job started');

  const campaigns = await prisma.warmupCampaign.findMany({
    where: {
      status: { in: [CampaignStatus.active, CampaignStatus.maintenance] },
    },
    include: {
      primaryMailbox: true,
      satellites: {
        where: { active: true },
        include: { satelliteMailbox: true },
      },
    },
  });

  if (campaigns.length === 0) {
    logger.info('Campaign dispatcher skipped: no active campaigns');
    return 'No active campaigns';
  }

  let processed = 0;

  for (const campaign of campaigns) {
    try {
      await syncCampaignDay(campaign.id);

      const refreshed = await prisma.warmupCampaign.findUnique({
        where: { id: campaign.id },
      });
      if (!refreshed) continue;

      if (!hasElapsedMinJitter(refreshed.lastSendAt)) {
        logger.debug({ campaignId: campaign.id }, 'Campaign within jitter window');
        continue;
      }

      if (
        !isWithinSendWindow(
          refreshed.timezone,
          refreshed.sendWindowStart,
          refreshed.sendWindowEnd,
        )
      ) {
        logger.debug({ campaignId: campaign.id }, 'Campaign outside send window');
        continue;
      }

      const schedule = await prisma.dailySchedule.findUnique({
        where: {
          campaignId_dayIndex: {
            campaignId: campaign.id,
            dayIndex: refreshed.currentDay,
          },
        },
      });

      if (!schedule || schedule.actualSends >= schedule.plannedSends) {
        continue;
      }

      await enqueueSendJob(campaign.id, new Date());
      const job = await claimDueSendJob(campaign.id);
      if (!job) continue;

      await executeCampaignSend(campaign.id, job.id);
      processed += 1;
    } catch (err) {
      logger.error({ err, campaignId: campaign.id }, 'Campaign dispatcher failed');
    }
  }

  logger.info('Campaign dispatcher job finished');
  return `Processed ${processed} send job(s)`;
}