> ## Documentation Index
> Fetch the complete documentation index at: https://docs.profy.cn/llms.txt
> Use this file to discover all available pages before exploring further.

# Webhook 事件接收

> 构建安全可靠的 Webhook 接收端点，处理 Profy 平台推送的用户授权、计费等实时事件通知

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

## 你将构建什么

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

```mermaid theme={null}
sequenceDiagram
    participant Profy as Profy 平台
    participant App as 你的应用
    participant DB as 数据库 / 队列

    Profy->>App: POST /webhook (JSON + X-Profy-Signature)
    App->>App: 验证 HMAC-SHA256 签名
    App->>App: 按 event_id 幂等去重
    App-->>Profy: 200 OK（立即响应）
    App->>DB: 异步处理事件
```

## 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 结构：

```json theme={null}
{
  "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"]
  }
}
```

## 前置条件

<CardGroup cols={3}>
  <Card title="Profy App" icon="grid-2">
    已在 Profy Studio 创建 App，获取 Client ID
  </Card>

  <Card title="Webhook URL" icon="link">
    在 Studio 配置了可公网访问的 Webhook 接收地址
  </Card>

  <Card title="Webhook Secret" icon="key">
    在 Studio 获取 Webhook 签名密钥（用于验证请求来源）
  </Card>
</CardGroup>

## Step 1: 注册 Webhook URL

在 Profy Studio 中完成 Webhook 配置：

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

```env .env theme={null}
PROFY_WEBHOOK_SECRET=whsec_your_webhook_secret_here
```

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

## Step 2: 创建 Webhook 接收端点

<CodeGroup>
  ```typescript webhook.ts (Express) theme={null}
  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);
  ```

  ```typescript webhook.ts (Hono) theme={null}
  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;
  ```

  ```python webhook.py (FastAPI) theme={null}
  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}
  ```
</CodeGroup>

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

## Step 3: 验证签名

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

<CodeGroup>
  ```typescript verify.ts theme={null}
  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")
    );
  }
  ```

  ```python verify.py theme={null}
  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)
  ```
</CodeGroup>

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

## Step 4: 处理事件

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

<CodeGroup>
  ```typescript handler.ts theme={null}
  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, "余额不足，部分功能可能受限");
  }
  ```

  ```python handler.py theme={null}
  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, "余额不足，部分功能可能受限")
  ```
</CodeGroup>

## Step 5: 幂等处理

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

<CodeGroup>
  ```typescript idempotency.ts theme={null}
  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);
  }
  ```

  ```python idempotency.py theme={null}
  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,
      )
  ```
</CodeGroup>

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

<CodeGroup>
  ```typescript handler.ts theme={null}
  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) {
      // ...
    }
  }
  ```

  ```python handler.py theme={null}
  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"])

      # ...
  ```
</CodeGroup>

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

## Step 6: 响应与重试

### 响应规范

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

### 最佳实践

<CodeGroup>
  ```typescript webhook.ts (Express) theme={null}
  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);
      });
    }
  );
  ```

  ```python webhook.py (FastAPI) theme={null}
  @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}
  ```
</CodeGroup>

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

## 本地开发调试

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

<CodeGroup>
  ```bash ngrok theme={null}
  ngrok http 3000
  ```

  ```bash cloudflared theme={null}
  cloudflared tunnel --url http://localhost:3000
  ```
</CodeGroup>

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

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

## 完整示例

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

<CodeGroup>
  ```typescript server.ts theme={null}
  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");
  });
  ```

  ```python server.py (FastAPI) theme={null}
  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}
  ```
</CodeGroup>

## 安全最佳实践

<CardGroup cols={2}>
  <Card title="始终验证签名" icon="shield-check">
    每个请求都必须通过 HMAC-SHA256 签名验证，拒绝一切未签名或签名错误的请求
  </Card>

  <Card title="使用 HTTPS" icon="lock">
    Webhook URL 必须使用 HTTPS，避免请求内容在传输中被窃听或篡改
  </Card>

  <Card title="幂等处理" icon="rotate">
    利用 `event_id` 做去重，确保同一事件被多次投递时只处理一次
  </Card>

  <Card title="超时保护" icon="clock">
    在 10 秒内返回 200。耗时逻辑放到后台队列，避免阻塞响应触发重试
  </Card>
</CardGroup>

其他建议：

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

## 下一步

<CardGroup cols={2}>
  <Card title="Next.js 全栈集成" icon="react" href="/zh/sdk/cookbook/nextjs-fullstack">
    OAuth 登录 + Token 管理 + 计费事件上报
  </Card>

  <Card title="按次计费 SaaS" icon="coins" href="/zh/sdk/cookbook/per-use-billing">
    PER\_USE 模式：固定价格按次扣费
  </Card>

  <Card title="Events API 参考" icon="book" href="/zh/api/post-event">
    查看完整的 Events API 文档
  </Card>

  <Card title="Token 管理最佳实践" icon="shield-check" href="/zh/sdk/cookbook/token-management">
    并发刷新、安全存储、降级策略
  </Card>
</CardGroup>