// Generated by carbonsteel. DO NOT EDIT. Source: asyncapi.yaml#/operations (initMessage)

import { Buffer } from 'buffer';
import { APIResource } from '../resource';
import { WebSocket } from '../_shims/index';
import { LMNT_API_VERSION } from '../version';
import {
  MessageQueue,
  WEBSOCKET_NORMAL_CLOSE,
  WEBSOCKET_OPEN_STATE,
  processBinaryFrame,
} from '../_websocketRuntime';

const URL_STREAMING = 'wss://api.lmnt.com/v1/ai/speech/stream';
const WS_PATH = '/v1/ai/speech/stream';
const DEFAULT_BASE_URL = 'https://api.lmnt.com';

/**
 * Compose the streaming WebSocket URL from the client base URL.
 *
 * Rewrites the scheme (`http(s)` -> `ws(s)`) and appends `WS_PATH`, so a custom
 * `baseURL` on the Lmnt client (e.g. staging or a local proxy) flows through to
 * the WebSocket connection.
 */
function wsUrlFromBase(baseURL: string): string {
  const s = baseURL.replace(/\/+$/, '');
  if (s.startsWith('https://')) return 'wss://' + s.slice('https://'.length) + WS_PATH;
  if (s.startsWith('http://')) return 'ws://' + s.slice('http://'.length) + WS_PATH;
  return s + WS_PATH;
}

export interface Timestamp {
  /**
   * The text segment
   */
  text: string;

  /**
   * The time at which the text starts, in seconds
   */
  start: number;

  /**
   * The spoken duration of the text segment, in seconds
   */
  duration: number;
}

export interface ErrorBody {
  /**
   * Slug identifying the error category.
   */
  type:
    | 'invalid_request_error'
    | 'authentication_error'
    | 'permission_error'
    | 'not_found_error'
    | 'payment_required_error'
    | 'rate_limit_error'
    | 'internal_server_error';

  /**
   * Human-readable error message.
   */
  message: string;
}

export type SpeechSessionMessage =
  | SpeechSessionAudio
  | SpeechSessionReady
  | SpeechSessionTimestamps
  | SpeechSessionFlushComplete
  | SpeechSessionResetComplete
  | SpeechSessionError;

/**
 * Binary audio data returned from the server
 */
export interface SpeechSessionAudio {
  type: 'audio';
  audio: ArrayBuffer | Buffer;
}

/**
 * First message sent by the server, confirming the session is established.
 */
export interface SpeechSessionReady {
  type: 'ready';
  request_id: string;
}

/**
 * Timestamps for the audio chunk that was just streamed, if requested in `init`.
 */
export interface SpeechSessionTimestamps {
  type: 'timestamps';
  timestamps?: Array<Timestamp>;
}

/**
 * Acknowledgement that a flush command has been completed.
 *
 * The `nonce` matches the one carried by the original flush, allowing you to determine when it has completed.
 */
export interface SpeechSessionFlushComplete {
  type: 'flush_complete';
  nonce: number;
}

/**
 * Acknowledgement that a reset command has been completed.
 *
 * The `nonce` matches the one carried by the original reset, allowing you to discard any remaining in-flight speech before the reset.
 */
export interface SpeechSessionResetComplete {
  type: 'reset_complete';
  nonce: number;
}

/**
 * Error envelope returned by the server. Connection closes immediately afterward.
 */
export interface SpeechSessionError {
  type: 'error';
  error: ErrorBody;
  request_id: string;
}

export class SpeechSession {
  private socket: WebSocket;
  private outMessages: any[] = [];
  private inMessages: MessageQueue;
  private nextNonce: number = 1;

  /** Per-session request id, populated when the server emits its `ready` frame. */
  request_id: string | null = null;

  constructor(apiKey: string, options: SpeechSessionParams, baseURL: string = DEFAULT_BASE_URL) {
    this.outMessages = [];
    this.inMessages = new MessageQueue();

    this.socket = new WebSocket(wsUrlFromBase(baseURL));
    this.setupWebSocket();

    this.sendMessage({
      'X-API-Key': apiKey,
      'lmnt-version': LMNT_API_VERSION,
      voice: options.voice,
      format: options.format,
      language: options.language,
      return_timestamps: options.return_timestamps,
      sample_rate: options.sample_rate,
    });
  }

  private setupWebSocket() {
    this.socket.onerror = this.onError.bind(this);
    this.socket.onmessage = this.onMessage.bind(this);
    this.socket.onopen = this.onOpen.bind(this);
    this.socket.onclose = this.onClose.bind(this);
  }

  private onError(event: any) {
    console.error('WebSocket error:', event);
    this.inMessages.finish();
  }

  private onOpen() {
    this.flushMessages();
  }

  private onClose(event: any) {
    if (event.code !== WEBSOCKET_NORMAL_CLOSE) {
      console.warn(`WebSocket closed unexpectedly with code: ${event.code}, reason: ${event.reason}`);
    }
    this.socket = null as any;
    this.inMessages.finish();
  }

  private onMessage(event: MessageEvent) {
    const isText = typeof event.data === 'string';
    this.inMessages.push({
      type: isText ? 'text' : 'binary',
      data: event.data,
    });
  }

  /**
   * Close the SpeechSession.
   */
  close() {
    if (this.socket) {
      this.socket.close();
      this.socket = null as any;
    }
  }

  /**
   * Send text to the server to append into the text stream.
   */
  sendText(text: string) {
    this.sendMessage({ type: 'text', text });
  }

  /**
   * Force the server to generate speech for all buffered text in the stream.
   *
   * @returns The nonce used for this flush command
   */
  sendFlush(): number {
    const nonce = this.nextNonce++;
    this.sendMessage({ type: 'flush', nonce });
    return nonce;
  }

  /**
   * Drop the server's buffered text without generating speech for it.
   *
   * @returns The nonce used for this reset command
   */
  sendReset(): number {
    const nonce = this.nextNonce++;
    this.sendMessage({ type: 'reset', nonce });
    return nonce;
  }

  /**
   * Inform the server you're done appending text to this session and want it to close when the server has finished dispatching speech.
   */
  sendFinish() {
    this.sendMessage({ type: 'finish' });
  }

  /**
   * Iterate over messages from the server.
   *
   * @returns AsyncIterator<SpeechSessionMessage>
   */
  async *[Symbol.asyncIterator](): AsyncIterator<SpeechSessionMessage> {
    while (true) {
      const message = await this.inMessages.next();
      if (message === null) {
        return;
      }

      if (message.type === 'text') {
        const parsed = this.parseTextMessage(message.data);
        if (parsed !== null) {
          yield parsed;
        }
      } else {
        const payload = await processBinaryFrame(message.data);
        yield { type: 'audio', audio: payload };
      }
    }
  }

  async streamAudioToFile(path: string): Promise<void> {
    const fs = await import('fs');
    const stream = fs.createWriteStream(path);
    for await (const message of this) {
      if (message.type === 'audio') {
        const chunk = Buffer.isBuffer(message.audio) ? message.audio : Buffer.from(message.audio);
        stream.write(chunk);
      }
    }
    await new Promise<void>((resolve, reject) => {
      stream.end((err: Error | null) => (err ? reject(err) : resolve()));
    });
  }

  private parseTextMessage(textData: string): SpeechSessionMessage | null {
    const messageJson = JSON.parse(textData);
    const messageType = messageJson?.type;

    if (messageType === 'ready') {
      this.request_id = messageJson.request_id ?? null;
      return messageJson as SpeechSessionReady;
    }

    if (messageType === 'timestamps') {
      return messageJson as SpeechSessionTimestamps;
    }

    if (messageType === 'flush_complete') {
      return messageJson as SpeechSessionFlushComplete;
    }

    if (messageType === 'reset_complete') {
      return messageJson as SpeechSessionResetComplete;
    }

    if (messageType === 'error') {
      return messageJson as SpeechSessionError;
    }

    return null;
  }

  private sendMessage(message: any) {
    this.outMessages.push(message);
    this.flushMessages();
  }

  private flushMessages() {
    if (this.socket?.readyState === WEBSOCKET_OPEN_STATE) {
      while (this.outMessages.length) {
        const message = this.outMessages.shift();
        this.socket.send(JSON.stringify(message));
      }
    }
  }
}

export class Sessions extends APIResource {
  /**
   * Stream text to our servers and receive generated speech in real-time. Great for latency-sensitive applications and situations where you don't have all the text upfront.
   */
  create(body: SpeechSessionParams) {
    return new SpeechSession(this._client.apiKey, body, this._client.baseURL);
  }
}

export interface SpeechSessionParams {
  /**
   * The voice ID to use for speech generation, obtained from 'List voices' API
   */
  voice: string;

  /**
   * The desired output format of the audio.
   * - `mp3`: 96kbps MP3 audio.
   * - `pcm_s16le`: PCM signed 16-bit little-endian audio.
   * - `pcm_f32le`: PCM 32-bit floating-point little-endian audio.
   * - `ulaw`: 8-bit G711 µ-law audio with a WAV header.
   * - `webm`: WebM format with Opus audio codec.
   */
  format?: 'mp3' | 'pcm_s16le' | 'pcm_f32le' | 'ulaw' | 'webm';

  language?:
    | 'auto'
    | 'ar'
    | 'as'
    | 'bn'
    | 'cs'
    | 'da'
    | 'de'
    | 'en'
    | 'es'
    | 'fi'
    | 'fr'
    | 'hi'
    | 'id'
    | 'it'
    | 'ja'
    | 'ko'
    | 'ml'
    | 'mr'
    | 'nl'
    | 'pl'
    | 'pt'
    | 'ru'
    | 'sk'
    | 'sv'
    | 'ta'
    | 'te'
    | 'th'
    | 'tr'
    | 'uk'
    | 'ur'
    | 'vi'
    | 'zh';

  /**
   * Controls whether the server will return timestamps for the generated speech
   */
  return_timestamps?: boolean;

  /**
   * The desired output audio sample rate
   */
  sample_rate?: 24000 | 16000 | 8000;
}

export declare namespace Sessions {
  export { type SpeechSessionParams as SpeechSessionParams };
  export { SpeechSession as SpeechSession };
}