Supabase Edge Functions 怎么构建支持断线重连与会话持久化的 WebSocket 服务【免费下载链接】supabaseThe Postgres development platform. Supabase gives you a dedicated Postgres database to build your web, mobile, and AI applications.项目地址: https://gitcode.com/GitHub_Trending/supa/supabaseSupabase Edge Functions 可以托管 WebSocket 服务器但 Edge 环境里的 worker 会在达到运行时长限制后被回收长连接随时可能中断同时函数本身无状态客户端断线后服务端记忆会丢失。Supabase 官方文档给出了一个可照做的参考实现用 Postgres 持久化会话与事件客户端携带sessionId和lastEventId重连服务端回放id lastEventId的事件从而构建一个断线可重连、消息不丢、重试不重复的聊天流。本文按“建表 → 写函数 → 本地验证 → 客户端重连 → 部署”的顺序实现这条路径。前提条件来自官方快速上手文档 Getting Started with Edge Functions已安装并配置 Supabase CLI本地运行和测试 Edge Functions 需要 Docker 或兼容运行时函数使用 TypeScript Deno runtime 编写这是 Edge Functions 目前唯一支持的组合。第一步创建 Postgres 持久化表会话持久化和事件回放都落在 Postgres 上。官方参考实现Resumable WebSockets with Edge Functions定义了四张表建表 SQL 如下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 );各表的分工对应官方文档说明的机制ws_sessions每个用户会话一行记录会话标识与最后事件位置ws_events每条消息带自增id重连时按id lastEventId回放(session_id, id)上的索引支撑该查询ws_idempotency_keys以(session_id, idempotency_key)为主键客户端重发同一消息时防止重复插入ws_live_connectionsunlogged表只记录当前活跃连接的元数据不承担持久化职责。第二步创建 WebSocket Edge Function在项目根目录已通过supabase init生成supabase/config.toml的目录创建函数supabase functions new websocket-proxy为什么 JWT 要走查询参数而不是请求头浏览器端的 WebSocket 客户端无法发送自定义请求头所以平台默认的 Authorization 头校验拿不到 JWT。官方文档Handling WebSockets给出的做法是serve 和 deploy 时显式传--no-verify-jwt跳过默认头校验然后自己从 URL 查询参数或自定义子协议里取 JWT 并验证。注意withSupabase包装器只校验请求头无法用于认证 WebSocket 客户端需要手动验证。函数实现下面是官方参考实现中websocket-proxy的完整代码写入supabase/functions/websocket-proxy/index.tsimport { 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 Promisevoid((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 })代码中有四个直接决定“断线重连 会话持久化”是否成立的点Deno.upgradeWebSocket(req, { idleTimeout: 0 })完成 HTTP 到 WebSocket 的升级。升级返回 response 之后 HTTP 请求即视为完成若不处理worker 可能在 socket 还开着时提前被回收。EdgeRuntime.waitUntil(closed)用一个在socket.onclose中才 resolve 的 promise 挂住 worker官方文档称之为防止 EarlyDrop过早退役空闲 worker的标准手法。事件回放连接建立后立即发session_init然后在queueMicrotask中查询ws_events里id lastEventId的事件逐条发出且带replay: true标记。客户端因此能补上断线窗口内丢失的消息。主动重启PREEMPTIVE_RESTART_MS 340_000到达时函数先发server_restarting再socket.close(1012, Service restart)让客户端在 worker 被平台强制杀掉之前就能有序重连。为什么是这个值平台的 Maximum Durationwall clock 限制在 Limits 文档中定义为 Free 计划 150s、付费计划 400s340s 的主动重启位于付费计划上限之内Free 计划下 worker 会先于该定时器到达 150s 被平台回收——两种情况都由客户端重连逻辑兜底。幂等写入客户端消息先对ws_idempotency_keys做upsertignoreDuplicates: true冲突说明是重试改为取回已有事件而不是再次插入ws_events。数据库访问方面部署后的 Edge Functions 已预配置 SSL 连接 Supabase 数据库无需额外配置见 Integrating with Supabase Database。第三步本地测试必须先改 config.toml这里有一个默认行为会直接破坏 WebSocket 调试本地supabase functions serve的实例在请求完成后会自动终止导致 WebSocket 连接无法保持。官方文档要求在supabase/config.toml中改为按 worker 存活[edge_runtime] policy per_worker注意事项官方文档原样给出的限制per_worker模式下函数不会在代码修改后自动热重载改完代码需要手动重新运行supabase functions serve。启动本地环境并 serve 函数本地服务跳过 JWT 头校验JWT 由函数内部验证supabase start # 首次运行会下载 Docker 镜像并启动全部本地服务可能需要几分钟 supabase functions serve websocket-proxy --no-verify-jwt函数本地运行在http://localhost:54321/functions/v1/websocket-proxyURL 规则与快速上手文档中http://localhost:54321/functions/v1/函数名一致。之后从浏览器或任意 WebSocket 客户端带上用户 JWT 连接例如连接串形如ws://localhost:54321/functions/v1/websocket-proxy?tokenUSER_JWT其中USER_JWT是读者自己 Supabase 项目里某个已登录用户的 session JWT需要自行替换。本地验证要点连接成功且 JWT 有效时收到第一条消息{type:session_init,session_id:...}sessionId首次连接时由函数生成客户端应保存它发送{type:user_message,idempotency_key:...,content:...}应答中带回服务端分配的event_id断开后用同一sessionId加上已收到的最大event_id作为lastEventId重连应收到带replay: true的补发事件。第四步浏览器客户端的重连逻辑官方示例客户端把sessionId和lastEventId存在sessionStorage中断线后携带这两个值重连并按官方说明采用指数退避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) } } }片段中有两处需要读者替换token传当前登录用户的 JWT从自己的 Auth 会话获取YOUR_PROJECT替换为自己的项目标识project ref。收到任何带event_id的消息时都用Math.max更新本地游标并写回sessionStorage这是重连时不重放、不遗漏的关键。收到server_restarting或连接onclose后调用重连逻辑即可——官方文档说明客户端“reconnects with exponential backoff”退避的具体参数文档未给出由客户端自行实现。官方文档解释这个模式为什么成立worker 重启时客户端用同一会话重连回放补上重连窗口的投递缺口幂等键防止客户端重试造成重复插入EdgeRuntime.waitUntil()防止空闲外观的 WebSocket worker 被提前终止。第五步部署部署流程与 Deploy to Production 一致。由于函数在内部自行验证 JWT部署时也需关闭平台的 JWT 头校验最稳妥的做法是写进supabase/config.toml保证各环境配置一致文档示例的写法[functions.websocket-proxy] verify_jwt false然后登录、关联项目并部署supabase login supabase projects list supabase link --project-ref your-project-id # 替换为上一步列出的项目 ID supabase functions deploy websocket-proxysupabase link --project-ref需要本机已有 DockerCLI 会自动回退到 API 部署也可显式加--use-api。部署成功后函数分发到全球边缘节点运行地址为https://[YOUR_PROJECT_ID].supabase.co/functions/v1/websocket-proxy客户端代码里的连接 URL 换成该地址即可。部署后的验证方式与本地一致带 JWT 连接确认收到session_init断开并携带lastEventId重连确认收到replay: true的事件用同一个idempotency_key重复发送确认不会在ws_events中产生第二条记录。限制与后续工作运行资源上限Limits最大内存 256MBworker 存活时长wall clockFree 计划 150s、付费计划 400s单请求 CPU 时间 2s。长连接场景下超过 wall clock 上限的 socket 会被平台结束客户端必须能处理这种情况。查询参数里的 JWT 可能被某些日志系统记录官方文档在“使用查询参数传 JWT”的方案中专门提示了这一点不想暴露到 URL 时可以改用Sec-WebSocket-Protocol子协议传 JWTjwt-TOKEN形式验证方式见 Handling WebSockets 的 “Using custom protocol” 小节。官方文档在 “Next steps” 中列出的后续项为所有ws_*表添加行级安全RLS策略为陈旧会话添加心跳与清理策略为事件 payload 添加结构化类型和输入校验添加断线率与回放延迟的观测面板。参考实现本身使用 admin client 绕过 RLS 直接读写这些表接入生产前 RLS 配置是必须补上的一环。【免费下载链接】supabaseThe Postgres development platform. Supabase gives you a dedicated Postgres database to build your web, mobile, and AI applications.项目地址: https://gitcode.com/GitHub_Trending/supa/supabase创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Supabase Edge Functions 怎么构建支持断线重连与会话持久化的 WebSocket 服务？
Supabase Edge Functions 怎么构建支持断线重连与会话持久化的 WebSocket 服务【免费下载链接】supabaseThe Postgres development platform. Supabase gives you a dedicated Postgres database to build your web, mobile, and AI applications.项目地址: https://gitcode.com/GitHub_Trending/supa/supabaseSupabase Edge Functions 可以托管 WebSocket 服务器但 Edge 环境里的 worker 会在达到运行时长限制后被回收长连接随时可能中断同时函数本身无状态客户端断线后服务端记忆会丢失。Supabase 官方文档给出了一个可照做的参考实现用 Postgres 持久化会话与事件客户端携带sessionId和lastEventId重连服务端回放id lastEventId的事件从而构建一个断线可重连、消息不丢、重试不重复的聊天流。本文按“建表 → 写函数 → 本地验证 → 客户端重连 → 部署”的顺序实现这条路径。前提条件来自官方快速上手文档 Getting Started with Edge Functions已安装并配置 Supabase CLI本地运行和测试 Edge Functions 需要 Docker 或兼容运行时函数使用 TypeScript Deno runtime 编写这是 Edge Functions 目前唯一支持的组合。第一步创建 Postgres 持久化表会话持久化和事件回放都落在 Postgres 上。官方参考实现Resumable WebSockets with Edge Functions定义了四张表建表 SQL 如下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 );各表的分工对应官方文档说明的机制ws_sessions每个用户会话一行记录会话标识与最后事件位置ws_events每条消息带自增id重连时按id lastEventId回放(session_id, id)上的索引支撑该查询ws_idempotency_keys以(session_id, idempotency_key)为主键客户端重发同一消息时防止重复插入ws_live_connectionsunlogged表只记录当前活跃连接的元数据不承担持久化职责。第二步创建 WebSocket Edge Function在项目根目录已通过supabase init生成supabase/config.toml的目录创建函数supabase functions new websocket-proxy为什么 JWT 要走查询参数而不是请求头浏览器端的 WebSocket 客户端无法发送自定义请求头所以平台默认的 Authorization 头校验拿不到 JWT。官方文档Handling WebSockets给出的做法是serve 和 deploy 时显式传--no-verify-jwt跳过默认头校验然后自己从 URL 查询参数或自定义子协议里取 JWT 并验证。注意withSupabase包装器只校验请求头无法用于认证 WebSocket 客户端需要手动验证。函数实现下面是官方参考实现中websocket-proxy的完整代码写入supabase/functions/websocket-proxy/index.tsimport { 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 Promisevoid((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 })代码中有四个直接决定“断线重连 会话持久化”是否成立的点Deno.upgradeWebSocket(req, { idleTimeout: 0 })完成 HTTP 到 WebSocket 的升级。升级返回 response 之后 HTTP 请求即视为完成若不处理worker 可能在 socket 还开着时提前被回收。EdgeRuntime.waitUntil(closed)用一个在socket.onclose中才 resolve 的 promise 挂住 worker官方文档称之为防止 EarlyDrop过早退役空闲 worker的标准手法。事件回放连接建立后立即发session_init然后在queueMicrotask中查询ws_events里id lastEventId的事件逐条发出且带replay: true标记。客户端因此能补上断线窗口内丢失的消息。主动重启PREEMPTIVE_RESTART_MS 340_000到达时函数先发server_restarting再socket.close(1012, Service restart)让客户端在 worker 被平台强制杀掉之前就能有序重连。为什么是这个值平台的 Maximum Durationwall clock 限制在 Limits 文档中定义为 Free 计划 150s、付费计划 400s340s 的主动重启位于付费计划上限之内Free 计划下 worker 会先于该定时器到达 150s 被平台回收——两种情况都由客户端重连逻辑兜底。幂等写入客户端消息先对ws_idempotency_keys做upsertignoreDuplicates: true冲突说明是重试改为取回已有事件而不是再次插入ws_events。数据库访问方面部署后的 Edge Functions 已预配置 SSL 连接 Supabase 数据库无需额外配置见 Integrating with Supabase Database。第三步本地测试必须先改 config.toml这里有一个默认行为会直接破坏 WebSocket 调试本地supabase functions serve的实例在请求完成后会自动终止导致 WebSocket 连接无法保持。官方文档要求在supabase/config.toml中改为按 worker 存活[edge_runtime] policy per_worker注意事项官方文档原样给出的限制per_worker模式下函数不会在代码修改后自动热重载改完代码需要手动重新运行supabase functions serve。启动本地环境并 serve 函数本地服务跳过 JWT 头校验JWT 由函数内部验证supabase start # 首次运行会下载 Docker 镜像并启动全部本地服务可能需要几分钟 supabase functions serve websocket-proxy --no-verify-jwt函数本地运行在http://localhost:54321/functions/v1/websocket-proxyURL 规则与快速上手文档中http://localhost:54321/functions/v1/函数名一致。之后从浏览器或任意 WebSocket 客户端带上用户 JWT 连接例如连接串形如ws://localhost:54321/functions/v1/websocket-proxy?tokenUSER_JWT其中USER_JWT是读者自己 Supabase 项目里某个已登录用户的 session JWT需要自行替换。本地验证要点连接成功且 JWT 有效时收到第一条消息{type:session_init,session_id:...}sessionId首次连接时由函数生成客户端应保存它发送{type:user_message,idempotency_key:...,content:...}应答中带回服务端分配的event_id断开后用同一sessionId加上已收到的最大event_id作为lastEventId重连应收到带replay: true的补发事件。第四步浏览器客户端的重连逻辑官方示例客户端把sessionId和lastEventId存在sessionStorage中断线后携带这两个值重连并按官方说明采用指数退避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) } } }片段中有两处需要读者替换token传当前登录用户的 JWT从自己的 Auth 会话获取YOUR_PROJECT替换为自己的项目标识project ref。收到任何带event_id的消息时都用Math.max更新本地游标并写回sessionStorage这是重连时不重放、不遗漏的关键。收到server_restarting或连接onclose后调用重连逻辑即可——官方文档说明客户端“reconnects with exponential backoff”退避的具体参数文档未给出由客户端自行实现。官方文档解释这个模式为什么成立worker 重启时客户端用同一会话重连回放补上重连窗口的投递缺口幂等键防止客户端重试造成重复插入EdgeRuntime.waitUntil()防止空闲外观的 WebSocket worker 被提前终止。第五步部署部署流程与 Deploy to Production 一致。由于函数在内部自行验证 JWT部署时也需关闭平台的 JWT 头校验最稳妥的做法是写进supabase/config.toml保证各环境配置一致文档示例的写法[functions.websocket-proxy] verify_jwt false然后登录、关联项目并部署supabase login supabase projects list supabase link --project-ref your-project-id # 替换为上一步列出的项目 ID supabase functions deploy websocket-proxysupabase link --project-ref需要本机已有 DockerCLI 会自动回退到 API 部署也可显式加--use-api。部署成功后函数分发到全球边缘节点运行地址为https://[YOUR_PROJECT_ID].supabase.co/functions/v1/websocket-proxy客户端代码里的连接 URL 换成该地址即可。部署后的验证方式与本地一致带 JWT 连接确认收到session_init断开并携带lastEventId重连确认收到replay: true的事件用同一个idempotency_key重复发送确认不会在ws_events中产生第二条记录。限制与后续工作运行资源上限Limits最大内存 256MBworker 存活时长wall clockFree 计划 150s、付费计划 400s单请求 CPU 时间 2s。长连接场景下超过 wall clock 上限的 socket 会被平台结束客户端必须能处理这种情况。查询参数里的 JWT 可能被某些日志系统记录官方文档在“使用查询参数传 JWT”的方案中专门提示了这一点不想暴露到 URL 时可以改用Sec-WebSocket-Protocol子协议传 JWTjwt-TOKEN形式验证方式见 Handling WebSockets 的 “Using custom protocol” 小节。官方文档在 “Next steps” 中列出的后续项为所有ws_*表添加行级安全RLS策略为陈旧会话添加心跳与清理策略为事件 payload 添加结构化类型和输入校验添加断线率与回放延迟的观测面板。参考实现本身使用 admin client 绕过 RLS 直接读写这些表接入生产前 RLS 配置是必须补上的一环。【免费下载链接】supabaseThe Postgres development platform. Supabase gives you a dedicated Postgres database to build your web, mobile, and AI applications.项目地址: https://gitcode.com/GitHub_Trending/supa/supabase创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考