Documentation

Backend pipeline with Neon

Receive AI webhooks and write structured data to Neon's serverless Postgres: with branching, autoscaling, and a native serverless driver.

Overview

Neon provides serverless Postgres with branching, autoscaling to zero, and a WebSocket-based serverless driver that works in edge environments. Use it with backend pipeline to build fully serverless data pipelines where AI-generated data lands directly in Postgres.

What you'll build:

  • A webhook handler using Neon's serverless driver
  • Database schema for AI-generated content
  • Connection pooling for production workloads
  • Callback to ModelRiver with enriched records

Database schema

SQL
1-- Create via Neon console or psql
2CREATE TABLE ai_content (
3 id UUID DEFAULT gen_random_uuid() PRIMARY KEY,
4 title TEXT NOT NULL,
5 body TEXT,
6 category TEXT DEFAULT 'general',
7 metadata JSONB DEFAULT '{}',
8 channel_id TEXT,
9 event_name TEXT,
10 created_at TIMESTAMPTZ DEFAULT now(),
11 updated_at TIMESTAMPTZ DEFAULT now()
12);
13 
14CREATE INDEX idx_ai_content_channel ON ai_content(channel_id);
15CREATE INDEX idx_ai_content_event ON ai_content(event_name);
16CREATE INDEX idx_ai_content_created ON ai_content(created_at DESC);

Webhook handler (Node.js)

TYPESCRIPT
1// Using @neondatabase/serverless driver
2import { neon } from "@neondatabase/serverless";
3import crypto from "crypto";
4 
5const sql = neon(process.env.DATABASE_URL!);
6 
7function verifySignature(
8 rawBody: string,
9 timestamp: string,
10 signature: string,
11 secret: string
12): boolean {
13 // Signed string: "<timestamp>.<raw JSON text of body.data>"
14 const start = rawBody.indexOf('"data":') + '"data":'.length;
15 const end = rawBody.lastIndexOf(',"timestamp":');
16 if (start < 7 || end < start) return false;
17 
18 const expected = crypto
19 .createHmac("sha256", secret)
20 .update(`${timestamp}.${rawBody.slice(start, end)}`)
21 .digest("hex");
22 return (
23 signature.length === expected.length &&
24 crypto.timingSafeEqual(Buffer.from(signature), Buffer.from(expected))
25 );
26}
27 
28export async function handleWebhook(req: Request): Promise<Response> {
29 const signature = req.headers.get("x-modelriver-signature") ?? "";
30 const timestamp = req.headers.get("x-modelriver-timestamp") ?? "";
31 const rawBody = await req.text();
32 
33 // 1. Verify signature
34 if (!verifySignature(rawBody, timestamp, signature, process.env.MODELRIVER_WEBHOOK_SECRET!)) {
35 return new Response(JSON.stringify({ error: "Invalid signature" }), { status: 401 });
36 }
37 
38 const payload = JSON.parse(rawBody).data; // the event payload is under body.data
39 const { type, event, ai_response, callback_url, customer_data, channel_id } = payload;
40 
41 // 2. Handle backend pipeline workflow
42 if (type === "task.ai_generated" && callback_url) {
43 try {
44 const aiData = ai_response?.data ?? {};
45 
46 // 3. Write to Neon
47 const [record] = await sql`
48 INSERT INTO ai_content (title, body, category, metadata, channel_id, event_name)
49 VALUES (
50 ${aiData.title ?? "Untitled"},
51 ${aiData.description ?? aiData.body ?? ""},
52 ${customer_data?.category ?? "general"},
53 ${JSON.stringify({
54 ...customer_data,
55 ai_model: ai_response?.meta?.used_model,
56 ai_provider: ai_response?.meta?.used_provider,
57 })}::jsonb,
58 ${channel_id},
59 ${event}
60 )
61 RETURNING id, created_at
62 `;
63 
64 // 4. Call back to ModelRiver
65 await fetch(callback_url, {
66 method: "POST",
67 headers: {
68 Authorization: `Bearer ${process.env.MODELRIVER_API_KEY}`,
69 "Content-Type": "application/json",
70 },
71 body: JSON.stringify({
72 data: {
73 ...aiData,
74 id: record.id,
75 saved_at: record.created_at,
76 },
77 task_id: `neon_${record.id}`,
78 metadata: {
79 database: "neon",
80 table: "ai_content",
81 record_id: record.id,
82 },
83 }),
84 });
85 
86 return new Response(JSON.stringify({ received: true }), { status: 200 });
87 
88 } catch (error: any) {
89 console.error("Error:", error);
90 
91 await fetch(callback_url, {
92 method: "POST",
93 headers: {
94 Authorization: `Bearer ${process.env.MODELRIVER_API_KEY}`,
95 "Content-Type": "application/json",
96 },
97 body: JSON.stringify({
98 data: {
99 processing_status: "failed",
100 processing_error: error.message,
101 },
102 }),
103 });
104 
105 return new Response(JSON.stringify({ received: true }), { status: 200 });
106 }
107 }
108 
109 return new Response(JSON.stringify({ received: true }), { status: 200 });
110}

Vercel Edge Function example

Neon's serverless driver works natively in edge environments:

TYPESCRIPT
1// app/api/webhooks/modelriver/route.ts
2import { NextRequest, NextResponse } from "next/server";
3import { neon } from "@neondatabase/serverless";
4 
5export const runtime = "edge";
6 
7const sql = neon(process.env.DATABASE_URL!);
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 [record] = await sql`
19 INSERT INTO ai_content (title, body, metadata)
20 VALUES (${aiData.title}, ${aiData.description}, ${JSON.stringify(payload.customer_data)}::jsonb)
21 RETURNING id, created_at
22 `;
23 
24 await fetch(payload.callback_url, {
25 method: "POST",
26 headers: {
27 Authorization: `Bearer ${process.env.MODELRIVER_API_KEY}`,
28 "Content-Type": "application/json",
29 },
30 body: JSON.stringify({
31 data: { ...aiData, id: record.id, saved_at: record.created_at },
32 task_id: `neon_${record.id}`,
33 }),
34 });
35 
36 return NextResponse.json({ received: true });
37 }
38 
39 return NextResponse.json({ received: true });
40}

Drizzle ORM integration

For type-safe database access:

TYPESCRIPT
1// db/schema.ts
2import { pgTable, uuid, text, jsonb, timestamp } from "drizzle-orm/pg-core";
3 
4export const aiContent = pgTable("ai_content", {
5 id: uuid("id").defaultRandom().primaryKey(),
6 title: text("title").notNull(),
7 body: text("body"),
8 category: text("category").default("general"),
9 metadata: jsonb("metadata").default({}),
10 channelId: text("channel_id"),
11 eventName: text("event_name"),
12 createdAt: timestamp("created_at", { withTimezone: true }).defaultNow(),
13 updatedAt: timestamp("updated_at", { withTimezone: true }).defaultNow(),
14});
15 
16// In your webhook handler
17import { drizzle } from "drizzle-orm/neon-http";
18import { neon } from "@neondatabase/serverless";
19import { aiContent } from "./db/schema";
20 
21const sql = neon(process.env.DATABASE_URL!);
22const db = drizzle(sql);
23 
24const [record] = await db
25 .insert(aiContent)
26 .values({
27 title: aiData.title,
28 body: aiData.description,
29 category: customerData?.category,
30 metadata: customerData,
31 channelId,
32 eventName: event,
33 })
34 .returning();

Database branching

Test AI data schemas without affecting production:

Bash
# Create a branch
neonctl branches create --name ai-content-v2
 
# Get connection string for the branch
neonctl connection-string ai-content-v2
 
# Test your webhook handler against the branch
MODELRIVER_WEBHOOK_SECRET=... DATABASE_URL=<branch-url> npm run dev
 
# Merge when satisfied

Best practices

  1. Use the serverless driver: @neondatabase/serverless uses WebSocket for edge compatibility.
  2. Use tagged template literals: Neon's sql function prevents SQL injection automatically.
  3. Use JSONB for metadata: Flexible schema for varying AI responses.
  4. Use database branching: Test schema changes safely before production.
  5. Use Drizzle ORM: Type-safe queries with automatic schema inference.

Next steps