The Leaky Abstraction: Why GenAI Streaming Breaks Your JSON (and How to Fix It)

// Resilient SSE & UTF-8 Stream Reassembler for Node.js / Cloud Run
import { StringDecoder } from 'node:string_decoder';
import { Transform, TransformCallback } from 'node:stream';

export interface SseEvent<T = unknown> {
  event?: string;
  data: T;
  id?: string;
  retry?: number;
}

export class SafeSseReassemblyStream extends Transform {
  private readonly decoder: StringDecoder;
  private buffer: string;

  constructor() {
    super({ readableObjectMode: true });
    this.decoder = new StringDecoder('utf8');
    this.buffer = '';
  }

  public _transform(
    chunk: Buffer | Uint8Array | string, 
    _encoding: string, 
    callback: TransformCallback
  ): void {
    try {
      const text = typeof chunk === 'string' 
        ? chunk 
        : this.decoder.write(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk));
      
      this.buffer += text;

      let delimiterIndex: number;
      while ((delimiterIndex = this.buffer.indexOf('\n\n')) !== -1) {
        const rawEvent = this.buffer.slice(0, delimiterIndex);
        this.buffer = this.buffer.slice(delimiterIndex + 2);

        const parsed = this.parseSseBlock(rawEvent);
        if (parsed) {
          this.push(parsed);
        }
      }

      callback();
    } catch (error) {
      callback(error instanceof Error ? error : new Error(String(error)));
    }
  }

  public _flush(callback: TransformCallback): void {
    try {
      this.buffer += this.decoder.end();

      if (this.buffer.trim().length > 0) {
        const parsed = this.parseSseBlock(this.buffer);
        if (parsed) {
          this.push(parsed);
        }
      }
      this.buffer = '';
      callback();
    } catch (error) {
      callback(error instanceof Error ? error : new Error(String(error)));
    }
  }

  private parseSseBlock(rawBlock: string): SseEvent | null {
    const lines = rawBlock.split(/\r?\n/);
    let eventType: string | undefined;
    let id: string | undefined;
    let retry: number | undefined;
    const dataLines: string[] = [];

    for (const line of lines) {
      if (line.startsWith(':') || line.trim() === '') continue;

      const colonIdx = line.indexOf(':');
      if (colonIdx === -1) continue;

      const field = line.slice(0, colonIdx).trim();
      const value = line.slice(colonIdx + 1).replace(/^\s/, '');

      switch (field) {
        case 'data': dataLines.push(value); break;
        case 'event': eventType = value; break;
        case 'id': id = value; break;
        case 'retry': retry = Number.parseInt(value, 10); break;
      }
    }

    if (dataLines.length === 0) return null;

    const rawData = dataLines.join('\n');
    let parsedData: unknown;

    try {
      const cleaned = rawData.trim();
      const parsedData = cleaned.startsWith('{') || cleaned.startsWith('[') 
        ? JSON.parse(cleaned) 
        : rawData;
    } catch {
      parsedData = rawData;
    }

    return {
      event: eventType,
      data: parsedData,
      id,
      retry,
    };
  }
}