/**
 * Connect-RPC envelope framing and compression.
 *
 * Connect-RPC frame format:
 *   [1 byte flags] [4 bytes big-endian length] [N bytes payload]
 *
 * Flags:
 *   0x01 = gzip compressed
 *   0x02 = end-of-stream (trailer frame, JSON payload)
 *   0x03 = compressed + end-of-stream
 *
 * IMPORTANT: Connect-RPC uses HTTP/1.1 POST, NOT HTTP/2 gRPC.
 * Content-Type: application/connect+proto
 */

import { gzipSync, gunzipSync } from 'zlib';
import { isUint8Array } from 'util/types';

// Hard ceiling on any single frame — both the wire length a frame may advertise
// and the number of bytes a gzip payload may inflate to. Bounding the DECOMPRESSED
// size (via zlib's maxOutputLength) is what stops a high-ratio "zip bomb" frame from
// a hijacked/compromised upstream from inflating to gigabytes and OOM-killing the
// single-process proxy before the resulting error can be caught (audit CONN-1).
// Same value/rationale as StreamingFrameParser's frame-length guard; mirrors the
// maxOutputLength pattern already used for PDF flate streams (pdf.js:52).
export const MAX_FRAME_SIZE = 16 * 1024 * 1024;

// ─── Compression helpers ───────────────────────────────────

export function gzip(buf) { return gzipSync(buf); }

export function gunzip(buf) { return gunzipSync(buf, { maxOutputLength: MAX_FRAME_SIZE }); }

export function tryGunzip(buf) {
  try { return gunzipSync(buf, { maxOutputLength: MAX_FRAME_SIZE }); }
  catch { return null; }
}

// ─── Envelope wrapping ─────────────────────────────────────

/**
 * Wrap protobuf bytes in a Connect-RPC envelope frame.
 */
export function wrapEnvelope(protoBuf, { compress = true } = {}) {
  let payload = protoBuf;
  let flags = 0;
  if (compress && payload.length > 0) {
    payload = gzipSync(payload);
    flags |= 0x01;
  }
  const frame = Buffer.alloc(5 + payload.length);
  frame[0] = flags;
  frame.writeUInt32BE(payload.length, 1);
  payload.copy(frame, 5);
  return frame;
}

/**
 * Wrap a request for sending (single envelope, gzipped).
 */
export function wrapRequest(protoBuf) {
  return wrapEnvelope(protoBuf, { compress: true });
}

/**
 * Build the end-of-stream trailer frame (JSON {}).
 */
export function endOfStreamEnvelope() {
  const trailer = Buffer.from('{}');
  const frame = Buffer.alloc(5 + trailer.length);
  frame[0] = 0x02; // end-of-stream, not compressed
  frame.writeUInt32BE(trailer.length, 1);
  trailer.copy(frame, 5);
  return frame;
}

// ─── Request unwrapping ────────────────────────────────────

/**
 * Unwrap a Connect-RPC request body → raw protobuf bytes.
 * Handles both envelope-wrapped and HTTP-level gzip.
 */
export function unwrapRequest(body, headers = {}) {
  let buf = Buffer.isBuffer(body) ? body : Buffer.from(body);

  // HTTP-level content-encoding gzip
  const encoding = headers['content-encoding'] || headers['connect-content-encoding'] || '';
  if (encoding === 'gzip') {
    buf = gunzipSync(buf, { maxOutputLength: MAX_FRAME_SIZE });
  }

  // Check if it's envelope-wrapped (flags byte + 4-byte length)
  if (buf.length >= 5) {
    const flags = buf[0];
    const len = buf.readUInt32BE(1);
    if (len === buf.length - 5 && (flags === 0 || flags === 1)) {
      let payload = buf.subarray(5);
      if (flags & 0x01) payload = gunzipSync(payload, { maxOutputLength: MAX_FRAME_SIZE });
      return payload;
    }
  }

  return buf;
}

// ─── Streaming frame parser ───────────────────────────────

/**
 * Stateful parser that buffers incoming data and yields complete frames.
 */
export class StreamingFrameParser {
  constructor() {
    this._chunks = [];
    this._head = 0;
    this._offset = 0;
    this._length = 0;
  }

  push(chunk) {
    if (!isUint8Array(chunk)) {
      // Preserve Buffer.concat's invalid-input error without copying valid data.
      Buffer.concat([Buffer.alloc(0), chunk]);
    }
    // Borrow transport chunks until consumed. Empty chunks must not accumulate
    // queue entries or turn a five-byte header lookup into an unbounded scan.
    if (chunk.length === 0) return;
    this._chunks.push(chunk);
    this._length += chunk.length;
  }

  _byteAt(relative) {
    let index = this._head;
    let offset = this._offset + relative;
    while (offset >= this._chunks[index].length) {
      offset -= this._chunks[index++].length;
    }
    return this._chunks[index][offset];
  }

  _copyPayload(length) {
    const payload = Buffer.alloc(length);
    if (length === 0) return payload;
    let index = this._head;
    let offset = this._offset + 5;
    while (offset >= this._chunks[index].length) {
      offset -= this._chunks[index++].length;
    }
    let written = 0;
    while (written < length) {
      const chunk = this._chunks[index++];
      const take = Math.min(length - written, chunk.length - offset);
      payload.set(chunk.subarray(offset, offset + take), written);
      written += take;
      offset = 0;
    }
    return payload;
  }

  _consume(length) {
    this._length -= length;
    while (length > 0) {
      const available = this._chunks[this._head].length - this._offset;
      if (length < available) {
        this._offset += length;
        break;
      }
      length -= available;
      this._chunks[this._head++] = null;
      this._offset = 0;
    }
    if (this._head === this._chunks.length) {
      this._chunks = [];
      this._head = 0;
    } else if (this._head * 2 >= this._chunks.length) {
      // Copy only references, only after at least as many entries were retired.
      // This bounds retained slots and amortizes compaction across consumed chunks.
      this._chunks = this._chunks.slice(this._head);
      this._head = 0;
    }
  }

  /** Drain all complete frames. Returns [{ flags, isEndStream, payload }]. */
  drain() {
    // Validate the advertised length before waiting for or allocating its body.
    // Reading five header bytes never walks the accumulated payload prefix.
    const frames = [];
    while (this._length >= 5) {
      let len = 0;
      for (let i = 1; i < 5; i++) len = len * 256 + this._byteAt(i);
      if (len > MAX_FRAME_SIZE) {
        throw new Error(`HTTP/2 frame size ${len} exceeds ${MAX_FRAME_SIZE}`);
      }
      if (this._length < 5 + len) break;

      const flags = this._byteAt(0);
      let payload = this._copyPayload(len);
      if (flags & 0x01) {
        // Bound the DECOMPRESSED size too: a high-ratio gzip frame that passes the
        // ≤16MB wire-length check above can still inflate to gigabytes. Without
        // maxOutputLength, gunzipSync accumulates up to kMaxLength (~2GB) and on a
        // memory-constrained host the OOM-killer can reap the process before the
        // throw below is caught (audit CONN-1).
        try { payload = gunzipSync(payload, { maxOutputLength: MAX_FRAME_SIZE }); }
        catch (err) {
          // Don't silently drop a frame whose body we couldn't decompress —
          // upstream is sending something we don't understand and the caller
          // needs to know so the stream can be cancelled rather than continued
          // with missing data.
          throw new Error(`Connect frame decompression failed: ${err.message}`);
        }
      }

      frames.push({
        flags,
        isEndStream: !!(flags & 0x02),
        payload,
      });
      // A decompression error must leave the failing frame queued, as before.
      this._consume(5 + len);
    }
    return frames;
  }
}

// ─── Connect-RPC headers ──────────────────────────────────

export function connectHeaders(extra = {}) {
  return {
    'Content-Type': 'application/connect+proto',
    'Connect-Protocol-Version': '1',
    'Connect-Accept-Encoding': 'gzip',
    'User-Agent': 'connect-es/2.0.0',
    ...extra,
  };
}