concord 2.4.7

use std::{
    collections::{HashMap, HashSet, VecDeque},
    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, oneshot};
use tokio::time::{Instant, sleep};
use tokio_tungstenite::{
    connect_async_with_config,
    tungstenite::{
        Message as WsMessage,
        client::IntoClientRequest,
        handshake::client::Request,
        protocol::{CloseFrame, WebSocketConfig},
    },
};

use super::{
    ActivityInfo, PresenceStatus,
    client::AppEventPublisher,
    events::AppEvent,
    fingerprint::{
        CLIENT_BROWSER, CLIENT_BROWSER_VERSION, ClientFingerprint, DISCORD_REFERRER_CURRENT,
        DISCORD_REFERRING_DOMAIN_CURRENT, discord_gateway_headers,
    },
    state::DiscordState,
};
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) fingerprint: Arc<ClientFingerprint>,
    pub(crate) state: Arc<RwLock<DiscordState>>,
    pub(crate) gateway_session_id: Arc<RwLock<Option<String>>>,
    pub(crate) event_publisher: AppEventPublisher,
}

/// 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);
// Discord applies this budget to every JSON event on one Gateway connection.
// WebSocket control frames such as Pong and Close are not Gateway events.
const GATEWAY_SEND_LIMIT: usize = 120;
const GATEWAY_SEND_WINDOW: Duration = Duration::from_secs(60);
const GUILD_MEMBER_REQUEST_INTERVAL: Duration = Duration::from_secs(30);
const GUILD_MEMBER_REQUEST_RESPONSE_TTL: Duration = Duration::from_secs(2 * 60);
const MAX_PENDING_GUILD_MEMBER_REQUESTS: usize = 512;
const MAX_SENT_GUILD_MEMBER_REQUESTS: usize = 512;
const MAX_GATEWAY_RETRY_DELAY: Duration = Duration::from_secs(30 * 60);

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(Clone)]
struct GatewaySender {
    // Heartbeats and session setup use the urgent queue so a backlog of UI
    // commands cannot delay connection health traffic once a slot is free.
    urgent_tx: mpsc::UnboundedSender<GatewaySendRequest>,
    normal_tx: mpsc::UnboundedSender<GatewaySendRequest>,
}

struct GatewaySendRequest {
    payload: String,
    completion: Option<oneshot::Sender<Result<(), String>>>,
}

#[derive(Default)]
struct GatewaySendWindow {
    sent_at: VecDeque<Instant>,
}

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

#[derive(Default)]
struct GuildMemberRequestLimiter {
    next_available: HashMap<Id<GuildMarker>, Instant>,
}

#[derive(Clone, Debug, Eq, PartialEq)]
enum GuildMemberRequestKind {
    Search {
        query: String,
        limit: u16,
        presences: bool,
    },
    ByIds {
        user_ids: Vec<Id<UserMarker>>,
        presences: bool,
    },
}

#[derive(Clone, Debug, Eq, PartialEq)]
struct GuildMemberRequest {
    guild_id: Id<GuildMarker>,
    nonce: String,
    kind: GuildMemberRequestKind,
}

#[derive(Clone, Debug)]
struct PendingGuildMemberRequest {
    request: GuildMemberRequest,
    send_at: Instant,
}

struct InFlightGuildMemberRequest {
    completion: oneshot::Receiver<Result<(), String>>,
}

struct ScheduledGuildMemberRequest {
    request: GuildMemberRequest,
    accepted: bool,
    retry_at: Option<Instant>,
}

struct SentGuildMemberRequest {
    request: GuildMemberRequest,
    sent_at: Instant,
}

struct GuildMemberRequestScheduler {
    limiter: GuildMemberRequestLimiter,
    pending: VecDeque<PendingGuildMemberRequest>,
    in_flight: Option<ScheduledGuildMemberRequest>,
    awaiting_response: VecDeque<SentGuildMemberRequest>,
    next_nonce: u64,
}

#[derive(Default)]
struct GatewaySessionResources {
    guild_member_requests: GuildMemberRequestScheduler,
    last_presence: Option<GatewayPresence>,
}

struct GatewayPresence {
    status: PresenceStatus,
    activities: Vec<ActivityInfo>,
}

impl GuildMemberRequestLimiter {
    fn reserve(&mut self, guild_id: Id<GuildMarker>, now: Instant) -> Duration {
        let send_at = self
            .next_available
            .get(&guild_id)
            .copied()
            .filter(|next| *next > now)
            .unwrap_or(now);
        self.next_available
            .insert(guild_id, send_at + GUILD_MEMBER_REQUEST_INTERVAL);
        send_at.saturating_duration_since(now)
    }
}

impl Default for GuildMemberRequestScheduler {
    fn default() -> Self {
        Self {
            limiter: GuildMemberRequestLimiter::default(),
            pending: VecDeque::new(),
            in_flight: None,
            awaiting_response: VecDeque::new(),
            next_nonce: 1,
        }
    }
}

impl GuildMemberRequest {
    fn payload(&self) -> String {
        match &self.kind {
            GuildMemberRequestKind::Search {
                query,
                limit,
                presences,
            } => request_guild_members_payload(
                self.guild_id,
                query,
                *limit,
                *presences,
                Some(&self.nonce),
            ),
            GuildMemberRequestKind::ByIds {
                user_ids,
                presences,
            } => request_guild_members_by_ids_payload(
                self.guild_id,
                user_ids,
                *presences,
                &self.nonce,
            ),
        }
    }

    fn is_coalescible_search(&self) -> bool {
        matches!(
            &self.kind,
            GuildMemberRequestKind::Search { query, limit, .. }
                if !query.is_empty() || *limit > 0
        )
    }
}

impl GuildMemberRequestScheduler {
    fn enqueue_search(
        &mut self,
        guild_id: Id<GuildMarker>,
        query: String,
        limit: u16,
        presences: bool,
        nonce: Option<String>,
        now: Instant,
    ) -> bool {
        self.prune_awaiting(now);
        let request = GuildMemberRequest {
            guild_id,
            nonce: nonce.unwrap_or_else(|| self.next_nonce()),
            kind: GuildMemberRequestKind::Search {
                query,
                limit,
                presences,
            },
        };
        self.awaiting_response
            .retain(|sent| sent.request.nonce != request.nonce);

        if request.is_coalescible_search()
            && let Some(pending) = self.pending.iter_mut().rev().find(|pending| {
                pending.request.guild_id == guild_id && pending.request.is_coalescible_search()
            })
        {
            pending.request = request;
            return true;
        }

        if self.pending.len() >= MAX_PENDING_GUILD_MEMBER_REQUESTS {
            let Some(position) = self
                .pending
                .iter()
                .position(|pending| pending.request.is_coalescible_search())
            else {
                return false;
            };
            self.pending.remove(position);
        }

        self.enqueue_request(request, now)
    }

    fn enqueue_by_ids(
        &mut self,
        guild_id: Id<GuildMarker>,
        user_ids: Vec<Id<UserMarker>>,
        presences: bool,
        now: Instant,
    ) {
        self.prune_awaiting(now);
        let mut remaining = user_ids;
        remaining.sort_unstable();
        remaining.dedup();

        let compatible_requests = self.pending.iter().filter(|pending| {
            pending.request.guild_id == guild_id
                && matches!(
                    pending.request.kind,
                    GuildMemberRequestKind::ByIds {
                        presences: pending_presences,
                        ..
                    } if pending_presences == presences
                )
        });
        let mut available = 0usize;
        for pending in compatible_requests {
            let GuildMemberRequestKind::ByIds { user_ids, .. } = &pending.request.kind else {
                continue;
            };
            remaining.retain(|user_id| !user_ids.contains(user_id));
            available += 100usize.saturating_sub(user_ids.len());
        }
        let new_request_count = remaining.len().saturating_sub(available).div_ceil(100);
        self.make_room_for_by_ids(new_request_count);

        for pending in self.pending.iter_mut().filter(|pending| {
            pending.request.guild_id == guild_id
                && matches!(
                    pending.request.kind,
                    GuildMemberRequestKind::ByIds {
                        presences: pending_presences,
                        ..
                    } if pending_presences == presences
                )
        }) {
            let GuildMemberRequestKind::ByIds { user_ids, .. } = &mut pending.request.kind else {
                continue;
            };
            remaining.retain(|user_id| !user_ids.contains(user_id));
            let available = 100usize.saturating_sub(user_ids.len());
            let added = remaining.len().min(available);
            user_ids.extend(remaining.drain(..added));
            if remaining.is_empty() {
                return;
            }
        }

        for user_ids in remaining.chunks(100) {
            let request = GuildMemberRequest {
                guild_id,
                nonce: self.next_nonce(),
                kind: GuildMemberRequestKind::ByIds {
                    user_ids: user_ids.to_vec(),
                    presences,
                },
            };
            self.enqueue_by_ids_request(request, now);
        }
    }

    fn make_room_for_by_ids(&mut self, additional: usize) {
        let overflow = self
            .pending
            .len()
            .saturating_add(additional)
            .saturating_sub(MAX_PENDING_GUILD_MEMBER_REQUESTS);
        if overflow == 0 {
            return;
        }
        let positions = self
            .pending
            .iter()
            .enumerate()
            .filter_map(|(index, pending)| pending.request.is_coalescible_search().then_some(index))
            .take(overflow)
            .collect::<Vec<_>>();
        for position in positions.into_iter().rev() {
            self.pending.remove(position);
        }
    }

    fn enqueue_request(&mut self, request: GuildMemberRequest, now: Instant) -> bool {
        if self.pending.len() >= MAX_PENDING_GUILD_MEMBER_REQUESTS {
            return false;
        }
        let guild_id = request.guild_id;
        let send_at = now + self.limiter.reserve(guild_id, now);
        self.pending
            .push_back(PendingGuildMemberRequest { request, send_at });
        true
    }

    fn enqueue_by_ids_request(&mut self, request: GuildMemberRequest, now: Instant) {
        // Hydration requests resolve concrete users already visible in the UI,
        // so losing them is worse than briefly exceeding the search queue's
        // soft memory bound. IDs are deduplicated and grouped into 100-member
        // payloads here, while the per-guild limiter still controls sends.
        let guild_id = request.guild_id;
        let send_at = now + self.limiter.reserve(guild_id, now);
        self.pending
            .push_back(PendingGuildMemberRequest { request, send_at });
    }

    fn next_nonce(&mut self) -> String {
        let nonce = format!("concord-{:016x}", self.next_nonce);
        self.next_nonce = self.next_nonce.wrapping_add(1).max(1);
        nonce
    }

    fn next_delay(&self, now: Instant) -> Option<Duration> {
        self.pending
            .iter()
            .map(|request| request.send_at.saturating_duration_since(now))
            .min()
    }

    fn start_due(&mut self, now: Instant) -> Option<String> {
        if self.in_flight.is_some() {
            return None;
        }
        let index = self
            .pending
            .iter()
            .enumerate()
            .filter(|(_, request)| request.send_at <= now)
            .min_by_key(|(_, request)| request.send_at)
            .map(|(index, _)| index)?;
        let pending = self
            .pending
            .remove(index)
            .expect("due guild member request exists");
        let payload = pending.request.payload();
        self.in_flight = Some(ScheduledGuildMemberRequest {
            request: pending.request,
            accepted: false,
            retry_at: None,
        });
        Some(payload)
    }

    fn complete_send(&mut self, sent_at: Instant) {
        let completed = self
            .in_flight
            .take()
            .expect("sent guild member request exists");
        let guild_id = completed.request.guild_id;
        if let Some(retry_at) = completed.retry_at {
            self.pending.push_front(PendingGuildMemberRequest {
                request: completed.request,
                send_at: retry_at,
            });
            self.reschedule_guild(guild_id, retry_at);
            return;
        }

        self.reschedule_guild(guild_id, sent_at + GUILD_MEMBER_REQUEST_INTERVAL);
        if completed.accepted {
            return;
        }

        self.prune_awaiting(sent_at);
        if self.awaiting_response.len() >= MAX_SENT_GUILD_MEMBER_REQUESTS {
            self.awaiting_response.pop_front();
        }
        self.awaiting_response.push_back(SentGuildMemberRequest {
            request: completed.request,
            sent_at,
        });
    }

    fn cancel_in_flight(&mut self, now: Instant) {
        let Some(in_flight) = self.in_flight.take() else {
            return;
        };
        if in_flight.accepted {
            return;
        }
        let guild_id = in_flight.request.guild_id;
        let send_at = in_flight
            .retry_at
            .unwrap_or(now + GUILD_MEMBER_REQUEST_INTERVAL);
        self.pending.push_front(PendingGuildMemberRequest {
            request: in_flight.request,
            send_at,
        });
        self.reschedule_guild(guild_id, send_at);
    }

    fn apply_rate_limit(
        &mut self,
        guild_id: Id<GuildMarker>,
        nonce: Option<&str>,
        retry_after: Duration,
        now: Instant,
    ) {
        self.prune_awaiting(now);
        let retry_at = now + retry_after;
        let has_newer_search = self.pending.iter().any(|pending| {
            pending.request.guild_id == guild_id && pending.request.is_coalescible_search()
        });

        let in_flight_matches = self.in_flight.as_ref().is_some_and(|in_flight| {
            in_flight.request.guild_id == guild_id
                && nonce.is_none_or(|nonce| in_flight.request.nonce == nonce)
        });
        if in_flight_matches {
            let in_flight = self
                .in_flight
                .as_mut()
                .expect("matching guild member request is in flight");
            if in_flight.request.is_coalescible_search() && has_newer_search {
                in_flight.accepted = true;
                in_flight.retry_at = None;
                self.reschedule_guild(guild_id, retry_at);
            } else {
                in_flight.retry_at = Some(retry_at);
                in_flight.accepted = false;
            }
            return;
        }

        let position = match nonce {
            Some(nonce) => self
                .awaiting_response
                .iter()
                .rposition(|sent| sent.request.guild_id == guild_id && sent.request.nonce == nonce),
            None => self
                .awaiting_response
                .iter()
                .rposition(|sent| sent.request.guild_id == guild_id),
        };
        if let Some(position) = position {
            let sent = self
                .awaiting_response
                .remove(position)
                .expect("rate-limited guild member request exists");
            if !sent.request.is_coalescible_search() || !has_newer_search {
                self.pending.push_front(PendingGuildMemberRequest {
                    request: sent.request,
                    send_at: retry_at,
                });
            }
        }
        self.reschedule_guild(guild_id, retry_at);
    }

    fn acknowledge(&mut self, nonce: &str) {
        if let Some(in_flight) = self.in_flight.as_mut()
            && in_flight.request.nonce == nonce
        {
            in_flight.accepted = true;
            return;
        }
        if let Some(position) = self
            .awaiting_response
            .iter()
            .rposition(|sent| sent.request.nonce == nonce)
        {
            self.awaiting_response.remove(position);
            return;
        }
        if let Some(position) = self
            .pending
            .iter()
            .rposition(|pending| pending.request.nonce == nonce)
        {
            self.pending.remove(position);
        }
    }

    fn reschedule_guild(&mut self, guild_id: Id<GuildMarker>, earliest: Instant) {
        let mut next_available = earliest;
        for pending in self
            .pending
            .iter_mut()
            .filter(|pending| pending.request.guild_id == guild_id)
        {
            pending.send_at = pending.send_at.max(next_available);
            next_available = pending.send_at + GUILD_MEMBER_REQUEST_INTERVAL;
        }
        self.limiter.next_available.insert(guild_id, next_available);
    }

    fn prune_awaiting(&mut self, now: Instant) {
        while self.awaiting_response.front().is_some_and(|sent| {
            now.saturating_duration_since(sent.sent_at) >= GUILD_MEMBER_REQUEST_RESPONSE_TTL
        }) {
            self.awaiting_response.pop_front();
        }
    }

    fn prepare_reconnect(&mut self, now: Instant) {
        self.prune_awaiting(now);
        self.cancel_in_flight(now);
        self.recover_awaiting(now + GUILD_MEMBER_REQUEST_INTERVAL);
    }

    fn recover_awaiting(&mut self, earliest: Instant) {
        let mut recovered = VecDeque::new();
        let mut recovered_guilds = HashSet::new();
        while let Some(sent) = self.awaiting_response.pop_back() {
            let superseded_search = sent.request.is_coalescible_search()
                && self.pending.iter().chain(recovered.iter()).any(|pending| {
                    pending.request.guild_id == sent.request.guild_id
                        && pending.request.is_coalescible_search()
                });
            if !superseded_search {
                recovered_guilds.insert(sent.request.guild_id);
                recovered.push_front(PendingGuildMemberRequest {
                    request: sent.request,
                    send_at: earliest,
                });
            }
        }
        recovered.append(&mut self.pending);
        self.pending = recovered;
        for guild_id in recovered_guilds {
            self.reschedule_guild(guild_id, earliest);
        }
    }

    fn start_new_session(&mut self, now: Instant) {
        self.prune_awaiting(now);
        self.cancel_in_flight(now);
        self.recover_awaiting(now);
        self.limiter = GuildMemberRequestLimiter::default();
        for pending in &mut self.pending {
            pending.send_at = now + self.limiter.reserve(pending.request.guild_id, now);
        }
    }
}

impl InFlightGuildMemberRequest {
    async fn wait(&mut self) -> Result<(), String> {
        (&mut self.completion)
            .await
            .map_err(|_| "gateway writer task stopped before send completed".to_owned())?
    }
}

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> {
    state: &'a Arc<RwLock<DiscordState>>,
    gateway_session_id: &'a Arc<RwLock<Option<String>>>,
    event_publisher: &'a AppEventPublisher,
}

#[derive(Clone, Copy)]
struct FrameContext<'a> {
    sequence_cell: &'a Arc<Mutex<Option<u64>>>,
    heartbeat_ack: &'a Arc<Mutex<HeartbeatAckState>>,
    sender: &'a GatewaySender,
    fingerprint: &'a ClientFingerprint,
    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,
    /// Whether the current connection reached READY or RESUMED. Read and
    /// cleared by the reconnect loop to reset the backoff after a healthy
    /// session.
    established: 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 resources = GatewaySessionResources::default();
    let mut backoff = RECONNECT_BASE_DELAY;
    let mut publish_gateway_closed = true;

    loop {
        let publish = GatewayPublishContext {
            state: &runtime.state,
            gateway_session_id: &runtime.gateway_session_id,
            event_publisher: &runtime.event_publisher,
        };
        let outcome = match connect_and_run(
            &token,
            &mut commands,
            &mut session,
            &mut resources,
            &runtime.fingerprint,
            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. A connection that reached READY/RESUMED was healthy,
        // so its disconnect restarts the backoff from the base delay.
        if std::mem::take(&mut session.established) {
            backoff = RECONNECT_BASE_DELAY;
        }
        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 {
            state: &runtime.state,
            gateway_session_id: &runtime.gateway_session_id,
            event_publisher: &runtime.event_publisher,
        };
        publish_gateway_event(publish, AppEvent::GatewayClosed).await;
    }
}

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

    let request = gateway_request(&url, fingerprint)?;
    let (ws, _response) =
        connect_async_with_config(request, 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 (sender, mut gateway_send_error_rx, gateway_writer_task) =
        spawn_gateway_sender(Arc::clone(&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(&sender, payload).await?;
        logging::debug("gateway", "RESUME sent");
    } else {
        resources
            .guild_member_requests
            .start_new_session(Instant::now());
        if session.has_received_ready && resources.last_presence.is_none() {
            let state = publish
                .state
                .read()
                .expect("discord state lock is not poisoned");
            resources.last_presence = current_gateway_presence(&state);
        }
        let reidentify_presence = session
            .has_received_ready
            .then_some(resources.last_presence.as_ref())
            .flatten();
        let payload = build_identify_payload(token, fingerprint, reidentify_presence);
        send_text(&sender, 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 sender_for_heartbeat = sender.clone();
    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 = Duration::from_millis(
        rand::thread_rng().gen_range(0..=heartbeat_interval.as_millis() as u64),
    );
    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(&sender_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 mut member_request_send: Option<InFlightGuildMemberRequest> = None;
    let outcome = loop {
        let member_request_delay = member_request_send
            .is_none()
            .then(|| resources.guild_member_requests.next_delay(Instant::now()))
            .flatten();
        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(
                                &sender,
                                command,
                                &mut subscription_deduper,
                                resources,
                            )
                        {
                            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,
                            sender: &sender,
                            fingerprint,
                            publish,
                        };
                        match handle_frame(
                            value,
                            session,
                            frame_context,
                            resources,
                        ).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;
            }
            Some(error) = gateway_send_error_rx.recv() => {
                log_and_publish_gateway_error(publish, error).await;
                break ConnectionOutcome::Resume;
            }
            send_result = async {
                member_request_send
                    .as_mut()
                    .expect("guard ensures a guild member request is in flight")
                    .wait()
                    .await
            }, if member_request_send.is_some() => {
                member_request_send
                    .take()
                    .expect("completed guild member request exists");
                if let Err(error) = send_result {
                    let message = format!("guild member request send failed: {error}");
                    log_and_publish_gateway_error(publish, message).await;
                    break ConnectionOutcome::Resume;
                }
                resources
                    .guild_member_requests
                    .complete_send(Instant::now());
            }
            _ = sleep(member_request_delay.unwrap_or_default()), if member_request_delay.is_some() => {
                let Some(payload) = resources
                    .guild_member_requests
                    .start_due(Instant::now())
                else {
                    continue;
                };
                let completion = match sender.enqueue_normal(payload) {
                    Ok(completion) => completion,
                    Err(error) => {
                        let message = format!("guild member request send failed: {error}");
                        log_and_publish_gateway_error(publish, message).await;
                        break ConnectionOutcome::Resume;
                    }
                };
                member_request_send = Some(InFlightGuildMemberRequest { completion });
            }
        }
    };

    if matches!(
        outcome,
        ConnectionOutcome::Resume | ConnectionOutcome::Reidentify
    ) {
        resources
            .guild_member_requests
            .prepare_reconnect(Instant::now());
    } else {
        resources
            .guild_member_requests
            .cancel_in_flight(Instant::now());
    }
    heartbeat_task.abort();
    gateway_writer_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))
}

fn gateway_request(url: &str, fingerprint: &ClientFingerprint) -> Result<Request, String> {
    let mut request = url
        .into_client_request()
        .map_err(|error| format!("websocket request failed: {error}"))?;
    request
        .headers_mut()
        .extend(discord_gateway_headers(fingerprint));
    Ok(request)
}

enum FrameOutcome {
    Continue,
    Resume,
    Reidentify,
}

async fn handle_frame(
    value: Value,
    session: &mut SessionState,
    context: FrameContext<'_>,
    resources: &mut GatewaySessionResources,
) -> 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;
            if dispatch_type == "RATE_LIMITED"
                && let Some(rate_limit) = gateway_guild_member_rate_limit(&value)
            {
                logging::debug(
                    "gateway",
                    format!(
                        "guild member requests rate limited: guild={} retry_after_ms={}",
                        rate_limit.guild_id.get(),
                        rate_limit.retry_after.as_millis()
                    ),
                );
                resources.guild_member_requests.apply_rate_limit(
                    rate_limit.guild_id,
                    rate_limit.nonce.as_deref(),
                    rate_limit.retry_after,
                    Instant::now(),
                );
            } else if dispatch_type == "GUILD_MEMBERS_CHUNK"
                && let Some(nonce) = gateway_guild_member_chunk_nonce(&value)
            {
                resources.guild_member_requests.acknowledge(nonce);
            }
            // 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;
                if let Some(installation_id) = ready_installation_id(d) {
                    match context.fingerprint.update_installation_id(installation_id) {
                        Ok(true) => {
                            logging::debug("fingerprint", "updated installation id from READY");
                        }
                        Ok(false) => {}
                        Err(error) => logging::debug(
                            "fingerprint",
                            format!("could not persist READY installation id: {error}"),
                        ),
                    }
                }
                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;
                session.established = true;
            } else if dispatch_type == "RESUMED" {
                session.established = true;
                publish_gateway_event(context.publish, AppEvent::GatewayResumed).await;
            }
            if let Some(parsed) = parse_user_account_dispatch(value) {
                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.sender, 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) {
    context.event_publisher.publish(event).await;
}

fn ready_installation_id(ready: &Value) -> Option<&str> {
    ready
        .get("apex_experiments")
        .and_then(|experiments| experiments.get("installation"))
        .and_then(Value::as_str)
}

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()
    }
}

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

    if let GatewayCommand::UpdatePresence { status, activities } = &command {
        resources.last_presence = Some(GatewayPresence {
            status: *status,
            activities: activities.clone(),
        });
    }
    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
                ),
            );
            if !resources.guild_member_requests.enqueue_search(
                guild_id,
                query,
                limit,
                presences,
                nonce,
                Instant::now(),
            ) {
                logging::debug(
                    "gateway",
                    "dropping guild member search because the session queue is full",
                );
            }
            return Ok(());
        }
        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
                ),
            );
            resources.guild_member_requests.enqueue_by_ids(
                guild_id,
                user_ids,
                presences,
                Instant::now(),
            );
            return Ok(());
        }
        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(()),
    };
    sender.enqueue_text(payload)
}

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(sender: &GatewaySender, payload: String) -> Result<(), String> {
    sender.send_urgent(payload).await
}

impl GatewaySender {
    async fn send_urgent(&self, payload: String) -> Result<(), String> {
        let (completion_tx, completion_rx) = oneshot::channel();
        self.urgent_tx
            .send(GatewaySendRequest {
                payload,
                completion: Some(completion_tx),
            })
            .map_err(|_| "gateway writer task stopped".to_owned())?;
        completion_rx
            .await
            .map_err(|_| "gateway writer task stopped before send completed".to_owned())?
    }

    fn enqueue_text(&self, payload: String) -> Result<(), String> {
        self.normal_tx
            .send(GatewaySendRequest {
                payload,
                completion: None,
            })
            .map_err(|_| "gateway writer task stopped".to_owned())
    }

    fn enqueue_normal(
        &self,
        payload: String,
    ) -> Result<oneshot::Receiver<Result<(), String>>, String> {
        let (completion_tx, completion_rx) = oneshot::channel();
        self.normal_tx
            .send(GatewaySendRequest {
                payload,
                completion: Some(completion_tx),
            })
            .map_err(|_| "gateway writer task stopped".to_owned())?;
        Ok(completion_rx)
    }
}

impl GatewaySendWindow {
    fn delay_at(&mut self, now: Instant) -> Option<Duration> {
        while self
            .sent_at
            .front()
            .is_some_and(|sent_at| now.duration_since(*sent_at) >= GATEWAY_SEND_WINDOW)
        {
            self.sent_at.pop_front();
        }
        if self.sent_at.len() < GATEWAY_SEND_LIMIT {
            return None;
        }
        self.sent_at
            .front()
            .map(|sent_at| (*sent_at + GATEWAY_SEND_WINDOW).duration_since(now))
    }

    fn record(&mut self, now: Instant) {
        self.sent_at.push_back(now);
    }
}

fn spawn_gateway_sender(
    writer: WriterHandle,
) -> (
    GatewaySender,
    mpsc::UnboundedReceiver<String>,
    tokio::task::JoinHandle<()>,
) {
    let (urgent_tx, urgent_rx) = mpsc::unbounded_channel();
    let (normal_tx, normal_rx) = mpsc::unbounded_channel();
    let (error_tx, error_rx) = mpsc::unbounded_channel();
    let task = tokio::spawn(run_gateway_sender(writer, urgent_rx, normal_rx, error_tx));
    (
        GatewaySender {
            urgent_tx,
            normal_tx,
        },
        error_rx,
        task,
    )
}

#[derive(Debug, Eq, PartialEq)]
struct GuildMemberRateLimit {
    guild_id: Id<GuildMarker>,
    nonce: Option<String>,
    retry_after: Duration,
}

fn gateway_guild_member_rate_limit(value: &Value) -> Option<GuildMemberRateLimit> {
    let data = value.get("d")?;
    if data.get("opcode").and_then(Value::as_u64) != Some(8) {
        return None;
    }
    let guild_id = data
        .get("meta")?
        .get("guild_id")?
        .as_str()?
        .parse::<u64>()
        .ok()
        .and_then(Id::new_checked)?;
    let retry_after = data.get("retry_after")?.as_f64()?;
    if !retry_after.is_finite() || retry_after < 0.0 {
        return None;
    }
    Some(GuildMemberRateLimit {
        guild_id,
        nonce: data
            .get("meta")
            .and_then(|meta| meta.get("nonce"))
            .and_then(Value::as_str)
            .map(str::to_owned),
        retry_after: Duration::from_secs_f64(
            retry_after.min(MAX_GATEWAY_RETRY_DELAY.as_secs_f64()),
        ),
    })
}

fn gateway_guild_member_chunk_nonce(value: &Value) -> Option<&str> {
    value
        .get("d")
        .and_then(|data| data.get("nonce"))
        .and_then(Value::as_str)
}

async fn run_gateway_sender(
    writer: WriterHandle,
    mut urgent_rx: mpsc::UnboundedReceiver<GatewaySendRequest>,
    mut normal_rx: mpsc::UnboundedReceiver<GatewaySendRequest>,
    error_tx: mpsc::UnboundedSender<String>,
) {
    let mut urgent = VecDeque::new();
    let mut normal = VecDeque::new();
    let mut urgent_open = true;
    let mut normal_open = true;
    let mut window = GatewaySendWindow::default();

    loop {
        drain_gateway_requests(&mut urgent_rx, &mut urgent, &mut urgent_open);
        drain_gateway_requests(&mut normal_rx, &mut normal, &mut normal_open);

        if urgent.is_empty() && normal.is_empty() {
            if !urgent_open && !normal_open {
                return;
            }
            tokio::select! {
                biased;
                request = urgent_rx.recv(), if urgent_open => {
                    match request {
                        Some(request) => urgent.push_back(request),
                        None => urgent_open = false,
                    }
                }
                request = normal_rx.recv(), if normal_open => {
                    match request {
                        Some(request) => normal.push_back(request),
                        None => normal_open = false,
                    }
                }
            }
            continue;
        }

        if let Some(delay) = window.delay_at(Instant::now()) {
            tokio::select! {
                biased;
                request = urgent_rx.recv(), if urgent_open => {
                    match request {
                        Some(request) => urgent.push_back(request),
                        None => urgent_open = false,
                    }
                }
                _ = sleep(delay) => {}
            }
            continue;
        }

        let request = urgent
            .pop_front()
            .or_else(|| normal.pop_front())
            .expect("gateway send queue is not empty");
        window.record(Instant::now());
        let result = {
            let mut writer = writer.lock().await;
            writer
                .send(WsMessage::Text(request.payload.into()))
                .await
                .map_err(|error| format!("websocket send failed: {error}"))
        };
        if let Some(completion) = request.completion {
            let _ = completion.send(result.clone());
        }
        if let Err(error) = result {
            let _ = error_tx.send(error);
            return;
        }
    }
}

fn drain_gateway_requests(
    receiver: &mut mpsc::UnboundedReceiver<GatewaySendRequest>,
    queue: &mut VecDeque<GatewaySendRequest>,
    open: &mut bool,
) {
    while *open {
        match receiver.try_recv() {
            Ok(request) => queue.push_back(request),
            Err(mpsc::error::TryRecvError::Empty) => return,
            Err(mpsc::error::TryRecvError::Disconnected) => {
                *open = false;
                return;
            }
        }
    }
}

fn build_identify_payload(
    token: &str,
    fingerprint: &ClientFingerprint,
    presence: Option<&GatewayPresence>,
) -> String {
    let mut properties = json!({
        "os": fingerprint.os,
        "browser": CLIENT_BROWSER,
        "device": "",
        "system_locale": fingerprint.system_locale,
        "browser_user_agent": fingerprint.user_agent,
        "browser_version": CLIENT_BROWSER_VERSION,
        "os_version": fingerprint.os_version,
        "referrer": "",
        "referring_domain": "",
        "referrer_current": DISCORD_REFERRER_CURRENT,
        "referring_domain_current": DISCORD_REFERRING_DOMAIN_CURRENT,
        "release_channel": "stable",
        "client_build_number": fingerprint.client_build_number,
        "client_event_source": Value::Null,
    });
    if let Some(installation_id) = fingerprint.installation_id() {
        properties["installation_id"] = Value::String(installation_id);
    }

    let presence = presence
        .map(|presence| gateway_presence_payload(&presence.status, &presence.activities))
        .unwrap_or_else(|| gateway_presence_payload(&PresenceStatus::Unknown, &[]));

    json!({
        "op": 2,
        "d": {
            "token": token,
            "capabilities": USER_ACCOUNT_CAPABILITIES,
            "properties": properties,
            "presence": presence,
            "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,
    nonce: &str,
) -> 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,
            "nonce": nonce,
        },
    })
    .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": gateway_presence_payload(&status, activities),
    })
    .to_string()
}

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

fn current_gateway_presence(state: &DiscordState) -> Option<GatewayPresence> {
    let user_id = state.current_user_id()?;
    Some(GatewayPresence {
        status: state.user_presence(user_id)?,
        activities: state.user_activities(user_id).to_vec(),
    })
}

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;