import {
  create,
  fromBinary,
  HeartBeatSchema,
  type Message,
  type MessageType,
  PacketBaseSchema,
  toBinary,
  type PacketBase,
} from "./packets/message_common_pb.js";
import {
  decodePacketBaseEnvelope,
  type FrameDecoder,
  type InboundGateway,
  WorkerGateway,
} from "./inbound_gateway.js";

export type { Message, MessageType } from "./packets/message_common_pb.js";
export {
  decodePacketBaseEnvelope,
  type DecodedEnvelope,
  type DecodedFrame,
  type FrameDecoder,
  FrameReorderBuffer,
  type InboundFrame,
  type InboundGateway,
  type InboundGatewayStats,
  WorkerGateway,
  type WorkerGatewayOptions,
} from "./inbound_gateway.js";

export class NotConnectedError extends Error {
  constructor() {
    super("not connected");
    this.name = "NotConnectedError";
  }
}

export type MessageParser<T extends Message = Message> = ((data: Uint8Array) => T) & {
  schema?: MessageType<T>;
};

export type ParserMap = Map<string, MessageParser>;

export interface Transport {
  writePacket(base: PacketBase): Promise<void>;
  close(): Promise<void>;
}

type RequestHandler = (
  msg: Message,
  nonce: number,
) => Message | Promise<Message | null | undefined> | null | undefined | void;

interface PendingRequest {
  expectedTypeName: string;
  timer: ReturnType<typeof setTimeout> | null;
  resolve: (msg: Message) => void;
  reject: (err: Error) => void;
}

interface QueueItem {
  base: PacketBase;
  resolve: () => void;
  reject: (err: Error) => void;
}

interface InboundItem {
  typeName: string;
  data: Uint8Array;
  nonce: number;
  responseNonce: number;
}

function getShortTypeName(typeName: string): string {
  const lastDot = typeName.lastIndexOf(".");
  if (lastDot < 0 || lastDot === typeName.length - 1) {
    return typeName;
  }
  return typeName.substring(lastDot + 1);
}

export function parserFromSchema<T extends Message>(schema: MessageType<T>): MessageParser<T> {
  const parser = ((data: Uint8Array) => fromBinary(schema, data)) as MessageParser<T>;
  parser.schema = schema;
  return parser;
}

export class Communicator {
  private static readonly MAX_WRITE_QUEUE_SIZE = 64;
  private static readonly MAX_NONCE = 2_147_483_647;

  private nonce = 0;
  private isAliveFlag = false;
  private parserMap: ParserMap = new Map();
  private schemaMap = new Map<string, MessageType>();
  private canonicalNameMap = new Map<string, string>();
  private handlers = new Map<string, Array<(msg: Message) => void>>();
  private reqHandlers = new Map<string, RequestHandler>();
  private pendingRequests = new Map<number, PendingRequest>();
  private writeQueue: QueueItem[] = [];
  private writeQueueWaiters: Array<() => void> = [];
  private writeQueueRunning = false;
  private inboundQueue: InboundItem[] = [];
  private inboundQueueRunning = false;
  private inboundQueueWaiters: Array<() => void> = [];
  private isClosing = false;
  private inFlightCount = 0;
  private closeResolver: (() => void) | null = null;
  private transport: Transport | null = null;
  private writeErrorHandler: ((err: Error) => void) | null = null;
  private frameErrorHandler: ((err: Error) => void) | null = null;
  private gateway: InboundGateway | null = null;
  private frameSeq = 0;
  private closedPromise: Promise<void> = Promise.resolve();
  private resolveClosed: (() => void) | null = null;

  private canonicalize(typeName: string): string {
    return this.canonicalNameMap.get(typeName) ??
      this.canonicalNameMap.get(getShortTypeName(typeName)) ??
      typeName;
  }

  initialize(transport: Transport, parserMap: ParserMap): void {
    this.transport = transport;
    this.closeGateway();
    this.frameSeq = 0;
    this.frameErrorHandler = null;
    this.parserMap = new Map();
    this.schemaMap = new Map();
    this.handlers = new Map();
    this.reqHandlers = new Map();
    this.pendingRequests = new Map();
    this.writeQueue = [];
    this.writeQueueWaiters = [];
    this.writeQueueRunning = false;
    this.inboundQueue = [];
    this.inboundQueueRunning = false;
    this.inFlightCount = 0;
    this.writeErrorHandler = null;
    this.nonce = 0;
    this.isAliveFlag = true;
    this.resetClosedPromise();

    const tempParsers = new Map<string, MessageParser>();
    for (const [typeName, parser] of parserMap) {
      tempParsers.set(typeName, parser);
    }
    const heartBeatParser = parserFromSchema(HeartBeatSchema);
    tempParsers.set(HeartBeatSchema.typeName, heartBeatParser);

    this.canonicalNameMap = new Map();
    const shortToFull = new Map<string, string>();
    for (const fullName of tempParsers.keys()) {
      const shortName = getShortTypeName(fullName);
      const existing = shortToFull.get(shortName);
      if (existing !== undefined && existing !== fullName) {
        throw new Error(`Duplicate alias mapping: ${shortName} maps to both ${existing} and ${fullName}`);
      }
      shortToFull.set(shortName, fullName);
    }

    for (const fullName of tempParsers.keys()) {
      this.canonicalNameMap.set(fullName, fullName);
      const shortName = getShortTypeName(fullName);
      this.canonicalNameMap.set(shortName, fullName);
    }

    for (const [typeName, parser] of tempParsers) {
      const canonical = this.canonicalize(typeName);
      this.parserMap.set(canonical, parser);
      if (parser.schema !== undefined) {
        this.schemaMap.set(canonical, parser.schema);
      }
    }
  }

  isAlive(): boolean {
    return this.isAliveFlag;
  }

  waitClosed(): Promise<void> {
    return this.closedPromise;
  }

  setWriteErrorHandler(fn: (err: Error) => void): void {
    this.writeErrorHandler = fn;
  }

  /**
   * inbound gateway가 raw frame decode에 실패했을 때 호출되는 handler를 등록한다. transport는 이를 통해
   * gateway path에서도 inline path와 동일한 parse-error disconnect semantics를 적용할 수 있다(write path의
   * {@link setWriteErrorHandler}와 대응). gateway가 없으면 {@link onReceivedFrame}의 inline decode 오류도
   * 이 handler로 보고된다.
   */
  setFrameErrorHandler(fn: (err: Error) => void): void {
    this.frameErrorHandler = fn;
  }

  /**
   * receive coordinator 앞단에 default receive gateway를 attach한다. 기본 gateway는 worker_threads hop 없이
   * raw frame에 내부 `seq`를 부여하고 inline decode 결과를 reorder/sink serialization으로 coordinator에 전달한다.
   * FIFO dispatch와 stateful 처리 소유권은 coordinator에 유지된다.
   *
   * opt-in이며 attach하지 않으면 {@link onReceivedFrame}은 호출 컨텍스트에서 inline decode한다. Node는
   * `worker_threads` 기반 `decode`를, browser는 Web Worker 기반 `decode`를 주입할 수 있고, 미주입 시 inline
   * decode가 safe fallback이 된다.
   */
  enableInboundGateway(
    options: { workers?: number; decode?: FrameDecoder } = {},
  ): void {
    this.attachInboundGateway(
      new WorkerGateway({
        workers: options.workers,
        decode: options.decode,
        sink: (frame) =>
          this.enqueueInbound(
            frame.typeName,
            frame.data,
            frame.incomingNonce,
            frame.responseNonce,
          ),
        // gateway decode 오류를 frame error handler로 노출해 inline path와 동일한 parse-error
        // disconnect semantics를 적용한다. gateway 자체는 transport를 종료하지 않는다.
        onError: (err) => this.frameErrorHandler?.(err),
      }),
    );
  }

  /**
   * custom {@link InboundGateway}를 receive coordinator 앞단에 설치한다. gateway sink는 coordinator 소유권을
   * 보존하기 위해 decoded frame을 {@link enqueueInbound}로 전달해야 한다. default receive gateway는
   * {@link enableInboundGateway}를 사용한다.
   */
  attachInboundGateway(gateway: InboundGateway): void {
    this.closeGateway();
    this.gateway = gateway;
    this.frameSeq = 0;
  }

  private closeGateway(): void {
    if (this.gateway !== null) {
      this.gateway.close();
      this.gateway = null;
    }
  }

  /**
   * transport read loop가 raw `PacketBase` frame을 ingest한다.
   *
   * gateway가 attach되어 있으면 frame에 내부 `seq`를 부여해 off-coordinator decode + reorder로 제출한다.
   * decode에 실패한 frame은 `seq` 순서에 맞춰 {@link setFrameErrorHandler}에 등록된 handler로 보고된다.
   *
   * gateway가 없으면 호출 컨텍스트에서 inline decode 후 {@link enqueueInbound}로 전달한다(safe fallback).
   */
  onReceivedFrame(raw: Uint8Array): void | Promise<void> {
    if (!this.isAliveFlag || this.isClosing) {
      return;
    }
    const gateway = this.gateway;
    if (gateway !== null) {
      this.frameSeq += 1;
      gateway.submit({ seq: this.frameSeq, bytes: raw });
      return;
    }
    let envelope;
    try {
      envelope = decodePacketBaseEnvelope(raw);
    } catch (err) {
      this.frameErrorHandler?.(toError(err));
      return;
    }
    // inline fallback에서는 enqueue Promise를 그대로 돌려준다. transport read loop가 이를 await해
    // 기존 inline decode 경로와 동일한 inbound queue backpressure(pause/resume)를 유지한다. gateway가
    // attach된 경로는 위에서 submit 후 즉시 반환되므로 decode가 read loop를 막지 않는다.
    return this.enqueueInbound(
      envelope.typeName,
      envelope.data,
      envelope.incomingNonce,
      envelope.responseNonce,
    );
  }

  private nextNonce(): number {
    let attempts = 0;
    while (attempts < Communicator.MAX_NONCE) {
      if (this.nonce >= Communicator.MAX_NONCE) {
        this.nonce = 0;
      }
      this.nonce += 1;
      if (!this.pendingRequests.has(this.nonce)) {
        return this.nonce;
      }
      attempts += 1;
    }
    throw new Error("All nonces are currently pending");
  }

  shutdown(): void {
    if (!this.isAliveFlag) {
      this.finishClosed();
      return;
    }

    this.isAliveFlag = false;
    this.isClosing = false;
    this.closeGateway();
    const pendingRequests = Array.from(this.pendingRequests.values());
    this.pendingRequests.clear();
    for (const pending of pendingRequests) {
      if (pending.timer !== null) {
        clearTimeout(pending.timer);
      }
      pending.reject(new NotConnectedError());
    }

    while (this.writeQueue.length > 0) {
      this.writeQueue.shift()?.reject(new NotConnectedError());
    }

    while (this.writeQueueWaiters.length > 0) {
      this.writeQueueWaiters.shift()?.();
    }

    this.inboundQueue = [];
    const waiters = [...this.inboundQueueWaiters];
    this.inboundQueueWaiters = [];
    for (const waiter of waiters) {
      waiter();
    }

    if (this.closeResolver !== null) {
      const resolve = this.closeResolver;
      this.closeResolver = null;
      resolve();
    }
    this.finishClosed();
  }

  async close(): Promise<void> {
    if (!this.isAliveFlag || this.isClosing) {
      return;
    }
    this.isClosing = true;
    this.closeGateway();
    if (this.inboundQueue.length > 0 || this.inFlightCount > 0) {
      await new Promise<void>((resolve) => {
        this.closeResolver = resolve;
      });
    }

    this.isAliveFlag = false;
    this.isClosing = false;

    const pendingRequests = Array.from(this.pendingRequests.values());
    this.pendingRequests.clear();
    for (const pending of pendingRequests) {
      if (pending.timer !== null) {
        clearTimeout(pending.timer);
      }
      pending.reject(new NotConnectedError());
    }

    while (this.writeQueue.length > 0) {
      this.writeQueue.shift()?.reject(new NotConnectedError());
    }

    while (this.writeQueueWaiters.length > 0) {
      this.writeQueueWaiters.shift()?.();
    }

    this.finishClosed();

    if (this.transport !== null) {
      await this.transport.close();
    }
  }

  async queuePacket(base: PacketBase): Promise<void> {
    if (!this.isAliveFlag || this.transport === null) {
      throw new NotConnectedError();
    }

    while (this.writeQueue.length >= Communicator.MAX_WRITE_QUEUE_SIZE) {
      await new Promise<void>((resolve) => {
        this.writeQueueWaiters.push(resolve);
      });
      if (!this.isAliveFlag || this.transport === null) {
        throw new NotConnectedError();
      }
    }

    const promise = new Promise<void>((resolve, reject) => {
      this.writeQueue.push({ base, resolve, reject });
    });
    if (!this.writeQueueRunning) {
      void this.runWriteQueue();
    }
    return promise;
  }

  async send(msg: Message): Promise<void> {
    if (!this.isAliveFlag) {
      throw new NotConnectedError();
    }

    const encoded = this.encodeMessage(msg);
    await this.queuePacket(
      create(PacketBaseSchema, {
        typeName: encoded.typeName,
        nonce: this.nextNonce(),
        data: encoded.data,
      }),
    );
  }

  async sendRequest<TRes extends Message>(
    req: Message,
    resType: MessageType<TRes>,
    timeoutMs = 30000,
  ): Promise<TRes> {
    if (!this.isAliveFlag) {
      throw new NotConnectedError();
    }

    const requestNonce = this.nextNonce();
    const encoded = this.encodeMessage(req);
    const effectiveTimeout = timeoutMs > 0 ? timeoutMs : 30000;

    return new Promise<TRes>((resolve, reject) => {
      // pending은 write 전에 등록한다. loopback에서 응답이 write resolve보다 먼저 도착할 수 있어야 하기
      // 때문이다. response timeout timer는 packet이 실제로 transport에 write된 뒤에만 시작한다. write
      // queue backpressure 대기 시간이 response timeout으로 집계되지 않게 하기 위함이다.
      this.pendingRequests.set(requestNonce, {
        expectedTypeName: resType.typeName,
        timer: null,
        resolve: (msg) => resolve(msg as TRes),
        reject,
      });

      void this.queuePacket(
        create(PacketBaseSchema, {
          typeName: encoded.typeName,
          nonce: requestNonce,
          data: encoded.data,
        }),
      ).then(
        () => {
          // write 완료 시점에 pending이 이미 응답/종료로 제거되었으면 timer를 걸지 않는다.
          const pending = this.pendingRequests.get(requestNonce);
          if (pending === undefined) {
            return;
          }
          pending.timer = setTimeout(() => {
            const timedOut = this.removePending(requestNonce);
            if (timedOut !== undefined) {
              timedOut.reject(new Error(`request timeout for nonce ${requestNonce}`));
            }
          }, effectiveTimeout);
        },
        (err: unknown) => {
          const pending = this.removePending(requestNonce);
          if (pending !== undefined) {
            pending.reject(toError(err));
          }
        },
      );
    });
  }

  addListener(typeName: string, fn: (msg: Message) => void): void {
    const canonical = this.canonicalize(typeName);
    if (this.reqHandlers.has(canonical)) {
      throw new Error(`type ${typeName} is already registered with addRequestListener`);
    }
    const handlers = this.handlers.get(canonical) ?? [];
    handlers.push(fn);
    this.handlers.set(canonical, handlers);
  }

  removeListeners(typeName: string): void {
    const canonical = this.canonicalize(typeName);
    this.handlers.delete(canonical);
  }

  addRequestListener(typeName: string, fn: RequestHandler): void {
    const canonical = this.canonicalize(typeName);
    if ((this.handlers.get(canonical) ?? []).length > 0) {
      throw new Error(`type ${typeName} is already registered with addListener`);
    }
    if (this.reqHandlers.has(canonical)) {
      throw new Error(`type ${typeName} is already registered with addRequestListener`);
    }
    this.reqHandlers.set(canonical, fn);
  }

  onReceivedData(typeName: string, data: Uint8Array, nonce = 0, responseNonce = 0): void {
    void this.enqueueInbound(typeName, data, nonce, responseNonce);
  }

  async enqueueInbound(typeName: string, data: Uint8Array, nonce = 0, responseNonce = 0): Promise<void> {
    if (!this.isAliveFlag || this.isClosing) {
      return;
    }
    if (responseNonce > 0) {
      this.handleResponse(typeName, data, responseNonce);
      return;
    }
    while (this.inboundQueue.length >= 64) {
      await new Promise<void>((resolve) => {
        this.inboundQueueWaiters.push(resolve);
      });
    }
    if (!this.isAliveFlag || this.isClosing) {
      return;
    }
    this.inboundQueue.push({ typeName, data, nonce, responseNonce });
    if (!this.inboundQueueRunning) {
      void this.runInboundQueue();
    }
  }

  private async runInboundQueue(): Promise<void> {
    if (this.inboundQueueRunning) {
      return;
    }
    this.inboundQueueRunning = true;
    try {
      while (this.inboundQueue.length > 0) {
        const item = this.inboundQueue.shift();
        if (item === undefined) {
          continue;
        }
        if (this.inboundQueue.length < 64 && this.inboundQueueWaiters.length > 0) {
          const waiter = this.inboundQueueWaiters.shift();
          if (waiter !== undefined) {
            waiter();
          }
        }
        await this.dispatchInbound(item);
      }
    } finally {
      this.inboundQueueRunning = false;
      if (this.inboundQueue.length > 0 && this.isAliveFlag) {
        void this.runInboundQueue();
      }
      this.checkCloseResolver();
    }
  }

  private checkCloseResolver(): void {
    if (this.inboundQueue.length === 0 && this.inFlightCount === 0 && this.closeResolver !== null) {
      const resolve = this.closeResolver;
      this.closeResolver = null;
      resolve();
    }
  }

  private async dispatchInbound(item: InboundItem): Promise<void> {
    this.inFlightCount++;
    try {
      const { typeName, data, nonce, responseNonce } = item;
      if (responseNonce > 0) {
        this.handleResponse(typeName, data, responseNonce);
        return;
      }

      const canonical = this.canonicalize(typeName);
      const reqHandler = this.reqHandlers.get(canonical);
      const listeners = [...(this.handlers.get(canonical) ?? [])];

      if (reqHandler !== undefined) {
        let message: Message;
        try {
          message = this.parse(typeName, data);
        } catch {
          return;
        }
        await this.runRequestHandler(reqHandler, message, nonce);
        return;
      }

      if (listeners.length === 0) {
        return;
      }

      let message: Message;
      try {
        message = this.parse(typeName, data);
      } catch {
        return;
      }
      for (const listener of listeners) {
        listener(message);
      }
    } finally {
      this.inFlightCount--;
      this.checkCloseResolver();
    }
  }

  private async runRequestHandler(
    handler: RequestHandler,
    message: Message,
    requestNonce: number,
  ): Promise<void> {
    try {
      const result = await handler(message, requestNonce);
      if (result == null || !this.isAliveFlag) {
        return;
      }

      const encoded = this.encodeMessage(result);
      await this.queuePacket(
        create(PacketBaseSchema, {
          typeName: encoded.typeName,
          nonce: this.nextNonce(),
          responseNonce: requestNonce,
          data: encoded.data,
        }),
      );
    } catch {
      return;
    }
  }

  private handleResponse(typeName: string, data: Uint8Array, responseNonce: number): void {
    const pending = this.removePending(responseNonce);
    if (pending === undefined) {
      return;
    }
    const canonicalExpected = this.canonicalize(pending.expectedTypeName);
    const canonicalReceived = this.canonicalize(typeName);
    if (canonicalReceived !== canonicalExpected) {
      pending.reject(
        new Error(
          `response type mismatch for nonce ${responseNonce}: expected ${canonicalExpected}, got ${typeName}`,
        ),
      );
      return;
    }

    try {
      pending.resolve(this.parse(typeName, data));
    } catch (err) {
      pending.reject(toError(err));
    }
  }

  private parse(typeName: string, data: Uint8Array): Message {
    const canonical = this.canonicalize(typeName);
    const parser = this.parserMap.get(canonical);
    if (parser === undefined) {
      throw new Error(`protobuf parser is not registered for type ${typeName}`);
    }
    return parser(data);
  }

  private removePending(nonce: number): PendingRequest | undefined {
    const pending = this.pendingRequests.get(nonce);
    if (pending === undefined) {
      return undefined;
    }
    if (pending.timer !== null) {
      clearTimeout(pending.timer);
    }
    this.pendingRequests.delete(nonce);
    return pending;
  }

  private encodeMessage(msg: Message): { typeName: string; data: Uint8Array } {
    const typeName = typeNameOf(msg);
    const canonical = this.canonicalize(typeName);
    const schema = this.schemaMap.get(canonical);
    if (schema === undefined) {
      throw new Error(`protobuf schema is not registered for type ${typeName}`);
    }
    return {
      typeName,
      data: toBinary(schema, msg),
    };
  }

  private async runWriteQueue(): Promise<void> {
    if (this.writeQueueRunning) {
      return;
    }
    this.writeQueueRunning = true;

    try {
      while (this.writeQueue.length > 0) {
        const item = this.writeQueue.shift();
        if (item === undefined) {
          continue;
        }
        if (this.writeQueueWaiters.length > 0) {
          this.writeQueueWaiters.shift()?.();
        }
        if (!this.isAliveFlag || this.transport === null) {
          item.reject(new NotConnectedError());
          continue;
        }

        try {
          await this.transport.writePacket(item.base);
          item.resolve();
        } catch (err) {
          const error = toError(err);
          item.reject(error);
          this.writeErrorHandler?.(error);
        }
      }
    } finally {
      this.writeQueueRunning = false;
      if (this.writeQueue.length > 0 && this.isAliveFlag) {
        void this.runWriteQueue();
      }
    }
  }

  private resetClosedPromise(): void {
    this.closedPromise = new Promise<void>((resolve) => {
      this.resolveClosed = resolve;
    });
  }

  private finishClosed(): void {
    if (this.resolveClosed !== null) {
      this.resolveClosed();
      this.resolveClosed = null;
    }
  }
}

export function typeNameOf(msgOrType: Message | MessageType): string {
  if ("$typeName" in msgOrType) {
    return msgOrType.$typeName;
  }
  return msgOrType.typeName;
}

export function addListenerTyped<T extends Message>(
  comm: Communicator,
  msgType: MessageType<T>,
  fn: (msg: T) => void,
): void {
  comm.addListener(msgType.typeName, (msg) => fn(msg as T));
}

export function addRequestListenerTyped<TReq extends Message, TRes extends Message>(
  comm: Communicator,
  reqType: MessageType<TReq>,
  fn: (req: TReq) => TRes | Promise<TRes | null | undefined> | null | undefined,
): void {
  comm.addRequestListener(reqType.typeName, (req) => fn(req as TReq));
}

export function sendRequestTyped<TReq extends Message, TRes extends Message>(
  comm: Communicator,
  req: TReq,
  resType: MessageType<TRes>,
  timeoutMs?: number,
): Promise<TRes> {
  return comm.sendRequest(req, resType, timeoutMs);
}

function toError(err: unknown): Error {
  return err instanceof Error ? err : new Error(String(err));
}
