Add email queue (#1)

* Add pgboss queue support

* Implement queue for sending emails

* Add migrations

apps/web/src/server/api/routers/email.ts

@@ -172,12 +172,21 @@ export const emailRouter = createTRPCRouter({
         where: {
           id: input.id,
         },
-        include: {
+        select: {
           emailEvents: {
             orderBy: {
               createdAt: "asc",
             },
           },
+          id: true,
+          createdAt: true,
+          latestStatus: true,
+          subject: true,
+          to: true,
+          from: true,
+          domainId: true,
+          text: true,
+          html: true,
         },
       });
 

apps/web/src/server/service/email-service.ts

@@ -1,7 +1,7 @@
 import { EmailContent } from "~/types";
 import { db } from "../db";
-import { sendEmailThroughSes, sendEmailWithAttachments } from "../aws/ses";
-import { APP_SETTINGS } from "~/utils/constants";
+import { UnsendApiError } from "~/server/public-api/api-error";
+import { queueEmail } from "./job-service";
 
 export async function sendEmail(
   emailContent: EmailContent & { teamId: number }
@@ -15,72 +15,34 @@ export async function sendEmail(
   });
 
   if (!domain) {
-    throw new Error(
-      "Domain of from email is wrong. Use the email verified by unsend"
-    );
+    throw new UnsendApiError({
+      code: "BAD_REQUEST",
+      message:
+        "Domain of from email is wrong. Use the email verified by unsend",
+    });
   }
 
   if (domain.status !== "SUCCESS") {
-    throw new Error("Domain is not verified");
-  }
-
-  const messageId = attachments
-    ? await sendEmailWithAttachments({
-        to,
-        from,
-        subject,
-        text,
-        html,
-        region: domain.region,
-        configurationSetName: getConfigurationSetName(
-          domain.clickTracking,
-          domain.openTracking
-        ),
-        attachments,
-      })
-    : await sendEmailThroughSes({
-        to,
-        from,
-        subject,
-        text,
-        html,
-        region: domain.region,
-        configurationSetName: getConfigurationSetName(
-          domain.clickTracking,
-          domain.openTracking
-        ),
-        attachments,
-      });
-
-  if (messageId) {
-    return await db.email.create({
-      data: {
-        to,
-        from,
-        subject,
-        text,
-        html,
-        sesEmailId: messageId,
-        teamId,
-        domainId: domain.id,
-      },
+    throw new UnsendApiError({
+      code: "BAD_REQUEST",
+      message: "Domain is not verified",
     });
   }
-}
 
-function getConfigurationSetName(
-  clickTracking: boolean,
-  openTracking: boolean
-) {
-  if (clickTracking && openTracking) {
-    return APP_SETTINGS.SES_CONFIGURATION_FULL;
-  }
-  if (clickTracking) {
-    return APP_SETTINGS.SES_CONFIGURATION_CLICK_TRACKING;
-  }
-  if (openTracking) {
-    return APP_SETTINGS.SES_CONFIGURATION_OPEN_TRACKING;
-  }
+  const email = await db.email.create({
+    data: {
+      to,
+      from,
+      subject,
+      text,
+      html,
+      teamId,
+      domainId: domain.id,
+      attachments: attachments ? JSON.stringify(attachments) : undefined,
+    },
+  });
 
-  return APP_SETTINGS.SES_CONFIGURATION_GENERAL;
+  queueEmail(email.id);
+
+  return email;
 }

apps/web/src/server/service/job-service.ts

@@ -0,0 +1,95 @@
+import pgBoss from "pg-boss";
+import { env } from "~/env";
+import { EmailAttachment, EmailContent } from "~/types";
+import { db } from "../db";
+import { sendEmailThroughSes, sendEmailWithAttachments } from "../aws/ses";
+import { getConfigurationSetName } from "~/utils/ses-utils";
+
+const boss = new pgBoss({
+  connectionString: env.DATABASE_URL,
+  archiveCompletedAfterSeconds: 60 * 60 * 24, // 24 hours
+  deleteAfterDays: 7, // 7 days
+});
+let started = false;
+
+async function getBoss() {
+  if (!started) {
+    await boss.start();
+    await boss.work(
+      "send_email",
+      {
+        teamConcurrency: env.SES_QUEUE_LIMIT,
+        teamSize: env.SES_QUEUE_LIMIT,
+        teamRefill: true,
+      },
+      executeEmail
+    );
+    started = true;
+  }
+  return boss;
+}
+
+export async function queueEmail(emailId: string) {
+  const boss = await getBoss();
+  await boss.send("send_email", { emailId, timestamp: Date.now() });
+}
+
+async function executeEmail(
+  job: pgBoss.Job<{ emailId: string; timestamp: number }>
+) {
+  console.log(
+    `[EmailJob]: Executing email job ${job.data.emailId}, time elapsed: ${Date.now() - job.data.timestamp}ms`
+  );
+
+  const email = await db.email.findUnique({
+    where: { id: job.data.emailId },
+  });
+
+  const domain = email?.domainId
+    ? await db.domain.findUnique({
+        where: { id: email?.domainId },
+      })
+    : null;
+
+  if (!email) {
+    console.log(`[EmailJob]: Email not found, skipping`);
+    return;
+  }
+
+  const attachments: Array<EmailAttachment> = email.attachments
+    ? JSON.parse(email.attachments)
+    : [];
+
+  const messageId = attachments.length
+    ? await sendEmailWithAttachments({
+        to: email.to,
+        from: email.from,
+        subject: email.subject,
+        text: email.text ?? undefined,
+        html: email.html ?? undefined,
+        region: domain?.region ?? env.AWS_DEFAULT_REGION,
+        configurationSetName: getConfigurationSetName(
+          domain?.clickTracking ?? false,
+          domain?.openTracking ?? false
+        ),
+        attachments,
+      })
+    : await sendEmailThroughSes({
+        to: email.to,
+        from: email.from,
+        subject: email.subject,
+        text: email.text ?? undefined,
+        html: email.html ?? undefined,
+        region: domain?.region ?? env.AWS_DEFAULT_REGION,
+        configurationSetName: getConfigurationSetName(
+          domain?.clickTracking ?? false,
+          domain?.openTracking ?? false
+        ),
+        attachments,
+      });
+
+  await db.email.update({
+    where: { id: email.id },
+    data: { sesEmailId: messageId, attachments: undefined },
+  });
+}