Documentation

Backend pipeline with Convex

Receive AI webhooks in Convex HTTP actions, write data with mutations, and stream results to clients with real-time queries: zero polling required.

Overview

Convex is a reactive database platform where queries automatically update when data changes. Combine it with ModelRiver backend pipeline and you get an end-to-end reactive pipeline: AI generates data → webhook writes to Convex → your React components update instantly.

What you'll build:

  • A Convex HTTP action that receives ModelRiver webhooks
  • A mutation for writing AI-generated data
  • Real-time queries that update the UI automatically
  • Callback to ModelRiver with record details

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 handler)

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. Verify signature: HMAC-SHA256 of "<timestamp>.<raw JSON text of body.data>"
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 // Constant-time comparison
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; // the event payload is under body.data
51 const { type, event, ai_response, callback_url, customer_data, channel_id } = payload;
52 
53 // 2. Handle backend pipeline workflow
54 if (type === "task.ai_generated" && callback_url) {
55 try {
56 const aiData = ai_response?.data ?? {};
57 
58 // 3. Write to Convex database via mutation
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. Call back to 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(`Callback failed: ${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:", 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 frontend (real-time)

Convex queries are reactive: the UI updates automatically when the webhook writes new data:

TSX
1// src/App.tsx
2import { useQuery } from "convex/react";
3import { api } from "../convex/_generated/api";
4 
5export function AiContentFeed() {
6 // This query automatically updates when new records are inserted!
7 const items = useQuery(api.aiContent.list, { limit: 10 });
8 
9 if (items === undefined) return <p>Loading...</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}

No WebSocket setup, no polling, no subscriptions: Convex handles reactivity automatically.

Best practices

  1. Use HTTP actions for webhooks: Convex HTTP actions can receive external HTTP requests.
  2. Use mutations for writes: All database writes go through validated mutations.
  3. Leverage reactive queries: No need for separate WebSocket/polling infrastructure.
  4. Use Web Crypto API: Convex HTTP actions run in a V8 environment with built-in Web Crypto.
  5. Store environment variables: Use Convex dashboard settings for API keys and secrets.

Next steps