// The RPC client against an in-process server that frames messages the way Effect RPC does.
import { afterEach, describe, expect, it } from "bun:test";
import type { Server, ServerWebSocket } from "bun";

import { RpcError, RpcSocket } from "../src/t3/rpc.ts";

type Frame = { _tag: string; id?: string; requestId?: string; tag?: string; payload?: unknown };

let server: Server<undefined> | undefined;
afterEach(() => server?.stop(true));

/** Starts a server whose `onRequest` decides how to answer each Request frame. */
function serve(onRequest: (ws: ServerWebSocket<undefined>, frame: Frame, received: Frame[]) => void) {
  const received: Frame[] = [];
  server = Bun.serve({
    port: 0,
    fetch: (req, srv) => (srv.upgrade(req) ? undefined : new Response("no", { status: 400 })),
    websocket: {
      message(ws, data) {
        const frame = JSON.parse(String(data)) as Frame;
        received.push(frame);
        if (frame._tag === "Request") onRequest(ws, frame, received);
      },
    },
  });
  return { url: `ws://localhost:${server.port}`, received };
}

const send = (ws: ServerWebSocket<undefined>, frame: object) => ws.send(JSON.stringify(frame));

describe("RpcSocket", () => {
  it("resolves a unary call with its success value", async () => {
    const { url } = serve((ws, frame) =>
      send(ws, { _tag: "Exit", requestId: frame.id, exit: { _tag: "Success", value: { echoed: frame.payload } } }),
    );
    const socket = await RpcSocket.connect(url);
    expect(await socket.call("server.probe", { a: 1 })).toEqual({ echoed: { a: 1 } });
    socket.close();
  });

  it("rejects with the server's typed error", async () => {
    const error = { _tag: "EnvironmentAuthorizationError", message: "missing scope", requiredScope: "orchestration:operate" };
    const { url } = serve((ws, frame) =>
      send(ws, { _tag: "Exit", requestId: frame.id, exit: { _tag: "Failure", cause: [{ _tag: "Fail", error }] } }),
    );
    const socket = await RpcSocket.connect(url);
    const failure = await socket.call("orchestration.dispatchCommand", {}).catch((e: unknown) => e);
    expect(failure).toBeInstanceOf(RpcError);
    expect((failure as RpcError).error).toEqual(error);
    socket.close();
  });

  it("acks every chunk so the server sends the next one", async () => {
    const { url, received } = serve((ws, frame) => {
      // Like Effect RPC, send one chunk, then wait for its Ack before the next.
      const values = [[1, 2], [3]];
      const next = () => {
        const value = values.shift();
        if (value) send(ws, { _tag: "Chunk", requestId: frame.id, values: value });
        else send(ws, { _tag: "Exit", requestId: frame.id, exit: { _tag: "Success", value: null } });
      };
      const waitForAck = setInterval(() => {
        const acks = received.filter((f) => f._tag === "Ack").length;
        if (acks === 2 - values.length) next();
        if (values.length === 0 && acks === 2) clearInterval(waitForAck);
      }, 2);
      next();
    });
    const socket = await RpcSocket.connect(url);
    const seen: unknown[] = [];
    await socket.stream("orchestration.subscribeShell", {}, (value) => seen.push(value)).done;
    expect(seen).toEqual([1, 2, 3]);
    expect(received.filter((f) => f._tag === "Ack")).toHaveLength(2);
    socket.close();
  });

  it("fails pending calls when the socket closes", async () => {
    const { url } = serve((ws) => ws.close());
    const socket = await RpcSocket.connect(url);
    expect(socket.call("server.probe", {})).rejects.toThrow("socket closed");
    await socket.closed;
  });

  it("closes a silent socket", async () => {
    // This server never answers Pings.
    const { url } = serve(() => {});
    const socket = await RpcSocket.connect(url, { pingIntervalMs: 10, silenceTimeoutMs: 40 });
    await socket.closed;
  });
});