pn-new-crm/tests/integration/documenso-webhook-completion-idempotency.test.ts

/**
 * Initiative 4 (e2e testing) — Documenso DOCUMENT_COMPLETED idempotency.
 *
 * CLAUDE.md states `handleDocumentCompleted` is idempotent: it early-returns
 * when `status='completed' && signedFileId` is already set, so Documenso's
 * 5xx-retry storm (and the reconciling poll worker) can't double-write the
 * signed PDF, re-clobber `documents.signedFileId`, or leak the first blob.
 *
 * Existing coverage gap (checked 2026-06-04):
 *  - `documents-completion-auto-deposit.test.ts` calls `handleDocumentCompleted`
 *    exactly ONCE per case (asserts the deposit folder/FK wiring), never a replay.
 *  - `documenso-webhook-route.test.ts` exercises the route, and its dedup case
 *    only covers the *route-level* `signatureHash` replay guard for an
 *    OPENED event with an identical body — NOT the *handler-level* idempotency
 *    gate inside `handleDocumentCompleted` (line ~1464 of documents.service.ts).
 *
 * This file fills that gap: replay DOCUMENT_COMPLETED 3× for the same document
 * and assert exactly one `files` row is minted by completion (single
 * `signedFileId` pointer, stable across replays) and exactly one
 * `audit_logs` provenance row (`action='create'`, `entityType='file'`,
 * `newValue.source='documenso_completion'`). Mocks all externals (Documenso
 * download + storage backend) per the integration-test convention.
 */

import { afterEach, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest';
import { and, eq, sql } from 'drizzle-orm';

import { db } from '@/lib/db';
import { documents, files, documentFolders, documentEvents } from '@/lib/db/schema/documents';
import { auditLogs } from '@/lib/db/schema/system';
import { user } from '@/lib/db/schema/users';
import { handleDocumentCompleted } from '@/lib/services/documents.service';
import { ensureSystemRoots } from '@/lib/services/document-folders.service';
import { makeClient, makePort } from '../helpers/factories';

// Module-scoped counters the mock factories write into. Plain mutable
// containers (not vi.fn) so they can be safely referenced from the hoisted
// `vi.mock` factory below without the "cannot access before init" trap.
const storagePuts: string[] = [];
const downloadCalls: { count: number } = { count: 0 };

// Stub the Documenso signed-PDF download so we never hit the network. A
// fresh non-empty buffer keeps the 0-byte guard in the handler happy.
vi.mock('@/lib/services/documenso-client', async (importOriginal) => {
  const real = await importOriginal<typeof import('@/lib/services/documenso-client')>();
  return {
    ...real,
    downloadSignedPdf: async () => {
      downloadCalls.count += 1;
      return Buffer.from('%PDF-1.4 idempotency stub\n');
    },
  };
});

// Stub storage so no MinIO is required. We count `put` calls to prove the
// idempotency gate short-circuits BEFORE re-uploading on the 2nd/3rd replay.
vi.mock('@/lib/storage', async (importOriginal) => {
  const real = await importOriginal<typeof import('@/lib/storage')>();
  const blobs = new Map<string, Buffer>();
  return {
    ...real,
    getStorageBackend: async () => ({
      name: 's3' as const,
      put: async (key: string, body: Buffer | NodeJS.ReadableStream) => {
        const buf = Buffer.isBuffer(body) ? body : Buffer.from('');
        blobs.set(key, buf);
        storagePuts.push(key);
        return { key, sizeBytes: buf.length, sha256: 'stub'.padEnd(64, '0') };
      },
      get: async (key: string) => {
        const { Readable } = await import('node:stream');
        return Readable.from([blobs.get(key) ?? Buffer.alloc(0)]);
      },
      head: async (key: string) => {
        const buf = blobs.get(key);
        return buf ? { sizeBytes: buf.length, contentType: 'application/pdf' } : null;
      },
      delete: async (key: string) => {
        blobs.delete(key);
      },
      presignUpload: async () => ({ url: 'http://stub', method: 'PUT' as const }),
      presignDownload: async () => ({ url: 'http://stub', expiresAt: new Date(Date.now() + 1000) }),
      listByPrefix: async (prefix: string) => [...blobs.keys()].filter((k) => k.startsWith(prefix)),
    }),
  };
});

/**
 * `handleDocumentCompleted` fires its audit-log write as `void createAuditLog(...)`
 * (fire-and-forget). Give the microtask + the single DB insert a beat to land
 * before we assert on `audit_logs`.
 */
async function flushAudit(): Promise<void> {
  await new Promise((r) => setTimeout(r, 150));
}

let TEST_USER_ID = '';

beforeAll(async () => {
  const [u] = await db.select({ id: user.id }).from(user).limit(1);
  if (!u) throw new Error('No user available; run pnpm db:seed first');
  TEST_USER_ID = u.id;
});

afterEach(() => {
  storagePuts.length = 0;
  downloadCalls.count = 0;
});

describe('handleDocumentCompleted · idempotency on webhook replay', () => {
  let portId: string;
  let clientId: string;

  beforeEach(async () => {
    storagePuts.length = 0;
    downloadCalls.count = 0;
    const port = await makePort();
    portId = port.id;

    await db.delete(documentFolders).where(eq(documentFolders.portId, portId));
    await ensureSystemRoots(portId, TEST_USER_ID);

    const client = await makeClient({ portId });
    clientId = client.id;
  });

  it('replaying DOCUMENT_COMPLETED 3× writes exactly one signed file + one audit row', async () => {
    const documensoId = `docu-idem-${Date.now()}`;
    const [doc] = await db
      .insert(documents)
      .values({
        portId,
        clientId,
        documentType: 'eoi',
        title: 'Idempotency replay EOI',
        status: 'partially_signed',
        documensoId,
        createdBy: 'seed',
      })
      .returning();

    // Three identical deliveries, exactly as Documenso would retry on a 5xx.
    await handleDocumentCompleted({ documentId: documensoId, portId });
    await handleDocumentCompleted({ documentId: documensoId, portId });
    await handleDocumentCompleted({ documentId: documensoId, portId });

    // ── Gate 1: storage.put fired exactly once. The 2nd/3rd replay must
    //    short-circuit at the `status==='completed' && signedFileId` guard
    //    BEFORE re-downloading + re-uploading. ──────────────────────────────
    expect(storagePuts).toHaveLength(1);
    expect(downloadCalls.count).toBe(1);

    // ── Gate 2: the document points at one stable signedFileId. ────────────
    const updatedDoc = await db.query.documents.findFirst({
      where: eq(documents.id, doc!.id),
    });
    expect(updatedDoc?.status).toBe('completed');
    expect(updatedDoc?.signedFileId).not.toBeNull();
    const signedFileId = updatedDoc!.signedFileId!;

    // ── Gate 3: exactly one `files` row exists for this completion. A
    //    double-write would leave a 2nd files row orphaned (no DB pointer). ──
    const fileRows = await db
      .select({ id: files.id, folderId: files.folderId })
      .from(files)
      .where(eq(files.portId, portId));
    expect(fileRows).toHaveLength(1);
    expect(fileRows[0]!.id).toBe(signedFileId);
    // And it was deposited into the client entity subfolder (folder_id set once).
    expect(fileRows[0]!.folderId).not.toBeNull();

    // ── Gate 4: exactly one audit_logs provenance row for the file create. ──
    await flushAudit();
    const auditRows = await db
      .select({ id: auditLogs.id, entityId: auditLogs.entityId })
      .from(auditLogs)
      .where(
        and(
          eq(auditLogs.portId, portId),
          eq(auditLogs.action, 'create'),
          eq(auditLogs.entityType, 'file'),
          sql`${auditLogs.newValue}->>'source' = 'documenso_completion'`,
        ),
      );
    expect(auditRows).toHaveLength(1);
    expect(auditRows[0]!.entityId).toBe(signedFileId);

    // ── Gate 5: only one 'completed' documentEvents row (the handler writes
    //    one per non-short-circuited pass). ──────────────────────────────────
    const completedEvents = await db
      .select({ id: documentEvents.id })
      .from(documentEvents)
      .where(
        and(eq(documentEvents.documentId, doc!.id), eq(documentEvents.eventType, 'completed')),
      );
    expect(completedEvents).toHaveLength(1);
  });

  it('a single delivery and a 3× replay converge on the same end state', async () => {
    // Control: prove the replay leaves the DB in the identical shape a single
    // delivery would — same signedFileId, same file count, same audit count.
    const singleId = `docu-idem-single-${Date.now()}`;
    const replayId = `docu-idem-replay-${Date.now()}`;

    const client2 = await makeClient({ portId });

    const [docSingle] = await db
      .insert(documents)
      .values({
        portId,
        clientId,
        documentType: 'eoi',
        title: 'Single delivery',
        status: 'partially_signed',
        documensoId: singleId,
        createdBy: 'seed',
      })
      .returning();
    const [docReplay] = await db
      .insert(documents)
      .values({
        portId,
        clientId: client2.id,
        documentType: 'eoi',
        title: 'Replayed delivery',
        status: 'partially_signed',
        documensoId: replayId,
        createdBy: 'seed',
      })
      .returning();

    await handleDocumentCompleted({ documentId: singleId, portId });

    await handleDocumentCompleted({ documentId: replayId, portId });
    await handleDocumentCompleted({ documentId: replayId, portId });
    await handleDocumentCompleted({ documentId: replayId, portId });

    const single = await db.query.documents.findFirst({ where: eq(documents.id, docSingle!.id) });
    const replay = await db.query.documents.findFirst({ where: eq(documents.id, docReplay!.id) });

    expect(single?.status).toBe('completed');
    expect(replay?.status).toBe('completed');
    expect(single?.signedFileId).not.toBeNull();
    expect(replay?.signedFileId).not.toBeNull();

    // Two completions across the two docs → exactly two file rows in the port.
    const fileCount = await db.select({ id: files.id }).from(files).where(eq(files.portId, portId));
    expect(fileCount).toHaveLength(2);

    await flushAudit();
    const auditRows = await db
      .select({ id: auditLogs.id })
      .from(auditLogs)
      .where(
        and(
          eq(auditLogs.portId, portId),
          eq(auditLogs.action, 'create'),
          eq(auditLogs.entityType, 'file'),
          sql`${auditLogs.newValue}->>'source' = 'documenso_completion'`,
        ),
      );
    // One provenance row per completed document — the 3× replay added zero extra.
    expect(auditRows).toHaveLength(2);
  });
});