Next.js：后端管道 – ModelRiver Docs

概述

Next.js 的 API 路由（App Router）是处理 ModelRiver 后端管道 Webhooks 的理想方案。您可以在接收结构化 AI 输出、运行服务端逻辑以及发起回调的整个过程中，都复用服务于前端的同一个 Next.js 应用。

您将构建的内容：

一个用于接收 ModelRiver 发出的 AI 生成数据的 Webhook 端点
用于安全处理 Webhook 的签名验证中间件
具有回调 ModelRiver 功能的后台处理逻辑
一个通过 WebSocket 接收最终结果的 React 前端

快速开始

安装依赖

Bash

npx create-next-app@latest my-ai-app --typescript --app
cd my-ai-app
npm install crypto

环境变量

Bash

# .env.local
MODELRIVER_API_KEY=mr_live_YOUR_API_KEY
MODELRIVER_WEBHOOK_SECRET=your_webhook_secret

Webhook 处理程序

API 路由

TYPESCRIPT

1// app/api/webhooks/modelriver/route.ts
2import { NextRequest, NextResponse } from "next/server";
3import crypto from "crypto";
4 
5function verifySignature(
6  rawBody: string,
7  timestamp: string,
8  signature: string,
9  secret: string
10): boolean {
11  // 签名字符串："<timestamp>.<原始请求体中 "data" 值的 JSON 文本>"（切勿重新编码）
12  const start = rawBody.indexOf('"data":');
13  const end = rawBody.lastIndexOf(',"timestamp":');
14  if (start === -1 || end === -1) return false;
15 
16  const expected = crypto
17    .createHmac("sha256", secret)
18    .update(`${timestamp}.${rawBody.slice(start + '"data":'.length, end)}`)
19    .digest("hex");
20 
21  const a = Buffer.from(signature);
22  const b = Buffer.from(expected);
23  return a.length === b.length && crypto.timingSafeEqual(a, b);
24}
25 
26export async function POST(request: NextRequest) {
27  const signature = request.headers.get("x-modelriver-signature");
28  const timestamp = request.headers.get("x-modelriver-timestamp") ?? "";
29  const rawBody = await request.text(); // 原始请求体：签名验证需要它
30 
31  // 1. 验证 Webhook 签名
32  if (!signature || !verifySignature(rawBody, timestamp, signature, process.env.MODELRIVER_WEBHOOK_SECRET!)) {
33    return NextResponse.json({ error: "Invalid signature" }, { status: 401 });
34  }
35 
36  const payload = JSON.parse(rawBody).data; // 事件载荷位于 "data" 之下
37  const { type, event, ai_response, callback_url, customer_data } = payload;
38 
39  // 2. 检查这是否是一个后端管道的工作流
40  if (type === "task.ai_generated" && callback_url) {
41    // 立即响应：在后台处理
42    // 注意：Next.js 在 Serverless 环境中不直接支持真正的后台任务，
43    // 因此我们使用 waitUntil 或在响应前进行处理
44    await processAndCallback(event, ai_response, callback_url, customer_data);
45 
46    return NextResponse.json({ received: true });
47  }
48 
49  // 标准 Webhook（无事件名称）
50  console.log("标准 Webhook:", payload);
51  return NextResponse.json({ received: true });
52}
53 
54async function processAndCallback(
55  event: string,
56  aiResponse: { data: Record<string, unknown> },
57  callbackUrl: string,
58  customerData: Record<string, unknown>
59) {
60  try {
61    let enrichedData = { ...aiResponse.data };
62 
63    // 3. 基于事件的自定义业务逻辑
64    if (event === "content_ready") {
65      // 保存到数据库
66      const savedRecord = await saveToDatabase(aiResponse.data);
67      enrichedData = {
68        ...enrichedData,
69        id: savedRecord.id,
70        slug: generateSlug(aiResponse.data.title as string),
71        saved_at: new Date().toISOString(),
72      };
73    }
74 
75    if (event === "review_complete") {
76      // 推送到外部 API
77      await postToExternalService(aiResponse.data);
78      enrichedData.posted = true;
79    }
80 
81    // 4. 将增强后的数据回调给 ModelRiver
82    const response = await fetch(callbackUrl, {
83      method: "POST",
84      headers: {
85        "Authorization": `Bearer ${process.env.MODELRIVER_API_KEY}`,
86        "Content-Type": "application/json",
87      },
88      body: JSON.stringify({
89        data: enrichedData,
90        task_id: `task_${Date.now()}`,
91        metadata: {
92          processed_at: new Date().toISOString(),
93          event,
94        },
95      }),
96    });
97 
98    if (!response.ok) {
99      throw new Error(`回调失败: ${response.status}`);
100    }
101 
102    console.log(`✅ 已针对事件发送回调: ${event}`);
103  } catch (error) {
104    console.error("处理 Webhook 时出错:", error);
105 
106    // 发送错误回调
107    await fetch(callbackUrl, {
108      method: "POST",
109      headers: {
110        "Authorization": `Bearer ${process.env.MODELRIVER_API_KEY}`,
111        "Content-Type": "application/json",
112      },
113      body: JSON.stringify({
114        data: {
115          processing_status: "failed",
116          processing_error: error instanceof Error ? error.message : "未知错误",
117        },
118      }),
119    });
120  }
121}
122 
123// 您的业务逻辑函数
124async function saveToDatabase(data: Record<string, unknown>) {
125  // 替换为您实际的数据库逻辑（Prisma、Drizzle 等）
126  return { id: `rec_${Date.now()}`, ...data };
127}
128 
129function generateSlug(title: string): string {
130  return title.toLowerCase().replace(/[^a-z0-9]+/g, "-").replace(/(^-|-$)/g, "");
131}
132 
133async function postToExternalService(data: Record<string, unknown>) {
134  // 替换为您实际的 API 调用
135  console.log("正在推送到外部服务:", data);
136}

触发异步请求

Server action

TYPESCRIPT

1// app/actions/ai.ts
2"use server";
3 
4export async function triggerContentGeneration(prompt: string) {
5  const response = await fetch("https://api.modelriver.com/v1/ai/async", {
6    method: "POST",
7    headers: {
8      "Authorization": `Bearer ${process.env.MODELRIVER_API_KEY}`,
9      "Content-Type": "application/json",
10    },
11    body: JSON.stringify({
12      model: "content_generator",
13      messages: [{ role: "user", content: prompt }],
14    }),
15  });
16 
17  const data = await response.json();
18  return {
19    channelId: data.channel_id,
20    wsToken: data.ws_token,
21    websocketUrl: data.websocket_url,
22    websocketChannel: data.websocket_channel,
23  };
24}

带有 WebSocket 的 React 组件

TSX

1// app/components/AiContentGenerator.tsx
2"use client";
3 
4import { useState, useEffect } from "react";
5import { triggerContentGeneration } from "@/app/actions/ai";
6 
7export function AiContentGenerator() {
8  const [result, setResult] = useState<Record<string, unknown> | null>(null);
9  const [status, setStatus] = useState<"idle" | "processing" | "complete">("idle");
10 
11  async function handleGenerate(prompt: string) {
12    setStatus("processing");
13 
14    const { channelId, wsToken, websocketChannel } = await triggerContentGeneration(prompt);
15 
16    // 连接 WebSocket 以获取实时结果
17    const ws = new WebSocket(
18      `wss://api.modelriver.com/socket/websocket?token=${encodeURIComponent(wsToken)}`
19    );
20 
21    ws.onopen = () => {
22      ws.send(JSON.stringify({
23        topic: websocketChannel,
24        event: "phx_join",
25        payload: {},
26        ref: "1",
27      }));
28    };
29 
30    ws.onmessage = (event) => {
31      const msg = JSON.parse(event.data);
32 
33      if (msg.event === "response") {
34        setResult(msg.payload.data);
35        setStatus("complete");
36        ws.close();
37      }
38    };
39  }
40 
41  return (
42    <div>
43      <button onClick={() => handleGenerate("生成产品描述")}>
44        {status === "processing" ? "正在处理..." : "生成"}
45      </button>
46      {result && <pre>{JSON.stringify(result, null, 2)}</pre>}
47    </div>
48  );
49}

Edge 运行时

对于 Vercel Edge Functions，Webhook 处理程序的工作方式相同：只需添加 edge runtime 的导出即可：

TYPESCRIPT

1// app/api/webhooks/modelriver/route.ts
2export const runtime = "edge";
3 
4// ... 与上面相同的处理程序代码，使用 Web Crypto API 替代 Node crypto 进行签名验证

注意： 在 Edge 上进行 HMAC 验证时，请使用 crypto.subtle.importKey 和 crypto.subtle.sign 替代 Node.js 的 crypto 模块。签名的仍是同一个 <timestamp>.<data 的 JSON 文本> 字符串，并将结果编码为十六进制。

最佳实践

快速响应：在处理繁重逻辑之前，先返回 200。在 Serverless 环境中，请使用 Vercel 的 waitUntil 执行后台工作。
验证每个 Webhook：始终校验 x-modelriver-signature 请求头。
使用 Server Actions 进行触发：将 API 密钥保留在服务端。
处理超时：您的回调必须在 10 分钟内到达 ModelRiver。
记录处理步骤：使用结构化日志记录，以便调试后端管道的工作流。

下一步

Nuxt.js 后端管道指南：Vue.js 的备选方案
Webhooks 参考：签名验证和重试政策
Next.js 集成：标准 ModelRiver + Next.js 的用法

Next.js 的后端管道

概述

快速开始

安装依赖

环境变量

Webhook 处理程序

API 路由

触发异步请求

Server action

带有 WebSocket 的 React 组件

Edge 运行时

最佳实践

下一步