database

Supabase 平台与 PostgreSQL 边缘函数实践

深入讲解 Supabase 平台架构与 PostgreSQL 边缘函数的实战方案。涵盖 Supabase 核心组件（PostgREST、Realtime、Storage、Auth）的整体架构、Row Level Security 策略与认证体系的联动设计、Deno 边缘函数的编写与部署、Database Webhooks 触发机制、pgvector 向量检索集成、Realtime 实时订阅的实现原理，以及结合自动生成的 REST API 进行前后端开发的全栈实践。

Leeting Yan 8 分钟阅读 3574 字

Supabase 的定位是 “开源 Firebase 替代方案”，但底层核心并非独立的文档数据库，而就是一个功能完备的 PostgreSQL 实例。它用 PostgREST 把 PostgreSQL 直接暴露为 REST API，用 Realtime 监听 WAL 变化推送实时消息，用 Edge Functions 扩展服务端逻辑，用 Storage 管理对象文件。理解 Supabase 的最佳方式，不是把它当作一个黑箱 PaaS，而是把它看作一套围绕 PostgreSQL 精心编排的开源工具链。

核心认知：Supabase 的强大在于"PostgreSQL 原生特性"——RLS、触发器、扩展——这些在 Supabase 上依然可用，甚至因为你不再需要写服务端 CRUD 而更显威力。

一、Supabase 架构概览

1.1 核心组件

┌───────────────────────────────────────────┐
│              Supabase Platform              │
│  ┌─────────┐  ┌──────────┐  ┌───────────┐ │
│  │ PostgREST│  │ Realtime │  │  Storage  │ │
│  │ (REST)   │  │ (WAL)    │  │ (S3 API)  │ │
│  └────┬────┘  └────┬─────┘  └─────┬─────┘ │
│       │            │              │        │
│       └────────────┼──────────────┘        │
│                    ▼                       │
│           ┌──────────────┐                 │
│           │ PostgreSQL   │                 │
│           │ + Auth Schema│                 │
│           │ + RLS        │                 │
│           └──────────────┘                 │
│                    ▲                       │
│       ┌────────────┼──────────┐           │
│       ▼            ▼          ▼           │
│  ┌────────┐  ┌──────────┐  ┌────────┐    │
│  │ Auth   │  │ Edge     │  │ pgvector│    │
│  │ (JWT)  │  │ Functions│  │ & others│    │
│  └────────┘  └──────────┘  └────────┘    │
└───────────────────────────────────────────┘
组件作用对应开源项目
PostgREST自动 REST APIPostgREST/postgrest
RealtimeWAL 变更实时推送supabase/realtime
AuthJWT 认证与用户管理supabase/gotrue
Storage对象存储supabase/storage-api
Edge Function边缘计算Deno Deploy

二、Row Level Security 与认证

2.1 RLS 的工作原理

Supabase 的 PostgREST 每个请求都携带 JWT，Supabase Auth 服务签发含 sub(uid) 的令牌，PostgreSQL 的 RLS 策略利用 auth.uid() 函数解析该令牌，决定行可见性。

-- 启用 RLS
ALTER TABLE todos ENABLE ROW LEVEL SECURITY;

-- 用户只能看到自己的数据
CREATE POLICY "Users can only see their own todos"
ON todos FOR SELECT
USING (auth.uid() = user_id);

-- 插入时自动绑定当前用户
CREATE POLICY "Users can create their own todos"
ON todos FOR INSERT
WITH CHECK (auth.uid() = user_id);

-- 更新/删除同理
CREATE POLICY "Users can update their own todos"
ON todos FOR UPDATE
USING (auth.uid() = user_id);
RLS 策略核心函数来源
auth.uid()从 JWT sub 字段解析当前用户 UUID
auth.role()返回角色名（authenticated、anon 等）
auth.jwt()返回完整 JWT payload（JSONB）

2.2 匿名访问策略

-- 允许匿名用户只读访问公开文章
CREATE POLICY "Public articles are readable"
ON articles FOR SELECT
USING (is_published = true);

-- 匿名访问策略语法
CREATE POLICY "Enable anon read"
ON articles FOR SELECT
TO anon
USING (is_published = true);

三、Edge Functions（Deno）

3.1 Edge Function 基础

Supabase Edge Functions 运行在 Deno Deploy 上，可以直接使用标准 Deno API。

// supabase/functions/hello/index.ts
import { serve } from "https://deno.land/std@0.177.0/http/server.ts";

serve(async (req) => {
  const { name } = await req.json();
  const data = { message: `Hello ${name || 'World'}!` };

  return new Response(JSON.stringify(data), {
    headers: { "Content-Type": "application/json" },
  });
});

部署：

supabase functions deploy hello

3.2 访问 PostgreSQL 数据库

// supabase/functions/process-order/index.ts
import { serve } from "https://deno.land/std@0.177.0/http/server.ts";
import { createClient } from "https://esm.sh/@supabase/supabase-js@2";

serve(async (req) => {
  const supabase = createClient(
    Deno.env.get("SUPABASE_URL")!,
    Deno.env.get("SUPABASE_SERVICE_ROLE_KEY")!  // 使用 service role 绕过 RLS
  );

  const { order_id } = await req.json();

  // 直接调用数据库函数或执行 RPC
  const { data, error } = await supabase
    .rpc('process_order', { order_id });

  if (error) {
    return new Response(JSON.stringify({ error: error.message }), {
      status: 500,
      headers: { "Content-Type": "application/json" },
    });
  }

  return new Response(JSON.stringify(data), {
    headers: { "Content-Type": "application/json" },
  });
});

注意：Edge Function 中使用 SUPABASE_SERVICE_ROLE_KEY 可绕过所有 RLS，适用于后台任务、定时作业。Never 把它泄露到客户端。

3.3 调用 Edge Function

// 前端调用
const { data, error } = await supabase.functions.invoke('process-order', {
  body: { order_id: 12345 }
});

四、Database Webhooks

4.1 Webhook 触发机制

Supabase Database Webhooks 实质是 PostgreSQL 触发器 + Edge Function 调用，当数据变更时自动触发外部 HTTP 请求。

-- 创建 Webhook 触发器：每次订单状态变更发送通知
CREATE OR REPLACE FUNCTION notify_order_change()
RETURNS trigger AS $$
BEGIN
    PERFORM net.http_post(
        url := 'https://your-project.supabase.co/functions/v1/order-webhook',
        headers := '{"Content-Type": "application/json", "Authorization": "Bearer ..."}'::jsonb,
        body := jsonb_build_object(
            'event', TG_OP,
            'order_id', NEW.id,
            'new_status', NEW.status
        )
    );
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER order_webhook_trigger
    AFTER UPDATE ON orders
    FOR EACH ROW
    WHEN (OLD.status IS DISTINCT FROM NEW.status)
    EXECUTE FUNCTION notify_order_change();

4.2 使用 pg_net 扩展

Supabase 已预装 pg_net 扩展，用于 PostgreSQL 内发起 HTTP 请求：

-- 检查扩展是否启用
SELECT * FROM pg_extension WHERE extname = 'pg_net';

-- 异步 HTTP 请求（非阻塞）
SELECT net.http_get('https://api.example.com/health');

五、向量检索与 pgvector

5.1 启用 pgvector

-- Supabase 已预装 pgvector，直接启用
CREATE EXTENSION IF NOT EXISTS vector;

-- 创建向量表
CREATE TABLE embeddings (
    id          uuid DEFAULT gen_random_uuid() PRIMARY KEY,
    content     text,
    embedding   vector(1536),  -- OpenAI text-embedding-ada-002 维度
    metadata    jsonb DEFAULT '{}'
);

5.2 向量操作

-- 插入向量（用小数数组表示）
INSERT INTO embeddings (content, embedding)
VALUES (
    'PostgreSQL 向量扩展',
    '[0.001, -0.023, 0.045, ...]'::vector  -- 1536 维
);

-- 余弦相似度检索（Top 5）
SELECT content,
       embedding <=> '[0.002, -0.015, ...]'::vector AS distance
FROM embeddings
ORDER BY embedding <=> '[0.002, -0.015, ...]'::vector
LIMIT 5;
运算符含义
<->L2 欧氏距离
<=>余弦距离（= 1 - cosine_similarity）
<#>内积

5.3 向量索引（IVFFlat / HNSW）

-- IVFFlat 索引
CREATE INDEX ON embeddings
USING ivfflat (embedding vector_cosine_ops)
WITH (lists = 100);  -- 列表数 ≈ 总行数 / 1000

-- HNSW 索引（Supabase 已支持 pgvector ≥ 0.5.0）
CREATE INDEX ON embeddings
USING hnsw (embedding vector_cosine_ops);

IVFFlat 需要预先建索引并传入 probes=10 参数调优；HNSW 不需要 probes，构建和查询延迟更优。

5.4 结合 Edge Function 做语义搜索

// supabase/functions/semantic-search/index.ts
import { serve } from "https://deno.land/std@0.177.0/http/server.ts";
import { createClient } from "https://esm.sh/@supabase/supabase-js@2";

serve(async (req) => {
  const { query, limit = 5 } = await req.json();
  const supabase = createClient(
    Deno.env.get("SUPABASE_URL")!,
    Deno.env.get("SUPABASE_SERVICE_ROLE_KEY")!
  );

  // 1. 调用 OpenAI 获取 query 的 embedding
  const embedRes = await fetch("https://api.openai.com/v1/embeddings", {
    method: "POST",
    headers: {
      "Authorization": `Bearer ${Deno.env.get("OPENAI_API_KEY")}`,
      "Content-Type": "application/json",
    },
    body: JSON.stringify({ input: query, model: "text-embedding-ada-002" }),
  });
  const { data: [{ embedding }] } = await embedRes.json();

  // 2. 在 Supabase 中做向量检索
  const { data: results } = await supabase
    .rpc('match_embeddings', {
      query_embedding: embedding,
      match_threshold: 0.78,
      match_count: limit,
    });

  return new Response(JSON.stringify(results), {
    headers: { "Content-Type": "application/json" },
  });
});

-- PostgreSQL 端：封装向量检索函数
CREATE OR REPLACE FUNCTION match_embeddings(
    query_embedding vector(1536),
    match_threshold float,
    match_count int
)
RETURNS TABLE(
    id uuid,
    content text,
    metadata jsonb,
    similarity float
) AS $$
BEGIN
    RETURN QUERY
    SELECT
        e.id,
        e.content,
        e.metadata,
        1 - (e.embedding <=> query_embedding) AS similarity
    FROM embeddings e
    WHERE 1 - (e.embedding <=> query_embedding) > match_threshold
    ORDER BY e.embedding <=> query_embedding
    LIMIT match_count;
END;
$$ LANGUAGE plpgsql;

六、Realtime 订阅

6.1 Realtime 工作原理

Supabase Realtime 监听 PostgreSQL 的 WAL（Write-Ahead Log），通过逻辑解码捕获变更事件，再通过 WebSocket 推送给客户端：

INSERT/UPDATE/DELETE → WAL → logical replication slot
                                 ↓
                          Realtime server
                                 ↓
                          WebSocket broadcast
                                 ↓
                          客户端 listener callback

6.2 客户端订阅

import { createClient } from '@supabase/supabase-js';

const supabase = createClient(SUPABASE_URL, SUPABASE_ANON_KEY);

// 订阅 orders 表的实时变更
const channel = supabase
  .channel('orders-channel')
  .on(
    'postgres_changes',
    { event: '*', schema: 'public', table: 'orders' },
    (payload) => {
      console.log('Change received!', payload);
      // payload = { new: {...}, old: {...}, eventType: 'INSERT'/'UPDATE'/'DELETE' }
    }
  )
  .subscribe();

// 按 RLS 过滤：用户只能收到自己订单的变更
const myOrders = supabase
  .channel('my-orders')
  .on(
    'postgres_changes',
    {
      event: '*',
      schema: 'public',
      table: 'orders',
      filter: 'user_id=eq.' + currentUserId,
    },
    (payload) => console.log('My order changed:', payload)
  )
  .subscribe();

6.3 广播（Broadcast）与 Presence

// Realtime 的广播功能：客户端间直接消息
const channel = supabase.channel('room-1');

channel
  .on('broadcast', { event: 'new-message' }, (payload) => {
    console.log('Received:', payload);
  })
  .subscribe();

// 发送广播
channel.send({
  type: 'broadcast',
  event: 'new-message',
  payload: { text: 'Hello!' },
});

// Presence：在线状态
channel
  .on('presence', { event: 'sync' }, () => {
    const users = channel.presenceState();
    console.log('Online users:', Object.keys(users).length);
  })
  .subscribe(async (status) => {
    if (status === 'SUBSCRIBED') {
      await channel.track({ online_at: new Date().toISOString() });
    }
  });

七、REST API 自动生成

7.1 PostgREST 自动映射

-- Supabase 自动将以下表暴露为 REST API
CREATE TABLE products (
    id      serial PRIMARY KEY,
    name    text NOT NULL,
    price   numeric(10,2),
    stock   int DEFAULT 0
);

HTTP 客户端可直接访问（带有 apikey 和 Authorization 头）：

# 读取
GET /rest/v1/products?select=*&stock=gt.0

# 插入
POST /rest/v1/products
Body: { "name": "Headphones", "price": 299, "stock": 10 }

# 更新
PATCH /rest/v1/products?id=eq.1
Body: { "price": 259 }

# 删除
DELETE /rest/v1/products?id=eq.1

7.2 客户端 SDK 封装

// Supabase JavaScript 客户端
const { data, error } = await supabase
  .from('products')
  .select('id, name, price')
  .gt('stock', 0)
  .order('price', { ascending: true })
  .limit(10);

// 嵌套查询
const { data } = await supabase
  .from('orders')
  .select(`
    id,
    created_at,
    user:users(email),
    items:order_items(
      product:products(name, price),
      quantity
    )
  `);

常见问题（FAQ）

PostgREST 的 RLS 生效顺序是什么？

请求到达 PostgREST → 解析 JWT → PostgreSQL SET LOCAL role → RLS 策略按角色评估。注意：API 请求的 Authorization: Bearer <token> 头中必须包含有效 JWT，否则退化为 anon 角色。

Edge Function 和 Database Function 有什么区别？

维度Edge FunctionPostgreSQL Function（RPC）
运行环境Deno / 边缘节点PostgreSQL 进程内
适用场景外部 API 调用、复杂业务逻辑纯数据处理、事务操作
延迟更高（网络往返）极低（数据库内执行）
安全性service_role_key 可绕过 RLS按调用者角色受 RLS 约束

pgvector IVFFlat 索引的 “probes” 是什么？

probes 决定查询时搜索多少个 IVF 列表。默认值 1 意味着最快但可能漏结果；值越大精度越高但越慢。常用 probes = lists / 10。对于 100 个 list，设 probes = 10 是较好的平衡。

Realtime 在大型表上性能如何？

Realtime 的性能取决于 WAL 解码速度而非表大小。但过滤大量变更时，每个客户端订阅都会消耗服务端资源。建议通过 RLS 或 filter 条件减少推送量，不要订阅无过滤条件的超大表。

相关阅读

延伸阅读

完整示例（一键复制）

-- ========== PostgreSQL 端：Supabase 项目初始化 ==========

-- 1. 启用 pgvector
CREATE EXTENSION IF NOT EXISTS vector;

-- 2. 示例表
CREATE TABLE IF NOT EXISTS todos (
    id          uuid DEFAULT gen_random_uuid() PRIMARY KEY,
    user_id     uuid NOT NULL,
    title       text NOT NULL,
    completed   boolean DEFAULT false,
    created_at  timestamptz DEFAULT NOW()
);

-- 3. RLS 策略
ALTER TABLE todos ENABLE ROW LEVEL SECURITY;

CREATE POLICY "Users can view own todos"
ON todos FOR SELECT
USING (auth.uid() = user_id);

CREATE POLICY "Users can insert own todos"
ON todos FOR INSERT
WITH CHECK (auth.uid() = user_id);

CREATE POLICY "Users can update own todos"
ON todos FOR UPDATE
USING (auth.uid() = user_id);

CREATE POLICY "Users can delete own todos"
ON todos FOR DELETE
USING (auth.uid() = user_id);

-- 4. 向量检索表
CREATE TABLE IF NOT EXISTS embeddings (
    id          uuid DEFAULT gen_random_uuid() PRIMARY KEY,
    content     text,
    embedding   vector(1536),
    metadata    jsonb DEFAULT '{}'
);

-- 5. 向量检索函数
CREATE OR REPLACE FUNCTION match_embeddings(
    query_embedding vector(1536),
    match_threshold float,
    match_count int
)
RETURNS TABLE(id uuid, content text, metadata jsonb, similarity float) AS $$
BEGIN
    RETURN QUERY
    SELECT
        e.id,
        e.content,
        e.metadata,
        1 - (e.embedding <=> query_embedding) AS similarity
    FROM embeddings e
    WHERE 1 - (e.embedding <=> query_embedding) > match_threshold
    ORDER BY e.embedding <=> query_embedding
    LIMIT match_count;
END;
$$ LANGUAGE plpgsql;

-- 6. Webhook 触发器示例
CREATE OR REPLACE FUNCTION notify_order_change()
RETURNS trigger AS $$
BEGIN
    PERFORM pg_notify('order_changes', json_build_object(
        'event', TG_OP,
        'order_id', NEW.id,
        'new_status', NEW.status
    )::text);
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

-- 7. 向量索引
CREATE INDEX IF NOT EXISTS idx_embeddings_hnsw
ON embeddings
USING hnsw (embedding vector_cosine_ops);

// ========== Deno Edge Function：语义搜索 ==========
// supabase/functions/semantic-search/index.ts

import { serve } from "https://deno.land/std@0.177.0/http/server.ts";
import { createClient } from "https://esm.sh/@supabase/supabase-js@2";

serve(async (req) => {
  const { query, limit = 5 } = await req.json();

  const supabase = createClient(
    Deno.env.get("SUPABASE_URL")!,
    Deno.env.get("SUPABASE_SERVICE_ROLE_KEY")!
  );

  // 获取 embedding（示例，实际应调用 OpenAI API）
  const embedRes = await fetch("https://api.openai.com/v1/embeddings", {
    method: "POST",
    headers: {
      "Authorization": `Bearer ${Deno.env.get("OPENAI_API_KEY")}`,
      "Content-Type": "application/json",
    },
    body: JSON.stringify({ input: query, model: "text-embedding-ada-002" }),
  });

  if (!embedRes.ok) {
    return new Response(JSON.stringify({ error: "Embedding failed" }), {
      status: 500,
      headers: { "Content-Type": "application/json" },
    });
  }

  const { data: [{ embedding }] } = await embedRes.json();

  const { data: results, error } = await supabase
    .rpc('match_embeddings', {
      query_embedding: embedding,
      match_threshold: 0.78,
      match_count: limit,
    });

  if (error) {
    return new Response(JSON.stringify({ error: error.message }), {
      status: 500,
      headers: { "Content-Type": "application/json" },
    });
  }

  return new Response(JSON.stringify({ query, results }), {
    headers: { "Content-Type": "application/json" },
  });
});

// ========== JavaScript 客户端 ==========
import { createClient } from '@supabase/supabase-js';

const supabase = createClient(
  process.env.SUPABASE_URL,
  process.env.SUPABASE_ANON_KEY
);

// 带 RLS 的 CRUD
const { data } = await supabase
  .from('todos')
  .select('*')
  .eq('completed', false)
  .order('created_at', { ascending: false });

// 插入（自动受 RLS 约束）
await supabase.from('todos').insert({
  title: 'Learn Supabase',
  user_id: (await supabase.auth.getUser()).data.user?.id
});

// Realtime 订阅
const channel = supabase
  .channel('todos-channel')
  .on(
    'postgres_changes',
    { event: '*', schema: 'public', table: 'todos' },
    (payload) => console.log('Change:', payload)
  )
  .subscribe();

// 调用 Edge Function
const { data: searchResults } = await supabase.functions.invoke(
  'semantic-search',
  { body: { query: '分布式数据库原理', limit: 3 } }
);

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「database」更多文章

  1. PostgreSQL 事件触发器与审计日志实现
  2. Kubernetes 上 PostgreSQL 运维与 CloudNativePG 实战
  3. PostgreSQL 统计信息与查询计划器：ANALYZE、pg_statistic 与代价模型