// One Deepgram streaming connection, carrying one participant's speech.
//
// Deepgram offers three speech-to-text paths. The pre-recorded REST endpoint is
// the cheaper one per minute, but its responses carry no CORS headers, so a
// browser cannot call it without a proxy server; the streaming WebSocket
// endpoint is reachable directly. Since CommonRoom has no server of its own,
// streaming is what we use.
//
// Browsers cannot set an Authorization header on a WebSocket, so the API key
// travels as a subprotocol — `Sec-WebSocket-Protocol: token, <key>`, which is
// Deepgram's documented scheme for client-side connections. The key therefore
// leaves the browser only in the TLS handshake with Deepgram itself, and it is
// never shared with the room.
//
// The connection is opened lazily on the first frame of speech and then held
// open across pauses with KeepAlive messages, which are not billed. A separate
// connection per speaker is what gives the transcript its attribution: there is
// no diarization to interpret, since each socket only ever hears one person.

const ENDPOINT = 'wss://api.deepgram.com/v1/listen'

/** Deepgram closes an idle socket after 10 s; the docs ask for a KeepAlive
 *  every 3-5 s. */
const TICK_MS = 3000
const KEEPALIVE_AFTER_MS = 2500
/** Give the socket back after a long silence rather than pinging it forever. */
const IDLE_CLOSE_MS = 120_000
/** Frames buffered while the socket is still opening (~4 s of speech). */
const QUEUE_CAP = 100
/** How long to wait for trailing results after asking the stream to close. */
const DRAIN_MS = 3000

const QUERY = {
  model: 'nova-3',
  language: 'en',
  smart_format: 'true',
  // Our own gate already segments speech, and interim results would triple the
  // message rate for text we would only overwrite.
  interim_results: 'false',
  encoding: 'linear16',
  channels: '1',
  endpointing: '400'
}

export interface DeepgramHandlers {
  transcript: (text: string) => void
  /** Samples actually written to the socket, for the cost readout. */
  sent: (samples: number) => void
  /** A problem worth surfacing. `fatal` means stop trying entirely. */
  failure: (message: string, fatal: boolean) => void
}

/** A subprotocol must be an RFC 7230 token, which excludes spaces and most
 *  punctuation. A key with a stray character would make the WebSocket
 *  constructor throw rather than fail as a rejected key, so it is checked up
 *  front where a comprehensible message can still be given. Deepgram keys are
 *  hexadecimal and pass comfortably. */
export const isUsableApiKey = (key: string): boolean =>
  /^[A-Za-z0-9!#$%&'*+\-.^_`|~]+$/.test(key)

export class DeepgramStream {
  private ws: WebSocket | null = null
  private queue: Int16Array[] = []
  private opened = false
  private failures = 0
  private lastAudioAt = 0
  private timer: number | null = null
  /** Set once the stream is winding down: no new audio, but results still
   *  arriving from the server are delivered. */
  private draining = false

  constructor(
    private apiKey: string,
    private sampleRate: number,
    private handlers: DeepgramHandlers
  ) {}

  send(pcm: Int16Array) {
    if (this.draining) return
    this.lastAudioAt = Date.now()
    if (!this.ws) this.open()
    const ws = this.ws
    if (!ws) return
    if (ws.readyState === WebSocket.CONNECTING) {
      // Everything up to here is speech we decided was worth paying for, so it
      // waits for the socket rather than being dropped — but only up to a
      // point, in case the connection never comes up.
      if (this.queue.length < QUEUE_CAP) this.queue.push(pcm)
      return
    }
    if (ws.readyState !== WebSocket.OPEN) return
    ws.send(pcm)
    this.handlers.sent(pcm.length)
  }

  /** End of utterance: ask Deepgram to flush what it has rather than waiting
   *  for its own endpointing to expire. */
  finalize() {
    const ws = this.ws
    if (ws && ws.readyState === WebSocket.OPEN) {
      ws.send(JSON.stringify({type: 'Finalize'}))
    }
  }

  /** Stop sending, but stay open briefly so the last utterance still lands. */
  stop() {
    if (this.draining) return
    this.draining = true
    this.queue = []
    if (this.timer !== null) clearInterval(this.timer)
    this.timer = null
    const ws = this.ws
    if (!ws) return
    if (ws.readyState === WebSocket.OPEN) {
      ws.send(JSON.stringify({type: 'CloseStream'}))
      window.setTimeout(() => ws.close(), DRAIN_MS)
    } else {
      ws.close()
    }
  }

  private open() {
    const params = new URLSearchParams({
      ...QUERY,
      sample_rate: String(this.sampleRate)
    })
    let ws: WebSocket
    try {
      ws = new WebSocket(`${ENDPOINT}?${params}`, ['token', this.apiKey])
    } catch {
      this.handlers.failure(
        'That Deepgram API key could not be used to open a connection — check it for stray characters.',
        true
      )
      return
    }
    ws.binaryType = 'arraybuffer'
    this.ws = ws
    this.opened = false

    ws.onopen = () => {
      this.opened = true
      this.failures = 0
      for (const pcm of this.queue.splice(0)) {
        ws.send(pcm)
        this.handlers.sent(pcm.length)
      }
    }
    ws.onmessage = ev => this.onMessage(ev)
    // An error is always followed by a close, which carries more information.
    ws.onerror = () => undefined
    ws.onclose = ev => this.onClose(ws, ev)

    if (this.timer === null) {
      this.timer = window.setInterval(() => this.tick(), TICK_MS)
    }
  }

  private onMessage(ev: MessageEvent) {
    if (typeof ev.data !== 'string') return
    let msg: {
      type?: string
      channel?: {alternatives?: {transcript?: string}[]}
      description?: string
      message?: string
    }
    try {
      msg = JSON.parse(ev.data)
    } catch {
      return
    }
    if (msg.type === 'Results') {
      const text = msg.channel?.alternatives?.[0]?.transcript
      if (typeof text === 'string' && text.trim()) this.handlers.transcript(text)
      return
    }
    if (msg.type === 'Error') {
      const why = msg.description ?? msg.message ?? 'unspecified'
      this.handlers.failure(`Deepgram reported an error: ${why}`, false)
    }
  }

  private onClose(ws: WebSocket, ev: CloseEvent) {
    // A socket we already gave up on (see the idle path in `tick`) may close
    // after its replacement is up; it has nothing left to say.
    if (this.ws !== ws) return
    const wasOpen = this.opened
    this.ws = null
    this.opened = false
    this.queue = []
    if (this.timer !== null) clearInterval(this.timer)
    this.timer = null
    if (this.draining) return

    if (!wasOpen) {
      // A browser deliberately hides the HTTP status of a failed WebSocket
      // handshake, so a rejected key and an unreachable network are
      // indistinguishable here. Say so, and stop after the second attempt
      // rather than reconnecting into a wall on every utterance.
      this.failures++
      this.handlers.failure(
        `Could not open a Deepgram connection${
          ev.reason ? ` (${ev.reason})` : ''
        } — check the API key and your network.`,
        this.failures >= 2
      )
      return
    }
    if (ev.code !== 1000) {
      this.handlers.failure(
        `The Deepgram connection dropped${
          ev.reason ? ` (${ev.reason})` : ''
        }; it will reopen when someone next speaks.`,
        false
      )
    }
  }

  private tick() {
    const ws = this.ws
    if (!ws || ws.readyState !== WebSocket.OPEN) return
    const idle = Date.now() - this.lastAudioAt
    if (idle > IDLE_CLOSE_MS) {
      // Nothing can be in flight after two minutes of silence, so this one is
      // released outright; the next utterance opens a fresh socket.
      ws.send(JSON.stringify({type: 'CloseStream'}))
      ws.close()
      this.ws = null
      if (this.timer !== null) clearInterval(this.timer)
      this.timer = null
      return
    }
    if (idle > KEEPALIVE_AFTER_MS) ws.send(JSON.stringify({type: 'KeepAlive'}))
  }
}