clemsbot/src/app/twitch.rs

use twitch_api2::{
    TwitchClient,
    types::UserId,
    twitch_oauth2::UserToken,
    pubsub::channel_points::Redemption,
};
use irc::{
    client::prelude::Message,
    proto::{Command, Response},
};
use tokio::sync::RwLock;
use tokio_tungstenite::tungstenite::Message as WsMessage;
use crate::app::{
    State,
    config::CommandExecutor,
    rhai_tools::ExecutorState,
};
use std::sync::Arc;

pub struct Twitch {
    pub client: TwitchClient<'static, reqwest::Client>,
    config: Arc<RwLock<Config>>,
}

impl Twitch {
    pub fn new(client: TwitchClient<'static, reqwest::Client>, config: Arc<RwLock<Config>>) -> Self {
        Self {
            client,
            config,
        }
    }

    pub async fn bot_token(&self) -> anyhow::Result<UserToken> {
        let mut bot_token = self.config.read().await.bot_token.as_ref().unwrap().clone();
        if bot_token.token.expires_in() <= std::time::Duration::from_secs(60 * 15) {
            bot_token.refresh_token(&reqwest::Client::new()).await?;
            self.config.write().await.bot_token.replace(bot_token.clone());
        }

        Ok(bot_token.token)
    }

    pub async fn user_token(&self) -> anyhow::Result<UserToken> {
        let mut user_token = self.config.read().await.user_token.as_ref().unwrap().clone();
        if user_token.token.expires_in() <= std::time::Duration::from_secs(60 * 15) {
            user_token.refresh_token(&reqwest::Client::new()).await?;
            self.config.write().await.user_token.replace(user_token.clone());
        }

        Ok(user_token.token)
    }
}

pub async fn handle_irc_event(state: Arc<State>, event: Message) -> anyhow::Result<()> {
    let channel_name = format!("#{}", state.channel_name);

    match &event.command {
        Command::Response(resp, _) if *resp == Response::RPL_WELCOME => {
            state.irc.read().await.send_join(&channel_name)?;
        }
        // FIXME: do correct checking here
        Command::PRIVMSG(channel, message) if *channel == channel_name => {
            on_privmsg(state, &event, &message).await?;
        }
        _ => {
            // eprintln!("{:#?}", c);
        }
    }

    Ok(())
}

async fn on_privmsg(state: Arc<State>, event: &Message, message: &str) -> anyhow::Result<()> {
    let initiator = event.source_nickname()
        .ok_or_else(|| anyhow::anyhow!("missing source"))?
    .to_string();

    let tags = event.tags.as_ref().ok_or_else(|| anyhow::anyhow!("missing tags"))?;

    let initiator_id = tags
        .iter()
        .find(|tag| tag.0 == "user-id")
        .and_then(|tag| tag.1.clone())
        .ok_or_else(|| anyhow::anyhow!("missing user id"))?;

    let badges: Vec<&str> = tags
        .iter()
        .find(|tag| tag.0 == "badges")
        .and_then(|tag| tag.1.as_ref())
        .map(|s| s.split(',').collect())
        .unwrap_or_default();

    let broadcaster = badges.iter().any(|badge| badge.starts_with("broadcaster/"));
    let moderator = badges.iter().any(|badge| badge.starts_with("moderator/"));
    let vip = badges.iter().any(|badge| badge.starts_with("vip/"));
    let subscriber = badges.iter().any(|badge| badge.starts_with("subscriber/"));

    let words: Vec<&str> = message.split(' ').collect();
    let command_name = words[0];
    let args = words[1..].iter()
        .map(ToString::to_string)
        .map(rhai::Dynamic::from)
        .collect();

    let command = state.config.read()
        .await
        .commands
        .iter().find(|command| command.name == command_name || command.aliases.iter().any(|alias| alias == command_name))
        .map(Clone::clone);
    let command: crate::app::config::Command = match command {
        Some(c) => c,
        None => return Ok(())
    };

    if let Some(gcd) = &command.cooldowns.global {
        if let Some(last_used) = state.global_cooldowns.read().await.get(&command.name).cloned() {
            if last_used.elapsed() < gcd.to_std().unwrap() {
                return Ok(());
            }
        }

        state.global_cooldowns.write().await.insert(command.name.clone(), Instant::now());
    }

    if let Some(cooldown) = &command.cooldowns.user {
        let user_id = UserId::new(&initiator_id);
        if let Some(last_used) = state.user_cooldowns.read().await.get(&(user_id.clone(), command.name.clone())).cloned() {
            if last_used.elapsed() < cooldown.to_std().unwrap() {
                return Ok(());
            }
        }

        state.user_cooldowns.write().await.insert((user_id, command.name.clone()), Instant::now());
    }

    match &command.executor {
        CommandExecutor::Text(t) => { state.irc_queue.send(t.to_string()).ok(); }
        CommandExecutor::Rhai(t) => {
            let (speedgame_name, speedgame_category) = state.speedgame.read().await.clone();

            let fn_state = ExecutorState {
                state: Arc::clone(&state),
                args,
                initiator,
                initiator_id: UserId::new(initiator_id),
                broadcaster,
                moderator,
                vip,
                subscriber,
                speedgame_name,
                speedgame_category,
            };
            let args = vec![rhai::Dynamic::from(fn_state)];
            crate::app::run_rhai(Arc::clone(&state), t, args);
        }
    }

    Ok(())
}

use twitch_api2::pubsub::{
    Response as PubSubResponse,
    TopicData,
    channel_points::ChannelPointsChannelV1Reply,
};
use std::time::Instant;
use twitch_api2::pubsub::video_playback::VideoPlaybackReply;
use crate::app::config::Config;
use twitch_api2::twitch_oauth2::TwitchToken;

pub async fn handle_pubsub(state: Arc<State>, event: WsMessage) -> anyhow::Result<()> {
    let json = match event {
        WsMessage::Text(json) => json,
        _ => return Ok(()),
    };

    let response = twitch_api2::pubsub::Response::parse(&json)?;

    match response {
        PubSubResponse::Message {data} => match data {
            TopicData::ChannelPointsChannelV1 { reply, .. } => handle_channel_points_reply(state, reply).await,
            TopicData::VideoPlaybackById { reply, .. } => handle_stream_status(state, reply).await,
            _ => Ok(()),
        },
        _ => Ok(()),
    }
}

async fn rename_a_split(state: Arc<State>, redemption: &Redemption) {
    let input = match &redemption.user_input {
        Some(i) => i,
        None => return,
    };
    let (old, new) = match input.split_once("=>")
        .or_else(|| input.split_once("->"))
    {
        Some(x) => x,
        None => return,
    };

    if let Err(e) = reqwest::Client::new()
        .post(&state.user_config.livesplit.server)
        .body(format!("{}\n{}", old.trim(), new.trim()))
        .send()
        .await
    {
        eprintln!("could not do split rename: {:?}", e);
    }
}

async fn handle_channel_points_reply(state: Arc<State>, reply: Box<ChannelPointsChannelV1Reply>) -> anyhow::Result<()> {
    let redemption = match *reply {
        ChannelPointsChannelV1Reply::RewardRedeemed { redemption, .. } => redemption,
        _ => return Ok(()),
    };

    if redemption.reward.id.to_string() == "2205b59c-62d4-4dc1-bf27-2a96f3fcf2ca" {
        let state = Arc::clone(&state);
        let redemption = redemption.clone();
        tokio::task::spawn(async move {
            rename_a_split(state, &redemption).await;
        });
    }

    let action = match state.config.read().await.redemptions.iter().find(|re| re.twitch_id == redemption.reward.id).map(Clone::clone) {
        Some(a) => a,
        None => return Ok(()),
    };

    let (speedgame_name, speedgame_category) = state.speedgame.read().await.clone();

    let args = redemption.user_input
        .map(|input| input.split(' ').map(ToOwned::to_owned).map(rhai::Dynamic::from).collect())
        .unwrap_or_default();
    let fn_state = ExecutorState {
        state: Arc::clone(&state),
        args,
        initiator: redemption.user.login.to_string(),
        initiator_id: redemption.user.id,
        broadcaster: false,
        moderator: false,
        vip: false,
        subscriber: false,
        speedgame_name,
        speedgame_category,
    };
    let args = vec![rhai::Dynamic::from(fn_state)];
    crate::app::run_rhai(Arc::clone(&state), &action.rhai, args);
    Ok(())
}

async fn handle_stream_status(state: Arc<State>, reply: Box<VideoPlaybackReply>) -> anyhow::Result<()> {
    let live = match *reply {
        VideoPlaybackReply::StreamUp { .. } => true,
        VideoPlaybackReply::StreamDown { .. } => false,
        _ => return Ok(()),
    };

    state.store_live_status(live).await;

    Ok(())
}