// Deferred SSE commit point.
//
// Why: the proxy currently writes HTTP 200 before the upstream attempt, so a
// pre-content 429/500 can only be delivered as an in-band SSE error frame after
// a 200. That loses the status and Retry-After. This wrapper holds the first
// frames until it can tell a real payload from an error, then either commits the
// SSE response or replaces it with a JSON error carrying the real status.
//
// It is deliberately protocol-agnostic at the byte level; the caller supplies
// the error-body shape. The first integration is the OpenAI chat route.

import { Buffer } from 'node:buffer';
import { StringDecoder } from 'node:string_decoder';
import { findFrameEnd, parseSseLine, parseSseDataJson } from './sse-framer.js';
import { connectErrorToHttp, normalizeOpenAIErrorType, clientRetryAfterSeconds } from './handlers/chat.js';

const DEFAULT_MAX_FRAMES = 48;
const DEFAULT_MAX_BYTES = 1024 * 1024;
const DEFAULT_BOOTSTRAP_MS = 5_000;
const MAX_BOOTSTRAP_MS = 14_000;

// A heartbeat frame is comment-only. A leading comment on a frame that also
// carries data must not make that data a bootstrap.
function isCommentFrame(frame) {
  let sawComment = false;
  for (const line of frame.split(/\r\n|\n|\r/)) {
    if (line === '') continue;
    if (!line.startsWith(':')) return false;
    sawComment = true;
  }
  return sawComment;
}

const KEEPALIVE_EVENT = /^(?:ping|keep-?alive)$/i;

// SSE event names that carry no payload. Holding one cannot delay content, and
// it keeps a pre-content error behind a ping from being flattened into a
// committed 200. The last `event` value in the frame wins, as in the framer.
function isKeepaliveFrame(frame) {
  let event = null;
  for (const line of frame.split(/\r\n|\n|\r/)) {
    const parsed = parseSseLine(line);
    if (parsed?.field === 'event') event = parsed.value;
  }
  return event !== null && KEEPALIVE_EVENT.test(event.trim());
}

// True when the text ends on a frame terminator, so no partial frame is left
// to carry. A lone trailing CRLF is not a terminator, so it takes the slow
// path; only a false "no" costs a scan, never a missed frame.
function endsAtFrameEnd(buf) {
  const n = buf.length;
  if (n < 2) return false;
  const last = buf[n - 1];
  const prev = buf[n - 2];
  if ((last !== 10 && last !== 13) || (prev !== 10 && prev !== 13)) return false;
  return !(prev === 13 && last === 10);
}

function dataPayload(frame) {
  const data = [];
  for (const line of frame.split(/\r\n|\n|\r/)) {
    const parsed = parseSseLine(line);
    if (parsed?.field === 'data') data.push(parsed.value);
  }
  return data.join('\n');
}

// Every JSON document a frame carries. An upstream may concatenate documents
// on one `data:` line; `parseSseDataJson` repairs that per line. A payload that
// starts like a JSON object but cannot be parsed into complete documents is
// malformed — the gate must surface a typed error, never treat it as "no error
// frame" and let the stream finish cleanly with zero content.
function frameDocuments(frame) {
  return parseSseDataJson(dataPayload(frame));
}

function malformedDataError() {
  const err = new Error('Upstream sent an SSE data payload that is not valid JSON and could not be repaired');
  // No `type`: `api_error` maps to 500 in TYPE_STATUS, but this is a transport
  // failure (the payload could not be parsed at all), so the status must come
  // from the unclassified-code path — 502 — and the body type is normalized
  // from that status.
  err.code = 'SSE_MALFORMED_JSON';
  return err;
}

function isOpenAIBootstrapChunk(obj) {
  // Role-only prime chunk: no content, no tool call, no finish. Holding it is
  // safe because a client cannot observe anything from it alone.
  if (!Array.isArray(obj?.choices) || obj.choices.length === 0) return false;
  return obj.choices.every((c) => {
    const d = c?.delta;
    if (!d || c?.finish_reason) return false;
    return Object.keys(d).every((k) => k === 'role'
      || ((k === 'content' || k === 'reasoning' || k === 'reasoning_content') && !d[k]));
  });
}

function isBootstrapFrame(frame, protocol, isBootstrap = null) {
  if (isCommentFrame(frame)) return true;
  if (isKeepaliveFrame(frame)) return true;
  const { docs, malformed } = frameDocuments(frame);
  if (malformed || !docs || docs.length === 0) return false;
  // Every document must be bootstrap. One content document in the frame is a
  // real payload and must commit the gate.
  if (typeof isBootstrap === 'function') return docs.every((obj) => !!isBootstrap(obj));
  if (protocol === 'openai') return docs.every((obj) => isOpenAIBootstrapChunk(obj));
  return false;
}

function errorFromFrame(frame, parseError = null) {
  const { docs, malformed } = frameDocuments(frame);
  if (malformed) return malformedDataError();
  if (!docs) return null;
  for (const obj of docs) {
    if (typeof parseError === 'function') {
      const parsed = parseError(obj);
      if (parsed) return parsed;
      continue;
    }
    if (obj && typeof obj === 'object' && obj.error && typeof obj.error === 'object') return obj.error;
  }
  return null;
}

const GRPC_STATUS = {
  INVALID_ARGUMENT: 400,
  OUT_OF_RANGE: 400,
  UNAUTHENTICATED: 401,
  PERMISSION_DENIED: 403,
  NOT_FOUND: 404,
  ALREADY_EXISTS: 409,
  ABORTED: 409,
  RESOURCE_EXHAUSTED: 429,
  INTERNAL: 500,
  UNIMPLEMENTED: 501,
  UNAVAILABLE: 503,
  DEADLINE_EXCEEDED: 504,
};

const TYPE_STATUS = {
  invalid_request_error: 400,
  not_found_error: 404,
  request_too_large: 413,
  rate_limit_error: 429,
  rate_limit_exceeded: 429,
  authentication_error: 401,
  permission_error: 403,
  insufficient_quota: 402,
  capacity_error: 503,
  api_error: 500,
  overloaded_error: 529,
  timeout_error: 504,
};

function mapStatus(error, mapError, protocol = '') {
  if (typeof mapError === 'function') return mapError(error).status;
  // Gemini's error envelope carries the HTTP-like status in numeric `code`
  // (e.g. 400/413/404), while `status` is the gRPC enum. Prefer the numeric
  // code so 413 is not flattened to INVALID_ARGUMENT/400.
  const numeric = Number.isInteger(error?.code)
    ? error.code
    : (Number.isInteger(error?.status) ? error.status : null);
  if (numeric >= 400 && numeric <= 599) return numeric;
  if (error?.status && GRPC_STATUS[error.status]) return GRPC_STATUS[error.status];
  const code = error?.__upstream_code
    || (typeof error?.code === 'string' ? error.code : null);
  // Messages is the one protocol whose Anthropic enum must win over the
  // internal Connect code (MODEL_BLOCKED → permission_error/403, while the
  // same code is 402 for OpenAI/Responses). Other protocols keep the code.
  if (protocol === 'messages' && error?.type && TYPE_STATUS[error.type]) return TYPE_STATUS[error.type];
  if (code) {
    const byCode = connectErrorToHttp(code);
    if (byCode.status !== 502) return byCode.status;
  }
  if (error?.type && TYPE_STATUS[error.type]) return TYPE_STATUS[error.type];
  return 502;
}

function retryAfterSeconds(error) {
  const ms = Number(error?.retry_after_ms ?? error?.retryAfterMs);
  if (Number.isFinite(ms) && ms > 0) return clientRetryAfterSeconds(ms);
  return null;
}

export function createDeferredSseResponse(realRes, {
  status = 200,
  headers = {},
  protocol = 'openai',
  requestId = null,
  errorBody = null,
  mapError = null,
  isBootstrap = null,
  parseError = null,
  maxFrames = DEFAULT_MAX_FRAMES,
  maxBytes = DEFAULT_MAX_BYTES,
  bootstrapTimeoutMs = DEFAULT_BOOTSTRAP_MS,
} = {}) {
  const bootstrapMs = Math.max(0, Math.min(Number(bootstrapTimeoutMs) || 0, MAX_BOOTSTRAP_MS));
  let committed = false;
  let ended = false;
  let headerStatus = status;
  let headerObj = { ...headers };
  let pending = [];
  let pendingBytes = 0;
  let pendingFrames = 0;
  // Incremental scan window (S3-F6): Buffer.concat over ALL pending buffers on
  // every write was O(writes × pending bytes) — a keepalive stream never
  // commits and turns the gate quadratic. `pendingText` holds only decoded
  // text not yet consumed; StringDecoder carries a multibyte char split across
  // writes; the window compacts once the consumed prefix exceeds 64 KiB.
  const pendingDecoder = new StringDecoder('utf8');
  let pendingText = '';
  let inspectOffset = 0;
  let bomChecked = false;
  let timer = null;
  let disposed = false;
  // O10: the first failure this stream surfaced, kept after it is delivered so
  // the caller can classify the request-ring row. The commit gate already
  // parses frames until it commits; from then on an error frame cannot change
  // the status any more, but it still has to reach the ring.
  let failure = null;
  // Post-commit failure scan: one write may carry several frames (content and
  // error), and one frame may span writes, so the scan walks frames and carries
  // the incomplete tail instead of parsing the whole write as a frame.
  let scanCarry = Buffer.alloc(0);

  const clearTimer = () => {
    if (timer) {
      clearTimeout(timer);
      timer = null;
    }
  };

  const commit = () => {
    if (committed || ended) return;
    if (realRes.writableEnded || realRes.destroyed) { ended = true; clearTimer(); return; }
    committed = true;
    clearTimer();
    realRes.writeHead(headerStatus, headerObj);
    if (pending.length === 0) realRes.flushHeaders?.();
    for (const buf of pending) realRes.write(buf);
    pending = [];
    pendingBytes = 0;
    pendingFrames = 0;
    pendingText = '';
    inspectOffset = 0;
  };

  const finishError = (error) => {
    if (ended) return;
    if (!failure) failure = error;
    ended = true;
    clearTimer();
    if (realRes.writableEnded || realRes.destroyed) return;
    const mapped = mapStatus(error, mapError, protocol);
    const retry = retryAfterSeconds(error);
    const fallbackType = (mapError ? mapError(error).type : connectErrorToHttp(error?.code).type);
    const type = normalizeOpenAIErrorType(
      error?.type && error.type !== 'api_error' ? error.type : fallbackType,
      mapped,
    );
    const body = errorBody
      ? errorBody(error, mapped)
      : {
        error: {
          message: error?.message || 'Upstream error',
          type,
          ...(error?.code ? { code: error.code } : {}),
        },
        ...(requestId ? { request_id: requestId } : {}),
      };
    const outHeaders = {};
    for (const [k, v] of Object.entries(headerObj)) {
      if (k.toLowerCase() !== 'content-type') outHeaders[k] = v;
    }
    outHeaders['content-type'] = 'application/json; charset=utf-8';
    if (retry) outHeaders['retry-after'] = String(retry);
    try {
      realRes.writeHead(mapped, outHeaders);
      headerStatus = mapped;
      committed = true;
      realRes.end(JSON.stringify(body));
    } catch {
      ended = true;
      // S3-F2: a throw here (e.g. an invalid header byte) used to leave the
      // socket half-open — `ended` flagged but nothing ever written — and the
      // caller held the upstream slot forever. Destroy so the connection
      // actually closes.
      try { realRes.destroy(); } catch { /* already gone */ }
    }
  };

  const scanFailure = (buf) => {
    if (failure) return;
    // A JSON error key cannot occur in a write without this byte sequence, so
    // skip the frame walk unless the write (or a carried frame) can hold one.
    // A write that stops mid-frame is always walked: its `"error"` may still be
    // in the next write, and the incomplete tail has to be carried.
    if (scanCarry.length === 0 && buf.indexOf('"error"') === -1 && endsAtFrameEnd(buf)) return;
    const text = scanCarry.length === 0
      ? buf.toString('utf8')
      : Buffer.concat([scanCarry, buf]).toString('utf8');
    let offset = 0;
    for (;;) {
      const found = findFrameEnd(text, offset);
      if (!found) break;
      const error = errorFromFrame(text.slice(offset, found.frameEnd), parseError);
      if (error) {
        failure = error;
        scanCarry = Buffer.alloc(0);
        return;
      }
      offset = found.next;
    }
    const rest = offset < text.length ? Buffer.from(text.slice(offset), 'utf8') : Buffer.alloc(0);
    scanCarry = rest.length <= maxBytes ? rest : Buffer.alloc(0);
  };

  const inspectPending = () => {
    let text = pendingText;
    // The wire may start with a UTF-8 BOM. It stays in `pending`, so the client
    // still receives it, but it must not hide the first frame from the gate.
    // Before the gate commits, `pending` always starts at the stream start, so
    // the BOM is always at index 0 of the first write; checking once keeps
    // `inspectOffset` in one coordinate space across writes.
    if (!bomChecked && text.length > 0) {
      bomChecked = true;
      if (text.charCodeAt(0) === 0xfeff) text = text.slice(1);
      pendingText = text;
    }
    let offset = inspectOffset;
    let framesSeen = 0;
    for (;;) {
      const found = findFrameEnd(text, offset);
      if (!found) break;
      framesSeen += 1;
      const frame = text.slice(offset, found.frameEnd);
      const error = errorFromFrame(frame, parseError);
      if (error) {
        finishError(error);
        return;
      }
      if (!isBootstrapFrame(frame, protocol, isBootstrap)) {
        commit();
        // Committing does not end the stream: an error frame after the
        // committing frame in this same write still has to reach the ring.
        if (!failure && !ended) scanFailure(Buffer.from(text.slice(found.next), 'utf8'));
        return;
      }
      offset = found.next;
      inspectOffset = offset;
    }
    pendingFrames += framesSeen;
    // Compact the consumed prefix so a long bootstrap run does not drag a
    // growing string through every write; 64 KiB amortizes the slice.
    if (inspectOffset > 65536) {
      pendingText = pendingText.slice(inspectOffset);
      inspectOffset = 0;
    }
    // Keep holding; a later content/error frame decides. The timer is the
    // escape hatch so a silent upstream still commits a live SSE response.
    if (pendingFrames > maxFrames || pendingBytes > maxBytes) commit();
  };

  // The bootstrap timer is armed by the FIRST write, not at construction.
  //
  // Measured P0 (2026-10-22): the gate used to start this timer when the
  // wrapper was created — before the handler acquired an account. The Cascade
  // stream handler can wait up to QUEUE_MAX_WAIT_MS (30 s) for admission and
  // only then discover the pool is exhausted; a construction-time 5 s timer
  // committed `200 text/event-stream` first, so the 429 and its Retry-After
  // could only arrive as an in-band error frame. Arming on the first write
  // keeps the response uncommitted for the whole admission window: a
  // pre-content error frame is then the first write and rewrites the status.
  // Once the handler is admitted it starts heartbeating, and this timer is the
  // escape hatch that still commits a live SSE response when upstream is silent.
  const armBootstrap = () => {
    if (bootstrapMs <= 0 || timer || committed || ended) return;
    timer = setTimeout(() => {
      if (!committed && !ended) commit();
    }, bootstrapMs);
    timer.unref?.();
  };

  return {
    get writableEnded() { return ended; },
    get headersSent() { return committed; },
    get statusCode() { return headerStatus; },
    set statusCode(v) { headerStatus = v; },
    // O10: first failure frame this stream carried (pre- or post-commit), or
    // null. Read by serveChatStream after the handler settles to classify the
    // request-ring row; it never changes what was sent to the client.
    get failure() { return failure; },
    setHeader(name, value) {
      if (committed) return realRes.setHeader(name, value);
      headerObj[name] = value;
      return this;
    },
    getHeader(name) {
      if (committed) return realRes.getHeader(name);
      return headerObj[name];
    },
    removeHeader(name) {
      if (committed) return realRes.removeHeader(name);
      delete headerObj[name];
    },
    writeHead(nextStatus, nextHeaders) {
      if (ended) return this;
      if (committed) return realRes.writeHead(nextStatus, nextHeaders);
      headerStatus = nextStatus;
      headerObj = { ...headerObj, ...(nextHeaders || {}) };
      return this;
    },
    flushHeaders() { commit(); },
    write(chunk, encoding, cb) {
      if (ended || disposed) return false;
      const buf = Buffer.isBuffer(chunk) ? chunk : Buffer.from(String(chunk), typeof encoding === 'string' ? encoding : 'utf8');
      if (committed) {
        // O10: a failure that lands after the commit gate is already on the
        // wire with HTTP 200, so it cannot rewrite the status — but the request
        // ring must still record its class. Walk the frames in this write (and
        // any frame carried over from the previous one) rather than parsing the
        // whole write as a single frame.
        scanFailure(buf);
        const ok = realRes.write(buf);
        if (typeof cb === 'function') process.nextTick(cb);
        return ok;
      }
      // First write of an uncommitted response: start the silence escape hatch
      // now. A write that commits immediately clears it in commit().
      armBootstrap();
      pending.push(buf);
      pendingBytes += buf.length;
      pendingText += pendingDecoder.write(buf);
      inspectPending();
      if (typeof cb === 'function') process.nextTick(cb);
      return true;
    },
    end(chunk, encoding, cb) {
      if (ended || disposed) return this;
      if (chunk !== undefined && chunk !== null) this.write(chunk, encoding);
      if (!committed && !ended) commit();
      if (!ended) {
        ended = true;
        clearTimer();
        realRes.end(typeof cb === 'function' ? cb : undefined);
      }
      return this;
    },
    on(event, listener) { realRes.on(event, listener); return this; },
    once(event, listener) { realRes.once(event, listener); return this; },
    removeListener(event, listener) { realRes.removeListener?.(event, listener); return this; },
    emit(event, ...args) { return realRes.emit?.(event, ...args); },
    // Cancellation releases the queued bytes, not just the timer: a route that
    // aborts before commit must not keep a held burst reachable from the gate.
    dispose() { disposed = true; ended = true; clearTimer(); pending = []; pendingBytes = 0; pendingFrames = 0; scanCarry = Buffer.alloc(0); pendingText = ''; },
    destroy(error) { disposed = true; ended = true; clearTimer(); pending = []; pendingBytes = 0; pendingFrames = 0; scanCarry = Buffer.alloc(0); pendingText = ''; return realRes.destroy?.(error); },
    get socket() { return realRes.socket; },
  };
}