MetonaSqlark/tests/engine/aria-page-store.test.ts

/**
 * AriaEngine — 页面化 SSTable 存储测试
 *
 * 覆盖：
 * 1. PageSSTableStore 单元：save/load 往返（页边界、非整数页）、delete 释放页面
 * 2. AriaEngine + OPFS（页面模式）：写入 → flush → close → reopen 数据完整
 * 3. 大 SSTable 多页 + pageIds 注入 meta 验证
 * 4. 页面损坏（CRC 校验失败）→ 打开自愈清理
 * 5. 页面模式与非页面模式混合（pageIds 兼容：新页面写 + 旧整 value 读）
 * 6. 页面化 + 加密 + 压缩组合
 */
import { AriaEngine } from '../../src/engine/aria/index';
import { PageSSTableStore } from '../../src/engine/aria/store/page_sstable_store';
import { FileManager } from '../../src/engine/aria/store/file_manager';
import { BufferPool } from '../../src/engine/aria/buffer/pool';
import { MemoryBackend } from '../../src/engine/aria/store/backend';
import { createSchema } from '../../src/table/schema';
import { PAGE_SIZE } from '../../src/engine/aria/types';

// ===================================================================
//  OPFS mock（共享工具）
// ===================================================================
import { resetOPFSMock } from '../helpers/storage-harness';
import { decode as decodeBytes } from '../helpers/assertions';

const SCHEMA = () => createSchema('users', {
  id: { type: 'string', primaryKey: true },
  name: { type: 'string' },
  note: { type: 'string' },
});

// ===================================================================
//  PageSSTableStore 单元
// ===================================================================
describe('AriaEngine — PageSSTableStore 单元', () => {
  it('save/load 往返：单页数据', async () => {
    const backend = new MemoryBackend();
    await backend.open('pg-unit-1');
    const fm = new FileManager(backend);
    await fm.init('pg-unit-1');
    const bp = new BufferPool(fm, 64);
    const store = new PageSSTableStore(fm, bp);

    const data = new Uint8Array(new TextEncoder().encode('hello page store'));
    await store.save(1, data);
    const pageIds = store.getPageIds(1)!;
    expect(pageIds.length).toBe(1);

    const back = await store.load(1, pageIds, data.byteLength);
    expect(back).not.toBeNull();
    expect(decodeBytes(back)).toBe('hello page store');

    // 页面已落盘（backend 有 pg_ 键）
    expect(await backend.exists('pg_1')).toBe(true);

    await store.delete(1, pageIds);
    expect(await backend.exists('pg_1')).toBe(false);
    await backend.close();
  });

  it('save/load 往返：多页数据（跨 4KB 边界）', async () => {
    const backend = new MemoryBackend();
    await backend.open('pg-unit-2');
    const fm = new FileManager(backend);
    await fm.init('pg-unit-2');
    const bp = new BufferPool(fm, 64);
    const store = new PageSSTableStore(fm, bp);

    // 3 页：4KB * 2 + 1KB
    const size = PAGE_SIZE * 2 + 1024;
    const data = new Uint8Array(size);
    let x = 42;
    for (let i = 0; i < size; i++) {
      x = (x * 31 + 7) & 0xffff;
      data[i] = x & 0xff;
    }
    await store.save(7, data);
    const pageIds = store.getPageIds(7)!;
    expect(pageIds.length).toBe(3);

    const back = await store.load(7, pageIds, size);
    expect(back).not.toBeNull();
    expect(back!.byteLength).toBe(size);
    expect(back).toEqual(data);
    await backend.close();
  });

  it('load 使用 totalSize 截断最后一页的 0 填充', async () => {
    const backend = new MemoryBackend();
    await backend.open('pg-unit-3');
    const fm = new FileManager(backend);
    await fm.init('pg-unit-3');
    const bp = new BufferPool(fm, 64);
    const store = new PageSSTableStore(fm, bp);

    // 非整数页：4096 + 100 字节
    const real = new Uint8Array(PAGE_SIZE + 100);
    for (let i = 0; i < real.length; i++) real[i] = i & 0xff;
    await store.save(9, real);
    const pageIds = store.getPageIds(9)!;
    expect(pageIds.length).toBe(2);

    const back = await store.load(9, pageIds, real.length);
    expect(back!.byteLength).toBe(real.length);
    expect(back).toEqual(real);
    await backend.close();
  });

  it('页面缺失（磁盘文件丢失 + 缓存清空）→ load 返回 null', async () => {
    const backend = new MemoryBackend();
    await backend.open('pg-unit-4');
    const fm = new FileManager(backend);
    await fm.init('pg-unit-4');
    const bp = new BufferPool(fm, 64);
    const store = new PageSSTableStore(fm, bp);

    await store.save(5, new Uint8Array(100));
    const pageIds = store.getPageIds(5)!;
    // 模拟崩溃重开：磁盘文件丢失 + 页面缓存清空
    await backend.delete('pg_' + pageIds[0]);
    bp.removePage(pageIds[0]);

    const back = await store.load(5, pageIds, 100);
    expect(back).toBeNull();
    await backend.close();
  });
});

// ===================================================================
//  AriaEngine + OPFS 页面模式集成
// ===================================================================
describe('AriaEngine — 页面化 SSTable 集成（OPFS）', () => {
  it('写入 → flush → close → reopen 数据完整（页面模式默认启用）', async () => {
    resetOPFSMock();
    const engine = new AriaEngine({
      storageBackend: 'opfs',
      memtableSizeThreshold: 64 * 1024 * 1024,
      checkpointInterval: 100000,
    });
    await engine.open('page-db-1', 1);
    await engine.createTable(SCHEMA());

    // 多行 + 大 note → 超过 4KB → 多页
    const rows = [] as Record<string, unknown>[];
    for (let i = 0; i < 30; i++) {
      rows.push({ id: `u-${i}`, name: `User${i}`, note: 'page-store-note-'.repeat(40) });
    }
    await engine.insert('users', rows);
    await (engine as any).lsm.flush();

    // 页面化落盘：backend 有 pg_ 键，meta 含 pageIds
    const backend = (engine as any).backend;
    const keys = await backend.listKeys();
    const pgKeys = keys.filter((k: string) => k.startsWith('pg_'));
    const sstKeys = keys.filter((k: string) => k.startsWith('sst_'));
    expect(pgKeys.length).toBeGreaterThan(0);
    expect(sstKeys.length).toBe(0); // 页面模式下不写整 value SSTable

    const metas = await (engine as any).lsm.sstableStore.listMeta();
    expect(metas.length).toBeGreaterThan(0);
    expect(metas[0].pageIds).toBeDefined();
    expect(metas[0].pageIds.length).toBeGreaterThan(0);

    await engine.close();

    // reopen：数据完整
    const engine2 = new AriaEngine({ storageBackend: 'opfs', checkpointInterval: 100000 });
    await engine2.open('page-db-1', 1);
    const found = await engine2.find('users', { table: 'users' });
    expect(found).toHaveLength(30);
    expect(found[0].note).toBe('page-store-note-'.repeat(40));
    expect(await engine2.count('users')).toBe(30);
    await engine2.close();
  });

  it('多级 compaction 后页面化数据仍完整', async () => {
    resetOPFSMock();
    const engine = new AriaEngine({
      storageBackend: 'opfs',
      memtableSizeThreshold: 16 * 1024,
      checkpointInterval: 100000,
    });
    await engine.open('page-db-2', 1);
    await engine.createTable(SCHEMA());

    // 大量小批量写入 → 多次 flush → 触发 compaction
    for (let batch = 0; batch < 8; batch++) {
      const rows = [] as Record<string, unknown>[];
      for (let i = 0; i < 20; i++) {
        rows.push({ id: `b${batch}-u${i}`, name: `U${i}`, note: 'x'.repeat(200) });
      }
      await engine.insert('users', rows);
      await (engine as any).lsm.flush();
    }
    // 强制压缩
    await (engine as any).lsm.compactLevel(0);

    await engine.close();
    const engine2 = new AriaEngine({ storageBackend: 'opfs', checkpointInterval: 100000 });
    await engine2.open('page-db-2', 1);
    expect(await engine2.count('users')).toBe(160);
    const one = await engine2.find('users', { table: 'users', where: { id: 'b3-u10' } });
    expect(one).toHaveLength(1);
    await engine2.close();
  });

  it('页面损坏（篡改 pg_ 文件）→ 打开自愈清理，其余数据可读', async () => {
    resetOPFSMock();
    const engine = new AriaEngine({
      storageBackend: 'opfs',
      memtableSizeThreshold: 64 * 1024 * 1024,
      checkpointInterval: 100000,
    });
    await engine.open('page-db-3', 1);
    await engine.createTable(SCHEMA());
    for (let batch = 0; batch < 2; batch++) {
      const rows = [] as Record<string, unknown>[];
      for (let i = 0; i < 20; i++) {
        rows.push({ id: `b${batch}-u${i}`, name: `U${i}`, note: 'n'.repeat(100) });
      }
      await engine.insert('users', rows);
      await (engine as any).lsm.flush();
    }
    await engine.close();

    // 篡改第一个 SSTable 的第一页
    const engine2 = new AriaEngine({ storageBackend: 'opfs', checkpointInterval: 100000 });
    await engine2.open('page-db-3', 1);
    const metas = await (engine2 as any).lsm.sstableStore.listMeta();
    expect(metas.length).toBeGreaterThanOrEqual(2);
    const victim = metas[0];
    const backend = (engine2 as any).backend;
    const pageRaw = await backend.read(`pg_${victim.pageIds[0]}`);
    const corrupted = new Uint8Array(pageRaw);
    corrupted[100] ^= 0xff;
    await backend.write(`pg_${victim.pageIds[0]}`, corrupted.buffer as ArrayBuffer);
    await engine2.close();

    // 重开：损坏页面的 SSTable 被清理，其余数据可读
    const engine3 = new AriaEngine({ storageBackend: 'opfs', checkpointInterval: 100000 });
    await engine3.open('page-db-3', 1);
    const rows3 = await engine3.find('users', { table: 'users' });
    expect(rows3.length).toBeGreaterThan(0);
    expect(rows3.length).toBeLessThan(40);
    await engine3.close();
  });

  it('页面模式与非页面模式混合兼容（pageIds 缺失 → 整 value 读取）', async () => {
    const opfs = resetOPFSMock();
    // 阶段 1：非页面模式写入（pageStorage: false → 整 value SSTable）
    const engine = new AriaEngine({
      storageBackend: 'opfs',
      pageStorage: false,
      memtableSizeThreshold: 64 * 1024 * 1024,
      checkpointInterval: 100000,
    });
    await engine.open('page-db-4', 1);
    await engine.createTable(SCHEMA());
    await engine.insert('users', [{ id: 'legacy-1', name: 'Old', note: 'whole-value' }]);
    await (engine as any).lsm.flush();
    await engine.close();

    const backendKeys1 = await opfs.listKeys();
    expect(backendKeys1.some((k) => k.startsWith('sst_'))).toBe(true);

    // 阶段 2：页面模式打开（默认 opfs → 启用），读旧数据 + 写新数据
    const engine2 = new AriaEngine({ storageBackend: 'opfs', checkpointInterval: 100000 });
    await engine2.open('page-db-4', 1);
    expect(await engine2.count('users')).toBe(1);
    const legacy = await engine2.find('users', { table: 'users', where: { id: 'legacy-1' } });
    expect(legacy[0].note).toBe('whole-value');
    // 新写入 → 页面化
    await engine2.insert('users', [{ id: 'new-1', name: 'New', note: 'paged' }]);
    await (engine2 as any).lsm.flush();
    await engine2.close();

    // 阶段 3：混合库重开，新旧数据都在
    const engine3 = new AriaEngine({ storageBackend: 'opfs', checkpointInterval: 100000 });
    await engine3.open('page-db-4', 1);
    expect(await engine3.count('users')).toBe(2);
    const both = await engine3.find('users', { table: 'users' });
    const byId = Object.fromEntries(both.map((r) => [r.id, r]));
    expect(byId['legacy-1'].note).toBe('whole-value');
    expect(byId['new-1'].note).toBe('paged');
    await engine3.close();
  });

  it('页面化 + 加密 + 压缩组合：往返完整', async () => {
    resetOPFSMock();
    const engine = new AriaEngine({
      storageBackend: 'opfs',
      compression: true,
      encryption: { password: 'page-pass' },
      memtableSizeThreshold: 64 * 1024 * 1024,
      checkpointInterval: 100000,
    });
    await engine.open('page-db-5', 1);
    await engine.createTable(SCHEMA());
    const rows = [] as Record<string, unknown>[];
    for (let i = 0; i < 20; i++) {
      rows.push({ id: `e-${i}`, name: `E${i}`, note: '加密的页面化内容'.repeat(30) });
    }
    await engine.insert('users', rows);
    await (engine as any).lsm.flush();
    await engine.close();

    const engine2 = new AriaEngine({
      storageBackend: 'opfs',
      compression: true,
      encryption: { password: 'page-pass' },
      checkpointInterval: 100000,
    });
    await engine2.open('page-db-5', 1);
    expect(await engine2.count('users')).toBe(20);
    const one = await engine2.find('users', { table: 'users', where: { id: 'e-7' } });
    expect(one[0].note).toBe('加密的页面化内容'.repeat(30));
    await engine2.close();
  });
});