Webhook 回呼指南

curl -X POST "https://vas-poc.vurbo.ai/api/v1/imports" \
  -H "X-API-Key: YOUR_API_KEY" \
  -F "file=@meeting.mp3" \
  -F 'transcription_languages=["zh-TW"]' \
  -F "recognition_mode=multi_speaker" \
  -F "callback_url=https://your-server.com/webhooks/vas"

curl -X POST "https://vas-poc.vurbo.ai/api/v1/broadcasts" \
  -H "X-API-Key: YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "transcription_language": "zh-TW",
    "translation_languages": ["en-US"],
    "callback_url": "https://your-server.com/webhooks/vas"
  }'

{
  "event": "recording.completed",
  "timestamp": "2026-02-24T12:00:00.000000Z",
  "delivery_id": "550e8400-e29b-41d4-a716-446655440000",
  "data": { ... }
}

{
  "event": "recording.completed",
  "timestamp": "2026-02-24T12:00:00.000000Z",
  "delivery_id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
  "data": {
    "task_id": "550e8400-e29b-41d4-a716-446655440000",
    "name": "會議錄音",
    "duration_ms": 3600000,
    "type_source": "realtime",
    "transcription_languages": ["zh-TW"],
    "translation_languages": ["en-US"]
  }
}

{
  "event": "recording.failed",
  "timestamp": "2026-02-24T12:05:00.000000Z",
  "delivery_id": "b2c3d4e5-f6a7-8901-bcde-f12345678901",
  "data": {
    "task_id": "550e8400-e29b-41d4-a716-446655440000",
    "error": "No audio was received during the recording, so nothing was saved.",
    "failure_source": "no_audio"
  }
}

{
  "event": "import.completed",
  "timestamp": "2026-02-24T12:00:00.000000Z",
  "delivery_id": "c3d4e5f6-a7b8-9012-cdef-123456789012",
  "data": {
    "import_id": "660e8400-e29b-41d4-a716-446655440001",
    "task_id": "550e8400-e29b-41d4-a716-446655440000",
    "name": "會議錄音",
    "duration_ms": 3600000
  }
}

{
  "event": "import.failed",
  "timestamp": "2026-02-24T12:05:00.000000Z",
  "delivery_id": "d4e5f6a7-b8c9-0123-defa-234567890123",
  "data": {
    "import_id": "660e8400-e29b-41d4-a716-446655440001",
    "error_code": "import_conversion_failed",
    "error_message": "音檔格式轉換失敗"
  }
}

{
  "event": "credit.low",
  "timestamp": "2026-02-24T12:05:00.000000Z",
  "delivery_id": "e5f6a7b8-c9d0-1234-efab-345678901234",
  "data": {
    "balance": "49.5",
    "threshold": "50.0"
  }
}

{
  "event": "credit.exhausted",
  "timestamp": "2026-02-24T12:05:00.000000Z",
  "delivery_id": "f6a7b8c9-d0e1-2345-fabc-456789012345",
  "data": {
    "balance": "0.0"
  }
}

signature_payload = timestamp + "." + raw_body
signature = "sha256=" + HMAC-SHA256(signature_payload, webhook_secret)

const crypto = require('crypto');

function verifyWebhookSignature(req, webhookSecret) {
  const timestamp = req.headers['x-vas-timestamp'];
  const signature = req.headers['x-vas-signature'];
  const body = req.rawBody; // 原始請求 body 字串

  const signaturePayload = `${timestamp}.${body}`;
  const expected = 'sha256=' + crypto
    .createHmac('sha256', webhookSecret)
    .update(signaturePayload)
    .digest('hex');

  return crypto.timingSafeEqual(
    Buffer.from(signature),
    Buffer.from(expected)
  );
}

import hmac
import hashlib

def verify_webhook_signature(timestamp, body, signature, webhook_secret):
    signature_payload = f"{timestamp}.{body}"
    expected = "sha256=" + hmac.new(
        webhook_secret.encode(),
        signature_payload.encode(),
        hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(signature, expected)

const MAX_AGE_SECONDS = 300; // 5 分鐘
const timestamp = parseInt(req.headers['x-vas-timestamp']);
const now = Math.floor(Date.now() / 1000);

if (Math.abs(now - timestamp) > MAX_AGE_SECONDS) {
  return res.status(401).json({ error: 'Timestamp too old' });
}

// 推薦：先回應 200，再非同步處理
app.post('/webhooks/vas', async (req, res) => {
  // 立即回應
  res.status(200).json({ received: true });

  // 非同步處理事件
  processWebhookEvent(req.body).catch(console.error);
});

app.post('/webhooks/vas', async (req, res) => {
  const { delivery_id } = req.body;

  // 檢查是否已處理過
  if (await isDeliveryProcessed(delivery_id)) {
    return res.status(200).json({ received: true, duplicate: true });
  }

  // 標記為已處理
  await markDeliveryProcessed(delivery_id);

  // 處理事件...
  res.status(200).json({ received: true });
});

app.post('/webhooks/vas', (req, res) => {
  if (!verifyWebhookSignature(req, WEBHOOK_SECRET)) {
    return res.status(401).json({ error: 'Invalid signature' });
  }
  // 處理事件...
});

const express = require('express');
const crypto = require('crypto');

const app = express();
const WEBHOOK_SECRET = 'your_webhook_secret_here';

// 保留原始 body 用於簽名驗證
app.use('/webhooks/vas', express.json({
  verify: (req, res, buf) => { req.rawBody = buf.toString(); }
}));

app.post('/webhooks/vas', (req, res) => {
  // 1. 驗證簽名
  const timestamp = req.headers['x-vas-timestamp'];
  const signature = req.headers['x-vas-signature'];
  const signaturePayload = `${timestamp}.${req.rawBody}`;
  const expected = 'sha256=' + crypto
    .createHmac('sha256', WEBHOOK_SECRET)
    .update(signaturePayload)
    .digest('hex');

  if (!crypto.timingSafeEqual(Buffer.from(signature || ''), Buffer.from(expected))) {
    return res.status(401).json({ error: 'Invalid signature' });
  }

  // 2. 立即回應
  res.status(200).json({ received: true });

  // 3. 處理事件
  const { event, data, delivery_id } = req.body;
  console.log(`Webhook received: ${event} (${delivery_id})`);

  switch (event) {
    case 'recording.completed':
      console.log(`Recording completed: task_id=${data.task_id}, duration=${data.duration_ms}ms`);
      // 載入逐字稿、通知使用者等...
      break;

    case 'recording.failed':
      console.log(`Recording failed: task_id=${data.task_id}, error=${data.error}`);
      // 通知使用者處理失敗...
      break;

    case 'import.completed':
      console.log(`Import completed: import_id=${data.import_id}, task_id=${data.task_id}`);
      break;

    case 'import.failed':
      console.log(`Import failed: import_id=${data.import_id}, error=${data.error_message}`);
      break;
  }
});

app.listen(3000, () => console.log('Webhook server running on port 3000'));

import hmac
import hashlib
import json
from flask import Flask, request, jsonify

app = Flask(__name__)
WEBHOOK_SECRET = "your_webhook_secret_here"

@app.route("/webhooks/vas", methods=["POST"])
def handle_webhook():
    # 1. 驗證簽名
    timestamp = request.headers.get("X-VAS-Timestamp", "")
    signature = request.headers.get("X-VAS-Signature", "")
    raw_body = request.get_data(as_text=True)

    signature_payload = f"{timestamp}.{raw_body}"
    expected = "sha256=" + hmac.new(
        WEBHOOK_SECRET.encode(),
        signature_payload.encode(),
        hashlib.sha256,
    ).hexdigest()

    if not hmac.compare_digest(signature, expected):
        return jsonify({"error": "Invalid signature"}), 401

    # 2. 處理事件
    payload = request.get_json()
    event = payload["event"]
    data = payload["data"]
    delivery_id = payload["delivery_id"]

    print(f"Webhook received: {event} ({delivery_id})")

    if event == "recording.completed":
        task_id = data["task_id"]
        print(f"Recording completed: {task_id}")
        # 載入逐字稿、通知使用者等...

    elif event == "recording.failed":
        print(f"Recording failed: {data['error']}")

    elif event == "import.completed":
        print(f"Import completed: import_id={data['import_id']}, task_id={data['task_id']}")

    elif event == "import.failed":
        print(f"Import failed: {data['error_message']}")

    return jsonify({"received": True}), 200

if __name__ == "__main__":
    app.run(port=3000)