diff --git a/apps/api/src/jobs/email-processor.ts b/apps/api/src/jobs/email-processor.ts index 305a7c5..97c6128 100644 --- a/apps/api/src/jobs/email-processor.ts +++ b/apps/api/src/jobs/email-processor.ts @@ -3,13 +3,14 @@ * Processes individual emails from the queue (for all sources: transactional, campaign, workflow) */ -import {CampaignStatus, EmailSourceType, EmailStatus} from '@plunk/db'; +import {EmailSourceType, EmailStatus} from '@plunk/db'; import type {SendEmailJobData} from '@plunk/types'; import {type Job, Worker} from 'bullmq'; import signale from 'signale'; import {DASHBOARD_URI, EMAIL_RATE_LIMIT_PER_SECOND} from '../app/constants.js'; import {prisma} from '../database/prisma.js'; +import {CampaignService} from '../services/CampaignService.js'; import {EmailService} from '../services/EmailService.js'; import {EventService} from '../services/EventService.js'; import {MeterService} from '../services/MeterService.js'; @@ -82,6 +83,12 @@ export async function createEmailWorker() { error: 'Project is disabled', }, }); + + // Cancelled emails are terminal for the campaign — finalize so it doesn't + // stay stuck in SENDING forever waiting on emails that will never be sent. + if (email.campaignId) { + await CampaignService.finalizeIfDone(email.campaignId); + } return; } @@ -226,56 +233,8 @@ export async function createEmailWorker() { sentAt: new Date().toISOString(), }); - // If this email belongs to a campaign, check if all campaign emails have been sent if (email.campaignId) { - const campaign = await prisma.campaign.findUnique({ - where: {id: email.campaignId}, - select: { - id: true, - name: true, - status: true, - totalRecipients: true, - projectId: true, - project: { - select: {name: true}, - }, - }, - }); - - // Only check if campaign is still in SENDING status - if (campaign && campaign.status === CampaignStatus.SENDING) { - // Count how many emails have been sent for this campaign - const sentCount = await prisma.email.count({ - where: { - campaignId: email.campaignId, - sentAt: {not: null}, - }, - }); - - // If all emails have been sent, mark campaign as SENT - if (sentCount >= campaign.totalRecipients) { - await prisma.campaign.update({ - where: {id: email.campaignId}, - data: { - status: CampaignStatus.SENT, - sentCount, - }, - }); - - signale.success( - `[EMAIL-PROCESSOR] Campaign ${campaign.name} completed: ${sentCount}/${campaign.totalRecipients} emails sent`, - ); - - // Send notification about campaign send completed - const {NtfyService} = await import('../services/NtfyService.js'); - await NtfyService.notifyCampaignSendCompleted( - campaign.name, - campaign.project.name, - campaign.projectId, - campaign.totalRecipients, - ); - } - } + await CampaignService.finalizeIfDone(email.campaignId); } } catch (error) { signale.error(`[EMAIL-PROCESSOR] Failed to send email ${emailId}:`, error); diff --git a/apps/api/src/services/CampaignService.ts b/apps/api/src/services/CampaignService.ts index 809c66f..95eb938 100644 --- a/apps/api/src/services/CampaignService.ts +++ b/apps/api/src/services/CampaignService.ts @@ -1,5 +1,5 @@ import type {Campaign, Contact, Prisma} from '@plunk/db'; -import {CampaignAudienceType, CampaignStatus, EmailSourceType, TemplateType} from '@plunk/db'; +import {CampaignAudienceType, CampaignStatus, EmailSourceType, EmailStatus, TemplateType} from '@plunk/db'; import type {CreateCampaignData, FilterCondition, PaginatedResponse, UpdateCampaignData} from '@plunk/types'; import {fromPrismaJson, toPrismaJson} from '@plunk/types'; import signale from 'signale'; @@ -511,45 +511,71 @@ export class CampaignService { // segment after totalRecipients was calculated are silently skipped. Without this // reconciliation, sentCount can never reach the original totalRecipients and the // campaign remains stuck in SENDING forever. - const [actualEmailCount, alreadySentCount] = await Promise.all([ - prisma.email.count({where: {campaignId}}), - prisma.email.count({where: {campaignId, sentAt: {not: null}}}), - ]); + const actualEmailCount = await prisma.email.count({where: {campaignId}}); await prisma.campaign.update({ where: {id: campaignId}, data: {totalRecipients: actualEmailCount}, }); - // If all emails were already processed by the time the last batch finished - // (race: worker was faster than the batch chain), finalize the campaign now. - if (actualEmailCount === 0 || alreadySentCount >= actualEmailCount) { - const finalCampaign = await prisma.campaign.findUnique({ - where: {id: campaignId}, - include: {project: {select: {name: true}}}, - }); - - if (finalCampaign && finalCampaign.status === CampaignStatus.SENDING) { - await prisma.campaign.update({ - where: {id: campaignId}, - data: {status: CampaignStatus.SENT, sentCount: alreadySentCount}, - }); - - signale.success( - `[CAMPAIGN] Campaign ${finalCampaign.name} finalized after last batch: ${alreadySentCount}/${actualEmailCount} emails sent`, - ); - - await NtfyService.notifyCampaignSendCompleted( - finalCampaign.name, - finalCampaign.project.name, - finalCampaign.projectId, - alreadySentCount, - ); - } - } + await this.finalizeIfDone(campaignId); } } + /** + * Finalize a SENDING campaign if every email has reached a terminal state. + * Terminal = sentAt is set OR status is FAILED. Counting FAILED as terminal + * unsticks campaigns where some emails couldn't be delivered (e.g. the project + * was disabled mid-send), so the campaign moves to SENT with a partial sentCount. + */ + public static async finalizeIfDone(campaignId: string): Promise { + const campaign = await prisma.campaign.findUnique({ + where: {id: campaignId}, + select: { + id: true, + name: true, + status: true, + totalRecipients: true, + projectId: true, + project: {select: {name: true}}, + }, + }); + + if (!campaign || campaign.status !== CampaignStatus.SENDING) { + return; + } + + const [processedCount, sentCount] = await Promise.all([ + prisma.email.count({ + where: { + campaignId, + OR: [{sentAt: {not: null}}, {status: EmailStatus.FAILED}], + }, + }), + prisma.email.count({where: {campaignId, sentAt: {not: null}}}), + ]); + + if (campaign.totalRecipients > 0 && processedCount < campaign.totalRecipients) { + return; + } + + await prisma.campaign.update({ + where: {id: campaignId}, + data: {status: CampaignStatus.SENT, sentCount}, + }); + + signale.success( + `[CAMPAIGN] Campaign ${campaign.name} finalized: ${sentCount}/${campaign.totalRecipients} emails sent`, + ); + + await NtfyService.notifyCampaignSendCompleted( + campaign.name, + campaign.project.name, + campaign.projectId, + sentCount, + ); + } + /** * Cancel a campaign */ diff --git a/apps/api/src/services/QueueService.ts b/apps/api/src/services/QueueService.ts index 747d5e2..ef8f38f 100644 --- a/apps/api/src/services/QueueService.ts +++ b/apps/api/src/services/QueueService.ts @@ -1,3 +1,4 @@ +import {CampaignStatus, EmailStatus} from '@plunk/db'; import {type Job, Queue} from 'bullmq'; import type {RedisOptions} from 'ioredis'; import signale from 'signale'; @@ -578,6 +579,39 @@ export class QueueService { } } + // Mark every still-PENDING email for this project as FAILED. We just stripped + // their queue jobs, so without this they'd sit as PENDING forever and any + // campaign waiting on them would stay stuck in SENDING. + const failed = await prisma.email.updateMany({ + where: {projectId, status: EmailStatus.PENDING}, + data: {status: EmailStatus.FAILED, error: 'Project is disabled'}, + }); + + if (failed.count > 0) { + signale.info(`[QUEUE] Marked ${failed.count} pending emails as failed for project ${projectId}`); + } + + // Finalize any in-flight campaigns. With the orphaned PENDING emails now FAILED + // (terminal), the campaign can move to SENT with a partial sentCount instead of + // staying stuck in SENDING. Reconcile totalRecipients first since the batch + // chain may have been cut short. + const sendingCampaigns = await prisma.campaign.findMany({ + where: {projectId, status: CampaignStatus.SENDING}, + select: {id: true}, + }); + + if (sendingCampaigns.length > 0) { + const {CampaignService} = await import('./CampaignService.js'); + for (const campaign of sendingCampaigns) { + const actualEmailCount = await prisma.email.count({where: {campaignId: campaign.id}}); + await prisma.campaign.update({ + where: {id: campaign.id}, + data: {totalRecipients: actualEmailCount}, + }); + await CampaignService.finalizeIfDone(campaign.id); + } + } + signale.info(`[QUEUE] Finished cancelling jobs for project ${projectId}`); }