Merge branch 'main' into feat/public-profiles

.env.example

@@ -103,6 +103,12 @@ NEXT_PRIVATE_STRIPE_WEBHOOK_SECRET=
 NEXT_PUBLIC_STRIPE_COMMUNITY_PLAN_MONTHLY_PRICE_ID=
 NEXT_PUBLIC_STRIPE_ENTERPRISE_PLAN_MONTHLY_PRICE_ID=
 
+# [[BACKGROUND JOBS]]
+NEXT_PRIVATE_JOBS_PROVIDER="local"
+NEXT_PRIVATE_TRIGGER_API_KEY=
+NEXT_PRIVATE_TRIGGER_API_URL=
+NEXT_PRIVATE_INNGEST_EVENT_KEY=
+
 # [[FEATURES]]
 # OPTIONAL: Leave blank to disable PostHog and feature flags.
 NEXT_PUBLIC_POSTHOG_KEY=""

.vscode/settings.json

@@ -5,11 +5,19 @@
   "editor.codeActionsOnSave": {
     "source.fixAll": "explicit"
   },
-  "eslint.validate": ["typescript", "typescriptreact", "javascript", "javascriptreact"],
+  "eslint.validate": [
+    "typescript",
+    "typescriptreact",
+    "javascript",
+    "javascriptreact"
+  ],
   "javascript.preferences.importModuleSpecifier": "non-relative",
   "javascript.preferences.useAliasesForRenames": false,
   "typescript.enablePromptUseWorkspaceTsdk": true,
   "files.eol": "\n",
   "editor.tabSize": 2,
-  "editor.insertSpaces": true
+  "editor.insertSpaces": true,
+  "[prisma]": {
+    "editor.defaultFormatter": "Prisma.prisma"
+  },
 }

apps/web/src/pages/api/jobs/[[...handler]].ts

@@ -0,0 +1,10 @@
+import { jobsClient } from '@documenso/lib/jobs/client';
+
+export const config = {
+  maxDuration: 300,
+  api: {
+    bodyParser: false,
+  },
+};
+
+export default jobsClient.getApiHandler();

docker/development/compose.yml

@@ -4,6 +4,13 @@ services:
   database:
     image: postgres:15
     container_name: database
+    volumes:
+      - documenso_database:/var/lib/postgresql/data
+    healthcheck:
+      test: ['CMD-SHELL', 'pg_isready -U ${POSTGRES_USER}']
+      interval: 10s
+      timeout: 5s
+      retries: 5
     environment:
       - POSTGRES_USER=documenso
       - POSTGRES_PASSWORD=password
@@ -33,5 +40,43 @@ services:
     entrypoint: sh
     command: -c 'mkdir -p /data/documenso && minio server /data --console-address ":9001" --address ":9002"'
 
+  triggerdotdev:
+    image: ghcr.io/triggerdotdev/trigger.dev:latest
+    container_name: triggerdotdev
+    environment:
+      - LOGIN_ORIGIN=http://localhost:3030
+      - APP_ORIGIN=http://localhost:3030
+      - PORT=3030
+      - REMIX_APP_PORT=3030
+      - MAGIC_LINK_SECRET=secret
+      - SESSION_SECRET=secret
+      - ENCRYPTION_KEY=deadbeefcafefeed
+      - DATABASE_URL=postgresql://trigger:password@triggerdotdev_database:5432/trigger
+      - DIRECT_URL=postgresql://trigger:password@triggerdotdev_database:5432/trigger
+      - RUNTIME_PLATFORM=docker-compose
+    ports:
+      - 3030:3030
+    depends_on:
+      - triggerdotdev_database
+
+  triggerdotdev_database:
+    container_name: triggerdotdev_database
+    image: postgres:15
+    volumes:
+      - triggerdotdev_database:/var/lib/postgresql/data
+    healthcheck:
+      test: ['CMD-SHELL', 'pg_isready -U ${POSTGRES_USER}']
+      interval: 10s
+      timeout: 5s
+      retries: 5
+    environment:
+      - POSTGRES_USER=trigger
+      - POSTGRES_PASSWORD=password
+      - POSTGRES_DB=trigger
+    ports:
+      - 54321:5432
+
 volumes:
   minio:
+  documenso_database:
+  triggerdotdev_database:

package.json

@@ -24,7 +24,9 @@
     "prisma:studio": "npm run with:env -- npm run prisma:studio -w @documenso/prisma",
     "with:env": "dotenv -e .env -e .env.local --",
     "reset:hard": "npm run clean && npm i && npm run prisma:generate",
-    "precommit": "npm install && git add package.json package-lock.json"
+    "precommit": "npm install && git add package.json package-lock.json",
+    "trigger:dev": "npm run with:env -- npx trigger-cli dev --handler-path=\"/api/jobs\"",
+    "inngest:dev": "inngest dev -u http://localhost:3000/api/jobs"
   },
   "packageManager": "npm@10.7.0",
   "engines": {
@@ -34,6 +36,7 @@
   "devDependencies": {
     "@commitlint/cli": "^17.7.1",
     "@commitlint/config-conventional": "^17.7.0",
+    "@trigger.dev/cli": "^2.3.18",
     "dotenv": "^16.3.1",
     "dotenv-cli": "^7.3.0",
     "eslint": "^8.40.0",
@@ -52,7 +55,9 @@
   ],
   "dependencies": {
     "@documenso/pdf-sign": "^0.1.0",
-    "next-runtime-env": "^3.2.0"
+    "inngest-cli": "^0.29.1",
+    "next-runtime-env": "^3.2.0",
+    "react": "18.2.0"
   },
   "overrides": {
     "next-auth": {
@@ -60,6 +65,10 @@
     },
     "next-contentlayer": {
       "next": "14.0.3"
-    }
+    },
+    "react": "18.2.0"
+  },
+  "trigger.dev": {
+    "endpointId": "documenso-app"
   }
 }

packages/lib/jobs/client.ts

@@ -0,0 +1,12 @@
+import { JobClient } from './client/client';
+import { SEND_CONFIRMATION_EMAIL_JOB_DEFINITION } from './definitions/send-confirmation-email';
+import { SEND_SIGNING_EMAIL_JOB_DEFINITION } from './definitions/send-signing-email';
+
+/**
+ * The `as const` assertion is load bearing as it provides the correct level of type inference for
+ * triggering jobs.
+ */
+export const jobsClient = new JobClient([
+  SEND_SIGNING_EMAIL_JOB_DEFINITION,
+  SEND_CONFIRMATION_EMAIL_JOB_DEFINITION,
+] as const);

packages/lib/jobs/client/_internal/job.ts

@@ -0,0 +1,61 @@
+import { z } from 'zod';
+
+import type { Json } from './json';
+
+export type SimpleTriggerJobOptions = {
+  id?: string;
+  name: string;
+  payload: unknown;
+  timestamp?: number;
+};
+
+export const ZSimpleTriggerJobOptionsSchema = z.object({
+  id: z.string().optional(),
+  name: z.string(),
+  payload: z.unknown().refine((x) => x !== undefined, { message: 'payload is required' }),
+  timestamp: z.number().optional(),
+});
+
+// Map the array to create a union of objects we may accept
+export type TriggerJobOptions<Definitions extends ReadonlyArray<JobDefinition> = []> = {
+  [K in keyof Definitions]: {
+    id?: string;
+    name: Definitions[K]['trigger']['name'];
+    payload: Definitions[K]['trigger']['schema'] extends z.ZodType<infer Shape> ? Shape : unknown;
+    timestamp?: number;
+  };
+}[number];
+
+// eslint-disable-next-line @typescript-eslint/no-explicit-any
+export type JobDefinition<Name extends string = string, Schema = any> = {
+  id: string;
+  name: string;
+  version: string;
+  enabled?: boolean;
+  trigger: {
+    name: Name;
+    schema?: z.ZodType<Schema>;
+  };
+  handler: (options: { payload: Schema; io: JobRunIO }) => Promise<Json | void>;
+};
+
+export interface JobRunIO {
+  // stableRun<T extends Json | void>(cacheKey: string, callback: (io: JobRunIO) => T | Promise<T>): Promise<T>;
+  runTask<T extends Json | void | undefined>(
+    cacheKey: string,
+    callback: () => Promise<T>,
+  ): Promise<T>;
+  triggerJob(cacheKey: string, options: SimpleTriggerJobOptions): Promise<unknown>;
+  wait(cacheKey: string, ms: number): Promise<void>;
+  logger: {
+    info(...args: unknown[]): void;
+    error(...args: unknown[]): void;
+    debug(...args: unknown[]): void;
+    warn(...args: unknown[]): void;
+    log(...args: unknown[]): void;
+  };
+}
+
+export const defineJob = <N extends string, T = unknown>(
+  job: JobDefinition<N, T>,
+): JobDefinition<N, T> => job;

packages/lib/jobs/client/_internal/json.ts

@@ -0,0 +1,14 @@
+/**
+ * Below type is borrowed from Trigger.dev's SDK, it may be moved elsewhere later.
+ */
+
+export type JsonPrimitive = string | number | boolean | null | undefined | Date | symbol;
+
+export type JsonArray = Json[];
+
+export type JsonRecord<T> = {
+  [Property in keyof T]: Json;
+};
+
+// eslint-disable-next-line @typescript-eslint/no-explicit-any
+export type Json<T = any> = JsonPrimitive | JsonArray | JsonRecord<T>;

packages/lib/jobs/client/base.ts

@@ -0,0 +1,19 @@
+import type { NextApiRequest, NextApiResponse } from 'next';
+
+import type { JobDefinition, SimpleTriggerJobOptions } from './_internal/job';
+
+export abstract class BaseJobProvider {
+  // eslint-disable-next-line @typescript-eslint/require-await
+  public async triggerJob(_options: SimpleTriggerJobOptions): Promise<void> {
+    throw new Error('Not implemented');
+  }
+
+  // eslint-disable-next-line @typescript-eslint/require-await
+  public defineJob<N extends string, T>(_job: JobDefinition<N, T>): void {
+    throw new Error('Not implemented');
+  }
+
+  public getApiHandler(): (req: NextApiRequest, res: NextApiResponse) => Promise<Response | void> {
+    throw new Error('Not implemented');
+  }
+}

packages/lib/jobs/client/client.ts

@@ -0,0 +1,30 @@
+import { match } from 'ts-pattern';
+
+import type { JobDefinition, TriggerJobOptions } from './_internal/job';
+import type { BaseJobProvider as JobClientProvider } from './base';
+import { InngestJobProvider } from './inngest';
+import { LocalJobProvider } from './local';
+import { TriggerJobProvider } from './trigger';
+
+export class JobClient<T extends ReadonlyArray<JobDefinition> = []> {
+  private _provider: JobClientProvider;
+
+  public constructor(definitions: T) {
+    this._provider = match(process.env.NEXT_PRIVATE_JOBS_PROVIDER)
+      .with('inngest', () => InngestJobProvider.getInstance())
+      .with('trigger', () => TriggerJobProvider.getInstance())
+      .otherwise(() => LocalJobProvider.getInstance());
+
+    definitions.forEach((definition) => {
+      this._provider.defineJob(definition);
+    });
+  }
+
+  public async triggerJob(options: TriggerJobOptions<T>) {
+    return this._provider.triggerJob(options);
+  }
+
+  public getApiHandler() {
+    return this._provider.getApiHandler();
+  }
+}

packages/lib/jobs/client/inngest.ts

@@ -0,0 +1,120 @@
+import type { NextApiRequest, NextApiResponse } from 'next';
+import type { NextRequest } from 'next/server';
+
+import type { Context, Handler, InngestFunction } from 'inngest';
+import { Inngest as InngestClient } from 'inngest';
+import type { Logger } from 'inngest/middleware/logger';
+import { serve as createPagesRoute } from 'inngest/next';
+import { json } from 'micro';
+
+import type { JobDefinition, JobRunIO, SimpleTriggerJobOptions } from './_internal/job';
+import { BaseJobProvider } from './base';
+
+export class InngestJobProvider extends BaseJobProvider {
+  private static _instance: InngestJobProvider;
+
+  private _client: InngestClient;
+  private _functions: Array<InngestFunction<InngestFunction.Options, Handler.Any, Handler.Any>> =
+    [];
+
+  private constructor(options: { client: InngestClient }) {
+    super();
+
+    this._client = options.client;
+  }
+
+  static getInstance() {
+    if (!this._instance) {
+      const client = new InngestClient({
+        id: 'documenso-app',
+        eventKey: process.env.NEXT_PRIVATE_INNGEST_EVENT_KEY,
+      });
+
+      this._instance = new InngestJobProvider({ client });
+    }
+
+    return this._instance;
+  }
+
+  public defineJob<N extends string, T>(job: JobDefinition<N, T>): void {
+    console.log('defining job', job.id);
+    const fn = this._client.createFunction(
+      {
+        id: job.id,
+        name: job.name,
+      },
+      {
+        event: job.trigger.name,
+      },
+      async (ctx) => {
+        const io = this.convertInngestIoToJobRunIo(ctx);
+
+        // We need to cast to any so we can deal with parsing later.
+        // eslint-disable-next-line @typescript-eslint/consistent-type-assertions, @typescript-eslint/no-explicit-any
+        let payload = ctx.event.data as any;
+
+        if (job.trigger.schema) {
+          payload = job.trigger.schema.parse(payload);
+        }
+
+        await job.handler({ payload, io });
+      },
+    );
+
+    this._functions.push(fn);
+  }
+
+  public async triggerJob(options: SimpleTriggerJobOptions): Promise<void> {
+    await this._client.send({
+      id: options.id,
+      name: options.name,
+      data: options.payload,
+      ts: options.timestamp,
+    });
+  }
+
+  public getApiHandler() {
+    const handler = createPagesRoute({
+      client: this._client,
+      functions: this._functions,
+    });
+
+    return async (req: NextApiRequest, res: NextApiResponse) => {
+      // Since body-parser is disabled for this route we need to patch in the parsed body
+      if (req.headers['content-type'] === 'application/json') {
+        Object.assign(req, {
+          body: await json(req),
+        });
+      }
+
+      // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
+      const nextReq = req as unknown as NextRequest;
+
+      return await handler(nextReq, res);
+    };
+  }
+
+  private convertInngestIoToJobRunIo(ctx: Context.Any & { logger: Logger }) {
+    const { step } = ctx;
+
+    return {
+      wait: step.sleep,
+      logger: {
+        ...ctx.logger,
+        log: ctx.logger.info,
+      },
+      runTask: async (cacheKey, callback) => {
+        const result = await step.run(cacheKey, callback);
+
+        // !: Not dealing with this right now but it should be correct.
+        // eslint-disable-next-line @typescript-eslint/consistent-type-assertions, @typescript-eslint/no-explicit-any
+        return result as any;
+      },
+      triggerJob: async (cacheKey, payload) =>
+        step.sendEvent(cacheKey, {
+          ...payload,
+          timestamp: payload.timestamp,
+        }),
+    } satisfies JobRunIO;
+  }
+}

packages/lib/jobs/client/local.ts

@@ -0,0 +1,351 @@
+import type { NextApiRequest, NextApiResponse } from 'next';
+
+import { sha256 } from '@noble/hashes/sha256';
+import { json } from 'micro';
+
+import { prisma } from '@documenso/prisma';
+import { BackgroundJobStatus, Prisma } from '@documenso/prisma/client';
+
+import { NEXT_PUBLIC_WEBAPP_URL } from '../../constants/app';
+import { sign } from '../../server-only/crypto/sign';
+import { verify } from '../../server-only/crypto/verify';
+import {
+  type JobDefinition,
+  type JobRunIO,
+  type SimpleTriggerJobOptions,
+  ZSimpleTriggerJobOptionsSchema,
+} from './_internal/job';
+import type { Json } from './_internal/json';
+import { BaseJobProvider } from './base';
+
+export class LocalJobProvider extends BaseJobProvider {
+  private static _instance: LocalJobProvider;
+
+  private _jobDefinitions: Record<string, JobDefinition> = {};
+
+  private constructor() {
+    super();
+  }
+
+  static getInstance() {
+    if (!this._instance) {
+      this._instance = new LocalJobProvider();
+    }
+
+    return this._instance;
+  }
+
+  public defineJob<N extends string, T>(definition: JobDefinition<N, T>) {
+    this._jobDefinitions[definition.id] = {
+      ...definition,
+      enabled: definition.enabled ?? true,
+    };
+  }
+
+  public async triggerJob(options: SimpleTriggerJobOptions) {
+    console.log({ jobDefinitions: this._jobDefinitions });
+
+    const eligibleJobs = Object.values(this._jobDefinitions).filter(
+      (job) => job.trigger.name === options.name,
+    );
+
+    console.log({ options });
+    console.log(
+      'Eligible jobs:',
+      eligibleJobs.map((job) => job.name),
+    );
+
+    await Promise.all(
+      eligibleJobs.map(async (job) => {
+        // Ideally we will change this to a createMany with returning later once we upgrade Prisma
+        // @see: https://github.com/prisma/prisma/releases/tag/5.14.0
+        const pendingJob = await prisma.backgroundJob.create({
+          data: {
+            jobId: job.id,
+            name: job.name,
+            version: job.version,
+            // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
+            payload: options.payload as Prisma.InputJsonValue,
+          },
+        });
+
+        await this.submitJobToEndpoint({
+          jobId: pendingJob.id,
+          jobDefinitionId: pendingJob.jobId,
+          data: options,
+        });
+      }),
+    );
+  }
+
+  public getApiHandler() {
+    return async (req: NextApiRequest, res: NextApiResponse) => {
+      if (req.method !== 'POST') {
+        res.status(405).send('Method not allowed');
+      }
+
+      const jobId = req.headers['x-job-id'];
+      const signature = req.headers['x-job-signature'];
+      const isRetry = req.headers['x-job-retry'] !== undefined;
+
+      // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
+      const options = await json(req)
+        .then(async (data) => ZSimpleTriggerJobOptionsSchema.parseAsync(data))
+        // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
+        .then((data) => data as SimpleTriggerJobOptions)
+        .catch(() => null);
+
+      if (!options) {
+        res.status(400).send('Bad request');
+        return;
+      }
+
+      const definition = this._jobDefinitions[options.name];
+
+      if (
+        typeof jobId !== 'string' ||
+        typeof signature !== 'string' ||
+        typeof options !== 'object'
+      ) {
+        res.status(400).send('Bad request');
+        return;
+      }
+
+      if (!definition) {
+        res.status(404).send('Job not found');
+        return;
+      }
+
+      if (definition && !definition.enabled) {
+        console.log('Attempted to trigger a disabled job', options.name);
+
+        res.status(404).send('Job not found');
+        return;
+      }
+
+      if (!signature || !verify(options, signature)) {
+        res.status(401).send('Unauthorized');
+        return;
+      }
+
+      if (definition.trigger.schema) {
+        const result = definition.trigger.schema.safeParse(options.payload);
+
+        if (!result.success) {
+          res.status(400).send('Bad request');
+          return;
+        }
+      }
+
+      console.log(`[JOBS]: Triggering job ${options.name} with payload`, options.payload);
+
+      let backgroundJob = await prisma.backgroundJob
+        .update({
+          where: {
+            id: jobId,
+            status: BackgroundJobStatus.PENDING,
+          },
+          data: {
+            status: BackgroundJobStatus.PROCESSING,
+            retried: {
+              increment: isRetry ? 1 : 0,
+            },
+            lastRetriedAt: isRetry ? new Date() : undefined,
+          },
+        })
+        .catch(() => null);
+
+      if (!backgroundJob) {
+        res.status(404).send('Job not found');
+        return;
+      }
+
+      try {
+        await definition.handler({
+          payload: options.payload,
+          io: this.createJobRunIO(jobId),
+        });
+
+        backgroundJob = await prisma.backgroundJob.update({
+          where: {
+            id: jobId,
+            status: BackgroundJobStatus.PROCESSING,
+          },
+          data: {
+            status: BackgroundJobStatus.COMPLETED,
+            completedAt: new Date(),
+          },
+        });
+      } catch (error) {
+        console.error(`[JOBS]: Job ${options.name} failed`, error);
+
+        const taskHasExceededRetries = error instanceof BackgroundTaskExceededRetriesError;
+        const jobHasExceededRetries =
+          backgroundJob.retried >= backgroundJob.maxRetries &&
+          !(error instanceof BackgroundTaskFailedError);
+
+        if (taskHasExceededRetries || jobHasExceededRetries) {
+          backgroundJob = await prisma.backgroundJob.update({
+            where: {
+              id: jobId,
+              status: BackgroundJobStatus.PROCESSING,
+            },
+            data: {
+              status: BackgroundJobStatus.FAILED,
+              completedAt: new Date(),
+            },
+          });
+
+          res.status(500).send('Task exceeded retries');
+          return;
+        }
+
+        backgroundJob = await prisma.backgroundJob.update({
+          where: {
+            id: jobId,
+            status: BackgroundJobStatus.PROCESSING,
+          },
+          data: {
+            status: BackgroundJobStatus.PENDING,
+          },
+        });
+
+        await this.submitJobToEndpoint({
+          jobId,
+          jobDefinitionId: backgroundJob.jobId,
+          data: options,
+        });
+      }
+
+      res.status(200).send('OK');
+    };
+  }
+
+  private async submitJobToEndpoint(options: {
+    jobId: string;
+    jobDefinitionId: string;
+    data: SimpleTriggerJobOptions;
+    isRetry?: boolean;
+  }) {
+    const { jobId, jobDefinitionId, data, isRetry } = options;
+
+    const endpoint = `${NEXT_PUBLIC_WEBAPP_URL()}/api/jobs/${jobDefinitionId}/${jobId}`;
+    const signature = sign(data);
+
+    const headers: Record<string, string> = {
+      'Content-Type': 'application/json',
+      'X-Job-Id': jobId,
+      'X-Job-Signature': signature,
+    };
+
+    if (isRetry) {
+      headers['X-Job-Retry'] = '1';
+    }
+
+    console.log('Submitting job to endpoint:', endpoint);
+    await Promise.race([
+      fetch(endpoint, {
+        method: 'POST',
+        body: JSON.stringify(data),
+        headers,
+      }).catch(() => null),
+      new Promise((resolve) => {
+        setTimeout(resolve, 150);
+      }),
+    ]);
+  }
+
+  private createJobRunIO(jobId: string): JobRunIO {
+    return {
+      runTask: async <T extends void | Json>(cacheKey: string, callback: () => Promise<T>) => {
+        const hashedKey = Buffer.from(sha256(cacheKey)).toString('hex');
+
+        let task = await prisma.backgroundJobTask.findFirst({
+          where: {
+            id: `task-${hashedKey}--${jobId}`,
+            jobId,
+          },
+        });
+
+        if (!task) {
+          task = await prisma.backgroundJobTask.create({
+            data: {
+              id: `task-${hashedKey}--${jobId}`,
+              name: cacheKey,
+              jobId,
+              status: BackgroundJobStatus.PENDING,
+            },
+          });
+        }
+
+        if (task.status === BackgroundJobStatus.COMPLETED) {
+          // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
+          return task.result as T;
+        }
+
+        if (task.retried >= 3) {
+          throw new BackgroundTaskExceededRetriesError('Task exceeded retries');
+        }
+
+        try {
+          const result = await callback();
+
+          task = await prisma.backgroundJobTask.update({
+            where: {
+              id: task.id,
+              jobId,
+            },
+            data: {
+              status: BackgroundJobStatus.COMPLETED,
+              result: result === null ? Prisma.JsonNull : result,
+              completedAt: new Date(),
+            },
+          });
+
+          return result;
+        } catch {
+          task = await prisma.backgroundJobTask.update({
+            where: {
+              id: task.id,
+              jobId,
+            },
+            data: {
+              status: BackgroundJobStatus.PENDING,
+              retried: {
+                increment: 1,
+              },
+            },
+          });
+
+          throw new BackgroundTaskFailedError('Task failed');
+        }
+      },
+      triggerJob: async (_cacheKey, payload) => await this.triggerJob(payload),
+      logger: {
+        debug: (...args) => console.debug(`[${jobId}]`, ...args),
+        error: (...args) => console.error(`[${jobId}]`, ...args),
+        info: (...args) => console.info(`[${jobId}]`, ...args),
+        log: (...args) => console.log(`[${jobId}]`, ...args),
+        warn: (...args) => console.warn(`[${jobId}]`, ...args),
+      },
+      // eslint-disable-next-line @typescript-eslint/require-await
+      wait: async () => {
+        throw new Error('Not implemented');
+      },
+    };
+  }
+}
+
+class BackgroundTaskFailedError extends Error {
+  constructor(message: string) {
+    super(message);
+    this.name = 'BackgroundTaskFailedError';
+  }
+}
+
+class BackgroundTaskExceededRetriesError extends Error {
+  constructor(message: string) {
+    super(message);
+    this.name = 'BackgroundTaskExceededRetriesError';
+  }
+}

packages/lib/jobs/client/trigger.ts

@@ -0,0 +1,73 @@
+import { createPagesRoute } from '@trigger.dev/nextjs';
+import type { IO } from '@trigger.dev/sdk';
+import { TriggerClient, eventTrigger } from '@trigger.dev/sdk';
+
+import type { JobDefinition, JobRunIO, SimpleTriggerJobOptions } from './_internal/job';
+import { BaseJobProvider } from './base';
+
+export class TriggerJobProvider extends BaseJobProvider {
+  private static _instance: TriggerJobProvider;
+
+  private _client: TriggerClient;
+
+  private constructor(options: { client: TriggerClient }) {
+    super();
+
+    this._client = options.client;
+  }
+
+  static getInstance() {
+    if (!this._instance) {
+      const client = new TriggerClient({
+        id: 'documenso-app',
+        apiKey: process.env.NEXT_PRIVATE_TRIGGER_API_KEY,
+        apiUrl: process.env.NEXT_PRIVATE_TRIGGER_API_URL,
+      });
+
+      this._instance = new TriggerJobProvider({ client });
+    }
+
+    return this._instance;
+  }
+
+  public defineJob<N extends string, T>(job: JobDefinition<N, T>): void {
+    this._client.defineJob({
+      id: job.id,
+      name: job.name,
+      version: job.version,
+      trigger: eventTrigger({
+        name: job.trigger.name,
+        schema: job.trigger.schema,
+      }),
+      run: async (payload, io) => job.handler({ payload, io: this.convertTriggerIoToJobRunIo(io) }),
+    });
+  }
+
+  public async triggerJob(options: SimpleTriggerJobOptions): Promise<void> {
+    await this._client.sendEvent({
+      id: options.id,
+      name: options.name,
+      payload: options.payload,
+      timestamp: options.timestamp ? new Date(options.timestamp) : undefined,
+    });
+  }
+
+  public getApiHandler() {
+    const { handler } = createPagesRoute(this._client);
+
+    return handler;
+  }
+
+  private convertTriggerIoToJobRunIo(io: IO) {
+    return {
+      wait: io.wait,
+      logger: io.logger,
+      runTask: async (cacheKey, callback) => io.runTask(cacheKey, callback),
+      triggerJob: async (cacheKey, payload) =>
+        io.sendEvent(cacheKey, {
+          ...payload,
+          timestamp: payload.timestamp ? new Date(payload.timestamp) : undefined,
+        }),
+    } satisfies JobRunIO;
+  }
+}

packages/lib/jobs/definitions/send-confirmation-email.ts

@@ -0,0 +1,30 @@
+import { z } from 'zod';
+
+import { sendConfirmationToken } from '../../server-only/user/send-confirmation-token';
+import type { JobDefinition } from '../client/_internal/job';
+
+const SEND_CONFIRMATION_EMAIL_JOB_DEFINITION_ID = 'send.signup.confirmation.email';
+
+const SEND_CONFIRMATION_EMAIL_JOB_DEFINITION_SCHEMA = z.object({
+  email: z.string().email(),
+  force: z.boolean().optional(),
+});
+
+export const SEND_CONFIRMATION_EMAIL_JOB_DEFINITION = {
+  id: SEND_CONFIRMATION_EMAIL_JOB_DEFINITION_ID,
+  name: 'Send Confirmation Email',
+  version: '1.0.0',
+  trigger: {
+    name: SEND_CONFIRMATION_EMAIL_JOB_DEFINITION_ID,
+    schema: SEND_CONFIRMATION_EMAIL_JOB_DEFINITION_SCHEMA,
+  },
+  handler: async ({ payload }) => {
+    await sendConfirmationToken({
+      email: payload.email,
+      force: payload.force,
+    });
+  },
+} as const satisfies JobDefinition<
+  typeof SEND_CONFIRMATION_EMAIL_JOB_DEFINITION_ID,
+  z.infer<typeof SEND_CONFIRMATION_EMAIL_JOB_DEFINITION_SCHEMA>
+>;

packages/lib/jobs/definitions/send-signing-email.ts

@@ -0,0 +1,171 @@
+import { createElement } from 'react';
+
+import { z } from 'zod';
+
+import { mailer } from '@documenso/email/mailer';
+import { render } from '@documenso/email/render';
+import DocumentInviteEmailTemplate from '@documenso/email/templates/document-invite';
+import { prisma } from '@documenso/prisma';
+import {
+  DocumentSource,
+  DocumentStatus,
+  RecipientRole,
+  SendStatus,
+} from '@documenso/prisma/client';
+
+import { NEXT_PUBLIC_WEBAPP_URL } from '../../constants/app';
+import { FROM_ADDRESS, FROM_NAME } from '../../constants/email';
+import {
+  RECIPIENT_ROLES_DESCRIPTION,
+  RECIPIENT_ROLE_TO_EMAIL_TYPE,
+} from '../../constants/recipient-roles';
+import { DOCUMENT_AUDIT_LOG_TYPE } from '../../types/document-audit-logs';
+import { ZRequestMetadataSchema } from '../../universal/extract-request-metadata';
+import { createDocumentAuditLogData } from '../../utils/document-audit-logs';
+import { renderCustomEmailTemplate } from '../../utils/render-custom-email-template';
+import { type JobDefinition } from '../client/_internal/job';
+
+const SEND_SIGNING_EMAIL_JOB_DEFINITION_ID = 'send.signing.requested.email';
+
+const SEND_SIGNING_EMAIL_JOB_DEFINITION_SCHEMA = z.object({
+  userId: z.number(),
+  documentId: z.number(),
+  recipientId: z.number(),
+  requestMetadata: ZRequestMetadataSchema.optional(),
+});
+
+export const SEND_SIGNING_EMAIL_JOB_DEFINITION = {
+  id: SEND_SIGNING_EMAIL_JOB_DEFINITION_ID,
+  name: 'Send Signing Email',
+  version: '1.0.0',
+  trigger: {
+    name: SEND_SIGNING_EMAIL_JOB_DEFINITION_ID,
+    schema: SEND_SIGNING_EMAIL_JOB_DEFINITION_SCHEMA,
+  },
+  handler: async ({ payload, io }) => {
+    const { userId, documentId, recipientId, requestMetadata } = payload;
+
+    const [user, document, recipient] = await Promise.all([
+      prisma.user.findFirstOrThrow({
+        where: {
+          id: userId,
+        },
+      }),
+      prisma.document.findFirstOrThrow({
+        where: {
+          id: documentId,
+          status: DocumentStatus.PENDING,
+        },
+        include: {
+          documentMeta: true,
+        },
+      }),
+      prisma.recipient.findFirstOrThrow({
+        where: {
+          id: recipientId,
+        },
+      }),
+    ]);
+
+    const { documentMeta } = document;
+
+    if (recipient.role === RecipientRole.CC) {
+      return;
+    }
+
+    const customEmail = document?.documentMeta;
+    const isDirectTemplate = document.source === DocumentSource.TEMPLATE_DIRECT_LINK;
+
+    const recipientEmailType = RECIPIENT_ROLE_TO_EMAIL_TYPE[recipient.role];
+
+    const { email, name } = recipient;
+    const selfSigner = email === user.email;
+    const { actionVerb } = RECIPIENT_ROLES_DESCRIPTION[recipient.role];
+    const recipientActionVerb = actionVerb.toLowerCase();
+
+    let emailMessage = customEmail?.message || '';
+    let emailSubject = `Please ${recipientActionVerb} this document`;
+
+    if (selfSigner) {
+      emailMessage = `You have initiated the document ${`"${document.title}"`} that requires you to ${recipientActionVerb} it.`;
+      emailSubject = `Please ${recipientActionVerb} your document`;
+    }
+
+    if (isDirectTemplate) {
+      emailMessage = `A document was created by your direct template that requires you to ${recipientActionVerb} it.`;
+      emailSubject = `Please ${recipientActionVerb} this document created by your direct template`;
+    }
+
+    const customEmailTemplate = {
+      'signer.name': name,
+      'signer.email': email,
+      'document.name': document.title,
+    };
+
+    const assetBaseUrl = NEXT_PUBLIC_WEBAPP_URL() || 'http://localhost:3000';
+    const signDocumentLink = `${NEXT_PUBLIC_WEBAPP_URL()}/sign/${recipient.token}`;
+
+    const template = createElement(DocumentInviteEmailTemplate, {
+      documentName: document.title,
+      inviterName: user.name || undefined,
+      inviterEmail: user.email,
+      assetBaseUrl,
+      signDocumentLink,
+      customBody: renderCustomEmailTemplate(emailMessage, customEmailTemplate),
+      role: recipient.role,
+      selfSigner,
+    });
+
+    await io.runTask('send-signing-email', async () => {
+      await mailer.sendMail({
+        to: {
+          name: recipient.name,
+          address: recipient.email,
+        },
+        from: {
+          name: FROM_NAME,
+          address: FROM_ADDRESS,
+        },
+        subject: renderCustomEmailTemplate(
+          documentMeta?.subject || emailSubject,
+          customEmailTemplate,
+        ),
+        html: render(template),
+        text: render(template, { plainText: true }),
+      });
+    });
+
+    await io.runTask('update-recipient', async () => {
+      await prisma.recipient.update({
+        where: {
+          id: recipient.id,
+        },
+        data: {
+          sendStatus: SendStatus.SENT,
+        },
+      });
+    });
+
+    await io.runTask('store-audit-log', async () => {
+      await prisma.documentAuditLog.create({
+        data: createDocumentAuditLogData({
+          type: DOCUMENT_AUDIT_LOG_TYPE.EMAIL_SENT,
+          documentId: document.id,
+          user,
+          requestMetadata,
+          data: {
+            emailType: recipientEmailType,
+            recipientId: recipient.id,
+            recipientName: recipient.name,
+            recipientEmail: recipient.email,
+            recipientRole: recipient.role,
+            isResending: false,
+          },
+        }),
+      });
+    });
+  },
+} as const satisfies JobDefinition<
+  typeof SEND_SIGNING_EMAIL_JOB_DEFINITION_ID,
+  z.infer<typeof SEND_SIGNING_EMAIL_JOB_DEFINITION_SCHEMA>
+>;

packages/lib/next-auth/auth-options.ts

@@ -14,11 +14,11 @@ import { prisma } from '@documenso/prisma';
 import { IdentityProvider, UserSecurityAuditLogType } from '@documenso/prisma/client';
 
 import { AppError, AppErrorCode } from '../errors/app-error';
+import { jobsClient } from '../jobs/client';
 import { isTwoFactorAuthenticationEnabled } from '../server-only/2fa/is-2fa-availble';
 import { validateTwoFactorAuthentication } from '../server-only/2fa/validate-2fa';
 import { getMostRecentVerificationTokenByUserId } from '../server-only/user/get-most-recent-verification-token-by-user-id';
 import { getUserByEmail } from '../server-only/user/get-user-by-email';
-import { sendConfirmationToken } from '../server-only/user/send-confirmation-token';
 import type { TAuthenticationResponseJSONSchema } from '../types/webauthn';
 import { ZAuthenticationResponseJSONSchema } from '../types/webauthn';
 import { extractNextAuthRequestMetadata } from '../universal/extract-request-metadata';
@@ -108,7 +108,12 @@ export const NEXT_AUTH_OPTIONS: AuthOptions = {
             mostRecentToken.expires.valueOf() <= Date.now() ||
             DateTime.fromJSDate(mostRecentToken.createdAt).diffNow('minutes').minutes > -5
           ) {
-            await sendConfirmationToken({ email });
+            await jobsClient.triggerJob({
+              name: 'send.signup.confirmation.email',
+              payload: {
+                email: user.email,
+              },
+            });
           }
 
           throw new Error(ErrorCode.UNVERIFIED_EMAIL);

packages/lib/package.json

@@ -32,10 +32,14 @@
     "@pdf-lib/fontkit": "^1.1.1",
     "@scure/base": "^1.1.3",
     "@sindresorhus/slugify": "^2.2.1",
+    "@trigger.dev/nextjs": "^2.3.18",
+    "@trigger.dev/sdk": "^2.3.18",
     "@upstash/redis": "^1.20.6",
     "@vvo/tzdb": "^6.117.0",
+    "inngest": "^3.19.13",
     "kysely": "^0.26.3",
     "luxon": "^3.4.0",
+    "micro": "^10.0.1",
     "nanoid": "^4.0.2",
     "next": "14.0.3",
     "next-auth": "4.24.5",
@@ -50,8 +54,8 @@
     "zod": "^3.22.4"
   },
   "devDependencies": {
+    "@playwright/browser-chromium": "1.43.0",
     "@types/luxon": "^3.3.1",
-    "@types/pg": "^8.11.4",
-    "@playwright/browser-chromium": "1.43.0"
+    "@types/pg": "^8.11.4"
   }
 }

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

@@ -1,16 +1,9 @@
-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';
 import { sealDocument } from '@documenso/lib/server-only/document/seal-document';
 import { updateDocument } from '@documenso/lib/server-only/document/update-document';
 import { DOCUMENT_AUDIT_LOG_TYPE } from '@documenso/lib/types/document-audit-logs';
 import type { RequestMetadata } from '@documenso/lib/universal/extract-request-metadata';
 import { putPdfFile } from '@documenso/lib/universal/upload/put-file';
 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 {
   DocumentSource,
@@ -21,11 +14,7 @@ import {
 } from '@documenso/prisma/client';
 import { WebhookTriggerEvents } from '@documenso/prisma/client';
 
-import { NEXT_PUBLIC_WEBAPP_URL } from '../../constants/app';
-import {
-  RECIPIENT_ROLES_DESCRIPTION,
-  RECIPIENT_ROLE_TO_EMAIL_TYPE,
-} from '../../constants/recipient-roles';
+import { jobsClient } from '../../jobs/client';
 import { getFile } from '../../universal/upload/get-file';
 import { insertFormValuesInPdf } from '../pdf/insert-form-values-in-pdf';
 import { triggerWebhook } from '../webhooks/trigger/trigger-webhook';
@@ -137,92 +126,15 @@ export const sendDocument = async ({
           return;
         }
 
-        const recipientEmailType = RECIPIENT_ROLE_TO_EMAIL_TYPE[recipient.role];
-
-        const { email, name } = recipient;
-        const selfSigner = email === user.email;
-        const { actionVerb } = RECIPIENT_ROLES_DESCRIPTION[recipient.role];
-        const recipientActionVerb = actionVerb.toLowerCase();
-
-        let emailMessage = customEmail?.message || '';
-        let emailSubject = `Please ${recipientActionVerb} this document`;
-
-        if (selfSigner) {
-          emailMessage = `You have initiated the document ${`"${document.title}"`} that requires you to ${recipientActionVerb} it.`;
-          emailSubject = `Please ${recipientActionVerb} your document`;
-        }
-
-        if (isDirectTemplate) {
-          emailMessage = `A document was created by your direct template that requires you to ${recipientActionVerb} it.`;
-          emailSubject = `Please ${recipientActionVerb} this document created by your direct template`;
-        }
-
-        const customEmailTemplate = {
-          'signer.name': name,
-          'signer.email': email,
-          'document.name': document.title,
-        };
-
-        const assetBaseUrl = NEXT_PUBLIC_WEBAPP_URL() || 'http://localhost:3000';
-        const signDocumentLink = `${NEXT_PUBLIC_WEBAPP_URL()}/sign/${recipient.token}`;
-
-        const template = createElement(DocumentInviteEmailTemplate, {
-          documentName: document.title,
-          inviterName: user.name || undefined,
-          inviterEmail: user.email,
-          assetBaseUrl,
-          signDocumentLink,
-          customBody: renderCustomEmailTemplate(emailMessage, customEmailTemplate),
-          role: recipient.role,
-          selfSigner,
-        });
-
-        await prisma.$transaction(
-          async (tx) => {
-            await mailer.sendMail({
-              to: {
-                address: email,
-                name,
-              },
-              from: {
-                name: FROM_NAME,
-                address: FROM_ADDRESS,
-              },
-              subject: customEmail?.subject
-                ? renderCustomEmailTemplate(customEmail.subject, customEmailTemplate)
-                : emailSubject,
-              html: render(template),
-              text: render(template, { plainText: true }),
-            });
-
-            await tx.recipient.update({
-              where: {
-                id: recipient.id,
-              },
-              data: {
-                sendStatus: SendStatus.SENT,
-              },
-            });
-
-            await tx.documentAuditLog.create({
-              data: createDocumentAuditLogData({
-                type: DOCUMENT_AUDIT_LOG_TYPE.EMAIL_SENT,
-                documentId: document.id,
-                user,
-                requestMetadata,
-                data: {
-                  emailType: recipientEmailType,
-                  recipientEmail: recipient.email,
-                  recipientName: recipient.name,
-                  recipientRole: recipient.role,
-                  recipientId: recipient.id,
-                  isResending: false,
-                },
-              }),
-            });
+        await jobsClient.triggerJob({
+          name: 'send.signing.requested.email',
+          payload: {
+            userId,
+            documentId,
+            recipientId: recipient.id,
+            requestMetadata,
           },
-          { timeout: 30_000 },
-        );
+        });
       }),
     );
   }

packages/lib/server-only/user/verify-email.ts

@@ -2,7 +2,7 @@ import { DateTime } from 'luxon';
 
 import { prisma } from '@documenso/prisma';
 
-import { sendConfirmationToken } from './send-confirmation-token';
+import { jobsClient } from '../../jobs/client';
 
 export type VerifyEmailProps = {
   token: string;
@@ -40,7 +40,12 @@ export const verifyEmail = async ({ token }: VerifyEmailProps) => {
       !mostRecentToken ||
       DateTime.now().minus({ hours: 1 }).toJSDate() > mostRecentToken.createdAt
     ) {
-      await sendConfirmationToken({ email: verificationToken.user.email });
+      await jobsClient.triggerJob({
+        name: 'send.signup.confirmation.email',
+        payload: {
+          email: verificationToken.user.email,
+        },
+      });
     }
 
     return valid;

packages/lib/universal/extract-request-metadata.ts

@@ -5,10 +5,12 @@ import { z } from 'zod';
 
 const ZIpSchema = z.string().ip();
 
-export type RequestMetadata = {
-  ipAddress?: string;
-  userAgent?: string;
-};
+export const ZRequestMetadataSchema = z.object({
+  ipAddress: ZIpSchema.optional(),
+  userAgent: z.string().optional(),
+});
+
+export type RequestMetadata = z.infer<typeof ZRequestMetadataSchema>;
 
 export const extractNextApiRequestMetadata = (req: NextApiRequest): RequestMetadata => {
   const parsedIp = ZIpSchema.safeParse(req.headers['x-forwarded-for'] || req.socket.remoteAddress);

packages/prisma/migrations/20240516022236_add_background_job_models/migration.sql

@@ -0,0 +1,37 @@
+-- CreateEnum
+CREATE TYPE "BackgroundJobStatus" AS ENUM ('PENDING', 'PROCESSING', 'COMPLETED', 'FAILED');
+
+-- CreateEnum
+CREATE TYPE "BackgroundJobTaskStatus" AS ENUM ('PENDING', 'COMPLETED', 'FAILED');
+
+-- CreateTable
+CREATE TABLE "BackgroundJob" (
+    "id" TEXT NOT NULL,
+    "status" "BackgroundJobStatus" NOT NULL DEFAULT 'PENDING',
+    "retried" INTEGER NOT NULL DEFAULT 0,
+    "maxRetries" INTEGER NOT NULL DEFAULT 3,
+    "jobId" TEXT NOT NULL,
+    "name" TEXT NOT NULL,
+    "version" TEXT NOT NULL,
+    "submittedAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "lastRetriedAt" TIMESTAMP(3),
+
+    CONSTRAINT "BackgroundJob_pkey" PRIMARY KEY ("id")
+);
+
+-- CreateTable
+CREATE TABLE "BackgroundJobTask" (
+    "id" TEXT NOT NULL,
+    "status" "BackgroundJobTaskStatus" NOT NULL DEFAULT 'PENDING',
+    "result" JSONB,
+    "retried" INTEGER NOT NULL DEFAULT 0,
+    "maxRetries" INTEGER NOT NULL DEFAULT 3,
+    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "updatedAt" TIMESTAMP(3) NOT NULL,
+    "jobId" TEXT NOT NULL,
+
+    CONSTRAINT "BackgroundJobTask_pkey" PRIMARY KEY ("id")
+);
+
+-- AddForeignKey
+ALTER TABLE "BackgroundJobTask" ADD CONSTRAINT "BackgroundJobTask_jobId_fkey" FOREIGN KEY ("jobId") REFERENCES "BackgroundJob"("id") ON DELETE CASCADE ON UPDATE CASCADE;

packages/prisma/migrations/20240516030441_add_more_timestamp_columns_for_background_jobs/migration.sql

@@ -0,0 +1,9 @@
+/*
+  Warnings:
+
+  - Added the required column `updatedAt` to the `BackgroundJob` table without a default value. This is not possible if the table is not empty.
+
+*/
+-- AlterTable
+ALTER TABLE "BackgroundJob" ADD COLUMN     "completedAt" TIMESTAMP(3),
+ADD COLUMN     "updatedAt" TIMESTAMP(3) NOT NULL;

packages/prisma/migrations/20240516032317_capture_payload_for_background_job/migration.sql

@@ -0,0 +1,2 @@
+-- AlterTable
+ALTER TABLE "BackgroundJob" ADD COLUMN     "payload" JSONB;

packages/prisma/migrations/20240516034546_update_background_job_task_model/migration.sql

@@ -0,0 +1,9 @@
+/*
+  Warnings:
+
+  - Added the required column `name` to the `BackgroundJobTask` table without a default value. This is not possible if the table is not empty.
+
+*/
+-- AlterTable
+ALTER TABLE "BackgroundJobTask" ADD COLUMN     "completedAt" TIMESTAMP(3),
+ADD COLUMN     "name" TEXT NOT NULL;

packages/prisma/schema.prisma

@@ -626,3 +626,53 @@ model SiteSettings {
   lastModifiedAt       DateTime @default(now())
   lastModifiedByUser   User?    @relation(fields: [lastModifiedByUserId], references: [id], onDelete: SetNull)
 }
+
+enum BackgroundJobStatus {
+  PENDING
+  PROCESSING
+  COMPLETED
+  FAILED
+}
+
+model BackgroundJob {
+  id         String              @id @default(cuid())
+  status     BackgroundJobStatus @default(PENDING)
+  payload    Json?
+  retried    Int                 @default(0)
+  maxRetries Int                 @default(3)
+
+  // Taken from the job definition
+  jobId   String
+  name    String
+  version String
+
+  submittedAt   DateTime  @default(now())
+  updatedAt     DateTime  @updatedAt
+  completedAt   DateTime?
+  lastRetriedAt DateTime?
+
+  tasks BackgroundJobTask[]
+}
+
+enum BackgroundJobTaskStatus {
+  PENDING
+  COMPLETED
+  FAILED
+}
+
+model BackgroundJobTask {
+  id     String                  @id
+  name   String
+  status BackgroundJobTaskStatus @default(PENDING)
+
+  result     Json?
+  retried    Int   @default(0)
+  maxRetries Int   @default(3)
+
+  createdAt   DateTime  @default(now())
+  updatedAt   DateTime  @updatedAt
+  completedAt DateTime?
+
+  jobId         String
+  backgroundJob BackgroundJob @relation(fields: [jobId], references: [id], onDelete: Cascade)
+}

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

@@ -5,6 +5,7 @@ import { env } from 'next-runtime-env';
 
 import { IS_BILLING_ENABLED } from '@documenso/lib/constants/app';
 import { AppError, AppErrorCode } from '@documenso/lib/errors/app-error';
+import { jobsClient } from '@documenso/lib/jobs/client';
 import { ErrorCode } from '@documenso/lib/next-auth/error-codes';
 import { createPasskey } from '@documenso/lib/server-only/auth/create-passkey';
 import { createPasskeyAuthenticationOptions } from '@documenso/lib/server-only/auth/create-passkey-authentication-options';
@@ -15,7 +16,6 @@ import { findPasskeys } from '@documenso/lib/server-only/auth/find-passkeys';
 import { compareSync } from '@documenso/lib/server-only/auth/hash';
 import { updatePasskey } from '@documenso/lib/server-only/auth/update-passkey';
 import { createUser } from '@documenso/lib/server-only/user/create-user';
-import { sendConfirmationToken } from '@documenso/lib/server-only/user/send-confirmation-token';
 import { extractNextApiRequestMetadata } from '@documenso/lib/universal/extract-request-metadata';
 
 import { authenticatedProcedure, procedure, router } from '../trpc';
@@ -52,7 +52,12 @@ export const authRouter = router({
 
       const user = await createUser({ name, email, password, signature, url });
 
-      await sendConfirmationToken({ email: user.email });
+      await jobsClient.triggerJob({
+        name: 'send.signup.confirmation.email',
+        payload: {
+          email: user.email,
+        },
+      });
 
       return user;
     } catch (err) {

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

@@ -2,13 +2,13 @@ import { TRPCError } from '@trpc/server';
 
 import { IS_BILLING_ENABLED } from '@documenso/lib/constants/app';
 import { AppError, AppErrorCode } from '@documenso/lib/errors/app-error';
+import { jobsClient } from '@documenso/lib/jobs/client';
 import { getSubscriptionsByUserId } from '@documenso/lib/server-only/subscription/get-subscriptions-by-user-id';
 import { deleteUser } from '@documenso/lib/server-only/user/delete-user';
 import { findUserSecurityAuditLogs } from '@documenso/lib/server-only/user/find-user-security-audit-logs';
 import { forgotPassword } from '@documenso/lib/server-only/user/forgot-password';
 import { getUserById } from '@documenso/lib/server-only/user/get-user-by-id';
 import { resetPassword } from '@documenso/lib/server-only/user/reset-password';
-import { sendConfirmationToken } from '@documenso/lib/server-only/user/send-confirmation-token';
 import { updatePassword } from '@documenso/lib/server-only/user/update-password';
 import { updateProfile } from '@documenso/lib/server-only/user/update-profile';
 import { updatePublicProfile } from '@documenso/lib/server-only/user/update-public-profile';
@@ -204,7 +204,12 @@ export const profileRouter = router({
       try {
         const { email } = input;
 
-        return await sendConfirmationToken({ email });
+        await jobsClient.triggerJob({
+          name: 'send.signup.confirmation.email',
+          payload: {
+            email,
+          },
+        });
       } catch (err) {
         console.error(err);
 

packages/tsconfig/process-env.d.ts

@@ -68,6 +68,19 @@ declare namespace NodeJS {
     //
     NEXT_PRIVATE_BROWSERLESS_URL?: string;
 
+    NEXT_PRIVATE_JOBS_PROVIDER?: 'trigger' | 'inngest' | 'local';
+
+    /**
+     * Trigger.dev environment variables
+     */
+    NEXT_PRIVATE_TRIGGER_API_KEY?: string;
+    NEXT_PRIVATE_TRIGGER_API_URL?: string;
+
+    /**
+     * Inngest environment variables
+     */
+    NEXT_PRIVATE_INNGEST_EVENT_KEY?: string;
+
     /**
      * Vercel environment variables
      */

turbo.json

@@ -115,6 +115,10 @@
     "NEXT_PRIVATE_STRIPE_WEBHOOK_SECRET",
     "NEXT_PRIVATE_GITHUB_TOKEN",
     "NEXT_PRIVATE_BROWSERLESS_URL",
+    "NEXT_PRIVATE_JOBS_PROVIDER",
+    "NEXT_PRIVATE_TRIGGER_API_KEY",
+    "NEXT_PRIVATE_TRIGGER_API_URL",
+    "NEXT_PRIVATE_INNGEST_EVENT_KEY",
     "CI",
     "VERCEL",
     "VERCEL_ENV",