Electric 通过数据库同步（Through-the-DB）写入模式：基于 PGlite 的本地优先完整实现

Electric 通过数据库同步Through-the-DB写入模式基于 PGlite 的本地优先完整实现【免费下载链接】electricThe agent platform built on sync.项目地址: https://gitcode.com/GitHub_Trending/el/electric导读本文深入解析 Electric 写模式Write Patterns示例中的第四种模式——Through-the-database sync通过数据库同步。该模式在本地嵌入一个 PGlite 数据库将读路径Electric 同步的不可变数据、写路径本地乐观状态 变更日志全部收敛到同一套数据库结构内组件只需读写一个统一的todos视图即可。读完本文你将掌握如何使用 shadow table影子表 视图 触发器构建本地写入通道如何通过pg_notify与ChangeLogSynchronizer实现自动后台同步以及该模式在回滚、并发 rebase 上的取舍。模式概述当读写都发生在本地数据库里Through-the-database sync 是 write-patterns 示例中四种写入模式的第四种也是把共享持久乐观状态第三种模式推进到极致的方案乐观状态不再存放在 valtio store 或 localStorage而是直接落在一个本地嵌入式 PGlite 数据库中。该示例同时使用了Electric 用于读路径同步read-path sync本地对单张数据库表的读写实际是一个视图共享、持久的乐观状态自动变更检测与后台同步。具体来说整个方案被拆成读路径与写路径两条管线读路径同步进本地通过 Electric 将服务端数据同步进一张不可变表immutable tabletodos_synced把本地乐观状态持久化到一张影子表shadow tabletodos_local读取时用一张视图todos把两者合并。写路径同步回服务端通过INSTEAD OF触发器自动检测本地写入将写入操作记录进变更日志表changes后台同步器把变更POST到 API 服务器。该模式的全部代码位于 patterns/4-through-the-db 目录核心文件为local-schema.sql本地数据库结构、db.tsPGlite 初始化与读路径接入、sync.ts写路径同步器、index.tsxReact 组件层。优势Benefits把读写都收进本地数据库带来的是完整的离线支持与共享乐观状态组件只与本地数据库交互完全不需要在网络层编写任何代码数据抓取fetching被 Electric 同步抽象掉数据发送sending被变更消息日志抽象掉由于乐观状态持久化在数据库中页面刷新、组件卸载都不会丢失未同步的写入。从使用场景看该模式尤其适合构建本地优先local-first软件移动端与桌面端应用协作与创作类软件。劣势Drawbacks代价同样明显原文档明确指出三点依赖较重本地嵌入式数据库是相对笨重的依赖本地 Schema 复杂影子表 触发器机制使客户端侧的 schema 定义变得复杂不再是一张普通业务表回滚处理困难后台同步意味着写被服务端拒绝时上下文已丢失。对照共享持久乐观状态模式——那里可以在处理用户输入的上下文内同步检测到写被拒而 through-the-DB 模式下重建这种上下文要困难得多。本地数据库结构三张表 一个视图local-schema.sql是整条读路径与本地写路径的基石包含两个核心表、一个视图和一个变更日志表。todos_synced不可变的同步态存储 Electric 从服务端复制过来的行应用代码只读不写CREATE TABLE IF NOT EXISTS todos_synced ( id UUID PRIMARY KEY, title TEXT NOT NULL, completed BOOLEAN NOT NULL, created_at TIMESTAMP WITH TIME ZONE NOT NULL, -- Bookkeeping column. write_id UUID );注意末尾的write_id字段这是贯穿整个模式的关键簿记列服务端在迁移 02-add-write-id.sql 中为todos表增加了同名可空列它的作用是让本地能够精确识别自己的哪一次写入已通过复制流回流而不是简单按行id匹配。todos_local可变的乐观态影子表本地写操作产生的乐观状态存放在这里CREATE TABLE IF NOT EXISTS todos_local ( id UUID PRIMARY KEY, title TEXT, completed BOOLEAN, created_at TIMESTAMP WITH TIME ZONE, -- Bookkeeping columns. changed_columns TEXT[], is_deleted BOOLEAN NOT NULL DEFAULT FALSE, write_id UUID NOT NULL );与todos_synced相比它多了三列簿记信息changed_columns记录本次本地写改了哪些列供视图合并时做列级覆盖判断is_deleted软删除标记配合视图 WHERE 条件实现已删除行对读取不可见write_id本次本地写入的唯一标识用于和回流数据做匹配。todos视图读写统一入口视图通过FULL OUTER JOIN合并两张表并依据changed_columns做列级覆盖CREATE OR REPLACE VIEW todos AS SELECT COALESCE(local.id, synced.id) AS id, CASE WHEN title ANY(local.changed_columns) THEN local.title ELSE synced.title END AS title, CASE WHEN completed ANY(local.changed_columns) THEN local.completed ELSE synced.completed END AS completed, CASE WHEN created_at ANY(local.changed_columns) THEN local.created_at ELSE synced.created_at END AS created_at FROM todos_synced AS synced FULL OUTER JOIN todos_local AS local ON synced.id local.id WHERE local.id IS NULL OR local.is_deleted FALSE;FULL OUTER JOIN保证两种行都能出现只有服务端数据本地无乐观状态、只有本地数据尚未同步的新行。WHERE local.id IS NULL OR local.is_deleted FALSE则把软删除的行从读取结果中剔除——这也解释了为什么删除要做成更新式的软删除只有 UPDATE 才能携带write_id参与并发 rebase真正的 DELETE 一旦发生就无法回退行被删后只能以 INSERT 重建。changes表写路径的变更日志本地对视图的每次写操作都会被INSTEAD OF触发器同时记录进变更日志这是后台同步器的数据源CREATE TABLE IF NOT EXISTS changes ( id BIGSERIAL PRIMARY KEY, operation TEXT NOT NULL, value JSONB NOT NULL, write_id UUID NOT NULL, transaction_id XID8 NOT NULL );operation取值insert/update/deletevalue是操作载荷JSONBtransaction_id记录写入时的事务 ID用于把同一事务内的一组变更打包成一次事务提交发给服务端。触发器机制合并、软删与写入捕获回流清理触发器乐观状态的自动回收本地写同步到服务端后会经由 Electric 复制流以todos_synced的新行形式回流。此时必须清理掉对应的本地乐观状态否则视图会出现本地值永远覆盖服务端值的问题。清理逻辑在delete_local_on_synced_insert_and_update_trigger中实现CREATE OR REPLACE FUNCTION delete_local_on_synced_insert_and_update_trigger() RETURNS TRIGGER AS $$ BEGIN DELETE FROM todos_local WHERE id NEW.id AND write_id IS NOT NULL AND write_id NEW.write_id; RETURN NEW; END; $$ LANGUAGE plpgsql;关键点在write_id NEW.write_id只在本地 write_id 与服务端回流行的 write_id 一致时才清除乐观状态。这样当其他用户并发更新了同一行时本地乐观状态不会被误清除从而得以在并发变更之上rebase重新合并这正是原文档强调的 merge 逻辑。删除场景则简单得多——delete_local_on_synced_delete_trigger直接按id清理CREATE OR REPLACE FUNCTION delete_local_on_synced_delete_trigger() RETURNS TRIGGER AS $$ BEGIN DELETE FROM todos_local WHERE id OLD.id; RETURN OLD; END; $$ LANGUAGE plpgsql;原文档在 SQL 注释中解释了原因删除可能并发发生但删除操作无法更新write_id也不可回退行一旦被删只能以 INSERT 重建因此按 ID 匹配是安全的。若需要可回退的并发删除可以改用软删除——它本质上就是 UPDATE。两个函数分别挂到AFTER INSERT OR UPDATE与AFTER DELETE触发器上见 local-schema.sql。INSTEAD OF触发器把视图写入转化为本地落库 变更记录PostgreSQL 允许为视图定义INSTEAD OF触发器来接管 DML这让应用代码可以直接对todos视图执行INSERT/UPDATE/DELETE。三个触发器函数对应三种操作以插入为例CREATE OR REPLACE FUNCTION todos_insert_trigger() RETURNS TRIGGER AS $$ DECLARE local_write_id UUID : gen_random_uuid(); BEGIN IF EXISTS (SELECT 1 FROM todos_synced WHERE id NEW.id) THEN RAISE EXCEPTION Cannot insert: id already exists in the synced table; END IF; IF EXISTS (SELECT 1 FROM todos_local WHERE id NEW.id) THEN RAISE EXCEPTION Cannot insert: id already exists in the local table; END IF; -- Insert into the local table. INSERT INTO todos_local (id, title, completed, created_at, changed_columns, write_id) VALUES (NEW.id, NEW.title, NEW.completed, NEW.created_at, ARRAY[title, completed, created_at], local_write_id); -- Record the write operation in the change log. INSERT INTO changes (operation, value, write_id, transaction_id) VALUES (insert, jsonb_build_object(id, NEW.id, title, NEW.title, completed, NEW.completed, created_at, NEW.created_at), local_write_id, pg_current_xact_id()); RETURN NEW; END; $$ LANGUAGE plpgsql;插入触发器做的事很直观校验 id 冲突 → 写入todos_local此时三列全标记为 changed→ 向changes写入一条insert记录。注意pg_current_xact_id()记录当前事务 ID用于后续按事务分组打包。更新触发器todos_update_trigger的逻辑最复杂需要回答这次更新改了什么若本地表无此行的记录与todos_synced逐列用IS DISTINCT FROM比较生成changed_cols后插入本地行若本地已有记录更新本地行并基于changed_columns与当前实际值重新计算哪些列仍算 changed只有既被标记过、值也确实与服务端不同的列才保留同时轮换write_id最后把update操作连同jsonb_strip_nulls(...)处理后的完整新值写入changes。删除触发器todos_delete_trigger实现软删除 变更记录在todos_local中 upsert 一条is_deleted TRUE的行保留/新生成write_id再向changes写入delete操作。正如前面分析软删除让删除具备了参与并发 rebase 的可能性。变更通知触发器驱动后台同步最后一个触发器是写路径同步的发令枪CREATE OR REPLACE FUNCTION changes_notify_trigger() RETURNS TRIGGER AS $$ BEGIN NOTIFY changes; RETURN NEW; END; $$ LANGUAGE plpgsql;每当changes表新增一行就通过 PostgreSQL 的LISTEN/NOTIFY机制向外发出changes主题通知ChangeLogSynchronizer正是订阅了这个通知才得以自动检测本地写入。读路径接线db.ts 中的 PGlite 初始化db.ts负责创建本地数据库并把 Electric 同步接进来const pglite: PGliteWithLive await PGlite.create(DATA_DIR, { extensions: { electric: electricSync(), live, }, }) await pglite.exec(localSchemaMigrations) await pglite.electric.syncShapeToTable({ shape: { url: TODOS_URL }, shapeKey: todos, table: todos_synced, primaryKey: [id], })几个要点数据目录idb://electric-write-patterns-example使用 IndexedDB 后端持久化在浏览器内刷新不丢electricSync()扩展让 PGlite 具备 Electric 同步能力live扩展提供实时查询支持启动时先执行 local-schema.sql 建立本地结构syncShapeToTable把TODOS_URL即 API 服务器代理的/todosshape 端点见 shared/app/config.ts同步进todos_synced表主键为id模块级Map缓存了loadingPromise避免重复初始化。写路径同步器ChangeLogSynchronizer 状态机sync.ts中的ChangeLogSynchronizer实现了一个极简但完整的后台同步状态机它维护两个状态位#hasChangedWhileProcessing处理期间是否又有新变更到达#statusidle|processing。启动start()通过db.listen(changes, handler)订阅变更通知PGlite 对 PostgreSQLLISTEN/NOTIFY的封装并立即执行一轮process()。处理循环process()query()用SELECT * FROM changes WHERE id 位置 ORDER BY id asc拉取当前位置之后的新变更游标位置记录在#positionsend(changes)按transaction_id用Object.groupBy分组、按事务 ID 排序组装成{ id, changes: [...] }[]数组通过共享的api.request见 shared/app/client.ts内置最长 3 分钟、1~20 秒递增的指数退避重试POST到/changes依据返回结果走三个分支acceptedHTTP 2xx→proceed(position)删除id position的已处理变更并推进游标rejectedHTTP 4xx→rollback()执行极为朴素的回滚retry网络错误或 HTTP 5xx→ 标记#hasChangedWhileProcessing true下一轮再试若处理期间又有新变更且未调用stop()立即递归进入下一轮否则状态回到idle。回滚策略是原文档点名提醒的坑async rollback(): Promisevoid { await this.#db.transaction(async (tx) { await tx.sqlDELETE from changes await tx.sqlDELETE from todos_local }) }任何一次写被服务端拒绝就清空全部本地状态与全部待同步变更。原文档明确指出这是 naive 策略并建议更精细的方案例如向用户说明发生了什么、或只清除与被拒绝写入存在因果依赖的本地状态从而最小化数据丢失。原文档也提示这条路径会引出大量复杂性可考虑借助既有框架详见官方 Writes 指南含框架清单。停止stop()置#shouldContinue false、中止进行中的请求并退订通知用于组件卸载时清理见 index.tsx。组件层只与本地视图打交道的 React UIindex.tsx展示了应用代码如何被简化到极致。Wrapper负责加载 PGlite、启动同步器并注入PGliteProviderThroughTheDB组件则完全基于本地数据库工作const results useLiveQueryTodo(SELECT * FROM todos ORDER BY created_at) await db.sql INSERT INTO todos (id, title, completed, created_at) VALUES (${uuidv4()}, ${title}, ${false}, ${new Date()}) 读取用useLiveQuery订阅todos视图任何本地写入或远端同步回流都会实时驱动 UI 刷新新增、勾选完成、删除都直接对视图执行db.sql语句由触发器完成乐观落库与变更记录模板其余部分与另外三种模式保持一致方便在同一个页面里横向对比四种写入模式的在线/离线行为差异示例主页说明见 examples/write-patterns/README.md。服务端配合POST /changes 事务回放写路径的终点是 shared/backend/api.js 中的POST /changes端点。它先用 zod 校验载荷transactionsSchema事务数组每项含id与changes[]变更项含operation、value、write_id然后在一个数据库事务内依次回放await client.query(BEGIN) data.forEach((tx) { tx.changes.forEach(({ operation, value, write_id }) { switch (operation) { case insert: createTodo(value.id, value.title, value.created_at, write_id); break case update: updateTodo(value.id, value.completed, write_id); break case delete: deleteTodo(value.id); break } }) }) await client.query(COMMIT)任一步出错则整体ROLLBACK并返回 500客户端据此判定为rejected而触发回滚。服务端写入todos表时带上write_id正是它让服务端接受 → Electric 回流 → 本地按 write_id 清理乐观状态这条闭环得以成立。此外该 API 还提供GET /todos代理转发 Electric shape 协议仅透传ELECTRIC_PROTOCOL_QUERY_PARAMS内的参数服务端固定tabletodos作为读路径的 shape 端点。如何运行该示例的运行方式与其他模式一致参见 examples/write-patterns/README.md#how-to-run。在仓库根目录安装并构建依赖pnpm install pnpm run -r build在examples/write-patterns目录下启动后端容器pnpm backend:up启动开发服务器pnpm dev结束后拆除后端容器pnpm backend:down总结这条模式的适用边界Through-the-DB 模式把共享持久乐观状态推进到了嵌入式数据库层面读路径由 Electric 同步进不可变表写路径由 shadow table INSTEAD OF触发器 变更日志表自动捕获ChangeLogSynchronizer通过NOTIFY订阅自动把变更按事务分组 POST 回服务端。它以明显的 schema 复杂度和依赖成本换来了应用代码完全离线化、零网络代码的纯粹本地优先体验。如果要动手改造优先关注两个地方一是delete_local_on_synced_insert_and_update_trigger中的 write_id 匹配逻辑并发 rebase 的正确性根基二是ChangeLogSynchronizer.rollback的全量清空策略生产环境需替换为因果感知的精细回滚。对这两个关键点的源码级推演分别见 local-schema.sql 与 sync.ts。【免费下载链接】electricThe agent platform built on sync.项目地址: https://gitcode.com/GitHub_Trending/el/electric创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考