devplacepy/nodes/extension/test/wsclient.test.mjs

import test from "node:test";
import assert from "node:assert/strict";
import { Fetcher } from "../src/Fetcher.js";
import { Limits } from "../src/Limits.js";
import { Policy } from "../src/Policy.js";
import { State, STATES } from "../src/State.js";
import { Usage } from "../src/Usage.js";
import { WsClient } from "../src/WsClient.js";
import { DESTINATIONS } from "../src/spec.js";

function memoryArea() {
  const store = {};
  return {
    async get(key) {
      return key in store ? { [key]: store[key] } : {};
    },
    async set(values) {
      Object.assign(store, values);
    },
  };
}

class FakeSocket {
  static instances = [];

  constructor(address) {
    this.address = address;
    this.readyState = 0;
    this.sent = [];
    this.closed = null;
    FakeSocket.instances.push(this);
  }

  open() {
    this.readyState = 1;
    this.onopen();
  }

  send(text) {
    this.sent.push(JSON.parse(text));
  }

  close(code, reason) {
    this.readyState = 3;
    this.closed = { code, reason };
  }

  receive(frame) {
    return this.onmessage({ data: typeof frame === "string" ? frame : JSON.stringify(frame) });
  }

  serverClose(code, reason = "") {
    this.readyState = 3;
    return this.onclose({ code, reason });
  }
}

class FakeTimers {
  constructor() {
    this.timeouts = [];
    this.intervals = [];
  }
  setTimeout(fn, ms) {
    const handle = { fn, ms, cleared: false };
    this.timeouts.push(handle);
    return handle;
  }
  clearTimeout(handle) {
    handle.cleared = true;
  }
  setInterval(fn, ms) {
    const handle = { fn, ms, cleared: false };
    this.intervals.push(handle);
    return handle;
  }
  clearInterval(handle) {
    handle.cleared = true;
  }
}

function build(overrides = {}) {
  FakeSocket.instances = [];
  const timers = new FakeTimers();
  const limits = new Limits();
  const usage = new Usage(memoryArea());
  const state = new State(memoryArea());
  const fetchLog = [];
  const jobFetch = overrides.jobFetch || (async () => new Response("page", { status: 200 }));
  const fetcher = new Fetcher({ policy: new Policy(), limits, usage, fetchImpl: jobFetch });
  const ticketFetch =
    overrides.ticketFetch ||
    (async (url, options) => {
      fetchLog.push({ url, options });
      return new Response(JSON.stringify({ ticket: "tk1", expires_in: 30, ws_path: "/nodes/ws" }), {
        status: 200,
      });
    });
  const client = new WsClient({
    config: { credentials: () => ({ serverUrl: "https://devplace.net", secret: "dpn_secretsecretsecret" }) },
    limits,
    fetcher,
    state,
    fetchImpl: ticketFetch,
    WebSocketImpl: FakeSocket,
    timers,
    random: overrides.random || (() => 0.5),
    clientVersion: "9.9.9",
    platform: "firefox",
  });
  return { client, timers, state, fetcher, limits, usage, fetchLog };
}

const welcome = (extra = {}) => ({
  type: "welcome",
  device_uid: "dev1",
  limits: { max_concurrent_per_node: 2, ping_seconds: 20 },
  destinations: DESTINATIONS.map((item) => ({ ...item })),
  settings: { bandwidth_mb_day: 500, max_concurrent: 2 },
  ...extra,
});

async function online(context) {
  await context.client.start();
  const socket = FakeSocket.instances.at(-1);
  socket.open();
  await socket.receive(welcome());
  return socket;
}

test("start fetches a ticket with the bearer secret and opens the socket", async () => {
  const context = build();
  await context.client.start();
  assert.equal(context.fetchLog[0].url, "https://devplace.net/nodes/ticket");
  assert.equal(context.fetchLog[0].options.headers.Authorization, "Bearer dpn_secretsecretsecret");
  assert.equal(context.fetchLog[0].options.credentials, "omit");
  const socket = FakeSocket.instances[0];
  assert.equal(socket.address, "wss://devplace.net/nodes/ws?ticket=tk1");
});

test("hello is sent on open with extension kind and fetch mode only", async () => {
  const context = build();
  await context.client.start();
  const socket = FakeSocket.instances[0];
  socket.open();
  const hello = socket.sent[0];
  assert.equal(hello.type, "hello");
  assert.equal(hello.protocol, 1);
  assert.equal(hello.kind, "extension");
  assert.deepEqual(hello.modes, ["fetch"]);
  assert.equal(hello.client_version, "9.9.9");
  assert.equal(hello.platform, "firefox");
  assert.deepEqual(hello.limits, { max_concurrent: 2 });
});

test("welcome puts the client online and starts the keepalive", async () => {
  const context = build();
  await online(context);
  assert.equal(context.state.snapshot().state, STATES.ONLINE);
  assert.equal(context.state.snapshot().deviceUid, "dev1");
  assert.equal(context.timers.intervals[0].ms, 20000);
});

test("keepalive timer sends pong and a server ping is answered with pong", async () => {
  const context = build();
  const socket = await online(context);
  context.timers.intervals[0].fn();
  await socket.receive({ type: "ping" });
  assert.deepEqual(
    socket.sent.filter((frame) => frame.type === "pong"),
    [{ type: "pong" }, { type: "pong" }],
  );
});

test("a job is executed and answered with job_result then usage", async () => {
  const context = build();
  const socket = await online(context);
  await socket.receive({
    type: "job",
    job_id: "j9",
    method: "GET",
    url: "https://www.reddit.com/r/x.json",
    timeout: 5,
  });
  await context.client.idle();
  const result = socket.sent.find((frame) => frame.type === "job_result");
  assert.equal(result.job_id, "j9");
  assert.equal(Buffer.from(result.body_b64, "base64").toString(), "page");
  assert.equal(socket.sent.some((frame) => frame.type === "usage"), true);
});

test("a job for a forbidden host is answered with job_error", async () => {
  const context = build();
  const socket = await online(context);
  await socket.receive({ type: "job", job_id: "j2", method: "GET", url: "https://evilreddit.com/", timeout: 5 });
  await context.client.idle();
  const error = socket.sent.find((frame) => frame.type === "job_error");
  assert.deepEqual(error, { type: "job_error", job_id: "j2", code: "host_not_allowed" });
});

test("a welcome that widens the allowlist is ignored for the widened entries", async () => {
  const context = build();
  await context.client.start();
  const socket = FakeSocket.instances[0];
  socket.open();
  await socket.receive(
    welcome({
      destinations: [
        ...DESTINATIONS.map((item) => ({ ...item })),
        { suffix: "evil.net", adapter: "x", methods: ["GET"], ports: [443], modes: ["fetch"] },
      ],
    }),
  );
  assert.match(context.state.snapshot().lastError, /evil\.net/);
  await socket.receive({ type: "job", job_id: "j3", method: "GET", url: "https://evil.net/", timeout: 5 });
  await context.client.idle();
  assert.equal(socket.sent.find((frame) => frame.type === "job_error").code, "host_not_allowed");
});

test("a job before welcome is a protocol fault and closes with 4400", async () => {
  const context = build();
  await context.client.start();
  const socket = FakeSocket.instances[0];
  socket.open();
  await socket.receive({ type: "job", job_id: "j", method: "GET", url: "https://reddit.com/", timeout: 5 });
  assert.equal(socket.closed.code, 4400);
  assert.equal(context.state.snapshot().state, STATES.BACKOFF);
});

test("malformed frames are rejected loudly", async () => {
  const context = build();
  const socket = await online(context);
  for (const bad of ["not json", "[]", JSON.stringify({ type: "nope" }), JSON.stringify({ type: "job", job_id: "x" })]) {
    await assert.rejects(context.client.handleMessage(bad));
  }
  await assert.rejects(context.client.handleMessage(new ArrayBuffer(4)));
  assert.equal(socket.sent.filter((frame) => frame.type === "job_result").length, 0);
});

test("kick halts without reconnecting", async () => {
  const context = build();
  const socket = await online(context);
  await socket.receive({ type: "kick", reason: "replaced" });
  assert.equal(context.state.snapshot().state, STATES.HALTED);
  assert.match(context.state.snapshot().detail, /replaced/);
  assert.equal(context.timers.timeouts.filter((handle) => !handle.cleared).length, 0);
  await context.client.ensure();
  assert.equal(FakeSocket.instances.length, 1);
});

test("a normal close schedules a full jitter reconnect within the bounds", async () => {
  const context = build({ random: () => 0.999 });
  const socket = await online(context);
  await socket.serverClose(1006);
  const timer = context.timers.timeouts.at(-1);
  assert.ok(timer.ms >= 1000 && timer.ms <= 60000, String(timer.ms));
  assert.equal(context.state.snapshot().state, STATES.BACKOFF);
  timer.fn();
  await new Promise((resolve) => setImmediate(resolve));
  assert.equal(FakeSocket.instances.length, 2);
});

test("backoff grows with attempts and never exceeds reconnect_max", async () => {
  const context = build({ random: () => 1 });
  await context.client.start();
  const delays = [];
  for (let index = 0; index < 12; index += 1) {
    const socket = FakeSocket.instances.at(-1);
    await socket.serverClose(1006);
    const timer = context.timers.timeouts.at(-1);
    delays.push(timer.ms);
    timer.fn();
    await new Promise((resolve) => setImmediate(resolve));
  }
  assert.ok(delays[1] > delays[0]);
  assert.ok(Math.max(...delays) <= 60000);
  assert.equal(delays.at(-1), 60000);
});

test("retry owner close reconnects at the minimum delay", async () => {
  const context = build({ random: () => 1 });
  const socket = await online(context);
  await socket.serverClose(4013);
  assert.equal(context.timers.timeouts.at(-1).ms, 1000);
});

test("a consent or device refusal halts instead of hammering the server", async () => {
  const context = build();
  await context.client.start();
  const socket = FakeSocket.instances[0];
  await socket.serverClose(1008, "no_consent");
  assert.equal(context.state.snapshot().state, STATES.HALTED);
  assert.match(context.state.snapshot().detail, /no_consent/);
});

test("a ticket 401 halts, a ticket 500 retries", async () => {
  const halted = build({
    ticketFetch: async () => new Response(JSON.stringify({ error: "x" }), { status: 401 }),
  });
  await halted.client.start();
  assert.equal(halted.state.snapshot().state, STATES.HALTED);
  assert.equal(FakeSocket.instances.length, 0);

  const retried = build({ ticketFetch: async () => new Response("boom", { status: 500 }) });
  await retried.client.start();
  assert.equal(retried.state.snapshot().state, STATES.BACKOFF);
});

test("stop closes the socket and clears timers", async () => {
  const context = build();
  const socket = await online(context);
  await context.client.stop("user");
  assert.equal(socket.closed.code, 1000);
  assert.equal(context.timers.intervals[0].cleared, true);
  assert.equal(context.state.snapshot().state, STATES.OFF);
});

test("settings_update can only lower the effective limits", async () => {
  const context = build();
  const socket = await online(context);
  await socket.receive({ type: "settings_update", settings: { bandwidth_mb_day: 10, max_concurrent: 1 } });
  assert.equal(context.limits.concurrency, 1);
  assert.equal(context.limits.bandwidthBytes, 10 * 1024 * 1024);
});