import assert from "node:assert/strict";
import { harness, Client, options } from "./server-harness.mjs";
import {
  getViewBounds,
  mergeChunkUpdate,
  mergeSnapshot,
  positionInView,
} from "../client/world-view.js";

// Isolated data and port: this check never edits the running demo world.
const opts = { port: "4012", ...options() },
  h = await harness(opts),
  clients = [],
  checks = [],
  transitions = [],
  world = "test_chunk_stream";
const key = (pos) => pos.join(","),
  sorted = (map) => [...map].sort(([a], [b]) => a.localeCompare(b)),
  view = {
    world,
    revision: 0,
    viewCenter: null,
    viewBounds: null,
    blocks: new Map(),
  };
let receiveError,
  explicitResync = false,
  snapshots = 0;

function observe(message) {
  if (receiveError) return;
  try {
    if (message.type === "error") throw Error(message.message);
    if (["welcome", "snapshot"].includes(message.type)) {
      if (message.type === "snapshot") {
        assert.ok(explicitResync, "Walking must not send a full snapshot");
        snapshots++;
      }
      assert.equal(message.world, world);
      assert.ok(message.features.includes("chunk_stream_v1"));
      view.revision = message.revision;
      view.viewCenter = message.view_center;
      view.viewBounds = getViewBounds(message);
      mergeSnapshot(view.blocks, message.blocks);
    } else if (message.type === "blocks") {
      assert.equal(message.revision, view.revision + 1);
      for (const change of message.changes) {
        if (!positionInView(change.pos, view.viewBounds)) continue;
        if (change.block) view.blocks.set(key(change.pos), change.block);
        else view.blocks.delete(key(change.pos));
      }
      view.revision = message.revision;
    } else if (message.type === "chunks") {
      assert.ok(!("spawn" in message), "Movement must not reset the camera");
      assert.ok(!("blocks" in message), "Movement must send section deltas");
      assert.equal(message.sections.length, 12);
      assert.equal(message.unload.length, 12);
      assert.equal(
        Math.abs(message.view_center[0] - message.from_center[0]),
        16,
      );
      assert.deepEqual(message.view_center.slice(1), [0, 0]);
      assert.ok(message.sections.some((section) => !section.blocks.length));
      const enteringIds = new Set(
          message.sections.flatMap((section) =>
            section.blocks.map((record) => record.block),
          ),
        ),
        materialIds = new Set(message.materials.map((material) => material.id));
      assert.deepEqual(materialIds, enteringIds, "Only entering materials sent");
      const departing = new Set(message.unload.map(key)),
        retained = [...view.blocks].filter(
          ([id]) =>
            !departing.has(
              key(id.split(",").map((value) => Math.floor(Number(value) / 16))),
            ),
        ),
        result = mergeChunkUpdate(view, message);
      assert.equal(result.status, "applied", "Real client accepts server delta");
      for (const [id, block] of retained)
        assert.equal(view.blocks.get(id), block, "Retained geometry is unchanged");
      transitions.push({
        from: message.from_center,
        to: message.view_center,
        entering_sections: message.sections.length,
        departing_sections: message.unload.length,
        retained_sections: 48 - message.unload.length,
        entering_blocks: message.sections.reduce(
          (count, section) => count + section.blocks.length,
          0,
        ),
        retained_blocks: retained.length,
        bytes: Buffer.byteLength(JSON.stringify(message)),
      });
    }
  } catch (error) {
    receiveError = error;
  }
}

function assertReceived() {
  if (receiveError) throw receiveError;
}

async function assertCurrentRegion() {
  assertReceived();
  const region = await h.control("world.read", {
    world,
    ...view.viewBounds,
  });
  assert.equal(view.revision, region.revision);
  assert.deepEqual(
    sorted(view.blocks),
    sorted(new Map(region.blocks.map((record) => [key(record.pos), record.block]))),
    "Merged client view must exactly match the authoritative region",
  );
}

async function edit(client, changes, operation) {
  const result = await h.control("world.edit", {
    world,
    expected_revision: view.revision,
    operation_id: operation,
    changes,
  });
  await client.next("blocks", (message) => message.revision === result.revision);
  await assertCurrentRegion();
}

async function walkTo(client, x, center, expectType = "chunks") {
  const initial = (await client.state()).players.find(
      (player) => player.id === client.id,
    ),
    strafe = x > initial.position[0] ? 1 : -1,
    seq = client.input({ strafe }),
    pulse = setInterval(() => client.input({ strafe }), 100);
  try {
    await client.state(
      (player, message) =>
        message.ack >= seq &&
        (strafe > 0 ? player.position[0] >= x : player.position[0] <= x),
      10000,
    );
  } finally {
    clearInterval(pulse);
    const stopped = client.input();
    await client.state((_player, message) => message.ack >= stopped);
  }
  const update = await client.next(
    expectType,
    (message) => message.view_center[0] === center,
  );
  assertReceived();
  return update;
}

try {
  const manifest = await h.get("/api/manifest");
  async function material(state) {
    const found = await h.control("catalog.search", { query: state, limit: 64 });
    const item = found.items.find((item) => item.state === state);
    assert.ok(item, `Catalog contains ${state}`);
    return item.id;
  }
  const stone = await material("minecraft:stone"),
    gold = await material("minecraft:gold_block"),
    diamond = await material("minecraft:diamond_block"),
    iron = await material("minecraft:iron_block");
  await h.control("world.create", { world });
  const plan = await h.control("build.plan", {
    world,
    expected_revision: 0,
    operations: [
      { type: "box", min: [-64, 0, -32], max: [79, 0, 31], block: stone },
    ],
  });
  await h.control("build.commit", {
    plan_id: plan.plan_id,
    operation_id: "chunk-stream-floor",
  });
  await h.control("arena.configure", {
    world,
    mode: "creative",
    spawn: [15, 2, 8],
  });
  const client = new Client(h.url);
  clients.push(client);
  client.ws.addEventListener("message", (event) => observe(JSON.parse(event.data)));
  await client.open;
  client.send({
    type: "join",
    protocol: 1,
    manifest_hash: manifest.hash,
    world,
    name: "ChunkStreamProbe",
    features: ["chunk_stream_v1"],
  });
  const welcome = await client.next("welcome");
  assertReceived();
  assert.deepEqual(view.viewCenter, [0, 0, 0]);
  assert.deepEqual(view.viewBounds, { min: [-32, -16, -32], max: [31, 31, 31] });
  await client.state((player) => player.position[1] < 1.15);
  await assertCurrentRegion();
  checks.push("Negotiated welcome supplies the complete 48-section view");

  await edit(
    client,
    [
      { pos: [5, 1, 10], block: gold },
      { pos: [-24, 1, 10], block: diamond },
      { pos: [40, 1, 10], block: iron },
    ],
    "chunk-stream-before-east",
  );
  assert.equal(view.blocks.has("40,1,10"), false);
  await walkTo(client, 18, 16);
  await assertCurrentRegion();
  assert.equal(view.blocks.get("5,1,10"), gold);
  assert.equal(view.blocks.has("-24,1,10"), false);
  assert.equal(view.blocks.get("40,1,10"), iron);
  checks.push("East crossing loads 12 sections, unloads 12, retains 36 without snapshot");

  await edit(
    client,
    [
      { pos: [5, 1, 10], block: 0 },
      { pos: [-24, 1, 10], block: gold },
      { pos: [40, 1, 10], block: 0 },
    ],
    "chunk-stream-while-away",
  );
  assert.equal(view.blocks.has("-24,1,10"), false);
  await walkTo(client, 14, 0);
  await assertCurrentRegion();
  assert.equal(view.blocks.get("-24,1,10"), gold);
  assert.equal(view.blocks.has("5,1,10"), false);
  checks.push("Retained deletions persist and returning sections include edits made while unloaded");

  await edit(
    client,
    [
      { pos: [24, 1, 10], block: gold },
      { pos: [-8, 1, 10], block: diamond },
      { pos: [-40, 1, 10], block: iron },
    ],
    "chunk-stream-before-negative",
  );
  await walkTo(client, -2, -16);
  await assertCurrentRegion();
  assert.equal(view.blocks.has("24,1,10"), false);
  assert.equal(view.blocks.get("-8,1,10"), diamond);
  assert.equal(view.blocks.get("-40,1,10"), iron);
  assert.equal(snapshots, 0);
  assert.equal(transitions.length, 3);
  assert.ok(transitions.every((transition) => transition.entering_blocks < welcome.blocks.length));
  checks.push("Negative-coordinate crossing matches authoritative region and sends only entering blocks");

  const beforeResync = sorted(view.blocks);
  explicitResync = true;
  client.send({ type: "resync" });
  await client.next("snapshot");
  assertReceived();
  assert.equal(snapshots, 1);
  assert.deepEqual(sorted(view.blocks), beforeResync);
  await assertCurrentRegion();
  checks.push("Explicit full resync remains available and preserves the current world contents");

  const legacy = new Client(h.url);
  clients.push(legacy);
  const legacyWelcome = await legacy.join(manifest, world, "LegacyChunkProbe");
  assert.ok(!legacyWelcome.features?.includes("chunk_stream_v1"));
  await walkTo(legacy, 18, 16, "snapshot");
  assert.equal(legacy.queue.some((message) => message.type === "chunks"), false);
  checks.push("Clients without feature negotiation retain protocol-1 snapshot fallback");

  await h.report({
    passed: true,
    checks,
    transitions,
    movement_snapshots: 0,
    explicit_resync_snapshots: snapshots,
    initial_blocks: welcome.blocks.length,
    initial_snapshot_bytes: Buffer.byteLength(JSON.stringify(welcome)),
    client_merge: "client/world-view.js",
    binary: h.binary,
  });
} catch (error) {
  await h.report({ passed: false, checks, transitions, error: error.stack, logs: h.logs().slice(-8000) });
  throw error;
} finally {
  await Promise.all(clients.map((client) => client.close()));
  await h.stop();
}