Local Development & Debugging

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

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

app.post("/webhooks/sent", (req, res) => {
  // Extract headers for signature verification
  const signature = req.get('x-webhook-signature');
  const webhookId = req.get('x-webhook-id');
  const timestamp = req.get('x-webhook-timestamp');
  const webhookSecret = process.env.SENT_WEBHOOK_SECRET;
  const rawBody = req.body.toString();

  if (!verifyWebhookSignature(rawBody, signature, webhookId, timestamp, webhookSecret)) {
    return res.status(401).send('Unauthorized');
  }

  // Parse event
  const event = JSON.parse(rawBody);

  // Immediate acknowledgment
  res.sendStatus(200);

  // Push to queue for background processing with a payload-derived
  // idempotency key (see the Handling Retries page)
  queue.add("webhook-processing", {
    eventId: idempotencyKey(event),
    field: event.field,
    payload: event.payload,
    timestamp: event.timestamp,
  });
});

from celery import Celery
from webhook_utils import verify_webhook_signature, idempotency_key

celery_app = Celery('tasks', broker='redis://localhost:6379/0')

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

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

    # Parse event
    event = request.get_json()

    # Immediate acknowledgment
    # Push to queue for background processing with a payload-derived
    # idempotency key (see the Handling Retries page)
    process_webhook.delay(
        event_id=idempotency_key(event),
        field=event['field'],
        payload=event['payload'],
        timestamp=event['timestamp']
    )

    return '', 200

import "yourapp/webhook"

// WebhookEvent maps the envelope: Field `json:"field"`, Event `json:"event"`,
// Timestamp `json:"timestamp"`, Payload `json:"payload"` (json.RawMessage)

func webhookHandler(w http.ResponseWriter, r *http.Request) {
    // Extract headers for signature verification
    signature := r.Header.Get("x-webhook-signature")
    webhookID := r.Header.Get("x-webhook-id")
    timestamp := r.Header.Get("x-webhook-timestamp")
    webhookSecret := os.Getenv("SENT_WEBHOOK_SECRET")

    body, _ := io.ReadAll(r.Body)

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

    // Parse event
    var event WebhookEvent
    json.Unmarshal(body, &event)

    // Immediate acknowledgment
    w.WriteHeader(http.StatusOK)

    // Push to queue for background processing with a payload-derived
    // idempotency key (idempotencyKey implements the Handling Retries rules)
    task := &Task{
        EventID:   idempotencyKey(event),
        Field:     event.Field,
        Payload:   event.Payload,
        Timestamp: event.Timestamp,
    }

    queue.Enqueue(task)
}

// Worker process
queue.process("webhook-processing", async (job) => {
  const { field, payload, timestamp } = job.data;

  try {
    if (field === 'message') {
      await handleMessageEvent(payload);
    } else if (field === 'templates') {
      await handleTemplateEvent(payload);
    }

    console.log(`Successfully processed ${field} event`);
  } catch (error) {
    console.error(`Failed to process ${field} event:`, error);
    throw error; // Will trigger retry based on queue configuration
  }
});

async function handleMessageEvent(payload) {
  const { message_id, message_status } = payload;

  if (message_status === 'DELIVERED') {
    // Update order status, send notification, etc.
    await updateOrderStatus(message_id, 'delivered');
  } else if (message_status === 'FAILED') {
    await notifyDeliveryFailure(message_id);
  }
}

async function handleTemplateEvent(payload) {
  const { template_id, status } = payload;

  if (status === 'APPROVED') {
    await enableTemplate(template_id);
  }
}

@celery_app.task(bind=True, max_retries=3)
def process_webhook(self, event_id, field, payload, timestamp):
    try:
        if field == 'message':
            handle_message_event(payload)
        elif field == 'templates':
            handle_template_event(payload)

        print(f"Successfully processed {field} event")
    except Exception as error:
        print(f"Failed to process {field} event: {error}")
        raise self.retry(exc=error, countdown=60)

def handle_message_event(payload):
    message_id = payload['message_id']
    message_status = payload['message_status']

    if message_status == 'DELIVERED':
        # Update order status, send notification, etc.
        update_order_status(message_id, 'delivered')
    elif message_status == 'FAILED':
        notify_delivery_failure(message_id)

def handle_template_event(payload):
    template_id = payload['template_id']
    template_name = payload['template_name']
    status = payload['status']

    if status == 'APPROVED':
        enable_template(template_id)

// Worker process
func processWebhookWorker(queue *Queue) {
    for task := range queue.Tasks {
        field := task.Field
        event := task.Event

        var err error
        switch field {
        case "message":
            err = handleMessageEvent(event)
        case "templates":
            err = handleTemplateEvent(event)
        }

        if err != nil {
            log.Printf("Failed to process %s event: %v", field, err)
            // Will trigger retry based on queue configuration
            queue.Retry(task)
        } else {
            log.Printf("Successfully processed %s event", field)
        }
    }
}

func handleMessageEvent(event WebhookEvent) error {
    var payload MessagePayload
    if err := json.Unmarshal(event.Payload, &payload); err != nil {
        return err
    }

    if payload.MessageStatus == "DELIVERED" {
        // Update order status, send notification, etc.
        return updateOrderStatus(payload.MessageID, "delivered")
    } else if payload.MessageStatus == "FAILED" {
        return notifyDeliveryFailure(payload.MessageID)
    }

    return nil
}

func handleTemplateEvent(event WebhookEvent) error {
    var payload TemplatePayload
    if err := json.Unmarshal(event.Payload, &payload); err != nil {
        return err
    }

    return syncTemplate(payload.TemplateID)
}