Webhooks

curl -X POST https://api.talqing.com/v1/webhooks \
  -H "Authorization: Bearer $TALQING_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "url": "https://hooks.example.com/talqing",
    "subscribed_events": ["session.completed", "recording.ready"]
  }'

const hook = await talqing.webhooks.create({
  createWebhookRequest: {
    url: "https://hooks.example.com/talqing",
    subscribed_events: ["session.completed", "recording.ready"],
  },
});
console.log(hook.secret); // shown once

hook = talqing.webhooks.create(
    url="https://hooks.example.com/talqing",
    subscribed_events=["session.completed", "recording.ready"],
)
print(hook["secret"])  # shown once

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 a = Buffer.from(header ?? "", "utf8");
  const b = Buffer.from(expected, "utf8");
  return a.length === b.length && crypto.timingSafeEqual(a, b);
}

app.post("/talqing", express.raw({ type: "application/json" }), (req, res) => {
  if (!verify(req.body, req.get("X-Talqing-Signature"), process.env.TALQING_WEBHOOK_SECRET)) {
    return res.sendStatus(401);
  }
  const event = JSON.parse(req.body.toString("utf8"));
  res.sendStatus(200); // answer first
  void handle(event); // then work
});

import hashlib
import hmac
import os

from flask import Flask, request

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


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


@app.post("/talqing")
def talqing_webhook():
    if not verify(request.get_data(), request.headers.get("X-Talqing-Signature")):
        return "", 401
    enqueue(request.get_json())  # answer first, work afterwards
    return "", 200