fix(A9/A10): 发布订阅接线 + 列对列比较与关联子查询（静默空结果根治）

A9 db.subscribe 对本地写入永不触发全库唯一调用 emit 的地方在 BroadcastChannel 收到**其它标签页**消息的分支里，于是 README:232「订阅表变更」与 site/docs.html:667-679 的示例（event.type: 'insert'|'update'|'delete'、event.row）全部不成立。根治方式：新增 src/engine/change-notifier.ts —— IStorageEngine 装饰器，把变更通知收敛到**引擎接口**这一个位置（三个写入入口 SQL/Table/Builder 与事务内写入都必须经过它），避免在三条路径上各写一份变更描述逻辑。事件语义（兑现文档承诺）：INSERT 逐行带 row+key；UPDATE/DELETE 写入前快照受影响行、成功后逐行发事件并带更新后/删除前的行；CLEAR/DDL 表级事件。订阅者返回 Promise 时被 await；订阅者抛错不影响写入结果（只上报 onError）。两个实现细节值得记录： 1. 引擎被装饰后，core 里 `this.engine instanceof HybridEngine` 恒为 false → Hybrid 跨标签页重载静默失效。新增 unwrapEngine() 对**内层**引擎做能力探测。 2. 外部事件（external）绝不能重新广播 —— 否则 A↔B 互相转发形成无限循环（实测 8 次以上且不终止）。已分离 emitExternal 路径。 A10 列对列比较与关联 IN 子查询静默空结果 1) `WHERE t.x = t.y`（唯一可解析的列对列写法）返回 []： - 引擎层 matchWhere 无 $col 上下文，把 `{ $col: ... }` 当普通对象比较； - executor.filterCorrelated 调用 matchWhere 时**没传** `{ $col: true }`。修复：engine 层遇到未解析操作数（$col/$subquery）时**放行**而非判假 —— 引擎的过滤只允许缩小候选集，最终判定始终由带上下文的 executor 完成； executor 侧补上 `{ $col: true }`。同时修正 MemoryEngine/AriaEngine 的索引下推：非原始值（对象）不走索引，否则 String({...}) 得到无意义键、查找为空并短路全表扫描 → 静默空结果。 2) `WHERE id IN (SELECT user_id FROM o WHERE o.user_id = u.id)` 返回 []：子查询执行**不传外层行上下文**，`u.id` 绑定为 null → 子查询空集 → `$in: []`。（结构相同的 EXISTS 走另一条分支、结果正确 —— 又一处"同一语义两条路径"。）修复：resolveOperatorSubqueries 接收并传递 contextRow；bindColumnRefs 递归进入 $subquery 绑定外层引用；新增 lookupOuterValue 先剥外层表名/别名前缀再取值（外层行键不带前缀，否则 `u.id` 取 undefined 被 `?? null` 静默成 null）。 ChangeNotifierEngine 能力转发装饰器只实现 IStorageEngine 声明的成员，导致： - 可选能力缺失时抛原生 Error，破坏 `NOT_SUPPORTED` 错误码契约（14 个用例失败） → 新增 requireCapability，统一抛 NOT_SUPPORTED 并保留方法名； - 接口外方法（analyzeTable/reindexTable/vacuum）在被包装后静默消失 → 新增 requireOptionalMethod 显式转发（ANALYZE/REINDEX/VACUUM 恢复可用）。新增 tests/v080-subscribe.test.ts（5 用例，四引擎 × 三种入口）、 tests/v080-correlated.test.ts（4 用例，含"关联 IN 与等价 EXISTS 结果一致"护栏）。
2026-09-14 22:29:51 +08:00
parent c0d85eeaab
commit 674da6b7b7
10 changed files with 804 additions and 48 deletions
@@ -6,6 +6,8 @@
 */

 import type { IStorageEngine } from './engine/interface';
+import { ChangeNotifierEngine } from './engine/change-notifier';
+import type { ChangeEvent } from './engine/change-notifier';
 import type { DatabaseConfig, ColumnDef } from './constants';
 import { DB_DEFAULTS, DatabaseError } from './constants';
 import { MemoryEngine } from './engine/memory';
@@ -81,10 +83,17 @@ export class MetonaSqlark {
      this.channel.onmessage = (event) => {
        const msg = event.data as { type?: string; table?: string } | null;
        if (!msg || msg.type !== 'change') return;
-        this.emit(msg.table ?? '', { type: 'external', table: msg.table ?? '' });
-        // Hybrid 引擎：从磁盘重载内存，保证读到其他标签页的最新数据
-        if (this.engine instanceof HybridEngine) {
-          (this.engine as HybridEngine).reloadMemoryFromDisk().catch(() => {
+        // v0.8.0: 外部事件只派发给本地订阅者，**不得再次广播** ——
+        // 否则两个标签页会互相转发形成无限广播循环（实测 8 次以上且不终止）。
+        void this.emitExternal(msg.table ?? '');
+        // Hybrid 引擎：从磁盘重载内存，保证读到其他标签页的最新数据。
+        //
+        // v0.8.0: 引擎现在被 ChangeNotifierEngine 装饰，`this.engine instanceof HybridEngine`
+        // 恒为 false —— 因此改为对**内层**引擎做能力探测。这也是审计指出的
+        // "用 instanceof 做引擎特判"的隐患：装饰器一加就静默失效。
+        const inner = this.unwrapEngine();
+        if (inner instanceof HybridEngine) {
+          (inner as HybridEngine).reloadMemoryFromDisk().catch(() => {
            // 重载失败不影响主流程（下次读可能短暂过期）
          });
        }
@@ -102,6 +111,28 @@ export class MetonaSqlark {
    // 打开连接
    await this.engine.open(this.name, this.version);

+    // v0.8.0（A9）：把引擎包进变更通知装饰器 —— **唯一**的变更事件汇聚点。
+    // 三个写入入口（SQL / Table API / QueryBuilder）与事务内写入都必须经过引擎接口，
+    // 因此在这里拦一次即可全覆盖，避免在三条路径上各写一份"变更描述"逻辑。
+    this.engine = new ChangeNotifierEngine(
+      this.engine,
+      (error) => this._onError(error),
+      (table) => this.broadcastChange(table),
+    );
+    this.notifier = this.engine as ChangeNotifierEngine;
+    // 把引擎层变更事件接入 db.subscribe 的订阅表（listeners）
+    this.notifier.addListener(async (event) => {
+      const set = this.listeners.get(`change:${event.table}`);
+      if (!set || set.size === 0) return;
+      for (const cb of [...set]) {
+        try {
+          await (cb as (e: ChangeEvent) => void | Promise<void>)(event);
+        } catch (error) {
+          this._onError(error as Error);
+        }
+      }
+    });
+
    // v0.4.2-fix (P2-7): 从库内加载持久化的迁移版本，
    // 重启后 migrateTo 从持久化版本继续执行，不再每次从 config.version 重置
    if (typeof this.engine.getMeta === 'function') {
@@ -169,7 +200,9 @@ export class MetonaSqlark {
        this.engine,
        name,
        this.executor,
-        (tableName) => this.broadcastChange(tableName),
+        // v0.8.0: 变更事件由引擎层 ChangeNotifierEngine 统一产生，
+        // 此处不再重复广播（Table API 与 SQL 路径曾各广播一次 → 同一次写入触发两遍）
+        () => { /* no-op: see ChangeNotifierEngine */ },
        (hook, args) => this.pluginManager.trigger(hook, ...args),
      );
      this.tableCache.set(name, t);
@@ -221,9 +254,8 @@ export class MetonaSqlark {
        await this.triggerStatementHooks(stmt, 'before');
        result = await this.executor.execute(stmt);
        await this.triggerStatementHooks(stmt, 'after', result);
-        // v0.3.2: 写语句广播表变更（多标签页同步）
-        const table = this.writeStatementTable(stmt);
-        if (table) this.broadcastChange(table);
+        // v0.8.0: 变更广播与订阅事件统一由引擎层 ChangeNotifierEngine 产生，
+        // 此处不再手工 broadcastChange（否则同一次写入会广播两次）。
      }
    } catch (error) {
      this._onError(error as Error);
@@ -470,20 +502,65 @@ export class MetonaSqlark {

  // ---- 发布订阅 ----

+  /** 变更通知引擎（init 后可用）；未初始化时为 null */
+  private notifier: ChangeNotifierEngine | null = null;
+
  private listeners: Map<string, Set<(data: unknown) => void>> = new Map();

-  /** 订阅表变更 */
-  subscribe(tableName: string, callback: (event: { type: string; row?: unknown; table?: string }) => void): () => void {
+  /**
+   * 订阅表变更。
+   *
+   * v0.8.0 修复：此前**本地写入永不触发** —— 全库唯一调用 `emit` 的地方在
+   * BroadcastChannel 收到其它标签页消息的分支里，因此 README「订阅表变更」与
+   * site/docs.html 的 `event.type: 'insert' | 'update' | 'delete'` 示例全都不成立。
+   * 现在本地写入（SQL / Table API / QueryBuilder / 事务内）都会产生事件。
+   *
+   * 现在返回的函数是**同步**退订函数（与既有 API 兼容）。
+   */
+  subscribe(
+    tableName: string,
+    callback: (event: ChangeEvent) => void | Promise<void>,
+  ): () => void {
    const key = `change:${tableName}`;
    if (!this.listeners.has(key)) this.listeners.set(key, new Set());
    this.listeners.get(key)!.add(callback as (data: unknown) => void);
-    return () => this.listeners.get(key)?.delete(callback as (data: unknown) => void);
+    return () => { this.listeners.get(key)?.delete(callback as (data: unknown) => void); };
  }

-  /** 触发变更事件 */
-  emit(tableName: string, event: { type: string; row?: unknown; table?: string }): void {
-    const key = `change:${tableName}`;
-    this.listeners.get(key)?.forEach((cb) => cb(event));
+  /**
+   * 手动触发变更事件（保留为公开 API：自定义写入路径可显式通知订阅者）。
+   * 现在也支持 await —— 订阅者的 Promise 会被等待。
+   */
+  async emit(tableName: string, event: Partial<ChangeEvent> & { type: ChangeEvent['type'] }): Promise<void> {
+    await this.dispatchChange({ table: tableName, ...event });
+  }
+
+  /**
+   * v0.8.0: 派发"来自其它标签页"的变更事件。
+   * 只走本地订阅者，不触发 onBroadcast（避免 A↔B 互相转发的无限循环）。
+   */
+  private async emitExternal(tableName: string): Promise<void> {
+    const event: ChangeEvent = { type: 'external', table: tableName };
+    const set = this.listeners.get(`change:${tableName}`);
+    if (!set) return;
+    for (const cb of [...set]) {
+      try {
+        await (cb as (e: ChangeEvent) => void | Promise<void>)(event);
+      } catch (error) {
+        this._onError(error as Error);
+      }
+    }
+  }
+
+  /** 内部：把一次变更同时派发给本地订阅者与跨标签页广播 */
+  private async dispatchChange(event: ChangeEvent): Promise<void> {
+    if (this.notifier) {
+      // notifier.dispatch 内部已包含跨标签页广播，这里不重复调用
+      await this.notifier.dispatch(event);
+      return;
+    }
+    // init 之前（notifier 尚未建立）也能广播
+    this.broadcastChange(event.table);
  }

  // ---- 多标签页同步（v0.3.2） ----
@@ -654,11 +731,21 @@ export class MetonaSqlark {
    this.ready = false;
  }

-  /** 获取底层引擎 */
+  /** 获取底层引擎（含变更通知装饰器） */
  getEngine(): IStorageEngine {
    return this.engine;
  }

+  /**
+   * v0.8.0: 取**未装饰**的真实存储引擎。
+   *
+   * 引擎在 init 时被 ChangeNotifierEngine 包了一层，因此需要引擎特化能力
+   * （如 HybridEngine.reloadMemoryFromDisk）时必须先解包，否则 instanceof 恒 false。
+   */
+  private unwrapEngine(): IStorageEngine {
+    return this.notifier ? this.notifier.getInner() : this.engine;
+  }
+
  // ---- 内部 ----

  private createEngine(): IStorageEngine {