Example: handle webhook events

import express from "express";
import { Webhook } from "svix";

const app = express();
const wh = new Webhook(process.env.WAZAPIN_WEBHOOK_SECRET!);

app.post("/webhooks/wazapin", express.raw({ type: "application/json" }), async (req, res) => {
  // 1. Verify webhook signature
  try {
    wh.verify(req.body, req.headers as Record<string, string>);
  } catch (err) {
    return res.status(403).send("Invalid signature");
  }

  const payload = JSON.parse(req.body.toString("utf8"));
  const eventId = req.headers["svix-id"] as string;

  // 2. Deduplicate using event ID
  if (await isAlreadyProcessed(eventId)) {
    return res.status(200).send("Duplicate acknowledged");
  }

  // 3. Acknowledge receipt quickly (within 3 seconds)
  res.status(200).send("OK");

  // 4. Process event asynchronously (outside request cycle)
  processEventAsync(payload, eventId).catch(console.error);
});

async function isAlreadyProcessed(id: string): Promise<boolean> {
  // Check your database/cache (e.g. Redis) here
  return false; 
}

async function processEventAsync(payload: any, eventId: string) {
  // Mark event as processed in your store to ensure idempotency
  await markAsProcessed(eventId);

  // Route by event properties
  if (payload.status) {
    // Delivery status update receipt
    return handleDeliveryStatus(payload);
  }

  if (payload.direction === "inbound") {
    switch (payload.msg_type) {
      case "text":
        return handleInboundText(payload);
      case "image":
      case "video":
      case "audio":
      case "document":
      case "sticker":
        return handleInboundMedia(payload);
      case "interactive":
        return handleInteractiveReplies(payload);
      default:
        console.log("Unhandled inbound message type:", payload.msg_type);
    }
  }
}

// Stubs for task guides
async function handleInboundText(payload: any) { /* See guide */ }
async function handleInboundMedia(payload: any) { /* See guide */ }
async function handleInteractiveReplies(payload: any) { /* See guide */ }
async function handleDeliveryStatus(payload: any) { /* See guide */ }
async function markAsProcessed(id: string) { /* Save state */ }

from fastapi import FastAPI, Request, HTTPException, BackgroundTasks
from svix.webhooks import Webhook, WebhookVerificationError
import os
import json

app = FastAPI()
wh = Webhook(os.environ["WAZAPIN_WEBHOOK_SECRET"])

@app.post("/webhooks/wazapin")
async def handle_webhook(request: Request, background_tasks: BackgroundTasks):
    body = await request.body()
    headers = dict(request.headers)
    event_id = headers.get("svix-id")

    # 1. Verify signature
    try:
        wh.verify(body, headers)
    except WebhookVerificationError:
        raise HTTPException(status_code=403, detail="Invalid signature")

    # 2. Check for duplicates
    if is_already_processed(event_id):
        return {"ok": True, "detail": "Duplicate"}

    payload = json.loads(body.decode("utf-8"))

    # 3. Schedule task to process asynchronously & return 200 OK
    background_tasks.add_task(process_event_async, payload, event_id)
    return {"ok": True}

def is_already_processed(event_id: str) -> bool:
    # Check Redis/DB here
    return False

def process_event_async(payload: dict, event_id: str):
    mark_as_processed(event_id)
    
    # Route event
    status = payload.get("status")
    direction = payload.get("direction")
    msg_type = payload.get("msg_type")

    if status:
        handle_delivery_status(payload)
    elif direction == "inbound":
        if msg_type == "text":
            handle_inbound_text(payload)
        elif msg_type in {"image", "video", "audio", "document", "sticker"}:
            handle_inbound_media(payload)
        elif msg_type == "interactive":
            handle_interactive_replies(payload)

def handle_inbound_text(payload): pass
def handle_inbound_media(payload): pass
def handle_interactive_replies(payload): pass
def handle_delivery_status(payload): pass
def mark_as_processed(event_id): pass