Supabaseで RAGチャット を作るチュートリアルは多い。pgvector を入れて、ベクトル検索を書く。動く。
でも、それを 「会社A・会社B・会社Cが同じアプリを使うけど、お互いのデータは絶対に見えない」 という SaaS にした瞬間、考えることが急に増える。
- 会社Aの社員が、会社Bのドキュメントを検索できちゃったらどうする?
- そのチェック、どこに書く? Edge Function? それとも DB?
- 大量のPDFを取り込む時、Edge Function がタイムアウトしないか?
この記事では、Supabase 公式ドキュメントが推奨するやり方に沿って、マルチテナントRAGの設計を組み立てる。コードよりも「なぜこうするか」に重心を置く。
参考になる公式ドキュメント(困ったらここに戻る):
設計の前提として押さえる3つのこと
RLS は pgvector の検索にも効く
Supabase で一番大事な機能が RLS (Row Level Security) 。「このユーザーはこの行を見ていい」というルールを、DB に直接書く仕組み。
そして公式が明言しているとおり、RLS は普通のSELECT だけでなく、pgvector のベクトル検索にもそのまま効く。つまり:
-- このSELECT に「自分のテナントだけ」というRLSが書いてあれば
select * from document_sections
order by embedding <#> query_embedding
limit 5;
-- ↑ 結果は自動的に「自分のテナントのチャンクだけ」に絞られる
権限チェックを Edge Function で書く必要はない。DB が勝手にやってくれる。
Edge Function には時間制限がある
Supabase の Edge Function は、Deno で動くサーバーレス関数。便利だけど、こういう制限がある(公式 Limits より):
| 項目 | 値 | 意味 |
|---|---|---|
| CPU 時間 | 2秒 / リクエスト | 計算処理の時間。API待ちは含まない |
| アイドルタイムアウト | 150秒 | レスポンス返さないと504 |
| Wall clock (Free) | 150秒 | 関数全体の最大稼働時間 |
| Wall clock (Paid) | 400秒 | 同上 |
| メモリ | 256MB |
重要なのは「CPU 2秒」。OpenAI API を呼んで待ってる時間は CPU を使ってないからカウントされない。だから LLM チャットは余裕。でも、PDFを解析してチャンクに分けて…みたいな処理は2秒で死ぬ。
DB でやる仕事と Edge Function でやる仕事を分ける
公式の指針(超ざっくり):
| やること | どこに書くか |
|---|---|
| データの集計、検索、JOIN、トランザクション | Postgres関数 (Database Function) |
| 外部API(OpenAI, Stripe...)を呼ぶ | Edge Function |
| 重い処理を非同期でやる | Queue + Cron + Edge Function |
これを意識せずに「全部 Edge Function でやる」と、CPU 2秒に引っかかったり、コードが Edge Function に集中して読みにくくなったりする。
どんなアプリを作るか
こういうアプリを想定する。
[会社A] [会社B]
├ ユーザーa1 ├ ユーザーb1
├ ユーザーa2 └ ユーザーb2
└ アップロードしたPDF
(社内マニュアル) └ アップロードしたPDF
(営業資料)
「うちのマニュアルでXXXってどう書いてあった?」
→ 会社A のドキュメントだけから検索して LLM が答える
→ 会社B のものは絶対に混ざらない
このアプリで最低限必要なテーブルはこの4つ。
tenants 会社マスタ
memberships 誰がどの会社に所属してるか
documents アップロードされたファイル
document_sections チャンクとベクトル
テーブルを作る
-- 会社マスタ
create table tenants (
id uuid primary key default gen_random_uuid(),
name text not null
);
-- 誰がどの会社のメンバーか
create table memberships (
user_id uuid references auth.users(id) on delete cascade,
tenant_id uuid references tenants(id) on delete cascade,
role text not null check (role in ('owner', 'member')),
primary key (user_id, tenant_id)
);
-- アップロードされたファイル単位
create table documents (
id uuid primary key default gen_random_uuid(),
tenant_id uuid not null references tenants(id) on delete cascade,
title text not null,
status text not null default 'pending' -- pending / ready / failed
);
-- チャンク (実際に検索される単位)
create table document_sections (
id bigint primary key generated always as identity,
document_id uuid not null references documents(id) on delete cascade,
content text not null,
embedding extensions.vector(1536)
);
-- RLS が見にいく列にはインデックスを張る (これ大事)
create index on memberships (user_id);
create index on documents (tenant_id);
create index on document_sections (document_id);
ポイント:memberships.user_id と documents.tenant_id には絶対インデックスを張る。あとで RLS がここを毎回見にくるので、ここが遅いと全部遅くなる。
RLS で「会社の壁」を作る(ここが核心)
「自分はどのテナントに所属してるか」を返す関数を用意する
毎回 memberships を JOIN するのは遅いし、ポリシーが読みにくくなる。なので、「現在のユーザーが所属するテナントID一覧」を返す関数を一つ作って、ポリシーから呼ぶ。
-- private スキーマを作る (公開しないため)
create schema if not exists private;
-- 現在のユーザーが所属するテナントIDの配列を返す
create or replace function private.user_tenant_ids()
returns uuid[]
language sql
stable
security definer -- RLS を貫通して memberships を読む
set search_path = '' -- 必須!(後述)
as $$
select array(
select tenant_id
from public.memberships -- 完全修飾名で書く
where user_id = auth.uid()
);
$$;
ここで気をつけることは3つ。これらは公式が "must" と言って強調しているやつ:
-
security definerを使うならset search_path = ''を必ず書く
→ 攻撃者が偽のスキーマを差し込んで関数の中身を乗っ取る攻撃を防ぐ -
関数本体ではテーブル名を
public.membershipsのように完全修飾で書く
→search_path = ''にしたので、これをやらないとテーブルが見つからない -
security definer関数はprivateスキーマに置く(=APIから直接呼ばれないようにする)
→ Supabase の API 設定で公開してあるスキーマには置かない
テーブルごとにポリシーを書く
-- 全テーブルで RLS をオン
alter table tenants enable row level security;
alter table memberships enable row level security;
alter table documents enable row level security;
alter table document_sections enable row level security;
-- memberships: 自分のメンバーシップだけ
create policy "own memberships"
on memberships for select
to authenticated
using ( user_id = (select auth.uid()) );
-- documents: 自分の所属テナントのドキュメント
create policy "tenant documents"
on documents for select
to authenticated
using ( tenant_id = any ((select private.user_tenant_ids())) );
-- document_sections: ドキュメント経由で同じく絞る
create policy "tenant document sections"
on document_sections for select
to authenticated
using (
document_id in (
select id from documents
-- ↑ documents の RLS が効くので、ここで自動的に絞られる
)
);
ここでの作法:
| やってること | なぜ |
|---|---|
to authenticated を必ず書く |
未ログインユーザーには最初から評価しない(速い、安全) |
(select auth.uid()) でラップ |
行ごとに評価せず、1回だけ評価される(公式ベンチマークで 94% 速くなる) |
(select private.user_tenant_ids()) でラップ |
同上。配列を1回だけ取得 |
using ( ... in (select ...) ) |
JOIN を避けてサブクエリにする(公式推奨) |
裏で何が起きているか
ユーザー a1(会社A所属)が、何も考えずにこう書いたとする:
select content from document_sections
order by embedding <#> '[0.1, 0.2, ...]'::vector
limit 5;
RLS が裏で勝手に追加するので、実際にはこうなる:
select content from document_sections
where document_id in (
select id from documents
where tenant_id = any (private.user_tenant_ids())
-- ↑ a1 さんは会社A のテナントIDだけ取れる
)
order by embedding <#> '[0.1, ...]'::vector
limit 5;
つまり、Edge Function 側で WHERE tenant_id = ? を書き忘れても、会社B のドキュメントは絶対に出てこない。これが「壊れない」マルチテナントの本体。
チャット応答の Edge Function を書く
ここからは TypeScript 。Edge Function でやることは2つだけ。
- ユーザーの質問を OpenAI で埋め込みベクトルに変換する
- その結果で DB を検索して、LLM に渡してストリーム応答を返す
// supabase/functions/chat/index.ts
import { createClient } from 'npm:@supabase/supabase-js@2'
Deno.serve(async (req) => {
// ⭐ ポイント: service_role ではなく anon key + ユーザーのJWT を使う
// こうすると auth.uid() が効いて、自動的に RLS が効く
const supabase = createClient(
Deno.env.get('SUPABASE_URL')!,
Deno.env.get('SUPABASE_ANON_KEY')!,
{
global: {
headers: { Authorization: req.headers.get('Authorization')! },
},
}
)
const { question } = await req.json()
// 1. 質問を埋め込みベクトルに (外部API)
const embedding = await generateEmbedding(question)
// 2. ベクトル検索 (RLS が自動で効くので、tenant_id を渡す必要すらない)
const { data: chunks, error } = await supabase.rpc('match_document_sections', {
query_embedding: embedding,
match_count: 5,
})
if (error) {
return new Response(error.message, { status: 403 })
}
// 3. LLM に渡してストリーミング応答
const stream = await streamChatCompletion(question, chunks)
return new Response(stream, {
headers: { 'Content-Type': 'text/event-stream' },
})
})
ベクトル検索用の DB関数 match_document_sections はこう:
create or replace function match_document_sections(
query_embedding extensions.vector(1536),
match_count int default 5
)
returns table (
id bigint,
content text,
similarity float
)
language sql
stable
-- ⭐ ここは security definer を「使わない」!
-- security invoker (デフォルト) のままで RLS を効かせる
as $$
select
id,
content,
1 - (embedding <#> query_embedding) as similarity
from document_sections
-- WHERE は書かない。RLS が自動で絞るから
order by embedding <#> query_embedding
limit match_count;
$$;
注意:この関数は security definer ではない(指定しないとデフォルトの invoker)。こっちを definer にすると RLS をすり抜けて全テナントから検索しちゃう。
整理:
| 関数 | security | 役割 |
|---|---|---|
private.user_tenant_ids() |
definer | RLSポリシーから呼ぶ。RLSをすり抜けて memberships を読む必要がある |
match_document_sections() |
invoker (デフォルト) | クライアントから呼ぶ。RLSを効かせたい |
「RLS から呼ぶ関数は definer、クライアントから呼ぶ関数は invoker」と覚えておくとよい。
ドキュメント取り込みは Edge Function でやらない
これが意外と落とし穴。
ユーザーが100ページのPDFをアップロードしたとする。これを取り込むには:
- テキスト抽出
- チャンクに分割(数百個)
- 各チャンクの埋め込み生成(OpenAIに数百回リクエスト)
- DBに保存
これを Edge Function で同期的にやろうとすると、ほぼ確実に Wall clock の 150〜400秒に引っかかる。CPU 2秒も超える可能性がある。
公式が用意してる解決策: Queue + Cron
実は Supabase 公式は Automatic embeddings という専用ガイドを用意していて、まさにこの問題を解決する方法を示している。仕組みはこう:
┌─────────────────────────┐
[1] チャンクをINSERT │ document_sections │
│ (embedding は NULL) │
└──────────┬──────────────┘
│ trigger
↓
┌─────────────────────────┐
[2] キューに積む │ pgmq queue │
│ 「このIDのembedding作って」│
└──────────┬──────────────┘
│
[3] 10秒ごとに ┌──────────┴──────────────┐
│ pg_cron │
│ キューから10件取り出す │
└──────────┬──────────────┘
│ pg_net で呼び出し
↓
[4] 短時間で ┌─────────────────────────┐
バッチ処理 │ Edge Function: embed │
│ 10件分の埋め込み生成 │
│ (これなら時間内に終わる) │
└──────────┬──────────────┘
│ UPDATE
↓
┌─────────────────────────┐
│ document_sections │
│ (embedding に値が入る) │
└─────────────────────────┘
このメリット:
- 1回の Edge Function 実行は短い(10件分だけ処理して終わる)
- 失敗しても自動でリトライされる(pgmq の visibility timeout で復活する)
- ユーザーには即レスポンス返せる(裏でじわじわ処理される)
具体的なコードは長いので公式ガイドを見てほしいけど、考え方としては:
- PDF アップロード → ドキュメントとチャンク(embedding=NULL)を INSERT して即レスポンス
- INSERT トリガが pgmq にジョブを積む
- pg_cron が 10秒ごとにジョブを取り出して Edge Function に渡す
- Edge Function は10件だけ処理して、UPDATE で embedding を埋める
これなら巨大な PDF でも「ちょっとずつ処理されていく」状態になる。
軽い処理なら EdgeRuntime.waitUntil() でもいい
「数チャンクだけ取り込みたい」みたいな軽いケースなら、Queue まで持ち出さずに Edge Function 内でバックグラウンドにする手もある:
Deno.serve(async (req) => {
// 即レスポンスを返す
EdgeRuntime.waitUntil(processInBackground(...))
return new Response('accepted', { status: 202 })
})
ただし、これも CPU 2秒・Wall clock 150〜400秒の制限の対象。処理時間が予測できないなら Queue + Cron のほうが安全。
全体像
「Postgres を太らせて、Edge Function を細くする」という意識が伝わるはず。
設計レビュー用チェックリスト
設計するときに見返すリスト。全部公式の推奨どおり。
RLS まわり
-
全テーブルで
enable row level securityしてある -
ポリシーには
to authenticatedを書いた -
auth.uid()は(select auth.uid())でラップした -
RLS で参照する列(
user_id,tenant_id等)にインデックス張った -
JOIN ではなく
INか配列で書いた
security definer まわり
-
definerは最適化や JOIN 迂回の時だけ使う(デフォルトはinvoker) -
使う時は必ず
set search_path = ''を書いた -
関数本体は
public.tableのように完全修飾名で書いた -
privateスキーマに置いた(API公開スキーマには置かない)
Edge Function まわり
-
DB アクセスは
service_roleではなくanon key + ユーザーJWTを使う - CPU 2秒で終わらない処理は Queue + Cron に逃がす
- DB でできる処理(検索・集計)は DB関数に書く
pgvector まわり
-
検索用の関数は
security invoker(デフォルト)で作る → RLS が効く -
本番投入前に
pg_prewarmでインデックスを暖めておく - HNSW を使う(Supabase 公式の推奨)
おわりに
Supabase は「Postgres をうまく使うためのプラットフォーム」。Firebase の代替だと思って表面だけ触ると、せっかくの強み(RLS、Postgres関数、pgmq、pg_cron)を活かせない。
マルチテナントRAG はこの強みを全部使う題材としていい例で、
- 権限は DB に書く(Edge Function は信用しない)
- 計算は DB に書く(Edge Function は外部APIだけ)
- 重い処理はキューに逃がす(Edge Function は短時間で終わる)
この3つを意識すれば、Supabase は本番運用に十分耐えるバックエンドになる。
困ったら公式ドキュメントに戻る。本記事もすべてそこに根拠がある。
参考
- RAG with Permissions | Supabase Docs
- Row Level Security | Supabase Docs
- Database Functions | Supabase Docs
- Automatic embeddings | Supabase Docs
- Edge Functions: Limits | Supabase Docs
- Edge Functions: Background Tasks | Supabase Docs
- Going to Production (AI) | Supabase Docs
- Supabase Queues | Supabase Docs