Webhooks

curl -X POST https://api.in.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"],
  },
});

// The only time the plaintext secret is ever returned. Store it now.
console.log(hook.secret);

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

# The only time the plaintext secret is ever returned. Store it now.
print(hook["secret"])

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");
  // timingSafeEqual throws on a length mismatch, so check length first.
  return a.length === b.length && crypto.timingSafeEqual(a, b);
}

// `express.raw` keeps req.body as the exact bytes we signed.
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 do the work
  },
);

async function handle(event) {
  if (event.type !== "session.completed") return;
  console.log(event.data.session_id, event.data.analysis.outcome);
}

app.listen(3000);

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, secret: str) -> 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():
    # request.get_data() is the exact bytes we signed. Do not use request.json here.
    if not verify(request.get_data(), request.headers.get("X-Talqing-Signature"), SECRET):
        return "", 401

    event = request.get_json()
    enqueue(event)  # answer first, do the work afterwards
    return "", 200

curl -G https://api.in.talqing.com/v1/webhooks/2e6a1f04-9c3b-4d17-8e52-b1d0a7c94f68/deliveries \
  -H "Authorization: Bearer $TALQING_API_KEY" \
  --data-urlencode "limit=20"

const page = await talqing.webhooks.deliveries({
  webhook_id: "2e6a1f04-9c3b-4d17-8e52-b1d0a7c94f68",
  limit: 20,
});

for (const attempt of page.items) {
  console.log(attempt.created_at, attempt.event_type, attempt.status, attempt.status_code);
}

page = talqing.webhooks.deliveries(
    "2e6a1f04-9c3b-4d17-8e52-b1d0a7c94f68", limit=20
)

for attempt in page["items"]:
    print(attempt["created_at"], attempt["event_type"], attempt["status"], attempt["error"])

curl -X POST \
  https://api.in.talqing.com/v1/webhooks/2e6a1f04-9c3b-4d17-8e52-b1d0a7c94f68/test \
  -H "Authorization: Bearer $TALQING_API_KEY"

const result = await talqing.webhooks.test({
  webhook_id: "2e6a1f04-9c3b-4d17-8e52-b1d0a7c94f68",
});

console.log(result.status, result.status_code, result.error);

result = talqing.webhooks.test("2e6a1f04-9c3b-4d17-8e52-b1d0a7c94f68")
print(result["status"], result["status_code"], result["error"])

curl -X POST \
  https://api.in.talqing.com/v1/webhooks/2e6a1f04-9c3b-4d17-8e52-b1d0a7c94f68/rotate-secret \
  -H "Authorization: Bearer $TALQING_API_KEY"

const rotated = await talqing.webhooks.rotateSecret({
  webhook_id: "2e6a1f04-9c3b-4d17-8e52-b1d0a7c94f68",
});

console.log(rotated.secret); // shown once

rotated = talqing.webhooks.rotate_secret("2e6a1f04-9c3b-4d17-8e52-b1d0a7c94f68")
print(rotated["secret"])  # shown once