Skip to content
Edge Functions

支持恢复的 WebSockets 与边缘函数

这个例子展示了如何在 Supabase Edge Functions 上构建一个可安全重新连接的聊天流，使用：

🌐 This example shows how to build a reconnect-safe chat stream on Supabase Edge Functions using:

  • WebSocket 升级 + JWT 认证
  • 基于 Postgres 的会话和事件持久化
  • 使用 lastEventId 重播事件
  • 带有 idempotency_key 的幂等用户消息
  • 在工作进程重启时客户端优雅地重新连接

参考实现：使用 Supabase Edge Functions 和 Postgres 构建可恢复的 WebSockets

🌐 Reference implementation: Building Resumable WebSockets with Supabase Edge Functions and Postgres

架构 #

🌐 Architecture

  1. 客户端使用用户 JWT 连接，还可以选择性地使用 sessionId 和 lastEventId。
  2. 这个函数验证身份，然后要么恢复会话，要么创建一个新会话。
  3. 每条消息都是写给 ws_events 的，并且 id 会递增。
  4. 重新连接时，服务器会重放 id > lastEventId 的事件。
  5. 客户端更新本地 lastEventId 后继续运行，不会丢失消息。

数据库架构 #

🌐 Database schema

create extension if not exists pgcrypto;
create table ws_sessions (
  id uuid primary key default gen_random_uuid(),
  user_id uuid not null,
  created_at timestamptz default now(),
  updated_at timestamptz default now(),
  last_event_id bigint default 0
);
create table ws_events (
  id bigint generated by default as identity primary key,
  session_id uuid not null references ws_sessions(id) on delete cascade,
  event_type text not null,
  payload jsonb not null,
  created_at timestamptz default now()
);
create index ws_events_session_id_id_idx on ws_events(session_id, id);
create table ws_idempotency_keys (
  session_id uuid not null references ws_sessions(id) on delete cascade,
  idempotency_key uuid not null,
  primary key(session_id, idempotency_key)
);
create unlogged table ws_live_connections (
  session_id uuid primary key,
  connected_at timestamptz default now(),
  last_seen_at timestamptz default now(),
  edge_region text
);

边缘函数（WebSocket 代理） #

🌐 Edge Function (WebSocket proxy)

在函数里使用 supabase functions serve --no-verify-jwt 并验证 JWT。

🌐 Use supabase functions serve --no-verify-jwt and validate JWT inside the function.

import { createAdminClient, createContextClient, verifyCredentials } from '@supabase/server/core'
const PREEMPTIVE_RESTART_MS = 340_000
function send(socket: WebSocket, payload: unknown) {
  if (socket.readyState === WebSocket.OPEN) {
    socket.send(JSON.stringify(payload))
  }
}
Deno.serve(async (req) => {
  const url = new URL(req.url)
  const token = url.searchParams.get('token')
  if (!token) return new Response('Missing token', { status: 401 })
  const { data: auth, error } = await verifyCredentials({ token, apikey: null }, { auth: 'user' })
  if (error || !auth?.userClaims?.id) {
    return new Response('Unauthorized', { status: 401 })
  }
  const admin = createAdminClient()
  const { socket, response } = Deno.upgradeWebSocket(req, { idleTimeout: 0 })
  // Prevent EarlyDrop by keeping a pending promise until socket close.
  let resolveClosed!: () => void
  const closed = new Promise<void>((resolve) => {
    resolveClosed = resolve
  })
  // @ts-ignore
  EdgeRuntime.waitUntil(closed)
  const requestedSessionId = url.searchParams.get('sessionId')
  const lastEventId = Number(url.searchParams.get('lastEventId') || 0)
  const sessionId = requestedSessionId ?? crypto.randomUUID()
  socket.onclose = () => {
    resolveClosed()
  }
  socket.onmessage = async (event) => {
    const msg = JSON.parse(event.data)
    if (msg.type === 'user_message') {
      const { error: idempotencyError } = await admin.from('ws_idempotency_keys').upsert(
        {
          session_id: sessionId,
          idempotency_key: msg.idempotency_key,
        },
        { onConflict: 'session_id,idempotency_key', ignoreDuplicates: true }
      )
      let userEvent
      if (idempotencyError) {
        // Conflict detected - this is a retry, fetch the existing event
        const { data: existingEvent } = await admin
          .from('ws_events')
          .select()
          .eq('session_id', sessionId)
          .eq('idempotency_key', msg.idempotency_key)
          .single()
        userEvent = existingEvent
      } else {
        // New idempotency key - insert the event
        const { data: newEvent } = await admin
          .from('ws_events')
          .insert({
            session_id: sessionId,
            event_type: 'user_message',
            payload: { content: msg.content },
          })
          .select()
          .single()
        userEvent = newEvent
      }
      send(socket, {
        type: 'user_message',
        payload: userEvent?.payload,
        event_id: userEvent?.id,
      })
    }
  }
  send(socket, { type: 'session_init', session_id: sessionId })
  queueMicrotask(async () => {
    const { data: replayEvents } = await admin
      .from('ws_events')
      .select('*')
      .eq('session_id', sessionId)
      .gt('id', lastEventId)
      .order('id')
    for (const event of replayEvents ?? []) {
      send(socket, {
        type: event.event_type,
        payload: event.payload,
        event_id: event.id,
        replay: true,
      })
    }
  })
  setTimeout(() => {
    send(socket, { type: 'server_restarting' })
    socket.close(1012, 'Service restart')
  }, PREEMPTIVE_RESTART_MS)
  return response
})

浏览器客户端 #

🌐 Browser client

客户端将 sessionId 和 lastEventId 存储在会话存储中，然后以指数退避方式重新连接。

🌐 The client stores sessionId and lastEventId in session storage, then reconnects with exponential backoff.

let sessionId = sessionStorage.getItem('ws_session_id')
let lastEventId = Number(sessionStorage.getItem('last_event_id') || 0)
function connect(token: string) {
  const url =
    `wss://YOUR_PROJECT.functions.supabase.co/websocket-proxy` +
    `?token=${encodeURIComponent(token)}` +
    `&lastEventId=${lastEventId}` +
    (sessionId ? `&sessionId=${sessionId}` : '')
  const ws = new WebSocket(url)
  ws.onmessage = (e) => {
    const msg = JSON.parse(e.data)
    if (msg.event_id) {
      lastEventId = Math.max(lastEventId, msg.event_id)
      sessionStorage.setItem('last_event_id', String(lastEventId))
    }
    if (msg.type === 'session_init') {
      sessionId = msg.session_id
      sessionStorage.setItem('ws_session_id', sessionId)
    }
  }
}

为什么这个模式有效 #

🌐 Why this pattern works

  • 如果工作者重启，客户端会用同一个会话重新连接。
  • Replay 会关闭由重连窗口引起的交付差距。
  • 幂等键可以防止客户端重试时重复插入。
  • EdgeRuntime.waitUntil() 可以防止看起来空闲的 WebSocket 工作线程意外提前终止。

下一步 #

🌐 Next steps

  • 为所有 ws_* 表添加行级安全策略。
  • 为过时的会话添加心跳和清理策略。
  • 添加结构化事件负载类型和输入验证。
  • 为断开率和重放延迟添加可观测性仪表板。