fomo-reducer/src/track_vcs.rs

use std::collections::BTreeMap;

use dashmap::DashMap;
use futures::{StreamExt, stream::FuturesUnordered};
use tokio::sync::watch;
use twilight_model::{
    gateway::payload::incoming::VoiceStateUpdate,
    id::{
        Id,
        marker::{ChannelMarker, GuildMarker, UserMarker},
    },
};

use crate::{OneToManyUniqueBTreeMapWithData, OneToOneBTreeMap, UserInVCData, VoiceStatus};

pub type GuildVoiceChannelToTextChannel =
    BTreeMap<Id<GuildMarker>, OneToOneBTreeMap<Id<ChannelMarker>, Id<ChannelMarker>>>;

type VCsInGuild = OneToManyUniqueBTreeMapWithData<Id<ChannelMarker>, Id<UserMarker>, UserInVCData>;

#[derive(Debug, Default)]
pub struct VCs(DashMap<Id<GuildMarker>, watch::Sender<VCsInGuild>>);

impl Extend<(Id<GuildMarker>, VCsInGuild)> for VCs {
    fn extend<T: IntoIterator<Item = (Id<GuildMarker>, VCsInGuild)>>(&mut self, iter: T) {
        for (id, guild_vcs) in iter {
            self.0.insert(id, watch::Sender::new(guild_vcs));
        }
    }
}

impl VCs {
    pub fn with_guild<R>(&self, id: Id<GuildMarker>, f: impl FnOnce(&VCsInGuild) -> R) -> R {
        f(&*self.0.entry(id).or_default().borrow())
    }

    pub fn update_guild<R>(&self, id: Id<GuildMarker>, f: impl FnOnce(&mut VCsInGuild) -> R) -> R {
        let mut ret_opt = None;
        self.0.entry(id).or_default().send_modify(|guild_vcs| {
            let ret = f(guild_vcs);
            _ = ret_opt.insert(ret);
        });
        let ret = ret_opt.unwrap();
        ret
    }

    pub fn subscribe_to_guild<R>(&self, id: Id<GuildMarker>) -> watch::Receiver<VCsInGuild> {
        self.0.entry(id).or_default().subscribe()
    }
}

#[tracing::instrument(skip(discord_client), ret)]
async fn initialize_user_in_vc(
    discord_client: &twilight_http::Client,
    guild_id: Id<GuildMarker>,
    user_id: Id<UserMarker>,
) -> Option<(Id<ChannelMarker>, UserInVCData)> {
    if let Ok(voice_state_res) = discord_client.user_voice_state(guild_id, user_id).await
        && let Ok(voice_state) = voice_state_res.model().await
    {
        tracing::info!(?user_id, ?voice_state);

        let voice_status = VoiceStatus::builder()
            .self_deafened(voice_state.self_deaf)
            .self_muted(voice_state.self_mute)
            .server_deafened(voice_state.deaf)
            .server_muted(voice_state.mute)
            .camming(voice_state.self_video)
            .streaming(voice_state.self_stream)
            .build();
        let user_in_vc_data = voice_status.into();

        voice_state
            .channel_id
            .map(|channel_id| (channel_id, user_in_vc_data))
    } else {
        None // TODO
    }
}

#[tracing::instrument(skip(discord_client), ret)]
async fn initialize_server_vcs(
    discord_client: &twilight_http::Client,
    id: Id<GuildMarker>,
) -> VCsInGuild {
    if let Ok(guild_members_res) = discord_client.guild_members(id).limit(999).await
        && let Ok(guild_members) = guild_members_res.model().await
    {
        FuturesUnordered::from_iter(guild_members.into_iter().map(|member| async move {
            (
                member.user.id,
                initialize_user_in_vc(discord_client, id, member.user.id).await,
            )
        }))
        .filter_map(
            |(user_id, channel_id_and_user_in_vc_data_option)| async move {
                channel_id_and_user_in_vc_data_option
                    .map(|(channel_id, user_in_vc_data)| (channel_id, user_id, user_in_vc_data))
            },
        )
        .collect()
        .await
    } else {
        Default::default()
    }
}

#[tracing::instrument(skip(discord_client), ret)]
pub async fn initialize_vcs(discord_client: &twilight_http::Client) -> VCs {
    if let Ok(guilds_res) = discord_client.current_user_guilds().limit(200).await
        && let Ok(guilds) = guilds_res.model().await
    {
        FuturesUnordered::from_iter(guilds.into_iter().map(|guild| async move {
            let guild_vcs = initialize_server_vcs(discord_client, guild.id).await;

            (guild.id, guild_vcs)
        }))
        .collect()
        .await
    } else {
        Default::default()
    }
}

#[tracing::instrument(skip(vcs))]
pub fn update_vcs(voice_state_update: &VoiceStateUpdate, vcs: &VCs) {
    let user_id = voice_state_update.user_id;
    match voice_state_update.guild_id {
        Some(guild_id) => match voice_state_update.channel_id {
            Some(channel_id) => {
                let voice_status = VoiceStatus::builder()
                    .self_deafened(voice_state_update.self_deaf)
                    .self_muted(voice_state_update.self_mute)
                    .server_deafened(voice_state_update.deaf)
                    .server_muted(voice_state_update.mute)
                    .camming(voice_state_update.self_video)
                    .streaming(voice_state_update.self_stream)
                    .build();
                let user_in_vc_data = voice_status.into();

                vcs.update_guild(guild_id, |guild_vcs| {
                    guild_vcs.insert(channel_id, user_id, user_in_vc_data)
                });

                tracing::info!(
                    ?guild_id,
                    ?channel_id,
                    ?user_id,
                    "connected or otherwise changed state while connected"
                );
            }

            None => {
                vcs.update_guild(guild_id, |guild_vcs| guild_vcs.remove_right(&user_id));

                tracing::info!(?guild_id, ?user_id, "disconnected");
            }
        },

        None => {
            tracing::error!("why doesn't this have a guild id attached?!");
        }
    }
}