feat: add queue for sending emails

docker/development/compose.yml

@@ -11,6 +11,17 @@ services:
     ports:
       - 54320:5432
 
+  queue:
+    image: postgres:15
+    container_name: queue
+    user: postgres
+    command: -c jit=off
+    environment:
+      POSTGRES_PASSWORD: password
+      POSTGRES_DB: queue
+    ports:
+      - 54321:5432
+
   inbucket:
     image: inbucket/inbucket
     container_name: mailserver

package-lock.json

@@ -8443,6 +8443,18 @@
         "node": ">= 6.0.0"
       }
     },
+    "node_modules/aggregate-error": {
+      "version": "3.1.0",
+      "resolved": "https://registry.npmjs.org/aggregate-error/-/aggregate-error-3.1.0.tgz",
+      "integrity": "sha512-4I7Td01quW/RpocfNayFdFVk1qSuoh0E7JrbRJ16nH01HhKFQ88INq9Sd+nd72zqRySlr9BmDA8xlEJ6vJMrYA==",
+      "dependencies": {
+        "clean-stack": "^2.0.0",
+        "indent-string": "^4.0.0"
+      },
+      "engines": {
+        "node": ">=8"
+      }
+    },
     "node_modules/ajv": {
       "version": "8.12.0",
       "resolved": "https://registry.npmjs.org/ajv/-/ajv-8.12.0.tgz",
@@ -9417,6 +9429,14 @@
       "resolved": "https://registry.npmjs.org/classnames/-/classnames-2.5.1.tgz",
       "integrity": "sha512-saHYOzhIQs6wy2sVxTM6bUDsQO4F50V9RQ22qBpEdCW+I+/Wmke2HOl6lS6dTpdxVhb88/I6+Hs+438c3lfUow=="
     },
+    "node_modules/clean-stack": {
+      "version": "2.2.0",
+      "resolved": "https://registry.npmjs.org/clean-stack/-/clean-stack-2.2.0.tgz",
+      "integrity": "sha512-4diC9HaTE+KRAMWhDhrGOECgWZxoevMc5TlkObMqNSsVU62PYzXZ/SMTjzyGAFF1YusgxGcSWTEXBhp0CPwQ1A==",
+      "engines": {
+        "node": ">=6"
+      }
+    },
     "node_modules/cli-cursor": {
       "version": "4.0.0",
       "resolved": "https://registry.npmjs.org/cli-cursor/-/cli-cursor-4.0.0.tgz",
@@ -10192,6 +10212,17 @@
       "integrity": "sha512-dcKFX3jn0MpIaXjisoRvexIJVEKzaq7z2rZKxf+MSr9TkdmHmsU4m2lcLojrj/FHl8mk5VxMmYA+ftRkP/3oKQ==",
       "devOptional": true
     },
+    "node_modules/cron-parser": {
+      "version": "4.9.0",
+      "resolved": "https://registry.npmjs.org/cron-parser/-/cron-parser-4.9.0.tgz",
+      "integrity": "sha512-p0SaNjrHOnQeR8/VnfGbmg9te2kfyYSQ7Sc/j/6DtPL3JQvKxmjO9TSjNFpujqV3vEYYBvNNvXSxzyksBWAx1Q==",
+      "dependencies": {
+        "luxon": "^3.2.1"
+      },
+      "engines": {
+        "node": ">=12.0.0"
+      }
+    },
     "node_modules/cross-fetch": {
       "version": "4.0.0",
       "resolved": "https://registry.npmjs.org/cross-fetch/-/cross-fetch-4.0.0.tgz",
@@ -10558,6 +10589,17 @@
         "url": "https://github.com/sponsors/ljharb"
       }
     },
+    "node_modules/delay": {
+      "version": "5.0.0",
+      "resolved": "https://registry.npmjs.org/delay/-/delay-5.0.0.tgz",
+      "integrity": "sha512-ReEBKkIfe4ya47wlPYf/gu5ib6yUG0/Aez0JQZQz94kiWtRQvZIQbTiehsnwHvLSWJnQdhVeqYue7Id1dKr0qw==",
+      "engines": {
+        "node": ">=10"
+      },
+      "funding": {
+        "url": "https://github.com/sponsors/sindresorhus"
+      }
+    },
     "node_modules/delayed-stream": {
       "version": "1.0.0",
       "resolved": "https://registry.npmjs.org/delayed-stream/-/delayed-stream-1.0.0.tgz",
@@ -13672,7 +13714,6 @@
       "version": "4.0.0",
       "resolved": "https://registry.npmjs.org/indent-string/-/indent-string-4.0.0.tgz",
       "integrity": "sha512-EdDDZu4A2OyIK7Lr/2zG+w5jmbuk1DVBnEwREQvBzspBJkCEbRa8GxU1lghYcaGJCnRWibjDXlq779X1/y5xwg==",
-      "dev": true,
       "engines": {
         "node": ">=8"
       }
@@ -17250,6 +17291,20 @@
         "url": "https://github.com/sponsors/sindresorhus"
       }
     },
+    "node_modules/p-map": {
+      "version": "4.0.0",
+      "resolved": "https://registry.npmjs.org/p-map/-/p-map-4.0.0.tgz",
+      "integrity": "sha512-/bjOqmgETBYB5BoEeGVea8dmvHb2m9GLy1E9W43yeyfP6QQCZGFNa+XRceJEuDB6zqr+gKpIAmlLebMpykw/MQ==",
+      "dependencies": {
+        "aggregate-error": "^3.0.0"
+      },
+      "engines": {
+        "node": ">=10"
+      },
+      "funding": {
+        "url": "https://github.com/sponsors/sindresorhus"
+      }
+    },
     "node_modules/p-try": {
       "version": "2.2.0",
       "resolved": "https://registry.npmjs.org/p-try/-/p-try-2.2.0.tgz",
@@ -17559,6 +17614,124 @@
         "is-reference": "^3.0.0"
       }
     },
+    "node_modules/pg": {
+      "version": "8.11.5",
+      "resolved": "https://registry.npmjs.org/pg/-/pg-8.11.5.tgz",
+      "integrity": "sha512-jqgNHSKL5cbDjFlHyYsCXmQDrfIX/3RsNwYqpd4N0Kt8niLuNoRNH+aazv6cOd43gPh9Y4DjQCtb+X0MH0Hvnw==",
+      "dependencies": {
+        "pg-connection-string": "^2.6.4",
+        "pg-pool": "^3.6.2",
+        "pg-protocol": "^1.6.1",
+        "pg-types": "^2.1.0",
+        "pgpass": "1.x"
+      },
+      "engines": {
+        "node": ">= 8.0.0"
+      },
+      "optionalDependencies": {
+        "pg-cloudflare": "^1.1.1"
+      },
+      "peerDependencies": {
+        "pg-native": ">=3.0.1"
+      },
+      "peerDependenciesMeta": {
+        "pg-native": {
+          "optional": true
+        }
+      }
+    },
+    "node_modules/pg-boss": {
+      "version": "9.0.3",
+      "resolved": "https://registry.npmjs.org/pg-boss/-/pg-boss-9.0.3.tgz",
+      "integrity": "sha512-cUWUiv3sr563yNy0nCZ25Tv5U0m59Y9MhX/flm0vTR012yeVCrqpfboaZP4xFOQPdWipMJpuu4g94HR0SncTgw==",
+      "dependencies": {
+        "cron-parser": "^4.0.0",
+        "delay": "^5.0.0",
+        "lodash.debounce": "^4.0.8",
+        "p-map": "^4.0.0",
+        "pg": "^8.5.1",
+        "serialize-error": "^8.1.0",
+        "uuid": "^9.0.0"
+      },
+      "engines": {
+        "node": ">=16"
+      }
+    },
+    "node_modules/pg-boss/node_modules/uuid": {
+      "version": "9.0.1",
+      "resolved": "https://registry.npmjs.org/uuid/-/uuid-9.0.1.tgz",
+      "integrity": "sha512-b+1eJOlsR9K8HJpow9Ok3fiWOWSIcIzXodvv0rQjVoOVNpWMpxf1wZNpt4y9h10odCNrqnYp1OBzRktckBe3sA==",
+      "funding": [
+        "https://github.com/sponsors/broofa",
+        "https://github.com/sponsors/ctavan"
+      ],
+      "bin": {
+        "uuid": "dist/bin/uuid"
+      }
+    },
+    "node_modules/pg-cloudflare": {
+      "version": "1.1.1",
+      "resolved": "https://registry.npmjs.org/pg-cloudflare/-/pg-cloudflare-1.1.1.tgz",
+      "integrity": "sha512-xWPagP/4B6BgFO+EKz3JONXv3YDgvkbVrGw2mTo3D6tVDQRh1e7cqVGvyR3BE+eQgAvx1XhW/iEASj4/jCWl3Q==",
+      "optional": true
+    },
+    "node_modules/pg-connection-string": {
+      "version": "2.6.4",
+      "resolved": "https://registry.npmjs.org/pg-connection-string/-/pg-connection-string-2.6.4.tgz",
+      "integrity": "sha512-v+Z7W/0EO707aNMaAEfiGnGL9sxxumwLl2fJvCQtMn9Fxsg+lPpPkdcyBSv/KFgpGdYkMfn+EI1Or2EHjpgLCA=="
+    },
+    "node_modules/pg-int8": {
+      "version": "1.0.1",
+      "resolved": "https://registry.npmjs.org/pg-int8/-/pg-int8-1.0.1.tgz",
+      "integrity": "sha512-WCtabS6t3c8SkpDBUlb1kjOs7l66xsGdKpIPZsg4wR+B3+u9UAum2odSsF9tnvxg80h4ZxLWMy4pRjOsFIqQpw==",
+      "engines": {
+        "node": ">=4.0.0"
+      }
+    },
+    "node_modules/pg-pool": {
+      "version": "3.6.2",
+      "resolved": "https://registry.npmjs.org/pg-pool/-/pg-pool-3.6.2.tgz",
+      "integrity": "sha512-Htjbg8BlwXqSBQ9V8Vjtc+vzf/6fVUuak/3/XXKA9oxZprwW3IMDQTGHP+KDmVL7rtd+R1QjbnCFPuTHm3G4hg==",
+      "peerDependencies": {
+        "pg": ">=8.0"
+      }
+    },
+    "node_modules/pg-protocol": {
+      "version": "1.6.1",
+      "resolved": "https://registry.npmjs.org/pg-protocol/-/pg-protocol-1.6.1.tgz",
+      "integrity": "sha512-jPIlvgoD63hrEuihvIg+tJhoGjUsLPn6poJY9N5CnlPd91c2T18T/9zBtLxZSb1EhYxBRoZJtzScCaWlYLtktg=="
+    },
+    "node_modules/pg-types": {
+      "version": "2.2.0",
+      "resolved": "https://registry.npmjs.org/pg-types/-/pg-types-2.2.0.tgz",
+      "integrity": "sha512-qTAAlrEsl8s4OiEQY69wDvcMIdQN6wdz5ojQiOy6YRMuynxenON0O5oCpJI6lshc6scgAY8qvJ2On/p+CXY0GA==",
+      "dependencies": {
+        "pg-int8": "1.0.1",
+        "postgres-array": "~2.0.0",
+        "postgres-bytea": "~1.0.0",
+        "postgres-date": "~1.0.4",
+        "postgres-interval": "^1.1.0"
+      },
+      "engines": {
+        "node": ">=4"
+      }
+    },
+    "node_modules/pgpass": {
+      "version": "1.0.5",
+      "resolved": "https://registry.npmjs.org/pgpass/-/pgpass-1.0.5.tgz",
+      "integrity": "sha512-FdW9r/jQZhSeohs1Z3sI1yxFQNFvMcnmfuj4WBMUTxOrAyLMaTcE1aAMBiTlbMNaXvBCQuVi0R7hd8udDSP7ug==",
+      "dependencies": {
+        "split2": "^4.1.0"
+      }
+    },
+    "node_modules/pgpass/node_modules/split2": {
+      "version": "4.2.0",
+      "resolved": "https://registry.npmjs.org/split2/-/split2-4.2.0.tgz",
+      "integrity": "sha512-UcjcJOWknrNkF6PLX83qcHM6KHgVKNkV62Y8a5uYDVv9ydGQVwAHMKqHdJje1VTWpljG0WYpCDhrCdAOYH4TWg==",
+      "engines": {
+        "node": ">= 10.x"
+      }
+    },
     "node_modules/picocolors": {
       "version": "1.0.0",
       "resolved": "https://registry.npmjs.org/picocolors/-/picocolors-1.0.0.tgz",
@@ -17817,6 +17990,41 @@
       "resolved": "https://registry.npmjs.org/postcss-value-parser/-/postcss-value-parser-4.2.0.tgz",
       "integrity": "sha512-1NNCs6uurfkVbeXG4S8JFT9t19m45ICnif8zWLd5oPSZ50QnwMfK+H3jv408d4jw/7Bttv5axS5IiHoLaVNHeQ=="
     },
+    "node_modules/postgres-array": {
+      "version": "2.0.0",
+      "resolved": "https://registry.npmjs.org/postgres-array/-/postgres-array-2.0.0.tgz",
+      "integrity": "sha512-VpZrUqU5A69eQyW2c5CA1jtLecCsN2U/bD6VilrFDWq5+5UIEVO7nazS3TEcHf1zuPYO/sqGvUvW62g86RXZuA==",
+      "engines": {
+        "node": ">=4"
+      }
+    },
+    "node_modules/postgres-bytea": {
+      "version": "1.0.0",
+      "resolved": "https://registry.npmjs.org/postgres-bytea/-/postgres-bytea-1.0.0.tgz",
+      "integrity": "sha512-xy3pmLuQqRBZBXDULy7KbaitYqLcmxigw14Q5sj8QBVLqEwXfeybIKVWiqAXTlcvdvb0+xkOtDbfQMOf4lST1w==",
+      "engines": {
+        "node": ">=0.10.0"
+      }
+    },
+    "node_modules/postgres-date": {
+      "version": "1.0.7",
+      "resolved": "https://registry.npmjs.org/postgres-date/-/postgres-date-1.0.7.tgz",
+      "integrity": "sha512-suDmjLVQg78nMK2UZ454hAG+OAW+HQPZ6n++TNDUX+L0+uUlLywnoxJKDou51Zm+zTCjrCl0Nq6J9C5hP9vK/Q==",
+      "engines": {
+        "node": ">=0.10.0"
+      }
+    },
+    "node_modules/postgres-interval": {
+      "version": "1.2.0",
+      "resolved": "https://registry.npmjs.org/postgres-interval/-/postgres-interval-1.2.0.tgz",
+      "integrity": "sha512-9ZhXKM/rw350N1ovuWHbGxnGh/SNJ4cnxHiM0rxE4VN41wsg8P8zWn9hv/buK00RP4WvlOyr/RBDiptyxVbkZQ==",
+      "dependencies": {
+        "xtend": "^4.0.0"
+      },
+      "engines": {
+        "node": ">=0.10.0"
+      }
+    },
     "node_modules/posthog-js": {
       "version": "1.93.2",
       "resolved": "https://registry.npmjs.org/posthog-js/-/posthog-js-1.93.2.tgz",
@@ -24926,6 +25134,7 @@
         "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/email/mailer.ts

@@ -1,3 +1,4 @@
+import type { SendMailOptions } from 'nodemailer';
 import { createTransport } from 'nodemailer';
 
 import { ResendTransport } from '@documenso/nodemailer-resend';
@@ -54,3 +55,4 @@ const getTransport = () => {
 };
 
 export const mailer = getTransport();
+export type MailOptions = SendMailOptions;

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-meta/upsert-document-meta.ts

@@ -2,12 +2,11 @@
 
 import { DOCUMENT_AUDIT_LOG_TYPE } from '@documenso/lib/types/document-audit-logs';
 import type { RequestMetadata } from '@documenso/lib/universal/extract-request-metadata';
-import {
+import { diffDocumentMetaChanges } from '@documenso/lib/utils/document-audit-logs';
-  createDocumentAuditLogData,
-  diffDocumentMetaChanges,
-} from '@documenso/lib/utils/document-audit-logs';
 import { prisma } from '@documenso/prisma';
 
+import { queueJob } from '../queue/job';
+
 export type CreateDocumentMetaOptions = {
   documentId: number;
   subject?: string;
@@ -65,8 +64,7 @@ export const upsertDocumentMeta = async ({
     },
   });
 
-  return await prisma.$transaction(async (tx) => {
+  const upsertedDocumentMeta = await prisma.documentMeta.upsert({
-    const upsertedDocumentMeta = await tx.documentMeta.upsert({
     where: {
       documentId,
     },
@@ -92,8 +90,9 @@ export const upsertDocumentMeta = async ({
   const changes = diffDocumentMetaChanges(originalDocumentMeta ?? {}, upsertedDocumentMeta);
 
   if (changes.length > 0) {
-      await tx.documentAuditLog.create({
+    await queueJob({
-        data: createDocumentAuditLogData({
+      job: 'create-document-audit-log',
+      args: {
         type: DOCUMENT_AUDIT_LOG_TYPE.DOCUMENT_META_UPDATED,
         documentId,
         user,
@@ -101,10 +100,9 @@ export const upsertDocumentMeta = async ({
         data: {
           changes: diffDocumentMetaChanges(originalDocumentMeta ?? {}, upsertedDocumentMeta),
         },
-        }),
+      },
     });
   }
 
   return upsertedDocumentMeta;
-  });
 };

packages/lib/server-only/document/complete-document-with-token.ts

@@ -2,15 +2,14 @@
 
 import { DOCUMENT_AUDIT_LOG_TYPE } from '@documenso/lib/types/document-audit-logs';
 import type { RequestMetadata } from '@documenso/lib/universal/extract-request-metadata';
-import { createDocumentAuditLogData } from '@documenso/lib/utils/document-audit-logs';
 import { prisma } from '@documenso/prisma';
 import { DocumentStatus, SigningStatus } from '@documenso/prisma/client';
 import { WebhookTriggerEvents } from '@documenso/prisma/client';
 
 import type { TRecipientActionAuth } from '../../types/document-auth';
+import { queueJob } from '../queue/job';
 import { triggerWebhook } from '../webhooks/trigger/trigger-webhook';
 import { sealDocument } from './seal-document';
-import { sendPendingEmail } from './send-pending-email';
 
 export type CompleteDocumentWithTokenOptions = {
   token: string;
@@ -93,8 +92,7 @@ export const completeDocumentWithToken = async ({
   //   throw new AppError(AppErrorCode.UNAUTHORIZED, 'Invalid authentication values');
   // }
 
-  await prisma.$transaction(async (tx) => {
+  await prisma.recipient.update({
-    await tx.recipient.update({
     where: {
       id: recipient.id,
     },
@@ -104,8 +102,9 @@ export const completeDocumentWithToken = async ({
     },
   });
 
-    await tx.documentAuditLog.create({
+  await queueJob({
-      data: createDocumentAuditLogData({
+    job: 'create-document-audit-log',
+    args: {
       type: DOCUMENT_AUDIT_LOG_TYPE.DOCUMENT_RECIPIENT_COMPLETED,
       documentId: document.id,
       user: {
@@ -120,8 +119,7 @@ export const completeDocumentWithToken = async ({
         recipientRole: recipient.role,
         // actionAuth: derivedRecipientActionAuth || undefined,
       },
-      }),
+    },
-    });
   });
 
   const pendingRecipients = await prisma.recipient.count({
@@ -134,7 +132,13 @@ export const completeDocumentWithToken = async ({
   });
 
   if (pendingRecipients > 0) {
-    await sendPendingEmail({ documentId, recipientId: recipient.id });
+    await queueJob({
+      job: 'send-pending-email',
+      args: {
+        documentId: document.id,
+        recipientId: recipient.id,
+      },
+    });
   }
 
   const documents = await prisma.document.updateMany({

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

@@ -3,10 +3,10 @@
 import { AppError, AppErrorCode } from '@documenso/lib/errors/app-error';
 import { DOCUMENT_AUDIT_LOG_TYPE } from '@documenso/lib/types/document-audit-logs';
 import type { RequestMetadata } from '@documenso/lib/universal/extract-request-metadata';
-import { createDocumentAuditLogData } from '@documenso/lib/utils/document-audit-logs';
 import { prisma } from '@documenso/prisma';
 import { WebhookTriggerEvents } from '@documenso/prisma/client';
 
+import { queueJob } from '../queue/job';
 import { triggerWebhook } from '../webhooks/trigger/trigger-webhook';
 
 export type CreateDocumentOptions = {
@@ -44,8 +44,7 @@ export const createDocument = async ({
     throw new AppError(AppErrorCode.NOT_FOUND, 'Team not found');
   }
 
-  return await prisma.$transaction(async (tx) => {
+  const document = await prisma.document.create({
-    const document = await tx.document.create({
     data: {
       title,
       documentDataId,
@@ -54,8 +53,9 @@ export const createDocument = async ({
     },
   });
 
-    await tx.documentAuditLog.create({
+  await queueJob({
-      data: createDocumentAuditLogData({
+    job: 'create-document-audit-log',
+    args: {
       type: DOCUMENT_AUDIT_LOG_TYPE.DOCUMENT_CREATED,
       documentId: document.id,
       user,
@@ -63,7 +63,7 @@ export const createDocument = async ({
       data: {
         title,
       },
-      }),
+    },
   });
 
   await triggerWebhook({
@@ -74,5 +74,4 @@ export const createDocument = async ({
   });
 
   return document;
-  });
 };

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

@@ -2,7 +2,6 @@
 
 import { createElement } from 'react';
 
-import { mailer } from '@documenso/email/mailer';
 import { render } from '@documenso/email/render';
 import DocumentCancelTemplate from '@documenso/email/templates/document-cancel';
 import { prisma } from '@documenso/prisma';
@@ -12,7 +11,7 @@ import { NEXT_PUBLIC_WEBAPP_URL } from '../../constants/app';
 import { FROM_ADDRESS, FROM_NAME } from '../../constants/email';
 import { DOCUMENT_AUDIT_LOG_TYPE } from '../../types/document-audit-logs';
 import type { RequestMetadata } from '../../universal/extract-request-metadata';
-import { createDocumentAuditLogData } from '../../utils/document-audit-logs';
+import { queueJob } from '../queue/job';
 
 export type DeleteDocumentOptions = {
   id: number;
@@ -61,11 +60,11 @@ export const deleteDocument = async ({
 
   // if the document is a draft, hard-delete
   if (status === DocumentStatus.DRAFT) {
-    return await prisma.$transaction(async (tx) => {
     // Currently redundant since deleting a document will delete the audit logs.
     // However may be useful if we disassociate audit lgos and documents if required.
-      await tx.documentAuditLog.create({
+    await queueJob({
-        data: createDocumentAuditLogData({
+      job: 'create-document-audit-log',
+      args: {
         documentId: id,
         type: DOCUMENT_AUDIT_LOG_TYPE.DOCUMENT_DELETED,
         user,
@@ -73,11 +72,10 @@ export const deleteDocument = async ({
         data: {
           type: 'HARD',
         },
-        }),
+      },
     });
 
-      return await tx.document.delete({ where: { id, status: DocumentStatus.DRAFT } });
+    return await prisma.document.delete({ where: { id, status: DocumentStatus.DRAFT } });
-    });
   }
 
   // if the document is pending, send cancellation emails to all recipients
@@ -93,7 +91,9 @@ export const deleteDocument = async ({
           assetBaseUrl,
         });
 
-        await mailer.sendMail({
+        await queueJob({
+          job: 'send-mail',
+          args: {
             to: {
               address: recipient.email,
               name: recipient.name,
@@ -105,15 +105,16 @@ export const deleteDocument = async ({
             subject: 'Document Cancelled',
             html: render(template),
             text: render(template, { plainText: true }),
+          },
         });
       }),
     );
   }
 
   // If the document is not a draft, only soft-delete.
-  return await prisma.$transaction(async (tx) => {
+  await queueJob({
-    await tx.documentAuditLog.create({
+    job: 'create-document-audit-log',
-      data: createDocumentAuditLogData({
+    args: {
       documentId: id,
       type: DOCUMENT_AUDIT_LOG_TYPE.DOCUMENT_DELETED,
       user,
@@ -121,10 +122,10 @@ export const deleteDocument = async ({
       data: {
         type: 'SOFT',
       },
-      }),
+    },
   });
 
-    return await tx.document.update({
+  return await prisma.document.update({
     where: {
       id,
     },
@@ -132,5 +133,4 @@ export const deleteDocument = async ({
       deletedAt: new Date().toISOString(),
     },
   });
-  });
 };

packages/lib/server-only/document/resend-document.tsx

@@ -1,6 +1,5 @@
 import { createElement } from 'react';
 
-import { mailer } from '@documenso/email/mailer';
 import { render } from '@documenso/email/render';
 import { DocumentInviteEmailTemplate } from '@documenso/email/templates/document-invite';
 import { FROM_ADDRESS, FROM_NAME } from '@documenso/lib/constants/email';
@@ -10,13 +9,13 @@ import {
 } from '@documenso/lib/constants/recipient-roles';
 import { DOCUMENT_AUDIT_LOG_TYPE } from '@documenso/lib/types/document-audit-logs';
 import type { RequestMetadata } from '@documenso/lib/universal/extract-request-metadata';
-import { createDocumentAuditLogData } from '@documenso/lib/utils/document-audit-logs';
 import { renderCustomEmailTemplate } from '@documenso/lib/utils/render-custom-email-template';
 import { prisma } from '@documenso/prisma';
 import { DocumentStatus, RecipientRole, SigningStatus } from '@documenso/prisma/client';
 import type { Prisma } from '@documenso/prisma/client';
 
 import { NEXT_PUBLIC_WEBAPP_URL } from '../../constants/app';
+import { queueJob } from '../queue/job';
 import { getDocumentWhereInput } from './get-document-by-id';
 
 export type ResendDocumentOptions = {
@@ -110,9 +109,9 @@ export const resendDocument = async ({
 
       const { actionVerb } = RECIPIENT_ROLES_DESCRIPTION[recipient.role];
 
-      await prisma.$transaction(
+      await queueJob({
-        async (tx) => {
+        job: 'send-mail',
-          await mailer.sendMail({
+        args: {
           to: {
             address: email,
             name,
@@ -126,10 +125,12 @@ export const resendDocument = async ({
             : `Please ${actionVerb.toLowerCase()} this document`,
           html: render(template),
           text: render(template, { plainText: true }),
+        },
       });
 
-          await tx.documentAuditLog.create({
+      await queueJob({
-            data: createDocumentAuditLogData({
+        job: 'create-document-audit-log',
+        args: {
           type: DOCUMENT_AUDIT_LOG_TYPE.EMAIL_SENT,
           documentId: document.id,
           user,
@@ -142,11 +143,8 @@ export const resendDocument = async ({
             recipientId: recipient.id,
             isResending: true,
           },
-            }),
-          });
         },
-        { timeout: 30_000 },
+      });
-      );
     }),
   );
 };

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

@@ -5,21 +5,20 @@ import path from 'node:path';
 import { PDFDocument } from 'pdf-lib';
 
 import PostHogServerClient from '@documenso/lib/server-only/feature-flags/get-post-hog-server-client';
-import { DOCUMENT_AUDIT_LOG_TYPE } from '@documenso/lib/types/document-audit-logs';
-import { createDocumentAuditLogData } from '@documenso/lib/utils/document-audit-logs';
 import { prisma } from '@documenso/prisma';
 import { DocumentStatus, RecipientRole, SigningStatus } from '@documenso/prisma/client';
 import { WebhookTriggerEvents } from '@documenso/prisma/client';
 import { signPdf } from '@documenso/signing';
 
+import { DOCUMENT_AUDIT_LOG_TYPE } from '../../types/document-audit-logs';
 import type { RequestMetadata } from '../../universal/extract-request-metadata';
 import { getFile } from '../../universal/upload/get-file';
 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;
@@ -126,8 +125,7 @@ export const sealDocument = async ({
     });
   }
 
-  await prisma.$transaction(async (tx) => {
+  await prisma.documentData.update({
-    await tx.documentData.update({
     where: {
       id: documentData.id,
     },
@@ -136,8 +134,9 @@ export const sealDocument = async ({
     },
   });
 
-    await tx.documentAuditLog.create({
+  await queueJob({
-      data: createDocumentAuditLogData({
+    job: 'create-document-audit-log',
+    args: {
       type: DOCUMENT_AUDIT_LOG_TYPE.DOCUMENT_COMPLETED,
       documentId: document.id,
       requestMetadata,
@@ -145,12 +144,14 @@ export const sealDocument = async ({
       data: {
         transactionId: nanoid(),
       },
-      }),
+    },
-    });
   });
 
   if (sendEmail && !isResealing) {
-    await sendCompletedEmail({ documentId, requestMetadata });
+    await queueJob({
+      job: 'send-completed-email',
+      args: { documentId, requestMetadata },
+    });
   }
 
   await triggerWebhook({

packages/lib/server-only/document/send-completed-email.ts

@@ -9,7 +9,7 @@ import { NEXT_PUBLIC_WEBAPP_URL } from '../../constants/app';
 import { DOCUMENT_AUDIT_LOG_TYPE } from '../../types/document-audit-logs';
 import type { RequestMetadata } from '../../universal/extract-request-metadata';
 import { getFile } from '../../universal/upload/get-file';
-import { createDocumentAuditLogData } from '../../utils/document-audit-logs';
+import { queueJob } from '../queue/job';
 
 export interface SendDocumentOptions {
   documentId: number;
@@ -86,8 +86,9 @@ export const sendCompletedEmail = async ({ documentId, requestMetadata }: SendDo
       ],
     });
 
-    await prisma.documentAuditLog.create({
+    await queueJob({
-      data: createDocumentAuditLogData({
+      job: 'create-document-audit-log',
+      args: {
         type: DOCUMENT_AUDIT_LOG_TYPE.EMAIL_SENT,
         documentId: document.id,
         user: null,
@@ -100,7 +101,7 @@ export const sendCompletedEmail = async ({ documentId, requestMetadata }: SendDo
           recipientRole: 'OWNER',
           isResending: false,
         },
-      }),
+      },
     });
   }
 
@@ -136,8 +137,9 @@ export const sendCompletedEmail = async ({ documentId, requestMetadata }: SendDo
         ],
       });
 
-      await prisma.documentAuditLog.create({
+      await queueJob({
-        data: createDocumentAuditLogData({
+        job: 'create-document-audit-log',
+        args: {
           type: DOCUMENT_AUDIT_LOG_TYPE.EMAIL_SENT,
           documentId: document.id,
           user: null,
@@ -150,7 +152,7 @@ export const sendCompletedEmail = async ({ documentId, requestMetadata }: SendDo
             recipientRole: recipient.role,
             isResending: false,
           },
-        }),
+        },
       });
     }),
   );

packages/lib/server-only/document/send-document.tsx

@@ -6,7 +6,6 @@ import { DocumentInviteEmailTemplate } from '@documenso/email/templates/document
 import { FROM_ADDRESS, FROM_NAME } from '@documenso/lib/constants/email';
 import { DOCUMENT_AUDIT_LOG_TYPE } from '@documenso/lib/types/document-audit-logs';
 import type { RequestMetadata } from '@documenso/lib/universal/extract-request-metadata';
-import { createDocumentAuditLogData } from '@documenso/lib/utils/document-audit-logs';
 import { renderCustomEmailTemplate } from '@documenso/lib/utils/render-custom-email-template';
 import { prisma } from '@documenso/prisma';
 import { DocumentStatus, RecipientRole, SendStatus } from '@documenso/prisma/client';
@@ -17,6 +16,7 @@ import {
   RECIPIENT_ROLES_DESCRIPTION,
   RECIPIENT_ROLE_TO_EMAIL_TYPE,
 } from '../../constants/recipient-roles';
+import { queueJob } from '../queue/job';
 import { triggerWebhook } from '../webhooks/trigger/trigger-webhook';
 
 export type SendDocumentOptions = {
@@ -113,8 +113,7 @@ export const sendDocument = async ({
 
       const { actionVerb } = RECIPIENT_ROLES_DESCRIPTION[recipient.role];
 
-      await prisma.$transaction(
+      // TODO: Move this to a seperate queue of it's own
-        async (tx) => {
       await mailer.sendMail({
         to: {
           address: email,
@@ -131,7 +130,7 @@ export const sendDocument = async ({
         text: render(template, { plainText: true }),
       });
 
-          await tx.recipient.update({
+      await prisma.recipient.update({
         where: {
           id: recipient.id,
         },
@@ -140,8 +139,9 @@ export const sendDocument = async ({
         },
       });
 
-          await tx.documentAuditLog.create({
+      await queueJob({
-            data: createDocumentAuditLogData({
+        job: 'create-document-audit-log',
+        args: {
           type: DOCUMENT_AUDIT_LOG_TYPE.EMAIL_SENT,
           documentId: document.id,
           user,
@@ -154,28 +154,25 @@ export const sendDocument = async ({
             recipientId: recipient.id,
             isResending: false,
           },
-            }),
-          });
         },
-        { timeout: 30_000 },
+      });
-      );
     }),
   );
 
-  const updatedDocument = await prisma.$transaction(async (tx) => {
   if (document.status === DocumentStatus.DRAFT) {
-      await tx.documentAuditLog.create({
+    await queueJob({
-        data: createDocumentAuditLogData({
+      job: 'create-document-audit-log',
+      args: {
         type: DOCUMENT_AUDIT_LOG_TYPE.DOCUMENT_SENT,
         documentId: document.id,
         requestMetadata,
         user,
         data: {},
-        }),
+      },
     });
   }
 
-    return await tx.document.update({
+  const updatedDocument = await prisma.document.update({
     where: {
       id: documentId,
     },
@@ -186,7 +183,6 @@ export const sendDocument = async ({
       Recipient: true,
     },
   });
-  });
 
   await triggerWebhook({
     event: WebhookTriggerEvents.DOCUMENT_SENT,

packages/lib/server-only/document/super-delete-document.ts

@@ -2,7 +2,6 @@
 
 import { createElement } from 'react';
 
-import { mailer } from '@documenso/email/mailer';
 import { render } from '@documenso/email/render';
 import DocumentCancelTemplate from '@documenso/email/templates/document-cancel';
 import { prisma } from '@documenso/prisma';
@@ -12,7 +11,7 @@ import { NEXT_PUBLIC_WEBAPP_URL } from '../../constants/app';
 import { FROM_ADDRESS, FROM_NAME } from '../../constants/email';
 import { DOCUMENT_AUDIT_LOG_TYPE } from '../../types/document-audit-logs';
 import type { RequestMetadata } from '../../universal/extract-request-metadata';
-import { createDocumentAuditLogData } from '../../utils/document-audit-logs';
+import { queueJob } from '../queue/job';
 
 export type SuperDeleteDocumentOptions = {
   id: number;
@@ -49,7 +48,9 @@ export const superDeleteDocument = async ({ id, requestMetadata }: SuperDeleteDo
           assetBaseUrl,
         });
 
-        await mailer.sendMail({
+        await queueJob({
+          job: 'send-mail',
+          args: {
             to: {
               address: recipient.email,
               name: recipient.name,
@@ -61,15 +62,15 @@ export const superDeleteDocument = async ({ id, requestMetadata }: SuperDeleteDo
             subject: 'Document Cancelled',
             html: render(template),
             text: render(template, { plainText: true }),
+          },
         });
       }),
     );
   }
 
-  // always hard delete if deleted from admin
+  await queueJob({
-  return await prisma.$transaction(async (tx) => {
+    job: 'create-document-audit-log',
-    await tx.documentAuditLog.create({
+    args: {
-      data: createDocumentAuditLogData({
       documentId: id,
       type: DOCUMENT_AUDIT_LOG_TYPE.DOCUMENT_DELETED,
       user,
@@ -77,9 +78,9 @@ export const superDeleteDocument = async ({ id, requestMetadata }: SuperDeleteDo
       data: {
         type: 'HARD',
       },
-      }),
+    },
   });
 
-    return await tx.document.delete({ where: { id } });
+  // always hard delete if deleted from admin
-  });
+  return await prisma.document.delete({ where: { id } });
 };

packages/lib/server-only/document/update-title.ts

@@ -2,9 +2,10 @@
 
 import { DOCUMENT_AUDIT_LOG_TYPE } from '@documenso/lib/types/document-audit-logs';
 import type { RequestMetadata } from '@documenso/lib/universal/extract-request-metadata';
-import { createDocumentAuditLogData } from '@documenso/lib/utils/document-audit-logs';
 import { prisma } from '@documenso/prisma';
 
+import { queueJob } from '../queue/job';
+
 export type UpdateTitleOptions = {
   userId: number;
   teamId?: number;
@@ -51,11 +52,10 @@ export const updateTitle = async ({
     return document;
   }
 
-  return await prisma.$transaction(async (tx) => {
   // Instead of doing everything in a transaction we can use our knowledge
   // of the current document title to ensure we aren't performing a conflicting
   // update.
-    const updatedDocument = await tx.document.update({
+  const updatedDocument = await prisma.document.update({
     where: {
       id: documentId,
       title: document.title,
@@ -65,8 +65,9 @@ export const updateTitle = async ({
     },
   });
 
-    await tx.documentAuditLog.create({
+  await queueJob({
-      data: createDocumentAuditLogData({
+    job: 'create-document-audit-log',
+    args: {
       type: DOCUMENT_AUDIT_LOG_TYPE.DOCUMENT_TITLE_UPDATED,
       documentId,
       user,
@@ -75,9 +76,8 @@ export const updateTitle = async ({
         from: document.title,
         to: updatedDocument.title,
       },
-      }),
+    },
   });
 
   return updatedDocument;
-  });
 };

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

@@ -1,11 +1,11 @@
 import { DOCUMENT_AUDIT_LOG_TYPE } from '@documenso/lib/types/document-audit-logs';
 import type { RequestMetadata } from '@documenso/lib/universal/extract-request-metadata';
-import { createDocumentAuditLogData } from '@documenso/lib/utils/document-audit-logs';
 import { prisma } from '@documenso/prisma';
 import { ReadStatus } from '@documenso/prisma/client';
 import { WebhookTriggerEvents } from '@documenso/prisma/client';
 
 import type { TDocumentAccessAuthTypes } from '../../types/document-auth';
+import { queueJob } from '../queue/job';
 import { triggerWebhook } from '../webhooks/trigger/trigger-webhook';
 import { getDocumentAndRecipientByToken } from './get-document-by-token';
 
@@ -33,8 +33,7 @@ export const viewedDocument = async ({
 
   const { documentId } = recipient;
 
-  await prisma.$transaction(async (tx) => {
+  await prisma.recipient.update({
-    await tx.recipient.update({
     where: {
       id: recipient.id,
     },
@@ -43,8 +42,9 @@ export const viewedDocument = async ({
     },
   });
 
-    await tx.documentAuditLog.create({
+  await queueJob({
-      data: createDocumentAuditLogData({
+    job: 'create-document-audit-log',
+    args: {
       type: DOCUMENT_AUDIT_LOG_TYPE.DOCUMENT_OPENED,
       documentId,
       user: {
@@ -59,8 +59,7 @@ export const viewedDocument = async ({
         recipientRole: recipient.role,
         accessAuth: recipientAccessAuth || undefined,
       },
-      }),
+    },
-    });
   });
 
   const document = await getDocumentAndRecipientByToken({ token, requireAccessAuth: false });

packages/lib/server-only/field/create-field.ts

@@ -2,7 +2,7 @@ import { prisma } from '@documenso/prisma';
 import type { FieldType, Team } from '@documenso/prisma/client';
 
 import type { RequestMetadata } from '../../universal/extract-request-metadata';
-import { createDocumentAuditLogData } from '../../utils/document-audit-logs';
+import { queueJob } from '../queue/job';
 
 export type CreateFieldOptions = {
   documentId: number;
@@ -103,8 +103,9 @@ export const createField = async ({
     },
   });
 
-  await prisma.documentAuditLog.create({
+  await queueJob({
-    data: createDocumentAuditLogData({
+    job: 'create-document-audit-log',
+    args: {
       type: 'FIELD_CREATED',
       documentId,
       user: {
@@ -119,7 +120,7 @@ export const createField = async ({
         fieldType: field.type,
       },
       requestMetadata,
-    }),
+    },
   });
 
   return field;

packages/lib/server-only/field/delete-field.ts

@@ -2,7 +2,7 @@ import { prisma } from '@documenso/prisma';
 import type { Team } from '@documenso/prisma/client';
 
 import type { RequestMetadata } from '../../universal/extract-request-metadata';
-import { createDocumentAuditLogData } from '../../utils/document-audit-logs';
+import { queueJob } from '../queue/job';
 
 export type DeleteFieldOptions = {
   fieldId: number;
@@ -67,8 +67,9 @@ export const deleteField = async ({
     });
   }
 
-  await prisma.documentAuditLog.create({
+  await queueJob({
-    data: createDocumentAuditLogData({
+    job: 'create-document-audit-log',
+    args: {
       type: 'FIELD_DELETED',
       documentId,
       user: {
@@ -83,7 +84,7 @@ export const deleteField = async ({
         fieldType: field.type,
       },
       requestMetadata,
-    }),
+    },
   });
 
   return field;

packages/lib/server-only/field/remove-signed-field-with-token.ts

@@ -2,10 +2,11 @@
 
 import { DOCUMENT_AUDIT_LOG_TYPE } from '@documenso/lib/types/document-audit-logs';
 import type { RequestMetadata } from '@documenso/lib/universal/extract-request-metadata';
-import { createDocumentAuditLogData } from '@documenso/lib/utils/document-audit-logs';
 import { prisma } from '@documenso/prisma';
 import { DocumentStatus, SigningStatus } from '@documenso/prisma/client';
 
+import { queueJob } from '../queue/job';
+
 export type RemovedSignedFieldWithTokenOptions = {
   token: string;
   fieldId: number;
@@ -65,9 +66,11 @@ export const removeSignedFieldWithToken = async ({
         fieldId: field.id,
       },
     });
+  });
 
-    await tx.documentAuditLog.create({
+  await queueJob({
-      data: createDocumentAuditLogData({
+    job: 'create-document-audit-log',
+    args: {
       type: DOCUMENT_AUDIT_LOG_TYPE.DOCUMENT_FIELD_UNINSERTED,
       documentId: document.id,
       user: {
@@ -79,7 +82,6 @@ export const removeSignedFieldWithToken = async ({
         field: field.type,
         fieldId: field.secondaryId,
       },
-      }),
+    },
-    });
   });
 };

packages/lib/server-only/field/set-fields-for-document.ts

@@ -8,6 +8,8 @@ import { prisma } from '@documenso/prisma';
 import type { Field, FieldType } from '@documenso/prisma/client';
 import { SendStatus, SigningStatus } from '@documenso/prisma/client';
 
+import { queueJob } from '../queue/job';
+
 export interface SetFieldsForDocumentOptions {
   userId: number;
   documentId: number;
@@ -155,8 +157,9 @@ export const setFieldsForDocument = async ({
 
         // Handle field updated audit log.
         if (field._persisted && changes.length > 0) {
-          await tx.documentAuditLog.create({
+          await queueJob({
-            data: createDocumentAuditLogData({
+            job: 'create-document-audit-log',
+            args: {
               type: DOCUMENT_AUDIT_LOG_TYPE.FIELD_UPDATED,
               documentId: documentId,
               user,
@@ -165,14 +168,15 @@ export const setFieldsForDocument = async ({
                 changes,
                 ...baseAuditLog,
               },
-            }),
+            },
           });
         }
 
         // Handle field created audit log.
         if (!field._persisted) {
-          await tx.documentAuditLog.create({
+          await queueJob({
-            data: createDocumentAuditLogData({
+            job: 'create-document-audit-log',
+            args: {
               type: DOCUMENT_AUDIT_LOG_TYPE.FIELD_CREATED,
               documentId: documentId,
               user,
@@ -180,7 +184,7 @@ export const setFieldsForDocument = async ({
               data: {
                 ...baseAuditLog,
               },
-            }),
+            },
           });
         }
 

packages/lib/server-only/field/sign-field-with-token.ts

@@ -12,9 +12,9 @@ import { AppError, AppErrorCode } from '../../errors/app-error';
 import { DOCUMENT_AUDIT_LOG_TYPE } from '../../types/document-audit-logs';
 import type { TRecipientActionAuth } from '../../types/document-auth';
 import type { RequestMetadata } from '../../universal/extract-request-metadata';
-import { createDocumentAuditLogData } from '../../utils/document-audit-logs';
 import { extractDocumentAuthMethods } from '../../utils/document-auth';
 import { isRecipientAuthorized } from '../document/is-recipient-authorized';
+import { queueJob } from '../queue/job';
 
 export type SignFieldWithTokenOptions = {
   token: string;
@@ -168,8 +168,9 @@ export const signFieldWithToken = async ({
       });
     }
 
-    await tx.documentAuditLog.create({
+    await queueJob({
-      data: createDocumentAuditLogData({
+      job: 'create-document-audit-log',
+      args: {
         type: DOCUMENT_AUDIT_LOG_TYPE.DOCUMENT_FIELD_INSERTED,
         documentId: document.id,
         user: {
@@ -199,7 +200,7 @@ export const signFieldWithToken = async ({
               }
             : undefined,
         },
-      }),
+      },
     });
 
     return updatedField;

packages/lib/server-only/field/update-field.ts

@@ -3,7 +3,8 @@ import type { FieldType, Team } from '@documenso/prisma/client';
 
 import { DOCUMENT_AUDIT_LOG_TYPE } from '../../types/document-audit-logs';
 import type { RequestMetadata } from '../../universal/extract-request-metadata';
-import { createDocumentAuditLogData, diffFieldChanges } from '../../utils/document-audit-logs';
+import { diffFieldChanges } from '../../utils/document-audit-logs';
+import { queueJob } from '../queue/job';
 
 export type UpdateFieldOptions = {
   fieldId: number;
@@ -77,8 +78,9 @@ export const updateField = async ({
       },
     });
 
-    await tx.documentAuditLog.create({
+    await queueJob({
-      data: createDocumentAuditLogData({
+      job: 'create-document-audit-log',
+      args: {
         type: DOCUMENT_AUDIT_LOG_TYPE.FIELD_UPDATED,
         documentId,
         user: {
@@ -94,7 +96,7 @@ export const updateField = async ({
           changes: diffFieldChanges(oldField, updatedField),
         },
         requestMetadata,
-      }),
+      },
     });
 
     return updatedField;

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,85 @@
+import type { WorkHandler } from 'pg-boss';
+
+import type { MailOptions } from '@documenso/email/mailer';
+import { mailer } from '@documenso/email/mailer';
+import { prisma } from '@documenso/prisma';
+
+import { initQueue } from '.';
+import type { CreateDocumentAuditLogDataOptions } from '../../utils/document-audit-logs';
+import { createDocumentAuditLogData } from '../../utils/document-audit-logs';
+import {
+  type SendDocumentOptions as SendCompletedDocumentOptions,
+  sendCompletedEmail,
+} from '../document/send-completed-email';
+import { type SendPendingEmailOptions, sendPendingEmail } from '../document/send-pending-email';
+
+type JobOptions = {
+  'send-mail': MailOptions;
+  'send-completed-email': SendCompletedDocumentOptions;
+  'send-pending-email': SendPendingEmailOptions;
+  'create-document-audit-log': CreateDocumentAuditLogDataOptions;
+};
+
+export const jobHandlers: {
+  [K in keyof JobOptions]: WorkHandler<JobOptions[K]>;
+} = {
+  'send-completed-email': async ({ id, name, data: { documentId, requestMetadata } }) => {
+    console.log('Running Queue: ', name, ' ', id);
+
+    await sendCompletedEmail({
+      documentId,
+      requestMetadata,
+    });
+  },
+  'send-pending-email': async ({ id, name, data: { documentId, recipientId } }) => {
+    console.log('Running Queue: ', name, ' ', id);
+
+    await sendPendingEmail({
+      documentId,
+      recipientId,
+    });
+  },
+  'send-mail': async ({ id, name, data: { attachments, to, from, subject, html, text } }) => {
+    console.log('Running Queue: ', name, ' ', id);
+
+    await mailer.sendMail({
+      to,
+      from,
+      subject,
+      html,
+      text,
+      attachments,
+    });
+  },
+
+  // Audit Logs Queue
+  'create-document-audit-log': async ({
+    name,
+    data: { documentId, type, requestMetadata, user, data },
+    id,
+  }) => {
+    console.log('Running Queue: ', name, ' ', id);
+
+    await prisma.documentAuditLog.create({
+      data: createDocumentAuditLogData({
+        type,
+        documentId,
+        requestMetadata,
+        user,
+        data,
+      }),
+    });
+  },
+};
+
+export const queueJob = async ({
+  job,
+  args,
+}: {
+  job: keyof JobOptions;
+  args?: JobOptions[keyof JobOptions];
+}) => {
+  const queue = await initQueue();
+
+  await queue.send(job, args ?? {});
+};

packages/lib/server-only/recipient/delete-recipient.ts

@@ -3,7 +3,7 @@ import type { Team } from '@documenso/prisma/client';
 import { SendStatus } from '@documenso/prisma/client';
 
 import type { RequestMetadata } from '../../universal/extract-request-metadata';
-import { createDocumentAuditLogData } from '../../utils/document-audit-logs';
+import { queueJob } from '../queue/job';
 
 export type DeleteRecipientOptions = {
   documentId: number;
@@ -73,15 +73,15 @@ export const deleteRecipient = async ({
     });
   }
 
-  const deletedRecipient = await prisma.$transaction(async (tx) => {
+  const deletedRecipient = await prisma.recipient.delete({
-    const deleted = await tx.recipient.delete({
     where: {
       id: recipient.id,
     },
   });
 
-    await tx.documentAuditLog.create({
+  await queueJob({
-      data: createDocumentAuditLogData({
+    job: 'create-document-audit-log',
+    args: {
       type: 'RECIPIENT_DELETED',
       documentId,
       user: {
@@ -96,10 +96,7 @@ export const deleteRecipient = async ({
         recipientRole: recipient.role,
       },
       requestMetadata,
-      }),
+    },
-    });
-
-    return deleted;
   });
 
   return deletedRecipient;

packages/lib/server-only/recipient/set-recipients-for-document.ts

@@ -17,6 +17,7 @@ import { RecipientRole } from '@documenso/prisma/client';
 import { SendStatus, SigningStatus } from '@documenso/prisma/client';
 
 import { AppError, AppErrorCode } from '../../errors/app-error';
+import { queueJob } from '../queue/job';
 
 export interface SetRecipientsForDocumentOptions {
   userId: number;
@@ -203,8 +204,9 @@ export const setRecipientsForDocument = async ({
 
         // Handle recipient updated audit log.
         if (recipient._persisted && changes.length > 0) {
-          await tx.documentAuditLog.create({
+          await queueJob({
-            data: createDocumentAuditLogData({
+            job: 'create-document-audit-log',
+            args: {
               type: DOCUMENT_AUDIT_LOG_TYPE.RECIPIENT_UPDATED,
               documentId: documentId,
               user,
@@ -213,14 +215,15 @@ export const setRecipientsForDocument = async ({
                 changes,
                 ...baseAuditLog,
               },
-            }),
+            },
           });
         }
 
         // Handle recipient created audit log.
         if (!recipient._persisted) {
-          await tx.documentAuditLog.create({
+          await queueJob({
-            data: createDocumentAuditLogData({
+            job: 'create-document-audit-log',
+            args: {
               type: DOCUMENT_AUDIT_LOG_TYPE.RECIPIENT_CREATED,
               documentId: documentId,
               user,
@@ -229,7 +232,7 @@ export const setRecipientsForDocument = async ({
                 ...baseAuditLog,
                 actionAuth: recipient.actionAuth || undefined,
               },
-            }),
+            },
           });
         }
 

packages/lib/server-only/recipient/update-recipient.ts

@@ -3,7 +3,8 @@ import type { RecipientRole, Team } from '@documenso/prisma/client';
 
 import { DOCUMENT_AUDIT_LOG_TYPE } from '../../types/document-audit-logs';
 import type { RequestMetadata } from '../../universal/extract-request-metadata';
-import { createDocumentAuditLogData, diffRecipientChanges } from '../../utils/document-audit-logs';
+import { diffRecipientChanges } from '../../utils/document-audit-logs';
+import { queueJob } from '../queue/job';
 
 export type UpdateRecipientOptions = {
   documentId: number;
@@ -75,8 +76,7 @@ export const updateRecipient = async ({
     throw new Error('Recipient not found');
   }
 
-  const updatedRecipient = await prisma.$transaction(async (tx) => {
+  const updatedRecipient = await prisma.recipient.update({
-    const persisted = await prisma.recipient.update({
     where: {
       id: recipient.id,
     },
@@ -87,11 +87,12 @@ export const updateRecipient = async ({
     },
   });
 
-    const changes = diffRecipientChanges(recipient, persisted);
+  const changes = diffRecipientChanges(recipient, updatedRecipient);
 
   if (changes.length > 0) {
-      await tx.documentAuditLog.create({
+    await queueJob({
-        data: createDocumentAuditLogData({
+      job: 'create-document-audit-log',
+      args: {
         type: DOCUMENT_AUDIT_LOG_TYPE.RECIPIENT_UPDATED,
         documentId: documentId,
         user: {
@@ -103,16 +104,15 @@ export const updateRecipient = async ({
         data: {
           changes,
           recipientId,
-            recipientEmail: persisted.email,
+          recipientEmail: updatedRecipient.email,
-            recipientName: persisted.name,
+          recipientName: updatedRecipient.name,
-            recipientRole: persisted.role,
+          recipientRole: updatedRecipient.role,
+        },
       },
-        }),
-      });
-
-      return persisted;
-    }
     });
 
+    return updatedRecipient;
+  }
+
   return updatedRecipient;
 };

packages/lib/server-only/team/create-team-email-verification.ts

@@ -2,7 +2,6 @@ import { createElement } from 'react';
 
 import { z } from 'zod';
 
-import { mailer } from '@documenso/email/mailer';
 import { render } from '@documenso/email/render';
 import { ConfirmTeamEmailTemplate } from '@documenso/email/templates/confirm-team-email';
 import { WEBAPP_BASE_URL } from '@documenso/lib/constants/app';
@@ -13,6 +12,8 @@ import { createTokenVerification } from '@documenso/lib/utils/token-verification
 import { prisma } from '@documenso/prisma';
 import { Prisma } from '@documenso/prisma/client';
 
+import { queueJob } from '../queue/job';
+
 export type CreateTeamEmailVerificationOptions = {
   userId: number;
   teamId: number;
@@ -122,7 +123,9 @@ export const sendTeamEmailVerificationEmail = async (
     token,
   });
 
-  await mailer.sendMail({
+  await queueJob({
+    job: 'send-mail',
+    args: {
       to: email,
       from: {
         name: FROM_NAME,
@@ -131,5 +134,6 @@ export const sendTeamEmailVerificationEmail = async (
       subject: `A request to use your email has been initiated by ${teamName} on Documenso`,
       html: render(template),
       text: render(template, { plainText: true }),
+    },
   });
 };

packages/lib/server-only/team/create-team-member-invites.ts

@@ -2,7 +2,6 @@ import { createElement } from 'react';
 
 import { nanoid } from 'nanoid';
 
-import { mailer } from '@documenso/email/mailer';
 import { render } from '@documenso/email/render';
 import type { TeamInviteEmailProps } from '@documenso/email/templates/team-invite';
 import { TeamInviteEmailTemplate } from '@documenso/email/templates/team-invite';
@@ -15,6 +14,8 @@ import { prisma } from '@documenso/prisma';
 import { TeamMemberInviteStatus } from '@documenso/prisma/client';
 import type { TCreateTeamMemberInvitesMutationSchema } from '@documenso/trpc/server/team-router/schema';
 
+import { queueJob } from '../queue/job';
+
 export type CreateTeamMemberInvitesOptions = {
   userId: number;
   userName: string;
@@ -148,7 +149,9 @@ export const sendTeamMemberInviteEmail = async ({
     ...emailTemplateOptions,
   });
 
-  await mailer.sendMail({
+  await queueJob({
+    job: 'send-mail',
+    args: {
       to: email,
       from: {
         name: FROM_NAME,
@@ -157,5 +160,6 @@ export const sendTeamMemberInviteEmail = async ({
       subject: `You have been invited to join ${emailTemplateOptions.teamName} on Documenso`,
       html: render(template),
       text: render(template, { plainText: true }),
+    },
   });
 };

packages/lib/server-only/team/delete-team-email.ts

@@ -1,6 +1,5 @@
 import { createElement } from 'react';
 
-import { mailer } from '@documenso/email/mailer';
 import { render } from '@documenso/email/render';
 import { TeamEmailRemovedTemplate } from '@documenso/email/templates/team-email-removed';
 import { WEBAPP_BASE_URL } from '@documenso/lib/constants/app';
@@ -8,6 +7,8 @@ import { FROM_ADDRESS, FROM_NAME } from '@documenso/lib/constants/email';
 import { TEAM_MEMBER_ROLE_PERMISSIONS_MAP } from '@documenso/lib/constants/teams';
 import { prisma } from '@documenso/prisma';
 
+import { queueJob } from '../queue/job';
+
 export type DeleteTeamEmailOptions = {
   userId: number;
   userEmail: string;
@@ -73,7 +74,9 @@ export const deleteTeamEmail = async ({ userId, userEmail, teamId }: DeleteTeamE
       teamUrl: team.url,
     });
 
-    await mailer.sendMail({
+    await queueJob({
+      job: 'create-document-audit-log',
+      args: {
         to: {
           address: team.owner.email,
           name: team.owner.name ?? '',
@@ -85,6 +88,7 @@ export const deleteTeamEmail = async ({ userId, userEmail, teamId }: DeleteTeamE
         subject: `Team email has been revoked for ${team.name}`,
         html: render(template),
         text: render(template, { plainText: true }),
+      },
     });
   } catch (e) {
     // Todo: Teams - Alert us.

packages/lib/server-only/team/request-team-ownership-transfer.ts

@@ -1,6 +1,5 @@
 import { createElement } from 'react';
 
-import { mailer } from '@documenso/email/mailer';
 import { render } from '@documenso/email/render';
 import { TeamTransferRequestTemplate } from '@documenso/email/templates/team-transfer-request';
 import { WEBAPP_BASE_URL } from '@documenso/lib/constants/app';
@@ -8,6 +7,8 @@ import { FROM_ADDRESS, FROM_NAME } from '@documenso/lib/constants/email';
 import { createTokenVerification } from '@documenso/lib/utils/token-verification';
 import { prisma } from '@documenso/prisma';
 
+import { queueJob } from '../queue/job';
+
 export type RequestTeamOwnershipTransferOptions = {
   /**
    * The ID of the user initiating the transfer.
@@ -93,7 +94,9 @@ export const requestTeamOwnershipTransfer = async ({
         token,
       });
 
-      await mailer.sendMail({
+      await queueJob({
+        job: 'create-document-audit-log',
+        args: {
           to: newOwnerUser.email,
           from: {
             name: FROM_NAME,
@@ -102,6 +105,7 @@ export const requestTeamOwnershipTransfer = async ({
           subject: `You have been requested to take ownership of team ${team.name} on Documenso`,
           html: render(template),
           text: render(template, { plainText: true }),
+        },
       });
     },
     { timeout: 30_000 },

packages/lib/utils/document-audit-logs.ts

@@ -25,7 +25,7 @@ import {
 import { ZRecipientAuthOptionsSchema } from '../types/document-auth';
 import type { RequestMetadata } from '../universal/extract-request-metadata';
 
-type CreateDocumentAuditLogDataOptions<T = TDocumentAuditLog['type']> = {
+export type CreateDocumentAuditLogDataOptions<T = TDocumentAuditLog['type']> = {
   documentId: number;
   type: T;
   data: Extract<TDocumentAuditLog, { type: T }>['data'];

packages/trpc/server/singleplayer-router/router.ts

@@ -2,12 +2,12 @@ import { createElement } from 'react';
 
 import { PDFDocument } from 'pdf-lib';
 
-import { mailer } from '@documenso/email/mailer';
 import { renderAsync } from '@documenso/email/render';
 import { DocumentSelfSignedEmailTemplate } from '@documenso/email/templates/document-self-signed';
 import { NEXT_PUBLIC_WEBAPP_URL } from '@documenso/lib/constants/app';
 import { FROM_ADDRESS, FROM_NAME, SERVICE_USER_EMAIL } from '@documenso/lib/constants/email';
 import { insertFieldInPDF } from '@documenso/lib/server-only/pdf/insert-field-in-pdf';
+import { queueJob } from '@documenso/lib/server-only/queue/job';
 import { alphaid } from '@documenso/lib/universal/id';
 import { getFile } from '@documenso/lib/universal/upload/get-file';
 import { putFile } from '@documenso/lib/universal/upload/put-file';
@@ -160,7 +160,9 @@ export const singleplayerRouter = router({
         ]);
 
         // Send email to signer.
-        await mailer.sendMail({
+        await queueJob({
+          job: 'send-mail',
+          args: {
             to: {
               address: signer.email,
               name: signer.name,
@@ -173,6 +175,7 @@ export const singleplayerRouter = router({
             html,
             text,
             attachments: [{ content: signedPdfBuffer, filename: documentName }],
+          },
         });
 
         return token;

turbo.json

@@ -93,6 +93,7 @@
     "NEXT_PRIVATE_STRIPE_API_KEY",
     "NEXT_PRIVATE_STRIPE_WEBHOOK_SECRET",
     "NEXT_PRIVATE_GITHUB_TOKEN",
+    "NEXT_RUNTIME",
     "CI",
     "VERCEL",
     "VERCEL_ENV",