文档

Convex 的后端管道

在 Convex HTTP Actions 中接收 AI Webhooks，通过 Mutations 写入数据，并利用实时查询将结果流式推送给客户端：无需任何轮询。

概述

Convex 是一个响应式数据库平台，当数据发生变化时，查询会自动更新。将其与 ModelRiver 后端管道 结合，您将获得一个端到端的响应式管道：AI 生成数据 → Webhook 写入 Convex → 您的 React 组件立即更新。

您将构建的内容：

  • 一个用于接收 ModelRiver Webhooks 的 Convex HTTP Action
  • 一个用于写入 AI 生成数据的 Mutation
  • 能够自动更新 UI 的实时查询 (Real-time queries)
  • 带有记录详情的回调 ModelRiver

架构 (Schema)

TYPESCRIPT
1// convex/schema.ts
2import { defineSchema, defineTable } from "convex/server";
3import { v } from "convex/values";
4 
5export default defineSchema({
6 aiContent: defineTable({
7 title: v.string(),
8 body: v.optional(v.string()),
9 category: v.string(),
10 metadata: v.any(),
11 channelId: v.optional(v.string()),
12 eventName: v.optional(v.string()),
13 source: v.string(),
14 })
15 .index("by_channel", ["channelId"])
16 .index("by_event", ["eventName"])
17 .index("by_creation", ["_creationTime"]),
18});

Mutation

TYPESCRIPT
1// convex/aiContent.ts
2import { mutation, query } from "./_generated/server";
3import { v } from "convex/values";
4 
5export const insert = mutation({
6 args: {
7 title: v.string(),
8 body: v.optional(v.string()),
9 category: v.string(),
10 metadata: v.any(),
11 channelId: v.optional(v.string()),
12 eventName: v.optional(v.string()),
13 },
14 handler: async (ctx, args) => {
15 const id = await ctx.db.insert("aiContent", {
16 ...args,
17 source: "modelriver",
18 });
19 return id;
20 },
21});
22 
23export const list = query({
24 args: {
25 limit: v.optional(v.number()),
26 },
27 handler: async (ctx, args) => {
28 return await ctx.db
29 .query("aiContent")
30 .order("desc")
31 .take(args.limit ?? 20);
32 },
33});
34 
35export const getByChannel = query({
36 args: {
37 channelId: v.string(),
38 },
39 handler: async (ctx, args) => {
40 return await ctx.db
41 .query("aiContent")
42 .withIndex("by_channel", (q) => q.eq("channelId", args.channelId))
43 .first();
44 },
45});

HTTP Action (Webhook 处理程序)

TYPESCRIPT
1// convex/http.ts
2import { httpRouter } from "convex/server";
3import { httpAction } from "./_generated/server";
4import { api } from "./_generated/api";
5 
6const http = httpRouter();
7 
8http.route({
9 path: "/webhooks/modelriver",
10 method: "POST",
11 handler: httpAction(async (ctx, request) => {
12 const signature = request.headers.get("x-modelriver-signature") ?? "";
13 const timestamp = request.headers.get("x-modelriver-timestamp") ?? "";
14 const rawBody = await request.text();
15 
16 // 1. 验证签名：对 "<timestamp>.<body.data 的原始 JSON 文本>" 计算 HMAC-SHA256
17 const start = rawBody.indexOf('"data":') + '"data":'.length;
18 const end = rawBody.lastIndexOf(',"timestamp":');
19 const webhookSecret = process.env.MODELRIVER_WEBHOOK_SECRET ?? "";
20 const encoder = new TextEncoder();
21 const key = await crypto.subtle.importKey(
22 "raw",
23 encoder.encode(webhookSecret),
24 { name: "HMAC", hash: "SHA-256" },
25 false,
26 ["sign"]
27 );
28 const sig = await crypto.subtle.sign(
29 "HMAC",
30 key,
31 encoder.encode(`${timestamp}.${rawBody.slice(start, end)}`)
32 );
33 const expected = Array.from(new Uint8Array(sig))
34 .map((b) => b.toString(16).padStart(2, "0"))
35 .join("");
36 
37 // 常量时间比较
38 let diff = expected.length ^ signature.length;
39 for (let i = 0; i < expected.length; i++) {
40 diff |= expected.charCodeAt(i) ^ signature.charCodeAt(i);
41 }
42 
43 if (start < 7 || end < start || diff !== 0) {
44 return new Response(JSON.stringify({ error: "Invalid signature" }), {
45 status: 401,
46 headers: { "Content-Type": "application/json" },
47 });
48 }
49 
50 const payload = JSON.parse(rawBody).data; // 事件负载位于 body.data 下
51 const { type, event, ai_response, callback_url, customer_data, channel_id } = payload;
52 
53 // 2. 处理后端管道的工作流
54 if (type === "task.ai_generated" && callback_url) {
55 try {
56 const aiData = ai_response?.data ?? {};
57 
58 // 3. 通过 Mutation 写入 Convex 数据库
59 const recordId = await ctx.runMutation(api.aiContent.insert, {
60 title: aiData.title ?? "Untitled",
61 body: aiData.description ?? aiData.body ?? "",
62 category: customer_data?.category ?? "general",
63 metadata: {
64 ...customer_data,
65 ai_model: ai_response?.meta?.used_model,
66 ai_provider: ai_response?.meta?.used_provider,
67 },
68 channelId: channel_id,
69 eventName: event,
70 });
71 
72 // 4. 回调 ModelRiver
73 const callbackResponse = await fetch(callback_url, {
74 method: "POST",
75 headers: {
76 Authorization: `Bearer ${process.env.MODELRIVER_API_KEY}`,
77 "Content-Type": "application/json",
78 },
79 body: JSON.stringify({
80 data: {
81 ...aiData,
82 id: recordId,
83 saved_at: new Date().toISOString(),
84 },
85 task_id: `convex_${recordId}`,
86 metadata: {
87 database: "convex",
88 table: "aiContent",
89 record_id: recordId,
90 },
91 }),
92 });
93 
94 if (!callbackResponse.ok) {
95 throw new Error(`回调失败: ${callbackResponse.status}`);
96 }
97 
98 return new Response(JSON.stringify({ received: true }), {
99 status: 200,
100 headers: { "Content-Type": "application/json" },
101 });
102 
103 } catch (error: any) {
104 console.error("错误:", error);
105 
106 await fetch(callback_url, {
107 method: "POST",
108 headers: {
109 Authorization: `Bearer ${process.env.MODELRIVER_API_KEY}`,
110 "Content-Type": "application/json",
111 },
112 body: JSON.stringify({
113 data: {
114 processing_status: "failed",
115 processing_error: error.message,
116 },
117 }),
118 });
119 
120 return new Response(JSON.stringify({ received: true }), {
121 status: 200,
122 headers: { "Content-Type": "application/json" },
123 });
124 }
125 }
126 
127 return new Response(JSON.stringify({ received: true }), {
128 status: 200,
129 headers: { "Content-Type": "application/json" },
130 });
131 }),
132});
133 
134export default http;

React 前端 (实时)

Convex 的查询是响应式的：当 Webhook 写入新数据时，UI 会自动更新：

TSX
1// src/App.tsx
2import { useQuery } from "convex/react";
3import { api } from "../convex/_generated/api";
4 
5export function AiContentFeed() {
6 // 当插入新记录时，此查询会自动更新！
7 const items = useQuery(api.aiContent.list, { limit: 10 });
8 
9 if (items === undefined) return <p>正在加载...</p>;
10 
11 return (
12 <div className="space-y-4">
13 {items.map((item) => (
14 <div key={item._id} className="p-4 border rounded-lg">
15 <h3 className="font-bold text-lg">{item.title}</h3>
16 <p className="text-zinc-600">{item.body}</p>
17 <div className="flex gap-2 mt-2">
18 <span className="text-xs bg-blue-100 text-blue-800 px-2 py-1 rounded">
19 {item.category}
20 </span>
21 <span className="text-xs text-zinc-400">
22 {new Date(item._creationTime).toLocaleString()}
23 </span>
24 </div>
25 </div>
26 ))}
27 </div>
28 );
29}

无需 WebSocket 配置、无需轮询、无需订阅：Convex 会自动处理响应性。

最佳实践

  1. 为 Webhooks 使用 HTTP Actions：Convex HTTP actions 可以接收外部 HTTP 请求。
  2. 为写入使用 Mutations：所有的数据库写入都通过经过验证的 mutations 进行。
  3. 利用响应式查询 (Reactive Queries)：无需独立的 WebSocket/轮询基础设施。
  4. 使用 Web Crypto API：Convex HTTP actions 运行在内置 Web Crypto 的 V8 环境中。
  5. 存储环境变量：使用 Convex 控制面板中的设置来存储 API 密钥和密钥。

下一步