core/tests/engine/sharded-delta-overlay.test.js

import { describe, test } from 'node:test';
import assert from 'node:assert/strict';
import fs from 'node:fs';
import os from 'node:os';
import path from 'node:path';
import { CondensedGraph } from '../../src/core/CondensedGraph.js';
import { ShardedSnapshotBuilder } from '../../src/core/shards/ShardedSnapshotBuilder.js';
import { ShardedSnapshot } from '../../src/core/shards/ShardedSnapshot.js';
import { FileShardStorage } from '../../src/core/shards/FileShardStorage.js';
import { DeltaShardBinary } from '../../src/core/shards/DeltaShardBinary.js';

function buildSnapshot(bucketSize = 4) {
  const graph = new CondensedGraph();

  const user0 = graph._ensureNode('user:0');
  const user1 = graph._ensureNode('user:1');
  const doc0 = graph._ensureNode('doc:0');
  const doc1 = graph._ensureNode('doc:1');

  graph.addEdge(user0, 'owner', doc0);
  graph.addEdge(user1, 'owner', doc0);

  graph.finalizePerfectHash();
  graph.finalizeWaveletAdjacency({ dropAdjacencyList: true });

  const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'sharded-delta-'));
  const builder = new ShardedSnapshotBuilder({ bucketSize, includeDirections: ['out', 'in'] });
  const manifest = builder.build(graph, dir);
  const storage = new FileShardStorage(dir);
  const snapshot = new ShardedSnapshot(manifest, storage, { cacheLimit: 8 });
  snapshot.initializeSync();

  return { graph, snapshot, dir, ids: { user0, user1, doc0, doc1 } };
}

function writeDeltaLayer(snapshot, dir, name, entries) {
  fs.mkdirSync(dir, { recursive: true });
  const merged = new Map();

  for (const entry of entries) {
    const shardMeta = snapshot._selectShardMeta(entry.relationId, entry.direction, entry.srcId);
    assert.ok(shardMeta, 'Missing shard meta for delta entry');
    const localSource = snapshot._localSource(entry.srcId, shardMeta);
    const key = shardMeta.cacheKey;
    let bucket = merged.get(key);
    if (!bucket) {
      bucket = { shardMeta, additions: [], removals: [] };
      merged.set(key, bucket);
    }
    for (const add of entry.additions) {
      bucket.additions.push({
        srcLocal: localSource,
        otherId: add.dstId,
        possBits: add.possBits,
        relBits: add.relBits
      });
    }
    for (const rem of entry.removals) {
      bucket.removals.push({
        srcLocal: localSource,
        otherId: rem.dstId
      });
    }
  }

  const shards = [];
  for (const bucket of merged.values()) {
    const shardMeta = bucket.shardMeta;
    const buffer = DeltaShardBinary.serialize({
      relationId: shardMeta.relationId,
      direction: shardMeta.direction,
      rangeStart: shardMeta.rangeStart,
      rangeEnd: shardMeta.rangeEnd,
      nodeCount: snapshot.nodeCount,
      additions: bucket.additions,
      removals: bucket.removals
    });

    const shardKey = `delta-${name}-${shardMeta.key}`;
    fs.writeFileSync(path.join(dir, shardKey), new Uint8Array(buffer));
    shards.push({
      key: shardKey,
      relationId: shardMeta.relationId,
      direction: shardMeta.direction,
      rangeStart: shardMeta.rangeStart,
      rangeEnd: shardMeta.rangeEnd,
      cacheKey: shardMeta.cacheKey
    });
  }

  return { shards };
}

// ADR-003: sharded snapshots are stubs in src/core/shards/ — tests are the spec for when the subsystem is implemented.
describe.skip('Sharded snapshot delta overlay', () => {
  test('adds and removes edges in overlay reads', () => {
    const { graph, snapshot, dir, ids } = buildSnapshot(4);
    const relId = graph.getRelationId('owner');

    const baseEdge = snapshot.findEdgeSync(ids.user0, relId, ids.doc0);
    assert.ok(baseEdge, 'Expected base edge');

    const deltaDir = path.join(dir, 'delta');
    const deltaManifest = writeDeltaLayer(snapshot, deltaDir, 'l1', [
      {
        relationId: relId,
        direction: 'out',
        srcId: ids.user0,
        additions: [],
        removals: [{ dstId: ids.doc0 }]
      },
      {
        relationId: relId,
        direction: 'out',
        srcId: ids.user1,
        additions: [{ dstId: ids.doc1, possBits: 65535, relBits: 65535 }],
        removals: []
      }
    ]);

    snapshot.setDeltaLayers([{ shards: deltaManifest.shards, storage: new FileShardStorage(deltaDir) }]);

    const removedEdge = snapshot.findEdgeSync(ids.user0, relId, ids.doc0);
    assert.equal(removedEdge, null);

    const addedEdge = snapshot.findEdgeSync(ids.user1, relId, ids.doc1);
    assert.ok(addedEdge, 'Expected added edge');

    const user0Edges = snapshot.getOutEdgesSync(ids.user0, relId);
    assert.equal(user0Edges.length, 0);

    const user1Edges = snapshot.getOutEdgesSync(ids.user1, relId);
    assert.equal(user1Edges.length, 2);

    fs.rmSync(dir, { recursive: true, force: true });
  });

  test('later delta layers override earlier ones', () => {
    const { graph, snapshot, dir, ids } = buildSnapshot(4);
    const relId = graph.getRelationId('owner');

    const layer1Dir = path.join(dir, 'delta-1');
    const layer2Dir = path.join(dir, 'delta-2');

    const layer1 = writeDeltaLayer(snapshot, layer1Dir, 'l1', [
      {
        relationId: relId,
        direction: 'out',
        srcId: ids.user0,
        additions: [{ dstId: ids.doc1, possBits: 65535, relBits: 65535 }],
        removals: []
      }
    ]);

    const layer2 = writeDeltaLayer(snapshot, layer2Dir, 'l2', [
      {
        relationId: relId,
        direction: 'out',
        srcId: ids.user0,
        additions: [],
        removals: [{ dstId: ids.doc1 }]
      }
    ]);

    snapshot.setDeltaLayers([
      { shards: layer1.shards, storage: new FileShardStorage(layer1Dir) },
      { shards: layer2.shards, storage: new FileShardStorage(layer2Dir) }
    ]);

    const edge = snapshot.findEdgeSync(ids.user0, relId, ids.doc1);
    assert.equal(edge, null);

    fs.rmSync(dir, { recursive: true, force: true });
  });
});