///|
fn voice_state_command(
  guild_id : @model.Id[@model.GuildMarker],
  channel_id : @model.Id[@model.ChannelMarker]?,
  self_mute : Bool,
  self_deaf : Bool,
) -> Json {
  let channel : Json = match channel_id {
    Some(id) => id.to_json()
    None => Json::null()
  }
  {
    "op": 4,
    "d": {
      "guild_id": guild_id.to_json(),
      "channel_id": channel,
      "self_mute": self_mute,
      "self_deaf": self_deaf,
    },
  }
}

///|
async fn GatewayCtx::request_voice_credentials(
  self : GatewayCtx,
  guild_id : @model.Id[@model.GuildMarker],
  channel_id : @model.Id[@model.ChannelMarker],
  self_mute : Bool,
  self_deaf : Bool,
  timeout_ms : Int,
) -> @voice.VoiceCredentials {
  let user_id = self.ready_.user.id
  let state = self.collector_.register(Events::voice_state_update(), update => {
    update.user_id == user_id && update.guild_id == Some(guild_id)
  })
  let server = self.collector_.register(Events::voice_server_update(), update => {
    update.guild_id == guild_id && update.endpoint is Value(_)
  })
  defer state.cancel()
  defer server.cancel()
  self.shard_.send(
    voice_state_command(guild_id, Some(channel_id), self_mute, self_deaf),
  )
  @async.with_task_group(group => {
    let state_task = group.spawn(no_wait=true, () => state.wait(timeout_ms~))
    let server_task = group.spawn(no_wait=true, () => server.wait(timeout_ms~))
    guard state_task.wait() is Some(state_update) else {
      server_task.cancel()
      raise @voice.VoiceError::JoinTimeout
    }
    guard server_task.wait() is Some(server_update) else {
      raise @voice.VoiceError::JoinTimeout
    }
    guard server_update.endpoint is Value(endpoint) else {
      raise @voice.VoiceError::JoinTimeout
    }
    {
      server_id: guild_id.to_string(),
      session_id: state_update.session_id,
      token: server_update.token,
      endpoint,
    }
  })
}

///|
async fn GatewayCtx::watch_voice_server_updates(
  self : GatewayCtx,
  guild_id : @model.Id[@model.GuildMarker],
  channel_id : @model.Id[@model.ChannelMarker],
  self_mute : Bool,
  self_deaf : Bool,
  timeout_ms : Int,
  connection : @voice.VoiceConnection,
  rejoin_in_progress : Ref[Bool],
) -> Unit noraise {
  let update = self.wait_for(Events::voice_server_update(), event => {
    event.guild_id == guild_id &&
    event.endpoint is Value(_) &&
    !rejoin_in_progress.val
  }) catch {
    _ => None
  }
  guard update is Some(_) && connection.state() is Ready else { return }
  connection.disconnect() catch {
    _ => return
  }
  try
    self.join_voice(guild_id, channel_id, self_mute~, self_deaf~, timeout_ms~)
  catch {
    _ => ()
  } noraise {
    _ => ()
  }
}

///|
async fn wait_ready_or_disconnect(
  connection : @voice.VoiceConnection,
  timeout_ms : Int,
) -> Unit {
  errdefer @async.protect_from_cancel(() => {
    connection.disconnect() catch {
      _ => ()
    }
  })
  connection.wait_ready(timeout_ms~) catch {
    error if @async.is_being_cancelled() => {
      @async.pause()
      raise error
    }
    error => raise error
  }
}

///|
/// Join a guild voice channel and wait until its voice transport is ready.
///
/// A second call for the same guild returns the registered connection; moving
/// an existing connection to another channel is outside this API. Discord may
/// send no response events when the channel is full and the bot lacks
/// `MOVE_MEMBERS`; that case raises `VoiceError::JoinTimeout`.
///
/// The bot must connect the main Gateway with the `GUILD_VOICE_STATES` intent.
/// This method raises `VoiceError::MissingIntent` before sending Opcode 4 when
/// that intent is absent.
///
/// Pass `telemetry` to observe non-fatal voice gateway and media diagnostics.
pub async fn GatewayCtx::join_voice(
  self : GatewayCtx,
  guild_id : @model.Id[@model.GuildMarker],
  channel_id : @model.Id[@model.ChannelMarker],
  self_mute? : Bool = false,
  self_deaf? : Bool = false,
  timeout_ms? : Int = 10000,
  telemetry? : (@voice.VoiceTelemetry) -> Unit = _ => (),
) -> @voice.VoiceConnection {
  let required_intents = Events::voice_state_update().required_intents()
  if !self.intents_.contains(required_intents) {
    raise @voice.VoiceError::MissingIntent(intent="GUILD_VOICE_STATES")
  }
  let join_gate = match self.voice_join_gates_.get(guild_id) {
    Some(gate) => gate
    None => {
      let gate = @async.Semaphore(1)
      self.voice_join_gates_[guild_id] = gate
      gate
    }
  }
  join_gate.acquire()
  defer join_gate.release()
  if self.voice_connections_.get(guild_id) is Some(existing) {
    if existing.state() is Closed(..) {
      self.voice_connections_.remove(guild_id)
    } else {
      return existing
    }
  }
  let credentials = self.request_voice_credentials(
    guild_id, channel_id, self_mute, self_deaf, timeout_ms,
  )
  let rejoin_in_progress = Ref(false)
  let leave = () => {
    defer self.voice_connections_.remove(guild_id)
    self.shard_.send(voice_state_command(guild_id, None, self_mute, self_deaf))
  }
  let rejoin = () => {
    rejoin_in_progress.val = true
    defer {
      rejoin_in_progress.val = false
    }
    self.request_voice_credentials(
      guild_id, channel_id, self_mute, self_deaf, timeout_ms,
    )
  }
  let connection = @voice.VoiceConnection::start(
    self.group_,
    credentials,
    user_id=self.ready_.user.id.to_string(),
    channel_id=channel_id.value(),
    leave~,
    rejoin~,
    telemetry~,
  )
  wait_ready_or_disconnect(connection, timeout_ms)
  self.voice_connections_[guild_id] = connection
  self.group_.spawn_bg(no_wait=true, allow_failure=true, () => {
    self.watch_voice_server_updates(
      guild_id, channel_id, self_mute, self_deaf, timeout_ms, connection, rejoin_in_progress,
    )
  })
  connection
}