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.ts | DI 辅助 | 注册引擎与投影（Projection） |
| di/tokens.ts | DI 令牌 | ShareDB 适配器令牌 ID |
| index.ts | 包入口 | 导出公共适配器表面 |
此外仓库中还实际存在ShareDbPubSubPublisher.ts（另一发布器实现）以及两个单元测试文件ShareDbRealtimeEngine.spec.ts、ShareDbPubSubPublisher.spec.ts，用于验证端到端链路（下文详述）。
依赖上，该包仅依赖@teable/v2-core、@teable/v2-di、@teamwork/websocket-json-stream@2.0.0、neverthrow@8.2.0与sharedb@5.2.2（见 package.json），保持极小的基础设施面，全部业务类型均来自 core。
三、发布器契约：IShareDbOpPublisher
ShareDbPublisher.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: ReadonlyArray<string>, op: ShareDbOp): Promise<Result<void, DomainError>>; }要点解读：
ShareDbOp是三种 ShareDB 操作类型的联合：CreateOp（创建文档）、DeleteOp（删除文档）、EditOp（提交操作），与实时引擎的 ensure / delete / applyChange 三个方法一一对应；- 发布不关心频道语义：
channels是只读字符串数组，由调用方（实时引擎）构造； - 返回值统一用
neverthrow的Result：成功为ok(undefined)，失败为携带DomainError的err(...)，与 v2 core 的错误体系（DomainError）保持一致，避免异常抛出破坏函数式流程。
这一端口设计使“发布操作”这一行为可替换：既可以直接提交给本地 ShareDB backend（ShareDbBackendPublisher），也可以转发给 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)； - del：fetch 后若文档不存在，先
doc.create({}, 'json0', ...)再doc.del(...)，并容忍 “Document already exists” 这类并发竞争错误； - edit：fetch 后调用
doc.submitOp(op.op, options, done)。
值得注意的两个实现细节：
- 提交选项固定携带
source: '@@v2-projection'：这是操作来源标记，ShareDB 会据此把该操作识别为服务端投影（Projection）产生的操作，避免回环广播给发起端； - 错误处理统一收敛：
done回调中把任意错误包装为DomainError（domainError.fromUnknown或domainError.unexpected），并记录logger.warn，同时无论成败都会connection.close()释放连接，避免连接泄漏。
4.2 ShareDbPubSubPublisher：经由 PubSub 中间件发布
ShareDbPubSubPublisher.ts 是另一实现，它只持有 ShareDB 的PubSub子集（Pick<PubSub, 'publish'>），把channels原样转发给pubsub.publish(channelList, op, cb)。这在多实例部署（如 Redis PubSub）场景下非常关键：ShareDB 的 PubSub 层负责把操作广播到其他实例，从而实现跨进程、跨节点的实时同步，而非像 Backend 版本那样只在单实例内部生效。
五、实时引擎：ShareDbRealtimeEngine如何把抽象变更翻译为 json0
ShareDbRealtimeEngine.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: 'json0'（ShareDB 内置的 JSON 操作类型），data为传入的初始值，v: 0表示版本起点，src/seq构成操作标识，m.ts记录毫秒时间戳元数据。
5.3 applyChange：变更到 json0 的映射
applyChange是适配器最富技术含量的部分：它把 v2 core 的领域变更RealtimeChange翻译为 json0 操作数组，见私有方法toJson0Op：
- set（对象字段替换）：若携带
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的ShareDbOp，v: 1表示删除操作作用于版本 1。
5.5 操作来源标记
所有操作都经过toProjectionSource(requestId)生成src，格式为@@v2-projection:${requestId ?? 'unknown'}，把当前请求上下文（requestId）编码进操作来源，既延续了@@v2-projection的服务端投影语义，又保留了可审计的请求溯源能力。
六、WebSocket 传输辅助：ShareDbWebSocketServer
ShareDbWebSocketServer.ts 解决“ShareDB 服务端如何接到 WebSocket 上”的问题：
- 构造器注入 ShareDB 实例（
ShareDbClass）与可选 logger； attach(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:*来源标记；- 注入的
IShareDbOpPublisher（Backend 或 PubSub 实现）按[collection, collection.docId]频道发布ShareDbOp； - ShareDB 校验版本并应用操作，通过
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），仅供参考
