feat: add seal-document sweep job and admin unsealed documents page (#2563)

packages/lib/jobs/client.ts

@@ -16,6 +16,7 @@ import { EXECUTE_WEBHOOK_JOB_DEFINITION } from './definitions/internal/execute-w
 import { EXPIRE_RECIPIENTS_SWEEP_JOB_DEFINITION } from './definitions/internal/expire-recipients-sweep';
 import { PROCESS_RECIPIENT_EXPIRED_JOB_DEFINITION } from './definitions/internal/process-recipient-expired';
 import { SEAL_DOCUMENT_JOB_DEFINITION } from './definitions/internal/seal-document';
+import { SEAL_DOCUMENT_SWEEP_JOB_DEFINITION } from './definitions/internal/seal-document-sweep';
 import { SYNC_EMAIL_DOMAINS_JOB_DEFINITION } from './definitions/internal/sync-email-domains';
 
 /**
@@ -29,6 +30,7 @@ export const jobsClient = new JobClient([
   SEND_ORGANISATION_MEMBER_LEFT_EMAIL_JOB_DEFINITION,
   SEND_TEAM_DELETED_EMAIL_JOB_DEFINITION,
   SEAL_DOCUMENT_JOB_DEFINITION,
+  SEAL_DOCUMENT_SWEEP_JOB_DEFINITION,
   SEND_PASSWORD_RESET_SUCCESS_EMAIL_JOB_DEFINITION,
   SEND_SIGNING_REJECTION_EMAILS_JOB_DEFINITION,
   SEND_RECIPIENT_SIGNED_EMAIL_JOB_DEFINITION,

packages/lib/jobs/definitions/internal/seal-document-sweep.handler.ts

@@ -0,0 +1,105 @@
+import { DocumentStatus, EnvelopeType, RecipientRole, SigningStatus } from '@prisma/client';
+import { DateTime } from 'luxon';
+
+import { kyselyPrisma, sql } from '@documenso/prisma';
+
+import { mapSecondaryIdToDocumentId } from '../../../utils/envelope';
+import { jobs } from '../../client';
+import type { JobRunIO } from '../../client/_internal/job';
+import type { TSealDocumentSweepJobDefinition } from './seal-document-sweep';
+
+export const run = async ({ io }: { payload: TSealDocumentSweepJobDefinition; io: JobRunIO }) => {
+  const now = DateTime.now();
+  const fifteenMinutesAgo = now.minus({ minutes: 15 }).toJSDate();
+  const sixHoursAgo = now.minus({ hours: 6 }).toJSDate();
+
+  // Find all PENDING envelopes that should have been sealed but weren't.
+  //
+  // A document is ready to seal when either:
+  //   1. All recipients are SIGNED or have role CC (normal completion)
+  //   2. Any recipient has REJECTED (rejection triggers immediate seal)
+  //
+  // We only look at documents where the last action was between 15 minutes
+  // and 6 hours ago. The lower bound avoids racing with the normal seal-document
+  // job that fires on completion. The upper bound stops us from endlessly retrying
+  // documents that are stuck due to a deeper issue (e.g. corrupt PDF).
+  const unsealedEnvelopes = await kyselyPrisma.$kysely
+    .selectFrom('Envelope')
+    .select(['Envelope.id', 'Envelope.secondaryId'])
+    .where('Envelope.status', '=', sql.lit(DocumentStatus.PENDING))
+    .where('Envelope.type', '=', sql.lit(EnvelopeType.DOCUMENT))
+    .where('Envelope.deletedAt', 'is', null)
+    // Ensure there is at least one recipient.
+    .where((eb) =>
+      eb.exists(eb.selectFrom('Recipient').whereRef('Recipient.envelopeId', '=', 'Envelope.id')),
+    )
+    // Document is ready to seal: all recipients are SIGNED/CC, or any recipient REJECTED.
+    .where((eb) =>
+      eb.or([
+        // Case 1: All recipients are either SIGNED or CC.
+        eb.not(
+          eb.exists(
+            eb
+              .selectFrom('Recipient')
+              .whereRef('Recipient.envelopeId', '=', 'Envelope.id')
+              .where('Recipient.signingStatus', '!=', sql.lit(SigningStatus.SIGNED))
+              .where('Recipient.role', '!=', sql.lit(RecipientRole.CC)),
+          ),
+        ),
+        // Case 2: Any recipient has rejected.
+        eb.exists(
+          eb
+            .selectFrom('Recipient')
+            .whereRef('Recipient.envelopeId', '=', 'Envelope.id')
+            .where('Recipient.signingStatus', '=', sql.lit(SigningStatus.REJECTED)),
+        ),
+      ]),
+    )
+    // Exclude envelopes where a recipient signed/rejected within the last 15 minutes
+    // to avoid racing with the standard completion flow.
+    .where((eb) =>
+      eb.not(
+        eb.exists(
+          eb
+            .selectFrom('Recipient')
+            .whereRef('Recipient.envelopeId', '=', 'Envelope.id')
+            .where('Recipient.signedAt', '>', fifteenMinutesAgo),
+        ),
+      ),
+    )
+    // Exclude envelopes where all activity is older than 6 hours.
+    // These are likely stuck due to a deeper issue and should not be retried.
+    .where((eb) =>
+      eb.exists(
+        eb
+          .selectFrom('Recipient')
+          .whereRef('Recipient.envelopeId', '=', 'Envelope.id')
+          .where('Recipient.signedAt', '>', sixHoursAgo),
+      ),
+    )
+    .limit(100)
+    .execute();
+
+  if (unsealedEnvelopes.length === 0) {
+    io.logger.info('No unsealed documents found');
+    return;
+  }
+
+  io.logger.info(`Found ${unsealedEnvelopes.length} unsealed documents`);
+
+  await Promise.allSettled(
+    unsealedEnvelopes.map(async (envelope) => {
+      const documentId = mapSecondaryIdToDocumentId(envelope.secondaryId);
+
+      io.logger.info(`Triggering seal for document ${documentId} (${envelope.id})`);
+
+      await jobs.triggerJob({
+        name: 'internal.seal-document',
+        payload: {
+          documentId,
+          isResealing: true,
+        },
+      });
+    }),
+  );
+};

packages/lib/jobs/definitions/internal/seal-document-sweep.ts

@@ -0,0 +1,30 @@
+import { z } from 'zod';
+
+import { type JobDefinition } from '../../client/_internal/job';
+
+const SEAL_DOCUMENT_SWEEP_JOB_DEFINITION_ID = 'internal.seal-document-sweep';
+
+const SEAL_DOCUMENT_SWEEP_JOB_DEFINITION_SCHEMA = z.object({});
+
+export type TSealDocumentSweepJobDefinition = z.infer<
+  typeof SEAL_DOCUMENT_SWEEP_JOB_DEFINITION_SCHEMA
+>;
+
+export const SEAL_DOCUMENT_SWEEP_JOB_DEFINITION = {
+  id: SEAL_DOCUMENT_SWEEP_JOB_DEFINITION_ID,
+  name: 'Seal Document Sweep',
+  version: '1.0.0',
+  trigger: {
+    name: SEAL_DOCUMENT_SWEEP_JOB_DEFINITION_ID,
+    schema: SEAL_DOCUMENT_SWEEP_JOB_DEFINITION_SCHEMA,
+    cron: '*/15 * * * *', // Every 15 minutes.
+  },
+  handler: async ({ payload, io }) => {
+    const handler = await import('./seal-document-sweep.handler');
+
+    await handler.run({ payload, io });
+  },
+} as const satisfies JobDefinition<
+  typeof SEAL_DOCUMENT_SWEEP_JOB_DEFINITION_ID,
+  TSealDocumentSweepJobDefinition
+>;

packages/lib/server-only/admin/admin-find-unsealed-documents.ts

@@ -0,0 +1,102 @@
+import { DocumentStatus, EnvelopeType, RecipientRole, SigningStatus } from '@prisma/client';
+
+import { kyselyPrisma, sql } from '@documenso/prisma';
+
+import type { FindResultResponse } from '../../types/search-params';
+
+export type AdminUnsealedDocument = {
+  id: string;
+  secondaryId: string;
+  title: string;
+  status: string;
+  createdAt: Date;
+  updatedAt: Date;
+  userId: number;
+  teamId: number;
+  ownerName: string | null;
+  ownerEmail: string;
+  lastSignedAt: Date | null;
+};
+
+export type AdminFindUnsealedDocumentsOptions = {
+  page?: number;
+  perPage?: number;
+};
+
+export const adminFindUnsealedDocuments = async ({
+  page = 1,
+  perPage = 20,
+}: AdminFindUnsealedDocumentsOptions): Promise<FindResultResponse<AdminUnsealedDocument[]>> => {
+  const offset = Math.max(page - 1, 0) * perPage;
+
+  const baseQuery = kyselyPrisma.$kysely
+    .selectFrom('Envelope')
+    .where('Envelope.status', '=', sql.lit(DocumentStatus.PENDING))
+    .where('Envelope.type', '=', sql.lit(EnvelopeType.DOCUMENT))
+    .where('Envelope.deletedAt', 'is', null)
+    // Must have at least one recipient.
+    .where((eb) =>
+      eb.exists(eb.selectFrom('Recipient').whereRef('Recipient.envelopeId', '=', 'Envelope.id')),
+    )
+    // Document is ready to seal: all recipients are SIGNED/CC, or any recipient REJECTED.
+    .where((eb) =>
+      eb.or([
+        // Case 1: All recipients are either SIGNED or CC.
+        eb.not(
+          eb.exists(
+            eb
+              .selectFrom('Recipient')
+              .whereRef('Recipient.envelopeId', '=', 'Envelope.id')
+              .where('Recipient.signingStatus', '!=', sql.lit(SigningStatus.SIGNED))
+              .where('Recipient.role', '!=', sql.lit(RecipientRole.CC)),
+          ),
+        ),
+        // Case 2: Any recipient has rejected.
+        eb.exists(
+          eb
+            .selectFrom('Recipient')
+            .whereRef('Recipient.envelopeId', '=', 'Envelope.id')
+            .where('Recipient.signingStatus', '=', sql.lit(SigningStatus.REJECTED)),
+        ),
+      ]),
+    );
+
+  const [data, countResult] = await Promise.all([
+    baseQuery
+      .innerJoin('User', 'User.id', 'Envelope.userId')
+      .select([
+        'Envelope.id',
+        'Envelope.secondaryId',
+        'Envelope.title',
+        'Envelope.status',
+        'Envelope.createdAt',
+        'Envelope.updatedAt',
+        'Envelope.userId',
+        'Envelope.teamId',
+        'User.name as ownerName',
+        'User.email as ownerEmail',
+      ])
+      .select((eb) =>
+        eb
+          .selectFrom('Recipient')
+          .whereRef('Recipient.envelopeId', '=', 'Envelope.id')
+          .select(sql<Date>`max("Recipient"."signedAt")`.as('lastSignedAt'))
+          .as('lastSignedAt'),
+      )
+      .orderBy('Envelope.createdAt', 'desc')
+      .limit(perPage)
+      .offset(offset)
+      .execute(),
+    baseQuery.select(({ fn }) => [fn.countAll().as('count')]).execute(),
+  ]);
+
+  const count = Number(countResult[0]?.count ?? 0);
+
+  return {
+    data: data as unknown as AdminUnsealedDocument[],
+    count,
+    currentPage: Math.max(page, 1),
+    perPage,
+    totalPages: Math.ceil(count / perPage),
+  };
+};

packages/trpc/server/admin-router/find-unsealed-documents.ts

@@ -0,0 +1,16 @@
+import { adminFindUnsealedDocuments } from '@documenso/lib/server-only/admin/admin-find-unsealed-documents';
+
+import { adminProcedure } from '../trpc';
+import {
+  ZFindUnsealedDocumentsRequestSchema,
+  ZFindUnsealedDocumentsResponseSchema,
+} from './find-unsealed-documents.types';
+
+export const findUnsealedDocumentsRoute = adminProcedure
+  .input(ZFindUnsealedDocumentsRequestSchema)
+  .output(ZFindUnsealedDocumentsResponseSchema)
+  .query(async ({ input }) => {
+    const { page, perPage } = input;
+
+    return await adminFindUnsealedDocuments({ page, perPage });
+  });

packages/trpc/server/admin-router/find-unsealed-documents.types.ts

@@ -0,0 +1,31 @@
+import { z } from 'zod';
+
+import { ZFindResultResponse, ZFindSearchParamsSchema } from '@documenso/lib/types/search-params';
+
+export const ZFindUnsealedDocumentsRequestSchema = ZFindSearchParamsSchema.pick({
+  page: true,
+  perPage: true,
+}).extend({
+  perPage: z.number().optional().default(20),
+});
+
+export const ZAdminUnsealedDocumentSchema = z.object({
+  id: z.string(),
+  secondaryId: z.string(),
+  title: z.string(),
+  status: z.string(),
+  createdAt: z.date(),
+  updatedAt: z.date(),
+  userId: z.number(),
+  teamId: z.number(),
+  ownerName: z.string().nullable(),
+  ownerEmail: z.string(),
+  lastSignedAt: z.date().nullable(),
+});
+
+export const ZFindUnsealedDocumentsResponseSchema = ZFindResultResponse.extend({
+  data: ZAdminUnsealedDocumentSchema.array(),
+});
+
+export type TFindUnsealedDocumentsRequest = z.infer<typeof ZFindUnsealedDocumentsRequestSchema>;
+export type TFindUnsealedDocumentsResponse = z.infer<typeof ZFindUnsealedDocumentsResponseSchema>;

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

@@ -13,6 +13,7 @@ import { findDocumentJobsRoute } from './find-document-jobs';
 import { findDocumentsRoute } from './find-documents';
 import { findEmailDomainsRoute } from './find-email-domains';
 import { findSubscriptionClaimsRoute } from './find-subscription-claims';
+import { findUnsealedDocumentsRoute } from './find-unsealed-documents';
 import { findUserTeamsRoute } from './find-user-teams';
 import { getAdminOrganisationRoute } from './get-admin-organisation';
 import { getEmailDomainRoute } from './get-email-domain';
@@ -65,6 +66,7 @@ export const adminRouter = router({
   },
   document: {
     find: findDocumentsRoute,
+    findUnsealed: findUnsealedDocumentsRoute,
     delete: deleteDocumentRoute,
     reseal: resealDocumentRoute,
     findJobs: findDocumentJobsRoute,