/**
 * HTTP/2 client for the local Windsurf language server binary.
 * Supports both gRPC and Connect-RPC protocols.
 *
 * Default: legacy gRPC framing (verified working with LS 2.12.5 against
 * production cascade flow). Set GRPC_PROTOCOL=connect to opt in to Connect
 * framing — note: as of v2.0.20, Connect default returned empty cascade_id
 * from StartCascade against the production LS, so we keep legacy as default
 * until the Connect response parser is debugged. Tracked for v2.0.22+.
 */

import http2 from 'http2';
import { log } from './config.js';
import { wrapRequest, unwrapRequest, gunzip, StreamingFrameParser } from './connect.js';
import { traceGrpcPayload } from './proto-trace.js';

const USE_CONNECT = process.env.GRPC_PROTOCOL === 'connect';
export const _USE_CONNECT_FOR_TEST = USE_CONNECT;

/**
 * Decode a raw upstream grpc-message trailer. Per the gRPC spec the value is
 * percent-encoded, but a misbehaving upstream can emit a malformed escape
 * (e.g. a bare '%'), and decodeURIComponent throws URIError on those. Since
 * this runs inside the stream 'end' listener, an unguarded throw escapes as
 * an uncaughtException and takes the whole proxy down. Fall back to the raw
 * value on decode failure so the status is still surfaced as an error.
 */
export function decodeGrpcMessage(grpcMessage) {
  try {
    return decodeURIComponent(grpcMessage);
  } catch {
    return grpcMessage;
  }
}

// ─── HTTP/2 session pool ───────────────────────────────────
//
// Previously every grpcUnary / grpcStream call did its own http2.connect()
// and client.close() — that's one TCP + HTTP/2 handshake per request, which
// under chat bursts (poll trajectory every 50 ms + per-chunk Send calls)
// was (a) wasting a SYN + SETTINGS round-trip per call and (b) burning
// ephemeral ports, eventually tripping EADDRNOTAVAIL. HTTP/2 is
// multiplexed — one session happily carries many concurrent streams, so we
// keep one session per LS port and let it handle all requests.
//
// The session is torn down (and a fresh one will be opened on demand) if
// it emits 'error' or 'close' — callers still see the error on their own
// `req` object because the stream error is delivered independently.

const _sessionPool = new Map();

function getSession(port) {
  const key = `127.0.0.1:${port}`;
  let session = _sessionPool.get(key);
  if (session && !session.destroyed && !session.closed) return session;

  // Use 127.0.0.1 (not "localhost") so the HTTP/2 client always connects via
  // IPv4 — the LS listens on 127.0.0.1, and on macOS "localhost" resolves to
  // ::1 first. With autoSelectFamily=false (set in config.js to fix ETIMEDOUT
  // on remote server.codeium.com connections), Node only tries the first DNS
  // result (::1) and gets ECONNREFUSED instead of falling back to 127.0.0.1.
  session = http2.connect(`http://127.0.0.1:${port}`);
  session.on('error', (err) => {
    log.debug(`HTTP/2 session error on port ${port}: ${err.message}`);
    if (_sessionPool.get(key) === session) _sessionPool.delete(key);
  });
  session.on('close', () => {
    if (_sessionPool.get(key) === session) _sessionPool.delete(key);
  });
  // The LS can hang up between requests; unref so an idle session doesn't
  // keep the Node event loop alive on its own.
  try { session.unref(); } catch {}
  _sessionPool.set(key, session);
  return session;
}

/**
 * Close the pooled session for a port (used when the underlying LS is
 * stopped so the next call opens a fresh session against whatever took
 * the port).
 */
export function closeSessionForPort(port) {
  const key = `127.0.0.1:${port}`;
  const session = _sessionPool.get(key);
  if (session) {
    try { session.close(); } catch {}
    _sessionPool.delete(key);
  }
}

/**
 * Wrap a protobuf payload for transport.
 * Connect mode: gzip-compressed connect envelope.
 * gRPC mode: uncompressed gRPC frame.
 *
 * Connect *unary* calls unwrap this again inside grpcUnary (the protocol wants a
 * bare message there); streaming calls send it as-is.
 */
export function grpcFrame(payload) {
  const buf = Buffer.isBuffer(payload) ? payload : Buffer.from(payload);
  if (USE_CONNECT) return wrapRequest(buf);
  const frame = Buffer.alloc(5 + buf.length);
  frame[0] = 0;
  frame.writeUInt32BE(buf.length, 1);
  buf.copy(frame, 5);
  return frame;
}

/**
 * Strip gRPC frame header (5 bytes) from a response buffer.
 * Returns the protobuf payload.
 */
export function stripGrpcFrame(buf) {
  if (buf.length >= 5 && buf[0] === 0) {
    const msgLen = buf.readUInt32BE(1);
    if (buf.length >= 5 + msgLen) {
      return buf.subarray(5, 5 + msgLen);
    }
  }
  return buf;
}

/**
 * Extract all gRPC frames from a buffer (may contain multiple concatenated frames).
 */
export function extractGrpcFrames(buf) {
  const frames = [];
  let offset = 0;
  while (offset + 5 <= buf.length) {
    const compressed = buf[offset];
    const msgLen = buf.readUInt32BE(offset + 1);
    if (compressed !== 0 || offset + 5 + msgLen > buf.length) break;
    frames.push(buf.subarray(offset + 5, offset + 5 + msgLen));
    offset += 5 + msgLen;
  }
  return frames;
}

// ─── Terminal truth ───────────────────────────────────────
//
// grpcStatus defaults to '0', so a response that ends without any grpc-status
// used to fall through to the success branch: a truncated or crashed LS reply
// surfaced as an empty assistant turn and the account pool never failed over
// (audit G4, probe G1). Presence is tracked separately from the value.
//
// Connect is exempt: its terminal truth is the HTTP status plus the end-stream
// JSON frame. The HTTP-status half is decided as soon as the response headers
// arrive (see the 'response' handlers) — a non-2xx body is an HTML/proxy page,
// and feeding it to the frame reader reported a nonsense frame size instead of
// the real cause (audit G2, probe C2) — so it never reaches this classifier.

/**
 * @returns {Error|null} the failure to surface, or null when the response is a
 *   success and its body may be parsed.
 */
function terminalFailure({ connect, grpcStatus, sawGrpcStatus, grpcMessage }) {
  if (connect) return null;
  if (!sawGrpcStatus) return new Error('gRPC response ended without grpc-status');
  if (grpcStatus !== '0') {
    return new Error(grpcMessage ? decodeGrpcMessage(grpcMessage) : `gRPC status ${grpcStatus}`);
  }
  return null;
}

/**
 * Split a Connect envelope stream, or return null when the buffer is not one.
 *
 * A spec-compliant Connect *unary* response body is a bare protobuf message;
 * the 5-byte envelope only applies to `application/connect+proto` streaming.
 * The LS has historically answered unary calls with envelopes, so both shapes
 * are accepted — but the discriminator is structural, never try/catch: a valid
 * protobuf message cannot begin with 0x00-0x03 (field number 0 is illegal, and
 * tags 0x01-0x03 are field 0 with wire types 1-3), which is exactly the first
 * byte of every envelope. That makes the two shapes unambiguous, so a corrupt
 * envelope is reported as such instead of being parsed as a message (G12).
 */
function scanConnectEnvelope(buf) {
  const frames = [];
  let offset = 0;
  while (offset < buf.length) {
    if (offset + 5 > buf.length) return null;
    const flags = buf[offset];
    if (flags & ~0x03) return null;
    const len = buf.readUInt32BE(offset + 1);
    if (offset + 5 + len > buf.length) return null;
    frames.push({ flags, isEndStream: !!(flags & 0x02), payload: buf.subarray(offset + 5, offset + 5 + len) });
    offset += 5 + len;
  }
  return frames;
}

/**
 * Decode a Connect unary response body.
 *
 * Returns the bare message plus any error the end-stream frame carried. A
 * trailer-only body decodes to an EMPTY message — never to the envelope bytes,
 * which callers then ran through parseFields as if they were protobuf (G1).
 */
function decodeConnectUnaryResponse(body) {
  if (body.length === 0 || body[0] > 0x03) return { payload: body, error: null };
  const frames = scanConnectEnvelope(body);
  if (!frames) throw new Error('Connect unary response is not a valid frame stream');
  const data = [];
  let error = null;
  for (const frame of frames) {
    if (frame.isEndStream) {
      if (!error) {
        try {
          const trailer = JSON.parse(frame.payload.toString());
          if (trailer.error) error = new Error(trailer.error.message || 'connect error');
        } catch {}
      }
      continue;
    }
    data.push(frame.flags & 0x01 ? gunzip(frame.payload) : frame.payload);
  }
  return { payload: Buffer.concat(data), error };
}

/**
 * Make a unary gRPC call to the language server.
 *
 * @param {number} port - Language server port
 * @param {string} csrfToken - CSRF token
 * @param {string} path - gRPC path (e.g. /exa.language_server_pb.LanguageServerService/StartCascade)
 * @param {Buffer} body - gRPC-framed request
 * @param {number} timeout - Timeout in ms
 * @returns {Promise<Buffer>} Protobuf response (stripped of gRPC frame)
 */
export function grpcUnary(port, csrfToken, path, body, timeout = 30000) {
  return new Promise((resolve, reject) => {
    // Guard against double-settling: req 'error' followed by session
    // 'error' (or a late 'end' after an abort) would otherwise call
    // resolve and reject both.
    let settled = false;
    const done = (fn, ...args) => {
      if (settled) return;
      settled = true;
      fn(...args);
    };

    // Connect unary RPCs use `application/proto` with a BARE message; the
    // envelope plus `application/connect+proto` belongs to streaming RPCs only
    // (connectrpc.com/docs/protocol). Callers hand every request through
    // grpcFrame(), which envelopes under GRPC_PROTOCOL=connect, so unwrap here
    // instead of teaching every call site which transport is active.
    //
    // Done before the timer is armed: unwrapRequest can throw on a corrupt
    // envelope, and a throw after setTimeout() would leave a timer whose
    // callback dereferences `req` before it exists.
    const requestBody = USE_CONNECT ? unwrapRequest(body, {}) : body;

    const client = getSession(port);
    const chunks = [];
    let timer;

    timer = setTimeout(() => {
      try { req.close?.(http2.constants.NGHTTP2_CANCEL); } catch {}
      done(reject, new Error('gRPC unary timeout'));
    }, timeout);

    const headers = USE_CONNECT ? {
      ':method': 'POST',
      ':path': path,
      'content-type': 'application/proto',
      'connect-protocol-version': '1',
      'user-agent': 'connect-es/2.0.0',
      'x-codeium-csrf-token': csrfToken,
    } : {
      ':method': 'POST',
      ':path': path,
      'content-type': 'application/grpc',
      'te': 'trailers',
      'user-agent': 'grpc-node/1.108.2',
      'x-codeium-csrf-token': csrfToken,
    };

    const req = client.request(headers);
    traceGrpcPayload({
      port,
      path,
      direction: 'request',
      body: requestBody,
      transport: USE_CONNECT ? 'connect' : 'grpc',
      framed: !USE_CONNECT,
    });
    req.on('data', (chunk) => chunks.push(chunk));

    let grpcStatus = '0', grpcMessage = '', sawGrpcStatus = false;

    // H-3 (ultracode audit 2026-07-13): a gRPC Trailers-Only response folds
    // grpc-status into the FIRST HEADERS frame (END_STREAM, no DATA) — it arrives
    // via the 'response' event, NOT 'trailers'. Without reading it, an immediate
    // upstream error (grpc-status 14/8/12) or an HTTP :status 4xx/5xx left
    // grpcStatus at '0' and was treated as an empty SUCCESS, blinding the account
    // pool's error classification/failover. Capture status from the response
    // headers too, then honour it in the end handler below.
    req.on('response', (headers) => {
      const status = Number(headers[':status'] || 0);
      if (headers['grpc-status'] != null) { grpcStatus = String(headers['grpc-status']); sawGrpcStatus = true; }
      if (headers['grpc-message'] != null) grpcMessage = String(headers['grpc-message']);
      // A non-2xx body is an error page (proxy/HTML), never a message. Decide at
      // the headers and stop reading so the real cause is reported instead of a
      // frame-size error produced by parsing HTML as frames (G2).
      if (status && (status < 200 || status >= 300)) {
        clearTimeout(timer);
        try { req.close?.(http2.constants.NGHTTP2_CANCEL); } catch {}
        done(reject, new Error(`upstream HTTP ${status}`));
      }
    });

    req.on('trailers', (trailers) => {
      // Trailers override/confirm; only take grpc-status when actually present so
      // a Trailers-Only status captured from headers isn't reset to '0'.
      if (trailers['grpc-status'] != null) { grpcStatus = String(trailers['grpc-status']); sawGrpcStatus = true; }
      if (trailers['grpc-message'] != null) grpcMessage = String(trailers['grpc-message']);
    });

    req.on('end', () => {
      clearTimeout(timer);
      if (settled) return;
      const failure = terminalFailure({ connect: USE_CONNECT, grpcStatus, sawGrpcStatus, grpcMessage });
      if (failure) { done(reject, failure); return; }
      const full = Buffer.concat(chunks);

      if (USE_CONNECT) {
        let decoded;
        try {
          decoded = decodeConnectUnaryResponse(full);
        } catch (err) {
          try { req.close?.(http2.constants.NGHTTP2_CANCEL); } catch {}
          done(reject, err);
          return;
        }
        if (decoded.error) { done(reject, decoded.error); return; }
        traceGrpcPayload({
          port,
          path,
          direction: 'response',
          body: decoded.payload,
          transport: 'connect',
          framed: false,
        });
        done(resolve, decoded.payload);
      } else {
        const frames = extractGrpcFrames(full);
        const payload = frames.length > 0 ? Buffer.concat(frames) : stripGrpcFrame(full);
        traceGrpcPayload({
          port,
          path,
          direction: 'response',
          body: payload,
          transport: 'grpc',
          framed: false,
        });
        done(resolve, payload);
      }
    });

    req.on('error', (err) => {
      clearTimeout(timer);
      done(reject, err);
    });

    req.write(requestBody);
    req.end();
  });
}

/**
 * Make a streaming gRPC call to the language server.
 * Yields parsed gRPC frame payloads as they arrive.
 *
 * @param {number} port
 * @param {string} csrfToken
 * @param {string} path
 * @param {Buffer} body
 * @param {object} opts - { onData, onEnd, onError, timeout, signal }
 */
export function grpcStream(port, csrfToken, path, body, opts = {}) {
  const { onData, onEnd, onError, timeout = 300000, signal } = opts;

  // GAP-2: caller-supplied cancellation. A client disconnect (or a server
  // shutdown drain) aborts `signal`; close the HTTP/2 stream with
  // NGHTTP2_CANCEL and surface an AbortError so callers stop burning
  // LS + account quota. No-op when no signal is supplied.
  const abortError = () => {
    const err = new Error('The operation was aborted');
    err.name = 'AbortError';
    err.code = 'ABORT_ERR';
    return err;
  };
  if (signal?.aborted) {
    // Already aborted before any stream was opened — settle without a request.
    onError?.(abortError());
    return;
  }

  // req may emit both 'end' and 'error' (or error twice) when the server
  // trailers report non-OK — flip this to only fire one callback.
  let settled = false;
  const client = getSession(port);
  let timer;
  let pendingChunks = [];
  let pendingHead = 0;
  let pendingOffset = 0;
  let pendingLength = 0;

  const pendingByteAt = (relative) => {
    let index = pendingHead;
    let offset = pendingOffset + relative;
    while (offset >= pendingChunks[index].length) {
      offset -= pendingChunks[index++].length;
    }
    return pendingChunks[index][offset];
  };

  const copyPendingPayload = (length) => {
    const payload = Buffer.alloc(length);
    if (length === 0) return payload;
    let index = pendingHead;
    let offset = pendingOffset + 5;
    while (offset >= pendingChunks[index].length) {
      offset -= pendingChunks[index++].length;
    }
    let written = 0;
    while (written < length) {
      const chunk = pendingChunks[index++];
      const take = Math.min(length - written, chunk.length - offset);
      payload.set(chunk.subarray(offset, offset + take), written);
      written += take;
      offset = 0;
    }
    return payload;
  };

  const consumePending = (length) => {
    pendingLength -= length;
    while (length > 0) {
      const available = pendingChunks[pendingHead].length - pendingOffset;
      if (length < available) {
        pendingOffset += length;
        break;
      }
      length -= available;
      pendingChunks[pendingHead++] = null;
      pendingOffset = 0;
    }
    if (pendingHead === pendingChunks.length) {
      pendingChunks = [];
      pendingHead = 0;
    } else if (pendingHead * 2 >= pendingChunks.length) {
      // Retired entries pay for this reference-only compaction; never shift().
      pendingChunks = pendingChunks.slice(pendingHead);
      pendingHead = 0;
    }
  };

  timer = setTimeout(() => {
    if (settled) return;
    settled = true;
    try { req.close?.(http2.constants.NGHTTP2_CANCEL); } catch {}
    onError?.(new Error('gRPC stream timeout'));
  }, timeout);

  const streamHeaders = USE_CONNECT ? {
    ':method': 'POST',
    ':path': path,
    'content-type': 'application/connect+proto',
    'connect-protocol-version': '1',
    'connect-accept-encoding': 'gzip',
    'user-agent': 'connect-es/2.0.0',
    'x-codeium-csrf-token': csrfToken,
  } : {
    ':method': 'POST',
    ':path': path,
    'content-type': 'application/grpc',
    'te': 'trailers',
    // T3 (Grok audit): the frame reader only handles uncompressed frames
    // (compressed===0) — a gzip/deflate frame is silently skipped, truncating
    // the stream. Advertise ONLY what we can actually decode so the LS never
    // sends a compressed frame we'd drop. (The LS is local loopback and rarely
    // compresses, but advertising a codec we can't read is a latent bug.)
    'grpc-accept-encoding': 'identity',
    'user-agent': 'grpc-node/1.108.2',
    'x-codeium-csrf-token': csrfToken,
  };

  const req = client.request(streamHeaders);
  traceGrpcPayload({
    port,
    path,
    direction: 'request',
    body,
    transport: USE_CONNECT ? 'connect' : 'grpc',
    framed: true,
  });

  if (signal) {
    signal.addEventListener('abort', () => {
      if (settled) return;
      settled = true;
      clearTimeout(timer);
      try { req.close?.(http2.constants.NGHTTP2_CANCEL); } catch {}
      onError?.(abortError());
    }, { once: true });
  }

  const connectParser = USE_CONNECT ? new StreamingFrameParser() : null;

  req.on('data', (chunk) => {
    if (settled) return;

    if (USE_CONNECT) {
      try {
        connectParser.push(chunk);
        for (const frame of connectParser.drain()) {
          if (frame.isEndStream) {
            try {
              const t = JSON.parse(frame.payload.toString());
              if (t.error) {
                settled = true;
                clearTimeout(timer);
                try { req.close?.(http2.constants.NGHTTP2_CANCEL); } catch {}
                onError?.(new Error(t.error.message || 'connect stream error'));
                return;
              }
            } catch {}
          } else {
            traceGrpcPayload({
              port,
              path,
              direction: 'response',
              body: frame.payload,
              transport: 'connect',
              framed: false,
            });
            onData?.(frame.payload);
          }
        }
      } catch (err) {
        settled = true;
        clearTimeout(timer);
        try { req.close?.(http2.constants.NGHTTP2_CANCEL); } catch {}
        onError?.(err);
      }
      return;
    }

    // Incoming HTTP/2 Buffer chunks are borrowed until consumed. Keep the guard
    // on aggregate pending bytes BEFORE draining, not on advertised frame size.
    if (chunk.length) {
      pendingChunks.push(chunk);
      pendingLength += chunk.length;
    }
    if (pendingLength > 100 * 1024 * 1024) {
      settled = true;
      clearTimeout(timer);
      try { req.close?.(http2.constants.NGHTTP2_CANCEL); } catch {}
      onError?.(new Error('gRPC frame too large (>100MB)'));
      return;
    }

    while (pendingLength >= 5) {
      const compressed = pendingByteAt(0);
      let msgLen = 0;
      for (let i = 1; i < 5; i++) msgLen = msgLen * 256 + pendingByteAt(i);
      if (pendingLength < 5 + msgLen) break;

      if (compressed === 0) {
        const payload = copyPendingPayload(msgLen);
        traceGrpcPayload({
          port,
          path,
          direction: 'response',
          body: payload,
          transport: 'grpc',
          framed: false,
        });
        onData?.(payload);
      }
      consumePending(5 + msgLen);
    }
  });

  let grpcStatus = '0', grpcMessage = '', sawGrpcStatus = false;

  // H-3 (ultracode audit 2026-07-13): capture Trailers-Only status from the
  // HEADERS frame ('response' event) — same fix as grpcUnary. Without it a
  // stream that errors immediately (Trailers-Only grpc-status, or HTTP :status
  // 4xx/5xx, END_STREAM no DATA) produced zero chunks → onEnd → a silent empty
  // assistant reply instead of an error the pool can classify.
  req.on('response', (headers) => {
    const status = Number(headers[':status'] || 0);
    if (headers['grpc-status'] != null) { grpcStatus = String(headers['grpc-status']); sawGrpcStatus = true; }
    if (headers['grpc-message'] != null) grpcMessage = String(headers['grpc-message']);
    // Fail at the headers for a non-2xx response: its body is an error page, and
    // the frame reader would report a bogus frame size for it instead of the
    // status (G2). Settle here so the data handler never parses that body.
    if (!settled && status && (status < 200 || status >= 300)) {
      settled = true;
      clearTimeout(timer);
      try { req.close?.(http2.constants.NGHTTP2_CANCEL); } catch {}
      onError?.(new Error(`upstream HTTP ${status}`));
    }
  });

  req.on('trailers', (trailers) => {
    if (trailers['grpc-status'] != null) { grpcStatus = String(trailers['grpc-status']); sawGrpcStatus = true; }
    if (trailers['grpc-message'] != null) grpcMessage = String(trailers['grpc-message']);
  });

  req.on('end', () => {
    clearTimeout(timer);
    if (settled) return;
    settled = true;
    const failure = terminalFailure({ connect: USE_CONNECT, grpcStatus, sawGrpcStatus, grpcMessage });
    if (failure) onError?.(failure);
    else onEnd?.();
  });

  req.on('error', (err) => {
    clearTimeout(timer);
    if (settled) return;
    settled = true;
    onError?.(err);
  });

  req.write(body);
  req.end();
}