Merge branch 'main' of gitea:jheuel/ohrwurm

Cargo.toml

@@ -7,20 +7,20 @@ license = "MIT"
 
 [dependencies]
 symphonia = { version = "0.5.2", features=["all", "opt-simd"] }
-songbird = { version = "0.4.0", features = ["driver", "gateway", "twilight", "rustls", "builtin-queue"] }
+songbird = { version = "0.5.0", features = ["driver", "gateway", "twilight", "rustls", "builtin-queue"] }
 tokio = { features = ["macros", "rt-multi-thread", "signal", "sync"], version = "1" }
 sqlx = { version = "0.8", features = [ "runtime-tokio", "sqlite", "chrono", "migrate"] }
 regex = { version = "1", features = ["unicode-case"] }
 futures = "0.3"
-reqwest = "0.11"
+reqwest = "0.12"
 tracing = "0.1"
 tracing-subscriber = "0.2"
-twilight-gateway = "0.15"
-twilight-http = "0.15"
-twilight-model = "0.15"
-twilight-standby = "0.15"
-twilight-cache-inmemory = "0.15"
-twilight-util = { version = "0.15", features=["builder"] }
+twilight-gateway = "0.16"
+twilight-http = "0.16"
+twilight-model = "0.16"
+twilight-standby = "0.16"
+twilight-cache-inmemory = "0.16"
+twilight-util = { version = "0.16", features=["builder"] }
 dotenv = "0.15.0"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"

src/commands/delete.rs

@@ -26,7 +26,7 @@ pub(crate) async fn delete(
         .http
         .channel_messages(msg.channel_id)
         .before(msg.id)
-        .limit(n)?
+        .limit(n)
         .await?
         .model()
         .await?;

src/commands/loop_queue.rs

@@ -1,4 +1,4 @@
-use crate::metadata::MetadataMap;
+use crate::metadata::Metadata;
 use crate::state::{State, StateRef};
 use async_trait::async_trait;
 use songbird::{Event, EventContext, EventHandler, TrackEvent};
@@ -99,9 +99,7 @@ impl EventHandler for TrackEndNotifier {
             let mut call = call_lock.lock().await;
 
             // get metadata from finished track
-            let old_typemap_lock = track_handle.typemap().read().await;
-            let old_metadata = old_typemap_lock.get::<MetadataMap>().unwrap();
-
+            let old_metadata = track_handle.data::<Metadata>();
             // enqueue track
             let handle = call.enqueue_with_preload(
                 old_metadata.src.clone().into(),
@@ -115,8 +113,8 @@ impl EventHandler for TrackEndNotifier {
             );
 
             // insert metadata into new track
-            let mut new_typemap = handle.typemap().write().await;
-            new_typemap.insert::<MetadataMap>(old_metadata.clone());
+            let mut _new_metadata = handle.data::<Metadata>();
+            _new_metadata = old_metadata.clone();
         }
         None
     }

src/commands/play.rs

@@ -1,5 +1,5 @@
 use crate::commands::join::join_channel;
-use crate::metadata::{Metadata, MetadataMap};
+use crate::metadata::Metadata;
 use crate::state::State;
 use crate::{colors, db};
 
@@ -266,7 +266,7 @@ pub(crate) async fn play(
                 .http
                 .interaction(interaction.application_id)
                 .update_response(&interaction.token)
-                .embeds(Some(&embeds))?
+                .embeds(Some(&embeds))
                 .await?;
             Ok(())
         }
@@ -348,7 +348,7 @@ pub(crate) async fn play_inner(
             .http
             .interaction(interaction.application_id)
             .update_response(&interaction.token)
-            .embeds(Some(&embeds))?
+            .embeds(Some(&embeds))
             .await
             .context("Could not send playlist loading message")?;
     }
@@ -401,8 +401,8 @@ pub(crate) async fn play_inner(
                                 }
                             }),
                         );
-                        let mut x = handle.typemap().write().await;
-                        x.insert::<MetadataMap>(Metadata {
+                        let mut _x = handle.data::<Metadata>();
+                        _x = Arc::new(Metadata {
                             title: metadata.title,
                             duration: metadata.duration,
                             url,
@@ -427,7 +427,7 @@ pub(crate) async fn play_inner(
                         .http
                         .interaction(interaction.application_id)
                         .update_response(&interaction.token)
-                        .embeds(Some(&embeds))?
+                        .embeds(Some(&embeds))
                         .await?;
                     return Ok(());
                 }
@@ -440,7 +440,7 @@ pub(crate) async fn play_inner(
         .http
         .interaction(interaction.application_id)
         .update_response(&interaction.token)
-        .embeds(Some(&embeds))?
+        .embeds(Some(&embeds))
         .await
         .context("Could not send final play message")?;
 

src/commands/queue.rs

@@ -1,6 +1,6 @@
 use songbird::tracks::TrackHandle;
 use twilight_model::channel::message::component::{ActionRow, Button, ButtonStyle};
-use twilight_model::channel::message::{Component, Embed, MessageFlags, ReactionType};
+use twilight_model::channel::message::{Component, Embed, EmojiReactionType, MessageFlags};
 use twilight_model::gateway::payload::incoming::InteractionCreate;
 use twilight_model::http::interaction::InteractionResponse;
 use twilight_model::http::interaction::InteractionResponseType;
@@ -8,7 +8,8 @@ use twilight_util::builder::embed::EmbedBuilder;
 use twilight_util::builder::InteractionResponseDataBuilder;
 
 use crate::colors;
-use crate::{metadata::MetadataMap, state::State};
+use crate::metadata::Metadata;
+use crate::state::State;
 use std::error::Error;
 
 pub(crate) const TRACKS_PER_PAGE: usize = 5;
@@ -38,10 +39,7 @@ pub(crate) async fn build_queue_embeds(queue: &[TrackHandle], page: usize) -> Ve
         .skip(TRACKS_PER_PAGE * page)
         .take(TRACKS_PER_PAGE)
     {
-        let map = track.typemap().read().await;
-        let metadata = map
-            .get::<MetadataMap>()
-            .expect("Could not get metadata map");
+        let metadata = track.data::<Metadata>();
         message.push_str(
             format!(
                 "* [{}]({})",
@@ -76,31 +74,34 @@ pub(crate) fn build_action_row(page: usize, n_pages: usize) -> Vec<Component> {
                 custom_id: Some(format!("page:{}", page as i32 - 1)),
                 style: ButtonStyle::Primary,
                 label: Some("Previous page".to_string()),
-                emoji: Some(ReactionType::Unicode {
+                emoji: Some(EmojiReactionType::Unicode {
                     name: "⬅️".to_string(),
                 }),
                 url: None,
                 disabled: page == 0,
+                sku_id: None,
             }),
             Component::Button(Button {
                 custom_id: Some(format!("page:{}", page)),
                 style: ButtonStyle::Primary,
                 label: Some("Refresh".to_string()),
-                emoji: Some(ReactionType::Unicode {
+                emoji: Some(EmojiReactionType::Unicode {
                     name: "🔄".to_string(),
                 }),
                 url: None,
                 disabled: false,
+                sku_id: None,
             }),
             Component::Button(Button {
                 custom_id: Some(format!("page:{}", page + 1)),
                 style: ButtonStyle::Primary,
                 label: Some("Next page".to_string()),
-                emoji: Some(ReactionType::Unicode {
+                emoji: Some(EmojiReactionType::Unicode {
                     name: "➡️".to_string(),
                 }),
                 url: None,
                 disabled: page >= n_pages - 1,
+                sku_id: None,
             }),
         ],
     })]
@@ -153,8 +154,8 @@ pub(crate) async fn queue(
         .http
         .interaction(interaction.application_id)
         .update_response(&interaction.token)
-        .embeds(Some(&embeds))?
-        .components(Some(&action_row))?
+        .embeds(Some(&embeds))
+        .components(Some(&action_row))
         .await?;
 
     Ok(())

src/handler.rs

@@ -14,6 +14,7 @@ use twilight_model::gateway::payload::incoming::InteractionCreate;
 use twilight_model::http::interaction::{InteractionResponse, InteractionResponseType};
 use twilight_util::builder::InteractionResponseDataBuilder;
 
+#[derive(Debug, Clone)]
 pub(crate) struct Handler {
     state: State,
 }

src/main.rs

@@ -12,18 +12,12 @@ mod utils;
 
 use crate::commands::get_chat_commands;
 use dotenv::dotenv;
-use futures::StreamExt;
-use signal::signal_handler;
 use songbird::{shards::TwilightMap, Songbird};
 use state::StateRef;
 use std::{env, error::Error, str::FromStr, sync::Arc, time::Duration};
-use tokio::select;
 use tracing::{debug, info};
 use twilight_cache_inmemory::InMemoryCache;
-use twilight_gateway::{
-    stream::{self, ShardEventStream},
-    Intents, Shard,
-};
+use twilight_gateway::{Event, EventTypeFlags, Intents, Shard, StreamExt as _};
 use twilight_http::Client as HttpClient;
 use twilight_model::id::Id;
 use twilight_standby::Standby;
@@ -38,7 +32,7 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
 
     info!("Starting up...");
 
-    let (mut shards, state) = {
+    let (shards, state) = {
         let db = env::var("DATABASE_URL").map_err(|_| "DATABASE_URL is not set")?;
         let options = SqliteConnectOptions::from_str(&db)
             .expect("could not create options")
@@ -71,7 +65,7 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
             | Intents::MESSAGE_CONTENT;
         let config = twilight_gateway::Config::new(token.clone(), intents);
         let shards: Vec<Shard> =
-            stream::create_recommended(&http, config, |_, builder| builder.build())
+            twilight_gateway::create_recommended(&http, config, |_, builder| builder.build())
                 .await?
                 .collect();
         let senders = TwilightMap::new(
@@ -105,45 +99,55 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
     info!("Ready to receive events");
 
     let handler = Handler::new(Arc::clone(&state));
-    let mut stop_rx = signal_handler();
-    let mut stream = ShardEventStream::new(shards.iter_mut());
-    loop {
-        select! {
-            biased;
-            _ = stop_rx.changed() => {
-                for guild in state.cache.iter().guilds() {
-                    if let Some(user) = state.cache.current_user() {
-                        if state.cache.voice_state(user.id, guild.id()).is_some() {
-                            debug!("Leaving guild {:?}", guild.id());
-                            state.songbird.leave(guild.id()).await?;
-                        }
-                    }
-                }
-                // need to grab next event to properly leave voice channels
-                stream.next().await;
-                break;
-            },
-            next = stream.next() => {
-                let event = match next {
-                    Some((_, Ok(event))) => event,
-                    Some((_, Err(source))) => {
-                        tracing::warn!(?source, "error receiving event");
-                        if source.is_fatal() {
-                            break;
-                        }
-                        continue;
-                    }
-                    None => break,
-                };
-                debug!("Event: {:?}", &event);
+    // let mut stop_rx = signal_handler();
+    let mut set = tokio::task::JoinSet::new();
 
-                state.cache.update(&event);
-                state.standby.process(&event);
-                state.songbird.process(&event).await;
-
-                handler.act(event).await?;
-            }
-        }
+    for shard in shards {
+        set.spawn(tokio::spawn(runner(shard, handler.clone(), state.clone())));
     }
+    set.join_next().await;
+
+    Ok(())
+}
+
+async fn runner(mut shard: Shard, handler: Handler, state: Arc<StateRef>) {
+    while let Some(item) = shard.next_event(EventTypeFlags::all()).await {
+        let event = match item {
+            Ok(event) => event,
+            Err(source) => {
+                tracing::warn!(?source, "error receiving event");
+
+                continue;
+            }
+        };
+
+        tokio::spawn({
+            let state = state.clone();
+            let handler = handler.clone();
+            async move {
+                handle_event(event, handler, state)
+                    .await
+                    .unwrap_or_else(|source| {
+                        tracing::warn!(?source, "error handling event");
+                    });
+            }
+        });
+    }
+}
+
+async fn handle_event(
+    event: Event,
+    handler: Handler,
+    state: Arc<StateRef>,
+) -> Result<(), Box<dyn Error>> {
+    state.standby.process(&event);
+    state.songbird.process(&event).await;
+    debug!("Event: {:?}", &event);
+
+    state.cache.update(&event);
+    state.standby.process(&event);
+    state.songbird.process(&event).await;
+
+    handler.act(event).await?;
     Ok(())
 }

src/metadata.rs

@@ -1,4 +1,4 @@
-use songbird::{input::YoutubeDl, typemap::TypeMapKey};
+use songbird::input::YoutubeDl;
 use std::time::Duration;
 
 #[derive(Clone)]
@@ -6,10 +6,5 @@ pub(crate) struct Metadata {
     pub(crate) title: Option<String>,
     pub(crate) duration: Option<Duration>,
     pub(crate) url: String,
-    pub(crate) src: YoutubeDl,
-}
-
-pub(crate) struct MetadataMap;
-impl TypeMapKey for MetadataMap {
-    type Value = Metadata;
+    pub(crate) src: YoutubeDl<'static>,
 }

src/signal.rs

@@ -1,21 +1,21 @@
-use tokio::{
-    select,
-    signal::unix::{signal, SignalKind},
-    sync::watch,
-};
+// use tokio::{
+//     select,
+//     signal::unix::{signal, SignalKind},
+//     sync::watch,
+// };
 
-pub(crate) fn signal_handler() -> watch::Receiver<()> {
-    let (stop_tx, stop_rx) = watch::channel(());
-    tokio::spawn(async move {
-        let mut sigterm = signal(SignalKind::terminate()).unwrap();
-        let mut sigint = signal(SignalKind::interrupt()).unwrap();
-        loop {
-            select! {
-                _ = sigterm.recv() => println!("Receive SIGTERM"),
-                _ = sigint.recv() => println!("Receive SIGTERM"),
-            };
-            stop_tx.send(()).unwrap();
-        }
-    });
-    stop_rx
-}
+// pub(crate) fn signal_handler() -> watch::Receiver<()> {
+//     let (stop_tx, stop_rx) = watch::channel(());
+//     tokio::spawn(async move {
+//         let mut sigterm = signal(SignalKind::terminate()).unwrap();
+//         let mut sigint = signal(SignalKind::interrupt()).unwrap();
+//         loop {
+//             select! {
+//                 _ = sigterm.recv() => println!("Receive SIGTERM"),
+//                 _ = sigint.recv() => println!("Receive SIGTERM"),
+//             };
+//             stop_tx.send(()).unwrap();
+//         }
+//     });
+//     stop_rx
+// }