WebHook
接收端实现
使用 FastAPI 实现包含验签、时间窗口、幂等和事件分发的 WebHook 接收端。
下面的示例展示完整处理顺序。内存集合只用于本地演示；生产环境必须换成具有唯一约束和过期策略的数据库或 Redis。
FastAPI 示例
from __future__ import annotations
import hashlib
import hmac
import json
from datetime import datetime, timedelta, timezone
from typing import Any
from fastapi import FastAPI, Header, HTTPException, Request
app = FastAPI()
WEBHOOK_SECRET = "replace-with-your-secret"
PROCESSED_EVENT_IDS: set[str] = set()
def build_signature(secret: str, timestamp: str, body_bytes: bytes) -> str:
signing_payload = timestamp.encode("utf-8") + b"." + body_bytes
digest = hmac.new(
secret.encode("utf-8"),
signing_payload,
hashlib.sha256,
).hexdigest()
return f"sha256={digest}"
def verify_signature(
secret: str,
timestamp: str,
body_bytes: bytes,
signature: str,
) -> bool:
expected = build_signature(secret, timestamp, body_bytes)
return hmac.compare_digest(expected, signature or "")
def verify_timestamp(timestamp: str, max_skew_minutes: int = 5) -> None:
try:
value = datetime.fromisoformat(timestamp.replace("Z", "+00:00"))
except ValueError as exc:
raise HTTPException(
status_code=400,
detail="Invalid X-Poco-Timestamp",
) from exc
now = datetime.now(timezone.utc)
if abs(now - value.astimezone(timezone.utc)) > timedelta(
minutes=max_skew_minutes
):
raise HTTPException(status_code=400, detail="Timestamp skew too large")
@app.post("/webhook/poco")
async def receive_poco_webhook(
request: Request,
x_poco_event: str | None = Header(default=None, alias="X-Poco-Event"),
x_poco_event_id: str | None = Header(default=None, alias="X-Poco-Event-Id"),
x_poco_timestamp: str | None = Header(
default=None,
alias="X-Poco-Timestamp",
),
x_poco_signature: str | None = Header(
default=None,
alias="X-Poco-Signature",
),
) -> dict[str, Any]:
if not x_poco_event or not x_poco_event_id or not x_poco_timestamp:
raise HTTPException(status_code=400, detail="Missing webhook headers")
body_bytes = await request.body()
if WEBHOOK_SECRET:
if not x_poco_signature:
raise HTTPException(status_code=401, detail="Missing signature")
verify_timestamp(x_poco_timestamp)
if not verify_signature(
WEBHOOK_SECRET,
x_poco_timestamp,
body_bytes,
x_poco_signature,
):
raise HTTPException(status_code=401, detail="Invalid signature")
try:
payload = json.loads(body_bytes.decode("utf-8"))
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
raise HTTPException(status_code=400, detail="Invalid JSON payload") from exc
if payload.get("event_type") != x_poco_event:
raise HTTPException(status_code=400, detail="Event type mismatch")
if payload.get("event_id") != x_poco_event_id:
raise HTTPException(status_code=400, detail="Event ID mismatch")
if x_poco_event_id in PROCESSED_EVENT_IDS:
return {"ok": True, "duplicate": True}
# Replace this with an atomic database or Redis operation in production.
PROCESSED_EVENT_IDS.add(x_poco_event_id)
if payload.get("is_test"):
return {"ok": True, "test": True}
if x_poco_event == "session.completed":
session = payload.get("session", {})
summary = payload.get("summary", {})
print(
"[Webhook] session.completed",
f"session_id={session.get('session_id')}",
f"kind={session.get('kind')}",
f"has_artifacts={summary.get('has_artifacts')}",
)
elif x_poco_event == "publish_page.published":
publish_page = payload.get("publish_page", {})
version = payload.get("version", {})
print(
"[Webhook] publish_page.published",
f"publish_key={publish_page.get('publish_key')}",
f"version_no={version.get('version_no')}",
)
else:
print("[Webhook] accepted event:", x_poco_event)
return {"ok": True}运行示例：
pip install fastapi uvicorn
uvicorn webhook_receiver:app --host 0.0.0.0 --port 8000生产环境需要替换的部分
幂等存储
PROCESSED_EVENT_IDS 在进程重启后会丢失，多实例之间也不共享。生产环境应使用数据库唯一键或 Redis 原子写入。幂等记录至少保存到你能够接受的重复处理窗口之后。
正确顺序是先可靠记录事件或入队，再返回 2xx。如果先返回成功再入队，进程崩溃可能导致事件丢失；如果业务处理完成后才返回，又容易触发超时和重复投递。
异步处理
接收端请求只完成：
- 验签与字段校验。
- 幂等登记。
- 写入可靠队列。
文件下载、IM 通知、第三方 API 调用和复杂数据库操作由队列消费者执行。
未知事件
部署新版本后可能出现接收端尚未处理的事件。只要验签和格式有效，建议记录并返回 2xx；不要返回 5xx 让 Poco 对一个业务上不支持的事件持续重试。根本解决方式是在 WebHook 配置中只勾选接收端支持的事件。
日志
建议记录 event_id、event_type、occurred_at、webhook.id、处理状态和耗时。不要记录签名密钥、API Key、密码或完整敏感附件内容。