Electric TanStack Query在 Electric 同步示例中实现乐观状态本地写入【免费下载链接】electricThe agent platform built on sync.项目地址: https://gitcode.com/GitHub_Trending/el/electric这篇技术文章基于 Electric 仓库中的 TanStack 演示website/sync/demos/tanstack.md讲解如何构建一个Electric 负责读路径同步、TanStack Query 负责本地写入与乐观状态的应用读完你将掌握useShape订阅、matchStream变更匹配、乐观 mutation 与流数据合并的完整实现方式并能按仓库内文档一步步在本地跑通该示例。方案总览读走 Electric写走 TanStack Query这个示例的定位在 README 中描述得很明确它使用 ElectricSQL 做读路径同步read path sync同时使用 TanStack Query 做带乐观状态的本地写入local writes with optimistic state。从 examples/tanstack/package.json 的依赖列表可以看出两条技术主线读路径electric-sql/client与electric-sql/react均为workspace:*即使用 monorepo 内源码构建的版本以及electric-sql/react提供的getShapeStream、useShape等 API写路径tanstack/react-queryv5.52.2、tanstack/react-query-persist-client与tanstack/query-sync-storage-persister用于 mutation 管理与离线持久化服务端辅助pgPostgreSQL 驱动、uuid以及 Vite React 18 的前端脚手架。数据模型非常简单items表只有一个id字段见 examples/.shared/db/migrations/01-create_items_table.sql-- Create a simple items table. CREATE TABLE IF NOT EXISTS items ( id TEXT PRIMARY KEY NOT NULL );迁移文件还会用generate_series预插入 10 条随机 UUID 记录方便演示时立刻看到数据。在本地运行示例该示例是 Electric monorepo 的一部分设计为 pnpm workspace 内的子包运行。完整操作步骤继承自 examples/tanstack/README.md在 monorepo 根目录安装并构建所有 workspace 包pnpm install pnpm run -r build进入示例目录后用 Docker Compose 启动后端服务Postgres Electriccd examples/tanstack pnpm backend:up注意 README 中特别提示backend:up会停止并删除此前任何其他示例后端容器挂载的 volume以确保示例总是从干净的数据库和磁盘启动。查看 examples/tanstack/package.json 可以发现该脚本的实际组成backend:up: PROJECT_NAMEtanstack-example pnpm -C ../../ run example-backend:up pnpm db:migrate,即先以PROJECT_NAMEtanstack-example启动共享的后端容器再执行db:migrate应用上述 SQL 迁移db:migrate使用../../.env.dev中的连接信息迁移目录为../.shared/db/migrations。启动开发服务器Vite 前端 Node 后端代理并发运行pnpm dev对应脚本为dev: concurrently \vite\ \node src/server/app.js\说明前端和代理服务端是同时拉起的。用完之后停止后端pnpm backend:down服务端一个把 /items 代理到 Electric Shape 流的 Node 服务前端的形状同步请求并不直连 Electric而是经由 examples/tanstack/src/server/app.js 这个 Node 代理。它监听PORT默认 3001提供三类端点GET /items代理到 Electric 的/v1/shape。关键逻辑是遍历请求的 query 参数只放行ELECTRIC_PROTOCOL_QUERY_PARAMS来自electric-sql/client中声明的电协议参数然后强制设置tableitems。如果环境里配置了ELECTRIC_SOURCE_ID/ELECTRIC_SOURCE_SECRET会附加source_id和secret参数用于多租户隔离。响应头会被复制剔除content-encoding、content-length等易出问题的头最后用Readable.fromWeb(response.body)把 Web 流转成 Node 流并pipeline到客户端实现字节透传。客户端提前断开产生的ERR_STREAM_PREMATURE_CLOSE会被静默忽略。POST /items解析 body 中的id直接INSERT INTO items (id) VALUES ($1)写入 PostgreSQL——这就是本地写入真正落库的路径。DELETE /items执行DELETE FROM items清空全表。另外/与/health返回健康检查所有响应都带 CORS 头Access-Control-Allow-Origin: *等。这种读走流代理、写走普通 SQL的拆分正是该示例架构的核心Electric 保证多客户端对items表变更的实时扇出而写入本身可以是任意普通 API这里为了演示简单直接对 PG 执行 SQL。前端骨架QueryClient、持久化与离线 mutationexamples/tanstack/src/App.tsx 搭起了 TanStack Query 的上下文const queryClient new QueryClient({ defaultOptions: { queries: { gcTime: Infinity, }, }, }) const persister createSyncStoragePersister({ storage: window.localStorage, }) PersistQueryClientProvider client{queryClient} persistOptions{{ persister }} Example / /PersistQueryClientProvider两个要点gcTime: Infinity让缓存永不回收——对于以 shape 流为主要数据来源的示例查询/变更缓存可以长期保留使用createSyncStoragePersisterPersistQueryClientProvider把 QueryClient 状态包括 pending mutation同步持久化到localStorage。注释里写明这是参照 TanStack Query 官方持久化离线 mutationpersisting offline mutations的指南实现的即使页面刷新、网络中断尚未完成的 mutation 也能在恢复后继续被追踪这正是乐观写入体验的组成部分。核心实现useShape 订阅 乐观 Mutation主要 Electric 代码集中在 examples/tanstack/src/Example.tsx这也是 demo 页面中嵌入展示的完整源码。按功能拆解形状定义与流订阅const baseApiUrl import.meta.env.VITE_SERVER_URL ?? http://localhost:3001 const itemsUrl new URL(/items, baseApiUrl) const itemShape () ({ url: itemsUrl.href, })shape 就是一个指向代理端点的 URL 对象VITE_SERVER_URL可覆盖默认的本机 3001 端口对应部署环境中的 Dockerfile / SST 部署配置。组件内部用useShapeItem(itemShape())拿到当前形状的全部数据items。写入createItem 同时等待API 成功与流中确认变更createItemExample.tsx体现了本示例最重要的模式——一次写入要拿到两个确认async function createItem(newId: string) { const itemsStream getShapeStreamItem(itemShape()) // Match the insert const findUpdatePromise matchStream({ stream: itemsStream, operations: [insert], matchFn: ({ message }) message.value.id newId, }) // Insert item const fetchPromise fetch(itemsUrl, { method: POST, body: JSON.stringify({ id: newId }), }) return await Promise.all([findUpdatePromise, fetchPromise]) }流程是先对同一 shape 建一个getShapeStream流并订阅等待其中出现恰好是这条新 id 的 insert 变更消息同时发出POST请求。Promise.all保证 mutation 只有在写入落库且变更已通过 Electric 流回传之后才算完成。clearItemsExample.tsx同理只是匹配delete操作且matchFn: () true——第一条 delete 变更即视为命中清空全表必然有删除发生只需确认流已恢复流动。matchStream把流变成一次性等待这个等待原语实现在 examples/tanstack/src/match-stream.ts只有约 50 行值得完整理解export async function matchStreamT extends Rowunknown({ stream, operations, matchFn, timeout 10000, }: { stream: ShapeStreamT operations: Arrayinsert | update | delete matchFn: ({ operationType, message }) boolean timeout?: number }): PromiseChangeMessageT { return new PromiseChangeMessageT((resolve, reject) { const unsubscribe stream.subscribe((messages) { const message messages.filter(isChangeMessage).find( (message) operations.includes(message.headers.operation) matchFn({ operationType: message.headers.operation, message }) ) if (message) return finish(message) }) const timeoutId setTimeout(() { console.error(matchStream timed out after ${timeout}ms) reject(matchStream timed out after ${timeout}ms) }, timeout) function finish(message: ChangeMessageT) { clearTimeout(timeoutId) unsubscribe() return resolve(message) } }) }实现要点对ShapeStream调用subscribe回调收到一批消息先用isChangeMessage来自electric-sql/client过滤出真正的变更消息排除快照等控制消息再按headers.operation是否在operations白名单中、且matchFn返回 true 来筛选目标消息命中后立即finish清理定时器、调用unsubscribe退订、resolve 出该ChangeMessage保证一次性语义且不留泄漏的订阅默认 10 秒超时超时打印错误并 reject——这为写后确认提供了有界等待避免网络或 Electric 故障导致 mutation 永久挂起。从源码结构看matchStream把 Electric 流式协议中持续推送的模型适配成命令式代码里的Promise 等待是流式数据源与 TanStack Query 这种基于 Promise 的 mutation 模型之间的关键粘合层。乐观状态onMutate、useMutationState 与流数据合并组件部分Example.tsx展示了三个配合add mutationuseMutation定义add-itemmutationKey: [add-item]mutationFn就是上面的createItemonMutate返回乐观项{ id }作为 mutation 的contextconst { mutateAsync: addItemMut } useMutation({ scope: { id: items }, mutationKey: [add-item], mutationFn: (newId: string) createItem(newId), onMutate: (id) { const optimisticItem: Item { id } return optimisticItem }, })读取 pending 提交useMutationState过滤status: pending的 mutation取出state.context得到正在提交中的乐观项列表submissions。由于 App 层配置了 localStorage 持久化刷新页面后这些 pending 提交仍可恢复展示。合并流数据与乐观数据const itemsMap new Mapstring, Item() if (!isClearing) { items.concat(submissions).forEach((item) { itemsMap.set(item.id, { ...itemsMap.get(item.id), ...item }) }) } else { submissions.forEach((item) itemsMap.set(item.id, item)) }源码注释解释得很直白合并 shape 数据与 fetcher 的乐观数据时按id去重因为存在竞态——useShape从流中收到的更新可能略早于actionmutation完成同一 item 会短暂地同时出现在items和submissions中用 Map 按键覆盖即可消除重复。特殊处理是清空进行中isClearing时只渲染 pending 提交、暂时隐藏流数据这样点 Clear 后界面立刻清空而不是等到删除变更从流里回来。clear mutation 顺带清理 pending 的 addclear-items的onMutate里遍历queryClient.getMutationCache().findAll({ mutationKey: [add-item] })把所有还在路上的 add mutation 从缓存中remove掉——既然全表都要清空那些即将插入的行不再需要展示。按钮交互上Add 点击时生成uuidv4()并触发addItemMutClear 传入当前items.length给clearItemsMut用于判断是否需要等待第一条 delete 变更。渲染则是对合并后的itemsMap逐项输出code{item.id}/code。端到端验证示例配有 Playwright 端到端测试 examples/tanstack/e2e/e2e.test.tsx共享配置见 examples/.shared/e2e 与 examples/.shared/playwright.config.ts可通过 examples/tanstack/package.json 中的脚本运行pnpm test:browser该脚本会先playwright install再执行playwright test ./e2e/用于回归验证点击 Add 后新行出现、点击 Clear 后列表清空这类依赖 Electric 流回传的完整用户路径。小结这个模式可复用在哪结合 demo 文档与示例源码可以提炼出Electric 读路径 TanStack Query 乐观写的三条核心做法shape 作为单一事实源UI 的主数据来自useShape而不是轮询 REST 接口写后确认用流而不是轮询matchStream把等待特定变更出现变成一个可超时、可退订的 Promise与 API 调用并行、一起 settle乐观项与流数据按键合并用useMutationState拿 pending 上下文按主键去重合并进渲染列表并用 mutation 缓存清理处理级联状态如清空时移除未完成的插入。仓库中还有更进一步的方向website/docs/sync/integrations/tanstack.md 介绍了 Electric 与 TanStack 的合作项目 TanStack DB嵌入式客户端数据库仓库内也提供了对应示例 examples/tanstack-db-web-starter 与 examples/tanstack-db-expo-starter可作为本示例之后继续深入 TanStack 生态同步方案时的参照。【免费下载链接】electricThe agent platform built on sync.项目地址: https://gitcode.com/GitHub_Trending/el/electric创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Electric + TanStack Query：在 Electric 同步示例中实现乐观状态本地写入
Electric TanStack Query在 Electric 同步示例中实现乐观状态本地写入【免费下载链接】electricThe agent platform built on sync.项目地址: https://gitcode.com/GitHub_Trending/el/electric这篇技术文章基于 Electric 仓库中的 TanStack 演示website/sync/demos/tanstack.md讲解如何构建一个Electric 负责读路径同步、TanStack Query 负责本地写入与乐观状态的应用读完你将掌握useShape订阅、matchStream变更匹配、乐观 mutation 与流数据合并的完整实现方式并能按仓库内文档一步步在本地跑通该示例。方案总览读走 Electric写走 TanStack Query这个示例的定位在 README 中描述得很明确它使用 ElectricSQL 做读路径同步read path sync同时使用 TanStack Query 做带乐观状态的本地写入local writes with optimistic state。从 examples/tanstack/package.json 的依赖列表可以看出两条技术主线读路径electric-sql/client与electric-sql/react均为workspace:*即使用 monorepo 内源码构建的版本以及electric-sql/react提供的getShapeStream、useShape等 API写路径tanstack/react-queryv5.52.2、tanstack/react-query-persist-client与tanstack/query-sync-storage-persister用于 mutation 管理与离线持久化服务端辅助pgPostgreSQL 驱动、uuid以及 Vite React 18 的前端脚手架。数据模型非常简单items表只有一个id字段见 examples/.shared/db/migrations/01-create_items_table.sql-- Create a simple items table. CREATE TABLE IF NOT EXISTS items ( id TEXT PRIMARY KEY NOT NULL );迁移文件还会用generate_series预插入 10 条随机 UUID 记录方便演示时立刻看到数据。在本地运行示例该示例是 Electric monorepo 的一部分设计为 pnpm workspace 内的子包运行。完整操作步骤继承自 examples/tanstack/README.md在 monorepo 根目录安装并构建所有 workspace 包pnpm install pnpm run -r build进入示例目录后用 Docker Compose 启动后端服务Postgres Electriccd examples/tanstack pnpm backend:up注意 README 中特别提示backend:up会停止并删除此前任何其他示例后端容器挂载的 volume以确保示例总是从干净的数据库和磁盘启动。查看 examples/tanstack/package.json 可以发现该脚本的实际组成backend:up: PROJECT_NAMEtanstack-example pnpm -C ../../ run example-backend:up pnpm db:migrate,即先以PROJECT_NAMEtanstack-example启动共享的后端容器再执行db:migrate应用上述 SQL 迁移db:migrate使用../../.env.dev中的连接信息迁移目录为../.shared/db/migrations。启动开发服务器Vite 前端 Node 后端代理并发运行pnpm dev对应脚本为dev: concurrently \vite\ \node src/server/app.js\说明前端和代理服务端是同时拉起的。用完之后停止后端pnpm backend:down服务端一个把 /items 代理到 Electric Shape 流的 Node 服务前端的形状同步请求并不直连 Electric而是经由 examples/tanstack/src/server/app.js 这个 Node 代理。它监听PORT默认 3001提供三类端点GET /items代理到 Electric 的/v1/shape。关键逻辑是遍历请求的 query 参数只放行ELECTRIC_PROTOCOL_QUERY_PARAMS来自electric-sql/client中声明的电协议参数然后强制设置tableitems。如果环境里配置了ELECTRIC_SOURCE_ID/ELECTRIC_SOURCE_SECRET会附加source_id和secret参数用于多租户隔离。响应头会被复制剔除content-encoding、content-length等易出问题的头最后用Readable.fromWeb(response.body)把 Web 流转成 Node 流并pipeline到客户端实现字节透传。客户端提前断开产生的ERR_STREAM_PREMATURE_CLOSE会被静默忽略。POST /items解析 body 中的id直接INSERT INTO items (id) VALUES ($1)写入 PostgreSQL——这就是本地写入真正落库的路径。DELETE /items执行DELETE FROM items清空全表。另外/与/health返回健康检查所有响应都带 CORS 头Access-Control-Allow-Origin: *等。这种读走流代理、写走普通 SQL的拆分正是该示例架构的核心Electric 保证多客户端对items表变更的实时扇出而写入本身可以是任意普通 API这里为了演示简单直接对 PG 执行 SQL。前端骨架QueryClient、持久化与离线 mutationexamples/tanstack/src/App.tsx 搭起了 TanStack Query 的上下文const queryClient new QueryClient({ defaultOptions: { queries: { gcTime: Infinity, }, }, }) const persister createSyncStoragePersister({ storage: window.localStorage, }) PersistQueryClientProvider client{queryClient} persistOptions{{ persister }} Example / /PersistQueryClientProvider两个要点gcTime: Infinity让缓存永不回收——对于以 shape 流为主要数据来源的示例查询/变更缓存可以长期保留使用createSyncStoragePersisterPersistQueryClientProvider把 QueryClient 状态包括 pending mutation同步持久化到localStorage。注释里写明这是参照 TanStack Query 官方持久化离线 mutationpersisting offline mutations的指南实现的即使页面刷新、网络中断尚未完成的 mutation 也能在恢复后继续被追踪这正是乐观写入体验的组成部分。核心实现useShape 订阅 乐观 Mutation主要 Electric 代码集中在 examples/tanstack/src/Example.tsx这也是 demo 页面中嵌入展示的完整源码。按功能拆解形状定义与流订阅const baseApiUrl import.meta.env.VITE_SERVER_URL ?? http://localhost:3001 const itemsUrl new URL(/items, baseApiUrl) const itemShape () ({ url: itemsUrl.href, })shape 就是一个指向代理端点的 URL 对象VITE_SERVER_URL可覆盖默认的本机 3001 端口对应部署环境中的 Dockerfile / SST 部署配置。组件内部用useShapeItem(itemShape())拿到当前形状的全部数据items。写入createItem 同时等待API 成功与流中确认变更createItemExample.tsx体现了本示例最重要的模式——一次写入要拿到两个确认async function createItem(newId: string) { const itemsStream getShapeStreamItem(itemShape()) // Match the insert const findUpdatePromise matchStream({ stream: itemsStream, operations: [insert], matchFn: ({ message }) message.value.id newId, }) // Insert item const fetchPromise fetch(itemsUrl, { method: POST, body: JSON.stringify({ id: newId }), }) return await Promise.all([findUpdatePromise, fetchPromise]) }流程是先对同一 shape 建一个getShapeStream流并订阅等待其中出现恰好是这条新 id 的 insert 变更消息同时发出POST请求。Promise.all保证 mutation 只有在写入落库且变更已通过 Electric 流回传之后才算完成。clearItemsExample.tsx同理只是匹配delete操作且matchFn: () true——第一条 delete 变更即视为命中清空全表必然有删除发生只需确认流已恢复流动。matchStream把流变成一次性等待这个等待原语实现在 examples/tanstack/src/match-stream.ts只有约 50 行值得完整理解export async function matchStreamT extends Rowunknown({ stream, operations, matchFn, timeout 10000, }: { stream: ShapeStreamT operations: Arrayinsert | update | delete matchFn: ({ operationType, message }) boolean timeout?: number }): PromiseChangeMessageT { return new PromiseChangeMessageT((resolve, reject) { const unsubscribe stream.subscribe((messages) { const message messages.filter(isChangeMessage).find( (message) operations.includes(message.headers.operation) matchFn({ operationType: message.headers.operation, message }) ) if (message) return finish(message) }) const timeoutId setTimeout(() { console.error(matchStream timed out after ${timeout}ms) reject(matchStream timed out after ${timeout}ms) }, timeout) function finish(message: ChangeMessageT) { clearTimeout(timeoutId) unsubscribe() return resolve(message) } }) }实现要点对ShapeStream调用subscribe回调收到一批消息先用isChangeMessage来自electric-sql/client过滤出真正的变更消息排除快照等控制消息再按headers.operation是否在operations白名单中、且matchFn返回 true 来筛选目标消息命中后立即finish清理定时器、调用unsubscribe退订、resolve 出该ChangeMessage保证一次性语义且不留泄漏的订阅默认 10 秒超时超时打印错误并 reject——这为写后确认提供了有界等待避免网络或 Electric 故障导致 mutation 永久挂起。从源码结构看matchStream把 Electric 流式协议中持续推送的模型适配成命令式代码里的Promise 等待是流式数据源与 TanStack Query 这种基于 Promise 的 mutation 模型之间的关键粘合层。乐观状态onMutate、useMutationState 与流数据合并组件部分Example.tsx展示了三个配合add mutationuseMutation定义add-itemmutationKey: [add-item]mutationFn就是上面的createItemonMutate返回乐观项{ id }作为 mutation 的contextconst { mutateAsync: addItemMut } useMutation({ scope: { id: items }, mutationKey: [add-item], mutationFn: (newId: string) createItem(newId), onMutate: (id) { const optimisticItem: Item { id } return optimisticItem }, })读取 pending 提交useMutationState过滤status: pending的 mutation取出state.context得到正在提交中的乐观项列表submissions。由于 App 层配置了 localStorage 持久化刷新页面后这些 pending 提交仍可恢复展示。合并流数据与乐观数据const itemsMap new Mapstring, Item() if (!isClearing) { items.concat(submissions).forEach((item) { itemsMap.set(item.id, { ...itemsMap.get(item.id), ...item }) }) } else { submissions.forEach((item) itemsMap.set(item.id, item)) }源码注释解释得很直白合并 shape 数据与 fetcher 的乐观数据时按id去重因为存在竞态——useShape从流中收到的更新可能略早于actionmutation完成同一 item 会短暂地同时出现在items和submissions中用 Map 按键覆盖即可消除重复。特殊处理是清空进行中isClearing时只渲染 pending 提交、暂时隐藏流数据这样点 Clear 后界面立刻清空而不是等到删除变更从流里回来。clear mutation 顺带清理 pending 的 addclear-items的onMutate里遍历queryClient.getMutationCache().findAll({ mutationKey: [add-item] })把所有还在路上的 add mutation 从缓存中remove掉——既然全表都要清空那些即将插入的行不再需要展示。按钮交互上Add 点击时生成uuidv4()并触发addItemMutClear 传入当前items.length给clearItemsMut用于判断是否需要等待第一条 delete 变更。渲染则是对合并后的itemsMap逐项输出code{item.id}/code。端到端验证示例配有 Playwright 端到端测试 examples/tanstack/e2e/e2e.test.tsx共享配置见 examples/.shared/e2e 与 examples/.shared/playwright.config.ts可通过 examples/tanstack/package.json 中的脚本运行pnpm test:browser该脚本会先playwright install再执行playwright test ./e2e/用于回归验证点击 Add 后新行出现、点击 Clear 后列表清空这类依赖 Electric 流回传的完整用户路径。小结这个模式可复用在哪结合 demo 文档与示例源码可以提炼出Electric 读路径 TanStack Query 乐观写的三条核心做法shape 作为单一事实源UI 的主数据来自useShape而不是轮询 REST 接口写后确认用流而不是轮询matchStream把等待特定变更出现变成一个可超时、可退订的 Promise与 API 调用并行、一起 settle乐观项与流数据按键合并用useMutationState拿 pending 上下文按主键去重合并进渲染列表并用 mutation 缓存清理处理级联状态如清空时移除未完成的插入。仓库中还有更进一步的方向website/docs/sync/integrations/tanstack.md 介绍了 Electric 与 TanStack 的合作项目 TanStack DB嵌入式客户端数据库仓库内也提供了对应示例 examples/tanstack-db-web-starter 与 examples/tanstack-db-expo-starter可作为本示例之后继续深入 TanStack 生态同步方案时的参照。【免费下载链接】electricThe agent platform built on sync.项目地址: https://gitcode.com/GitHub_Trending/el/electric创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考