import * as childProcess from "node:child_process";
import * as fs from "node:fs";
import * as path from "node:path";
import * as readline from "node:readline";
import { fileURLToPath } from "node:url";

import {
  addListenerTyped,
  addRequestListenerTyped,
  parserFromSchema,
} from "../src/communicator.js";
import { create, TestDataSchema } from "../src/packets/message_common_pb.js";
import { TcpClient } from "../src/tcp_client.js";
import { TcpServer } from "../src/tcp_server.js";
import { NodeWsClient } from "../src/node_ws_client.js";
import { NodeWsServer } from "../src/node_ws_server.js";

const __filename = fileURLToPath(import.meta.url);
const repoRoot = path.resolve(path.dirname(__filename), "../..");
const goDir = path.join(repoRoot, "go");
const CERT_PATH = path.join(repoRoot, "typescript/test/certs/server.crt");
const KEY_PATH = path.join(repoRoot, "typescript/test/certs/server.key");

const HOST = "127.0.0.1";
const TCP_PORT = 29806;
const WS_PORT = 29808;
const TLS_TCP_PORT = 29810;
const WSS_PORT = 29812;
const WS_PATH = "/";
const PROCESS_TIMEOUT_MS = 20_000;
const STREAM_JOIN_TIMEOUT_MS = 5_000;
const SERVER_OBSERVATION_MS = 200;

type Mode = "tcp" | "ws" | "tls" | "wss";
type Phase = "send-push" | "requests" | "legacy-receive";

async function main(): Promise<void> {
  console.log(`INFO typeName ts=${TestDataSchema.typeName}`);
  const selectedTransport = process.env.PROTO_SOCKET_CROSSTEST_TRANSPORT;
  try {
    if (selectedTransport === undefined || selectedTransport === "tcp") {
      await runTcp();
    }
    if (selectedTransport === undefined || selectedTransport === "ws") {
      await runWs();
    }
    if (selectedTransport === undefined || selectedTransport === "tls") {
      await runTls();
    }
    if (selectedTransport === undefined || selectedTransport === "wss") {
      await sleep(150);
      await runWss();
    }
    console.log("PASS all typescript-server/go-client crosstests passed");
  } catch (err) {
    console.error(`FAIL crosstest error=${errorMessage(err)}`);
    process.exitCode = 1;
  }
}

async function runTcpLegacySimpleReceive(): Promise<void> {
  let resolveReceived: ((value: boolean) => void) | null = null;
  const received = new Promise<boolean>((resolve) => {
    resolveReceived = resolve;
  });
  const server = new TcpServer(HOST, TCP_PORT, (socket) => new TcpClient(socket, 0, 0, parserMap()));
  server.onClientConnected = (client) => {
    addListenerTyped(client.communicator, TestDataSchema, (data) => {
      if (resolveReceived !== null) {
        resolveReceived(data.index === 101);
        resolveReceived = null;
      }
    });
  };
  await withServer(server, async () => {
    await runGoClient("tcp", TCP_PORT, "legacy-receive", new Set());
    const ok = await withTimeout(received, SERVER_OBSERVATION_MS, "TCP legacy simple-name server timeout");
    if (!ok) {
      throw new Error("TCP legacy simple-name server received unexpected data");
    }
    console.log("PASS scenario=tcp-legacy-simple-receive detail=server-received-legacy-alias");
  });
}

async function runTcp(): Promise<void> {
  await runTcpSendPush();
  await sleep(150);
  await runTcpRequests();
  await sleep(150);
  await runTcpLegacySimpleReceive();
}

async function runWs(): Promise<void> {
  await runWsSendPush();
  await sleep(150);
  await runWsRequests();
}

async function runTls(): Promise<void> {
  await runTlsTcpSendPush();
  await sleep(150);
  await runTlsTcpRequests();
}

async function runWss(): Promise<void> {
  await runWssSendPush();
  await sleep(150);
  await runWssRequests();
}

async function runTcpSendPush(): Promise<void> {
  let resolveReceived: ((value: boolean) => void) | null = null;
  const received = new Promise<boolean>((resolve) => {
    resolveReceived = resolve;
  });
  const server = new TcpServer(HOST, TCP_PORT, (socket) => new TcpClient(socket, 0, 0, parserMap()));
  server.onClientConnected = (client) => {
    addListenerTyped(client.communicator, TestDataSchema, (data) => {
      console.log(`SERVER_RECEIVED index=${data.index} message=${data.message}`);
      const valid = data.index === 101 && data.message === "fire from go client";
      if (resolveReceived !== null) {
        resolveReceived(valid);
        resolveReceived = null;
      }
      if (valid) {
        void client.communicator.send(
          create(TestDataSchema, {
            index: 200,
            message: "push from typescript server",
          }),
        );
      }
    });
  };
  await withServer(server, async () => {
    await runGoClient("tcp", TCP_PORT, "send-push", new Set(["1", "2"]));
    const ok = await withTimeout(received, SERVER_OBSERVATION_MS, "TCP send-push server did not receive expected data");
    if (!ok) {
      throw new Error("TCP send-push server received unexpected data");
    }
  });
}

async function runTcpRequests(): Promise<void> {
  const server = new TcpServer(HOST, TCP_PORT, (socket) => new TcpClient(socket, 0, 0, parserMap()));
  server.onClientConnected = (client) => {
    addRequestListenerTyped(client.communicator, TestDataSchema, (req) =>
      create(TestDataSchema, {
        index: req.index * 2,
        message: `echo: ${req.message}`,
      }),
    );
  };
  await withServer(server, () => runGoClient("tcp", TCP_PORT, "requests", new Set(["3", "4"])));
}

async function runWsSendPush(): Promise<void> {
  let resolveReceived: ((value: boolean) => void) | null = null;
  const received = new Promise<boolean>((resolve) => {
    resolveReceived = resolve;
  });
  const server = new NodeWsServer(HOST, WS_PORT, WS_PATH, (ws) => new NodeWsClient(ws, 0, 0, parserMap()));
  server.onClientConnected = (client) => {
    addListenerTyped(client.communicator, TestDataSchema, (data) => {
      console.log(`SERVER_RECEIVED index=${data.index} message=${data.message}`);
      const valid = data.index === 101 && data.message === "fire from go client";
      if (resolveReceived !== null) {
        resolveReceived(valid);
        resolveReceived = null;
      }
      if (valid) {
        void client.communicator.send(
          create(TestDataSchema, {
            index: 200,
            message: "push from typescript server",
          }),
        );
      }
    });
  };
  await withServer(server, async () => {
    await runGoClient("ws", WS_PORT, "send-push", new Set(["1", "2"]));
    const ok = await withTimeout(received, SERVER_OBSERVATION_MS, "WS send-push server did not receive expected data");
    if (!ok) {
      throw new Error("WS send-push server received unexpected data");
    }
  });
}

async function runWsRequests(): Promise<void> {
  const server = new NodeWsServer(HOST, WS_PORT, WS_PATH, (ws) => new NodeWsClient(ws, 0, 0, parserMap()));
  server.onClientConnected = (client) => {
    addRequestListenerTyped(client.communicator, TestDataSchema, (req) =>
      create(TestDataSchema, {
        index: req.index * 2,
        message: `echo: ${req.message}`,
      }),
    );
  };
  await withServer(server, () => runGoClient("ws", WS_PORT, "requests", new Set(["3", "4"])));
}

async function runTlsTcpSendPush(): Promise<void> {
  let resolveReceived: ((value: boolean) => void) | null = null;
  const received = new Promise<boolean>((resolve) => {
    resolveReceived = resolve;
  });
  const server = new TcpServer(
    HOST,
    TLS_TCP_PORT,
    (socket) => new TcpClient(socket, 0, 0, parserMap()),
    serverTlsOptions(),
  );
  server.onClientConnected = (client) => {
    addListenerTyped(client.communicator, TestDataSchema, (data) => {
      console.log(`SERVER_RECEIVED index=${data.index} message=${data.message}`);
      const valid = data.index === 101 && data.message === "fire from go client";
      if (resolveReceived !== null) {
        resolveReceived(valid);
        resolveReceived = null;
      }
      if (valid) {
        void client.communicator.send(
          create(TestDataSchema, {
            index: 200,
            message: "push from typescript server",
          }),
        );
      }
    });
  };
  await withServer(server, async () => {
    await runGoClient("tls", TLS_TCP_PORT, "send-push", new Set(["1", "2"]), CERT_PATH);
    const ok = await withTimeout(received, SERVER_OBSERVATION_MS, "TLS TCP send-push server did not receive expected data");
    if (!ok) {
      throw new Error("TLS TCP send-push server received unexpected data");
    }
  });
}

async function runTlsTcpRequests(): Promise<void> {
  const server = new TcpServer(
    HOST,
    TLS_TCP_PORT,
    (socket) => new TcpClient(socket, 0, 0, parserMap()),
    serverTlsOptions(),
  );
  server.onClientConnected = (client) => {
    addRequestListenerTyped(client.communicator, TestDataSchema, (req) =>
      create(TestDataSchema, {
        index: req.index * 2,
        message: `echo: ${req.message}`,
      }),
    );
  };
  await withServer(server, () => runGoClient("tls", TLS_TCP_PORT, "requests", new Set(["3", "4"]), CERT_PATH));
}

async function runWssSendPush(): Promise<void> {
  let resolveReceived: ((value: boolean) => void) | null = null;
  const received = new Promise<boolean>((resolve) => {
    resolveReceived = resolve;
  });
  const server = new NodeWsServer(
    HOST,
    WSS_PORT,
    WS_PATH,
    (ws) => new NodeWsClient(ws, 0, 0, parserMap()),
    serverTlsOptions(),
  );
  server.onClientConnected = (client) => {
    addListenerTyped(client.communicator, TestDataSchema, (data) => {
      console.log(`SERVER_RECEIVED index=${data.index} message=${data.message}`);
      const valid = data.index === 101 && data.message === "fire from go client";
      if (resolveReceived !== null) {
        resolveReceived(valid);
        resolveReceived = null;
      }
      if (valid) {
        void client.communicator.send(
          create(TestDataSchema, {
            index: 200,
            message: "push from typescript server",
          }),
        );
      }
    });
  };
  await withServer(server, async () => {
    await runGoClient("wss", WSS_PORT, "send-push", new Set(["1", "2"]), CERT_PATH);
    const ok = await withTimeout(received, SERVER_OBSERVATION_MS, "WSS send-push server did not receive expected data");
    if (!ok) {
      throw new Error("WSS send-push server received unexpected data");
    }
  });
}

async function runWssRequests(): Promise<void> {
  const server = new NodeWsServer(
    HOST,
    WSS_PORT,
    WS_PATH,
    (ws) => new NodeWsClient(ws, 0, 0, parserMap()),
    serverTlsOptions(),
  );
  server.onClientConnected = (client) => {
    addRequestListenerTyped(client.communicator, TestDataSchema, (req) =>
      create(TestDataSchema, {
        index: req.index * 2,
        message: `echo: ${req.message}`,
      }),
    );
  };
  await withServer(server, () => runGoClient("wss", WSS_PORT, "requests", new Set(["3", "4"]), CERT_PATH));
}

async function withServer(
  server: { start: () => Promise<void>; stop: () => Promise<void> },
  body: () => Promise<void>,
): Promise<void> {
  await server.start();
  try {
    await body();
  } finally {
    await server.stop();
  }
}

async function runGoClient(
  mode: Mode,
  port: number,
  phase: Phase,
  expected: Set<string>,
  cert?: string,
): Promise<void> {
  await runClientProcess(
    "go",
    [
      "run",
      "./crosstest/typescript_go_client",
      `--mode=${mode}`,
      `--port=${port}`,
      `--phase=${phase}`,
      ...(cert === undefined ? [] : [`--cert=${cert}`]),
    ],
    goDir,
    `go-client ${mode}/${phase}`,
    expected,
  );
}

async function runClientProcess(
  command: string,
  args: string[],
  cwd: string,
  label: string,
  expected: Set<string>,
): Promise<void> {
  const child = childProcess.spawn(command, args, {
    cwd,
    detached: process.platform !== "win32",
    stdio: ["ignore", "pipe", "pipe"],
  });

  if (child.stdout === null || child.stderr === null) {
    throw new Error(`${label} stdio is not piped`);
  }

  const resultLines: string[] = [];
  const stdoutDone = collectLines(child.stdout, (line) => {
    console.log(line);
    if (line.startsWith("PASS ") || line.startsWith("FAIL ")) {
      resultLines.push(line);
    }
  });
  const stderrDone = collectLines(child.stderr, (line) => {
    console.error(line);
  });

  let timedOut = false;
  const timer = setTimeout(() => {
    timedOut = true;
    killProcessTree(child);
  }, PROCESS_TIMEOUT_MS);

  let exitCode: number;
  try {
    exitCode = await new Promise<number>((resolve, reject) => {
      child.once("error", reject);
      child.once("close", (code) => resolve(code ?? -1));
    });
  } finally {
    clearTimeout(timer);
  }

  await withTimeout(
    Promise.all([stdoutDone, stderrDone]),
    STREAM_JOIN_TIMEOUT_MS,
    `${label} streams did not finish`,
  );
  if (timedOut) {
    throw new Error(`${label} timed out`);
  }

  validateResultLines(label, exitCode, resultLines, expected);
}

function collectLines(stream: NodeJS.ReadableStream, onLine: (line: string) => void): Promise<void> {
  const rl = readline.createInterface({ input: stream });
  return new Promise((resolve, reject) => {
    rl.on("line", onLine);
    rl.once("close", resolve);
    rl.once("error", reject);
  });
}

function validateResultLines(label: string, exitCode: number, lines: string[], expected: Set<string>): void {
  const failed = lines.filter((line) => line.startsWith("FAIL "));
  const passed = new Set<string>();
  for (const line of lines) {
    if (!line.startsWith("PASS ")) {
      continue;
    }
    const match = /scenario=([^ ]+)/.exec(line);
    if (match !== null) {
      passed.add(match[1]);
    }
  }

  const missing = [...expected].filter((scenario) => !passed.has(scenario));
  if (exitCode !== 0 || failed.length > 0 || missing.length > 0) {
    throw new Error(`${label} failed exitCode=${exitCode} failed=${JSON.stringify(failed)} missing=${JSON.stringify(missing)}`);
  }
}

function parserMap() {
  return new Map([[TestDataSchema.typeName, parserFromSchema(TestDataSchema)]]);
}

function serverTlsOptions(): { cert: Buffer; key: Buffer } {
  return {
    cert: fs.readFileSync(CERT_PATH),
    key: fs.readFileSync(KEY_PATH),
  };
}

function killProcessTree(child: childProcess.ChildProcess): void {
  if (process.platform !== "win32" && child.pid !== undefined) {
    try {
      process.kill(-child.pid, "SIGKILL");
      return;
    } catch {
      // fallthrough to direct kill
    }
  }
  child.kill("SIGKILL");
}

async function withTimeout<T>(promise: Promise<T>, timeoutMs: number, label: string): Promise<T> {
  let timer: ReturnType<typeof setTimeout> | undefined;
  try {
    return await Promise.race([
      promise,
      new Promise<T>((_, reject) => {
        timer = setTimeout(() => reject(new Error(label)), timeoutMs);
      }),
    ]);
  } finally {
    if (timer !== undefined) {
      clearTimeout(timer);
    }
  }
}

function errorMessage(err: unknown): string {
  return err instanceof Error ? err.message : String(err);
}

function sleep(ms: number): Promise<void> {
  return new Promise((resolve) => {
    setTimeout(resolve, ms);
  });
}

void main();
