CronJob to process Messages (#14)

Author: dcodesdev

RELEASE_NOTES.md

@@ -1,10 +1,13 @@
 ### ✨ Features
 
+- Uses cron job to process queued campaigns, processes messages in batches, much more efficient for large campaigns.
+- Theme toggle updated.
+- Loading state for API keys.
 - New features and improvements.
 
 ### 🐛 Bug Fixes
 
-- Template preview fixed.
+- Recipient count on campaign page was incorrect.
 - Various bug fixes and optimizations.
 
 ### 📚 Docs

apps/backend/src/campaign/mutation.ts

@@ -3,13 +3,7 @@ import { authProcedure } from "../trpc"
 import { prisma } from "../utils/prisma"
 import { TRPCError } from "@trpc/server"
 import pMap from "p-map"
-import { LinkTracker } from "../lib/LinkTracker"
-import { v4 as uuidV4 } from "uuid"
 import { Mailer } from "../lib/Mailer"
-import {
-  replacePlaceholders,
-  PlaceholderDataKey,
-} from "../utils/placeholder-parser"
 
 const createCampaignSchema = z.object({
   title: z.string().min(1, "Campaign title is required"),
@@ -378,105 +372,16 @@ export const startCampaign = authProcedure
       })
     }
 
-    const updatedCampaign = await prisma.$transaction(async (tx) => {
-      const status =
-        campaign.scheduledAt && campaign.scheduledAt > new Date()
-          ? "SCHEDULED"
-          : "SENDING"
-
-      const linkTracker = new LinkTracker(tx)
-
-      const messagesData = await Promise.all(
-        Array.from(subscribers.values()).map(async (subscriber) => {
-          if (!campaign.content) {
-            throw new TRPCError({
-              code: "BAD_REQUEST",
-              message: "Campaign must have content",
-            })
-          }
-
-          if (!generalSettings.baseURL) {
-            throw new TRPCError({
-              code: "BAD_REQUEST",
-              message:
-                "Base URL must be configured in settings before running a campaign",
-            })
-          }
-
-          const messageId = uuidV4()
-          let content = campaign.Template
-            ? campaign.Template.content.replace(
-                /{{content}}/g,
-                campaign.content
-              )
-            : campaign.content
-
-          const placeholderData: Partial<Record<PlaceholderDataKey, string>> = {
-            "subscriber.email": subscriber.email,
-            "campaign.name": campaign.title,
-            "campaign.subject": campaign.subject ?? "",
-            "organization.name": organization.name,
-            unsubscribe_link: `${generalSettings.baseURL}/unsubscribe?sid=${subscriber.id}&cid=${campaign.id}&mid=${messageId}`,
-            current_date: new Date().toLocaleDateString("en-CA"),
-          }
-
-          if (campaign.openTracking) {
-            content += `<img src="${generalSettings.baseURL}/img/${messageId}/img.png" alt="" width="1" height="1" style="display:none" />`
-          }
-
-          if (subscriber.name) {
-            placeholderData["subscriber.name"] = subscriber.name
-          }
-
-          if (subscriber.Metadata) {
-            for (const meta of subscriber.Metadata) {
-              placeholderData[`subscriber.metadata.${meta.key}`] = meta.value
-            }
-          }
-
-          content = replacePlaceholders(content, placeholderData)
-
-          const { content: finalContent, trackedIds } =
-            await linkTracker.replaceMessageContentWithTrackedLinks(
-              content,
-              campaign.id,
-              generalSettings.baseURL
-            )
-
-          return {
-            message: {
-              id: messageId,
-              subscriberId: subscriber.id,
-              content: finalContent,
-            },
-            subscriberTrackedLinks: trackedIds.map((trackedLinkId) => ({
-              messageId,
-              subscriberId: subscriber.id,
-              trackedLinkId,
-            })),
-          }
-        })
-      )
-
-      const createdCampaign = await tx.campaign.update({
-        where: { id: campaign.id },
-        data: {
-          status,
-          Messages: {
-            create: messagesData.map(({ message }) => message),
-          },
-        },
-        include: {
-          Messages: true,
-          _count: {
-            select: {
-              Messages: true,
-            },
-          },
-        },
-      })
+    const status =
+      campaign.scheduledAt && campaign.scheduledAt > new Date()
+        ? "SCHEDULED"
+        : "SENDING"
 
-      return createdCampaign
+    const updatedCampaign = await prisma.campaign.update({
+      where: { id: campaign.id },
+      data: {
+        status,
+      },
     })
 
     return { campaign: updatedCampaign }

apps/backend/src/campaign/query.ts

@@ -114,19 +114,7 @@ export const getCampaign = authProcedure
         Template: true,
         CampaignLists: {
           include: {
-            List: {
-              include: {
-                _count: {
-                  select: {
-                    ListSubscribers: {
-                      where: {
-                        unsubscribedAt: null,
-                      },
-                    },
-                  },
-                },
-              },
-            },
+            List: true,
           },
         },
       },
@@ -139,6 +127,46 @@ export const getCampaign = authProcedure
       })
     }
 
+    const listSubscribers = await prisma.listSubscriber.findMany({
+      where: {
+        listId: {
+          in: campaign.CampaignLists.map((cl) => cl.listId),
+        },
+        unsubscribedAt: null,
+      },
+      select: {
+        id: true,
+      },
+      distinct: ["subscriberId"],
+    })
+
+    // Add the count to each list for backward compatibility
+    const campaignWithCounts = {
+      ...campaign,
+      CampaignLists: await Promise.all(
+        campaign.CampaignLists.map(async (cl) => {
+          const count = await prisma.listSubscriber.count({
+            where: {
+              listId: cl.listId,
+              unsubscribedAt: null,
+            },
+          })
+
+          return {
+            ...cl,
+            List: {
+              ...cl.List,
+              _count: {
+                ListSubscribers: count,
+              },
+            },
+          }
+        })
+      ),
+      // Add the unique subscriber count directly to the campaign object
+      uniqueRecipientCount: listSubscribers.length,
+    }
+
     const promises = {
       totalMessages: prisma.message.count({
         where: {
@@ -198,7 +226,7 @@ export const getCampaign = authProcedure
     const result = await resolveProps(promises)
 
     return {
-      campaign,
+      campaign: campaignWithCounts,
       stats: {
         totalMessages: result.totalMessages,
         queuedMessages: result.queuedMessages,

apps/backend/src/cron/cron.ts

@@ -1,6 +1,7 @@
 import cron from "node-cron"
 import { sendMessagesCron } from "./sendMessages"
 import { dailyMaintenanceCron } from "./dailyMaintenance"
+import { processQueuedCampaigns } from "./processQueuedCampaigns"
 
 type CronJob = {
   name: string
@@ -23,7 +24,18 @@ const dailyMaintenanceJob: CronJob = {
   enabled: true,
 }
 
-const cronJobs: CronJob[] = [sendMessagesJob, dailyMaintenanceJob]
+const processQueuedCampaignsJob: CronJob = {
+  name: "process-queued-campaigns",
+  schedule: "* * * * * *", // Runs every second
+  job: processQueuedCampaigns,
+  enabled: true,
+}
+
+const cronJobs: CronJob[] = [
+  sendMessagesJob,
+  dailyMaintenanceJob,
+  processQueuedCampaignsJob,
+]
 
 export const initializeCronJobs = () => {
   const scheduledJobs = cronJobs