Use these handlers as your starting point. Each example verifies the X-Webhook-Signature header against the raw request body using your endpoint secret.
Always verify against the raw request body before parsing JSON. Any body transformation can invalidate the signature.

​
Reusable verify function

import crypto from "crypto";

export function verifyWebhook(
  rawBody: string | Buffer,
  timestampHeader: string,
  signatureHeader: string,
  secret: string,
  toleranceSeconds = 300,
) {
  const timestamp = Number(timestampHeader);
  if (!timestamp) throw new Error("Missing or invalid webhook timestamp");

  const age = Math.floor(Date.now() / 1000) - timestamp;
  if (age > toleranceSeconds) {
    throw new Error("Timestamp outside tolerance");
  }

  const body = typeof rawBody === "string" ? rawBody : rawBody.toString("utf8");
  const expected = crypto
    .createHmac("sha256", secret)
    .update(`${timestamp}.${body}`, "utf8")
    .digest("hex");

  const received = String(signatureHeader).replace(/^sha256=/i, "").trim();
  const a = Buffer.from(expected, "utf8");
  const b = Buffer.from(received, "utf8");

  if (a.length !== b.length || !crypto.timingSafeEqual(a, b)) {
    throw new Error("Invalid signature");
  }

  return JSON.parse(body);
}

​
Express (Node.js)

import express from "express";
import { verifyWebhook } from "./verify";

const app = express();

app.post(
  "/webhooks/zquence",
  express.raw({ type: "application/json" }),
  async (req, res) => {
    let event;
    try {
      event = verifyWebhook(
        req.body,
        String(req.headers["x-webhook-timestamp"] || ""),
        String(req.headers["x-webhook-signature"] || ""),
        process.env.ZQUENCE_WEBHOOK_SECRET!,
      );
    } catch {
      return res.status(400).send("Invalid webhook");
    }

    await enqueue(event);
    return res.sendStatus(200);
  },
);

​
Next.js (App Router)

import { NextResponse } from "next/server";
import { verifyWebhook } from "@/lib/verify";

export const runtime = "nodejs";
export const dynamic = "force-dynamic";

export async function POST(req: Request) {
  const body = await req.text();
  const timestamp = req.headers.get("x-webhook-timestamp") || "";
  const signature = req.headers.get("x-webhook-signature") || "";

  let event;
  try {
    event = verifyWebhook(body, timestamp, signature, process.env.ZQUENCE_WEBHOOK_SECRET!);
  } catch {
    return new NextResponse("Invalid webhook", { status: 400 });
  }

  await enqueue(event);
  return NextResponse.json({ received: true });
}

​
Flask (Python)

from flask import Flask, request
from verify import verify_webhook
import os

app = Flask(__name__)

@app.post("/webhooks/zquence")
def handle():
    try:
        event = verify_webhook(
            request.data,
            request.headers.get("X-Webhook-Timestamp", ""),
            request.headers.get("X-Webhook-Signature", ""),
            os.environ["ZQUENCE_WEBHOOK_SECRET"],
        )
    except ValueError:
        return "Invalid webhook", 400

    enqueue(event)
    return "", 200

​
Response codes

Current delivery behavior is based only on HTTP success vs failure:
  • 2xx → delivery succeeds and stops retrying.
  • Non-2xx → delivery is marked failed and retried if attempts remain.
  • Timeout after 10 seconds → delivery is marked failed and retried if attempts remain.
Keep your handler small: verify, enqueue, return 200. Do not do slow business logic inline.