文档

PlanetScale 的后端管道

接收 AI Webhooks 并将结构化数据直接写入 PlanetScale 的无服务器 MySQL：利用分支功能实现安全的架构演进。

概述

PlanetScale 提供了由 Vitess 驱动的无服务器 MySQL 兼容数据库。其分支模型 (Branching model) 让您可以安全地迭代数据库架构，而无服务器驱动程序则可以在 Edge 环境中原生运行：这使其成为 Webhook 驱动的 AI 数据管道的理想选择。

您将构建的内容：

  • 一个将 AI 数据写入 PlanetScale 的 Webhook 处理程序
  • 用于存储 AI 生成内容的数据库架构 (Schema)
  • 适用于 Edge/Serverless 环境的无服务器驱动程序用法
  • 携带增强后的记录回调 ModelRiver

数据库架构 (Schema)

SQL
1-- 通过 PlanetScale 控制台或 CLI 创建
2CREATE TABLE ai_content (
3 id BIGINT AUTO_INCREMENT PRIMARY KEY,
4 title VARCHAR(500) NOT NULL,
5 body TEXT,
6 category VARCHAR(100) DEFAULT 'general',
7 metadata JSON,
8 channel_id VARCHAR(36),
9 event_name VARCHAR(100),
10 created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
11 updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
12 INDEX idx_channel (channel_id),
13 INDEX idx_event (event_name)
14);

Webhook 处理程序 (Node.js)

TYPESCRIPT
1// 使用 @planetscale/database 无服务器驱动程序
2import { connect } from "@planetscale/database";
3import crypto from "crypto";
4 
5const db = connect({
6 host: process.env.PLANETSCALE_HOST,
7 username: process.env.PLANETSCALE_USERNAME,
8 password: process.env.PLANETSCALE_PASSWORD,
9});
10 
11function verifySignature(
12 rawBody: string,
13 timestamp: string,
14 signature: string,
15 secret: string
16): boolean {
17 // 签名字符串："<timestamp>.<body.data 的原始 JSON 文本>"
18 const start = rawBody.indexOf('"data":') + '"data":'.length;
19 const end = rawBody.lastIndexOf(',"timestamp":');
20 if (start < 7 || end < start) return false;
21 
22 const expected = crypto
23 .createHmac("sha256", secret)
24 .update(`${timestamp}.${rawBody.slice(start, end)}`)
25 .digest("hex");
26 return (
27 signature.length === expected.length &&
28 crypto.timingSafeEqual(Buffer.from(signature), Buffer.from(expected))
29 );
30}
31 
32export async function handleWebhook(req: Request): Promise<Response> {
33 const signature = req.headers.get("x-modelriver-signature") ?? "";
34 const timestamp = req.headers.get("x-modelriver-timestamp") ?? "";
35 const rawBody = await req.text();
36 
37 // 1. 验证签名
38 if (!verifySignature(rawBody, timestamp, signature, process.env.MODELRIVER_WEBHOOK_SECRET!)) {
39 return new Response(JSON.stringify({ error: "Invalid signature" }), { status: 401 });
40 }
41 
42 const payload = JSON.parse(rawBody).data; // 事件负载位于 body.data 下
43 const { type, event, ai_response, callback_url, customer_data, channel_id } = payload;
44 
45 // 2. 处理后端管道的工作流
46 if (type === "task.ai_generated" && callback_url) {
47 try {
48 const aiData = ai_response?.data ?? {};
49 
50 // 3. 写入 PlanetScale
51 const result = await db.execute(
52 `INSERT INTO ai_content (title, body, category, metadata, channel_id, event_name)
53 VALUES (?, ?, ?, ?, ?, ?)`,
54 [
55 aiData.title ?? "Untitled",
56 aiData.description ?? aiData.body ?? "",
57 customer_data?.category ?? "general",
58 JSON.stringify({
59 ...customer_data,
60 ai_model: ai_response?.meta?.used_model,
61 ai_provider: ai_response?.meta?.used_provider,
62 }),
63 channel_id,
64 event,
65 ]
66 );
67 
68 const recordId = result.insertId;
69 
70 // 4. 回调 ModelRiver
71 await fetch(callback_url, {
72 method: "POST",
73 headers: {
74 Authorization: `Bearer ${process.env.MODELRIVER_API_KEY}`,
75 "Content-Type": "application/json",
76 },
77 body: JSON.stringify({
78 data: {
79 ...aiData,
80 id: recordId,
81 saved_at: new Date().toISOString(),
82 },
83 task_id: `planetscale_${recordId}`,
84 metadata: {
85 database: "planetscale",
86 table: "ai_content",
87 record_id: recordId,
88 },
89 }),
90 });
91 
92 return new Response(JSON.stringify({ received: true }), { status: 200 });
93 
94 } catch (error: any) {
95 console.error("错误:", error);
96 
97 await fetch(callback_url, {
98 method: "POST",
99 headers: {
100 Authorization: `Bearer ${process.env.MODELRIVER_API_KEY}`,
101 "Content-Type": "application/json",
102 },
103 body: JSON.stringify({
104 data: {
105 processing_status: "failed",
106 processing_error: error.message,
107 },
108 }),
109 });
110 
111 return new Response(JSON.stringify({ received: true }), { status: 200 });
112 }
113 }
114 
115 return new Response(JSON.stringify({ received: true }), { status: 200 });
116}

Next.js API 路由示例

TYPESCRIPT
1// app/api/webhooks/modelriver/route.ts
2import { NextRequest, NextResponse } from "next/server";
3import { connect } from "@planetscale/database";
4 
5const db = connect({
6 url: process.env.DATABASE_URL, // PlanetScale 连接字符串
7});
8 
9export async function POST(request: NextRequest) {
10 const rawBody = await request.text();
11 // ... 使用 rawBody 验证 x-modelriver-signature（见上文） ...
12 
13 const payload = JSON.parse(rawBody).data; // 事件负载位于 body.data 下
14 
15 if (payload.type === "task.ai_generated" && payload.callback_url) {
16 const aiData = payload.ai_response?.data ?? {};
17 
18 const result = await db.execute(
19 "INSERT INTO ai_content (title, body, metadata) VALUES (?, ?, ?)",
20 [aiData.title, aiData.description, JSON.stringify(payload.customer_data)]
21 );
22 
23 await fetch(payload.callback_url, {
24 method: "POST",
25 headers: {
26 Authorization: `Bearer ${process.env.MODELRIVER_API_KEY}`,
27 "Content-Type": "application/json",
28 },
29 body: JSON.stringify({
30 data: { ...aiData, id: result.insertId },
31 task_id: `ps_${result.insertId}`,
32 }),
33 });
34 
35 return NextResponse.json({ received: true });
36 }
37 
38 return NextResponse.json({ received: true });
39}

架构分支 (Branching)

PlanetScale 的分支模型让您可以安全地演进 AI 内容架构：

Bash
# 创建开发分支
pscale branch create my-db ai-content-v2
 
# 添加新列
pscale shell my-db ai-content-v2
> ALTER TABLE ai_content ADD COLUMN embedding_id VARCHAR(100);
> ALTER TABLE ai_content ADD COLUMN quality_score FLOAT;
 
# 创建部署请求
pscale deploy-request create my-db ai-content-v2
 
# 审核后进行部署
pscale deploy-request deploy my-db <id>

最佳实践

  1. 使用无服务器驱动程序：@planetscale/database 可以在 Edge/Serverless 环境中运行。
  2. 使用参数化查询：始终使用 ? 占位符以防止 SQL 注入。
  3. 将元数据存储为 JSON：灵活的架构，适应各种 AI 响应形式。
  4. 使用数据库分支：在部署到生产环境之前，使用 AI 数据测试架构变更。
  5. 在 channel_id 上建立索引：实现快速的事件追踪查询。

下一步