import type { Capabilities, Provider, Request, Response, StreamEvent } from '@wrongstack/core';
import { ProviderError, StreamHangError } from '@wrongstack/core';
import { parseProviderHttpError } from './error-parse.js';
import { isDebugStreamEnabled, pushDebugChunkStats } from './stream-debug-state.js';
import { Readable } from 'node:stream';

type Response2 = {
  ok: boolean;
  status: number;
  text(): Promise<string>;
  body: ReadableStream<Uint8Array> | NodeJS.ReadableStream | null;
};

/** Configuration for WireAdapter stream-level debugging and hang detection. */
export interface WireAdapterStreamOptions {
  /**
   * When true, accumulate per-chunk stats into the shared debug-sink
   * (stream-debug-state.ts). The sink batches every 200 ms and pushes to
   * a registered callback. The CLI default callback writes to stderr; the
   * TUI replaces it with a reducer dispatch that renders in StatusBar line 3,
   * keeping all output inside Ink's layout.
   *
   * Controlled by WRONGSTACK_DEBUG_STREAM=1 env var or the runtime
   * /settings debug-stream toggle.
   */
  debugStream?: boolean | undefined;
  /**
   * Maximum time (ms) to wait for the next chunk of data before declaring
   * a stream hang. Default: 60_000 (60 seconds). Set to 0 to disable.
   * When a hang is detected, a StreamHangError is thrown so the agent
   * loop can retry the iteration.
   */
  streamHangTimeoutMs?: number | undefined;
}

/** Validate fetchImpl response has required fields; normalize missing body to null. */
function validateResponse(res: unknown): asserts res is Response2 {
  const r = res as Record<string, unknown> | undefined;
  if (r === undefined || typeof r.ok !== 'boolean' || typeof r.status !== 'number') {
    throw new Error('fetchImpl returned invalid response shape — expected { ok, status, text, body }');
  }
  // If body is absent, null, or undefined on a plain object (not a native Response
  // with a read-only getter), normalize it to null so callers can safely use it.
  // Native Response objects always have a body getter — no mutation needed.
  if (!('body' in r) || r.body === undefined) {
    // Only set on plain objects — native Response.body is read-only
    const proto = Object.getPrototypeOf(r);
    if (proto === Object.prototype || proto === null) {
      r.body = null;
    }
  }
}

async function safeText(res: Response2): Promise<string> {
  try {
    return await res.text();
  } catch {
    return '';
  }
}

/**
 * Feed debug-chunk stats into the shared singleton sink. The sink batches
 * and throttles writes so the TUI can render them inside Ink's StatusBar
 * line 3 (~5 Hz) instead of raw stderr interfering with the terminal layout.
 */
function logRawChunk(
  _providerId: string,
  _chunkIndex: number,
  bytes: Uint8Array,
  deltaMs: number,
): void {
  pushDebugChunkStats(bytes.length, deltaMs);
}

const DEFAULT_STREAM_HANG_TIMEOUT_MS = 60_000;

/**
 * Shared HTTP mechanics for streaming providers.
 * Providers extend this to get:
 *   - canonical error handling (ProviderError with retryable flag)
 *   - SSE body parsing via parseSSE()
 *   - abort signal wiring
 *   - optional raw-stream debug logging
 *   - optional stream hang detection
 *
 * Subclasses implement the abstract members to provide their specific wire format.
 */
export abstract class WireAdapter implements Provider {
  abstract readonly id: string;
  abstract readonly capabilities: Capabilities;

  protected readonly debugStream: boolean;
  protected readonly streamHangTimeoutMs: number;

  constructor(
    protected readonly apiKey: string,
    protected readonly baseUrl: string,
    public readonly fetchImpl: typeof fetch = fetch,
    streamOpts: WireAdapterStreamOptions = {},
  ) {
    if (!apiKey) throw new Error(`${this.constructor.name}: apiKey required`);
    this.debugStream = streamOpts.debugStream ?? false;
    this.streamHangTimeoutMs = streamOpts.streamHangTimeoutMs ?? DEFAULT_STREAM_HANG_TIMEOUT_MS;
  }

  async complete(req: Request, opts: { signal: AbortSignal }): Promise<Response> {
    const { aggregateStream } = await import('./aggregate.js');
    return aggregateStream(this.stream(req, opts));
  }

  async *stream(req: Request, opts: { signal: AbortSignal }): AsyncIterable<StreamEvent> {
    const url = this.buildUrl(req);
    const headers = this.buildHeaders(req);
    const body = this.buildBody(req);

    let httpRes: Response2;
    try {
      const raw = await this.fetchImpl(url, {
        method: 'POST',
        headers,
        body: JSON.stringify(body),
        signal: opts.signal,
      });
      validateResponse(raw);
      httpRes = raw as Response2;
    } catch (err) {
      if (opts.signal.aborted) throw err;
      throw new ProviderError(err instanceof Error ? err.message : String(err), 0, true, this.id, {
        cause: err,
        body: { message: err instanceof Error ? err.message : String(err) },
      });
    }

    if (!httpRes.ok) {
      const text = await safeText(httpRes);
      throw this.translateError(httpRes.status, text);
    }

    let sseBody = httpRes.body;
    if (!sseBody) {
      // No body — emit nothing
      return;
    }

    // Layer 1: debug logging — wrap the stream to log raw bytes.
    // Checks both the instance-level option (set at construction) AND the
    // runtime singleton (flipped via /settings or setDebugStreamEnabled) so
    // toggles take effect on the next request without recreating providers.
    if (this.debugStream || isDebugStreamEnabled()) {
      sseBody = this.wrapDebugStream(sseBody);
    }

    // Layer 2: hang detection — wrap with timeout-aware reader
    if (this.streamHangTimeoutMs > 0) {
      sseBody = this.wrapWithHangDetection(sseBody, req.model);
    }

    yield* this.parseStream(sseBody, req.model);
  }

  /**
   * Wrap a readable stream body to log a compact status line per incoming
   * byte chunk to stderr. This is a diagnostic tool for tracking stream
   * activity — chunk count, sizes, and inter-chunk deltas — without
   * printing payload contents.
   */
  private wrapDebugStream(
    body: ReadableStream<Uint8Array> | NodeJS.ReadableStream,
  ): ReadableStream<Uint8Array> | NodeJS.ReadableStream {
    // Node.js Readable stream — use async iterator
    if (isNodeReadable(body)) {
      return this.wrapDebugNodeStream(body as NodeJS.ReadableStream) as NodeJS.ReadableStream;
    }
    // Web ReadableStream — wrap reader
    return this.wrapDebugWebStream(body as ReadableStream<Uint8Array>);
  }

  private wrapDebugNodeStream(body: NodeJS.ReadableStream): NodeJS.ReadableStream {
    let lastChunkTime = Date.now();
    let chunkIndex = 0;
    const providerId = this.id;

    return Readable.from(
      (async function* () {
        for await (const chunk of body) {
          const bytes: Uint8Array =
            typeof chunk === 'string'
              ? new TextEncoder().encode(chunk)
              : new Uint8Array(chunk.buffer, chunk.byteOffset, chunk.byteLength);
          const now = Date.now();
          logRawChunk(providerId, chunkIndex++, bytes, now - lastChunkTime);
          lastChunkTime = now;
          yield chunk;
        }
      })(),
    );
  }

  private wrapDebugWebStream(
    body: ReadableStream<Uint8Array>,
  ): ReadableStream<Uint8Array> {
    let lastChunkTime = Date.now();
    let chunkIndex = 0;
    const self = this;
    const reader = body.getReader();

    return new ReadableStream<Uint8Array>({
      async pull(controller) {
        const { done, value } = await reader.read();
        if (done) {
          controller.close();
          return;
        }
        if (value) {
          const now = Date.now();
          logRawChunk(self.id, chunkIndex++, value, now - lastChunkTime);
          lastChunkTime = now;
        }
        controller.enqueue(value);
      },
      cancel(reason) {
        reader.cancel(reason);
      },
    });
  }

  /**
   * Wrap a readable stream to detect hangs — when no data arrives for
   * longer than `streamHangTimeoutMs`. When a hang is detected, throws
   * `StreamHangError` so the caller can retry or fall back.
   */
  private wrapWithHangDetection(
    body: ReadableStream<Uint8Array> | NodeJS.ReadableStream,
    model: string,
  ): ReadableStream<Uint8Array> | NodeJS.ReadableStream {
    if (isNodeReadable(body)) {
      return this.wrapHangNodeStream(body as NodeJS.ReadableStream, model);
    }
    return this.wrapHangWebStream(body as ReadableStream<Uint8Array>, model);
  }

  private wrapHangNodeStream(
    body: NodeJS.ReadableStream,
    model: string,
  ): NodeJS.ReadableStream {
    // Node Readable → Web ReadableStream, then use the race-based
    // web wrapper that properly detects hangs even when no chunks arrive.
    // The for-await approach only checks BETWEEN chunks — a stalled stream
    // that never yields another chunk would freeze indefinitely.
    const webStream = Readable.toWeb(body as Readable);
    const wrappedWeb = this.wrapHangWebStream(webStream as ReadableStream<Uint8Array>, model);
    return Readable.fromWeb(wrappedWeb as unknown as ReadableStream) as unknown as NodeJS.ReadableStream;
  }

  private wrapHangWebStream(
    body: ReadableStream<Uint8Array>,
    model: string,
  ): ReadableStream<Uint8Array> {
    const startTime = Date.now();
    let bytesReceived = 0;
    const timeout = this.streamHangTimeoutMs;
    const providerId = this.id;
    const reader = body.getReader();

    return new ReadableStream<Uint8Array>({
      async pull(controller) {
        // Race the read against a hang timeout
        const readPromise = reader.read();
        const timeoutPromise = new Promise<{ timedOut: true }>((resolve) => {
          setTimeout(() => resolve({ timedOut: true }), timeout);
        });

        const result = await Promise.race([readPromise, timeoutPromise]);

        if ('timedOut' in result && result.timedOut) {
          // The read is still pending — this is a hang.
          // Cancel the reader and throw.
          reader.cancel('stream hang detected').catch((err) => console.debug(`[wire-adapter] cancel after stream hang failed: ${err}`));
          const elapsedMs = Date.now() - startTime;
          throw new StreamHangError({
            providerId,
            model,
            hangTimeoutMs: timeout,
            bytesReceived,
            elapsedMs,
          });
        }

        const { done, value } = result as Awaited<ReturnType<typeof reader.read>>;
        if (done) {
          controller.close();
          return;
        }
        if (value) {
          bytesReceived += value.length;
        }
        controller.enqueue(value);
      },
      cancel(reason) {
        reader.cancel(reason);
      },
    });
  }

  // ─── Abstract / overridable ───────────────────────────────────────────────

  /** HTTP endpoint for this provider's chat completions / messages API. */
  protected abstract buildUrl(req: Request): string;

  /** Per-request headers. `apiKey` is already in scope — call `super.buildHeaders` first. */
  protected buildHeaders(_req: Request): Record<string, string> {
    return {
      'content-type': 'application/json',
      accept: 'text/event-stream',
    };
  }

  /** Map Request fields to the wire request body. */
  protected abstract buildBody(req: Request): Record<string, unknown>;

  /** Translate wire SSE events into canonical StreamEvent[]. */
  protected abstract parseStream(
    body: ReadableStream<Uint8Array> | NodeJS.ReadableStream | null,
    fallbackModel: string,
  ): AsyncIterable<StreamEvent>;

  /** Build a ProviderError from an HTTP failure response. */
  protected translateError(status: number, body: string): ProviderError {
    return parseProviderHttpError(this.id, status, body);
  }
}

function isNodeReadable(b: unknown): boolean {
  return (
    !!b &&
    typeof b === 'object' &&
    typeof (b as { pipe?: unknown | undefined }).pipe === 'function' &&
    typeof (b as { on?: unknown | undefined }).on === 'function'
  );
}
