fix: enhance campaign finalization process to handle pending emails and ensure accurate status updates

2026-05-06 21:48:54 +02:00
parent 4ddafdc041
commit 2529c9b428
3 changed files with 100 additions and 81 deletions
@@ -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);
@@ -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<void> {
+    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
   */
@@ -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}`);
  }