Test Intercom webhooks locally ngrok, localhost, webhook replay

// Minimal receiver: verify the signature over the raw request body (bytes)
// npm i express
import crypto from "node:crypto";
import express from "express";

const app = express();

app.post(
  "/webhook",
  express.raw({ type: "*/*", limit: "2mb" }),
  (req, res) => {
    const secret = process.env.WEBHOOK_SECRET ?? "replace_me";
    const signature = req.header("x-hub-signature") ?? "";
    const rawBody = Buffer.isBuffer(req.body) ? req.body : Buffer.from(req.body ?? "");

    if (!signature) return res.status(401).send("missing signature header");

    const payload = rawBody.toString("utf8");
    const digest = crypto.createHmac("sha1", secret).update(payload, "utf8").digest("hex");
    const expected = "sha1=" + digest;

    const a = Buffer.from(signature);
    const b = Buffer.from(expected);
    if (a.length !== b.length) return res.status(401).send("bad signature");
    if (!crypto.timingSafeEqual(a, b)) return res.status(401).send("bad signature");

    // Authentic. Ack fast.
    res.sendStatus(202);
  }
);

app.listen(3000, () => console.log("listening on :3000"));

# Minimal receiver: verify the signature over the raw request body (bytes)
# pip install fastapi uvicorn
import hashlib
import hmac
import os
from fastapi import FastAPI, Request, Response

app = FastAPI()

@app.post("/webhook")
async def webhook(request: Request) -> Response:
    secret = os.getenv("WEBHOOK_SECRET", "replace_me").encode("utf-8")
    signature = request.headers.get("x-hub-signature", "")
    raw_body = await request.body()

    if not signature:
        return Response("missing signature header", status_code=401)

    payload = raw_body
    digest = hmac.new(secret, payload, hashlib.sha1).hexdigest()
    expected = "sha1=" + digest

    if not hmac.compare_digest(expected, signature):
        return Response("bad signature", status_code=401)

    # Authentic. Ack fast.
    return Response(status_code=202)

# Run: uvicorn app:app --reload --port 3000

// Minimal local consumer loop (REST pull):
// 1) GET .../next
// 2) parse X-Hooque-Meta (ackUrl/nackUrl/rejectUrl)
// 3) POST ackUrl on success, otherwise nack/reject
//
// Node 18+ (fetch built-in)
const QUEUE_NEXT_URL =
  process.env.HOOQUE_QUEUE_NEXT_URL ??
  "https://app.hooque.io/queues/cons_dev_webhooks/next";
const TOKEN = process.env.HOOQUE_TOKEN ?? "hq_tok_replace_me";

const headers = { Authorization: `Bearer ${TOKEN}` };

async function processLocally(payload) {
  // TODO: call your local code paths here
  console.log("received event:", payload?.type ?? payload?.id ?? "event");
}

while (true) {
  const resp = await fetch(QUEUE_NEXT_URL, { headers });

  if (resp.status === 204) break; // queue is empty
  if (!resp.ok) throw new Error(`Hooque next() failed: ${resp.status}`);

  const payload = await resp.json();
  const meta = JSON.parse(resp.headers.get("X-Hooque-Meta") ?? "{}");

  try {
    await processLocally(payload);
    await fetch(meta.ackUrl, { method: "POST", headers });
  } catch (err) {
    const reason = String(err);
    const isPermanent = err?.name === "ValidationError";
    const url = isPermanent ? meta.rejectUrl : meta.nackUrl;

    await fetch(url, {
      method: "POST",
      headers: { ...headers, "Content-Type": "application/json" },
      body: JSON.stringify({ reason }),
    });
  }
}

# Minimal local consumer loop (REST pull):
# 1) GET .../next
# 2) parse X-Hooque-Meta (ackUrl/nackUrl/rejectUrl)
# 3) POST ackUrl on success, otherwise nack/reject
import json
import os
import requests

QUEUE_NEXT_URL = os.getenv(
    "HOOQUE_QUEUE_NEXT_URL",
    "https://app.hooque.io/queues/cons_dev_webhooks/next",
)
TOKEN = os.getenv("HOOQUE_TOKEN", "hq_tok_replace_me")
headers = {"Authorization": f"Bearer {TOKEN}"}

def process_locally(payload: dict) -> None:
    print("received event:", payload.get("type") or payload.get("id") or "event")
    # TODO: call your local code paths here

while True:
    resp = requests.get(QUEUE_NEXT_URL, headers=headers, timeout=30)

    if resp.status_code == 204:
        break
    if resp.status_code >= 400:
        raise RuntimeError(f"Hooque next() failed: {resp.status_code} {resp.text}")

    payload = resp.json()
    meta = json.loads(resp.headers.get("X-Hooque-Meta", "{}"))

    try:
        process_locally(payload)
        requests.post(meta["ackUrl"], headers=headers, timeout=30)
    except Exception as err:
        reason = str(err)
        is_permanent = err.__class__.__name__ == "ValidationError"
        url = meta["rejectUrl"] if is_permanent else meta["nackUrl"]
        requests.post(
            url,
            headers={**headers, "Content-Type": "application/json"},
            json={"reason": reason},
            timeout=30,
        )