/**
 * Inbound queue ordering stress / benchmark harness (same-language TypeScript baseline).
 *
 * Milestone `inbound-queue-ordering`의 verify Epic 기준선을 만든다. 절대 성능 합격선을 고정하지 않고
 * local 환경 baseline(throughput, p50/p95/p99 latency, peak memory)을 남기며, 안정성 합격선만 hard fail로
 * 둔다: nonce mismatch 0, response type mismatch 0, unexpected timeout 0, connection별 FIFO 위반 0,
 * 종료 후 pending/queue leak 0.
 *
 * 출력 계약(stdout): 각 측정 축마다 한 줄의 `ROW|...` 레코드를 내고, 마지막에 `SUMMARY|...` 한 줄을 낸다.
 * 사람이 읽는 진행 로그는 stderr로 보낸다. 안정성 위반이 하나라도 있으면 process exit code 1.
 *
 * 측정 축(profile):
 * - roundtrip       : 동일 connection request-response, concurrency 1/16/64/256, latency 분포.
 * - burst           : 단일 connection fire-and-forget burst, dispatch 순서/누락/leak.
 * - sustained       : 일정 시간 지속 request-response, throughput/p95/p99/peak memory.
 * - parallel        : 다중 connection 동시 request-response, connection별 FIFO/nonce 독립성.
 * - gateway         : communicator `onReceivedFrame` opt-in 경로에서 worker_threads gateway(on) vs
 *                     inline decode fallback(off) 비교 baseline. TcpClient는 inline decode로 enqueue하므로
 *                     gateway 효과는 이 frame-ingest 경로에서만 켤 수 있고, 작은 payload에서는 이득이
 *                     불명확하여 inline fallback을 안전 기본값으로 본다.
 */

import * as net from "node:net";
import { availableParallelism } from "node:os";

import {
  addListenerTyped,
  Communicator,
  parserFromSchema,
  type ParserMap,
  type Transport,
} from "../src/communicator.js";
import { createNodeWorkerGateway } from "../src/node_inbound_gateway.js";
import {
  create,
  PacketBaseSchema,
  TestDataSchema,
  toBinary,
  type TestData,
} from "../src/packets/message_common_pb.js";
import { connectTcp, TcpClient } from "../src/tcp_client.js";
import { TcpServer } from "../src/tcp_server.js";

/** stress harness 전용 TCP client factory. heartbeat 비활성(interval=0)으로 측정 noise를 줄인다. */
function newTcpClient(socket: net.Socket): TcpClient {
  return new TcpClient(socket, 0, 0, parserMap());
}

const HOST = "127.0.0.1";
const ALL_PROFILES = ["roundtrip", "burst", "sustained", "parallel", "gateway"] as const;
type Profile = (typeof ALL_PROFILES)[number];
type Mode = "quick" | "full";

/** 안정성 위반 카운터. 0이 아니면 해당 축은 FAIL이고 프로세스도 non-zero exit한다. */
interface Stability {
  timeouts: number;
  nonceMismatch: number;
  typeMismatch: number;
  fifoViolations: number;
  pendingLeak: number;
}

interface Row {
  profile: Profile;
  axis: string;
  requests: number;
  throughputRps: number;
  p50: number;
  p95: number;
  p99: number;
  gateway: string;
  memMb: number;
  stability: Stability;
}

const rows: Row[] = [];

function newStability(): Stability {
  return { timeouts: 0, nonceMismatch: 0, typeMismatch: 0, fifoViolations: 0, pendingLeak: 0 };
}

function stabilityViolations(s: Stability): number {
  return s.timeouts + s.nonceMismatch + s.typeMismatch + s.fifoViolations + s.pendingLeak;
}

function rowStatus(row: Row): "PASS" | "FAIL" {
  return stabilityViolations(row.stability) === 0 ? "PASS" : "FAIL";
}

function emitRow(row: Row): void {
  rows.push(row);
  const s = row.stability;
  process.stdout.write(
    [
      "ROW",
      row.profile,
      row.axis,
      String(row.requests),
      row.throughputRps.toFixed(1),
      row.p50.toFixed(3),
      row.p95.toFixed(3),
      row.p99.toFixed(3),
      String(s.timeouts),
      String(s.nonceMismatch),
      String(s.typeMismatch),
      String(s.fifoViolations),
      String(s.pendingLeak),
      row.gateway,
      row.memMb.toFixed(1),
      rowStatus(row),
    ].join("|") + "\n",
  );
}

function log(line: string): void {
  process.stderr.write(line + "\n");
}

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

function percentile(sorted: number[], p: number): number {
  if (sorted.length === 0) {
    return 0;
  }
  const rank = Math.ceil((p / 100) * sorted.length) - 1;
  const idx = Math.min(sorted.length - 1, Math.max(0, rank));
  return sorted[idx];
}

function memMb(): number {
  return process.memoryUsage().rss / (1024 * 1024);
}

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

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

/** request 실패를 안정성 카운터의 적절한 항목으로 분류한다. */
function classifyRequestError(message: string, stability: Stability): void {
  if (message.includes("timeout")) {
    stability.timeouts += 1;
  } else if (message.includes("type mismatch")) {
    stability.typeMismatch += 1;
  } else {
    stability.nonceMismatch += 1;
  }
}

/** request-response echo + per-connection FIFO 감시를 붙인 TCP 서버를 띄우고 body를 실행한다. */
async function withRequestServer(
  stability: Stability,
  body: (port: number) => Promise<void>,
): Promise<void> {
  const server = new TcpServer(HOST, 0, newTcpClient);
  server.onClientConnected = (client) => {
    let lastNonce = 0;
    // raw addRequestListener로 nonce를 받아 connection별 FIFO(단조 증가) 위반을 감시한다.
    client.communicator.addRequestListener(TestDataSchema.typeName, (msg, nonce) => {
      if (nonce < lastNonce) {
        stability.fifoViolations += 1;
      }
      lastNonce = nonce;
      const data = msg as TestData;
      return create(TestDataSchema, { index: data.index * 2, message: `echo:${data.message}` });
    });
  };
  await server.start();
  try {
    await body(server.port);
  } finally {
    await server.stop();
  }
}

/** 한 connection에서 concurrency C로 request batch를 보내고 latency를 모은다. */
async function runRequestLoad(
  client: TcpClient,
  total: number,
  concurrency: number,
  timeoutMs: number,
  stability: Stability,
): Promise<number[]> {
  const latencies: number[] = [];
  let issued = 0;
  let nextIndex = 0;

  async function oneRequest(index: number): Promise<void> {
    const started = performance.now();
    try {
      const res = await client.communicator.sendRequest(
        create(TestDataSchema, { index, message: `req-${index}` }),
        TestDataSchema,
        timeoutMs,
      );
      const elapsed = performance.now() - started;
      latencies.push(elapsed);
      if (res.index !== index * 2 || res.message !== `echo:req-${index}`) {
        stability.nonceMismatch += 1;
      }
    } catch (err) {
      classifyRequestError(errorMessage(err), stability);
    }
  }

  while (issued < total) {
    const batchSize = Math.min(concurrency, total - issued);
    const batch: Array<Promise<void>> = [];
    for (let i = 0; i < batchSize; i += 1) {
      batch.push(oneRequest(nextIndex));
      nextIndex += 1;
    }
    issued += batchSize;
    await Promise.all(batch);
  }

  return latencies;
}

/** latency 분포가 없는 throughput 중심 축(burst/gateway)을 위한 row emit. */
function emitThroughputRow(
  profile: Profile,
  axis: string,
  count: number,
  elapsedMs: number,
  stability: Stability,
  gateway: string,
  observedMemMb: number,
): void {
  emitRow({
    profile,
    axis,
    requests: count,
    throughputRps: elapsedMs > 0 ? (count / elapsedMs) * 1000 : 0,
    p50: 0,
    p95: 0,
    p99: 0,
    gateway,
    memMb: observedMemMb,
    stability,
  });
}

function summarizeLatencies(
  profile: Profile,
  axis: string,
  latencies: number[],
  elapsedMs: number,
  stability: Stability,
  gateway: string,
  observedMemMb: number,
): void {
  const sorted = [...latencies].sort((a, b) => a - b);
  const throughput = elapsedMs > 0 ? (latencies.length / elapsedMs) * 1000 : 0;
  emitRow({
    profile,
    axis,
    requests: latencies.length,
    throughputRps: throughput,
    p50: percentile(sorted, 50),
    p95: percentile(sorted, 95),
    p99: percentile(sorted, 99),
    gateway,
    memMb: observedMemMb,
    stability,
  });
}

async function profileRoundtrip(mode: Mode): Promise<void> {
  const concurrencies = [1, 16, 64, 256];
  const batches = mode === "quick" ? 2 : 20;
  const timeoutMs = mode === "quick" ? 5_000 : 15_000;
  log(`[roundtrip] mode=${mode} concurrencies=${concurrencies.join(",")} batches/level=${batches}`);

  for (const concurrency of concurrencies) {
    const stability = newStability();
    const total = concurrency * batches;
    await withRequestServer(stability, async (port) => {
      const client = await connectTcp(HOST, port, 0, 0, parserMap());
      try {
        const started = performance.now();
        const latencies = await runRequestLoad(client, total, concurrency, timeoutMs, stability);
        const elapsed = performance.now() - started;
        summarizeLatencies("roundtrip", `concurrency=${concurrency}`, latencies, elapsed, stability, "off", memMb());
      } finally {
        await client.close();
        if (!leakClear(client)) {
          stability.pendingLeak += 1;
        }
      }
    });
    log(`[roundtrip] concurrency=${concurrency} done violations=${stabilityViolations(stability)}`);
  }
}

/** close 후 communicator가 더 이상 살아있지 않은지로 pending/queue leak을 근사한다. */
function leakClear(client: TcpClient): boolean {
  return !client.communicator.isAlive();
}

async function profileBurst(mode: Mode): Promise<void> {
  const counts = mode === "quick" ? [200] : [1_000, 10_000];
  log(`[burst] mode=${mode} counts=${counts.join(",")}`);

  for (const count of counts) {
    const stability = newStability();
    let received = 0;
    let lastIndex = -1;
    let resolveDone: (() => void) | null = null;
    const done = new Promise<void>((resolve) => {
      resolveDone = resolve;
    });

    const server = new TcpServer(HOST, 0, newTcpClient);
    server.onClientConnected = (client) => {
      addListenerTyped(client.communicator, TestDataSchema, (msg) => {
        if (msg.index <= lastIndex) {
          stability.fifoViolations += 1;
        }
        lastIndex = msg.index;
        received += 1;
        if (received >= count && resolveDone !== null) {
          resolveDone();
          resolveDone = null;
        }
      });
    };
    await server.start();
    const client = await connectTcp(HOST, server.port, 0, 0, parserMap());
    const started = performance.now();
    try {
      // fire-and-forget를 순차 await하여 등록 순서를 보존한다(writeQueue backpressure도 함께 검증).
      for (let i = 0; i < count; i += 1) {
        await client.communicator.send(create(TestDataSchema, { index: i, message: `b-${i}` }));
      }
      await withTimeout(done, mode === "quick" ? 10_000 : 60_000, `burst ${count} dispatch timeout`);
      const elapsed = performance.now() - started;
      if (received !== count) {
        stability.pendingLeak += 1;
      }
      emitThroughputRow("burst", `count=${count}`, count, elapsed, stability, "off", memMb());
    } catch (err) {
      stability.timeouts += 1;
      log(`[burst] count=${count} error=${errorMessage(err)}`);
      emitThroughputRow("burst", `count=${count}`, received, performance.now() - started, stability, "off", memMb());
    } finally {
      await client.close();
      await server.stop();
      if (!leakClear(client)) {
        stability.pendingLeak += 1;
      }
    }
    log(`[burst] count=${count} received=${received} violations=${stabilityViolations(stability)}`);
  }
}

async function profileSustained(mode: Mode): Promise<void> {
  const durationMs = mode === "quick" ? 2_000 : 30_000;
  const concurrency = 16;
  const timeoutMs = 15_000;
  log(`[sustained] mode=${mode} duration=${durationMs}ms concurrency=${concurrency}`);

  const stability = newStability();
  let peakMem = memMb();
  await withRequestServer(stability, async (port) => {
    const client = await connectTcp(HOST, port, 0, 0, parserMap());
    const latencies: number[] = [];
    let nextIndex = 0;
    const deadline = performance.now() + durationMs;
    try {
      while (performance.now() < deadline) {
        const batch: Array<Promise<void>> = [];
        for (let i = 0; i < concurrency; i += 1) {
          const index = nextIndex;
          nextIndex += 1;
          const started = performance.now();
          batch.push(
            client.communicator
              .sendRequest(
                create(TestDataSchema, { index, message: `s-${index}` }),
                TestDataSchema,
                timeoutMs,
              )
              .then((res) => {
                latencies.push(performance.now() - started);
                if (res.index !== index * 2 || res.message !== `echo:s-${index}`) {
                  stability.nonceMismatch += 1;
                }
              })
              .catch((err: unknown) => {
                classifyRequestError(errorMessage(err), stability);
              }),
          );
        }
        await Promise.all(batch);
        const current = memMb();
        if (current > peakMem) {
          peakMem = current;
        }
      }
      const elapsed = durationMs;
      summarizeLatencies("sustained", `duration=${durationMs}ms`, latencies, elapsed, stability, "off", peakMem);
    } finally {
      await client.close();
      if (!leakClear(client)) {
        stability.pendingLeak += 1;
      }
    }
  });
  log(`[sustained] done peakMemMb=${peakMem.toFixed(1)} violations=${stabilityViolations(stability)}`);
}

async function profileParallel(mode: Mode): Promise<void> {
  const clientCount = mode === "quick" ? 4 : 16;
  const perClient = mode === "quick" ? 50 : 500;
  const concurrency = 8;
  const timeoutMs = 15_000;
  log(`[parallel] mode=${mode} clients=${clientCount} perClient=${perClient}`);

  const stability = newStability();
  await withRequestServer(stability, async (port) => {
    const clients = await Promise.all(
      Array.from({ length: clientCount }, () => connectTcp(HOST, port, 0, 0, parserMap())),
    );
    const allLatencies: number[] = [];
    const started = performance.now();
    try {
      await Promise.all(
        clients.map(async (client) => {
          const latencies = await runRequestLoad(client, perClient, concurrency, timeoutMs, stability);
          allLatencies.push(...latencies);
        }),
      );
      const elapsed = performance.now() - started;
      summarizeLatencies(
        "parallel",
        `clients=${clientCount}`,
        allLatencies,
        elapsed,
        stability,
        "off",
        memMb(),
      );
    } finally {
      await Promise.all(clients.map(async (client) => client.close()));
      for (const client of clients) {
        if (!leakClear(client)) {
          stability.pendingLeak += 1;
        }
      }
    }
  });
  log(`[parallel] done violations=${stabilityViolations(stability)}`);
}

const NOOP_TRANSPORT: Transport = {
  writePacket: async () => {},
  close: async () => {},
};

/**
 * gateway on/off baseline. communicator의 `onReceivedFrame` opt-in 경로에 raw PacketBase frame을 흘려
 * dispatch 순서/throughput을 비교한다. off=inline decode, on=worker_threads decode pool.
 * 작은 payload의 in-process 측정이므로 성능 이득이 불명확할 수 있고, 안정성(순서/leak) 0 위반만 합격선으로 둔다.
 */
async function profileGateway(mode: Mode): Promise<void> {
  const count = mode === "quick" ? 500 : 5_000;
  log(`[gateway] mode=${mode} frames=${count}`);

  const frames: Uint8Array[] = [];
  for (let i = 0; i < count; i += 1) {
    const base = create(PacketBaseSchema, {
      typeName: TestDataSchema.typeName,
      nonce: i + 1,
      data: toBinary(TestDataSchema, create(TestDataSchema, { index: i, message: `g-${i}` })),
    });
    frames.push(toBinary(PacketBaseSchema, base));
  }

  for (const gatewayOn of [false, true]) {
    const stability = newStability();
    const comm = new Communicator();
    comm.initialize(NOOP_TRANSPORT, parserMap());
    comm.setFrameErrorHandler(() => {
      stability.typeMismatch += 1;
    });

    let dispatched = 0;
    let lastIndex = -1;
    let resolveDone: (() => void) | null = null;
    const done = new Promise<void>((resolve) => {
      resolveDone = resolve;
    });
    addListenerTyped(comm, TestDataSchema, (msg) => {
      if (msg.index <= lastIndex) {
        stability.fifoViolations += 1;
      }
      lastIndex = msg.index;
      dispatched += 1;
      if (dispatched >= count && resolveDone !== null) {
        resolveDone();
        resolveDone = null;
      }
    });

    if (gatewayOn) {
      const gateway = createNodeWorkerGateway({
        sink: (frame) =>
          comm.enqueueInbound(frame.typeName, frame.data, frame.incomingNonce, frame.responseNonce),
        onError: () => {
          stability.typeMismatch += 1;
        },
        workers: Math.max(2, Math.min(4, availableParallelism())),
      });
      comm.attachInboundGateway(gateway);
    }

    const started = performance.now();
    try {
      for (const raw of frames) {
        comm.onReceivedFrame(raw);
      }
      await withTimeout(done, mode === "quick" ? 15_000 : 60_000, `gateway dispatch timeout`);
      const elapsed = performance.now() - started;
      if (dispatched !== count) {
        stability.pendingLeak += 1;
      }
      emitThroughputRow("gateway", `frames=${count}`, count, elapsed, stability, gatewayOn ? "on" : "off", memMb());
    } catch (err) {
      stability.timeouts += 1;
      log(`[gateway] on=${gatewayOn} error=${errorMessage(err)}`);
      emitThroughputRow(
        "gateway",
        `frames=${count}`,
        dispatched,
        performance.now() - started,
        stability,
        gatewayOn ? "on" : "off",
        memMb(),
      );
    } finally {
      await comm.close();
    }
    log(`[gateway] on=${gatewayOn} dispatched=${dispatched} violations=${stabilityViolations(stability)}`);
  }
}

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 parseArgs(argv: string[]): { mode: Mode; profiles: Profile[] } {
  let mode: Mode = "quick";
  let profiles: Profile[] = [...ALL_PROFILES];
  for (const arg of argv) {
    if (arg === "--quick") {
      mode = "quick";
    } else if (arg === "--full") {
      mode = "full";
    } else if (arg.startsWith("--mode=")) {
      const value = arg.slice("--mode=".length);
      mode = value === "full" ? "full" : "quick";
    } else if (arg.startsWith("--profiles=") || arg.startsWith("--profile=")) {
      const value = arg.slice(arg.indexOf("=") + 1);
      const requested = value
        .split(",")
        .map((p) => p.trim())
        .filter((p) => p.length > 0);
      const selected = requested.filter((p): p is Profile => (ALL_PROFILES as readonly string[]).includes(p));
      if (selected.length > 0) {
        profiles = selected;
      }
    }
  }
  return { mode, profiles };
}

async function main(): Promise<void> {
  const { mode, profiles } = parseArgs(process.argv.slice(2));
  log(`INFO stress harness mode=${mode} profiles=${profiles.join(",")} typeName=${TestDataSchema.typeName}`);

  const runners: Record<Profile, (mode: Mode) => Promise<void>> = {
    roundtrip: profileRoundtrip,
    burst: profileBurst,
    sustained: profileSustained,
    parallel: profileParallel,
    gateway: profileGateway,
  };

  for (const profile of profiles) {
    await runners[profile](mode);
  }

  const totalViolations = rows.reduce((acc, row) => acc + stabilityViolations(row.stability), 0);
  const status = totalViolations === 0 ? "PASS" : "FAIL";
  process.stdout.write(
    `SUMMARY|status=${status}|mode=${mode}|profiles=${profiles.join(",")}|rows=${rows.length}|stability_violations=${totalViolations}\n`,
  );
  if (status !== "PASS") {
    process.exitCode = 1;
  }
}

void main().catch((err: unknown) => {
  process.stdout.write(`SUMMARY|status=FAIL|error=${errorMessage(err)}\n`);
  log(`FATAL ${errorMessage(err)}`);
  process.exitCode = 1;
});