mumble-web2/gui/src/lib.rs

use app::Chat;
use app::Command;
use app::ConnectionState;
use app::STATE;
use asynchronous_codec::FramedRead;
use asynchronous_codec::FramedWrite;
use color_eyre::eyre::{bail, Error};
use dioxus::prelude::*;
use futures::select;
use futures::FutureExt as _;
use futures::SinkExt as _;
use futures::StreamExt as _;
use futures_channel::mpsc::UnboundedSender;
pub use imp::spawn;
use msghtml::process_message_html;
use mumble_protocol::control::msgs;
use mumble_protocol::control::ControlCodec;
use mumble_protocol::control::ControlPacket;
use mumble_protocol::voice::VoicePacket;
use mumble_protocol::voice::VoicePacketPayload;
use mumble_protocol::Clientbound;
use mumble_protocol::Serverbound;
use mumble_web2_common::GuiConfig;
use once_cell::sync::Lazy;
use std::collections::hash_map::Entry;
use std::collections::HashMap;
use std::time::Duration;
use tracing::debug;
use tracing::error;
use tracing::info;

pub mod app;
pub mod imp;
mod msghtml;

//pub static CONFIG: Lazy<GuiConfig> = Lazy::new(|| imp::load_config().unwrap_or_default());
pub static CONFIG: async_cell::sync::AsyncCell<GuiConfig> = async_cell::sync::AsyncCell::new();

pub async fn network_entrypoint(mut event_rx: UnboundedReceiver<Command>) {
    loop {
        let Some(Command::Connect { address, username }) = event_rx.next().await else {
            panic!("did not receive connect command")
        };

        *STATE.server.write() = Default::default();
        *STATE.status.write() = ConnectionState::Connecting;
        if let Err(error) = imp::network_connect(address, username, &mut event_rx).await {
            error!("could not connect {:?}", error);
            *STATE.status.write() = ConnectionState::Failed(error.to_string());
        } else {
            *STATE.status.write() = ConnectionState::Disconnected;
        }
    }
}

pub async fn network_loop<R: imp::ImpRead, W: imp::ImpWrite>(
    username: String,
    event_rx: &mut UnboundedReceiver<Command>,
    mut reader: FramedRead<R, ControlCodec<Serverbound, Clientbound>>,
    mut writer: FramedWrite<W, ControlCodec<Serverbound, Clientbound>>,
) -> Result<(), Error> {
    let (mut send_chan, mut writer_recv_chan) = futures_channel::mpsc::unbounded();
    spawn(async move {
        while let Some(msg) = writer_recv_chan.next().await {
            if !matches!(msg, ControlPacket::Ping(_) | ControlPacket::UDPTunnel(_)) {
                info!("sending packet {:#?}", msg);
            }
            if let Err(e) = writer.send(msg).await {
                error!("error sending packet {:?}", e);
                break;
            }
        }
    });

    // Get version packet
    let version = match reader.next().await {
        Some(Ok(v)) => v,
        Some(Err(err)) => bail!("bad version packet: {err:?}"),
        None => bail!("no version was recieved"),
    };
    info!("got version packet {:#?}", version);

    // Send version packet
    let mut msg = msgs::Version::new();
    msg.set_version(0x000010204);
    msg.set_release(format!("{} {}", "mumbleweb2", "6.9.0"));
    //msg.set_os("Chrome".to_string());
    send_chan.send(msg.into()).await.unwrap();

    // Send authenticate packet
    let mut msg = msgs::Authenticate::new();
    msg.set_username(username);
    msg.set_opus(true);
    send_chan.send(msg.into()).await.unwrap();

    // Spawn worker to send pings
    {
        let mut send_chan = send_chan.clone();
        spawn(async move {
            loop {
                if let Err(_) = send_chan.send(msgs::Ping::new().into()).await {
                    break;
                }

                imp::sleep(Duration::from_millis(3000)).await;
            }
        });
    }

    let mut audio = imp::AudioSystem::new()?;
    {
        let send_chan = send_chan.clone();
        let mut sequence_num = 0;
        audio.start_recording(move |opus_frame| {
            let _ =
                send_chan.unbounded_send(ControlPacket::UDPTunnel(Box::new(VoicePacket::Audio {
                    _dst: std::marker::PhantomData,
                    target: 0,
                    session_id: (),
                    seq_num: sequence_num,
                    payload: VoicePacketPayload::Opus(opus_frame.into(), false),
                    position_info: None,
                })));
            sequence_num = sequence_num.wrapping_add(2);
        });
    }

    // Create map of session_id -> AudioDecoder
    let mut decoder_map = HashMap::new();

    let mut reader_future = reader.next().fuse();
    let mut command_future = event_rx.next();

    loop {
        select! {
            packet = reader_future => {
                reader_future = reader.next().fuse();
                match packet {
                    Some(Ok(msg)) => {
                        if !matches!(msg, ControlPacket::UDPTunnel(_) | ControlPacket::Ping(_)) {
                            info!("receiving packet {:#?}", msg);
                        }
                        let res = accept_packet(msg, &mut audio, &mut decoder_map);
                        if let Err(err) = res {
                            error!("error accepting packet {:?}", err)
                        }
                    },
                    Some(Err(err)) => {
                        error!("error receiving packet {:?}", err)
                    },
                    None => break,
                }
            }
            command = command_future => {
                command_future = event_rx.next();
                if let Some(command) = &command {
                    info!("issuing command {:#?}", command);
                }
                match command {
                    Some(Command::Disconnect) => break,
                    Some(command) => {
                        let res = accept_command(command, &mut send_chan);
                        if let Err(err) = res {
                            info!("error accepting command {:?}", err)
                        }
                    }
                    None => continue,
                }
            }
        }
    }
    let _ = send_chan.close();

    Ok(())
}

fn accept_command(
    command: Command,
    send_chan: &mut UnboundedSender<ControlPacket<mumble_protocol::Serverbound>>,
) -> Result<(), Error> {
    use Command::*;
    let Some(session) = STATE.server.read().session else {
        bail!("no session id")
    };

    match command {
        SendChat { markdown, channels } => {
            use markdown::*;
            let blocks = tokenize(&markdown);
            let html_text = match blocks.as_slice() {
                [Block::Paragraph(par)] => match par.as_slice() {
                    [Span::Text(text)] => text.to_string(),
                    _ => to_html(&markdown)
                        .trim()
                        .strip_prefix("<p>")
                        .unwrap()
                        .strip_suffix("</p>")
                        .unwrap()
                        .to_string(),
                },
                _ => to_html(&markdown).trim().to_string(),
            };

            {
                let mut server = STATE.server.write();
                let Some(me) = server.session else {
                    bail!("not signed in with a session id")
                };
                server.chat.push(Chat {
                    raw: markdown,
                    dangerous_html: html_text.clone(),
                    sender: Some(me),
                })
            }

            let mut u = msgs::TextMessage::new();
            u.set_message(html_text.to_string());
            u.set_channel_id(channels);
            let _ = send_chan.unbounded_send(u.into());
        }
        SendFile {
            ref bytes,
            name,
            mime,
            channels,
        } => {
            use base64::{display::Base64Display, prelude::BASE64_STANDARD};
            let html = match mime {
                Some(mime) if mime.type_() == "image" => format!(
                    "<img src=\"data:{};base64,{}\" />",
                    mime,
                    Base64Display::new(bytes, &BASE64_STANDARD)
                ),
                Some(mime) => format!(
                    "<a href=\"data:{};base64,{}\" download>{name}</a>",
                    mime,
                    Base64Display::new(bytes, &BASE64_STANDARD)
                ),
                None => format!(
                    "<a href=\"data:application/octet-stream;base64,{}\" download>{name}</a>",
                    Base64Display::new(bytes, &BASE64_STANDARD)
                ),
            };

            {
                let mut server = STATE.server.write();
                let Some(me) = server.session else {
                    bail!("not signed in with a session id")
                };
                server.chat.push(Chat {
                    raw: "".to_string(),
                    dangerous_html: html.clone(),
                    sender: Some(me),
                })
            }

            let mut u = msgs::TextMessage::new();
            u.set_message(html);
            u.set_channel_id(channels);
            let _ = send_chan.unbounded_send(u.into());
        }
        SetMute { mute } => {
            let mut u = msgs::UserState::new();
            u.set_session(session);
            u.set_self_mute(mute);
            let _ = send_chan.unbounded_send(u.into());
        }
        SetDeaf { deaf } => {
            let mut u = msgs::UserState::new();
            u.set_session(session);
            u.set_self_deaf(deaf);
            let _ = send_chan.unbounded_send(u.into());
        }
        EnterChannel { channel, user } => {
            let mut u = msgs::UserState::new();
            u.set_session(user);
            u.set_channel_id(channel);
            let _ = send_chan.unbounded_send(u.into());
        }
        Connect { .. } | Disconnect => (),
    }

    Ok(())
}

fn accept_packet(
    msg: ControlPacket<mumble_protocol::Clientbound>,
    audio_context: &mut imp::AudioSystem,
    player_map: &mut HashMap<u32, imp::AudioPlayer>,
) -> Result<(), Error> {
    match msg {
        ControlPacket::UDPTunnel(u) => {
            match *u.clone() {
                mumble_protocol::voice::VoicePacket::Audio {
                    _dst,
                    target,
                    session_id,
                    seq_num,
                    payload,
                    position_info,
                } => {
                    // Get or create audio decoder for this user
                    let audio_player = match player_map.entry(session_id) {
                        Entry::Occupied(occupied_entry) => occupied_entry.into_mut(),
                        Entry::Vacant(vacant_entry) => {
                            vacant_entry.insert(audio_context.create_player()?)
                        }
                    };
                    // This will over time (as users join and leave) leak
                    // AudioDecoders, MediaStreamTrackGenerators, MediaStreams, and MediaStreamAudioSourceNodes.
                    // A better way to handle this would be to delete and create all the audio
                    // infra on channel join and update it as new users join the channel, dropping
                    // any audio packets that come in the meantime.
                    if let VoicePacketPayload::Opus(audio_payload, end_bit) = payload {
                        audio_player.play_opus(audio_payload.as_ref());
                        //console::log_1(&"Oueued audio chunk for decoding".into());
                    }
                }
                _ => {
                    unreachable!("TCP tunnels UDP packets should not contain pings");
                    // I think?
                }
            }
        }
        ControlPacket::ChannelState(u) => {
            let mut server = STATE.server.write();
            let id = u.get_channel_id();

            let state = server.channels.entry(id).or_default();
            let new_parent = if u.has_parent() {
                if let Some(parent) = state.parent.and_then(|p| server.channels.get_mut(&p)) {
                    parent.children.remove(&id);
                }

                let parent_id = u.get_parent();
                let parent = server.channels.entry(parent_id).or_default();
                if u.has_position() && u.get_position() as usize <= parent.children.len() {
                    // TODO: what if positions are received out of order? we need to sort afterwards?
                    parent.children.insert_before(u.get_position() as usize, id);
                } else {
                    parent.children.insert(id);
                }
                Some(parent_id)
            } else {
                None
            };

            let state = server.channels.entry(id).or_default();
            state.parent = new_parent;
            if u.has_name() {
                state.name = u.get_name().to_string();
            }
        }
        ControlPacket::ChannelRemove(u) => {
            let mut server = STATE.server.write();
            let id = u.get_channel_id();
            if let Some(channel) = server.channels.remove(&id) {
                if let Some(parent) = channel.parent.and_then(|p| server.channels.get_mut(&p)) {
                    parent.children.remove(&id);
                }
            }
        }
        ControlPacket::UserState(u) => {
            let mut server = STATE.server.write();
            let server = &mut *server;
            let id = u.get_session();

            let state_entry = server.users.entry(id);
            let new = matches!(state_entry, std::collections::hash_map::Entry::Vacant(_));
            let state = state_entry.or_default();
            // the server might now send a channel_id if the user is in channel=0
            if u.has_channel_id() || new {
                if let Some(parent) = server.channels.get_mut(&state.channel) {
                    parent.users.remove(&id);
                }

                let channel_id = u.get_channel_id();
                server
                    .channels
                    .entry(channel_id)
                    .or_default()
                    .users
                    .insert(id);
                state.channel = channel_id;
            }

            if u.has_name() {
                state.name = u.get_name().to_string();
            }
            if u.has_mute() {
                state.mute = u.get_mute();
            }
            if u.has_deaf() {
                state.deaf = u.get_deaf();
            }
            if u.has_self_mute() {
                state.self_mute = u.get_self_mute();
            }
            if u.has_self_deaf() {
                state.self_deaf = u.get_self_deaf();
            }
        }
        ControlPacket::UserRemove(u) => {
            let mut server = STATE.server.write();
            let id = u.get_session();
            if let Some(state) = server.users.remove(&id) {
                if let Some(parent) = server.channels.get_mut(&state.channel) {
                    parent.users.remove(&id);
                }
            }
        }
        ControlPacket::TextMessage(u) => {
            let mut server = STATE.server.write();
            if u.has_message() {
                let text = u.get_message().to_string();
                server.chat.push(Chat {
                    sender: if u.has_actor() {
                        Some(u.get_actor())
                    } else {
                        None
                    },
                    dangerous_html: process_message_html(&text),
                    raw: text,
                });
            }
        }
        ControlPacket::ServerSync(u) => {
            *STATE.status.write() = ConnectionState::Connected;
            let mut server = STATE.server.write();
            if u.has_welcome_text() {
                let text = u.get_welcome_text().to_string();
                server.chat.push(Chat {
                    sender: None,
                    dangerous_html: process_message_html(&text),
                    raw: text,
                });
            }
            if u.has_session() {
                server.session = Some(u.get_session());
            }
        }
        _ => {}
    }

    Ok(())
}