E-Signature API Webhooks: Get Notified the Instant a Document Is Signed

 [ Incoming Webhook Event ]
         │
         ▼
 ┌───────────────────────┐
 │   Signature Valid?    │
 └───────────┬───────────┘
             ├───────────────── No ──────────────► [ 401 Unauthorized / Drop ]
             │
            Yes
             │
             ▼
 ┌───────────────────────┐
 │ Ingest & Ack (200 OK) │
 └───────────┬───────────┘
             │
             ▼
 ┌───────────────────────┐
 │ Push to Message Queue │◄────────────────────────────────┐
 └───────────┬───────────┘                                 │
             │                                             │
             ▼                                             │
 ┌───────────────────────┐                                 │
 │    BullMQ / Celery    │                                 │
 │     Worker Pulls      │                                 │
 └───────────┬───────────┘                                 │
             │                                             │
      Worker Processes                                     │
             │                                             │
      Did it succeed?                                      │
       /           \                                       │
     Yes            No                                     │
     /                \                                    │
    ▼                  ▼                                   │
[ Success / Ack ]  [ Max Retries Reached? ]                │
                       /            \                      │
                     Yes             No                    │
                     /                 \                   │
                    ▼                   ▼                  │
        [ Dead Letter Queue ]    [ Calculate Backoff ]     │
        [ (DLQ) for Alerts  ]    [ (Exponential + Jitter) ]│
                                         │                 │
                                         └─────────────────┘

// worker.ts
import { Queue, Worker, Job } from 'bullmq';
import { Webhook } from 'standard-webhooks';
import { PrismaClient } from '@prisma/client';

const prisma = new PrismaClient();
const WEBHOOK_SECRET = process.env.SIGNBEE_WEBHOOK_SECRET || 'whsec_your_secret';

// Initialize Redis-backed queue with exponential retry backoff
export const webhookQueue = new Queue('webhook-processing', {
  connection: { host: '127.0.0.1', port: 6379 },
  defaultJobOptions: {
    attempts: 5,
    backoff: {
      type: 'exponential',
      delay: 2000, // Retries in 2s, 4s, 8s, 16s, 32s
    },
    removeOnComplete: true,
  },
});

// 1. Ingestion Endpoint logic (Express / Next.js API Route)
export async function handleIncomingWebhook(reqBody: string, headers: Record<string, string>) {
  const signature = headers['x-signbee-signature'];
  const timestamp = headers['x-signbee-timestamp'];
  const eventId = headers['x-signbee-event-id'];

  if (!signature || !timestamp || !eventId) {
    throw new Error('Missing webhook verification headers');
  }

  // Validate the signature using the svix standard webhook specification
  const wh = new Webhook(WEBHOOK_SECRET);
  try {
    wh.verify(reqBody, {
      'webhook-id': eventId,
      'webhook-signature': signature,
      'webhook-timestamp': timestamp,
    });
  } catch (err) {
    throw new Error('Signature verification failed: unauthorized payload');
  }

  // Enqueue job into BullMQ immediately for async processing
  const payload = JSON.parse(reqBody);
  await webhookQueue.add(
    `event-${eventId}`,
    { eventId, payload },
    { jobId: eventId } // Redis job deduplication helps prevent race conditions
  );
}

// 2. Concurrency-limited background worker
const worker = new Worker('webhook-processing', async (job: Job) => {
  const { eventId, payload } = job.data;
  const { event, document_id, signer_email, signed_at } = payload;

  console.log(`[Worker] Processing event: ${event} | ID: ${eventId}`);

  // Ensure strict idempotency at database level
  const existingEvent = await prisma.processedWebhook.findUnique({
    where: { id: eventId }
  });

  if (existingEvent) {
    console.log(`[Worker] Duplicate event detected (ID: ${eventId}). Skipping processing.`);
    return;
  }

  // Execute updates inside an atomic database transaction
  await prisma.$transaction(async (tx) => {
    // Record event as processed
    await tx.processedWebhook.create({
      data: { id: eventId, eventType: event }
    });

    if (event === 'document.signed') {
      // Sync document signing state
      await tx.contract.update({
        where: { documentId: document_id },
        data: {
          status: 'SIGNED',
          signedAt: signed_at ? new Date(signed_at) : new Date(),
          signerEmail: signer_email
        }
      });
      
      // Perform downstream workflows (e.g. provisioning access)
      await triggerProvisioningWorkflow(signer_email, document_id);
    }
  });
}, {
  connection: { host: '127.0.0.1', port: 6379 },
  limiter: {
    max: 15, // Process maximum of 15 jobs per second to prevent DB pool exhaustion
    duration: 1000,
  }
});

async function triggerProvisioningWorkflow(email: string, docId: string) {
  // downstream integration code...
}

worker.on('failed', (job, err) => {
  console.error(`Job ${job?.id} failed ultimately: ${err.message}`);
});

# tasks.py
import os
import logging
from celery import Celery
from standardwebhooks.webhooks import Webhook
from db_connection import get_db_pool # Mock connection utility

logger = logging.getLogger(__name__)

# Configure Celery to use Redis as a broker and status backend
REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379/0")
app = Celery("signbee_tasks", broker=REDIS_URL, backend=REDIS_URL)

app.conf.update(
    task_serializer="json",
    result_serializer="json",
    accept_content=["json"],
    timezone="UTC",
    enable_utc=True,
    task_acks_late=True, # Acknowledge only after processing completes
    task_reject_on_worker_lost=True,
)

SIGNBEE_WEBHOOK_SECRET = os.getenv("SIGNBEE_WEBHOOK_SECRET", "whsec_your_secret")

@app.task(
    bind=True,
    max_retries=5,
    default_retry_delay=2, # Initial retry wait of 2s
    autoretry_for=(Exception,), # Automatically retry on database errors/timeouts
    retry_backoff=True, # Exponential delays: 2s, 4s, 8s, 16s...
    retry_backoff_max=300, # Cap retry spacing to 5 minutes
    retry_jitter=True # Avoid thundering herd with randomized offset
)
def process_webhook_task(self, event_id, payload):
    logger.info(f"Processing webhook task {event_id} (Attempt {self.request.retries + 1})")
    
    event_type = payload.get("event")
    document_id = payload.get("document_id")
    signer_email = payload.get("signer_email")
    signed_at = payload.get("signed_at")

    db_pool = get_db_pool()
    with db_pool.getconn() as conn:
        with conn.cursor() as cur:
            # 1. Idempotency Check: search if event_id exists
            cur.execute("SELECT id FROM processed_webhooks WHERE id = %s;", (event_id,))
            if cur.fetchone():
                logger.info(f"Event {event_id} already processed. Aborting task safely.")
                return True
            
            # 2. Transactional Update: update db state & log transaction
            if event_type == "document.signed":
                cur.execute(
                    """
                    UPDATE client_contracts 
                    SET status = 'SIGNED', signed_at = %s, signer = %s 
                    WHERE document_id = %s;
                    """,
                    (signed_at, signer_email, document_id)
                )
                
                # Mark as processed in the same database transaction
                cur.execute(
                    "INSERT INTO processed_webhooks (id, event_type) VALUES (%s, %s);",
                    (event_id, event_type)
                )
                conn.commit()
                logger.info(f"Contract database sync successful for document {document_id}")
                
                # Schedule downstream workflow task asynchronously
                trigger_post_signing_actions.delay(document_id, signer_email)
                
    return True

@app.task
def trigger_post_signing_actions(document_id, email):
    # Triggers emails, ERP updates, sales notification CRM pipelines
    logger.info(f"Executing downstream automation for contract {document_id} signed by {email}")