Supabase: Backend pipeline – ModelRiver Docs

Overview

Supabase combines Postgres, real-time subscriptions, and Edge Functions in one platform. This makes it ideal for backend pipeline: your Edge Function receives the webhook, writes AI data directly to Postgres, and Supabase Realtime pushes the changes to connected clients.

What you'll build:

A Supabase Edge Function that receives ModelRiver webhooks
Database schema for storing AI-generated content
Real-time subscription for instant frontend updates
Callback to ModelRiver with enriched database records

Database schema

SQL

1-- supabase/migrations/001_ai_content.sql
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  source text default 'modelriver',
11  created_at timestamptz default now(),
12  updated_at timestamptz default now()
13);
14 
15-- Enable real-time
16alter publication supabase_realtime add table ai_content;
17 
18-- Row-level security
19alter table ai_content enable row level security;
20 
21create policy "Users can view their content"
22  on ai_content for select
23  using (metadata->>'user_id' = auth.uid()::text);

Edge Function webhook handler

TYPESCRIPT

1// supabase/functions/modelriver-webhook/index.ts
2import { serve } from "https://deno.land/[email protected]/http/server.ts";
3import { createClient } from "https://esm.sh/@supabase/supabase-js@2";
4 
5const supabase = createClient(
6  Deno.env.get("SUPABASE_URL")!,
7  Deno.env.get("SUPABASE_SERVICE_ROLE_KEY")!
8);
9 
10serve(async (req) => {
11  if (req.method !== "POST") {
12    return new Response("Method not allowed", { status: 405 });
13  }
14 
15  const signature = req.headers.get("x-modelriver-signature") ?? "";
16  const timestamp = req.headers.get("x-modelriver-timestamp") ?? "";
17  const rawBody = await req.text();
18  const webhookSecret = Deno.env.get("MODELRIVER_WEBHOOK_SECRET") ?? "";
19 
20  // 1. Verify signature: HMAC-SHA256 of "<timestamp>.<raw JSON text of body.data>"
21  const start = rawBody.indexOf('"data":') + '"data":'.length;
22  const end = rawBody.lastIndexOf(',"timestamp":');
23  const key = await crypto.subtle.importKey(
24    "raw",
25    new TextEncoder().encode(webhookSecret),
26    { name: "HMAC", hash: "SHA-256" },
27    false,
28    ["sign"]
29  );
30  const sig = await crypto.subtle.sign(
31    "HMAC",
32    key,
33    new TextEncoder().encode(`${timestamp}.${rawBody.slice(start, end)}`)
34  );
35  const expected = Array.from(new Uint8Array(sig))
36    .map((b) => b.toString(16).padStart(2, "0"))
37    .join("");
38 
39  // Constant-time comparison
40  let diff = expected.length ^ signature.length;
41  for (let i = 0; i < expected.length; i++) {
42    diff |= expected.charCodeAt(i) ^ signature.charCodeAt(i);
43  }
44 
45  if (start < 7 || end < start || diff !== 0) {
46    return new Response(JSON.stringify({ error: "Invalid signature" }), { status: 401 });
47  }
48 
49  const payload = JSON.parse(rawBody).data; // the event payload is under body.data
50  const { type, event, ai_response, callback_url, customer_data, channel_id } = payload;
51 
52  // 2. Handle backend pipeline workflow
53  if (type === "task.ai_generated" && callback_url) {
54    try {
55      const aiData = ai_response?.data ?? {};
56 
57      // 3. Write to Supabase
58      const { data: record, error } = await supabase
59        .from("ai_content")
60        .insert({
61          title: aiData.title ?? "Untitled",
62          body: aiData.description ?? aiData.body ?? "",
63          category: customer_data?.category ?? "general",
64          metadata: {
65            ...customer_data,
66            ai_model: ai_response?.meta?.used_model,
67            ai_provider: ai_response?.meta?.used_provider,
68          },
69          channel_id,
70          event_name: event,
71        })
72        .select()
73        .single();
74 
75      if (error) throw error;
76 
77      // 4. Call back to ModelRiver with database record
78      const callbackResponse = await fetch(callback_url, {
79        method: "POST",
80        headers: {
81          Authorization: `Bearer ${Deno.env.get("MODELRIVER_API_KEY")}`,
82          "Content-Type": "application/json",
83        },
84        body: JSON.stringify({
85          data: {
86            ...aiData,
87            id: record.id,
88            saved_at: record.created_at,
89            supabase_url: `${Deno.env.get("SUPABASE_URL")}/rest/v1/ai_content?id=eq.${record.id}`,
90          },
91          task_id: `supabase_${record.id}`,
92          metadata: {
93            database: "supabase",
94            table: "ai_content",
95            record_id: record.id,
96          },
97        }),
98      });
99 
100      if (!callbackResponse.ok) {
101        throw new Error(`Callback failed: ${callbackResponse.status}`);
102      }
103 
104      return new Response(JSON.stringify({ received: true }), { status: 200 });
105 
106    } catch (error) {
107      console.error("Error:", error);
108 
109      // Send error callback
110      await fetch(callback_url, {
111        method: "POST",
112        headers: {
113          Authorization: `Bearer ${Deno.env.get("MODELRIVER_API_KEY")}`,
114          "Content-Type": "application/json",
115        },
116        body: JSON.stringify({
117          data: {
118            processing_status: "failed",
119            processing_error: error.message,
120          },
121        }),
122      });
123 
124      return new Response(JSON.stringify({ received: true }), { status: 200 });
125    }
126  }
127 
128  return new Response(JSON.stringify({ received: true }), { status: 200 });
129});

Deploy the Edge Function

Bash

supabase functions deploy modelriver-webhook

Set secrets

Bash

supabase secrets set MODELRIVER_API_KEY=mr_live_YOUR_API_KEY
supabase secrets set MODELRIVER_WEBHOOK_SECRET=your_webhook_secret

Frontend with Realtime

Listen for new records in real time using Supabase Realtime subscriptions: