Supabase：后端管道 – ModelRiver Docs

概述

Supabase 在一个平台中结合了 Postgres、实时订阅和 Edge Functions。这使其成为后端管道的理想选择：您的 Edge Function 接收 Webhook，直接将 AI 数据写入 Postgres，然后 Supabase Realtime 将变更推送到已连接的客户端。

您将构建的内容：

一个用于接收 ModelRiver Webhooks 的 Supabase Edge Function
用于存储 AI 生成内容的数据库架构 (Schema)
用于即时前端更新的实时订阅
携带增强后的数据库记录回调 ModelRiver

数据库架构 (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-- 启用实时功能
16alter publication supabase_realtime add table ai_content;
17 
18-- 行级安全性 (RLS)
19alter table ai_content enable row level security;
20 
21create policy "用户可以查看自己的内容"
22  on ai_content for select
23  using (metadata->>'user_id' = auth.uid()::text);

Edge Function Webhook 处理程序

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. 验证签名：对 "<timestamp>.<body.data 的原始 JSON 文本>" 计算 HMAC-SHA256
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  // 常量时间比较
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; // 事件负载位于 body.data 下
50  const { type, event, ai_response, callback_url, customer_data, channel_id } = payload;
51 
52  // 2. 处理后端管道的工作流
53  if (type === "task.ai_generated" && callback_url) {
54    try {
55      const aiData = ai_response?.data ?? {};
56 
57      // 3. 写入 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. 携带数据库记录回调 ModelRiver
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(`回调失败: ${callbackResponse.status}`);
102      }
103 
104      return new Response(JSON.stringify({ received: true }), { status: 200 });
105 
106    } catch (error) {
107      console.error("错误:", error);
108 
109      // 发送错误回调
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});

部署 Edge Function

Bash

supabase functions deploy modelriver-webhook

设置环境变量 (Secrets)

Bash

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

带有实时功能的前端

使用 Supabase Realtime 订阅实时监听新记录：