Electric + YJS 协同编辑实战：用 Postgres 做后端，搭建多人在线 CodeMirror 编辑器（examples/yjs 全解析）

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 所总结的，典型工作流只有四步：

1. 开发者暴露一个 Shape 代理端点，用于授权 Shape 请求；
2. 客户端为Y.Doc定义一条 Shape 来同步变更；
3. 开发者暴露一个写 API 处理 Yjs 更新；
4. Y-Electric 自动在所有已连接客户端之间共享更新。

目录结构速览

二、环境准备：在 pnpm monorepo 中安装与构建

该示例是 ElectricSQL monorepo 的一部分，必须作为 pnpm workspace 的一员构建运行，因为它依赖@electric-sql/y-electric（workspace:*）等本地包。

2.1 安装全部工作区依赖

先进入仓库根目录：

cd ../../

安装并构建所有 workspace 包与示例（注意示例依赖的@electric-sql/client、y-electric等都会一并构建）：

pnpm install pnpm run -r build

2.2 回到示例目录并启动

cd examples/yjs

启动后端服务（Postgres + Electric），使用 Docker Compose：

pnpm backend:up

注意：backend:up总是会停止并删除其他示例后端容器挂载的 volume。这是有意为之，确保示例每次都在干净的数据库与磁盘上启动。因此不要在跑别的示例（如 todo-app、tanstack）时执行该命令。

该命令实际展开为两步（见 examples/yjs/package.json）：先执行仓库根部的example-backend:up（以PROJECT_NAME=yjs标识容器），随后立即执行数据库迁移：

PROJECT_NAME=yjs pnpm -C ../../ run example-backend:up && pnpm db:migrate

db:migrate使用@databases/pg-migrations应用./db/migrations目录下的 SQL，并通过根目录.env.dev注入连接信息：

dotenv -e ../../.env.dev -- pnpm exec pg-migrations apply --directory ./db/migrations

2.3 分别启动服务端与客户端

服务端（Hono +tsx watch热重载）：

pnpm dev:server

客户端（Vite，默认监听src/client）：

pnpm 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_update

01-create_yjs_tables.sql 的第一张表把 Yjs 文档产生的二进制增量更新逐条落库：

CREATE TABLE ydoc_update( id SERIAL PRIMARY KEY, room TEXT, update BYTEA NOT NULL );

• room用于区分不同协同房间（示例中房间固定为electric-demo）；
• update以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) );

配合 Upsert（ON 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 查询参数区分两种写入：

• ?room=<room>：文档更新，写入ydoc_update；
• ?room=<room>&client_id=<id>：Awareness 更新，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:password@localhost:54321/electric与本地 Docker Compose 一致，生产环境通过DATABASE_URL注入。

五、客户端实现：ElectricProvider 装配与 CodeMirror 接入

客户端核心在 electric-editor.tsx：创建Y.Doc与Awareness，配置ElectricProvider，再把它接进 CodeMirror。

5.1 创建 Yjs 文档与 Awareness

const 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 配置 ElectricProvider

const options: ElectricProviderOptions<UpdateTableSchema, 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 说明如下：

parser: parseToDecoder的实现见 src/client/common/utils.ts：Postgres 的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继承自ObservableV2（lib0/observable），本质上是「ShapeStream 订阅者 + Yjs 更新发送者」的合体。

6.1 上行：文档更新如何发出

1. Y.Doc的每次本地编辑触发update事件，applyDocumentUpdate把增量缓存进pendingChanges，并通过debounceMs合并批量（y-electric.ts）；
2. sendOperations把缓存的更新用encoding.writeVarUint8Array编码，PUT到documentUpdates.sendUrl（y-electric.ts）；
3. 发送失败时，更新会重新batch回缓存并断开连接（触发disconnect），等待下次重连时重发——这就是断网不丢编辑的保证。

6.2 下行：如何从 Postgres 读到别人的编辑

connect()内部创建ShapeStream（来自@electric-sql/client），并把它与恢复点合并：

const operationsStream = new ShapeStream<RowWithDocumentUpdate>({ ...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回放到本地 Awareness（y-electric.ts）；
• 断开：disconnect()会先主动广播removeAwarenessStates通知其他客户端自己下线，再清空本地状态（y-electric.ts）。

6.4 恢复状态（ResumeState）的组成

ResumeState（types.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.ts（globals 模式，匹配*.{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