Webhook events

import crypto from "node:crypto";
import express from "express";

const app = express();

function verify(rawBody, header, secret) {
  const expected =
    "sha256=" + crypto.createHmac("sha256", secret).update(rawBody).digest("hex");
  const sent = Buffer.from(header ?? "", "utf8");
  const mine = Buffer.from(expected, "utf8");
  // Lengths must match before timingSafeEqual, which throws otherwise.
  return sent.length === mine.length && crypto.timingSafeEqual(sent, mine);
}

app.post(
  "/talqing/webhook",
  express.raw({ type: "application/json" }),
  (req, res) => {
    if (!verify(req.body, req.get("X-Talqing-Signature"), process.env.TALQING_WEBHOOK_SECRET)) {
      return res.status(401).end();
    }
    const event = JSON.parse(req.body.toString("utf8"));
    // Answer now, work later — you have five seconds and there is no retry.
    res.status(204).end();
    void handle(event);
  },
);

import hashlib
import hmac
import json
import os

from fastapi import BackgroundTasks, FastAPI, Header, Request, Response

app = FastAPI()
SECRET = os.environ["TALQING_WEBHOOK_SECRET"]


def verify(raw: bytes, header: str | None) -> bool:
    expected = "sha256=" + hmac.new(SECRET.encode(), raw, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, header or "")


@app.post("/talqing/webhook")
async def receive(
    request: Request,
    background: BackgroundTasks,
    x_talqing_signature: str | None = Header(default=None),
) -> Response:
    raw = await request.body()
    if not verify(raw, x_talqing_signature):
        return Response(status_code=401)
    event = json.loads(raw)
    # Answer now, work later — you have five seconds and there is no retry.
    background.add_task(handle, event)
    return Response(status_code=204)