import { fromBinary, PacketBaseSchema } from "./packets/message_common_pb.js";

/**
 * 수신 측이 gateway 제출 전에 부여하는 내부 raw 프레임이다.
 *
 * `seq`는 단조 증가하는 내부 시퀀스로, 병렬 decode 후 입력 순서를 복원하는 용도로만 쓰인다.
 * wire format이나 공개 프로토콜 필드에는 절대 노출하지 않는다.
 */
export interface InboundFrame {
  seq: number;
  bytes: Uint8Array;
}

/** gateway decode 결과의 envelope 부분이다(순서 복원용 `seq` 제외). */
export interface DecodedEnvelope {
  typeName: string;
  data: Uint8Array;
  incomingNonce: number;
  responseNonce: number;
}

/**
 * gateway가 PacketBase envelope을 decode한 결과다. `Communicator`의 inbound dispatch가 기대하는
 * envelope 필드와, 입력 순서 복원을 위한 원본 `seq`를 함께 싣는다.
 */
export interface DecodedFrame extends DecodedEnvelope {
  seq: number;
}

/**
 * 수신 coordinator 앞단에서 순수 raw-bytes 작업(PacketBase envelope decode/경량 전처리)만 병렬
 * 처리하는 worker gateway 계약이다.
 *
 * gateway는 stateful dispatch, `pendingRequests`, listener, write queue를 소유하지 않는다. decode
 * 결과는 입력 `seq` 순서로 sink에 전달되어 coordinator가 결정적으로 dispatch할 수 있게 한다.
 */
/**
 * gateway cleanup/drain 진단용 backlog 스냅샷이다. close/cancel/drain이 실제로 자원을 비웠는지,
 * repeated run에서 backlog가 누적되지 않는지 숫자로 관측하기 위한 계측 계약이다.
 *
 * `queued`/`reorderPending`/`sinkPending`/`workerPending`는 close 후 모두 0이어야 하며, 0이 아니면
 * cleanup leak이다. `active`는 close 시점에 await 중이던 in-flight decode loop가 drain되는 동안 잠시
 * 남을 수 있으므로 backlog leak 판정에는 포함하지 않는다.
 */
export interface InboundGatewayStats {
  /** sink 전달 대기 중인(아직 decode loop가 집어가지 않은) 큐 길이다. */
  queued: number;
  /** 현재 in-flight인 decode loop 수다(최대 `workers`). */
  active: number;
  /** reorder 버퍼가 선행 `seq`를 기다리며 보류 중인 결과 수다. */
  reorderPending: number;
  /** sink chain에 스케줄됐지만 아직 전달되지 않은 결과 수다. */
  sinkPending: number;
  /** worker pool(off-thread decode)에 제출됐지만 아직 응답이 오지 않은 decode 수다. */
  workerPending: number;
  /** close()가 호출되어 추가 전달이 멈춘 상태인지 여부다. */
  closed: boolean;
}

export interface InboundGateway {
  /** raw 프레임을 gateway에 제출한다. fire-and-forget이며 decode 결과는 이후 sink에 `seq` 순서로 전달된다. */
  submit(frame: InboundFrame): void;
  /** gateway 자원을 해제하고 추가 전달을 멈춘다. */
  close(): void;
  /**
   * 현재 backlog/close 상태 스냅샷을 반환한다. cleanup/drain 계측용이며, 기존 custom gateway 구현과의
   * 호환을 위해 optional이다. 미구현 gateway는 진단 backlog를 0으로 간주한다.
   */
  stats?(): InboundGatewayStats;
}

/** raw 프레임을 PacketBase envelope으로 decode하는 순수 함수다. 공유 상태가 없어 병렬 실행에 안전하다. */
export type FrameDecoder = (bytes: Uint8Array) => DecodedEnvelope | Promise<DecodedEnvelope>;

/** raw bytes를 PacketBase envelope으로 inline decode한다. gateway 미설정 시의 safe fallback 경로다. */
export function decodePacketBaseEnvelope(bytes: Uint8Array): DecodedEnvelope {
  const base = fromBinary(PacketBaseSchema, bytes);
  return {
    typeName: base.typeName,
    data: base.data,
    incomingNonce: base.nonce,
    responseNonce: base.responseNonce,
  };
}

interface GatewayResult {
  seq: number;
  frame: DecodedFrame | null;
  error: Error | null;
}

/**
 * out-of-order로 도착할 수 있는 결과에서 입력 순서를 복원한다. `seq`는 1부터 연속이라고 가정한다.
 * 결과는 앞선 모든 `seq`가 release될 때까지 버퍼되며, 이후 자신과 연속된 후속 결과까지 순서대로 반환한다.
 * concurrent 사용에 안전하지 않으며 collector 단독에서만 호출한다.
 */
export class FrameReorderBuffer {
  private nextSeq = 1;
  private readonly pending = new Map<number, GatewayResult>();

  /** `res`를 버퍼하고 지금 순서대로 release 가능한 결과들을 `seq` 순으로 반환한다. */
  release(res: GatewayResult): GatewayResult[] {
    if (res.seq < this.nextSeq) {
      // 이미 release되었거나 중복/지연 도착이므로 두 번 방출하지 않도록 무시한다.
      return [];
    }
    this.pending.set(res.seq, res);
    const ready: GatewayResult[] = [];
    for (;;) {
      const r = this.pending.get(this.nextSeq);
      if (r === undefined) {
        break;
      }
      ready.push(r);
      this.pending.delete(this.nextSeq);
      this.nextSeq += 1;
    }
    return ready;
  }

  /** 선행 `seq`를 기다리며 보류 중인 결과 수다. cleanup 계측에서 reorder backlog로 관측한다. */
  pendingCount(): number {
    return this.pending.size;
  }

  /** 버퍼된 결과를 비운다. gateway close 시 stale result가 contiguous release를 유발하지 않게 한다. */
  clear(): void {
    this.pending.clear();
  }
}

export interface WorkerGatewayOptions {
  /** decode 결과를 입력 `seq` 순서로 받는 coordinator sink다. 반환 Promise는 직렬화되어 backpressure를 전달한다. */
  sink: (frame: DecodedFrame) => void | Promise<void>;
  /** raw bytes decode 함수. 기본은 inline PacketBase decode이며, Web Worker 등 off-thread decode를 주입할 수 있다. */
  decode?: FrameDecoder;
  /** decode 오류를 `seq` 순서로 보고한다. transport가 inline path와 동일한 parse-error 정책을 적용할 수 있게 한다. */
  onError?: (err: Error) => void;
  /** 동시 in-flight decode 상한. `<= 0`이면 4를 적용한다. */
  workers?: number;
}

/**
 * async worker-pool gateway다. 제출된 프레임을 최대 `workers`개의 async slot이 동시에 decode하고, 단일
 * reorder 버퍼가 결과를 `seq`로 재정렬해 입력 순서로 sink에 전달한다. sink 호출은 단일 chain으로 직렬화되어
 * 순서가 보존되므로, stateful dispatch는 수신 coordinator에 남는다.
 *
 * 기본 decode는 호출 스레드에서 PacketBase envelope을 inline 처리하므로 worker 미지원 환경의 safe
 * fallback이 된다. browser에서는 Web Worker 기반 `decode`를, Node에서는 `worker_threads` 기반
 * `decode`를 주입해 실제 off-thread 처리를 적용할 수 있다.
 *
 * decode에 실패한 프레임은 sink가 아니라 `onError`로 `seq` 순서에 맞춰 보고되어 한 프레임의 실패가 이후
 * 프레임을 막지 않는다. gateway 자체는 transport를 종료하지 않는다.
 */
export class WorkerGateway implements InboundGateway {
  private readonly decode: FrameDecoder;
  private readonly sink: (frame: DecodedFrame) => void | Promise<void>;
  private readonly onError?: (err: Error) => void;
  private readonly maxWorkers: number;
  private readonly reorder = new FrameReorderBuffer();
  private readonly jobs: InboundFrame[] = [];
  private activeWorkers = 0;
  private closed = false;
  private sinkChain: Promise<void> = Promise.resolve();
  // sink chain에 스케줄됐지만 아직 전달되지 않은 결과 수. cleanup 계측의 sinkPending으로 관측한다.
  private sinkPending = 0;

  constructor(options: WorkerGatewayOptions) {
    this.decode = options.decode ?? decodePacketBaseEnvelope;
    this.sink = options.sink;
    this.onError = options.onError;
    const workers = options.workers ?? 0;
    this.maxWorkers = workers > 0 ? workers : 4;
  }

  submit(frame: InboundFrame): void {
    if (this.closed) {
      return;
    }
    this.jobs.push(frame);
    if (this.activeWorkers < this.maxWorkers) {
      this.activeWorkers += 1;
      void this.workerLoop();
    }
  }

  close(): void {
    if (this.closed) {
      return;
    }
    // 추가 전달을 멈춘다(InboundGateway.close() 계약). queued jobs와 reorder pending을 비워 close 이후
    // 완료되는 in-flight decode가 stale result를 sink/onError로 흘리거나 contiguous release를 유발하지 못하게
    // 한다. closed guard는 collect/sinkChain 전달 직전에도 적용된다.
    this.closed = true;
    this.jobs.length = 0;
    this.reorder.clear();
    // 스케줄된 sink 전달을 폐기하므로 backlog 계측도 즉시 0으로 본다. 이미 큐된 sinkChain 콜백은
    // closed guard로 전달을 건너뛰며, 아래 finally의 가드가 카운터를 음수로 떨어뜨리지 않는다.
    this.sinkPending = 0;
  }

  /** 현재 backlog/close 상태 스냅샷을 반환한다(cleanup/drain 계측용). */
  stats(): InboundGatewayStats {
    return {
      queued: this.jobs.length,
      active: this.activeWorkers,
      reorderPending: this.reorder.pendingCount(),
      sinkPending: this.sinkPending,
      // 기본 WorkerGateway는 별도 off-thread worker pool을 두지 않는다. Node worker gateway wrapper가
      // pool pending을 합성해 채운다.
      workerPending: 0,
      closed: this.closed,
    };
  }

  private async workerLoop(): Promise<void> {
    try {
      for (;;) {
        const frame = this.jobs.shift();
        if (frame === undefined) {
          return;
        }
        let result: GatewayResult;
        try {
          const env = await this.decode(frame.bytes);
          result = { seq: frame.seq, frame: { seq: frame.seq, ...env }, error: null };
        } catch (err) {
          result = { seq: frame.seq, frame: null, error: toError(err) };
        }
        // decode가 await 중에 close되었을 수 있다. close 이후에는 stale result/error를 전달하지 않는다.
        if (this.closed) {
          return;
        }
        this.collect(result);
      }
    } finally {
      this.activeWorkers -= 1;
    }
  }

  private collect(result: GatewayResult): void {
    if (this.closed) {
      return;
    }
    const ready = this.reorder.release(result);
    if (ready.length === 0) {
      return;
    }
    // reorder는 contiguous prefix만 release하므로 batch들은 호출 순서가 곧 seq 순서다. sinkChain으로
    // 직렬화해 async sink가 입력 순서를 깨지 않게 한다.
    this.sinkPending += ready.length;
    this.sinkChain = this.sinkChain.then(async () => {
      for (const r of ready) {
        try {
          // 스케줄 이후 close되었으면 남은 전달을 건너뛴다(stale delivery 금지). 단 카운터는 finally에서
          // 정리해야 하므로 return이 아니라 continue로 남은 항목을 소비한다.
          if (this.closed) {
            continue;
          }
          if (r.error !== null) {
            // decode 오류를 seq 순서로 보고하고 계속 진행한다. transport 종료 여부는 onError가 결정한다.
            this.onError?.(r.error);
            continue;
          }
          if (r.frame !== null) {
            await this.sink(r.frame);
          }
        } finally {
          // close()가 sinkPending을 0으로 리셋했을 수 있으므로 음수로 떨어뜨리지 않는다.
          if (this.sinkPending > 0) {
            this.sinkPending -= 1;
          }
        }
      }
    });
  }
}

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