teable v2 实时架构解析：adapter-realtime-sharedb 适配器从 Op 发布到 WebSocket 传输的完整实现

teable v2 实时架构解析adapter-realtime-sharedb 适配器从 Op 发布到 WebSocket 传输的完整实现【免费下载链接】teable✨ AI Spreadsheet for Business项目地址: https://gitcode.com/GitHub_Trending/te/teable导读本文围绕 teable 仓库packages/v2/adapter-realtime-sharedb适配器包展开讲解它如何基于 ShareDB 为 v2 核心层提供IRealtimeEngine实时引擎实现通过可插拔的 Op 发布器Publisher将创建、编辑、删除操作发布到 ShareDB并提供轻量级 WebSocket 传输辅助。读完本文你将掌握该适配器的职责边界、每个源文件的具体作用、json0 操作转换细节、双发布器Backend 直连与 PubSub 中间件的实现差异以及 DI 注册与包导出的完整装配方式可直接用于理解或扩展 teable 的实时协作链路。一、包级职责适配器在 v2 架构中的定位按 ARCHITECTURE.md 的声明本包承担三项核心职责为 v2 core 提供 ShareDB 支撑的IRealtimeEngine实现—— 这是核心契约core 层定义实时引擎抽象本适配器把抽象的“变更应用”翻译成 ShareDB 操作通过可插拔的 Publisher 发布 ShareDB 操作create / edit / delete—— 发布方式不绑死既可直连 ShareDB backend也可走 ShareDB 的 PubSub为 ShareDB 服务端提供小型 WebSocket 传输辅助—— 负责把 WebSocket 连接桥接成 ShareDB 所需的流。从包名与目录结构packages/v2/adapter-realtime-sharedb/看它属于 v2 适配器族adapter-*与adapter-db-postgres-*、adapter-repository-postgres、adapter-realtime-broadcastchannel等并列共同构成“core 领域逻辑 可替换基础设施适配器”的分层设计。二、文件清单适配器包的构成全览ARCHITECTURE.md 为每个文件标注了“角色 目的”本包源码目录与之一一对应文件角色职责ARCHITECTURE.md架构说明描述适配器包范围ShareDbPublisher.ts适配器端口定义 Op 发布器契约与类型ShareDbBackendPublisher.ts适配器辅助通过 ShareDB backend 提交操作ShareDbRealtimeEngine.ts实时适配器把IRealtimeEngine映射为 ShareDB 操作ShareDbWebSocketServer.ts传输辅助将 ShareDB 绑定到 WebSocket 服务websocket-json-stream.d.ts类型垫片声明 WebSocket JSON 流模块类型di/register.tsDI 辅助注册引擎与投影Projectiondi/tokens.tsDI 令牌ShareDB 适配器令牌 IDindex.ts包入口导出公共适配器表面此外仓库中还实际存在ShareDbPubSubPublisher.ts另一发布器实现以及两个单元测试文件ShareDbRealtimeEngine.spec.ts、ShareDbPubSubPublisher.spec.ts用于验证端到端链路下文详述。依赖上该包仅依赖teable/v2-core、teable/v2-di、teamwork/websocket-json-stream2.0.0、neverthrow8.2.0与sharedb5.2.2见 package.json保持极小的基础设施面全部业务类型均来自 core。三、发布器契约IShareDbOpPublisherShareDbPublisher.ts 定义了整个适配器的核心端口import { type DomainError } from teable/v2-core; import type { Result } from neverthrow; import type { CreateOp, DeleteOp, EditOp } from sharedb; export type ShareDbOp CreateOp | DeleteOp | EditOp; export interface IShareDbOpPublisher { publish(channels: ReadonlyArraystring, op: ShareDbOp): PromiseResultvoid, DomainError; }要点解读ShareDbOp是三种 ShareDB 操作类型的联合CreateOp创建文档、DeleteOp删除文档、EditOp提交操作与实时引擎的 ensure / delete / applyChange 三个方法一一对应发布不关心频道语义channels是只读字符串数组由调用方实时引擎构造返回值统一用neverthrow的Result成功为ok(undefined)失败为携带DomainError的err(...)与 v2 core 的错误体系DomainError保持一致避免异常抛出破坏函数式流程。这一端口设计使“发布操作”这一行为可替换既可以直接提交给本地 ShareDB backendShareDbBackendPublisher也可以转发给 ShareDB PubSub 中间件ShareDbPubSubPublisher。四、两种 Publisher 实现Backend 直连与 PubSub 中间件4.1 ShareDbBackendPublisher通过 backend 直接提交ShareDbBackendPublisher.ts 使用 ShareDB backend 的connect()建立一个内部连接然后按操作类型执行 fetch → create / del / submitOp 的标准三步流程create先doc.fetch检查若doc.type已存在说明文档已创建直接跳过幂等否则调用doc.create(op.create.data, op.create.type, options, done)delfetch 后若文档不存在先doc.create({}, json0, ...)再doc.del(...)并容忍 “Document already exists” 这类并发竞争错误editfetch 后调用doc.submitOp(op.op, options, done)。值得注意的两个实现细节提交选项固定携带source: v2-projection这是操作来源标记ShareDB 会据此把该操作识别为服务端投影Projection产生的操作避免回环广播给发起端错误处理统一收敛done回调中把任意错误包装为DomainErrordomainError.fromUnknown或domainError.unexpected并记录logger.warn同时无论成败都会connection.close()释放连接避免连接泄漏。4.2 ShareDbPubSubPublisher经由 PubSub 中间件发布ShareDbPubSubPublisher.ts 是另一实现它只持有 ShareDB 的PubSub子集PickPubSub, publish把channels原样转发给pubsub.publish(channelList, op, cb)。这在多实例部署如 Redis PubSub场景下非常关键ShareDB 的 PubSub 层负责把操作广播到其他实例从而实现跨进程、跨节点的实时同步而非像 Backend 版本那样只在单实例内部生效。五、实时引擎ShareDbRealtimeEngine如何把抽象变更翻译为 json0ShareDbRealtimeEngine.ts 标注injectable()通过构造器注入发布器inject(v2ShareDbTokens.publisher)实现IRealtimeEngine的三个方法ensure、applyChange、delete。5.1 文档标识解析三个方法都先调用RealtimeDocIdValue.parse(docId)来自 v2 core把领域层的RealtimeDocId解析为{ collection, docId }二元组解析失败则直接返回err。频道channels统一构造为[collection, \${collection}.${documentId}]——即“整集合”与“单文档”两级订阅粒度。5.2 ensure创建文档ensure构造一条create类型的ShareDbOp其中type: json0ShareDB 内置的 JSON 操作类型data为传入的初始值v: 0表示版本起点src/seq构成操作标识m.ts记录毫秒时间戳元数据。5.3 applyChange变更到 json0 的映射applyChange是适配器最富技术含量的部分它把 v2 core 的领域变更RealtimeChange翻译为 json0 操作数组见私有方法toJson0Opset对象字段替换若携带oldValue则生成{ p: path, oi: newValue, od: oldValue }json0 对象替换带旧值可做冲突检测否则仅{ p: path, oi: newValue }insert列表插入生成{ p: [...path, index], li: value }路径拼上插入下标delete列表删除需要生成多条操作且从后往前删除for (let i change.count - 1; i 0; i--)保证删除过程中前面的下标始终有效——这是 json0 列表语义的经典陷阱源码注释明确说明“to keep indices valid”空变更数组直接返回domainError.validation({ message: No changes to apply })。批量变更通过flatMap展开为单个 json0 操作序列随同v: options?.version ?? 0一起提交把版本并发控制交给 ShareDB 校验。5.4 delete删除文档delete构造del: true的ShareDbOpv: 1表示删除操作作用于版本 1。5.5 操作来源标记所有操作都经过toProjectionSource(requestId)生成src格式为v2-projection:${requestId ?? unknown}把当前请求上下文requestId编码进操作来源既延续了v2-projection的服务端投影语义又保留了可审计的请求溯源能力。六、WebSocket 传输辅助ShareDbWebSocketServerShareDbWebSocketServer.ts 解决“ShareDB 服务端如何接到 WebSocket 上”的问题构造器注入 ShareDB 实例ShareDbClass与可选 loggerattach(server)订阅任意满足{ on(connection, listener) }形状的 WebSocket 服务如ws、Node HTTP upgrade 等这是最小接口约束不依赖具体框架handleConnection中用teamwork/websocket-json-stream把 socket 包装为 JSON 流再交给shareDb.listen(stream, request)同时过滤掉 “WebSocket CLOSING or CLOSED.” 这类正常关闭噪音其余错误以logger.warn记录。websocket-json-stream.d.ts则是纯类型垫片teamwork/websocket-json-stream未自带类型故在包内以declare module声明默认导出为any并在 index.ts 开头通过/// reference path./websocket-json-stream.d.ts /引用。七、DI 装配令牌、注册与投影7.1 令牌di/tokens.ts 定义了唯一令牌export const v2ShareDbTokens { publisher: Symbol(v2.adapter.realtime.sharedb.publisher), } as const;发布器以实例registerInstance注入引擎与投影以类register注入且生命周期为Lifecycle.Singleton。7.2 注册函数与硬性依赖检查di/register.ts 导出registerV2ShareDbRealtime(c, config)config.publisher缺失时抛出Invalid v2 ShareDB realtime config注册ShareDbRealtimeEngine为v2CoreTokens.realtimeEngine的实现硬性依赖校验若容器中未注册v2CoreTokens.tableRepository或v2CoreTokens.tableMapper直接抛错ShareDB realtime requires tableRepository and tableMapper registrations——说明实时引擎依赖表仓储与映射器随后批量注册 11 个实时投影类均Lifecycle.Singleton覆盖表与字段生命周期TableCreatedRealtimeProjection、FieldCreated/Deleted/Updated/OptionsAddedRealtimeProjection、ComputedActivityRealtimeProjection、ViewColumnMetaUpdatedRealtimeProjection以及记录操作RecordCreated/Updated/ReorderedRealtimeProjection、RecordsBatchCreated/Updated/DeletedRealtimeProjection。这些投影类来自 v2 core 的application/projections如 FieldCreatedRealtimeProjection.ts、RecordCreatedRealtimeProjection.ts是“把领域事件折叠为实时文档快照”的消费者与 ShareDB 文档的 create/edit/delete 一一呼应。八、端到端验证测试如何佐证整条链路ShareDbRealtimeEngine.spec.ts 直接演示了“服务端 WebSocket 客户端订阅 引擎发布”的完整闭环可作为集成参考startShareDbRuntime创建new ShareDb()后端用ws包在随机端口port: 0启动 WebSocketServer路径为/socket并shareDbWebSocket.attach(wsServer)客户端侧用new WebSocket(url)new Connection(socket)connection.get(collection, docId)建立订阅readyPromise 等待首次 fetch 完成发布侧则分别用ShareDbBackendPublisher与ShareDbPubSubPublisher构造引擎并调用ensure/applyChange/delete断言订阅端快照与变更事件符合预期。该测试同时覆盖了两种 Publisher证明“发布器可插拔”不是纸面设计而是被单元测试验证过的真实约束。九、整体数据流小结结合上述源码一次典型的实时变更可概括为v2 core 领域层产生变更如记录更新由实时投影捕获ShareDbRealtimeEngine.applyChange把RealtimeChange翻译为 json0 操作并带上v2-projection:*来源标记注入的IShareDbOpPublisherBackend 或 PubSub 实现按[collection, collection.docId]频道发布ShareDbOpShareDB 校验版本并应用操作通过ShareDbWebSocketServer桥接的 WebSocket 连接把变更推送给订阅客户端。这一设计把“领域实时语义”RealtimeChange与“协作协议实现”json0/ShareDB彻底解耦core 只依赖IRealtimeEngine抽象而具体是 ShareDB、BroadcastChannel 还是其他实现由 DI 注册决定——这也是 teable v2 适配器架构的核心价值所在。【免费下载链接】teable✨ AI Spreadsheet for Business项目地址: https://gitcode.com/GitHub_Trending/te/teable创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考