use std::{
    collections::{HashMap, HashSet},
    sync::{Arc, RwLock},
    time::Duration,
};

use crate::discord::ids::{
    Id,
    marker::{ChannelMarker, GuildMarker, UserMarker},
};
use futures::{SinkExt, StreamExt};
use rand::Rng;
use serde_json::{Value, json};
use tokio::sync::{Mutex, mpsc, watch};
use tokio::time::sleep;
use tokio_tungstenite::{
    connect_async_with_config,
    tungstenite::{
        Message as WsMessage,
        protocol::{CloseFrame, WebSocketConfig},
    },
};

use super::{
    ActivityInfo, PresenceStatus,
    client::publish_app_event,
    events::{AppEvent, SequencedAppEvent},
    fingerprint::{
        CLIENT_BROWSER, CLIENT_BROWSER_VERSION, client_build_number, discord_web_os,
        discord_web_os_version, discord_web_user_agent,
    },
    state::{DiscordState, SnapshotRevision},
    voice::{self, VoiceRuntimeEvent},
};
use crate::logging;

mod parser;

pub(in crate::discord) use parser::parse_activity;
use parser::parse_user_account_dispatch;
pub(crate) use parser::{parse_channel_info, parse_message_info};

#[derive(Clone, Debug, Eq, PartialEq)]
pub enum GatewayCommand {
    RequestGuildMembers {
        guild_id: Id<GuildMarker>,
        query: String,
        limit: u16,
        presences: bool,
        nonce: Option<String>,
    },
    RequestGuildMembersByIds {
        guild_id: Id<GuildMarker>,
        user_ids: Vec<Id<UserMarker>>,
        presences: bool,
    },
    SubscribeDirectMessage {
        channel_id: Id<ChannelMarker>,
    },
    SubscribeGuildChannel {
        guild_id: Id<GuildMarker>,
        channel_id: Id<ChannelMarker>,
    },
    UpdateMemberListSubscription {
        guild_id: Id<GuildMarker>,
        channel_id: Id<ChannelMarker>,
        ranges: Vec<(u32, u32)>,
    },
    UpdateVoiceState {
        /// `None` for DM and group-DM calls, which Discord joins with a null
        /// `guild_id` and the DM `channel_id` as the voice target.
        guild_id: Option<Id<GuildMarker>>,
        channel_id: Option<Id<ChannelMarker>>,
        self_mute: bool,
        self_deaf: bool,
    },
    UpdatePresence {
        status: PresenceStatus,
        activities: Vec<ActivityInfo>,
    },
    Shutdown,
}

#[derive(Clone)]
pub(crate) struct GatewayRuntime {
    pub(crate) effects_tx: mpsc::Sender<SequencedAppEvent>,
    pub(crate) snapshots_tx: watch::Sender<SnapshotRevision>,
    pub(crate) state: Arc<RwLock<DiscordState>>,
    pub(crate) revision: Arc<RwLock<SnapshotRevision>>,
    pub(crate) gateway_session_id: Arc<RwLock<Option<String>>>,
    pub(crate) publish_lock: Arc<Mutex<()>>,
    pub(crate) voice_events_tx: mpsc::UnboundedSender<VoiceRuntimeEvent>,
}

/// Discord user-account gateway endpoint. We pin to `v=9` because the v9
/// dispatch shapes line up with everything `parse_user_account_event` already
/// understands. `compress=false` keeps the wire human-readable. Switching to
/// `zlib-stream` is a follow-up.
const GATEWAY_URL: &str = "wss://gateway.discord.gg/?v=9&encoding=json";

/// Bitmask Discord checks before delivering user-account-only payloads such as
/// `READY_SUPPLEMENTAL.merged_presences.friends` and per-friend
/// `PRESENCE_UPDATE` dispatches. Without these bits set Discord assumes the
/// session is a bot and silently drops friend presence streaming.
///
/// Bits enabled (sum 253):
///   0  LAZY_USER_NOTIFICATIONS
///   2  VERSIONED_READ_STATES
///   3  VERSIONED_USER_GUILD_SETTINGS
///   4  DEDUPE_USER_OBJECTS
///   5  PRIORITIZED_READY_PAYLOAD
///   6  MULTIPLE_GUILD_EXPERIMENT_POPULATIONS
///   7  NON_CHANNEL_READ_STATES
const USER_ACCOUNT_CAPABILITIES: u64 = 253;

// Some user-account READY payloads exceed tungstenite's default 16 MiB frame
// cap before Discord has a chance to split the initial state across follow-up
// dispatches. Keep the limit bounded, but large enough for accounts with many
// guilds and channels until gateway compression is implemented.
const GATEWAY_WEBSOCKET_LIMIT: usize = 64 << 20;

const RECONNECT_BASE_DELAY: Duration = Duration::from_millis(500);
const RECONNECT_MAX_DELAY: Duration = Duration::from_secs(30);

type GatewayStream =
    tokio_tungstenite::WebSocketStream<tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>>;

/// Shared, lockable WebSocket sink. Both the heartbeat task and the main
/// dispatch loop need to send over the same connection, so the sink lives
/// behind a `Mutex<Arc<…>>` instead of being moved into either side.
type WriterHandle = Arc<Mutex<futures::stream::SplitSink<GatewayStream, WsMessage>>>;

#[derive(Default)]
struct SubscriptionDeduper {
    direct_messages: HashSet<Id<ChannelMarker>>,
    guild_channels: HashMap<GuildChannelSubscriptionKey, Vec<(u32, u32)>>,
}

impl SubscriptionDeduper {
    fn should_send(&mut self, command: &GatewayCommand) -> bool {
        match command {
            GatewayCommand::SubscribeDirectMessage { channel_id } => {
                self.direct_messages.insert(*channel_id)
            }
            GatewayCommand::SubscribeGuildChannel {
                guild_id,
                channel_id,
            } => self.should_send_guild_channel(*guild_id, *channel_id, &[(0, 99)]),
            GatewayCommand::UpdateMemberListSubscription {
                guild_id,
                channel_id,
                ranges,
            } => self.should_send_guild_channel(*guild_id, *channel_id, ranges),
            _ => true,
        }
    }

    fn should_send_guild_channel(
        &mut self,
        guild_id: Id<GuildMarker>,
        channel_id: Id<ChannelMarker>,
        ranges: &[(u32, u32)],
    ) -> bool {
        let key = GuildChannelSubscriptionKey {
            guild_id,
            channel_id,
        };
        if self
            .guild_channels
            .get(&key)
            .is_some_and(|last_ranges| last_ranges == ranges)
        {
            return false;
        }
        self.guild_channels.insert(key, ranges.to_vec());
        true
    }
}

#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
struct GuildChannelSubscriptionKey {
    guild_id: Id<GuildMarker>,
    channel_id: Id<ChannelMarker>,
}

#[derive(Clone, Copy)]
struct GatewayPublishContext<'a> {
    effects_tx: &'a mpsc::Sender<SequencedAppEvent>,
    snapshots_tx: &'a watch::Sender<SnapshotRevision>,
    state: &'a Arc<RwLock<DiscordState>>,
    revision: &'a Arc<RwLock<SnapshotRevision>>,
    gateway_session_id: &'a Arc<RwLock<Option<String>>>,
    publish_lock: &'a Arc<Mutex<()>>,
    voice_events_tx: &'a mpsc::UnboundedSender<VoiceRuntimeEvent>,
}

#[derive(Clone, Copy)]
struct FrameContext<'a> {
    sequence_cell: &'a Arc<Mutex<Option<u64>>>,
    heartbeat_ack: &'a Arc<Mutex<HeartbeatAckState>>,
    writer: &'a WriterHandle,
    publish: GatewayPublishContext<'a>,
}

#[derive(Default)]
struct HeartbeatAckState {
    awaiting_ack: bool,
}

impl HeartbeatAckState {
    fn mark_heartbeat_sent(&mut self) -> bool {
        if self.awaiting_ack {
            return false;
        }
        self.awaiting_ack = true;
        true
    }

    fn mark_ack_received(&mut self) {
        self.awaiting_ack = false;
    }
}

/// What to do after one connection lifecycle ends.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum ConnectionOutcome {
    /// The websocket dropped or Discord asked us to reconnect. Try to RESUME
    /// using the saved session_id + sequence number.
    Resume,
    /// Authentication failed or Discord told us the session is dead. Throw
    /// the saved session away and start over with a fresh IDENTIFY.
    Reidentify,
    /// The downstream consumers went away, so stop the loop entirely.
    Stop,
    /// Discord rejected this gateway session in a way that retrying the same
    /// token or shard configuration cannot fix. Keep the UI alive so it can
    /// show the published gateway error.
    Fatal,
}

/// Mutable session bookkeeping that survives reconnects. We only persist what
/// op-6 RESUME needs (session_id + last seq) plus the resume URL Discord
/// hands us in READY.
#[derive(Default)]
struct SessionState {
    session_id: Option<String>,
    resume_url: Option<String>,
    last_sequence: Option<u64>,
    has_received_ready: bool,
}

impl SessionState {
    fn clear(&mut self) {
        self.session_id = None;
        self.resume_url = None;
        self.last_sequence = None;
    }

    fn can_resume(&self) -> bool {
        self.session_id.is_some()
    }

    fn next_url(&self) -> String {
        match self.resume_url.as_deref() {
            // Discord embeds `?v=...&encoding=...` already, but it costs
            // nothing to append our own and helps when the resume URL is bare.
            Some(url) if !url.is_empty() => format!("{url}/?v=9&encoding=json"),
            _ => GATEWAY_URL.to_owned(),
        }
    }
}

pub async fn run_gateway(
    token: String,
    mut commands: mpsc::UnboundedReceiver<GatewayCommand>,
    runtime: GatewayRuntime,
) {
    let mut session = SessionState::default();
    let mut backoff = RECONNECT_BASE_DELAY;
    let mut publish_gateway_closed = true;

    loop {
        let publish = GatewayPublishContext {
            effects_tx: &runtime.effects_tx,
            snapshots_tx: &runtime.snapshots_tx,
            state: &runtime.state,
            revision: &runtime.revision,
            gateway_session_id: &runtime.gateway_session_id,
            publish_lock: &runtime.publish_lock,
            voice_events_tx: &runtime.voice_events_tx,
        };
        let outcome = match connect_and_run(&token, &mut commands, &mut session, publish).await {
            Ok(outcome) => outcome,
            Err(error) => {
                logging::error("gateway", format!("connection error: {error}"));
                publish_gateway_event(
                    publish,
                    AppEvent::GatewayError {
                        message: format!("connection error: {error}"),
                    },
                )
                .await;
                ConnectionOutcome::Resume
            }
        };

        match outcome {
            ConnectionOutcome::Stop => break,
            ConnectionOutcome::Resume => {
                if !session.can_resume() {
                    // No saved session, fall through to a clean IDENTIFY.
                }
            }
            ConnectionOutcome::Reidentify => {
                session.clear();
                *runtime
                    .gateway_session_id
                    .write()
                    .expect("gateway session id lock is not poisoned") = None;
            }
            ConnectionOutcome::Fatal => {
                publish_gateway_closed = false;
                break;
            }
        }

        // Exponential backoff with full jitter so a flapping network doesn't
        // hammer Discord. Successful sessions reset the delay below.
        let jitter = rand::thread_rng().gen_range(0..=backoff.as_millis() as u64);
        let delay = Duration::from_millis(jitter);
        logging::debug(
            "gateway",
            format!("reconnecting in {}ms", delay.as_millis()),
        );
        sleep(delay).await;
        backoff = (backoff * 2).min(RECONNECT_MAX_DELAY);
    }

    if publish_gateway_closed {
        let publish = GatewayPublishContext {
            effects_tx: &runtime.effects_tx,
            snapshots_tx: &runtime.snapshots_tx,
            state: &runtime.state,
            revision: &runtime.revision,
            gateway_session_id: &runtime.gateway_session_id,
            publish_lock: &runtime.publish_lock,
            voice_events_tx: &runtime.voice_events_tx,
        };
        publish_gateway_event(publish, AppEvent::GatewayClosed).await;
    }
}

async fn connect_and_run(
    token: &str,
    commands: &mut mpsc::UnboundedReceiver<GatewayCommand>,
    session: &mut SessionState,
    publish: GatewayPublishContext<'_>,
) -> Result<ConnectionOutcome, String> {
    let url = session.next_url();
    logging::debug("gateway", format!("connecting to {url}"));

    let (ws, _response) = connect_async_with_config(&url, Some(gateway_websocket_config()), false)
        .await
        .map_err(|error| format!("websocket connect failed: {error}"))?;
    let (writer, mut reader) = ws.split();
    let writer = Arc::new(Mutex::new(writer));
    let mut subscription_deduper = SubscriptionDeduper::default();

    // Discord must speak first with op-10 HELLO carrying heartbeat_interval.
    // If the first frame is anything else, fail fast and try a clean
    // re-identify.
    let hello_frame = match reader.next().await {
        Some(Ok(WsMessage::Text(text))) => text,
        Some(Ok(WsMessage::Close(frame))) => {
            let message = websocket_close_message("websocket closed before HELLO", frame.as_ref());
            log_and_publish_gateway_error(publish, message).await;
            return Ok(ConnectionOutcome::Reidentify);
        }
        Some(Ok(_)) => return Err("unexpected non-text frame before HELLO".to_owned()),
        Some(Err(error)) => return Err(format!("read HELLO failed: {error}")),
        None => return Err("connection closed before HELLO".to_owned()),
    };
    let hello: Value =
        serde_json::from_str(&hello_frame).map_err(|error| format!("HELLO parse: {error}"))?;
    if hello.get("op").and_then(Value::as_u64) != Some(10) {
        return Err(format!(
            "first frame was not HELLO: {}",
            hello.get("op").and_then(Value::as_u64).unwrap_or_default()
        ));
    }
    let heartbeat_interval_ms = hello
        .get("d")
        .and_then(|d| d.get("heartbeat_interval"))
        .and_then(Value::as_u64)
        .unwrap_or(41250);
    let heartbeat_interval = Duration::from_millis(heartbeat_interval_ms);

    // Either resume with the saved session or send a fresh IDENTIFY. RESUME
    // tells Discord to replay missed dispatches. This is good for transient drops.
    // IDENTIFY rebuilds the world from scratch.
    if session.can_resume() {
        let payload = build_resume_payload(token, session);
        send_text(&writer, payload).await?;
        logging::debug("gateway", "RESUME sent");
    } else {
        let payload = build_identify_payload(token);
        send_text(&writer, payload).await?;
        logging::debug("gateway", "IDENTIFY sent");
    }

    // Background heartbeat task driven by Discord's interval. We jitter the
    // first beat per the API recommendation. The task reads the latest seq
    // from a shared atomic via the sequence cell.
    let writer_for_heartbeat = Arc::clone(&writer);
    let sequence_cell: Arc<Mutex<Option<u64>>> = Arc::new(Mutex::new(session.last_sequence));
    let sequence_for_heartbeat = Arc::clone(&sequence_cell);
    let heartbeat_ack: Arc<Mutex<HeartbeatAckState>> = Arc::default();
    let heartbeat_ack_for_task = Arc::clone(&heartbeat_ack);
    let (heartbeat_timeout_tx, mut heartbeat_timeout_rx) = mpsc::unbounded_channel();
    let initial_jitter = {
        let jitter_ms =
            rand::thread_rng().gen_range(0..=heartbeat_interval.as_millis().min(2_000) as u64);
        Duration::from_millis(jitter_ms)
    };
    let heartbeat_task = tokio::spawn(async move {
        sleep(initial_jitter).await;
        loop {
            {
                let mut state = heartbeat_ack_for_task.lock().await;
                if !state.mark_heartbeat_sent() {
                    logging::error("gateway", "heartbeat ACK timeout; reconnecting");
                    let _ = heartbeat_timeout_tx.send(());
                    break;
                }
            }
            let seq = *sequence_for_heartbeat.lock().await;
            let payload = json!({"op": 1, "d": seq}).to_string();
            if let Err(error) = send_text(&writer_for_heartbeat, payload).await {
                logging::error("gateway", format!("heartbeat send failed: {error}"));
                let _ = heartbeat_timeout_tx.send(());
                break;
            }
            sleep(heartbeat_interval).await;
        }
    });

    // Main loop: race incoming frames against outgoing user commands. The
    // heartbeat task is already running on its own cadence in the background.
    let outcome = loop {
        tokio::select! {
            biased;

            maybe_command = commands.recv() => {
                match maybe_command {
                    Some(command) => {
                        if let GatewayCommand::Shutdown = command {
                            if let Err(error) = close_websocket(&writer).await {
                                let message = format!("gateway shutdown failed: {error}");
                                log_and_publish_gateway_error(publish, message).await;
                            }
                            break ConnectionOutcome::Stop;
                        } else if let Err(error) =
                            dispatch_command(&writer, command, &mut subscription_deduper).await
                        {
                            let message = format!("command send failed: {error}");
                            log_and_publish_gateway_error(publish, message).await;
                            break ConnectionOutcome::Resume;
                        }
                    }
                    None => break ConnectionOutcome::Stop,
                }
            }
            frame = reader.next() => {
                match frame {
                    Some(Ok(WsMessage::Text(text))) => {
                        let value: Value = match serde_json::from_str(&text) {
                            Ok(value) => value,
                            Err(error) => {
                                logging::debug(
                                    "gateway",
                                    format!("ignoring non-JSON frame: {error}"),
                                );
                                continue;
                            }
                        };
                        let frame_context = FrameContext {
                            sequence_cell: &sequence_cell,
                            heartbeat_ack: &heartbeat_ack,
                            writer: &writer,
                            publish,
                        };
                        match handle_frame(
                            value,
                            &text,
                            session,
                            frame_context,
                        ).await {
                            FrameOutcome::Continue => {}
                            FrameOutcome::Resume => break ConnectionOutcome::Resume,
                            FrameOutcome::Reidentify => break ConnectionOutcome::Reidentify,
                        }
                    }
                    Some(Ok(WsMessage::Binary(_))) => {
                        // Compression isn't enabled in the IDENTIFY, so binary
                        // frames are unexpected. Log and ignore rather than
                        // panic on bad input.
                        logging::debug("gateway", "ignoring unexpected binary frame");
                    }
                    Some(Ok(WsMessage::Ping(payload))) => {
                        let mut writer = writer.lock().await;
                        if let Err(error) = writer.send(WsMessage::Pong(payload)).await {
                            let message = format!("websocket pong send failed: {error}");
                            log_and_publish_gateway_error(publish, message).await;
                            break ConnectionOutcome::Resume;
                        }
                    }
                    Some(Ok(WsMessage::Pong(_))) | Some(Ok(WsMessage::Frame(_))) => {}
                    Some(Ok(WsMessage::Close(frame))) => {
                        let outcome = close_outcome(frame.as_ref());
                        let message = websocket_close_message("websocket closed", frame.as_ref());
                        log_and_publish_gateway_error(publish, message).await;
                        break outcome;
                    }
                    Some(Err(error)) => {
                        let message = format!("websocket read error: {error}");
                        log_and_publish_gateway_error(publish, message).await;
                        break ConnectionOutcome::Resume;
                    }
                    None => {
                        let message = "websocket closed without frame".to_owned();
                        log_and_publish_gateway_error(publish, message).await;
                        break ConnectionOutcome::Resume;
                    }
                }
            }
            _ = heartbeat_timeout_rx.recv() => {
                break ConnectionOutcome::Resume;
            }
        }
    };

    heartbeat_task.abort();
    Ok(outcome)
}

fn gateway_websocket_config() -> WebSocketConfig {
    WebSocketConfig::default()
        .max_message_size(Some(GATEWAY_WEBSOCKET_LIMIT))
        .max_frame_size(Some(GATEWAY_WEBSOCKET_LIMIT))
}

enum FrameOutcome {
    Continue,
    Resume,
    Reidentify,
}

async fn handle_frame(
    value: Value,
    raw: &str,
    session: &mut SessionState,
    context: FrameContext<'_>,
) -> FrameOutcome {
    let op = value.get("op").and_then(Value::as_u64).unwrap_or_default();
    match op {
        // Dispatch
        0 => {
            if let Some(seq) = value.get("s").and_then(Value::as_u64) {
                session.last_sequence = Some(seq);
                *context.sequence_cell.lock().await = Some(seq);
            }
            let dispatch_type = value.get("t").and_then(Value::as_str).unwrap_or("");
            let mut publish_reidentified = false;
            // Capture the session_id and resume_url from READY so a later
            // disconnect can RESUME instead of redoing the heavy initial sync.
            if dispatch_type == "READY"
                && let Some(d) = value.get("d")
            {
                let was_reidentify = session.has_received_ready;
                session.session_id = d
                    .get("session_id")
                    .and_then(Value::as_str)
                    .map(str::to_owned);
                session.resume_url = d
                    .get("resume_gateway_url")
                    .and_then(Value::as_str)
                    .map(str::to_owned);
                *context
                    .publish
                    .gateway_session_id
                    .write()
                    .expect("gateway session id lock is not poisoned") = session.session_id.clone();
                if was_reidentify {
                    publish_reidentified = true;
                }
                session.has_received_ready = true;
            } else if dispatch_type == "RESUMED" {
                publish_gateway_event(context.publish, AppEvent::GatewayResumed).await;
            }
            if let Some(parsed) = parse_user_account_dispatch(raw) {
                publish_gateway_event(
                    context.publish,
                    AppEvent::GatewayDispatchReceived {
                        dispatch: parsed.dispatch,
                    },
                )
                .await;
                for app_event in parsed.events {
                    publish_gateway_event(context.publish, app_event).await;
                }
            }
            if publish_reidentified {
                publish_gateway_event(context.publish, AppEvent::GatewayReidentified).await;
            }
            FrameOutcome::Continue
        }
        // Answer Discord heartbeat requests immediately. The background task
        // only paces our own heartbeat sends.
        1 => {
            let seq = *context.sequence_cell.lock().await;
            let payload = json!({"op": 1, "d": seq}).to_string();
            context.heartbeat_ack.lock().await.mark_heartbeat_sent();
            if let Err(error) = send_text(context.writer, payload).await {
                let message = format!("heartbeat response send failed: {error}");
                log_and_publish_gateway_error(context.publish, message).await;
            }
            FrameOutcome::Continue
        }
        // Discord wants us to drop and resume. Saved session_id and seq make
        // the resume cheap.
        7 => {
            logging::debug("gateway", "RECONNECT requested");
            FrameOutcome::Resume
        }
        // `d` tells us whether an invalid session is resumable. Anything else
        // means we have to throw it away.
        9 => {
            let resumable = value.get("d").and_then(Value::as_bool).unwrap_or(false);
            logging::debug("gateway", format!("INVALID_SESSION resumable={resumable}"));
            if resumable {
                FrameOutcome::Resume
            } else {
                FrameOutcome::Reidentify
            }
        }
        11 => {
            context.heartbeat_ack.lock().await.mark_ack_received();
            FrameOutcome::Continue
        }
        other => {
            logging::debug("gateway", format!("unhandled gateway op={other}"));
            FrameOutcome::Continue
        }
    }
}

async fn publish_gateway_event(context: GatewayPublishContext<'_>, event: AppEvent) {
    publish_app_event(
        context.effects_tx,
        context.snapshots_tx,
        context.state,
        context.revision,
        context.publish_lock,
        &event,
    )
    .await;
    voice::forward_app_event(context.voice_events_tx, &event);
}

async fn log_and_publish_gateway_error(context: GatewayPublishContext<'_>, message: String) {
    logging::error("gateway", &message);
    publish_gateway_event(context, AppEvent::GatewayError { message }).await;
}

fn close_outcome(frame: Option<&CloseFrame>) -> ConnectionOutcome {
    let Some(frame) = frame else {
        return ConnectionOutcome::Resume;
    };
    close_code_outcome(u16::from(frame.code))
}

fn close_code_outcome(code: u16) -> ConnectionOutcome {
    // Authentication and gateway configuration failures are not transient.
    // Retrying the same IDENTIFY would hide the real problem behind Loading...
    // and can loop forever for codes such as 4004.
    match code {
        4004 | 4010..=4014 => ConnectionOutcome::Fatal,
        4007 | 4009 => ConnectionOutcome::Reidentify,
        4000..=4003 | 4005 | 4008 => ConnectionOutcome::Resume,
        _ => ConnectionOutcome::Reidentify,
    }
}

fn websocket_close_message(context: &str, frame: Option<&CloseFrame>) -> String {
    if let Some(frame) = frame {
        format!(
            "{context}: code={} reason={:?}",
            u16::from(frame.code),
            frame.reason.as_str()
        )
    } else {
        context.to_owned()
    }
}

async fn dispatch_command(
    writer: &WriterHandle,
    command: GatewayCommand,
    subscription_deduper: &mut SubscriptionDeduper,
) -> Result<(), String> {
    if !subscription_deduper.should_send(&command) {
        logging::debug("gateway", "skipping duplicate channel subscription");
        return Ok(());
    }

    let payload = match command {
        GatewayCommand::RequestGuildMembers {
            guild_id,
            query,
            limit,
            presences,
            nonce,
        } => {
            logging::debug(
                "gateway",
                format!(
                    "requesting guild members: guild={} query_len={} limit={} presences={}",
                    guild_id.get(),
                    query.len(),
                    limit,
                    presences
                ),
            );
            request_guild_members_payload(guild_id, &query, limit, presences, nonce.as_deref())
        }
        GatewayCommand::RequestGuildMembersByIds {
            guild_id,
            user_ids,
            presences,
        } => {
            logging::debug(
                "gateway",
                format!(
                    "requesting guild members by id: guild={} users={} presences={}",
                    guild_id.get(),
                    user_ids.len(),
                    presences
                ),
            );
            request_guild_members_by_ids_payload(guild_id, &user_ids, presences)
        }
        GatewayCommand::SubscribeDirectMessage { channel_id } => {
            logging::debug(
                "gateway",
                format!("subscribing to DM: channel={}", channel_id.get()),
            );
            direct_message_subscribe_payload(channel_id)
        }
        GatewayCommand::SubscribeGuildChannel {
            guild_id,
            channel_id,
        } => {
            logging::debug(
                "gateway",
                format!(
                    "subscribing to guild channel: guild={} channel={}",
                    guild_id.get(),
                    channel_id.get()
                ),
            );
            guild_channel_subscribe_payload(guild_id, channel_id, &[(0, 99)])
        }
        GatewayCommand::UpdateMemberListSubscription {
            guild_id,
            channel_id,
            ranges,
        } => {
            logging::debug(
                "gateway",
                format!(
                    "updating member list ranges: guild={} channel={} ranges={:?}",
                    guild_id.get(),
                    channel_id.get(),
                    ranges
                ),
            );
            guild_channel_subscribe_payload(guild_id, channel_id, &ranges)
        }
        GatewayCommand::UpdateVoiceState {
            guild_id,
            channel_id,
            self_mute,
            self_deaf,
        } => {
            logging::debug(
                "gateway",
                format!(
                    "updating voice state: guild={} channel={} self_mute={} self_deaf={}",
                    guild_id.map(|id| id.get()).unwrap_or_default(),
                    channel_id.map(|id| id.get()).unwrap_or_default(),
                    self_mute,
                    self_deaf,
                ),
            );
            voice_state_update_payload(guild_id, channel_id, self_mute, self_deaf)
        }
        GatewayCommand::UpdatePresence { status, activities } => {
            logging::debug(
                "gateway",
                format!(
                    "updating presence status: {} activities={}",
                    status.label(),
                    activities.len()
                ),
            );
            presence_update_payload(status, &activities)
        }
        GatewayCommand::Shutdown => return Ok(()),
    };
    send_text(writer, payload).await
}

async fn close_websocket(writer: &WriterHandle) -> Result<(), String> {
    let mut writer = writer.lock().await;
    writer
        .close()
        .await
        .map_err(|error| format!("websocket close failed: {error}"))
}

async fn send_text(writer: &WriterHandle, payload: String) -> Result<(), String> {
    let mut writer = writer.lock().await;
    writer
        .send(WsMessage::Text(payload.into()))
        .await
        .map_err(|error| format!("websocket send failed: {error}"))
}

fn build_identify_payload(token: &str) -> String {
    let os = discord_web_os();
    let os_version = discord_web_os_version();
    let user_agent = discord_web_user_agent();
    json!({
        "op": 2,
        "d": {
            "token": token,
            "capabilities": USER_ACCOUNT_CAPABILITIES,
            "properties": {
                "os": os,
                "browser": CLIENT_BROWSER,
                "device": "",
                "system_locale": "en-US",
                "browser_user_agent": user_agent,
                "browser_version": CLIENT_BROWSER_VERSION,
                "os_version": os_version,
                "referrer": "",
                "referring_domain": "",
                "referrer_current": "",
                "referring_domain_current": "",
                "release_channel": "stable",
                "client_build_number": client_build_number(),
                "client_event_source": Value::Null,
            },
            "presence": {
                "status": PresenceStatus::Online.gateway_status(),
                "since": 0,
                "activities": [],
                "afk": false,
            },
            "compress": false,
            "client_state": {
                "guild_versions": {},
                "highest_last_message_id": "0",
                "read_state_version": 0,
                "user_guild_settings_version": -1,
                "user_settings_version": -1,
                "private_channels_version": "0",
                "api_code_version": 0,
            },
        },
    })
    .to_string()
}

fn build_resume_payload(token: &str, session: &SessionState) -> String {
    json!({
        "op": 6,
        "d": {
            "token": token,
            "session_id": session.session_id.as_deref().unwrap_or_default(),
            "seq": session.last_sequence.unwrap_or_default(),
        },
    })
    .to_string()
}

fn request_guild_members_payload(
    guild_id: Id<GuildMarker>,
    query: &str,
    limit: u16,
    presences: bool,
    nonce: Option<&str>,
) -> String {
    let mut data = json!({
        "guild_id": guild_id.to_string(),
        "query": query,
        "limit": limit,
        "presences": presences,
    });
    if let Some(nonce) = nonce {
        data["nonce"] = json!(nonce);
    }
    json!({
        "op": 8,
        "d": data,
    })
    .to_string()
}

fn request_guild_members_by_ids_payload(
    guild_id: Id<GuildMarker>,
    user_ids: &[Id<UserMarker>],
    presences: bool,
) -> String {
    let user_ids = user_ids
        .iter()
        .take(100)
        .map(|user_id| user_id.to_string())
        .collect::<Vec<_>>();
    json!({
        "op": 8,
        "d": {
            "guild_id": guild_id.to_string(),
            "user_ids": user_ids,
            "presences": presences,
        },
    })
    .to_string()
}

fn direct_message_subscribe_payload(channel_id: Id<ChannelMarker>) -> String {
    json!({
        "op": 13,
        "d": {
            "channel_id": channel_id.to_string(),
        },
    })
    .to_string()
}

fn guild_channel_subscribe_payload(
    guild_id: Id<GuildMarker>,
    channel_id: Id<ChannelMarker>,
    ranges: &[(u32, u32)],
) -> String {
    let ranges_json: Vec<[u32; 2]> = ranges.iter().map(|(start, end)| [*start, *end]).collect();
    json!({
        "op": 37,
        "d": {
            "subscriptions": {
                guild_id.to_string(): {
                    "typing": true,
                    "activities": true,
                    "threads": true,
                    "channels": {
                        channel_id.to_string(): ranges_json,
                    },
                },
            },
        },
    })
    .to_string()
}

fn voice_state_update_payload(
    guild_id: Option<Id<GuildMarker>>,
    channel_id: Option<Id<ChannelMarker>>,
    self_mute: bool,
    self_deaf: bool,
) -> String {
    // A null `guild_id` tells Discord this is a DM or group-DM call; the
    // `channel_id` then points at the DM channel rather than a guild voice
    // channel.
    json!({
        "op": 4,
        "d": {
            "guild_id": guild_id.map(|guild_id| guild_id.to_string()),
            "channel_id": channel_id.map(|channel_id| channel_id.to_string()),
            "self_mute": self_mute,
            "self_deaf": self_deaf,
        },
    })
    .to_string()
}

fn presence_update_payload(status: PresenceStatus, activities: &[ActivityInfo]) -> String {
    json!({
        "op": 3,
        "d": {
            "since": 0,
            "activities": activities.iter().map(activity_gateway_payload).collect::<Vec<_>>(),
            "status": status.gateway_status(),
            "afk": false,
        },
    })
    .to_string()
}

fn activity_gateway_payload(activity: &ActivityInfo) -> Value {
    let mut value = json!({
        "name": activity.name.as_str(),
        "type": activity.kind.gateway_code(),
    });
    if let Some(details) = activity.details.as_deref() {
        value["details"] = json!(details);
    }
    if let Some(state) = activity.state.as_deref() {
        value["state"] = json!(state);
    }
    if let Some(url) = activity.url.as_deref() {
        value["url"] = json!(url);
    }
    // A Custom status carries its emoji here. Without it a status change would
    // re-broadcast the activity and drop the emoji.
    if let Some(emoji) = activity.emoji.as_ref() {
        let mut node = json!({ "name": emoji.name.as_str() });
        if let Some(id) = emoji.id {
            node["id"] = json!(id.get().to_string());
        }
        if emoji.animated {
            node["animated"] = json!(true);
        }
        value["emoji"] = node;
    }
    if let Some(application_id) = activity.application_id.as_deref() {
        value["application_id"] = json!(application_id);
    }
    if let Some(timestamps) = activity.timestamps.as_ref() {
        let mut node = json!({});
        if let Some(start) = timestamps.start {
            node["start"] = json!(start);
        }
        if let Some(end) = timestamps.end {
            node["end"] = json!(end);
        }
        value["timestamps"] = node;
    }
    if let Some(assets) = activity.assets.as_ref() {
        let mut node = json!({});
        if let Some(large_image) = assets.large_image.as_deref() {
            node["large_image"] = json!(large_image);
        }
        if let Some(large_text) = assets.large_text.as_deref() {
            node["large_text"] = json!(large_text);
        }
        if let Some(small_image) = assets.small_image.as_deref() {
            node["small_image"] = json!(small_image);
        }
        if let Some(small_text) = assets.small_text.as_deref() {
            node["small_text"] = json!(small_text);
        }
        value["assets"] = node;
    }
    if let Some(party) = activity.party.as_ref() {
        let mut node = json!({});
        if let Some(id) = party.id.as_deref() {
            node["id"] = json!(id);
        }
        if let Some((current, max)) = party.size {
            node["size"] = json!([current, max]);
        }
        value["party"] = node;
    }
    // User-account presence encodes buttons as a parallel pair: an array of
    // labels under `buttons` and their URLs under `metadata.button_urls`. This
    // differs from the bot `[{label, url}]` shape.
    if !activity.buttons.is_empty() {
        let labels: Vec<&str> = activity
            .buttons
            .iter()
            .map(|button| button.label.as_str())
            .collect();
        let urls: Vec<&str> = activity
            .buttons
            .iter()
            .map(|button| button.url.as_str())
            .collect();
        value["buttons"] = json!(labels);
        value["metadata"] = json!({ "button_urls": urls });
    }
    value
}

#[cfg(test)]
mod tests {
    use crate::discord::fingerprint::{
        CLIENT_BROWSER, CLIENT_BROWSER_VERSION, CLIENT_BUILD_NUMBER, discord_web_os,
        discord_web_os_version, discord_web_user_agent,
    };
    use crate::discord::ids::{
        Id,
        marker::{ChannelMarker, EmojiMarker, GuildMarker, UserMarker},
    };
    use crate::discord::{ActivityEmoji, ActivityInfo, ActivityKind, PresenceStatus};
    use serde_json::json;

    use super::{
        ConnectionOutcome, GATEWAY_WEBSOCKET_LIMIT, GatewayCommand, HeartbeatAckState,
        SessionState, SubscriptionDeduper, USER_ACCOUNT_CAPABILITIES, build_identify_payload,
        build_resume_payload, close_code_outcome, direct_message_subscribe_payload,
        gateway_websocket_config, guild_channel_subscribe_payload, presence_update_payload,
        request_guild_members_by_ids_payload, request_guild_members_payload,
        voice_state_update_payload,
    };

    #[test]
    fn gateway_websocket_config_allows_large_ready_payloads() {
        let config = gateway_websocket_config();

        assert_eq!(config.max_message_size, Some(GATEWAY_WEBSOCKET_LIMIT));
        assert_eq!(config.max_frame_size, Some(GATEWAY_WEBSOCKET_LIMIT));
    }

    #[test]
    fn identify_payload_carries_user_account_capabilities() {
        let payload: serde_json::Value =
            serde_json::from_str(&build_identify_payload("dummy-token"))
                .expect("identify payload should be valid json");
        assert_eq!(payload["op"].as_u64(), Some(2));
        assert_eq!(
            payload["d"]["capabilities"].as_u64(),
            Some(USER_ACCOUNT_CAPABILITIES)
        );
        assert_eq!(
            payload["d"]["properties"]["os"].as_str(),
            Some(discord_web_os())
        );
        assert_eq!(
            payload["d"]["properties"]["browser"].as_str(),
            Some(CLIENT_BROWSER)
        );
        assert_eq!(
            payload["d"]["properties"]["browser_user_agent"].as_str(),
            Some(discord_web_user_agent().as_str())
        );
        assert_eq!(
            payload["d"]["properties"]["browser_version"].as_str(),
            Some(CLIENT_BROWSER_VERSION)
        );
        assert_eq!(
            payload["d"]["properties"]["os_version"].as_str(),
            Some(discord_web_os_version().as_str())
        );
        assert_eq!(
            payload["d"]["properties"]["client_build_number"].as_u64(),
            Some(CLIENT_BUILD_NUMBER)
        );
        assert_eq!(payload["d"]["compress"].as_bool(), Some(false));
        assert_eq!(payload["d"]["presence"]["status"].as_str(), Some("online"));
    }

    #[test]
    fn presence_update_payload_maps_statuses_for_gateway() {
        let online_payload: serde_json::Value =
            serde_json::from_str(&presence_update_payload(PresenceStatus::Online, &[]))
                .expect("presence payload should be valid json");
        assert_eq!(online_payload["op"].as_u64(), Some(3));
        assert_eq!(online_payload["d"]["status"].as_str(), Some("online"));
        assert_eq!(online_payload["d"]["since"].as_u64(), Some(0));
        assert_eq!(online_payload["d"]["activities"], json!([]));
        assert_eq!(online_payload["d"]["afk"].as_bool(), Some(false));

        let idle_payload: serde_json::Value =
            serde_json::from_str(&presence_update_payload(PresenceStatus::Idle, &[]))
                .expect("presence payload should be valid json");
        assert_eq!(idle_payload["d"]["status"].as_str(), Some("idle"));

        let dnd_payload: serde_json::Value =
            serde_json::from_str(&presence_update_payload(PresenceStatus::DoNotDisturb, &[]))
                .expect("presence payload should be valid json");
        assert_eq!(dnd_payload["d"]["status"].as_str(), Some("dnd"));

        let offline_payload: serde_json::Value =
            serde_json::from_str(&presence_update_payload(PresenceStatus::Offline, &[]))
                .expect("presence payload should be valid json");
        assert_eq!(offline_payload["d"]["status"].as_str(), Some("invisible"));
    }

    #[test]
    fn presence_update_payload_carries_custom_status_emoji() {
        let mut activity = ActivityInfo::test(ActivityKind::Custom, "");
        activity.emoji = Some(ActivityEmoji {
            name: "wave".to_owned(),
            id: Some(Id::<EmojiMarker>::new(50)),
            animated: true,
        });
        let payload: serde_json::Value = serde_json::from_str(&presence_update_payload(
            PresenceStatus::Online,
            &[activity],
        ))
        .expect("presence payload should be valid json");
        let emoji = &payload["d"]["activities"][0]["emoji"];
        assert_eq!(emoji["name"].as_str(), Some("wave"));
        assert_eq!(emoji["id"].as_str(), Some("50"));
        assert_eq!(emoji["animated"].as_bool(), Some(true));
    }

    #[test]
    fn presence_update_payload_includes_manual_activity() {
        let activity = ActivityInfo::playing("Concord");
        let payload: serde_json::Value = serde_json::from_str(&presence_update_payload(
            PresenceStatus::Online,
            &[activity],
        ))
        .expect("presence payload should be valid json");

        assert_eq!(
            payload["d"]["activities"][0]["name"].as_str(),
            Some("Concord")
        );
        assert_eq!(payload["d"]["activities"][0]["type"].as_u64(), Some(0));
    }

    #[test]
    fn presence_update_payload_serializes_rich_activity_fields() {
        let activity = ActivityInfo {
            timestamps: Some(crate::discord::ActivityTimestamps {
                start: Some(1_700_000_000_000),
                end: None,
            }),
            assets: Some(crate::discord::ActivityAssets {
                large_image: Some("cover".to_owned()),
                large_text: Some("On the main menu".to_owned()),
                small_image: None,
                small_text: None,
            }),
            party: Some(crate::discord::ActivityParty {
                id: Some("party-1".to_owned()),
                size: Some((2, 5)),
            }),
            buttons: vec![crate::discord::ActivityButton {
                label: "Join".to_owned(),
                url: "https://example.com/join".to_owned(),
            }],
            ..ActivityInfo::playing("Concord")
        };
        let payload: serde_json::Value = serde_json::from_str(&presence_update_payload(
            PresenceStatus::Online,
            &[activity],
        ))
        .expect("presence payload should be valid json");
        let entry = &payload["d"]["activities"][0];

        assert_eq!(
            entry["timestamps"]["start"].as_i64(),
            Some(1_700_000_000_000)
        );
        assert!(entry["timestamps"].get("end").is_none());
        assert_eq!(entry["assets"]["large_image"].as_str(), Some("cover"));
        assert_eq!(
            entry["assets"]["large_text"].as_str(),
            Some("On the main menu")
        );
        assert!(entry["assets"].get("small_image").is_none());
        assert_eq!(entry["party"]["id"].as_str(), Some("party-1"));
        assert_eq!(entry["party"]["size"], json!([2, 5]));
        assert_eq!(entry["buttons"], json!(["Join"]));
        assert_eq!(
            entry["metadata"]["button_urls"],
            json!(["https://example.com/join"])
        );
    }

    #[test]
    fn fatal_gateway_close_codes_do_not_retry_identify() {
        for code in [4004, 4010, 4011, 4012, 4013, 4014] {
            assert_eq!(close_code_outcome(code), ConnectionOutcome::Fatal, "{code}");
        }
        assert_eq!(close_code_outcome(4007), ConnectionOutcome::Reidentify);
        assert_eq!(close_code_outcome(4009), ConnectionOutcome::Reidentify);
        assert_eq!(close_code_outcome(4000), ConnectionOutcome::Resume);
    }

    #[test]
    fn resume_payload_uses_saved_session_id_and_seq() {
        let session = SessionState {
            session_id: Some("sess-123".to_owned()),
            last_sequence: Some(42),
            ..SessionState::default()
        };
        let payload: serde_json::Value =
            serde_json::from_str(&build_resume_payload("dummy-token", &session))
                .expect("resume payload should be valid json");
        assert_eq!(payload["op"].as_u64(), Some(6));
        assert_eq!(payload["d"]["session_id"].as_str(), Some("sess-123"));
        assert_eq!(payload["d"]["seq"].as_u64(), Some(42));
    }

    #[test]
    fn heartbeat_ack_state_detects_missing_ack_before_next_heartbeat() {
        let mut state = HeartbeatAckState::default();

        assert!(state.mark_heartbeat_sent());
        assert!(!state.mark_heartbeat_sent());
        state.mark_ack_received();
        assert!(state.mark_heartbeat_sent());
    }

    #[test]
    fn request_guild_members_payload_supports_full_load_and_search_shapes() {
        let search_payload: serde_json::Value =
            serde_json::from_str(&request_guild_members_payload(
                Id::<GuildMarker>::new(10),
                "alic",
                10,
                false,
                Some("mention-ac-10-alic"),
            ))
            .expect("payload should be valid json");

        assert_eq!(
            search_payload,
            json!({
                "op": 8,
                "d": {
                    "guild_id": "10",
                    "query": "alic",
                    "limit": 10,
                    "presences": false,
                    "nonce": "mention-ac-10-alic"
                }
            })
        );

        let full_load_payload: serde_json::Value = serde_json::from_str(
            &request_guild_members_payload(Id::<GuildMarker>::new(10), "", 0, true, None),
        )
        .expect("payload should be valid json");

        assert_eq!(full_load_payload["op"].as_u64(), Some(8));
        assert_eq!(full_load_payload["d"]["guild_id"].as_str(), Some("10"));
        assert_eq!(full_load_payload["d"]["query"].as_str(), Some(""));
        assert_eq!(full_load_payload["d"]["limit"].as_u64(), Some(0));
        assert_eq!(full_load_payload["d"]["presences"].as_bool(), Some(true));
        assert!(full_load_payload["d"].get("nonce").is_none());
    }

    #[test]
    fn request_guild_members_by_ids_payload_matches_web_shape() {
        let payload: serde_json::Value =
            serde_json::from_str(&request_guild_members_by_ids_payload(
                Id::<GuildMarker>::new(10),
                &[Id::<UserMarker>::new(20), Id::<UserMarker>::new(30)],
                false,
            ))
            .expect("payload should be valid json");

        assert_eq!(
            payload,
            json!({
                "op": 8,
                "d": {
                    "guild_id": "10",
                    "user_ids": ["20", "30"],
                    "presences": false
                }
            })
        );
    }

    #[test]
    fn direct_message_subscribe_payload_matches_expected_shape() {
        let payload: serde_json::Value = serde_json::from_str(&direct_message_subscribe_payload(
            Id::<ChannelMarker>::new(20),
        ))
        .expect("payload should be valid json");

        assert_eq!(
            payload,
            json!({
                "op": 13,
                "d": {
                    "channel_id": "20"
                }
            })
        );
    }

    #[test]
    fn guild_channel_subscribe_payload_matches_shape_and_member_ranges() {
        for (ranges, expected_ranges) in [
            (&[(0, 99)][..], json!([[0, 99]])),
            (
                &[(0, 99), (100, 199), (200, 299)][..],
                json!([[0, 99], [100, 199], [200, 299]]),
            ),
        ] {
            let payload: serde_json::Value =
                serde_json::from_str(&guild_channel_subscribe_payload(
                    Id::<GuildMarker>::new(10),
                    Id::<ChannelMarker>::new(20),
                    ranges,
                ))
                .expect("payload should be valid json");

            assert_eq!(payload["op"].as_u64(), Some(37));
            assert_eq!(payload["d"]["subscriptions"]["10"]["typing"], json!(true));
            assert_eq!(
                payload["d"]["subscriptions"]["10"]["activities"],
                json!(true)
            );
            assert_eq!(payload["d"]["subscriptions"]["10"]["threads"], json!(true));
            assert_eq!(
                payload["d"]["subscriptions"]["10"]["channels"]["20"],
                expected_ranges
            );
            if ranges == &[(0, 99)][..] {
                assert_eq!(
                    payload,
                    json!({
                        "op": 37,
                        "d": {
                            "subscriptions": {
                                "10": {
                                    "typing": true,
                                    "activities": true,
                                    "threads": true,
                                    "channels": {
                                        "20": [[0, 99]]
                                    }
                                }
                            }
                        }
                    })
                );
            }
        }
    }

    #[test]
    fn subscription_deduper_skips_exact_duplicate_channel_subscriptions() {
        let guild_id = Id::<GuildMarker>::new(10);
        let channel_id = Id::<ChannelMarker>::new(20);
        let other_channel_id = Id::<ChannelMarker>::new(30);
        let mut deduper = SubscriptionDeduper::default();

        assert!(deduper.should_send(&GatewayCommand::SubscribeDirectMessage { channel_id }));
        assert!(!deduper.should_send(&GatewayCommand::SubscribeDirectMessage { channel_id }));
        assert!(
            deduper.should_send(&GatewayCommand::SubscribeDirectMessage {
                channel_id: other_channel_id,
            })
        );

        assert!(deduper.should_send(&GatewayCommand::SubscribeGuildChannel {
            guild_id,
            channel_id,
        }));
        assert!(
            !deduper.should_send(&GatewayCommand::SubscribeGuildChannel {
                guild_id,
                channel_id,
            })
        );

        assert!(
            deduper.should_send(&GatewayCommand::UpdateMemberListSubscription {
                guild_id,
                channel_id,
                ranges: vec![(0, 99), (100, 199)],
            })
        );
        assert!(
            !deduper.should_send(&GatewayCommand::UpdateMemberListSubscription {
                guild_id,
                channel_id,
                ranges: vec![(0, 99), (100, 199)],
            })
        );
        assert!(
            deduper.should_send(&GatewayCommand::UpdateMemberListSubscription {
                guild_id,
                channel_id,
                ranges: vec![(0, 99)],
            })
        );
        assert!(
            !deduper.should_send(&GatewayCommand::UpdateMemberListSubscription {
                guild_id,
                channel_id,
                ranges: vec![(0, 99)],
            })
        );
        assert!(
            deduper.should_send(&GatewayCommand::RequestGuildMembersByIds {
                guild_id,
                user_ids: vec![Id::new(40)],
                presences: false,
            })
        );
    }

    #[test]
    fn voice_state_update_payload_joins_and_leaves_voice_channel() {
        let join_payload: serde_json::Value = serde_json::from_str(&voice_state_update_payload(
            Some(Id::<GuildMarker>::new(10)),
            Some(Id::<ChannelMarker>::new(20)),
            true,
            false,
        ))
        .expect("voice join payload should be valid json");
        assert_eq!(join_payload["op"].as_u64(), Some(4));
        assert_eq!(join_payload["d"]["guild_id"].as_str(), Some("10"));
        assert_eq!(join_payload["d"]["channel_id"].as_str(), Some("20"));
        assert_eq!(join_payload["d"]["self_mute"].as_bool(), Some(true));
        assert_eq!(join_payload["d"]["self_deaf"].as_bool(), Some(false));

        let leave_payload: serde_json::Value = serde_json::from_str(&voice_state_update_payload(
            Some(Id::<GuildMarker>::new(10)),
            None,
            true,
            false,
        ))
        .expect("voice leave payload should be valid json");
        assert!(leave_payload["d"]["channel_id"].is_null());

        // A DM or group-DM call joins with a null guild and the DM channel as
        // the voice target.
        let dm_call_payload: serde_json::Value = serde_json::from_str(&voice_state_update_payload(
            None,
            Some(Id::<ChannelMarker>::new(30)),
            false,
            false,
        ))
        .expect("dm call payload should be valid json");
        assert!(dm_call_payload["d"]["guild_id"].is_null());
        assert_eq!(dm_call_payload["d"]["channel_id"].as_str(), Some("30"));
    }
}
