// Peer delivery with no browser: the `peer` bundle and peer.js under node, an asker and its
// holders in one process over links in memory and one local HTTP link, the corpus's world
// behind them, and faults planted on the holders' side.
//
//   node verify_peer.mjs <wai root> [<pkg dir>] [<file to write the session's records to>]
//
// <pkg dir> is the bundle built for the web (default: pkg/ beside this file):
//   cargo build --release --target wasm32-unknown-unknown --features peer
//   wasm-bindgen --target web --out-dir peer/pkg --out-name wai_peer target/wasm32-unknown-unknown/release/wai_web.wasm
// The session file is what `peer_delivery_verify.py --session` recomputes. Exit 0 iff every
// check holds. Every service binds port 0 and says its port; all are closed before exit.

import { writeFileSync } from "node:fs";
import { dirname, join, resolve } from "node:path";
import { fileURLToPath } from "node:url";
import { PeerClient, PeerServer, memoryLinks, memoryStore, overHttp, overLink, hex, unhex } from "./peer.js";
import { checks, httpHolder, listen, loadBundle, loadWorld, sha256 } from "./harness.mjs";

const HERE = dirname(fileURLToPath(import.meta.url));
const [rootArg, pkgArg, sessionOut] = process.argv.slice(2);
if (!rootArg) {
  console.log("usage: node verify_peer.mjs <wai root> [<pkg dir>] [<session file>]");
  process.exit(2);
}
const w = await loadBundle(resolve(pkgArg || join(HERE, "pkg")));
const world = loadWorld(resolve(rootArg));
const { expect, done } = checks();
const [A, B] = world.groups;
const ON = { ask: true, serve: true, max_object_bytes: 1 << 22, relay_only: false };
const seed = (n) => new Uint8Array(32).fill(n);
const servers = [];
const closing = [];

// The build gives the corpus every outcome it states.
const corpus = JSON.parse(w.peerCorpusFailures(world.text));
expect(corpus.length === 0, `the wasm32 build gives every case of the corpus its outcome${corpus.length ? ": " + corpus.slice(0, 3).join(" | ") : ""}`);

const originLog = [];
const origin = (who) => async (group, index) => {
  originLog.push(`${who} ${group === A.link ? "A" : "B"}${index}`);
  return world.groups.find((g) => g.link === group).objects[index].envelope;
};

/** A holder that holds `held` ([group, positions]) and lists `listed` (default: what it holds). */
async function holder(n, held, { listed = held, leafCount = (g) => g.objects.length, alter = [] } = {}) {
  const inner = memoryStore();
  for (const [g, positions] of held) for (const i of positions) await inner.put(g.link, i, g.objects[i].envelope);
  // A planted fault: a holder that sends other bytes than it holds, which §4.3 forbids.
  const store = {
    ...inner,
    get: async (group, index) => {
      const b = await inner.get(group, index);
      if (!b || !alter.some(([g, i]) => g.link === group && i === index)) return b;
      const other = Uint8Array.from(b);
      other[other.length - 1] ^= 1;
      return other;
    },
  };
  const s = new PeerServer(w, { seed: seed(n), policy: ON, setting: { serve: true }, store });
  const now = Date.now();
  s.publish(now - 5000, now + 3600000, listed.map(([g, positions]) => ({ group: g.link, leaf_count: leafCount(g), held: positions })));
  return s;
}

/** A connection to `server` over a link in memory, served until it closes. */
function linked(server) {
  const [mine, theirs] = memoryLinks();
  servers.push(server.serve(theirs, { idle: 5000 }));
  return overLink(w, mine, { timeout: 2000 });
}

const range = (n) => Array.from({ length: n }, (_, i) => i);
const hm = await holder(0xa1, [[A, [0, 1, 2]]], { alter: [[A, 0]] });
const hc = await holder(0xa2, [[A, range(6)]], { leafCount: (g) => g.objects.length + 1 });
const h1 = await holder(0xa3, [[A, [0, 1, 2, 3]], [B, [1]]]);
const hl = await holder(0xa4, [[A, [0, 1, 2, 3, 5]]], { listed: [[A, range(6)]] });
const hh = await holder(0xa5, [[B, range(3)]]);
const httpLog = [];
const http = await listen(httpHolder(hh, (l) => httpLog.push(l)), "the HTTP holder");
closing.push(http);
const httpBase = `http://127.0.0.1:${http.address().port}/`;

// ── An asker whose viewer has not turned asking on.
{
  let opened = 0;
  const c = new PeerClient(w, { seed: seed(0xb0), policy: ON, setting: {}, origin: origin("a0") });
  for (const g of world.groups) c.holdGroup(g.receipt, g.objects.map((o) => o.receipt));
  for (const s of [hm, h1]) await c.addNeighbour(() => (opened++, linked(s)));
  await c.addNeighbour(() => (opened++, overHttp(w, httpBase)));
  const got = await c.get(A.link, 0);
  expect(!c.mayAsk && opened === 0, "setting off: no connection to a holder is opened");
  expect(got.from === "origin" && got.route === "origin reason=peer_not_permitted" && got.verified, "setting off: the object is the origin's, and verifies");
  expect((await c.acknowledge(1, 2)).length === 0 && httpLog.length === 0 && h1.stats.wants === 0, "setting off: nothing is asked of a holder and nothing signed");
  c.close();
}
// ── A holder whose viewer has not turned serving on signs no record and serves no link.
{
  const s = new PeerServer(w, { seed: seed(0xa9), policy: ON, setting: { ask: true }, store: memoryStore() });
  let refused = "";
  try {
    s.publish(1, 2, [{ group: A.link, leaf_count: 6, held: [0] }]);
  } catch (e) {
    refused = e.message;
  }
  const [mine, theirs] = memoryLinks();
  await s.serve(theirs);
  const failed = await overLink(w, mine, { timeout: 200 }).record().then(() => "", (e) => e.reason);
  expect(!s.mayServe && refused === "peer_not_permitted" && s.record() === null && failed === "peer_link_failed", "serving off: no record is signed and a link is closed unanswered");
}

// ── The asker, both switches on: the altering holder first, then the miscounting one, the
//    honest peer, the one that lists what it lacks, and the HTTP holder.
const t0 = Date.now();
const asker = new PeerClient(w, { seed: seed(0xb1), policy: ON, setting: { ask: true }, origin: origin("a1") });
for (const g of world.groups) asker.holdGroup(g.receipt, g.objects.map((o) => o.receipt));
const keys = {};
for (const [name, s] of [["hm", hm], ["hc", hc], ["h1", h1], ["hl", hl]]) keys[name] = await asker.addNeighbour(() => linked(s));
keys.hh = await asker.addNeighbour(() => overHttp(w, httpBase));
expect(Object.entries(keys).every(([n, k]) => k === { hm, hc, h1, hl, hh }[n].key), "each holder's record arrives over its link and names its key");
expect(JSON.stringify(asker.records()) === JSON.stringify(["stands", "reject reason=holdings_leaf_count_mismatch", "stands", "stands", "stands"]), "a record that states a group of six as seven is refused whole");

const got = {};
for (const [name, g] of [["A", A], ["B", B]]) for (const i of range(g.objects.length)) got[`${name}${i}`] = await asker.get(g.link, i);
const from = (k) => (got[k].from === "origin" ? "origin" : Object.keys(keys).find((n) => keys[n] === got[k].from));
const outcomes = (k) => got[k].attempts.map((a) => `${Object.keys(keys).find((n) => keys[n] === a.holder)}:${a.outcome}`).join(", ");

let verified = true;
for (const [name, g] of [["A", A], ["B", B]])
  for (const i of range(g.objects.length)) {
    const r = got[`${name}${i}`];
    verified &&= r.verified && (await sha256(r.bytes)) === (await sha256(g.objects[i].envelope));
  }
expect(verified, "every object the asker holds verified under its receipt and is the publisher's bytes");
expect(outcomes("A0") === "hm:reject step=4 reason=content_hash_mismatch, h1:taken", `altered bytes are refused and the next holder serves (A0: ${outcomes("A0")})`);
expect(["A1", "A2", "A3"].every((k) => outcomes(k) === "h1:taken"), "the holder that served other bytes is asked nothing more (A1 to A3 from h1 alone)");
expect(hm.stats.wants === 1 && hm.stats.acks.length === 0, "the altering holder was asked once and is sent no acknowledgement");
expect(hc.stats.wants === 0, "the holder whose record overstates the group is never asked");
expect(outcomes("A4") === "hl:miss reason=holdings_overstated" && from("A4") === "origin", `a holder that lists what it lacks answers miss, and the origin serves (A4: ${outcomes("A4")})`);
expect(got.A5.attempts.length === 0 && from("A5") === "origin" && hl.stats.wants === 1, "its record is set aside: the next object it listed is not asked of it (A5)");
expect(from("B0") === "hh" && from("B2") === "hh" && from("B1") === "h1", `objects come over the HTTP link and, past one data frame, over a link (B0 ${from("B0")}, B1 ${from("B1")}, B2 ${from("B2")})`);
expect(JSON.stringify(asker.records()) === JSON.stringify(["set-aside reason=peer_bytes_refused", "reject reason=holdings_leaf_count_mismatch", "stands", "set-aside reason=holdings_overstated", "stands"]), "the records' labels after the session");
expect(originLog.filter((l) => l.startsWith("a1 ")).join(",") === "a1 A4,a1 A5", `the origin served only what no peer did (${originLog.filter((l) => l.startsWith("a1 ")).join(",")})`);
expect(httpLog.length > 0 && httpLog.every((l) => !l.cookie && !l.authorization && !l.referer), "requests to the HTTP holder carried no cookie, credential or referrer");

const acks = await asker.acknowledge(t0 * 1000, Date.now() * 1000 + 1);
const by = (name) => acks.find((a) => a.upstream === keys[name]);
expect(acks.length === 2 && by("h1") && by("hh") && acks.every((a) => a.sent && a.taken), "one acknowledgement each to the two holders that served, each taken by its holder");
expect(!acks.some((a) => a.upstream === keys.hm || a.upstream === keys.hl || a.upstream === keys.hc), "no acknowledgement names the altering holder, the overstating one or the miscounting one");
const taken = (s) => s.stats.acks.filter((a) => a.outcome === "taken");
const size = (names) => names.reduce((n, k) => n + got[k].bytes.length, 0);
expect(taken(h1).length === 1 && taken(h1)[0].objects === 5 && taken(h1)[0].bytes === size(["A0", "A1", "A2", "A3", "B1"]) && taken(h1)[0].downstream === asker.key, "h1's acknowledgement states the five objects it served and their bytes");
expect(taken(hh).length === 1 && taken(hh)[0].objects === 2 && taken(hh)[0].bytes === size(["B0", "B2"]), "the HTTP holder's states the two it served and their bytes");
// An acknowledgement replayed to its holder names objects already acknowledged.
{
  const again = hh.takeAck("http", Uint8Array.from([...lenPrefix(by("hh").record), ...unhex(by("hh").record), ...unhex(by("hh").items)]));
  expect(again.outcome === "refuse" && again.reason === "ack_not_served", "an acknowledgement sent twice is not taken twice");
  // And one for another holder is not this holder's.
  const other = h1.takeAck("link-1", Uint8Array.from([...lenPrefix(by("hh").record), ...unhex(by("hh").record), ...unhex(by("hh").items)]));
  expect(other.outcome === "refuse" && other.reason === "ack_wrong_party", "an acknowledgement naming another holder is refused");
}
function lenPrefix(recordHex) {
  const n = recordHex.length / 2;
  return [n >>> 24, (n >>> 16) & 255, (n >>> 8) & 255, n & 255];
}

// ── Holders that break §4.4, written out frame by frame.
{
  const group = unhex(A.link);
  const rogue = async (behave, { max = 1 << 22 } = {}) => {
    const [mine, theirs] = memoryLinks();
    const run = (async () => {
      try {
        await theirs.recv(2000);
        await behave(theirs);
      } catch {
        // the asker closed the link
      }
    })();
    const r = await overLink(w, mine, { timeout: 300 }).want(group, 2, max).then(() => "taken", (e) => e.reason);
    await run;
    return r;
  };
  const bytes = A.objects[2].envelope;
  expect((await rogue((l) => l.send(w.peerHead(group, 3, bytes.length)))) === "peer_frame_unexpected", "a head for another object than was asked is refused");
  expect((await rogue((l) => l.send(w.peerHead(group, 2, 4097)), { max: 4096 })) === "peer_object_too_large", "a head past the bound is refused before any of its bytes");
  expect((await rogue(async (l) => (await l.send(w.peerHead(group, 2, bytes.length)), await l.send(w.peerData(new Uint8Array(bytes.length + 1)))))) === "peer_frame_unexpected", "a byte more than the head stated is refused");
  expect((await rogue(async (l) => (await l.send(w.peerHead(group, 2, bytes.length)), l.close()))) === "peer_link_failed", "a link that closes before the reply is whole has failed");
  expect((await rogue((l) => l.send(new Uint8Array([9, 0])))) === "peer_frame_malformed", "a frame that does not read is refused");
  expect((await rogue(() => new Promise((ok) => setTimeout(ok, 500)))) === "peer_link_failed", "a holder that does not answer in time has failed");
  expect((await rogue((l) => l.send(w.peerMiss(group, 5, 1)))) === "peer_frame_unexpected", "a miss for another object is refused");
}
// ── An HTTP holder that states a length past the bound, and one that sends more than it stated.
{
  const big = await listen((req, res) => res.writeHead(200, { "content-length": 5000 }).end(new Uint8Array(5000)), "a holder past the bound");
  closing.push(big);
  const r = await overHttp(w, `http://127.0.0.1:${big.address().port}/`).want(unhex(A.link), 0, 4096).then(() => "taken", (e) => e.reason);
  expect(r === "peer_object_too_large", "over HTTP, a stated length past the bound is refused");
  const down = await overHttp(w, "http://127.0.0.1:9/", { timeout: 500 }).want(unhex(A.link), 0, 4096).then(() => "taken", (e) => e.reason);
  expect(down === "peer_link_failed", "over HTTP, a request that fails is a failed link");
}

if (sessionOut) {
  writeFileSync(sessionOut, JSON.stringify({ asker: asker.key, holders: keys, acks: acks.map(({ upstream, record, items }) => ({ upstream, record, items })), expected: { [keys.h1]: { objects: 5, bytes: size(["A0", "A1", "A2", "A3", "B1"]) }, [keys.hh]: { objects: 2, bytes: size(["B0", "B2"]) } } }, null, 1) + "\n");
  console.log(`  the session's records: ${sessionOut}`);
}
asker.close();
for (const s of closing) s.close();
await Promise.all(servers);
process.exit(done(33));