reactive-engine

Reactive Engine

一个基于类型化节点构建的响应式状态系统，由以下五个包组成：

包	提供内容
@virtuoso.dev/reactive-engine-core	节点（Cell、Stream、Trigger、Resource、DerivedCell）、Engine、运算符、组合器和 e 命名空间
@virtuoso.dev/reactive-engine-react	EngineProvider、状态钩子、本地和远程诊断钩子
@virtuoso.dev/reactive-engine-query	基于节点实现数据获取的 Query 和 Mutation
@virtuoso.dev/reactive-engine-router	Route、Layout、Guard、Router——路由本身就是节点
@virtuoso.dev/reactive-engine-storage	linkCellToStorage（localStorage / sessionStorage / cookie）

它也在内部为 @virtuoso.dev/data-table 提供支持——远程控制表格时也适用相同的概念，而 data-table 的架构是基于该引擎构建应用的参考（building-on-the-engine）。

心智模型：节点是定义，引擎持有状态

节点构造函数返回惰性引用（运行时为符号），在模块作用域中定义，并以 $ 结尾。它们不持有状态。Engine 实例会在首次使用时惰性激活节点，并持有其值——使用同一节点的两个引擎拥有彼此独立的状态。这使基于引擎的库可以重复使用：一张模块作用域的图，每个组件实例对应一个引擎。

import { Cell, Stream, Engine, e } from '@virtuoso.dev/reactive-engine-core'

const count$ = Cell(0)
const engine = new Engine()
engine.sub(count$, (value) => console.log('count:', value))
engine.pub(count$, 1) // logs 'count: 1'

节点类型——根据状态特性选择：

节点	状态	用途
Cell(initial, distinct?)	有状态，具有当前值	应用状态、设置
DerivedCell(initial, source$)	有状态，跟踪一个源	只读计算状态
Stream<T>(distinct?)	无状态，发出值	事件、命令
Trigger()	无状态，无值	“发生了某事”的信号（重新获取、重置）
Resource(factory)	按引擎初始化的工厂	需要设置或释放的对象

默认情况下，节点使用引用相等（===）进行去重：发布相同的引用（或相等的基本类型值）不会再次发出。更新时务必发布新的引用（new Map(old)、展开运算符）。传入 false 可始终发出；也可传入比较函数 (prev, next) => boolean（true = 相等 = 抑制发出）；首次发出时，prev 为 undefined。

三种 API 形式（最重要的区别）

同一组动词有三种形式——使用错误的形式是最主要的错误来源：

形式	示例	位置	效果
模块级组合器	link、pipe、sub、changeWith、combine / e.*	任意位置，包括模块顶层	延迟连接；节点激活时，每个引擎应用一次
引擎实例方法	engine.pub、engine.sub、engine.getValue	任何持有引擎的位置	立即在该引擎上生效
上下文绑定工具函数	独立的 pub、getValue、pubIn / e.pub 等	仅限订阅回调和节点初始化函数内部	作用于当前正在执行的引擎

在其他位置调用上下文绑定工具函数会抛出 No active engine found. You can use getValue only in the context of node subscription callbacks. 错误——请改用 engine.* 方法、React hooks，或将代码移入订阅中。注意其中的不对称性：独立的 sub / link / changeWith 在模块级使用是安全的（它们会延迟执行）；独立的 pub / getValue 则不安全。

e 导出将所有组合器、运算符和上下文工具函数打包到一个命名空间中——用于连接代码块时非常地道：

e.link(
  e.pipe(
    toggleTheme$,
    e.withLatestFrom(theme$),
    e.map(([, t]) => (t === 'light' ? 'dark' : 'light'))
  ),
  theme$
)

连接图

• link(source$, sink$)——转发发射；pipe(source$, ...ops)——派生出一个新节点
• combine(a$, b$, ...)——任意来源发射时，发出 [a, b, ...]；withLatestFrom(...)（运算符）——被动读取其他节点，仅在来源发射时才发射
• changeWith(cell$, source$, (cell, value) => next)——以 reducer 风格根据流更新 cell（返回一个新引用）
• sub、subMultiple、singletonSub（替换之前的单例订阅——适用于由 React 渲染驱动的回调）
• 运算符：map(fn, distinct?)、filter（支持类型守卫）、mapTo、scan、once、throttleTime、debounceTime、delayWithMicrotask、onNext、handlePromise

发布具有事务性：engine.pubIn({ [a$]: 1, [b$]: 2 }) 会合并为一个周期；图按拓扑顺序执行，因此菱形依赖会以一致的输入向下游发射一次。发射是同步的；没有错误或完成通道（异常会中止该周期——请在边界处处理错误）。订阅时不会重放当前值；需要时使用 getValue 读取 cell。参见事务。

React 集成

import { EngineProvider, useCellValue, usePublisher } from '@virtuoso.dev/reactive-engine-react'

const Counter = () => {
  const count = useCellValue(count$)
  const setCount = usePublisher(count$)
  return <button onClick={() => setCount(count + 1)}>{count}</button>
}

• useCellValue(cell$) 会订阅并触发重新渲染；useCellValues(a$, b$, ...) 返回一个元组，并使用单个组合订阅（对于会一起变化的 cell，优先使用它）；useCell 与 useState 的形式相同；usePublisher(node$) 返回一个稳定的发布函数
• EngineProvider 的 props：initFn(engine)（一次性设置：注册节点、使用 pubIn 设定初始值、连接桥接）、initWith（初始 cell 值）、updateFn / updateDeps（重新发布发生变化的 props；对于回调 props，请使用 singletonSub，这样重新渲染时会替换而不是叠加）、engineId、engineRef 和 diagnostics（在 initFn 之前附加）
• 远程 hooks（useRemoteCellValue、useRemotePublisher、useRemoteCell、useRemoteCellValues、useRemoteEngineDiagnostics）通过 useEngineRef() 或字符串 engineId 访问另一个 provider 的引擎——在该引擎挂载前，值 hooks 会返回 undefined；诊断观察会等待并自动附加
• 组件仅负责投影：读取 cell、发布操作。逻辑放在模块级连接代码中，可使用不依赖 React 的纯 Engine 进行测试

使用诊断排查传播问题

当某个值缺失、重复、意外被抑制、经由错误的引擎路由，或在同步传播期间抛出异常时，请使用诊断：

1. 使用 describeNode 标记根节点、预期接收节点和中间节点。仅当内置的有界摘要无法说明应用专用值时，才添加范围窄的 summarize 函数。
2. 在正确的生命周期中附加诊断：EngineProvider.diagnostics 包含 initFn；useEngineDiagnostics 会在提交后观察最近的引擎；useRemoteEngineDiagnostics 适用于同级工具，并遵循引擎的可用状态。由于 React Strict Mode 可能会多次执行初始化，因此在开发环境中应将 provider 初始化记录视为至少一次。
3. 先禁用值捕获。按顺序读取根节点和事件。检查 distinct-suppressed、no-candidate、剪枝事件和 error.phase，然后再启用 captureValues: 'summary'。
4. 如果状态似乎来自错误的实例，请比较 engineInstanceId 和 engineLabel。对于同步嵌套传播或父子传播，请使用 transactionId、origin 和 parentCycle。异步工作会启动新事务，需要应用层关联。
5. 仅为验证假设所需的节点启用摘要。导出前先脱敏。节点摘要函数会接收实时值，不得修改这些值或产生副作用。
6. 复现问题后移除观察器。Core 和 React 不会保留历史记录；请基于有界外部存储构建诊断 UI，而不要使用 EngineProvider 状态。

请阅读传播诊断，了解记录模型、安全规则和证据与原因对照表。请阅读React 集成，了解 provider、本地钩子、远程钩子、Strict Mode 和 SSR 时机。

数据获取（reactive-engine-query）

export const tasksQuery = Query<{ listId: string }, Task[]>({
  queryFn: async ({ listId }, signal) => {
    const res = await fetch(`/api/tasks?listId=${listId}`, { signal })
    if (!res.ok) throw new Error('Failed to fetch tasks')
    return res.json()
  },
  initialParams: { listId: '' },
})

Query 会返回节点：data$（包含 { type: 'pending' | 'success' | 'error', data, error, isLoading, isFetching, ... } 的 cell）、params$（发布新参数以重新获取）、refetch$、invalidate$（继续显示过期数据，isFetching: true，并重新执行）、unload$（中止并清除）、enabled$。查询会在激活和参数变化时自动执行；每次执行都会通过 signal 中止前一次执行。选项：retry（默认 3，指数退避）、refetchInterval、initialData、enabled。

Mutation({ mutationFn, onSuccess?, onError? }) 会返回 data$、mutate$、reset$。Mutation 永远不会自动执行——请向 mutate$ 发布参数。在图中将 mutation 连接到查询以触发重新获取，而不要在组件中处理：

e.link(
  e.pipe(
    e.merge(createTask.data$, deleteTask.data$),
    e.filter((r) => r.type === 'success'),
    e.map(() => undefined, false) // distinct=false so consecutive successes each fire
  ),
  tasksQuery.refetch$
)

路由（reactive-engine-router）

export const user$ = Route('/users/{id:number}', UserPage) // params typed from the path
const root = Layout('/', ({ children }) => <main>{children}</main>)

<Router routes={[user$]} layouts={[root]} guards={[authGuard]} />

匹配时，路由节点会发出类型化参数；否则会发出 null。通过向路由节点发布参数来导航（usePublisher(user$)({ id: 42 })）。将路由参数传入查询参数，以便在导航时获取数据。Guard(pattern, fn) 会在匹配的导航上运行——其上下文提供 continue()、navigate()、redirect()；异步 guard 会在 Suspense 中渲染路由。布局按 URL 前缀匹配，并按具体程度嵌套；LayoutSlotPortal/LayoutSlot/LayoutSlotFill 可让页面填充布局区域。默认情况下，Router 会与浏览器历史记录同步。

持久化（reactive-engine-storage）

linkCellToStorage(theme$, { storageType: 'localStorage', key: 'app-theme', debounceMs: 300 })

storageType: 'localStorage' | 'sessionStorage' | 'cookie'（cookie 会添加 cookieOptions）。读取操作会延迟到引擎首次激活该 cell 时执行，并且每个引擎只执行一次；写入会进行防抖；localStorage 链接会跨标签页同步。引擎的 id 会为键添加命名空间（my-app:app-theme），因此多个实例可以独立持久化。兼容 SSR（没有 window 时不执行操作）。

在引擎上构建组件或库

遵循数据表架构（完整模式）：

• 模块作用域的节点图 + 每个组件实例一个引擎（通过 EngineProvider）
• 公共 API = 输入的操作流（setX$、resetX$），输出的状态单元（xState$，通常是 DerivedCell）；连线逻辑保持在内部
• 属性通过 initFn/updateFn 中的 pubIn 流入单元；回调属性通过 singletonSub 处理
• 将可选功能拆分为单独的模块（子路径导出），并将其连线逻辑附加到核心节点——由于初始化是惰性的，未使用的功能不会产生开销
• 通过 engineRef/engineId 和针对导出节点的远程钩子实现外部控制
• 通过双向订阅函数连接命令式子系统，并确保清理逻辑触及 engine.onDispose

常见错误

• 在订阅回调之外调用独立的 pub/getValue/pubIn——请使用 engine.* 方法或 React 钩子。
• 修改同一个对象并再次发布——distinct 基于引用相等性；不会有任何内容发出。请发布新的引用。
• 认为节点本身会保存状态——值保存在引擎中；同一个 Cell 在两个引擎中有两个值。
• 认为 Stream 订阅会在订阅时触发并提供当前值——流是无状态的；只有 Cell 有值。
• 未设置 distinct: false，就通过 map 将数据导入触发器——连续相同的映射值会被过滤。
• 通过更改参数来触发 Mutation——只有将参数发布到 mutate$ 时，变更才会运行。
• 在会随 React 渲染而重复运行的代码中使用普通的 sub——订阅会不断累积；请使用 singletonSub。
• 在图中创建环路——环路会挂起或循环；请通过操作流或 delayWithMicrotask 打破一个方向。

参考资料

各软件包的指南见 references/：

• references/core/ — 核心概念（节点类型、distinct、e 命名空间）、引擎与生命周期（激活、API 风格、子引擎、释放）、事务（环路、拓扑执行、与 RxJS 的差异）、运算符与组合器（完整参考）、基于引擎构建（来自数据表的架构模式）、传播诊断（记录、安全性、故障排查）、README
• references/react/ — React 集成（EngineProvider 生命周期、钩子、远程钩子、SSR）、README
• references/query/ — 查询与变更（结果状态、生命周期、连线模式）、README
• references/router/ — 路由（路径语法、导航、布局、守卫）、README
• references/storage/ — 存储链接（时序、命名空间、跨标签页同步）、README

@virtuoso.dev/reactive-engine-core 中导出符号的 JSDoc 内容详尽——如果不确定某个签名，请阅读随软件包提供的类型定义。