feat: add job queue

packages/lib/package.json

@@ -27,18 +27,19 @@
     "@next-auth/prisma-adapter": "1.0.7",
     "@noble/ciphers": "0.4.0",
     "@noble/hashes": "1.3.2",
+    "@node-rs/bcrypt": "^1.10.0",
     "@pdf-lib/fontkit": "^1.1.1",
     "@scure/base": "^1.1.3",
     "@sindresorhus/slugify": "^2.2.1",
     "@upstash/redis": "^1.20.6",
     "@vvo/tzdb": "^6.117.0",
-    "@node-rs/bcrypt": "^1.10.0",
     "luxon": "^3.4.0",
     "nanoid": "^4.0.2",
     "next": "14.0.3",
     "next-auth": "4.24.5",
     "oslo": "^0.17.0",
     "pdf-lib": "^1.17.1",
+    "pg-boss": "^9.0.3",
     "react": "18.2.0",
     "remeda": "^1.27.1",
     "stripe": "^12.7.0",

packages/lib/server-only/document/seal-document.ts

@@ -18,8 +18,8 @@ import { putFile } from '../../universal/upload/put-file';
 import { flattenAnnotations } from '../pdf/flatten-annotations';
 import { insertFieldInPDF } from '../pdf/insert-field-in-pdf';
 import { normalizeSignatureAppearances } from '../pdf/normalize-signature-appearances';
+import { queueJob } from '../queue/job';
 import { triggerWebhook } from '../webhooks/trigger/trigger-webhook';
-import { sendCompletedEmail } from './send-completed-email';
 
 export type SealDocumentOptions = {
   documentId: number;
@@ -150,7 +150,10 @@ export const sealDocument = async ({
   });
 
   if (sendEmail && !isResealing) {
-    await sendCompletedEmail({ documentId, requestMetadata });
+    await queueJob({
+      job: 'send-completed-email',
+      args: { documentId, requestMetadata },
+    });
   }
 
   await triggerWebhook({

packages/lib/server-only/queue/index.ts

@@ -0,0 +1,52 @@
+import type { WorkHandler } from 'pg-boss';
+import PgBoss from 'pg-boss';
+
+import { jobHandlers } from './job';
+
+type QueueState = {
+  isReady: boolean;
+  queue: PgBoss | null;
+};
+
+let initPromise: Promise<PgBoss> | null = null;
+const state: QueueState = {
+  isReady: false,
+  queue: null,
+};
+
+export async function initQueue() {
+  if (state.isReady) {
+    return state.queue as PgBoss;
+  }
+
+  if (initPromise) {
+    return initPromise;
+  }
+
+  initPromise = (async () => {
+    const queue = new PgBoss({
+      connectionString: 'postgres://postgres:password@127.0.0.1:54321/queue',
+
+      schema: 'documenso_queue',
+    });
+
+    try {
+      await queue.start();
+    } catch (error) {
+      console.error('Failed to start queue', error);
+    }
+
+    await Promise.all(
+      Object.entries(jobHandlers).map(async ([job, jobHandler]) => {
+        await queue.work(job, jobHandler as WorkHandler<unknown>);
+      }),
+    );
+
+    state.isReady = true;
+    state.queue = queue;
+
+    return queue;
+  })();
+
+  return initPromise;
+}

packages/lib/server-only/queue/job.ts

@@ -0,0 +1,34 @@
+import type { WorkHandler } from 'pg-boss';
+
+import { initQueue } from '.';
+import {
+  type SendDocumentOptions as SendCompletedDocumentOptions,
+  sendCompletedEmail,
+} from '../document/send-completed-email';
+
+type JobOptions = {
+  'send-completed-email': SendCompletedDocumentOptions;
+};
+
+export const jobHandlers: {
+  [K in keyof JobOptions]: WorkHandler<JobOptions[K]>;
+} = {
+  'send-completed-email': async ({ data }) => {
+    await sendCompletedEmail({
+      documentId: data.documentId,
+      requestMetadata: data.requestMetadata,
+    });
+  },
+};
+
+export const queueJob = async ({
+  job,
+  args,
+}: {
+  job: keyof JobOptions;
+  args?: JobOptions[keyof JobOptions];
+}) => {
+  const queue = await initQueue();
+
+  await queue.send(job, args ?? {});
+};