Electric YJS 协同编辑实战用 Postgres 做后端搭建多人在线 CodeMirror 编辑器examples/yjs 全解析【免费下载链接】electricThe agent platform built on sync.项目地址: https://gitcode.com/GitHub_Trending/el/electric导读本文基于 examples/yjs 示例讲解如何用 Yjs 的连接 Provider「Y-Electric」把多人在线协同编辑CodeMirror 编辑器 Yjs CRDT Awareness 光标/在线状态无缝接入 Electric 与 Postgres文档更新与在线状态全部通过 Postgres 表持久化再借由 Electric 的 Shape 同步机制广播给所有客户端无需额外部署任何实时消息基础设施。读完本文你将掌握如何在 pnpm monorepo 中一键启动该示例、数据库表与触发器如何设计、写 API 与 Shape 代理如何实现、ElectricProvider的每个配置项如何理解以及 Y-Electric 在底层是如何用 ShapeStream Yjs 协议完成增量同步与断线恢复的。一、示例概览为什么协同编辑只需要一个 Postgres传统实时协同编辑如多人共享文档通常需要自建 WebSocket 服务、操作转换服务器或专门的实时后端。examples/yjs展示了另一条路径数据面Yjs 的Y.Doc产生二进制增量更新update由示例自带的轻量写 API 落库为 Postgres 的BYTEA字段同步面Electric 通过 Shape 机制监听 Postgres 表变化把新增的更新行以增量流SSE推送给所有订阅客户端在线状态面Yjs 的 Awareness光标、选中、在线用户等临时状态同样以表的形式存储并复用同一条 Shape 通道分发。因此整个系统只需要「Postgres Electric 你自己的一个 HTTP 写端点」Y-Electric 负责把 Yjs 生态与 Electric 的读路径粘合起来。正如 packages/y-electric/README.md 所总结的典型工作流只有四步开发者暴露一个 Shape 代理端点用于授权 Shape 请求客户端为Y.Doc定义一条 Shape 来同步变更开发者暴露一个写 API 处理 Yjs 更新Y-Electric 自动在所有已连接客户端之间共享更新。目录结构速览路径作用examples/yjs/db/migrations/01-create_yjs_tables.sql文档更新表、Awareness 表及清理触发器examples/yjs/src/server/server.tsHono 写 API接收更新落库 Shape 代理examples/yjs/src/client/components/electric-editor.tsx客户端ElectricProvider装配与 CodeMirror 接入examples/yjs/src/client/common/utils.tsPostgresbytea→ YjsDecoder的解析工具packages/y-electric/srcY-Electric Provider 核心实现二、环境准备在 pnpm monorepo 中安装与构建该示例是 ElectricSQL monorepo 的一部分必须作为 pnpm workspace 的一员构建运行因为它依赖electric-sql/y-electricworkspace:*等本地包。2.1 安装全部工作区依赖先进入仓库根目录cd ../../安装并构建所有 workspace 包与示例注意示例依赖的electric-sql/client、y-electric等都会一并构建pnpm install pnpm run -r build2.2 回到示例目录并启动cd examples/yjs启动后端服务Postgres Electric使用 Docker Composepnpm backend:up注意backend:up总是会停止并删除其他示例后端容器挂载的 volume。这是有意为之确保示例每次都在干净的数据库与磁盘上启动。因此不要在跑别的示例如 todo-app、tanstack时执行该命令。该命令实际展开为两步见 examples/yjs/package.json先执行仓库根部的example-backend:up以PROJECT_NAMEyjs标识容器随后立即执行数据库迁移PROJECT_NAMEyjs pnpm -C ../../ run example-backend:up pnpm db:migratedb:migrate使用databases/pg-migrations应用./db/migrations目录下的 SQL并通过根目录.env.dev注入连接信息dotenv -e ../../.env.dev -- pnpm exec pg-migrations apply --directory ./db/migrations2.3 分别启动服务端与客户端服务端Hono tsx watch热重载pnpm dev:server客户端Vite默认监听src/clientpnpm dev:client本地默认端口约定Electric API 为http://localhost:3000示例服务端为http://localhost:3002见 src/server/server.ts 的PORT逻辑与客户端VITE_SERVER_URL默认值Postgres 为localhost:54321见服务端DATABASE_URL回退值。也可以用一条命令同时拉起两端pnpm start-all结束时停止后端并清理容器pnpm backend:down三、数据库设计两张表承载「文档更新」与「Awareness」3.1 文档更新表 ydoc_update01-create_yjs_tables.sql 的第一张表把 Yjs 文档产生的二进制增量更新逐条落库CREATE TABLE ydoc_update( id SERIAL PRIMARY KEY, room TEXT, update BYTEA NOT NULL );room用于区分不同协同房间示例中房间固定为electric-demoupdate以BYTEA保存 Yjs update 的二进制内容每条 update 都是 Yjs 协议的增量片段客户端通过「从某偏移量订阅新行」的方式增量拉取并Y.applyUpdate回放即可重建完整的文档状态。3.2 Awareness 表 ydoc_awareness在线状态光标、选择、在线用户是易失数据其持久化策略与文档更新不同——每个客户端只保留一行最新状态以(client_id, room)为主键CREATE TABLE ydoc_awareness( client_id TEXT, room TEXT, update BYTEA NOT NULL, updated_at TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY (client_id, room) );配合 UpsertON CONFLICT ... DO UPDATE写入Awareness 表始终只保存每个客户端的最新快照。3.3 过期清理触发器客户端离线后 Provider 无法可靠检测其消失因此在数据库侧用触发器做垃圾回收每当插入或更新 Awareness 行时把同房间内 30 秒未更新的旧行删除。CREATE OR REPLACE FUNCTION gc_awareness_timeouts() RETURNS TRIGGER AS $$ BEGIN DELETE FROM ydoc_awareness WHERE updated_at (CURRENT_TIMESTAMP - INTERVAL 30 seconds) AND room NEW.room; RETURN NEW; END; $$ LANGUAGE plpgsql; CREATE TRIGGER gc_awareness_timeouts_trigger AFTER INSERT OR UPDATE ON ydoc_awareness FOR EACH ROW EXECUTE FUNCTION gc_awareness_timeouts();这套表结构在 packages/y-electric/README.md 中有等价的通用模板ydoc_updates/ydoc_awareness与 UUID 主键版本你可以直接按需复制改造。四、服务端实现轻量写 API Shape 代理Y-Electric 的同步链路是「Electric 管读、你的服务管写」因此服务端只需要两件事接收更新写库、把 Shape 请求代理给 Electric。示例用 Hono 实现约 200 行完整代码见 examples/yjs/src/server/server.ts。4.1 统一写端点 PUT /api/update服务端暴露单个PUT /api/update端点通过 URL 查询参数区分两种写入?roomroom文档更新写入ydoc_update?roomroomclient_ididAwareness 更新Upsert 到ydoc_awareness。app.put(/api/update, async (c: Context) { const requestParams await parseRequest(c) if (!requestParams.isValid) { return c.json({ error: requestParams }, 400) } if (client_id in requestParams) { await upsertAwarenessUpdate(requestParams, pool) } else { await saveUpdate(requestParams, pool) } return c.json({}) })请求体就是 Yjs 更新的二进制流application/octet-stream服务端原样读取为Uint8Array后落库export async function saveUpdate({ room, update }: Update, pool: Pool) { const q INSERT INTO ydoc_update (room, update) VALUES ($1, $2) await pool.query(q, params) } export async function upsertAwarenessUpdate({ room, client_id, update }, pool) { const q INSERT INTO ydoc_awareness (room, client_id, update) VALUES ($1, $2, $3) ON CONFLICT (client_id, room) DO UPDATE SET update $3, updated_at now() await pool.query(q, params) }对应地Y-Electric 源码中的send函数 正是用PUT加Content-Type: application/octet-stream把编码后的更新发往sendUrl所以你的写端点应接受PUT与二进制 body。4.2 Shape 代理端点 /shape-proxy/v1/shape客户端不从 Electric 直连取 Shape便于以后接入鉴权而是统一走示例服务端的代理。代理把客户端的查询参数原样转发给 Electric 的/v1/shape并透传响应头app.get(/shape-proxy/v1/shape, async (c: Context) { const electricUrl process.env.ELECTRIC_URL || http://localhost:3000 const originUrl new URL(${electricUrl}/v1/shape) url.searchParams.forEach((value, key) originUrl.searchParams.set(key, value)) // ...转发 headers若配置了 ELECTRIC_SOURCE_ID / ELECTRIC_SOURCE_SECRET 则附加 source_id 与 secret })代理还特意保留了 Electric 同步所需的关键响应头——electric-offset、electric-handle、electric-schema、electric-total-count——并处理了content-encoding可能带来的解析问题。这正是ElectricProvider恢复增量位置所依赖的元数据。4.3 其他细节服务端开启cors()允许任意来源exposeHeaders同样声明了electric-offset、electric-handle、electric-schema、electric-cursor等头src/server/server.ts提供/health健康检查供容器编排与负载均衡器使用连接串回退默认值postgresql://postgres:passwordlocalhost:54321/electric与本地 Docker Compose 一致生产环境通过DATABASE_URL注入。五、客户端实现ElectricProvider 装配与 CodeMirror 接入客户端核心在 electric-editor.tsx创建Y.Doc与Awareness配置ElectricProvider再把它接进 CodeMirror。5.1 创建 Yjs 文档与 Awarenessconst ydoc new Y.Doc() const awareness new Awareness(ydoc) awareness.setLocalStateField(user, { name: user.color, color: user.color, colorLight: user.light, })每个标签页随机挑选一种用户颜色lib0/random用于光标与用户名渲染这就是 Awareness 中「我是谁」的信息来源。5.2 离线持久化IndexedDB 本地恢复状态const databaseProvider new IndexeddbPersistence(user.color, ydoc) const resumeStateProvider new LocalStorageResumeStateProvider(user.color)IndexeddbPersistence是 Yjs 生态标准的数据库 Provider把Y.Doc全量存进浏览器 IndexedDB实现离线可用与秒开LocalStorageResumeStateProvider来自 packages/y-electric/src/local-storage-resume-state.ts把「已同步到的 Shape 位置offset/handle与文档状态向量」存进 localStorage。有了恢复点客户端重连时只需拉取增量而不是重新传输整个文档。5.3 配置 ElectricProviderconst options: ElectricProviderOptionsUpdateTableSchema, UpdateTableSchema { doc: ydoc, documentUpdates: { shape: { url: shapeUrl.href, params: { table: ydoc_update, where: room ${room}, }, parser: parseToDecoder, liveSse: true, }, sendUrl: new URL(/api/update?room${room}, serverUrl), getUpdateFromRow: (row) row.update, }, awarenessUpdates: { shape: { url: shapeUrl.href, params: { table: ydoc_awareness, where: room ${room} }, parser: parseToDecoder, liveSse: true, }, sendUrl: new URL(/api/update?room${room}client_id${ydoc.clientID}, serverUrl), protocol: awareness, getUpdateFromRow: (row) row.update, }, resumeState: resumeStateProvider.load(), debounceMs: 100, }各配置项的语义与底层影响对照 packages/y-electric/src/types.ts 说明如下配置项含义doc要同步的Y.Doc实例documentUpdates.shape文档更新的 Shape 配置table指定ydoc_updatewhere按room过滤parser用parseToDecoder把bytea十六进制串解析成 YjsDecoderliveSse: true开启 SSE 实时推送本地开发需 HTTPS 才能用documentUpdates.sendUrl文档更新的写端点PUT 二进制documentUpdates.getUpdateFromRow从行对象中取出更新列示例为row.update这使 Y-Electric 能适配任意后端表结构见 packages/y-electric/README.mdawarenessUpdates.shape / sendUrl同理但针对ydoc_awareness表sendUrl需附带client_idawarenessUpdates.protocol传入共享的Awareness实例resumeState启动时的恢复点resumeStateProvider.load()读取debounceMs文档更新的防抖窗口毫秒。示例设为100编辑时合并高频更新为0或省略则立即发送见types.ts注释与 y-electric.ts 的scheduleSendOperationsconnect可选默认true构造后自动连接fetchClient可选自定义 fetch 实现用于写请求parser: parseToDecoder的实现见 src/client/common/utils.tsPostgres 的bytea默认以\x前缀的十六进制字符串传输工具函数先去前缀、再按字节转成Uint8Array最终交给lib0/decoding生成 Yjs 解码器。该工具与 packages/y-electric/src/utils.ts 完全同源。5.4 接入 CodeMirror 与连接控制文档加载完成后IndexeddbPersistence触发synced再创建 Provider 和编辑器provider.current new ElectricProvider(options) resumeStateUnsubscribeHandler resumeStateProvider.subscribeToResumeState(provider.current) provider.current.on(status, statusHandler) const ytext ydoc.getText(room) const state EditorState.create({ doc: ytext.toString(), extensions: [ keymap.of([...yUndoManagerKeymap]), basicSetup, javascript(), EditorView.lineWrapping, yCollab(ytext, awareness), ], })yCollab是y-codemirror.next提供的绑定把 CodeMirror 文本与Y.Text双向同步subscribeToResumeState(provider)订阅 Provider 的resumeState事件并把最新恢复点写回 localStorage实现见 local-storage-resume-state.ts界面上提供「connect / disconnect」按钮通过provider.disconnect()与provider.connect()切换网络用来直观演示离线与重连行为electric-editor.tsx。六、底层原理Y-Electric 是如何完成同步的深入 packages/y-electric/src/y-electric.ts可以看到ElectricProvider继承自ObservableV2lib0/observable本质上是「ShapeStream 订阅者 Yjs 更新发送者」的合体。6.1 上行文档更新如何发出Y.Doc的每次本地编辑触发update事件applyDocumentUpdate把增量缓存进pendingChanges并通过debounceMs合并批量y-electric.tssendOperations把缓存的更新用encoding.writeVarUint8Array编码PUT到documentUpdates.sendUrly-electric.ts发送失败时更新会重新batch回缓存并断开连接触发disconnect等待下次重连时重发——这就是断网不丢编辑的保证。6.2 下行如何从 Postgres 读到别人的编辑connect()内部创建ShapeStream来自electric-sql/client并把它与恢复点合并const operationsStream new ShapeStreamRowWithDocumentUpdate({ ...this.documentUpdates.shape, ...this.resumeState.document, // 从上次的 offset/handle 续传 signal: abortController.signal, })收到变更消息后operationsShapeHandler从行中解码出更新并回放到本地文档const decoder this.documentUpdates.getUpdateFromRow(message.value) while (decoder.pos ! decoder.arr.length) { const operation decoding.readVarUint8Array(decoder) Y.applyUpdate(this.doc, operation, server) }注意Y.applyUpdate(..., server)的 origin 标记——回放来自服务端的更新时applyDocumentUpdate直接 return避免把自己的写入再发回去形成回环y-electric.ts。当收到up-to-date控制消息时Provider 记录当前 offset/handle、编码当前状态向量、标记synced并触发resumeState事件y-electric.ts。6.3 Awareness 的双向通道上行Awareness 的update事件仅本地、且已连接时把发生变化的 client 集合编码成encodeAwarenessUpdate后 PUT 给写端点数据库侧用ON CONFLICT保持每客户端一行y-electric.ts下行awarenessShapeHandler处理ydoc_awareness的 Shape 流——delete操作触发removeAwarenessStates客户端离线的信号新增/更新行则applyAwarenessUpdate回放到本地 Awarenessy-electric.ts断开disconnect()会先主动广播removeAwarenessStates通知其他客户端自己下线再清空本地状态y-electric.ts。6.4 恢复状态ResumeState的组成ResumeStatetypes.ts由两部分组成export type ResumeState { document?: { offset: Offset handle: string } stableStateVector?: Uint8Array }offset/handle是 Shape 的续传游标决定从哪一行继续拉取stableStateVector是文档最后一次同步成功时的状态向量。构造 Provider 时如果存在该向量会用Y.encodeStateAsUpdate(doc, stableStateVector)算出「本地与已同步状态之间的差异」作为pendingChanges优先上传y-electric.ts从而避免在 IndexedDB 中已有历史的情况下重复传输整个文档。七、扩展与部署7.1 生产部署SST AWS Neon示例提供了完整的云部署配置 examples/yjs/sst.config.ts通过createDatabaseForCloudElectric创建 Neon 数据库并自动执行./db/migrations迁移后端以容器方式部署到共享 ECS 集群cluster.addService环境变量注入ELECTRIC_URL、DATABASE_URL、ELECTRIC_SOURCE_ID、ELECTRIC_SOURCE_SECRET并通过/health做健康检查前端以sst.aws.StaticSite部署构建时传入VITE_SERVER_URL指向后端服务域名Docker 镜像由 examples/yjs/Dockerfile 构建多阶段构建、pnpm install --frozen-lockfile、build:server产出dist/server后以node dist/server/server.js运行。7.2 换成自己的房间与表要开新的协同房间改客户端room常量并保持where过滤与sendUrl查询参数一致即可。若你的表字段名不同例如op列代替update只需修改getUpdateFromRow: (row) row.op其余逻辑不变——这正是 Y-Electric「适配任意后端 schema」的设计packages/y-electric/README.md。7.3 测试与验证服务端导出honoApp便于集成测试仓库配置了 vitest.config.tsglobals 模式匹配*.{test,spec}.{js,ts}浏览器端行为可通过 playwright.config.ts 扩展端到端测试验证多标签页同步、Awareness 展示与断线重连。八、小结examples/yjs完整演示了「Yjs 协作编辑 Electric 增量同步 Postgres 持久化」的最小闭环数据库两张表加一个清理触发器、服务端一个写端点加一个 Shape 代理、客户端一次ElectricProvider装配就得到了支持离线、断线恢复、在线状态与实时多端同步的协同编辑器。其核心思想——写路径自持、读路径交给 Electric——也正是 Y-Electric 作为 Yjs connection provider 能在整个 Yjs 生态和既有应用中即插即用的原因。如果你想在自己项目里复刻建议从本示例的迁移脚本与server.ts起步再按ElectricProviderOptions的类型定义逐项对齐你的表结构与接口即可。【免费下载链接】electricThe agent platform built on sync.项目地址: https://gitcode.com/GitHub_Trending/el/electric创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考