folders structure

apps/master/Cargo.toml

@@ -0,0 +1,18 @@
+[package]
+name = "master"
+version = "0.1.0"
+edition = "2024"
+
+[dependencies]
+async-nats = { version = "0.46.0" }
+postcard = { version = "1.1.3", features = ["use-std"] }
+serenity = { version = "0.12.5", default-features = false, features = [
+    "client",
+    "gateway",
+    "rustls_backend",
+    "model",
+    "collector",
+] }
+tokio = { version = "1.49.0", features = ["macros", "rt-multi-thread"] }
+types = { path = "../../libs/types" }
+uuid = { version = "1.21.0" }

apps/master/src/commands/mod.rs

@@ -0,0 +1,3 @@
+pub mod modal;
+pub mod ping;
+pub mod testnats;

apps/master/src/commands/modal.rs

@@ -0,0 +1,28 @@
+use serenity::{builder::*, model::prelude::*, prelude::*, utils::CreateQuickModal};
+
+pub async fn run(ctx: &Context, interaction: &CommandInteraction) -> Result<(), serenity::Error> {
+    let modal = CreateQuickModal::new("About you")
+        .timeout(std::time::Duration::from_secs(600))
+        .short_field("First name")
+        .short_field("Last name")
+        .paragraph_field("Hobbies and interests");
+    let response = interaction.quick_modal(ctx, modal).await?.unwrap();
+
+    let inputs = response.inputs;
+    let (first_name, last_name, hobbies) = (&inputs[0], &inputs[1], &inputs[2]);
+
+    response
+        .interaction
+        .create_response(
+            ctx,
+            CreateInteractionResponse::Message(CreateInteractionResponseMessage::new().content(
+                format!("**Name**: {first_name} {last_name}\n\nHobbies and interests: {hobbies}"),
+            )),
+        )
+        .await?;
+    Ok(())
+}
+
+pub fn register() -> CreateCommand {
+    CreateCommand::new("modal").description("Asks some details about you")
+}

apps/master/src/commands/ping.rs

@@ -0,0 +1,9 @@
+use serenity::{builder::CreateCommand, model::application::ResolvedOption};
+
+pub fn run(_options: &[ResolvedOption]) -> String {
+    "Hey, I'm alive!".to_string()
+}
+
+pub fn register() -> CreateCommand {
+    CreateCommand::new("ping").description("A ping command")
+}

apps/master/src/commands/testnats.rs

@@ -0,0 +1,91 @@
+use std::str::from_utf8;
+
+use postcard::to_stdvec;
+use serenity::{
+    all::{
+        CommandInteraction,
+        Context,
+        CreateCommandOption,
+        CreateInteractionResponse,
+        CreateInteractionResponseMessage,
+        EditInteractionResponse,
+    },
+    builder::CreateCommand,
+    model::application::{CommandOptionType, ResolvedOption, ResolvedValue},
+};
+use types::{
+    jobs::{InnerStruct, Job, JobKind},
+    misc::new_uuid_v4,
+};
+
+pub async fn run(
+    ctx: &Context,
+    interaction: &CommandInteraction,
+    nats_client: &async_nats::Client,
+) -> Result<(), serenity::Error> {
+    let options = interaction.data.options();
+
+    if let Some(ResolvedOption {
+        value: ResolvedValue::String(value),
+        ..
+    }) = options.first()
+    {
+        interaction
+            .create_response(
+                ctx,
+                CreateInteractionResponse::Message(
+                    CreateInteractionResponseMessage::new()
+                        .content(format!("Searching: {value}...")),
+                ),
+            )
+            .await?;
+
+        let job = Job {
+            uuid: new_uuid_v4(),
+            kind: JobKind::Download,
+            inner: InnerStruct {
+                str: value.to_string(),
+            },
+        };
+
+        println!("job {:?}", job);
+
+        let response = match nats_client
+            .request("corro-dj.download", to_stdvec(&job).unwrap().into())
+            .await
+        {
+            Ok(resp) => resp,
+            Err(_why) => return Err(serenity::Error::Other("send error")),
+        };
+
+        println!("response: {:?}", from_utf8(&response.payload).unwrap());
+
+        interaction
+            .edit_response(
+                ctx,
+                EditInteractionResponse::new()
+                    .content(format!("path: {}", from_utf8(&response.payload).unwrap())),
+            )
+            .await?;
+    } else {
+        interaction
+            .create_response(
+                ctx,
+                CreateInteractionResponse::Message(
+                    CreateInteractionResponseMessage::new()
+                        .content("Please provide a valid string"),
+                ),
+            )
+            .await?;
+    }
+    Ok(())
+}
+
+pub fn register() -> CreateCommand {
+    CreateCommand::new("testnats")
+        .description("test nats")
+        .add_option(
+            CreateCommandOption::new(CommandOptionType::String, "str", "random string")
+                .required(false),
+        )
+}

apps/master/src/main.rs

@@ -0,0 +1,92 @@
+mod commands;
+
+use std::env;
+
+use serenity::{
+    Client,
+    all::{Context, EventHandler, GatewayIntents},
+    async_trait,
+    builder::{CreateInteractionResponse, CreateInteractionResponseMessage},
+    model::{
+        application::{Command, Interaction},
+        gateway::Ready,
+    },
+};
+
+struct Handler {
+    nats_client: async_nats::Client,
+}
+
+#[async_trait]
+impl EventHandler for Handler {
+    async fn interaction_create(&self, ctx: Context, interaction: Interaction) {
+        if let Interaction::Command(command) = interaction {
+            println!("Received command interaction: {command:#?}");
+
+            let content = match command.data.name.as_str() {
+                "testnats" => {
+                    commands::testnats::run(&ctx, &command, &self.nats_client)
+                        .await
+                        .unwrap();
+                    None
+                }
+                "ping" => Some(commands::ping::run(&command.data.options())),
+                "modal" => {
+                    commands::modal::run(&ctx, &command).await.unwrap();
+                    None
+                }
+                _ => Some("not implemented :(".to_string()),
+            };
+
+            if let Some(content) = content {
+                let data = CreateInteractionResponseMessage::new().content(content);
+                let builder = CreateInteractionResponse::Message(data);
+                if let Err(why) = command.create_response(&ctx.http, builder).await {
+                    println!("Cannot respond to slash command: {why}");
+                }
+            }
+        }
+    }
+
+    async fn ready(&self, ctx: Context, ready: Ready) {
+        println!("{} is connected!", ready.user.name);
+
+        let commands = Command::set_global_commands(
+            &ctx.http,
+            vec![
+                commands::ping::register(),
+                commands::modal::register(),
+                commands::testnats::register(),
+            ],
+        )
+        .await;
+
+        println!("I now have the following slash commands: {commands:#?}");
+    }
+}
+
+#[tokio::main]
+async fn main() {
+    // Configure the client with your Discord bot token in the environment.
+    let token = env::var("DISCORD_TOKEN").expect("Expected a token in the environment");
+
+    let nats_client = async_nats::connect("nats://localhost:4222")
+        .await
+        .expect("Error creating nats client");
+
+    let handler = Handler { nats_client };
+
+    // Build our client.
+    let mut discord_client = Client::builder(token, GatewayIntents::empty())
+        .event_handler(handler)
+        .await
+        .expect("Error creating discord client");
+
+    // Finally, start a single shard, and start listening to events.
+    //
+    // Shards will automatically attempt to reconnect, and will perform exponential backoff until
+    // it reconnects.
+    if let Err(why) = discord_client.start().await {
+        println!("Client error: {why:?}");
+    }
+}

apps/worker/Cargo.toml

@@ -0,0 +1,14 @@
+[package]
+name = "worker"
+version = "0.1.0"
+edition = "2024"
+
+[dependencies]
+async-nats = { version = "0.46.0" }
+futures = { version = "0.3.32" }
+futures-executor = { version = "0.3.32" }
+postcard = { version = "1.1.3", features = ["use-std"] }
+tokio = { version = "1.49.0", features = ["macros", "rt-multi-thread"] }
+types = { path = "../../libs/types" }
+which = { version = "8.0.0" }
+yt-dlp = { version = "2.1.0" }

apps/worker/src/main.rs

@@ -0,0 +1,64 @@
+use futures::StreamExt;
+use postcard::from_bytes;
+use types::jobs::Job;
+use which::which;
+use yt_dlp::{Downloader, client::Libraries};
+
+#[tokio::main]
+async fn main() {
+    let nats_client = async_nats::connect("nats://localhost:4222")
+        .await
+        .expect("Error creating nats client");
+
+    let mut subscriber = nats_client
+        .queue_subscribe("corro-dj.*", "download".to_string())
+        .await
+        .unwrap();
+
+    let libraries = Libraries::new(which("yt-dlp").unwrap(), which("ffmpeg").unwrap());
+
+    let downloader = Downloader::builder(libraries, "output")
+        .build()
+        .await
+        .unwrap();
+
+    // Receive and process messages
+    while let Some(message) = subscriber.next().await {
+        println!("Received message {:?}", message);
+        let result: Job = from_bytes(&message.payload).unwrap();
+        println!("{:?}", result);
+
+        let video = match downloader.fetch_video_infos(result.inner.str).await {
+            Ok(video) => video,
+            Err(why) => {
+                println!("{:?}", why);
+                continue;
+            }
+        };
+
+        let audio_path = match downloader
+            .download_audio_stream_with_quality(
+                &video,
+                format!("{}.opus", video.id),
+                yt_dlp::model::AudioQuality::Best,
+                yt_dlp::model::AudioCodecPreference::Opus,
+            )
+            .await
+        {
+            Ok(path) => path,
+            Err(why) => {
+                println!("{:?}", why);
+                continue;
+            }
+        };
+
+        println!("reply: {:?}", audio_path);
+
+        if let Some(reply) = message.reply {
+            nats_client
+                .publish(reply, format!("{:?}", audio_path).into())
+                .await
+                .unwrap();
+        }
+    }
+}