import * as crypto from "node:crypto";
import { EventEmitter } from "node:events";
import * as net from "node:net";
import * as tls from "node:tls";
import WebSocketClient, { type ClientOptions as WebSocketClientOptions } from "ws";

import { BaseClient } from "./base_client.js";
import { type ParserMap } from "./communicator.js";
import { PacketBaseSchema, toBinary, type PacketBase } from "./packets/message_common_pb.js";

const WS_GUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11";
const MAX_WS_PAYLOAD_SIZE = 64 << 20;
const WEBSOCKET_HANDSHAKE_TIMEOUT_MS = 1_000;

export type NodeWssConnectOptions = tls.ConnectionOptions;

type NodeSocket = net.Socket | tls.TLSSocket;

class ReadBuffer {
  private chunks: Buffer[] = [];
  private headOffset = 0;
  private totalLength = 0;

  append(chunk: Buffer): void {
    if (chunk.byteLength === 0) return;
    this.chunks.push(chunk);
    this.totalLength += chunk.byteLength;
  }

  get length(): number {
    return this.totalLength;
  }

  peekBytes(dest: Uint8Array, destOffset: number, offset: number, len: number): void {
    if (offset + len > this.totalLength) {
      throw new Error("Out of bounds peek");
    }
    let bytesToSkip = this.headOffset + offset;
    let chunkIdx = 0;

    while (chunkIdx < this.chunks.length) {
      const chunk = this.chunks[chunkIdx]!;
      if (bytesToSkip < chunk.byteLength) {
        break;
      }
      bytesToSkip -= chunk.byteLength;
      chunkIdx++;
    }

    let written = 0;
    while (written < len && chunkIdx < this.chunks.length) {
      const chunk = this.chunks[chunkIdx]!;
      const chunkAvail = chunk.byteLength - bytesToSkip;
      const toCopy = Math.min(len - written, chunkAvail);
      chunk.copy(dest, destOffset + written, bytesToSkip, bytesToSkip + toCopy);
      written += toCopy;
      bytesToSkip = 0;
      chunkIdx++;
    }
  }

  peek(offset: number): number {
    const virtualOffset = this.headOffset + offset;
    const firstChunk = this.chunks[0];
    if (firstChunk && virtualOffset < firstChunk.byteLength) {
      return firstChunk[virtualOffset]!;
    }
    return this.peekSlow(offset);
  }

  private peekSlow(offset: number): number {
    if (offset >= this.totalLength) {
      throw new Error("Out of bounds peek");
    }
    let bytesToSkip = this.headOffset + offset;
    for (let i = 0; i < this.chunks.length; i++) {
      const chunk = this.chunks[i]!;
      if (bytesToSkip < chunk.byteLength) {
        return chunk[bytesToSkip]!;
      }
      bytesToSkip -= chunk.byteLength;
    }
    throw new Error("Out of bounds peek");
  }

  readUInt16BE(offset: number): number {
    const virtualOffset = this.headOffset + offset;
    const firstChunk = this.chunks[0];
    if (firstChunk && virtualOffset + 2 <= firstChunk.byteLength) {
      return firstChunk.readUInt16BE(virtualOffset);
    }
    const temp = Buffer.allocUnsafe(2);
    this.peekBytes(temp, 0, offset, 2);
    return temp.readUInt16BE(0);
  }

  readBigUInt64BE(offset: number): bigint {
    const virtualOffset = this.headOffset + offset;
    const firstChunk = this.chunks[0];
    if (firstChunk && virtualOffset + 8 <= firstChunk.byteLength) {
      return firstChunk.readBigUInt64BE(virtualOffset);
    }
    const temp = Buffer.allocUnsafe(8);
    this.peekBytes(temp, 0, offset, 8);
    return temp.readBigUInt64BE(0);
  }

  take(len: number): Buffer {
    if (len === 0) {
      return Buffer.alloc(0);
    }
    if (len > this.totalLength) {
      throw new Error("Not enough bytes in buffer");
    }

    const firstChunk = this.chunks[0]!;
    const firstAvail = firstChunk.byteLength - this.headOffset;

    let result: Buffer;
    if (firstAvail >= len) {
      result = firstChunk.subarray(this.headOffset, this.headOffset + len);
      this.skip(len);
    } else {
      result = Buffer.allocUnsafe(len);
      this.peekBytes(result, 0, 0, len);
      this.skip(len);
    }

    return result;
  }

  skip(len: number): void {
    if (len > this.totalLength) {
      throw new Error("Cannot skip more than total length");
    }
    this.totalLength -= len;
    while (len > 0) {
      const firstChunk = this.chunks[0]!;
      const firstAvail = firstChunk.byteLength - this.headOffset;
      if (firstAvail > len) {
        this.headOffset += len;
        len = 0;
      } else {
        len -= firstAvail;
        this.chunks.shift();
        this.headOffset = 0;
      }
    }
  }
}

export class NodeWebSocket {
  static readonly CONNECTING = 0;
  static readonly OPEN = 1;
  static readonly CLOSING = 2;
  static readonly CLOSED = 3;

  readyState = NodeWebSocket.OPEN;

  private readonly events = new EventEmitter();
  private readBuf = new ReadBuffer();
  private fragments: Buffer[] = [];
  private fragmentOpcode: number | null = null;
  private closeEmitted = false;

  constructor(
    private readonly socket: NodeSocket,
    private readonly maskOutgoing: boolean,
    initialData: Buffer = Buffer.alloc(0),
  ) {
    this.events.on("error", () => {});

    // writeFrame은 header와 payload를 별도 write로 보낸다. Nagle's algorithm + delayed-ACK가
    // 겹치면 두 번째 작은 write가 ~40ms 지연되므로 no-delay를 명시해 즉시 flush한다.
    socket.setNoDelay?.(true);

    socket.on("data", (chunk) => {
      this.onData(chunk);
    });
    socket.on("error", (err) => {
      this.events.emit("error", err);
    });
    socket.once("close", () => {
      this.emitClose();
    });

    if (initialData.byteLength > 0) {
      queueMicrotask(() => this.onData(initialData));
    }
  }

  on(event: "message", listener: (data: Uint8Array) => void): this;
  on(event: "close", listener: () => void): this;
  on(event: "error", listener: (err: Error) => void): this;
  on(event: string, listener: (...args: any[]) => void): this {
    this.events.on(event, listener);
    return this;
  }

  once(event: "close", listener: () => void): this;
  once(event: "error", listener: (err: Error) => void): this;
  once(event: string, listener: (...args: any[]) => void): this {
    this.events.once(event, listener);
    return this;
  }

  off(event: "message", listener: (data: Uint8Array) => void): this;
  off(event: "close", listener: () => void): this;
  off(event: "error", listener: (err: Error) => void): this;
  off(event: string, listener: (...args: any[]) => void): this {
    this.events.off(event, listener);
    return this;
  }

  send(data: Uint8Array, callback?: (err?: Error | null) => void): void {
    if (this.readyState !== NodeWebSocket.OPEN) {
      callback?.(new Error("WebSocket is not open"));
      return;
    }
    this.writeFrame(0x2, data, callback);
  }

  close(): void {
    if (this.readyState === NodeWebSocket.CLOSED || this.readyState === NodeWebSocket.CLOSING) {
      return;
    }

    this.readyState = NodeWebSocket.CLOSING;
    this.writeFrame(0x8, Buffer.alloc(0), () => {
      this.socket.end();
    });
  }

  terminate(): void {
    this.readyState = NodeWebSocket.CLOSED;
    this.socket.destroy();
    this.emitClose();
  }

  pause(): void {
    this.socket.pause();
  }

  resume(): void {
    this.socket.resume();
  }

  private onData(chunk: Buffer): void {
    if (this.readyState === NodeWebSocket.CLOSED) {
      return;
    }

    this.readBuf.append(chunk);
    try {
      while (this.tryReadFrame()) {
        continue;
      }
    } catch (err) {
      this.events.emit("error", err instanceof Error ? err : new Error(String(err)));
      this.terminate();
    }
  }

  private tryReadFrame(): boolean {
    if (this.readBuf.length < 2) {
      return false;
    }

    const first = this.readBuf.peek(0);
    const second = this.readBuf.peek(1);
    const fin = (first & 0x80) !== 0;
    const opcode = first & 0x0f;
    const masked = (second & 0x80) !== 0;
    let payloadLength = second & 0x7f;
    let offset = 2;

    if (payloadLength === 126) {
      if (this.readBuf.length < offset + 2) {
        return false;
      }
      payloadLength = this.readBuf.readUInt16BE(offset);
      offset += 2;
    } else if (payloadLength === 127) {
      if (this.readBuf.length < offset + 8) {
        return false;
      }
      const longLength = this.readBuf.readBigUInt64BE(offset);
      if (longLength > BigInt(MAX_WS_PAYLOAD_SIZE)) {
        throw new Error("WebSocket payload exceeds maximum size");
      }
      payloadLength = Number(longLength);
      offset += 8;
    }

    const expectedMasked = !this.maskOutgoing;
    if (masked !== expectedMasked) {
      throw new Error("invalid WebSocket frame masking");
    }
    if (payloadLength > MAX_WS_PAYLOAD_SIZE) {
      throw new Error("WebSocket payload exceeds maximum size");
    }

    const maskOffset = masked ? 4 : 0;
    const frameLength = offset + maskOffset + payloadLength;
    if (this.readBuf.length < frameLength) {
      return false;
    }

    let mask: Buffer | null = null;
    if (masked) {
      mask = Buffer.allocUnsafe(4);
      this.readBuf.peekBytes(mask, 0, offset, 4);
    }

    this.readBuf.skip(offset + maskOffset);
    const payload = this.readBuf.take(payloadLength);
    if (mask !== null) {
      unmaskInPlace(payload, mask);
    }

    this.handleFrame(opcode, fin, payload);
    return true;
  }

  private handleFrame(opcode: number, fin: boolean, payload: Buffer): void {
    if (opcode >= 0x8) {
      if (!fin || payload.byteLength > 125) {
        throw new Error("invalid WebSocket control frame");
      }
      this.handleControlFrame(opcode, payload);
      return;
    }

    if (opcode === 0x0) {
      if (this.fragmentOpcode === null) {
        throw new Error("unexpected WebSocket continuation frame");
      }
      this.fragments.push(payload);
      if (fin) {
        const data = Buffer.concat(this.fragments);
        const fragmentOpcode = this.fragmentOpcode;
        this.fragments = [];
        this.fragmentOpcode = null;
        this.handleMessageFrame(fragmentOpcode, data);
      }
      return;
    }

    if (opcode !== 0x1 && opcode !== 0x2) {
      throw new Error(`unsupported WebSocket opcode: ${opcode}`);
    }

    if (!fin) {
      this.fragmentOpcode = opcode;
      this.fragments = [payload];
      return;
    }

    this.handleMessageFrame(opcode, payload);
  }

  private handleControlFrame(opcode: number, payload: Buffer): void {
    if (opcode === 0x8) {
      if (this.readyState === NodeWebSocket.OPEN) {
        this.readyState = NodeWebSocket.CLOSING;
        this.writeFrame(0x8, payload, () => {
          this.socket.end();
        });
      } else {
        this.socket.end();
      }
      return;
    }
    if (opcode === 0x9) {
      this.writeFrame(0xA, payload);
    }
  }

  private handleMessageFrame(opcode: number, payload: Buffer): void {
    if (opcode !== 0x2) {
      throw new Error("unsupported WebSocket text frame");
    }
    this.events.emit("message", payload);
  }

  private writeFrame(opcode: number, payload: Uint8Array, callback?: (err?: Error | null) => void): void {
    if (this.socket.destroyed) {
      callback?.(new Error("socket is closed"));
      return;
    }
    const { header, payload: encodedPayload } = encodeFrameParts(opcode, payload, this.maskOutgoing);
    this.socket.write(header);
    this.socket.write(encodedPayload, callback);
  }

  private emitClose(): void {
    if (this.closeEmitted) {
      return;
    }
    this.closeEmitted = true;
    this.readyState = NodeWebSocket.CLOSED;
    this.events.emit("close");
  }
}

export class NodeWsClient extends BaseClient {
  private readonly ws: NodeWebSocket;

  private parseQueue = Promise.resolve();

  constructor(ws: NodeWebSocket, intervalSec: number, waitSec: number, parserMap: ParserMap) {
    super(intervalSec, waitSec, () => closeWebSocket(ws));
    this.ws = ws;
    this.initBase(parserMap);

    ws.on("message", (data) => {
      if (typeof (ws as any).pause === "function") {
        (ws as any).pause();
      } else if ((ws as any)._socket && typeof (ws as any)._socket.pause === "function") {
        (ws as any)._socket.pause();
      }
      this.parseQueue = this.parseQueue.then(async () => {
        try {
          await this.onMessage(data);
        } catch {
          // ignore error to keep the chain alive
        } finally {
          if (typeof (ws as any).resume === "function") {
            (ws as any).resume();
          } else if ((ws as any)._socket && typeof (ws as any)._socket.resume === "function") {
            (ws as any)._socket.resume();
          }
        }
      });
    });
    ws.on("close", () => {
      void this.onDisconnected();
    });
    ws.on("error", () => {
      void this.onDisconnected();
    });
  }

  async writePacket(base: PacketBase): Promise<void> {
    const data = toBinary(PacketBaseSchema, base);
    await new Promise<void>((resolve, reject) => {
      this.ws.send(data, (err) => {
        if (err) {
          reject(err);
          return;
        }
        resolve();
      });
    });
  }

  private async onMessage(data: Uint8Array): Promise<void> {
    // raw PacketBase binary message를 communicator로 넘긴다. gateway 미설정 시 inline decode + enqueue를
    // await해 기존 backpressure를 유지하고, gateway 설정 시 off-thread decode 경로로 제출된다. decode
    // 실패는 setFrameErrorHandler(initBase)에서 onDisconnected로 처리한다.
    await this.communicator.onReceivedFrame(data);
    void this.sendHeartbeat();
  }
}

export async function connectNodeWs(
  host: string,
  port: number,
  path: string,
  intervalSec: number,
  waitSec: number,
  parserMap: ParserMap,
): Promise<NodeWsClient> {
  const socket = await connectPlainSocket(host, port);
  return connectNodeWebSocket(socket, host, port, path, intervalSec, waitSec, parserMap);
}

export async function connectNodeWss(
  host: string,
  port: number,
  path: string,
  wsOptions: NodeWssConnectOptions,
  intervalSec: number,
  waitSec: number,
  parserMap: ParserMap,
): Promise<NodeWsClient> {
  return connectPackageWebSocket(host, port, path, wsOptions, intervalSec, waitSec, parserMap);
}

export function createWebSocketAccept(key: string): string {
  return crypto.createHash("sha1").update(`${key}${WS_GUID}`).digest("base64");
}

async function connectPlainSocket(host: string, port: number): Promise<net.Socket> {
  return new Promise<net.Socket>((resolve, reject) => {
    const socket = net.createConnection({ host, port });
    const cleanup = () => {
      socket.off("connect", onConnect);
      socket.off("error", onError);
    };
    const onConnect = () => {
      cleanup();
      resolve(socket);
    };
    const onError = (err: Error) => {
      cleanup();
      reject(err);
    };

    socket.once("connect", onConnect);
    socket.once("error", onError);
  });
}

async function connectTlsSocket(
  host: string,
  port: number,
  options: tls.ConnectionOptions,
): Promise<tls.TLSSocket> {
  return new Promise<tls.TLSSocket>((resolve, reject) => {
    const servername = options.servername ?? (net.isIP(host) === 0 ? host : undefined);
    const socket = tls.connect({
      ...options,
      host,
      port,
      servername,
    });
    const cleanup = () => {
      socket.off("secureConnect", onSecureConnect);
      socket.off("error", onError);
    };
    const onSecureConnect = () => {
      cleanup();
      resolve(socket);
    };
    const onError = (err: Error) => {
      cleanup();
      reject(err);
    };

    socket.once("secureConnect", onSecureConnect);
    socket.once("error", onError);
  });
}

async function connectNodeWebSocket(
  socket: NodeSocket,
  host: string,
  port: number,
  path: string,
  intervalSec: number,
  waitSec: number,
  parserMap: ParserMap,
): Promise<NodeWsClient> {
  return new Promise<NodeWsClient>((resolve, reject) => {
    const key = crypto.randomBytes(16).toString("base64");
    let response = Buffer.alloc(0);

    const timer = setTimeout(() => {
      fail(new Error("websocket handshake timed out"));
    }, WEBSOCKET_HANDSHAKE_TIMEOUT_MS);
    const cleanup = () => {
      clearTimeout(timer);
      socket.off("data", onData);
      socket.off("error", onError);
      socket.off("close", onClose);
    };
    const fail = (err: Error) => {
      cleanup();
      socket.destroy();
      reject(err);
    };
    const onError = (err: Error) => {
      fail(err);
    };
    const onClose = () => {
      fail(new Error("socket closed during WebSocket handshake"));
    };
    const onData = (chunk: Buffer) => {
      response = Buffer.concat([response, chunk]);
      const headerEnd = response.indexOf("\r\n\r\n");
      if (headerEnd < 0) {
        return;
      }

      const header = response.subarray(0, headerEnd).toString("latin1");
      const remaining = response.subarray(headerEnd + 4);
      try {
        verifyHandshakeResponse(header, key);
      } catch (err) {
        fail(err instanceof Error ? err : new Error(String(err)));
        return;
      }

      cleanup();
      resolve(new NodeWsClient(new NodeWebSocket(socket, true, remaining), intervalSec, waitSec, parserMap));
    };

    socket.on("data", onData);
    socket.once("error", onError);
    socket.once("close", onClose);
    socket.write(buildHandshakeRequest(host, port, path, key));
  });
}

async function connectPackageWebSocket(
  host: string,
  port: number,
  path: string,
  options: tls.ConnectionOptions,
  intervalSec: number,
  waitSec: number,
  parserMap: ParserMap,
): Promise<NodeWsClient> {
  return new Promise<NodeWsClient>((resolve, reject) => {
    const requestPath = path.startsWith("/") ? path : `/${path}`;
    const socket = new WebSocketClient(`wss://${host}:${port}${requestPath}`, {
      ...options,
      servername: options.servername ?? (net.isIP(host) === 0 ? host : undefined),
    } as WebSocketClientOptions);

    const timer = setTimeout(() => {
      fail(new Error("websocket handshake timed out"));
    }, WEBSOCKET_HANDSHAKE_TIMEOUT_MS);
    const cleanup = () => {
      clearTimeout(timer);
      socket.off("error", onError);
      socket.off("open", onOpen);
    };
    const fail = (err: Error) => {
      cleanup();
      socket.on("error", () => {});
      try {
        socket.terminate();
      } catch {
        // The ws package throws if terminate is called before open completes.
      }
      reject(err);
    };
    const onError = (err: Error) => {
      cleanup();
      reject(err);
    };
    const onOpen = () => {
      cleanup();
      resolve(new NodeWsClient(socket as unknown as NodeWebSocket, intervalSec, waitSec, parserMap));
    };

    socket.once("error", onError);
    socket.once("open", onOpen);
  });
}

function buildHandshakeRequest(host: string, port: number, path: string, key: string): string {
  const requestPath = path.startsWith("/") ? path : `/${path}`;
  return [
    `GET ${requestPath} HTTP/1.1`,
    `Host: ${host}:${port}`,
    "Upgrade: websocket",
    "Connection: Upgrade",
    `Sec-WebSocket-Key: ${key}`,
    "Sec-WebSocket-Version: 13",
    "\r\n",
  ].join("\r\n");
}

function verifyHandshakeResponse(header: string, key: string): void {
  const lines = header.split("\r\n");
  const statusLine = lines.shift() ?? "";
  if (!/^HTTP\/1\.[01] 101(?: |$)/.test(statusLine)) {
    throw new Error(`invalid WebSocket handshake status: ${statusLine}`);
  }

  const headers = new Map<string, string>();
  for (const line of lines) {
    const index = line.indexOf(":");
    if (index <= 0) {
      continue;
    }
    headers.set(line.slice(0, index).trim().toLowerCase(), line.slice(index + 1).trim());
  }

  const accept = headers.get("sec-websocket-accept");
  if (accept !== createWebSocketAccept(key)) {
    throw new Error("invalid WebSocket handshake accept header");
  }
}

function encodeFrame(opcode: number, payload: Buffer, mask: boolean): Buffer {
  const { header, payload: encodedPayload } = encodeFrameParts(opcode, payload, mask);
  return Buffer.concat([header, encodedPayload]);
}

function encodeFrameParts(
  opcode: number,
  payload: Uint8Array,
  mask: boolean,
): { header: Buffer; payload: Uint8Array } {
  const payloadLength = payload.byteLength;
  const lengthBytes = payloadLength < 126 ? 0 : payloadLength <= 0xffff ? 2 : 8;
  const maskBytes = mask ? 4 : 0;
  const header = Buffer.allocUnsafe(2 + lengthBytes + maskBytes);
  let offset = 0;

  header[offset] = 0x80 | opcode;
  offset += 1;
  if (payloadLength < 126) {
    header[offset] = (mask ? 0x80 : 0) | payloadLength;
    offset += 1;
  } else if (payloadLength <= 0xffff) {
    header[offset] = (mask ? 0x80 : 0) | 126;
    offset += 1;
    header.writeUInt16BE(payloadLength, offset);
    offset += 2;
  } else {
    header[offset] = (mask ? 0x80 : 0) | 127;
    offset += 1;
    header.writeBigUInt64BE(BigInt(payloadLength), offset);
    offset += 8;
  }

  if (!mask) {
    return { header, payload };
  }

  const maskingKey = crypto.randomBytes(4);
  maskingKey.copy(header, offset);
  return { header, payload: maskPayload(payload, maskingKey) };
}

function xorInPlace(payload: Buffer, mask: Uint8Array): void {
  const m0 = mask[0]!;
  const m1 = mask[1]!;
  const m2 = mask[2]!;
  const m3 = mask[3]!;
  const len = payload.byteLength;
  let i = 0;
  for (; i + 3 < len; i += 4) {
    payload[i] ^= m0;
    payload[i + 1] ^= m1;
    payload[i + 2] ^= m2;
    payload[i + 3] ^= m3;
  }
  for (; i < len; i++) {
    payload[i] ^= mask[i & 3]!;
  }
}

function unmaskInPlace(payload: Buffer, mask: Buffer): void {
  xorInPlace(payload, mask);
}

function maskPayload(payload: Uint8Array, mask: Uint8Array): Buffer {
  const result = Buffer.from(payload);
  xorInPlace(result, mask);
  return result;
}

async function closeWebSocket(ws: NodeWebSocket): Promise<void> {
  if (ws.readyState === NodeWebSocket.CLOSED) {
    return;
  }

  await new Promise<void>((resolve) => {
    const cleanup = () => {
      clearTimeout(timer);
      ws.off("close", onDone);
      ws.off("error", onDone);
    };
    const onDone = () => {
      cleanup();
      resolve();
    };
    const timer = setTimeout(() => {
      ws.terminate();
      onDone();
    }, 1000);

    ws.once("close", onDone);
    ws.once("error", onDone);

    if (ws.readyState === NodeWebSocket.CLOSING) {
      return;
    }
    try {
      ws.close();
    } catch {
      ws.terminate();
      onDone();
    }
  });
}
