wrap lock function

apps/master/src/commands/mod.rs

@@ -1,2 +1,42 @@
+use nats_libs::stream::{JetstreamClient, StreamClient};
+use serenity::all::{CommandInteraction, Context};
+use types::error::CorroError;
+
+use crate::common::Services;
+
 pub mod ping;
 pub mod play;
+
+pub async fn queue_lock<'a, F, Fut>(
+    services: &'a Services,
+    ctx: &'a Context,
+    interaction: &'a CommandInteraction,
+    func: F,
+) -> Result<(), CorroError>
+where
+    F: Fn(&'a Services, &'a Context, &'a CommandInteraction) -> Fut,
+    Fut: Future<Output = Result<(), CorroError>> + 'a,
+{
+    let guild_id = interaction.guild_id.unwrap();
+
+    match services.jetstream_client.lock(guild_id.into()).await {
+        Ok(sequence) => {
+            let result = func(services, ctx, interaction).await;
+
+            match services.jetstream_stream.unlock(sequence).await {
+                Ok(unlocked) => {
+                    if !unlocked {
+                        return Err(CorroError {
+                            error_type: types::error::CorroErrorType::NatsError,
+                            message: "Not unlocked".to_string(),
+                        });
+                    }
+                }
+                Err(why) => return Err(why),
+            }
+
+            result
+        }
+        Err(why) => Err(why),
+    }
+}

apps/master/src/main.rs

@@ -4,7 +4,6 @@ mod common;
 use std::env;
 
 use async_nats::jetstream;
-use nats_libs::stream::{JetstreamClient, StreamClient};
 use serenity::{
     all::{Context, EventHandler, GatewayIntents},
     async_trait,
@@ -16,7 +15,7 @@ use serenity::{
 use serenity_libs::functions::CustomInteraction;
 use types::error::{CorroError, CorroErrorType};
 
-use crate::common::Services;
+use crate::{commands::queue_lock, common::Services};
 
 struct Handler {
     services: Services,
@@ -29,22 +28,7 @@ impl EventHandler for Handler {
             println!("Received command interaction: {command:?}");
 
             if let Err(why) = match command.data.name.as_str() {
-                "play" => {
-                    let guild_id = command.guild_id.unwrap();
-
-                    if let Ok(sequence) = self
-                        .services
-                        .jetstream_client
-                        .lock(guild_id.into())
-                        .await
-                    {
-                        let result = commands::play::run(&self.services, &ctx, &command).await;
-                        let _ = self.services.jetstream_stream.unlock(sequence).await;
-                        result
-                    } else {
-                        Err(CorroError { error_type: CorroErrorType::NatsError, message: format!("Cannot lock {guild_id}") })
-                    }
-                }
+                "play" => queue_lock(&self.services, &ctx, &command, commands::play::run).await,
                 "ping" => commands::ping::run(&ctx, &command).await,
                 _ => Err(CorroError {
                     error_type: CorroErrorType::CommandError,

apps/worker/Cargo.toml

@@ -23,4 +23,4 @@ symphonia = { version = "0.5.5" }
 tokio = { version = "1.50.0", features = ["macros", "rt-multi-thread"] }
 types = { path = "../../libs/types" }
 which = { version = "8.0.2" }
-yt-dlp = { version = "2.6.0" }
+yt-dlp = { version = "2.7.0" }