import { PiProcessError } from './process-errors';

export const DEFAULT_MAX_PI_RPC_LINE_BYTES = 1024 * 1024;

type StrictLfJsonlFramerOptions = {
  maxLineBytes?: number;
  onRecord(record: unknown): void;
};

function protocolError(message: string): PiProcessError {
  return new PiProcessError('PI_RPC_PROTOCOL_ERROR', message);
}

export class StrictLfJsonlFramer {
  private readonly decoder = new TextDecoder('utf-8', { fatal: true });
  private readonly maxLineBytes: number;
  private readonly onRecord: (record: unknown) => void;
  private buffered = Buffer.alloc(0);
  private finished = false;

  constructor(options: StrictLfJsonlFramerOptions) {
    const maxLineBytes = options.maxLineBytes ?? DEFAULT_MAX_PI_RPC_LINE_BYTES;
    if (!Number.isSafeInteger(maxLineBytes) || maxLineBytes <= 0) {
      throw new Error('maxLineBytes must be a positive safe integer');
    }
    this.maxLineBytes = maxLineBytes;
    this.onRecord = options.onRecord;
  }

  push(chunk: Uint8Array | string): void {
    if (this.finished) throw protocolError('Pi RPC stdout continued after stream end');
    const bytes = typeof chunk === 'string'
      ? Buffer.from(chunk, 'utf8')
      : Buffer.from(chunk.buffer, chunk.byteOffset, chunk.byteLength);
    let offset = 0;

    while (offset < bytes.length) {
      const newline = bytes.indexOf(0x0a, offset);
      if (newline === -1) {
        this.append(bytes.subarray(offset));
        return;
      }
      this.append(bytes.subarray(offset, newline));
      this.emitLine();
      offset = newline + 1;
    }
  }

  finish(): void {
    if (this.finished) return;
    this.finished = true;
    if (this.buffered.length > 0) {
      throw protocolError('Pi RPC stdout ended with a partial line');
    }
  }

  private append(segment: Uint8Array): void {
    if (this.buffered.length + segment.byteLength > this.maxLineBytes) {
      throw protocolError(`Pi RPC stdout line exceeded ${this.maxLineBytes} bytes`);
    }
    if (segment.byteLength === 0) return;
    this.buffered = this.buffered.length === 0
      ? Buffer.from(segment)
      : Buffer.concat([this.buffered, segment], this.buffered.length + segment.byteLength);
  }

  private emitLine(): void {
    let line = this.buffered;
    this.buffered = Buffer.alloc(0);
    if (line.at(-1) === 0x0d) line = line.subarray(0, -1);
    if (line.length === 0) throw protocolError('Pi RPC stdout emitted a blank line');

    let source: string;
    try {
      source = this.decoder.decode(line);
    } catch (error) {
      throw new PiProcessError('PI_RPC_PROTOCOL_ERROR', 'Pi RPC stdout was not valid UTF-8', {
        cause: error,
      });
    }

    try {
      this.onRecord(JSON.parse(source));
    } catch (error) {
      if (error instanceof PiProcessError) throw error;
      throw new PiProcessError('PI_RPC_PROTOCOL_ERROR', 'Pi RPC stdout contained malformed JSON', {
        cause: error,
      });
    }
  }
}