Webhook 事件接收

Webhook 事件系统正在持续完善中。本文档展示的事件类型和签名格式为推荐实现模式，具体以平台发布的最新文档为准。

你将构建什么

本教程将带你构建一个 Webhook 接收服务，安全地接收并处理 Profy 平台推送的实时事件。你将实现签名验证、事件分发、幂等处理和异步消费的完整流程。

Webhook 事件类型

事件类型	触发时机	Payload 示例
`user.authorized`	用户完成 OAuth 授权	`{ "user_id": "u_abc", "app_id": "app_123", "scopes": ["events:write"] }`
`user.revoked`	用户撤销应用授权	`{ "user_id": "u_abc", "app_id": "app_123" }`
`billing.event_processed`	计费事件扣费成功	`{ "user_id": "u_abc", "event_name": "ai_generation", "amount": 10, "balance_remaining": 490 }`
`billing.insufficient_balance`	用户余额不足，扣费失败	`{ "user_id": "u_abc", "event_name": "ai_generation", "amount": 10, "balance": 3 }`

每个 Webhook 请求的 JSON 结构：

{
  "event_id": "evt_20260602_xK9mN3",
  "event_type": "user.authorized",
  "created_at": "2026-06-02T09:15:00Z",
  "payload": {
    "user_id": "u_abc",
    "app_id": "app_123",
    "scopes": ["events:write"]
  }
}

前置条件

Profy App

已在 Profy Studio 创建 App，获取 Client ID

Webhook URL

在 Studio 配置了可公网访问的 Webhook 接收地址

Webhook Secret

在 Studio 获取 Webhook 签名密钥（用于验证请求来源）

Step 1: 注册 Webhook URL

在 Profy Studio 中完成 Webhook 配置：

进入 App 管理 → 选择你的 App → Webhook 设置
填写你的 Webhook 接收地址（如 https://your-app.com/api/webhook）
选择需要订阅的事件类型
保存后复制自动生成的 Webhook Secret

.env

PROFY_WEBHOOK_SECRET=whsec_your_webhook_secret_here

Webhook Secret 是验证请求来源的唯一凭据。请妥善保管，不要提交到版本控制或暴露给客户端。

Step 2: 创建 Webhook 接收端点

import express from "express";
import { verifySignature } from "./verify";
import { handleWebhookEvent } from "./handler";

const app = express();

app.post(
  "/api/webhook",
  express.raw({ type: "application/json" }),
  async (req, res) => {
    const signature = req.headers["x-profy-signature"] as string;
    const body = req.body as Buffer;

    if (!verifySignature(body, signature)) {
      res.status(401).json({ error: "Invalid signature" });
      return;
    }

    const event = JSON.parse(body.toString());

    res.status(200).json({ received: true });

    handleWebhookEvent(event).catch(console.error);
  }
);

app.listen(3000);

import { Hono } from "hono";
import { verifySignature } from "./verify";
import { handleWebhookEvent } from "./handler";

const app = new Hono();

app.post("/api/webhook", async (c) => {
  const signature = c.req.header("x-profy-signature") ?? "";
  const body = await c.req.arrayBuffer();
  const rawBody = Buffer.from(body);

  if (!verifySignature(rawBody, signature)) {
    return c.json({ error: "Invalid signature" }, 401);
  }

  const event = JSON.parse(rawBody.toString());

  c.executionCtx.waitUntil(handleWebhookEvent(event));

  return c.json({ received: true }, 200);
});

export default app;

from fastapi import FastAPI, Request, HTTPException
import asyncio
from verify import verify_signature
from handler import handle_webhook_event

app = FastAPI()

@app.post("/api/webhook")
async def webhook(request: Request):
    signature = request.headers.get("x-profy-signature", "")
    body = await request.body()

    if not verify_signature(body, signature):
        raise HTTPException(status_code=401, detail="Invalid signature")

    event = await request.json()

    asyncio.create_task(handle_webhook_event(event))

    return {"received": True}

先返回 200，再异步处理。Profy 平台在超时未收到响应时会触发重试，快速应答可避免重复投递。

Step 3: 验证签名

Profy 使用 HMAC-SHA256 对请求 body 签名，签名值通过 X-Profy-Signature header 传递。格式为 sha256=<hex_digest>。

import { createHmac, timingSafeEqual } from "node:crypto";

const WEBHOOK_SECRET = process.env.PROFY_WEBHOOK_SECRET!;

export function verifySignature(body: Buffer, signature: string): boolean {
  if (!signature.startsWith("sha256=")) return false;

  const expected = createHmac("sha256", WEBHOOK_SECRET)
    .update(body)
    .digest("hex");

  const received = signature.slice("sha256=".length);

  if (expected.length !== received.length) return false;

  return timingSafeEqual(
    Buffer.from(expected, "hex"),
    Buffer.from(received, "hex")
  );
}

import hmac
import hashlib
import os

WEBHOOK_SECRET = os.environ["PROFY_WEBHOOK_SECRET"]

def verify_signature(body: bytes, signature: str) -> bool:
    if not signature.startswith("sha256="):
        return False

    expected = hmac.new(
        WEBHOOK_SECRET.encode(),
        body,
        hashlib.sha256,
    ).hexdigest()

    received = signature[len("sha256="):]

    return hmac.compare_digest(expected, received)

必须使用 timingSafeEqual（Node）或 hmac.compare_digest（Python）进行比较。普通字符串比较存在时序攻击风险。

Step 4: 处理事件

按 event_type 分发到对应的处理函数。

interface WebhookEvent {
  event_id: string;
  event_type: string;
  created_at: string;
  payload: Record<string, unknown>;
}

export async function handleWebhookEvent(event: WebhookEvent) {
  switch (event.event_type) {
    case "user.authorized":
      await onUserAuthorized(event.payload);
      break;
    case "user.revoked":
      await onUserRevoked(event.payload);
      break;
    case "billing.event_processed":
      await onBillingProcessed(event.payload);
      break;
    case "billing.insufficient_balance":
      await onInsufficientBalance(event.payload);
      break;
    default:
      console.warn(`Unhandled event type: ${event.event_type}`);
  }
}

async function onUserAuthorized(payload: Record<string, unknown>) {
  const userId = payload.user_id as string;
  await db.user.upsert({ externalId: userId, status: "active" });
}

async function onUserRevoked(payload: Record<string, unknown>) {
  const userId = payload.user_id as string;
  await db.user.update({ externalId: userId, status: "revoked" });
  await tokenStore.delete(userId);
}

async function onBillingProcessed(payload: Record<string, unknown>) {
  const { user_id, event_name, amount, balance_remaining } = payload as {
    user_id: string;
    event_name: string;
    amount: number;
    balance_remaining: number;
  };
  await db.billingLog.create({ userId: user_id, event_name, amount, balance_remaining });
}

async function onInsufficientBalance(payload: Record<string, unknown>) {
  const userId = payload.user_id as string;
  await notificationService.send(userId, "余额不足，部分功能可能受限");
}

from typing import Any

async def handle_webhook_event(event: dict[str, Any]) -> None:
    event_type = event["event_type"]
    payload = event["payload"]

    handlers = {
        "user.authorized": on_user_authorized,
        "user.revoked": on_user_revoked,
        "billing.event_processed": on_billing_processed,
        "billing.insufficient_balance": on_insufficient_balance,
    }

    handler = handlers.get(event_type)
    if handler:
        await handler(payload)
    else:
        print(f"Unhandled event type: {event_type}")


async def on_user_authorized(payload: dict[str, Any]) -> None:
    user_id = payload["user_id"]
    await db.user.upsert(external_id=user_id, status="active")


async def on_user_revoked(payload: dict[str, Any]) -> None:
    user_id = payload["user_id"]
    await db.user.update(external_id=user_id, status="revoked")
    await token_store.delete(user_id)


async def on_billing_processed(payload: dict[str, Any]) -> None:
    await db.billing_log.create(
        user_id=payload["user_id"],
        event_name=payload["event_name"],
        amount=payload["amount"],
        balance_remaining=payload["balance_remaining"],
    )


async def on_insufficient_balance(payload: dict[str, Any]) -> None:
    user_id = payload["user_id"]
    await notification_service.send(user_id, "余额不足，部分功能可能受限")

Step 5: 幂等处理

网络抖动或超时重试可能导致同一事件被投递多次。使用 event_id 做幂等去重。

const PROCESSED_TTL_MS = 7 * 24 * 60 * 60 * 1000;

export async function isProcessed(eventId: string): Promise<boolean> {
  const existing = await redis.get(`webhook:processed:${eventId}`);
  return existing !== null;
}

export async function markProcessed(eventId: string): Promise<void> {
  await redis.set(`webhook:processed:${eventId}`, "1", "PX", PROCESSED_TTL_MS);
}

PROCESSED_TTL_SECONDS = 7 * 24 * 60 * 60

async def is_processed(event_id: str) -> bool:
    return await redis.exists(f"webhook:processed:{event_id}")

async def mark_processed(event_id: str) -> None:
    await redis.set(
        f"webhook:processed:{event_id}",
        "1",
        ex=PROCESSED_TTL_SECONDS,
    )

在主处理流程中加入幂等检查：

import { isProcessed, markProcessed } from "./idempotency";

export async function handleWebhookEvent(event: WebhookEvent) {
  if (await isProcessed(event.event_id)) return;

  await markProcessed(event.event_id);

  switch (event.event_type) {
    // ...
  }
}

from idempotency import is_processed, mark_processed

async def handle_webhook_event(event: dict[str, Any]) -> None:
    if await is_processed(event["event_id"]):
        return

    await mark_processed(event["event_id"])

    # ...

如果没有 Redis，也可以用数据库 UNIQUE(event_id) 约束实现幂等。插入失败即表示已处理过。

Step 6: 响应与重试

响应规范

你的响应	Profy 的行为
`2xx`	投递成功，不再重试
`4xx`（非 429）	投递失败，不重试（视为客户端永久错误）
`429`	限流，稍后重试
`5xx` / 超时	按指数退避重试（最多 5 次，间隔 30s → 2m → 8m → 30m → 2h）

最佳实践

app.post(
  "/api/webhook",
  express.raw({ type: "application/json" }),
  async (req, res) => {
    const signature = req.headers["x-profy-signature"] as string;
    const body = req.body as Buffer;

    if (!verifySignature(body, signature)) {
      res.status(401).json({ error: "Invalid signature" });
      return;
    }

    const event = JSON.parse(body.toString());

    res.status(200).json({ received: true });

    handleWebhookEvent(event).catch((err) => {
      console.error(`Failed to process event ${event.event_id}:`, err);
    });
  }
);

@app.post("/api/webhook")
async def webhook(request: Request):
    signature = request.headers.get("x-profy-signature", "")
    body = await request.body()

    if not verify_signature(body, signature):
        raise HTTPException(status_code=401, detail="Invalid signature")

    event = await request.json()

    asyncio.create_task(handle_webhook_event(event))

    return {"received": True}

不要在返回 200 之前执行耗时操作（如数据库写入、外部 API 调用）。Profy 平台的投递超时为 10 秒。

本地开发调试

本地开发时，Webhook URL 需要公网可达。推荐使用隧道工具将本地端口暴露到公网。

ngrok http 3000

cloudflared tunnel --url http://localhost:3000

获取到临时公网地址后，将其填入 Profy Studio 的 Webhook URL 配置中（如 https://abc123.ngrok-free.app/api/webhook）。

调试时可以在 Profy Studio 的 Webhook 日志中查看每次投递的请求/响应详情和重试记录。

完整示例

以下是可直接运行的完整示例：

import express from "express";
import { createHmac, timingSafeEqual } from "node:crypto";

const app = express();
const WEBHOOK_SECRET = process.env.PROFY_WEBHOOK_SECRET!;
const processedEvents = new Map<string, number>();

function verifySignature(body: Buffer, signature: string): boolean {
  if (!signature.startsWith("sha256=")) return false;

  const expected = createHmac("sha256", WEBHOOK_SECRET)
    .update(body)
    .digest("hex");

  const received = signature.slice("sha256=".length);

  if (expected.length !== received.length) return false;

  return timingSafeEqual(
    Buffer.from(expected, "hex"),
    Buffer.from(received, "hex")
  );
}

function isProcessed(eventId: string): boolean {
  return processedEvents.has(eventId);
}

function markProcessed(eventId: string): void {
  processedEvents.set(eventId, Date.now());
  if (processedEvents.size > 10_000) {
    const cutoff = Date.now() - 7 * 24 * 60 * 60 * 1000;
    for (const [id, ts] of processedEvents) {
      if (ts < cutoff) processedEvents.delete(id);
    }
  }
}

interface WebhookEvent {
  event_id: string;
  event_type: string;
  created_at: string;
  payload: Record<string, unknown>;
}

async function handleEvent(event: WebhookEvent) {
  if (isProcessed(event.event_id)) return;
  markProcessed(event.event_id);

  switch (event.event_type) {
    case "user.authorized":
      console.log("User authorized:", event.payload.user_id);
      break;
    case "user.revoked":
      console.log("User revoked:", event.payload.user_id);
      break;
    case "billing.event_processed":
      console.log("Billing processed:", event.payload);
      break;
    case "billing.insufficient_balance":
      console.log("Insufficient balance:", event.payload.user_id);
      break;
    default:
      console.warn("Unknown event:", event.event_type);
  }
}

app.post(
  "/api/webhook",
  express.raw({ type: "application/json" }),
  async (req, res) => {
    const signature = req.headers["x-profy-signature"] as string;
    const body = req.body as Buffer;

    if (!verifySignature(body, signature)) {
      res.status(401).json({ error: "Invalid signature" });
      return;
    }

    const event: WebhookEvent = JSON.parse(body.toString());
    res.status(200).json({ received: true });

    handleEvent(event).catch((err) => {
      console.error(`Event ${event.event_id} failed:`, err);
    });
  }
);

app.listen(3000, () => {
  console.log("Webhook receiver running on :3000");
});

import hashlib
import hmac
import os
import time
from typing import Any

from fastapi import FastAPI, Request, HTTPException

app = FastAPI()

WEBHOOK_SECRET = os.environ["PROFY_WEBHOOK_SECRET"]
processed_events: dict[str, float] = {}


def verify_signature(body: bytes, signature: str) -> bool:
    if not signature.startswith("sha256="):
        return False

    expected = hmac.new(
        WEBHOOK_SECRET.encode(), body, hashlib.sha256
    ).hexdigest()
    received = signature[len("sha256="):]

    return hmac.compare_digest(expected, received)


def is_processed(event_id: str) -> bool:
    return event_id in processed_events


def mark_processed(event_id: str) -> None:
    processed_events[event_id] = time.time()
    if len(processed_events) > 10_000:
        cutoff = time.time() - 7 * 24 * 60 * 60
        to_remove = [eid for eid, ts in processed_events.items() if ts < cutoff]
        for eid in to_remove:
            del processed_events[eid]


async def handle_event(event: dict[str, Any]) -> None:
    if is_processed(event["event_id"]):
        return
    mark_processed(event["event_id"])

    match event["event_type"]:
        case "user.authorized":
            print(f"User authorized: {event['payload']['user_id']}")
        case "user.revoked":
            print(f"User revoked: {event['payload']['user_id']}")
        case "billing.event_processed":
            print(f"Billing processed: {event['payload']}")
        case "billing.insufficient_balance":
            print(f"Insufficient balance: {event['payload']['user_id']}")
        case _:
            print(f"Unknown event: {event['event_type']}")


@app.post("/api/webhook")
async def webhook(request: Request):
    signature = request.headers.get("x-profy-signature", "")
    body = await request.body()

    if not verify_signature(body, signature):
        raise HTTPException(status_code=401, detail="Invalid signature")

    import json
    event = json.loads(body)

    await handle_event(event)

    return {"received": True}

安全最佳实践

始终验证签名

每个请求都必须通过 HMAC-SHA256 签名验证，拒绝一切未签名或签名错误的请求

使用 HTTPS

Webhook URL 必须使用 HTTPS，避免请求内容在传输中被窃听或篡改

幂等处理

利用 event_id 做去重，确保同一事件被多次投递时只处理一次

超时保护

在 10 秒内返回 200。耗时逻辑放到后台队列，避免阻塞响应触发重试

其他建议：

IP 白名单：如果你的防火墙支持，可以只放行 Profy 平台的出口 IP 段
Secret 轮换：定期在 Studio 重新生成 Webhook Secret，并在应用侧平滑切换（支持同时验证新旧两个 Secret 的过渡期）
日志与监控：记录每个 Webhook 的 event_id、处理结果、耗时，便于排查丢失或延迟问题

下一步

Next.js 全栈集成

OAuth 登录 + Token 管理 + 计费事件上报

按次计费 SaaS

PER_USE 模式：固定价格按次扣费

Events API 参考

查看完整的 Events API 文档

Token 管理最佳实践

并发刷新、安全存储、降级策略

​你将构建什么

​Webhook 事件类型

​前置条件

Profy App

Webhook URL

Webhook Secret

​Step 1: 注册 Webhook URL

​Step 2: 创建 Webhook 接收端点

​Step 3: 验证签名

​Step 4: 处理事件

​Step 5: 幂等处理

​Step 6: 响应与重试

​响应规范

​最佳实践

​本地开发调试

​完整示例

​安全最佳实践

始终验证签名

使用 HTTPS

幂等处理

超时保护

​下一步

Next.js 全栈集成

按次计费 SaaS

Events API 参考

Token 管理最佳实践

你将构建什么

Webhook 事件类型

前置条件

Step 1: 注册 Webhook URL

Step 2: 创建 Webhook 接收端点

Step 3: 验证签名

Step 4: 处理事件

Step 5: 幂等处理

Step 6: 响应与重试

响应规范

最佳实践

本地开发调试

完整示例

安全最佳实践

下一步