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,
  reclaimStaleSendJobs,
} 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.paused) {
    logger.warn({ campaignId }, 'Primary mailbox is paused — skipping send');
    if (jobId) await completeSendJob(jobId, false, 'Primary mailbox paused');
    return;
  }

  if (primary.status === InboxStatus.error) {
    logger.warn(
      { campaignId, primary: primary.email },
      'Primary mailbox in error — proceeding with satellite send (IMAP not required)',
    );
  }

  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;
  }

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

    await withInboxGuard(satellite, async () => {
      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() },
        });
      });

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

    if (jobId) await completeSendJob(jobId, true);
  } catch (err) {
    const message = err instanceof Error ? err.message : 'Campaign send failed';
    if (jobId) await completeSendJob(jobId, false, message);
    throw err;
  }
}

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

  await reclaimStaleSendJobs();

  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)`;
}