Webhook 事件接收

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

你将构建什么

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

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"]
  }
}

前置条件

袋袋 App

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

Webhook URL

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

Webhook Secret

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

Step 1: 注册 Webhook URL

在袋袋 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，再异步处理。袋袋平台在超时未收到响应时会触发重试，快速应答可避免重复投递。

Step 3: 验证签名

袋袋使用 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: 响应与重试

响应规范

你的响应	袋袋的行为
`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 调用）。袋袋平台的投递超时为 10 秒。

本地开发调试

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

ngrok http 3000

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

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

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

下一步

Next.js 全栈集成

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

按次计费 SaaS

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

Events API 参考

查看完整的 Events API 文档

Token 管理最佳实践

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

开始使用

认证与密钥

API 能力指南

SDK

集成

Cookbook

参考

Webhook 事件接收

你将构建什么

Webhook 事件类型

前置条件

袋袋 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 事件类型

​前置条件

袋袋 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: 响应与重试

响应规范

最佳实践

本地开发调试

完整示例

安全最佳实践

下一步