pn-new-crm/src/jobs/processors/documenso-poll.ts

import { and, eq, inArray, isNotNull } from 'drizzle-orm';

import { db } from '@/lib/db';
import { documents, documentSigners } from '@/lib/db/schema/documents';
import { getDocument as getDocumensoDoc } from '@/lib/services/documenso-client';
import {
  handleRecipientSigned,
  handleDocumentCompleted,
  handleDocumentExpired,
} from '@/lib/services/documents.service';
import { logger } from '@/lib/logger';

export async function processDocumensoPoll(): Promise<void> {
  // Find all documents that are in-progress signing and have a documensoId
  const pendingDocs = await db.query.documents.findMany({
    where: and(
      inArray(documents.status, ['sent', 'partially_signed']),
      isNotNull(documents.documensoId),
    ),
  });

  if (pendingDocs.length === 0) return;

  logger.info({ count: pendingDocs.length }, 'Polling Documenso for document statuses');

  for (const doc of pendingDocs) {
    if (!doc.documensoId) continue;

    try {
      // Pass the doc's portId so the client uses per-port credentials
      // (admin-set Documenso URL/key/version), not the global env fallback.
      // Without portId a multi-port deployment hits the wrong instance and
      // 401s every poll.
      const remoteDoc = await getDocumensoDoc(doc.documensoId, doc.portId);

      // Reconcile signer statuses
      for (const remoteRecipient of remoteDoc.recipients) {
        if (remoteRecipient.status === 'SIGNED') {
          // Check if we already have this signer as signed locally
          const localSigner = await db.query.documentSigners.findFirst({
            where: and(
              eq(documentSigners.documentId, doc.id),
              eq(documentSigners.signerEmail, remoteRecipient.email),
            ),
          });

          if (localSigner && localSigner.status !== 'signed') {
            logger.info(
              { documentId: doc.id, email: remoteRecipient.email, portId: doc.portId },
              'Reconciling signed signer from poll',
            );
            // Thread portId from the workflow's port context so the webhook
            // handlers run port-scoped lookups (resolveWebhookDocument) rather
            // than the port-ambiguous fallback.
            await handleRecipientSigned({
              documentId: doc.documensoId,
              recipientEmail: remoteRecipient.email,
              portId: doc.portId,
            });
          }
        }
      }

      // Reconcile document status
      if (remoteDoc.status === 'COMPLETED' && doc.status !== 'completed') {
        logger.info(
          { documentId: doc.id, portId: doc.portId },
          'Reconciling completed document from poll',
        );
        await handleDocumentCompleted({ documentId: doc.documensoId, portId: doc.portId });
      } else if (remoteDoc.status === 'EXPIRED' && doc.status !== 'expired') {
        logger.info(
          { documentId: doc.id, portId: doc.portId },
          'Reconciling expired document from poll',
        );
        await handleDocumentExpired({ documentId: doc.documensoId, portId: doc.portId });
      }
    } catch (err) {
      logger.error(
        { err, documentId: doc.id, documensoId: doc.documensoId },
        'Documenso poll failed for document',
      );
    }
  }
}