///|
/// A source of pre-encoded 20 ms Opus frames. `None` marks end of stream.
pub(open) trait AudioSource {
  async fn next_frame(Self) -> Bytes?
}

///|
/// Discord's canonical Opus silence frame.
pub const SILENCE_FRAME : Bytes = b"\xF8\xFF\xFE"

///|
async fn pace_until(deadline : Int64) -> Unit {
  let remaining = deadline - @clock.now_ms()
  if remaining > 0L {
    @async.sleep(remaining.to_int())
  }
}

///|
async fn send_opus_packet(
  cipher : TransportCipher,
  udp : &VoiceUdp,
  dave : DaveMachine?,
  sequence : UInt16,
  timestamp : UInt,
  ssrc : UInt,
  frame : Bytes,
  on_dave_error : (String) -> Unit,
) -> Bool {
  let header = build_rtp_header(sequence~, timestamp~, ssrc~)
  let encrypted_frame = match dave {
    Some(machine) =>
      Some(machine.encrypt_opus_frame(ssrc~, frame)) catch {
        DaveInvalid(reason~) => {
          on_dave_error(reason)
          None
        }
        DaveUnavailable(reason~) => {
          on_dave_error(reason)
          None
        }
        DaveInternal(reason~) => {
          on_dave_error(reason)
          None
        }
      }
    None => Some(frame)
  }
  guard encrypted_frame is Some(encrypted) else { return false }
  udp.send(cipher.seal(header~, encrypted))
  true
}

///|
priv struct SendLoopState {
  mut sequence : UInt16
  mut timestamp : UInt
  mut deadline : Int64
}

///|
/// Internal paced sender used by the M4 connection driver.
pub async fn run_send_loop(
  source : &AudioSource,
  cipher : TransportCipher,
  udp : &VoiceUdp,
  gateway : VoiceGateway,
  ssrc~ : UInt,
  initial_sequence? : UInt16 = 0,
  initial_timestamp? : UInt = 0,
  stop? : Ref[Bool],
  frame_duration_ms? : Int = 20,
  dave? : DaveMachine,
  on_dave_error? : (String) -> Unit = _ => (),
) -> Unit {
  let state = SendLoopState::{
    sequence: initial_sequence,
    timestamp: initial_timestamp,
    deadline: @clock.now_ms(),
  }
  gateway.send_json(encode_speaking(ssrc~, flags=1))
  for ;; {
    if stop is Some(flag) && flag.val {
      break
    }
    guard source.next_frame() is Some(frame) else { break }
    let sent = send_opus_packet(
      cipher,
      udp,
      dave,
      state.sequence,
      state.timestamp,
      ssrc,
      frame,
      on_dave_error,
    )
    if sent {
      state.sequence += 1
    }
    state.timestamp += 960
    state.deadline += frame_duration_ms.to_int64()
    pace_until(state.deadline)
  }
  // Discord recommends five silence frames to flush decoder state on stop.
  for _ in 0..<5 {
    let sent = send_opus_packet(
      cipher,
      udp,
      dave,
      state.sequence,
      state.timestamp,
      ssrc,
      SILENCE_FRAME,
      on_dave_error,
    )
    if sent {
      state.sequence += 1
    }
    state.timestamp += 960
    state.deadline += frame_duration_ms.to_int64()
    pace_until(state.deadline)
  }
  gateway.send_json(encode_speaking(ssrc~, flags=0))
}