Documentation

Backend pipeline with PlanetScale

Receive AI webhooks and write structured data directly to PlanetScale's serverless MySQL: with branching for safe schema evolution.

Overview

PlanetScale provides a serverless MySQL-compatible database powered by Vitess. Its branching model lets you iterate on schemas safely, while the serverless driver works natively in edge environments: making it ideal for webhook-driven AI data pipelines.

What you'll build:

  • A webhook handler that writes AI data to PlanetScale
  • Database schema for AI-generated content
  • Serverless driver usage for edge/serverless environments
  • Callback to ModelRiver with enriched records

Database schema

SQL
1-- Create via PlanetScale console or 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 handler (Node.js)

TYPESCRIPT
1// Using @planetscale/database serverless driver
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 // Signed string: "<timestamp>.<raw JSON text of body.data>"
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. Verify signature
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; // the event payload is under body.data
43 const { type, event, ai_response, callback_url, customer_data, channel_id } = payload;
44 
45 // 2. Handle backend pipeline workflow
46 if (type === "task.ai_generated" && callback_url) {
47 try {
48 const aiData = ai_response?.data ?? {};
49 
50 // 3. Write to 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. Call back to 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:", 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 route example

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 connection string
7});
8 
9export async function POST(request: NextRequest) {
10 const rawBody = await request.text();
11 // ... verify x-modelriver-signature against rawBody (see above) ...
12 
13 const payload = JSON.parse(rawBody).data; // the event payload is under 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}

Schema branching

PlanetScale's branching model lets you evolve your AI content schema safely:

Bash
# Create a development branch
pscale branch create my-db ai-content-v2
 
# Add new columns
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;
 
# Create deploy request
pscale deploy-request create my-db ai-content-v2
 
# Deploy after review
pscale deploy-request deploy my-db <id>

Best practices

  1. Use the serverless driver: @planetscale/database works in edge/serverless environments.
  2. Use parameterized queries: Always use ? placeholders to prevent SQL injection.
  3. Store metadata as JSON: Flexible schema for varying AI response shapes.
  4. Use database branching: Test schema changes with AI data before deploying to production.
  5. Index channel_id: Enable fast lookups for event tracking.

Next steps