Webhook Development & Debugging

// Test webhook handler with different event types
app.post("/webhooks/sent", express.json(), (req, res) => {
  console.log(`Received: ${req.body.event.type}`);
  console.log("Data:", JSON.stringify(req.body.data, null, 2));

  switch (req.body.event.type) {
    case "message.status.updated":
      handleMessageStatus(req.body.data);
      break;
    case "message.inbound.received":
      handleInboundMessage(req.body.data);
      break;
    case "template.approval.updated":
      handleTemplateApproval(req.body.data);
      break;
    default:
      console.log(`Unhandled event type: ${req.body.event.type}`);
  }

  res.sendStatus(200);
});

from flask import Flask, request

app = Flask(__name__)

@app.route('/webhooks/sent', methods=['POST'])
def handle_webhook():
    event_data = request.json
    event_type = event_data['event']['type']

    print(f"Received: {event_type}")
    print(f"Data: {event_data['data']}")

    if event_type == "message.status.updated":
        handle_message_status(event_data['data'])
    elif event_type == "message.inbound.received":
        handle_inbound_message(event_data['data'])
    elif event_type == "template.approval.updated":
        handle_template_approval(event_data['data'])
    else:
        print(f"Unhandled event type: {event_type}")

    return '', 200

func webhookHandler(w http.ResponseWriter, r *http.Request) {
    var event WebhookEvent
    if err := json.NewDecoder(r.Body).Decode(&event); err != nil {
        http.Error(w, err.Error(), http.StatusBadRequest)
        return
    }

    log.Printf("Received: %s", event.Event.Type)
    log.Printf("Data: %+v", event.Data)

    switch event.Event.Type {
    case "message.status.updated":
        handleMessageStatus(event.Data)
    case "message.inbound.received":
        handleInboundMessage(event.Data)
    case "template.approval.updated":
        handleTemplateApproval(event.Data)
    default:
        log.Printf("Unhandled event type: %s", event.Event.Type)
    }

    w.WriteHeader(http.StatusOK)
}

const processedEvents = new Set(); // Use Redis/database in production

app.post("/webhooks/sent", express.json(), (req, res) => {
  const eventId = req.body.id;

  // Check if already processed
  if (processedEvents.has(eventId)) {
    console.log(`Duplicate event ${eventId}, skipping`);
    return res.sendStatus(200);
  }

  // Process the event
  try {
    processWebhookEvent(req.body);
    processedEvents.add(eventId);
    res.sendStatus(200);
  } catch (error) {
    console.error("Processing failed:", error);
    res.sendStatus(500); // This will trigger a retry
  }
});

processed_events = set()  # Use Redis/database in production

@app.route('/webhooks/sent', methods=['POST'])
def handle_webhook():
    event_data = request.json
    event_id = event_data['id']

    # Check if already processed
    if event_id in processed_events:
        print(f"Duplicate event {event_id}, skipping")
        return '', 200

    # Process the event
    try:
        process_webhook_event(event_data)
        processed_events.add(event_id)
        return '', 200
    except Exception as error:
        print(f"Processing failed: {error}")
        return '', 500  # This will trigger a retry

var processedEvents = make(map[string]bool) // Use Redis/database in production
var mu sync.Mutex

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

    mu.Lock()
    defer mu.Unlock()

    // Check if already processed
    if processedEvents[event.ID] {
        log.Printf("Duplicate event %s, skipping", event.ID)
        w.WriteHeader(http.StatusOK)
        return
    }

    // Process the event
    if err := processWebhookEvent(event); err != nil {
        log.Printf("Processing failed: %v", err)
        w.WriteHeader(http.StatusInternalServerError)
        return
    }

    processedEvents[event.ID] = true
    w.WriteHeader(http.StatusOK)
}

async function handleWebhook(eventData) {
  const { id: eventId, event, data } = 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: event.type,
        processedAt: new Date(),
        status: "processing",
      },
    });

    // Process the actual business logic
    await processBusinessLogic(data);

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

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

async def handle_webhook(event_data: dict, db: Session):
    event_id = event_data['id']
    event_type = event_data['event']['type']
    data = event_data['data']

    # 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=event_type,
            processed_at=datetime.now(),
            status="processing"
        )
        db.add(webhook_event)
        db.flush()

        # Process the actual business logic
        await process_business_logic(data)

        # 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
    _, err = tx.ExecContext(ctx,
        "INSERT INTO webhook_events (event_id, event_type, processed_at, status) VALUES ($1, $2, $3, $4)",
        eventID, eventData.Event.Type, time.Now(), "processing",
    )
    if err != nil {
        return err
    }

    // Process the actual business logic
    if err := processBusinessLogic(eventData.Data); 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
}