Next.js Webhook Handling and Event-Driven Architecture

Building robust webhook handling in Next.js enables real-time responsiveness and event-driven architecture. This guide covers webhook security, retry mechanisms, event processing, and distributed system patterns for production applications.

TL;DR

Build production-ready event-driven systems using webhooks with Next.js and Supabase. You'll implement secure webhook endpoints, reliable event processing, retry mechanisms, and distributed system patterns for scalable applications.

Prerequisites

Next.js 14+ with API Routes experience
Supabase database and Edge Functions knowledge
Understanding of HTTP protocols and security
Basic knowledge of distributed systems concepts

Problem Statement

Modern applications need to react to events from multiple sources: payment processors, third-party APIs, user actions, and system events. Simple polling is inefficient and doesn't scale. Event-driven architecture with webhooks provides real-time responsiveness but introduces complexity around security, reliability, and error handling.

What Are Webhooks and How Do They Work?

Webhooks push data to your application when events occur, while polling requires your app to repeatedly check for changes. Webhooks are more efficient, provide real-time updates, and reduce server load, but require proper security and error handling.

How Do You Secure Webhook Endpoints in Next.js?

Verify webhook signatures using HMAC, validate the source IP if possible, use HTTPS only, implement rate limiting, and validate payload structure. Never trust webhook data without verification - treat it as potentially malicious input.

What Happens When Webhook Processing Fails?

Implement retry mechanisms with exponential backoff, use dead letter queues for failed events, monitor webhook health, and consider using message queues for reliability. Most webhook providers retry failed deliveries automatically, so your endpoint should handle duplicate events gracefully.

Step-by-Step Walkthrough

1. Webhook Infrastructure Setup

Create a robust webhook handling system:

-- Webhook events tracking table
CREATE TABLE webhook_events (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  event_id TEXT UNIQUE NOT NULL, -- External event ID for idempotency
  source TEXT NOT NULL, -- 'stripe', 'github', 'supabase', etc.
  event_type TEXT NOT NULL,
  payload JSONB NOT NULL,
  status TEXT NOT NULL DEFAULT 'pending', -- 'pending', 'processing', 'completed', 'failed'
  attempts INTEGER DEFAULT 0,
  last_attempt_at TIMESTAMPTZ,
  completed_at TIMESTAMPTZ,
  error_message TEXT,
  created_at TIMESTAMPTZ DEFAULT NOW()
);

-- Event processing jobs queue
CREATE TABLE event_jobs (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  webhook_event_id UUID REFERENCES webhook_events(id) ON DELETE CASCADE,
  job_type TEXT NOT NULL,
  payload JSONB NOT NULL,
  status TEXT NOT NULL DEFAULT 'queued', -- 'queued', 'processing', 'completed', 'failed'
  scheduled_for TIMESTAMPTZ DEFAULT NOW(),
  attempts INTEGER DEFAULT 0,
  max_attempts INTEGER DEFAULT 3,
  created_at TIMESTAMPTZ DEFAULT NOW(),
  updated_at TIMESTAMPTZ DEFAULT NOW()
);

-- Indexes for performance
CREATE INDEX idx_webhook_events_status ON webhook_events(status);
CREATE INDEX idx_webhook_events_source_type ON webhook_events(source, event_type);
CREATE INDEX idx_event_jobs_status_scheduled ON event_jobs(status, scheduled_for);

2. Secure Webhook Endpoint

Implement a secure, generic webhook handler:

// app/api/webhooks/[source]/route.ts
import { NextRequest, NextResponse } from 'next/server'
import { createClient } from '@/lib/supabase/server'
import { verifyWebhookSignature } from '@/lib/webhook-security'
import { queueEventJob } from '@/lib/event-queue'

interface WebhookConfig {
  secretKey: string
  signatureHeader: string
  verifySignature: (payload: string, signature: string, secret: string) => boolean
}

const WEBHOOK_CONFIGS: Record<string, WebhookConfig> = {
  stripe: {
    secretKey: process.env.STRIPE_WEBHOOK_SECRET!,
    signatureHeader: 'stripe-signature',
    verifySignature: verifyStripeSignature
  },
  github: {
    secretKey: process.env.GITHUB_WEBHOOK_SECRET!,
    signatureHeader: 'x-hub-signature-256',
    verifySignature: verifyGitHubSignature
  },
  supabase: {
    secretKey: process.env.SUPABASE_WEBHOOK_SECRET!,
    signatureHeader: 'x-supabase-signature',
    verifySignature: verifySupabaseSignature
  }
}

export async function POST(
  request: NextRequest,
  { params }: { params: { source: string } }
) {
  const source = params.source
  const config = WEBHOOK_CONFIGS[source]

  if (!config) {
    return NextResponse.json(
      { error: 'Unknown webhook source' },
      { status: 400 }
    )
  }

  try {
    // Get raw body for signature verification
    const body = await request.text()
    const signature = request.headers.get(config.signatureHeader)

    if (!signature) {
      return NextResponse.json(
        { error: 'Missing signature' },
        { status: 400 }
      )
    }

    // Verify webhook signature
    if (!config.verifySignature(body, signature, config.secretKey)) {
      return NextResponse.json(
        { error: 'Invalid signature' },
        { status: 401 }
      )
    }

    const payload = JSON.parse(body)
    const eventId = extractEventId(source, payload)
    const eventType = extractEventType(source, payload)

    // Store webhook event with idempotency check
    const supabase = createClient()
    const { data: existingEvent } = await supabase
      .from('webhook_events')
      .select('id, status')
      .eq('event_id', eventId)
      .single()

    if (existingEvent) {
      // Event already processed or processing
      return NextResponse.json({ 
        message: 'Event already processed',
        status: existingEvent.status 
      })
    }

    // Store new webhook event
    const { data: webhookEvent, error } = await supabase
      .from('webhook_events')
      .insert({
        event_id: eventId,
        source,
        event_type: eventType,
        payload,
        status: 'pending'
      })
      .select()
      .single()

    if (error) throw error

    // Queue processing job
    await queueEventJob(webhookEvent.id, eventType, payload)

    return NextResponse.json({ 
      message: 'Webhook received',
      event_id: eventId 
    })

  } catch (error) {
    console.error('Webhook processing error:', error)
    return NextResponse.json(
      { error: 'Internal server error' },
      { status: 500 }
    )
  }
}

// Helper functions for different webhook sources
function extractEventId(source: string, payload: any): string {
  switch (source) {
    case 'stripe':
      return payload.id
    case 'github':
      return payload.delivery || `${payload.repository?.id}-${Date.now()}`
    case 'supabase':
      return payload.record?.id || `${payload.table}-${Date.now()}`
    default:
      return `${source}-${Date.now()}`
  }
}

function extractEventType(source: string, payload: any): string {
  switch (source) {
    case 'stripe':
      return payload.type
    case 'github':
      return payload.action ? `${payload.action}` : 'unknown'
    case 'supabase':
      return `${payload.type}.${payload.table}`
    default:
      return 'unknown'
  }
}

3. Webhook Signature Verification

Implement secure signature verification for different providers:

// lib/webhook-security.ts
import crypto from 'crypto'

export function verifyStripeSignature(
  payload: string,
  signature: string,
  secret: string
): boolean {
  const elements = signature.split(',')
  const signatureElements = elements.reduce((acc, element) => {
    const [key, value] = element.split('=')
    acc[key] = value
    return acc
  }, {} as Record<string, string>)

  const timestamp = signatureElements.t
  const signatures = [signatureElements.v1]

  const expectedSignature = crypto
    .createHmac('sha256', secret)
    .update(`${timestamp}.${payload}`)
    .digest('hex')

  return signatures.some(sig => 
    crypto.timingSafeEqual(
      Buffer.from(sig, 'hex'),
      Buffer.from(expectedSignature, 'hex')
    )
  )
}

export function verifyGitHubSignature(
  payload: string,
  signature: string,
  secret: string
): boolean {
  const expectedSignature = crypto
    .createHmac('sha256', secret)
    .update(payload)
    .digest('hex')

  const actualSignature = signature.replace('sha256=', '')

  return crypto.timingSafeEqual(
    Buffer.from(actualSignature, 'hex'),
    Buffer.from(expectedSignature, 'hex')
  )
}

export function verifySupabaseSignature(
  payload: string,
  signature: string,
  secret: string
): boolean {
  const expectedSignature = crypto
    .createHmac('sha256', secret)
    .update(payload)
    .digest('base64')

  return crypto.timingSafeEqual(
    Buffer.from(signature),
    Buffer.from(expectedSignature)
  )
}

4. Event Processing Queue System

Build a reliable job queue for processing webhook events:

// lib/event-queue.ts
import { createClient } from '@/lib/supabase/server'

export async function queueEventJob(
  webhookEventId: string,
  jobType: string,
  payload: any,
  scheduledFor?: Date
) {
  const supabase = createClient()

  const { error } = await supabase
    .from('event_jobs')
    .insert({
      webhook_event_id: webhookEventId,
      job_type: jobType,
      payload,
      scheduled_for: scheduledFor?.toISOString() || new Date().toISOString()
    })

  if (error) throw error
}

export async function processEventJobs() {
  const supabase = createClient()

  // Get pending jobs
  const { data: jobs, error } = await supabase
    .from('event_jobs')
    .select('*')
    .eq('status', 'queued')
    .lte('scheduled_for', new Date().toISOString())
    .order('created_at', { ascending: true })
    .limit(10)

  if (error) throw error

  for (const job of jobs) {
    await processJob(job)
  }
}

async function processJob(job: any) {
  const supabase = createClient()

  try {
    // Mark job as processing
    await supabase
      .from('event_jobs')
      .update({ 
        status: 'processing',
        updated_at: new Date().toISOString()
      })
      .eq('id', job.id)

    // Process based on job type
    await processJobByType(job.job_type, job.payload)

    // Mark as completed
    await supabase
      .from('event_jobs')
      .update({ 
        status: 'completed',
        updated_at: new Date().toISOString()
      })
      .eq('id', job.id)

    // Mark webhook event as completed
    await supabase
      .from('webhook_events')
      .update({ 
        status: 'completed',
        completed_at: new Date().toISOString()
      })
      .eq('id', job.webhook_event_id)

  } catch (error) {
    console.error('Job processing failed:', error)

    const newAttempts = job.attempts + 1
    const shouldRetry = newAttempts < job.max_attempts

    if (shouldRetry) {
      // Schedule retry with exponential backoff
      const retryDelay = Math.pow(2, newAttempts) * 1000 // 2s, 4s, 8s...
      const scheduledFor = new Date(Date.now() + retryDelay)

      await supabase
        .from('event_jobs')
        .update({
          status: 'queued',
          attempts: newAttempts,
          scheduled_for: scheduledFor.toISOString(),
          updated_at: new Date().toISOString()
        })
        .eq('id', job.id)
    } else {
      // Mark as failed
      await supabase
        .from('event_jobs')
        .update({
          status: 'failed',
          attempts: newAttempts,
          updated_at: new Date().toISOString()
        })
        .eq('id', job.id)

      await supabase
        .from('webhook_events')
        .update({
          status: 'failed',
          error_message: error instanceof Error ? error.message : 'Unknown error'
        })
        .eq('id', job.webhook_event_id)
    }
  }
}

async function processJobByType(jobType: string, payload: any) {
  switch (jobType) {
    case 'payment.succeeded':
      await handlePaymentSucceeded(payload)
      break
    case 'user.created':
      await handleUserCreated(payload)
      break
    case 'subscription.updated':
      await handleSubscriptionUpdated(payload)
      break
    default:
      console.warn(`Unknown job type: ${jobType}`)
  }
}

5. Event Handlers Implementation

Create specific handlers for different event types:

// lib/event-handlers.ts
import { createClient } from '@/lib/supabase/server'
import { sendEmail } from '@/lib/email'
import { updateUserPlan } from '@/lib/billing'

export async function handlePaymentSucceeded(payload: any) {
  const supabase = createClient()

  const { customer, amount, currency, metadata } = payload.data.object

  // Update user's payment status
  const { error } = await supabase
    .from('payments')
    .insert({
      stripe_payment_id: payload.data.object.id,
      customer_id: customer,
      amount,
      currency,
      status: 'succeeded',
      metadata
    })

  if (error) throw error

  // Send confirmation email
  await sendEmail({
    to: metadata.user_email,
    template: 'payment-confirmation',
    data: { amount, currency }
  })

  // Update user plan if this was a subscription payment
  if (metadata.subscription_id) {
    await updateUserPlan(metadata.user_id, metadata.plan_id)
  }
}

export async function handleUserCreated(payload: any) {
  const supabase = createClient()
  const { user } = payload

  // Create user profile
  const { error } = await supabase
    .from('user_profiles')
    .insert({
      user_id: user.id,
      email: user.email,
      created_at: user.created_at
    })

  if (error) throw error

  // Send welcome email
  await sendEmail({
    to: user.email,
    template: 'welcome',
    data: { name: user.user_metadata?.name || 'there' }
  })

  // Create default workspace
  await supabase
    .from('workspaces')
    .insert({
      name: 'My Workspace',
      owner_id: user.id
    })
}

export async function handleSubscriptionUpdated(payload: any) {
  const supabase = createClient()
  const subscription = payload.data.object

  // Update subscription in database
  const { error } = await supabase
    .from('subscriptions')
    .upsert({
      stripe_subscription_id: subscription.id,
      customer_id: subscription.customer,
      status: subscription.status,
      current_period_start: new Date(subscription.current_period_start * 1000),
      current_period_end: new Date(subscription.current_period_end * 1000),
      plan_id: subscription.items.data[0]?.price?.id
    })

  if (error) throw error

  // Handle subscription status changes
  if (subscription.status === 'canceled') {
    await handleSubscriptionCanceled(subscription)
  } else if (subscription.status === 'past_due') {
    await handleSubscriptionPastDue(subscription)
  }
}

async function handleSubscriptionCanceled(subscription: any) {
  const supabase = createClient()

  // Downgrade user to free plan
  const { data: user } = await supabase
    .from('users')
    .select('id, email')
    .eq('stripe_customer_id', subscription.customer)
    .single()

  if (user) {
    await updateUserPlan(user.id, 'free')

    await sendEmail({
      to: user.email,
      template: 'subscription-canceled',
      data: { cancellation_date: new Date() }
    })
  }
}

6. Cron Job for Processing Events

Set up automated event processing:

// app/api/cron/process-events/route.ts
import { NextResponse } from 'next/server'
import { processEventJobs } from '@/lib/event-queue'

export async function GET(request: Request) {
  // Verify cron secret to prevent unauthorized access
  const authHeader = request.headers.get('authorization')
  if (authHeader !== `Bearer ${process.env.CRON_SECRET}`) {
    return NextResponse.json({ error: 'Unauthorized' }, { status: 401 })
  }

  try {
    await processEventJobs()
    return NextResponse.json({ message: 'Events processed successfully' })
  } catch (error) {
    console.error('Event processing failed:', error)
    return NextResponse.json(
      { error: 'Event processing failed' },
      { status: 500 }
    )
  }
}

7. Event Monitoring and Observability

Implement comprehensive monitoring:

// lib/webhook-monitoring.ts
import { createClient } from '@/lib/supabase/server'

export async function getWebhookMetrics(timeframe: 'hour' | 'day' | 'week' = 'day') {
  const supabase = createClient()

  const timeframeSql = {
    hour: "created_at >= NOW() - INTERVAL '1 hour'",
    day: "created_at >= NOW() - INTERVAL '1 day'",
    week: "created_at >= NOW() - INTERVAL '1 week'"
  }

  // Get event counts by status
  const { data: statusCounts } = await supabase
    .from('webhook_events')
    .select('status, count(*)')
    .gte('created_at', getTimeframeStart(timeframe))
    .group('status')

  // Get event counts by source
  const { data: sourceCounts } = await supabase
    .from('webhook_events')
    .select('source, count(*)')
    .gte('created_at', getTimeframeStart(timeframe))
    .group('source')

  // Get failed events with details
  const { data: failedEvents } = await supabase
    .from('webhook_events')
    .select('*')
    .eq('status', 'failed')
    .gte('created_at', getTimeframeStart(timeframe))
    .order('created_at', { ascending: false })
    .limit(10)

  // Get average processing time
  const { data: processingTimes } = await supabase
    .rpc('get_avg_processing_time', { timeframe })

  return {
    statusCounts,
    sourceCounts,
    failedEvents,
    avgProcessingTime: processingTimes?.[0]?.avg_time || 0
  }
}

function getTimeframeStart(timeframe: string): string {
  const now = new Date()
  switch (timeframe) {
    case 'hour':
      return new Date(now.getTime() - 60 * 60 * 1000).toISOString()
    case 'day':
      return new Date(now.getTime() - 24 * 60 * 60 * 1000).toISOString()
    case 'week':
      return new Date(now.getTime() - 7 * 24 * 60 * 60 * 1000).toISOString()
    default:
      return new Date(now.getTime() - 24 * 60 * 60 * 1000).toISOString()
  }
}

// Database function for calculating processing times
/*
CREATE OR REPLACE FUNCTION get_avg_processing_time(timeframe TEXT)
RETURNS TABLE(avg_time INTERVAL) AS $$
BEGIN
  RETURN QUERY
  SELECT AVG(completed_at - created_at) as avg_time
  FROM webhook_events
  WHERE status = 'completed'
    AND created_at >= CASE
      WHEN timeframe = 'hour' THEN NOW() - INTERVAL '1 hour'
      WHEN timeframe = 'day' THEN NOW() - INTERVAL '1 day'
      WHEN timeframe = 'week' THEN NOW() - INTERVAL '1 week'
      ELSE NOW() - INTERVAL '1 day'
    END;
END;
$$ LANGUAGE plpgsql;
*/

8. Event Replay and Recovery

Implement event replay for failed or missed events:

// lib/event-replay.ts
import { createClient } from '@/lib/supabase/server'
import { queueEventJob } from '@/lib/event-queue'

export async function replayFailedEvents(
  source?: string,
  eventType?: string,
  fromDate?: Date
) {
  const supabase = createClient()

  let query = supabase
    .from('webhook_events')
    .select('*')
    .eq('status', 'failed')

  if (source) query = query.eq('source', source)
  if (eventType) query = query.eq('event_type', eventType)
  if (fromDate) query = query.gte('created_at', fromDate.toISOString())

  const { data: failedEvents, error } = await query
    .order('created_at', { ascending: true })

  if (error) throw error

  for (const event of failedEvents) {
    // Reset event status
    await supabase
      .from('webhook_events')
      .update({ status: 'pending' })
      .eq('id', event.id)

    // Queue new processing job
    await queueEventJob(event.id, event.event_type, event.payload)
  }

  return { replayedCount: failedEvents.length }
}

export async function replayEventById(eventId: string) {
  const supabase = createClient()

  const { data: event, error } = await supabase
    .from('webhook_events')
    .select('*')
    .eq('id', eventId)
    .single()

  if (error) throw error

  // Reset status and queue for processing
  await supabase
    .from('webhook_events')
    .update({ status: 'pending' })
    .eq('id', eventId)

  await queueEventJob(event.id, event.event_type, event.payload)

  return { message: 'Event queued for replay' }
}

Common Pitfalls

1. Not Implementing Idempotency

Problem: Processing the same webhook multiple times causes data inconsistency.
Solution: Always use unique event IDs and check for existing events before processing.

2. Synchronous Processing

Problem: Long-running webhook processing causes timeouts and retries.
Solution: Acknowledge webhooks immediately and process asynchronously with job queues.

3. Insufficient Error Handling

Problem: Failed webhooks are lost without proper retry mechanisms.
Solution: Implement exponential backoff, dead letter queues, and monitoring for failed events.

Production Considerations

Security Hardening

Always verify webhook signatures using timing-safe comparison
Implement rate limiting per source to prevent abuse
Use HTTPS only and validate SSL certificates
Log all webhook attempts for security auditing

Reliability and Scaling

Use database transactions for atomic event processing
Implement circuit breakers for external service calls
Consider using message queues (Redis, RabbitMQ) for high-volume scenarios
Set up monitoring and alerting for webhook failures

Performance Optimization

Index webhook events table on frequently queried columns
Implement connection pooling for database operations
Use batch processing for high-volume events
Consider event streaming for real-time requirements

推荐订阅源

DEV Community