Webhook Development & Debugging

import crypto from 'crypto';

function verifyWebhookSignature(rawBody, signature, secret) {
  if (!signature || !secret) {
    return false;
  }

  // Extract hash from signature header
  const [algorithm, hash] = signature.split('=');
  if (algorithm !== 'sha256') {
    return false;
  }

  // Compute expected signature
  const expectedSignature = crypto
    .createHmac('sha256', secret)
    .update(rawBody)
    .digest('hex');

  // Verify signature using timing-safe comparison
  try {
    return crypto.timingSafeEqual(
      Buffer.from(hash, 'hex'),
      Buffer.from(expectedSignature, 'hex')
    );
  } catch {
    return false;
  }
}

export { verifyWebhookSignature };

import hmac
import hashlib
from typing import Optional

def verify_webhook_signature(
    raw_body: bytes,
    signature: Optional[str],
    secret: Optional[str]
) -> bool:
    if not signature or not secret:
        return False

    # Extract hash from signature header
    try:
        algorithm, hash_value = signature.split('=')
        if algorithm != 'sha256':
            return False
    except ValueError:
        return False

    # Compute expected signature
    expected_signature = hmac.new(
        secret.encode('utf-8'),
        raw_body,
        hashlib.sha256
    ).hexdigest()

    # Verify signature using timing-safe comparison
    return hmac.compare_digest(hash_value, expected_signature)

package webhook

import (
    "crypto/hmac"
    "crypto/sha256"
    "crypto/subtle"
    "encoding/hex"
    "strings"
)

// VerifyWebhookSignature verifies the webhook signature
func VerifyWebhookSignature(rawBody []byte, signature, secret string) bool {
    if signature == "" || secret == "" {
        return false
    }

    // Extract hash from signature header
    parts := strings.SplitN(signature, "=", 2)
    if len(parts) != 2 || parts[0] != "sha256" {
        return false
    }
    hash := parts[1]

    // Compute expected signature
    mac := hmac.New(sha256.New, []byte(secret))
    mac.Write(rawBody)
    expectedSignature := hex.EncodeToString(mac.Sum(nil))

    // Verify signature using timing-safe comparison
    return subtle.ConstantTimeCompare(
        []byte(hash),
        []byte(expectedSignature),
    ) == 1
}

import express from 'express';
import { verifyWebhookSignature } from './webhook-utils.js';

const app = express();

// Validate required environment variables
if (!process.env.SENT_WEBHOOK_SECRET) {
  throw new Error('SENT_WEBHOOK_SECRET environment variable is required');
}

// Use raw body parser for signature verification
app.use('/webhooks/sent', express.raw({ type: 'application/json' }));

app.post('/webhooks/sent', (req, res) => {
  // Verify signature
  const signature = req.get('x-webhook-signature');
  const webhookSecret = process.env.SENT_WEBHOOK_SECRET;

  if (!verifyWebhookSignature(req.body, signature, webhookSecret)) {
    return res.status(401).send('Unauthorized');
  }

  // Parse verified webhook
  const event = JSON.parse(req.body);
  console.log(`Received: ${event.field}`);
  console.log("Payload:", JSON.stringify(event.payload, null, 2));

  // Process based on event type
  if (event.field === 'messages') {
    handleMessageEvent(event.payload);
  } else if (event.field === 'templates') {
    handleTemplateEvent(event.payload);
  } else {
    console.log(`Unknown event type: ${event.field}`);
  }

  res.sendStatus(200);
});

function handleMessageEvent(payload) {
  const status = payload.message_status;
  console.log(`Message ${payload.message_id}: ${status}`);

  // Your business logic here
  if (status === 'DELIVERED') {
    // Update order status, send notification, etc.
  }
}

function handleTemplateEvent(payload) {
  console.log(`Template ${payload.template_id}: ${payload.status}`);

  // Your business logic here
  if (payload.status === 'approved') {
    // Enable template for use, notify user, etc.
  }
}

app.listen(3000);

import os
from flask import Flask, request
from webhook_utils import verify_webhook_signature

app = Flask(__name__)

# Validate required environment variables
if not os.environ.get('SENT_WEBHOOK_SECRET'):
    raise ValueError('SENT_WEBHOOK_SECRET environment variable is required')

@app.route('/webhooks/sent', methods=['POST'])
def handle_webhook():
    # Verify signature
    signature = request.headers.get('x-webhook-signature')
    webhook_secret = os.environ.get('SENT_WEBHOOK_SECRET')
    raw_body = request.get_data()

    if not verify_webhook_signature(raw_body, signature, webhook_secret):
        return 'Unauthorized', 401

    # Parse verified webhook
    event = request.get_json()
    print(f"Received: {event['field']}")
    print(f"Payload: {event['payload']}")

    # Process based on event type
    if event['field'] == 'messages':
        handle_message_event(event['payload'])
    elif event['field'] == 'templates':
        handle_template_event(event['payload'])
    else:
        print(f"Unknown event type: {event['field']}")

    return '', 200

def handle_message_event(payload):
    status = payload['message_status']
    print(f"Message {payload['message_id']}: {status}")

    # Your business logic here
    if status == 'DELIVERED':
        # Update order status, send notification, etc.
        pass

def handle_template_event(payload):
    print(f"Template {payload['template_id']}: {payload['status']}")

    # Your business logic here
    if payload['status'] == 'approved':
        # Enable template for use, notify user, etc.
        pass

if __name__ == '__main__':
    app.run(port=3000)

package main

import (
    "encoding/json"
    "io"
    "log"
    "net/http"
    "os"

    "yourapp/webhook"
)

type WebhookEvent struct {
    Field     string                 `json:"field"`
    Payload   map[string]interface{} `json:"payload"`
    Timestamp string                 `json:"timestamp"`
}

func webhookHandler(w http.ResponseWriter, r *http.Request) {
    // Read and verify signature
    signature := r.Header.Get("x-webhook-signature")
    webhookSecret := os.Getenv("SENT_WEBHOOK_SECRET")

    body, err := io.ReadAll(r.Body)
    if err != nil {
        http.Error(w, "Error reading body", http.StatusBadRequest)
        return
    }

    if !webhook.VerifyWebhookSignature(body, signature, webhookSecret) {
        http.Error(w, "Unauthorized", http.StatusUnauthorized)
        return
    }

    // Parse verified webhook
    var event WebhookEvent
    if err := json.Unmarshal(body, &event); err != nil {
        http.Error(w, "Invalid JSON", http.StatusBadRequest)
        return
    }

    log.Printf("Received: %s", event.Field)
    log.Printf("Payload: %+v", event.Payload)

    // Process based on event type
    switch event.Field {
    case "messages":
        handleMessageEvent(event.Payload)
    case "templates":
        handleTemplateEvent(event.Payload)
    default:
        log.Printf("Unknown event type: %s", event.Field)
    }

    w.WriteHeader(http.StatusOK)
}

func handleMessageEvent(payload map[string]interface{}) {
    status := payload["message_status"].(string)
    messageID := payload["message_id"].(string)
    log.Printf("Message %s: %s", messageID, status)

    // Your business logic here
    if status == "DELIVERED" {
        // Update order status, send notification, etc.
    }
}

func handleTemplateEvent(payload map[string]interface{}) {
    status := payload["status"].(string)
    templateID := payload["template_id"].(string)
    log.Printf("Template %s: %s", templateID, status)

    // Your business logic here
    if status == "approved" {
        // Enable template for use, notify user, etc.
    }
}

func main() {
    // Validate required environment variables
    if os.Getenv("SENT_WEBHOOK_SECRET") == "" {
        log.Fatal("SENT_WEBHOOK_SECRET environment variable is required")
    }

    http.HandleFunc("/webhooks/sent", webhookHandler)
    log.Println("Webhook server listening on port 3000")
    log.Fatal(http.ListenAndServe(":3000", nil))
}

async function handleWebhook(eventData) {
  const { id: eventId, field, payload, timestamp } = eventData;

  // Check if event already processed
  const existing = await db.webhookEvents.findUnique({
    where: { eventId },
  });

  if (existing) {
    console.log(`Event ${eventId} already processed`);
    return; // Idempotent - already handled
  }

  // Use transaction to ensure atomicity
  await db.$transaction(async (tx) => {
    // Record that we're processing this event
    await tx.webhookEvents.create({
      data: {
        eventId,
        eventType: field,
        payload: JSON.stringify(payload),
        processedAt: new Date(),
        status: "processing",
      },
    });

    // Process the actual business logic
    if (field === 'messages') {
      await handleMessageEvent(payload, tx);
    } else if (field === 'templates') {
      await handleTemplateEvent(payload, tx);
    }

    // Mark as completed
    await tx.webhookEvents.update({
      where: { eventId },
      data: { status: "completed" },
    });
  });
}

from sqlalchemy.orm import Session
from sqlalchemy.exc import IntegrityError
from datetime import datetime

async def handle_webhook(event_data: dict, db: Session):
    event_id = event_data.get('id')
    field = event_data.get('field')
    payload = event_data.get('payload')

    # Check if event already processed
    existing = db.query(WebhookEvent).filter_by(event_id=event_id).first()

    if existing:
        print(f"Event {event_id} already processed")
        return  # Idempotent - already handled

    # Use transaction to ensure atomicity
    try:
        # Record that we're processing this event
        webhook_event = WebhookEvent(
            event_id=event_id,
            event_type=field,
            payload=json.dumps(payload),
            processed_at=datetime.now(),
            status="processing"
        )
        db.add(webhook_event)
        db.flush()

        # Process the actual business logic
        if field == 'messages':
            await handle_message_event(payload, db)
        elif field == 'templates':
            await handle_template_event(payload, db)

        # Mark as completed
        webhook_event.status = "completed"
        db.commit()
    except IntegrityError:
        db.rollback()
        print(f"Event {event_id} already processed (race condition)")
    except Exception as e:
        db.rollback()
        raise e

func handleWebhook(ctx context.Context, eventData WebhookEvent, db *sql.DB) error {
    eventID := eventData.ID

    // Check if event already processed
    var exists bool
    err := db.QueryRowContext(ctx,
        "SELECT EXISTS(SELECT 1 FROM webhook_events WHERE event_id = $1)",
        eventID,
    ).Scan(&exists)

    if err != nil {
        return err
    }

    if exists {
        log.Printf("Event %s already processed", eventID)
        return nil // Idempotent - already handled
    }

    // Use transaction to ensure atomicity
    tx, err := db.BeginTx(ctx, nil)
    if err != nil {
        return err
    }
    defer tx.Rollback()

    // Record that we're processing this event
    payloadJSON, _ := json.Marshal(eventData.Payload)
    _, err = tx.ExecContext(ctx,
        "INSERT INTO webhook_events (event_id, event_type, payload, processed_at, status) VALUES ($1, $2, $3, $4, $5)",
        eventID, eventData.Field, payloadJSON, time.Now(), "processing",
    )
    if err != nil {
        return err
    }

    // Process the actual business logic
    if eventData.Field == "messages" {
        if err := handleMessageEvent(eventData.Payload, tx); err != nil {
            return err
        }
    } else if eventData.Field == "templates" {
        if err := handleTemplateEvent(eventData.Payload, tx); err != nil {
            return err
        }
    }

    // Mark as completed
    _, err = tx.ExecContext(ctx,
        "UPDATE webhook_events SET status = $1 WHERE event_id = $2",
        "completed", eventID,
    )
    if err != nil {
        return err
    }

    return tx.Commit()
}

import { Queue, Worker } from 'bullmq';

const webhookQueue = new Queue("webhook-processing", {
  connection: { host: "localhost", port: 6379 },
});

// Add to queue
app.post("/webhooks/sent", express.json(), async (req, res) => {
  res.sendStatus(200);

  await webhookQueue.add("process-event", req.body, {
    attempts: 3,
    backoff: {
      type: "exponential",
      delay: 5000,
    },
  });
});

// Process from queue
const worker = new Worker(
  "webhook-processing",
  async (job) => {
    const eventData = job.data;
    await processWebhookEvent(eventData);
  },
  { connection: { host: "localhost", port: 6379 } }
);

from celery import Celery
from kombu import Exchange, Queue

# Configure Celery
celery_app = Celery('webhook_processor', broker='redis://localhost:6379/0')

celery_app.conf.task_routes = {
    'process_webhook_event': {'queue': 'webhook-processing'}
}

celery_app.conf.task_queues = (
    Queue('webhook-processing', Exchange('webhook-processing'), routing_key='webhook'),
)

# Add to queue
@app.route('/webhooks/sent', methods=['POST'])
def handle_webhook():
    event_data = request.json

    # Queue with retry configuration
    process_webhook_event.apply_async(
        args=[event_data],
        retry=True,
        retry_policy={
            'max_retries': 3,
            'interval_start': 5,
            'interval_step': 5,
            'interval_max': 15,
        }
    )

    return '', 200

# Process from queue
@celery_app.task(bind=True, max_retries=3)
def process_webhook_event(self, event_data):
    try:
        # Process webhook event
        handle_event(event_data)
    except Exception as exc:
        raise self.retry(exc=exc, countdown=5)

import (
    "github.com/gomodule/redigo/redis"
    "encoding/json"
)

// Queue configuration
type WebhookQueue struct {
    pool *redis.Pool
}

func NewWebhookQueue() *WebhookQueue {
    return &WebhookQueue{
        pool: &redis.Pool{
            MaxIdle: 10,
            Dial: func() (redis.Conn, error) {
                return redis.Dial("tcp", "localhost:6379")
            },
        },
    }
}

// Add to queue
func webhookHandler(w http.ResponseWriter, r *http.Request) {
    var event WebhookEvent
    json.NewDecoder(r.Body).Decode(&event)

    w.WriteHeader(http.StatusOK)

    // Add to Redis queue
    conn := queue.pool.Get()
    defer conn.Close()

    data, _ := json.Marshal(event)
    conn.Do("LPUSH", "webhook-processing", data)
}

// Process from queue
func (q *WebhookQueue) ProcessWorker() {
    conn := q.pool.Get()
    defer conn.Close()

    for {
        reply, err := redis.ByteSlices(conn.Do("BRPOP", "webhook-processing", 0))
        if err != nil {
            log.Printf("Queue error: %v", err)
            continue
        }

        var event WebhookEvent
        if err := json.Unmarshal(reply[1], &event); err != nil {
            log.Printf("Unmarshal error: %v", err)
            continue
        }

        if err := processWebhookEvent(event); err != nil {
            log.Printf("Processing error: %v", err)
            // Re-queue with retry logic
            conn.Do("LPUSH", "webhook-processing:retry", reply[1])
        }
    }
}

app.post("/webhooks/sent", express.json(), (req, res) => {
  try {
    processEvent(req.body);
    res.sendStatus(200); // Success
  } catch (error) {
    if (error.code === "TEMPORARY_FAILURE") {
      // Database connection lost, service unavailable, etc.
      res.sendStatus(500); // Retry
    } else if (error.code === "INVALID_DATA") {
      // Malformed event data, business logic error
      console.error("Permanent failure:", error);
      res.sendStatus(200); // Don't retry
    } else {
      // Unknown error - safe to retry
      res.sendStatus(500);
    }
  }
});

@app.route('/webhooks/sent', methods=['POST'])
def handle_webhook():
    try:
        process_event(request.json)
        return '', 200  # Success
    except Exception as error:
        if hasattr(error, 'code') and error.code == "TEMPORARY_FAILURE":
            # Database connection lost, service unavailable, etc.
            return '', 500  # Retry
        elif hasattr(error, 'code') and error.code == "INVALID_DATA":
            # Malformed event data, business logic error
            print(f"Permanent failure: {error}")
            return '', 200  # Don't retry
        else:
            # Unknown error - safe to retry
            return '', 500

func webhookHandler(w http.ResponseWriter, r *http.Request) {
    var event WebhookEvent
    json.NewDecoder(r.Body).Decode(&event)

    if err := processEvent(event); err != nil {
        switch err.(type) {
        case *TemporaryFailureError:
            // Database connection lost, service unavailable, etc.
            w.WriteHeader(http.StatusInternalServerError) // Retry
        case *InvalidDataError:
            // Malformed event data, business logic error
            log.Printf("Permanent failure: %v", err)
            w.WriteHeader(http.StatusOK) // Don't retry
        default:
            // Unknown error - safe to retry
            w.WriteHeader(http.StatusInternalServerError)
        }
        return
    }

    w.WriteHeader(http.StatusOK) // Success
}