nu_plugin_twitch 0.1.1

use std::time::Duration;

use nu_engine::command_prelude::*;
use nu_plugin::EvaluatedCall;
use nu_plugin::{EngineInterface, PluginCommand};
use nu_protocol::{CustomValue, LabeledError, PipelineData, Signature, SyntaxShape, Type};
use tokio::time::timeout;
use twitch_irc::login::StaticLoginCredentials;
use twitch_irc::message::ServerMessage;
use twitch_irc::{ClientConfig, SecureTCPTransport, TwitchIRCClient};

const POLL_TIMEOUT: Duration = Duration::from_micros(500);

use crate::TwitchPlugin;
use crate::values::TwitchNotice;

pub struct TwitchNotices;

impl PluginCommand for TwitchNotices {
    type Plugin = TwitchPlugin;

    fn name(&self) -> &str {
        "twitch notices"
    }

    fn description(&self) -> &str {
        "stream twitch notices of the given channel"
    }

    fn signature(&self) -> Signature {
        Signature::build(PluginCommand::name(self))
            .required("CHANNEL", SyntaxShape::String, "channel to listen to")
            .input_output_type(
                Type::Nothing,
                Type::List(Box::new(Type::Record(Box::new([
                    ("channel".to_string(), Type::String),
                    ("sender".to_string(), Type::String),
                    ("message".to_string(), Type::String),
                    ("event_id".to_string(), Type::String),
                    ("event_type".to_string(), Type::String),
                    ("event".to_string(), Type::record()),
                    ("timestamp".to_string(), Type::Date),
                ])))),
            )
    }

    fn run(
        &self,
        plugin: &TwitchPlugin,
        engine: &EngineInterface,
        call: &EvaluatedCall,
        _input: PipelineData,
    ) -> Result<PipelineData, LabeledError> {
        plugin.setup_tracing();

        let channel_login: String = call.req(0)?;
        let signals = engine.signals().clone();

        let (tx, mut rx) = tokio::sync::mpsc::channel(256);
        plugin.reactor.spawn(async move {
            let config = ClientConfig::default();
            let (mut incoming_messages, client) =
                TwitchIRCClient::<SecureTCPTransport, StaticLoginCredentials>::new(config);

            // Try to sanitize channel by put it in lowercase
            let channel_login = channel_login.clone().to_lowercase();
            tracing::info!(
                task = "receiver",
                phase = "joining",
                channel = channel_login.clone()
            );

            // Malformed channel login (most likely non-ascii char)
            if let Err(e) = client.join(channel_login.clone()) {
                tracing::error!(task = "receiver", phase = "joining", "{e:?}");
                return;
            }

            loop {
                // Timeout so that terminal interuptions do not have to wait for the next message
                let Ok(msg) = timeout(POLL_TIMEOUT, incoming_messages.recv()).await else {
                    // Interupted in the terminal
                    if signals.interrupted() {
                        break;
                    } else {
                        continue;
                    };
                };

                let Some(ServerMessage::UserNotice(notice)) = msg else {
                    // The twitch_irc library indicated that the connection is closed
                    if msg.is_none() {
                        break;
                    } else {
                        continue;
                    }
                };

                tracing::trace!(
                    task = "receiver",
                    type = "message",
                    channel = channel_login,
                    "{notice:?}"
                );
                tx.send(TwitchNotice::from(notice)).await.unwrap();
            }
        });

        let span = call.head;
        let iter = std::iter::from_fn(move || {
            rx.blocking_recv()
                .map(|msg| msg.to_base_value(span).unwrap())
        });

        Ok(iter.into_pipeline_data(call.head, engine.signals().clone()))
    }
}