MetonaSqlark/tests/v080-subscribe.test.ts

/**
 * v0.8.0 回归 —— A9 发布订阅（本地写入必须触发订阅者）
 *
 * 修复前：`db.subscribe(table, fn)` 对**本地写入永不触发** —— 全库唯一调用 `emit`
 * 的地方在 BroadcastChannel 收到其它标签页消息的分支里。于是 README:232
 * 「订阅表变更」与 site/docs.html:667-679 的示例（`event.type: 'insert' | 'update' |
 * 'delete'`、`event.row`）全部不成立；唯一实际触发条件是 `multiTabSync: true`
 * 且收到 `external` 事件。
 *
 * 修复方式：用 `ChangeNotifierEngine` 装饰器把变更通知收敛到**引擎接口**这一个位置。
 * 三个写入入口（SQL 语句 / Table API / QueryBuilder）与事务内写入都必须经过引擎，
 * 因此拦一次即可全覆盖，避免在三条路径上各写一份变更描述逻辑（那正是本项目
 * 反复出现"同一语义多份实现 → 漂移"的模式）。
 *
 * 覆盖：四个引擎 × 三种写入入口 × 事件内容（type/table/row/key/count）+
 * 退订 + async 订阅者 + 跨标签页广播不形成回路。
 */
import { MetonaSqlark } from '../src/core';
import type { ChangeEvent } from '../src/engine/change-notifier';

const MODES: Array<[string, Record<string, unknown>]> = [
  ['memory', {}],
  ['disk', {}],
  ['hybrid', {}],
  ['aria', { diskEngine: 'memory' }],
];

describe('[v0.8.0] A9 发布订阅 —— SQL 写入路径', () => {
  test('INSERT/UPDATE/DELETE 逐行事件，携带行内容与主键', async () => {
    for (const [mode, extra] of MODES) {
      const db = await MetonaSqlark.create({ name: `v080-a9-${mode}`, mode, ...extra } as never);
      await db.defineTable('t', { id: { type: 'string', primaryKey: true }, v: { type: 'number' } });
      const events: ChangeEvent[] = [];
      const unsub = db.subscribe('t', (e) => { events.push(e); });

      await db.query("INSERT INTO t VALUES ('a',1),('b',2)");
      await db.query("UPDATE t SET v = 9 WHERE id = 'a'");
      await db.query("DELETE FROM t WHERE id = 'b'");

      expect(events.map((e) => e.type)).toEqual(['insert', 'insert', 'update', 'delete']);
      // 逐行事件带主键与行内容
      expect(events.filter((e) => e.type === 'insert').map((e) => e.key).sort()).toEqual(['a', 'b']);
      const update = events.find((e) => e.type === 'update')!;
      expect(update.row).toMatchObject({ id: 'a', v: 9 });
      expect(update.table).toBe('t');
      const del = events.find((e) => e.type === 'delete')!;
      expect(del.row).toMatchObject({ id: 'b', v: 2 });

      // 退订后不再收到
      unsub();
      await db.query("INSERT INTO t VALUES ('c',3)");
      expect(events).toHaveLength(4);

      await db.close();
    }
  });
});

describe('[v0.8.0] A9 发布订阅 —— Table API / QueryBuilder 路径', () => {
  test('insert / insertMany / update / delete / clear 全部产生事件', async () => {
    for (const [mode, extra] of MODES) {
      const db = await MetonaSqlark.create({ name: `v080-a9b-${mode}`, mode, ...extra } as never);
      await db.defineTable('t', { id: { type: 'string', primaryKey: true }, v: { type: 'number' } });
      const events: ChangeEvent[] = [];
      db.subscribe('t', (e) => { events.push(e); });

      await db.table('t').insert({ id: 'a', v: 1 });
      await db.table('t').insertMany([{ id: 'b', v: 2 }, { id: 'c', v: 3 }]);
      await db.table('t').update({ v: 9 }).where({ id: 'a' }).execute();
      await db.table('t').delete().where({ id: 'b' }).execute();
      await db.table('t').clear();

      expect(events.map((e) => e.type))
        .toEqual(['insert', 'insert', 'insert', 'update', 'delete', 'clear']);
      await db.close();
    }
  });
});

describe('[v0.8.0] A9 发布订阅 —— 语义护栏', () => {
  test('异步订阅者会被 await（写完即通知完）', async () => {
    const db = await MetonaSqlark.create({ name: 'v080-a9-async', mode: 'memory' });
    await db.defineTable('t', { id: { type: 'string', primaryKey: true } });
    const seen: string[] = [];
    db.subscribe('t', async (e) => {
      await new Promise((r) => setTimeout(r, 1));
      seen.push(`${e.type}:${e.key ?? ''}`);
    });
    await db.query("INSERT INTO t VALUES ('x')");
    expect(seen).toEqual(['insert:x']);
    await db.close();
  });

  test('订阅者抛错不影响写入结果（写入已成功）', async () => {
    const errors: Error[] = [];
    const db = await MetonaSqlark.create({
      name: 'v080-a9-throw',
      mode: 'memory',
      onError: (e: Error) => { errors.push(e); },
    } as never);
    await db.defineTable('t', { id: { type: 'string', primaryKey: true } });
    db.subscribe('t', () => { throw new Error('subscriber boom'); });

    await expect(db.query("INSERT INTO t VALUES ('x')")).resolves.toEqual(['x']);
    const rows = await db.query('SELECT COUNT(*) AS c FROM t') as Array<{ c: number }>;
    expect(rows[0].c).toBe(1);
    expect(errors.some((e) => e.message === 'subscriber boom')).toBe(true);
    await db.close();
  });

  test('手动 emit 仍可用，并补全 table 字段', async () => {
    const db = await MetonaSqlark.create({ name: 'v080-a9-emit', mode: 'memory' });
    const events: ChangeEvent[] = [];
    db.subscribe('t', (e) => { events.push(e); });
    await db.emit('t', { type: 'insert', row: { id: 'z' } });
    expect(events).toEqual([{ table: 't', type: 'insert', row: { id: 'z' } }]);
    await db.close();
  });
});