rework to use songbird

4 files changed, 209 insertions, 170 deletions

diff --git a/src/commands/meme/create.rs b/src/commands/meme/create.rs
index 97c5276..1c12f2a 100644
--- a/src/commands/meme/create.rs
+++ b/src/commands/meme/create.rs
@@ -1,50 +1,41 @@
-use std::{
-    io::Read,
-    process::{
-        Command,
-        Stdio,
-    },
-};
+use std::process::Stdio;
 
+use anyhow::anyhow;
 use diesel::result::Error as DieselError;
+use lazy_static::lazy_static;
 use log::{
     debug,
     error,
     warn,
 };
 use serenity::{
+    all::ReactionType,
     framework::standard::{
         macros::command,
         Args,
+        CommandError,
+        CommandResult,
         Delimiter,
     },
+    futures::TryFutureExt,
     model::channel::Message,
     prelude::*,
 };
-use url::Url;
-
-use anyhow::anyhow;
-use lazy_static::lazy_static;
-use serenity::{
-    all::ReactionType,
-    framework::standard::{
-        CommandError,
-        CommandResult,
-    },
-    futures::TryFutureExt,
+use tap::Pipe;
+use tokio::{
+    io::AsyncReadExt,
+    process::Command,
 };
+use url::Url;
 
 use crate::{
-    audio::{
-        parse_times,
-        ytdl_url,
-    },
     db::{
         connection,
         Audio,
         Image,
         NewMeme,
     },
+    parse_times,
     util,
     FFMPEG_COMMAND,
 };
@@ -77,12 +68,12 @@ pub async fn addmeme(ctx: &Context, msg: &Message, args: Args) -> CommandResult
             .await;
     }
 
-    let image_id = image
-        .map(|att| {
-            let data = att.download()?;
-            Image::create(&mut conn, &att.filename, data, msg.author.id.get())
-        })
-        .transpose()?;
+    let mut image_id = None;
+
+    if let Some(att) = image {
+        let data = att.download().await?;
+        image_id = Some(Image::create(&mut conn, &att.filename, data, msg.author.id.get())?);
+    };
 
     let save_result = NewMeme {
         title,
@@ -96,7 +87,9 @@ pub async fn addmeme(ctx: &Context, msg: &Message, args: Args) -> CommandResult
 
     use diesel::result::DatabaseErrorKind;
     match save_result {
-        Ok(_) => msg.react(&ctx, "👌"),
+        Ok(_) => {
+            msg.react(&ctx, ReactionType::Unicode("👌".to_string())).await?;
+        },
         Err(e) => {
             if let Some(DieselError::DatabaseError(DatabaseErrorKind::UniqueViolation, _)) =
                 e.downcast_ref::<DieselError>()
@@ -111,6 +104,8 @@ pub async fn addmeme(ctx: &Context, msg: &Message, args: Args) -> CommandResult
             return Err(e.into());
         },
     }
+
+    Ok(())
 }
 
 #[command]
@@ -131,7 +126,7 @@ pub async fn addaudiomeme(ctx: &Context, msg: &Message, args: Args) -> CommandRe
     let opts = elems[1..].join(" ");
     let (start, end) = parse_times(opts);
 
-    let youtube_url = ytdl_url(audio_link.as_str())?;
+    let youtube_url = util::ytdl_url(audio_link.as_str()).await?;
 
     let duration_opts = if let Some(e) = end {
         vec![
@@ -178,18 +173,17 @@ pub async fn addaudiomeme(ctx: &Context, msg: &Message, args: Args) -> CommandRe
 
     let mut conn = connection()?;
 
-    let image = msg
-        .attachments
-        .first()
-        .ok_or(anyhow!("no attachment"))
-        .and_then(|att| {
-            let data = att.download()?;
-            Image::create(&mut conn, &att.filename, data, msg.author.id.get())
-        })
-        .ok();
+    let image_att = msg.attachments.first().ok_or(anyhow!("no attachment"));
+
+    let mut image_id = None;
+
+    if let Ok(att) = image_att {
+        let data = att.download().await?;
+        image_id = Image::create(&mut conn, &att.filename, data, msg.author.id.get())?.pipe(Some);
+    }
 
     let mut audio_data = Vec::new();
-    let bytes = audio_reader.read_to_end(&mut audio_data)?;
+    let bytes = audio_reader.read_to_end(&mut audio_data).await?;
 
     if bytes == 0 {
         debug!("read 0 bytes from audio reader");
@@ -203,7 +197,7 @@ pub async fn addaudiomeme(ctx: &Context, msg: &Message, args: Args) -> CommandRe
     let save_result = NewMeme {
         title,
         content: text,
-        image_id: image,
+        image_id,
         audio_id: Some(audio_id),
         metadata_id: 0,
     }
@@ -212,7 +206,9 @@ pub async fn addaudiomeme(ctx: &Context, msg: &Message, args: Args) -> CommandRe
 
     use diesel::result::DatabaseErrorKind;
     match save_result {
-        Ok(_) => msg.react(&ctx, ReactionType::Unicode("👌".to_owned())),
+        Ok(_) => {
+            msg.react(&ctx, ReactionType::Unicode("👌".to_owned())).await?;
+        },
         Err(e) => {
             if let Some(DieselError::DatabaseError(DatabaseErrorKind::UniqueViolation, _)) =
                 e.downcast_ref::<DieselError>()
@@ -224,7 +220,9 @@ pub async fn addaudiomeme(ctx: &Context, msg: &Message, args: Args) -> CommandRe
                     .await;
             }
 
-            return Err(e);
+            return Err(e.into());
         },
     }
+
+    Ok(())
 }
diff --git a/src/commands/meme/history.rs b/src/commands/meme/history.rs
index 5e200b1..ed50e27 100644
--- a/src/commands/meme/history.rs
+++ b/src/commands/meme/history.rs
@@ -1,7 +1,11 @@
+use anyhow::anyhow;
 use diesel::{
     result::Error as DieselError,
     NotFound,
+    PgConnection,
 };
+use itertools::Itertools;
+use lazy_static::lazy_static;
 use log::{
     debug,
     error,
@@ -11,25 +15,23 @@ use serenity::{
     framework::standard::{
         macros::command,
         Args,
+        CommandError,
+        CommandResult,
+    },
+    futures::{
+        StreamExt,
+        TryFutureExt,
+        TryStreamExt,
     },
     model::channel::Message,
     prelude::*,
 };
+use tap::Pipe;
 use timeago::{
     Formatter,
     TimeUnit,
 };
 
-use anyhow::anyhow;
-use lazy_static::lazy_static;
-use serenity::{
-    framework::standard::{
-        CommandError,
-        CommandResult,
-    },
-    futures::TryFutureExt,
-};
-
 use crate::{
     db::{
         self,
@@ -39,7 +41,6 @@ use crate::{
         Metadata,
     },
     util,
-    Result,
     CONFIG,
 };
 
@@ -53,10 +54,10 @@ lazy_static! {
     };
 }
 
-static CLEAN_DATE_FORMAT: &'static str = "%b %-e %Y";
+static CLEAN_DATE_FORMAT: &str = "%b %-e %Y";
 
 #[command]
-#[aliases("what")]
+#[aliases("what", "hwaet", "hwæt")]
 pub async fn wat(ctx: &Context, msg: &Message, _: Args) -> CommandResult {
     let mut conn = connection()?;
 
@@ -80,7 +81,7 @@ pub async fn wat(ctx: &Context, msg: &Message, _: Args) -> CommandResult {
     match meme {
         Ok(ref meme) => {
             let metadata = Metadata::find(&mut conn, meme.metadata_id)?;
-            let author = CONFIG.discord.guild().member(&ctx, metadata.created_by as u64)?;
+            let author = CONFIG.discord.guild().member(&ctx, metadata.created_by as u64).await?;
 
             util::send(
                 ctx,
@@ -98,22 +99,22 @@ pub async fn wat(ctx: &Context, msg: &Message, _: Args) -> CommandResult {
         Err(e) => {
             if let Some(NotFound) = e.downcast_ref::<DieselError>() {
                 info!("last meme not found in database");
-                return util::send(ctx, msg.channel_id, "heuueueeeeh?", msg.tts).await;
+                return util::send(ctx, msg.channel_id, "heuueueeeeh?", msg.tts)
+                    .await
+                    .map_err(CommandError::from);
             }
 
             util::send(ctx, msg.channel_id, "do i look like i know what a jpeg is", msg.tts)
                 .await?;
-            return Err(e);
+            return Err(e.into());
         },
     };
 
-    meme.map(|_| {})
+    meme.map(|_| {}).map_err(CommandError::from)
 }
 
 #[command]
 pub async fn history(ctx: &Context, msg: &Message, mut args: Args) -> CommandResult {
-    use itertools::Itertools;
-
     let mut conn = connection()?;
 
     let n = args.single_quoted::<usize>().unwrap_or(CONFIG.default_hist);
@@ -135,66 +136,76 @@ pub async fn history(ctx: &Context, msg: &Message, mut args: Args) -> CommandRes
     }
 
     info!("reporting meme history (len {})", n);
-    let resp = records
-        .into_iter()
-        .enumerate()
-        .rev()
-        .map(|(i, rec)| {
-            let dt = chrono::DateTime::from_utc(rec.time, chrono::Utc {});
-            let ago = TIME_FORMATTER.convert((chrono::Utc::now() - dt).to_std().unwrap());
 
-            let rand = if rec.random {
-                "R, "
-            } else {
-                ""
-            };
-            Meme::find(&mut conn, rec.meme_id)
-                .and_then(|meme| {
-                    Metadata::find(&mut conn, meme.metadata_id).map(|metadata| (metadata, meme))
-                })
-                .map(|(metadata, meme)| {
-                    let author_name = CONFIG
-                        .discord
-                        .guild()
-                        .member(&ctx, metadata.created_by as u64)
-                        .map(|m| m.display_name().into_owned())
-                        .unwrap_or("???".to_owned());
-                    let invoker_name = CONFIG
-                        .discord
-                        .guild()
-                        .member(&ctx, rec.user_id as u64)
-                        .map(|m| m.display_name().into_owned())
-                        .unwrap_or("???".to_owned());
-                    format!(
-                        "{}. [{}{}] \"{}\" by {} ({}). invoked by {}.",
-                        i + 1,
-                        rand,
-                        ago,
-                        meme.title,
-                        author_name,
-                        metadata.created.date().format(CLEAN_DATE_FORMAT),
-                        invoker_name
-                    )
-                })
-                .unwrap_or_else(|e| {
-                    if let Some(variant) = e.downcast_ref::<DieselError>() {
-                        if *variant != NotFound {
-                            error!("error encountered loading meme history: {}", e);
-                        }
-                    }
+    let resp = serenity::futures::stream::iter(records.into_iter().enumerate().rev())
+        .then(|(i, rec)| ir_info(ctx, i, rec, &mut conn))
+        .try_collect::<Vec<String>>()
+        .await?;
 
-                    let invoker_name = CONFIG
-                        .discord
-                        .guild()
-                        .member(&ctx, rec.user_id as u64)
-                        .map(|m| m.display_name().into_owned())
-                        .unwrap_or("???".to_owned());
-                    format!("{}. [{}{}] not found. invoked by {}.", i + 1, rand, ago, invoker_name)
-                })
-        })
-        .join("\n");
+    let resp = resp.join("\n");
 
-    util::send(ctx, msg.channel_id, &resp, false).await
+    util::send(ctx, msg.channel_id, &resp, false).await.map_err(CommandError::from)
+}
+
+async fn ir_info(
+    ctx: &Context,
+    i: usize,
+    rec: InvocationRecord,
+    conn: &mut PgConnection,
+) -> Result<String, CommandError> {
+    let dt = chrono::DateTime::from_utc(rec.time, chrono::Utc {});
+    let ago = TIME_FORMATTER.convert((chrono::Utc::now() - dt).to_std().unwrap());
+
+    let rand = if rec.random {
+        "R, "
+    } else {
+        ""
+    };
+
+    let meme = Meme::find(conn, rec.meme_id)
+        .and_then(|meme| Metadata::find(conn, meme.metadata_id).map(|metadata| (metadata, meme)));
+
+    let invoker_name = CONFIG
+        .discord
+        .guild()
+        .member(&ctx, rec.user_id as u64)
+        .await
+        .map(|m| m.display_name().to_owned())
+        .unwrap_or("???".to_owned());
+
+    let result = match meme {
+        Ok((metadata, meme)) => {
+            let author_name = CONFIG
+                .discord
+                .guild()
+                .member(&ctx, metadata.created_by as u64)
+                .await
+                .map(|m| m.display_name().to_owned())
+                .unwrap_or("???".to_owned());
+
+            format!(
+                "{}. [{}{}] \"{}\" by {} ({}). invoked by {}.",
+                i + 1,
+                rand,
+                ago,
+                meme.title,
+                author_name,
+                metadata.created.date().format(CLEAN_DATE_FORMAT),
+                invoker_name
+            )
+        },
+        Err(e) => {
+            if let Some(variant) = e.downcast_ref::<DieselError>() {
+                if *variant != NotFound {
+                    error!("error encountered loading meme history: {}", e);
+                }
+            }
+
+            format!("{}. [{}{}] not found. invoked by {}.", i + 1, rand, ago, invoker_name)
+        },
+    };
+
+    Ok(result)
 }
 
 #[command]
@@ -211,11 +222,12 @@ pub async fn stats(ctx: &Context, msg: &Message, _: Args) -> CommandResult {
 
     debug!("reporting stats");
 
-    let rand_user: User = UserId::new(stats.most_random_meme_user).to_user(&ctx)?;
-    let direct_user: User = UserId::new(stats.most_directly_named_meme_user).to_user(&ctx)?;
+    let rand_user: User = UserId::new(stats.most_random_meme_user).to_user(&ctx).await?;
+    let direct_user: User = UserId::new(stats.most_directly_named_meme_user).to_user(&ctx).await?;
 
-    let rand_user = rand_user.nick_in(&ctx, CONFIG.discord.guild()).unwrap_or(rand_user.name);
-    let direct_user = direct_user.nick_in(&ctx, CONFIG.discord.guild()).unwrap_or(direct_user.name);
+    let rand_user = rand_user.nick_in(&ctx, CONFIG.discord.guild()).await.unwrap_or(rand_user.name);
+    let direct_user =
+        direct_user.nick_in(&ctx, CONFIG.discord.guild()).await.unwrap_or(direct_user.name);
 
     let s = format!(
         r#"
@@ -270,15 +282,14 @@ and *{}* was the most-memed overall ({})"#,
 
 #[command]
 pub async fn memers(ctx: &Context, msg: &Message, _args: Args) -> CommandResult {
-    use db;
-    use itertools::Itertools;
     use serenity::model::id::UserId;
 
     let s = db::memers()?
         .into_iter()
-        .map(|info| {
-            let user = UserId::new(info.user_id).to_user(&ctx)?;
-            let username = user.nick_in(&ctx, CONFIG.discord.guild()).unwrap_or(user.name);
+        .pipe(serenity::futures::stream::iter)
+        .then(|info| async move {
+            let user = UserId::new(info.user_id).to_user(&ctx).await?;
+            let username = user.nick_in(&ctx, CONFIG.discord.guild()).await.unwrap_or(user.name);
 
             let res = format!(
                 "**{}**: {} total, {} random, {} specific. favorite meme: *{}* ({})",
@@ -290,9 +301,10 @@ pub async fn memers(ctx: &Context, msg: &Message, _args: Args) -> CommandResult
                 info.most_used_meme_count,
             );
 
-            Ok(res)
+            Result::<_, CommandError>::Ok(res)
         })
-        .collect::<Result<Vec<_>>>()?
+        .try_collect::<Vec<String>>()
+        .await?
         .into_iter()
         .join("\n");
 
@@ -301,11 +313,9 @@ pub async fn memers(ctx: &Context, msg: &Message, _args: Args) -> CommandResult
 
 #[command]
 pub async fn query(ctx: &Context, msg: &Message, mut args: Args) -> CommandResult {
-    use std::borrow::Borrow;
-
-    use itertools::Itertools;
     use regex::Regex;
     use serenity::model::id::UserId;
+    use std::borrow::Borrow;
 
     use crate::{
         db,
@@ -318,11 +328,10 @@ pub async fn query(ctx: &Context, msg: &Message, mut args: Args) -> CommandResul
         static ref AGE_REGEX: Regex = Regex::new(r"(?i)(?:age|order)=(.*)").unwrap();
     }
 
-    let guild = msg.channel_id.to_channel(&ctx)?.guild().ok_or(anyhow!("couldn't find guild"))?;
-
-    let guild = guild.read().guild(&ctx).ok_or(anyhow!("couldn't find guild"))?;
+    let guild =
+        msg.channel_id.to_channel(&ctx).await?.guild().ok_or(anyhow!("couldn't find guild"))?;
 
-    let guild = guild.read();
+    let guild = guild.guild(&ctx).ok_or(anyhow!("couldn't find guild"))?;
 
     let creator: Option<u64> = {
         let creator = args.quoted().current().map(|s| CREATOR_REGEX.is_match(s)).unwrap_or(false);
@@ -354,11 +363,13 @@ pub async fn query(ctx: &Context, msg: &Message, mut args: Args) -> CommandResul
         }
     };
 
-    let result = db::query_meme(args.rest(), creator, order)?
-        .into_iter()
-        .map(|(meme, metadata)| {
-            let user = UserId::new(metadata.created_by as u64).to_user(&ctx)?;
-            let username = user.nick_in(&ctx, CONFIG.discord.guild()).unwrap_or(user.name);
+    let iter = db::query_meme(args.rest(), creator, order)?.into_iter();
+
+    let result = iter
+        .pipe(serenity::futures::stream::iter)
+        .then(|(meme, metadata)| async move {
+            let user = UserId::new(metadata.created_by as u64).to_user(&ctx).await?;
+            let username = user.nick_in(&ctx, CONFIG.discord.guild()).await.unwrap_or(user.name);
 
             Ok(format!(
                 "*{}* by **{}** ({}). text length: **{}**, image: **{}**, audio: **{}**",
@@ -368,9 +379,12 @@ pub async fn query(ctx: &Context, msg: &Message, mut args: Args) -> CommandResul
                 meme.content.map_or(0, |s| s.len()),
                 meme.image_id.map_or("NO", |_s| "YES"),
                 meme.audio_id.map_or("NO", |_s| "YES"),
-            ))
+            )) as Result<String, CommandError>
         })
-        .collect::<Result<Vec<_>>>()?
+        .try_collect::<Vec<String>>()
+        .await;
+
+    let result = result?
         .into_iter()
         .scan(0, |state, line| {
             *state = *state + line.len() + 1;
diff --git a/src/commands/meme/invoke.rs b/src/commands/meme/invoke.rs
index 03c6251..13996da 100644
--- a/src/commands/meme/invoke.rs
+++ b/src/commands/meme/invoke.rs
@@ -30,37 +30,37 @@ use crate::{
 #[command]
 #[aliases("mem")]
 pub async fn meme(ctx: &Context, msg: &Message, args: Args) -> CommandResult {
-    _meme(ctx, msg, args, AudioPlayback::Optional)
+    _meme(ctx, msg, args, AudioPlayback::Optional).await
 }
 
 #[command]
 pub async fn omen(ctx: &Context, msg: &Message, _args: Args) -> CommandResult {
     let args = Args::new("", &[]);
-    _meme(ctx, msg, args, AudioPlayback::Optional)
+    _meme(ctx, msg, args, AudioPlayback::Optional).await
 }
 
 #[command]
 pub async fn silentomen(ctx: &Context, msg: &Message, _args: Args) -> CommandResult {
     let args = Args::new("", &[]);
-    _meme(ctx, msg, args, AudioPlayback::Prohibited)
+    _meme(ctx, msg, args, AudioPlayback::Prohibited).await
 }
 
 #[command]
 pub async fn audioomen(ctx: &Context, msg: &Message, _args: Args) -> CommandResult {
     let args = Args::new("", &[]);
-    _meme(ctx, msg, args, AudioPlayback::Required)
+    _meme(ctx, msg, args, AudioPlayback::Required).await
 }
 
 #[command]
 #[aliases("audiomeme", "audiomem")]
 pub async fn audio_meme(ctx: &Context, msg: &Message, args: Args) -> CommandResult {
-    _meme(ctx, msg, args, AudioPlayback::Required)
+    _meme(ctx, msg, args, AudioPlayback::Required).await
 }
 
 #[command]
 #[aliases("silentmeme", "silentmem")]
 pub async fn silent_meme(ctx: &Context, msg: &Message, args: Args) -> CommandResult {
-    _meme(ctx, msg, args, AudioPlayback::Prohibited)
+    _meme(ctx, msg, args, AudioPlayback::Prohibited).await
 }
 
 #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
@@ -152,7 +152,7 @@ async fn rand_meme(
 #[command]
 #[aliases("rarememe", "raremem")]
 pub async fn rare_meme(ctx: &Context, msg: &Message, _args: Args) -> CommandResult {
-    let should_audio = ctx.users_listening()?;
+    let should_audio = util::users_listening(ctx).await?;
 
     let mut conn = connection()?;
 
@@ -160,7 +160,7 @@ pub async fn rare_meme(ctx: &Context, msg: &Message, _args: Args) -> CommandResu
     match meme {
         Ok(meme) => {
             InvocationRecord::create(&mut conn, msg.author.id.get(), msg.id.get(), meme.id, true)?;
-            send_meme(ctx, &meme, &mut conn, msg)
+            send_meme(ctx, &meme, &mut conn, msg).await
         },
         Err(e) => {
             match e.downcast_ref::<DieselError>() {
@@ -177,7 +177,7 @@ pub async fn rare_meme(ctx: &Context, msg: &Message, _args: Args) -> CommandResu
                 .map_err(CommandError::from)
                 .await?;
 
-            Err(e)
+            Err(e.into())
         },
     }
 }
diff --git a/src/commands/meme/mod.rs b/src/commands/meme/mod.rs
index 31d9b78..24fc50d 100644
--- a/src/commands/meme/mod.rs
+++ b/src/commands/meme/mod.rs
@@ -3,6 +3,7 @@ use log::debug;
 use rand::random;
 use serenity::{
     all::ReactionType,
+    async_trait,
     builder::{
         CreateAttachment,
         CreateMessage,
@@ -14,13 +15,21 @@ use serenity::{
     model::channel::Message,
     prelude::*,
 };
+use songbird::input::{
+    core::io::MediaSource,
+    AudioStream,
+    AudioStreamError,
+    Compose,
+    Input,
+};
 
 use crate::{
-    audio::{
-        PlayArgs,
-        PlayQueue,
+    commands::songbird,
+    db::{
+        Audio,
+        Meme,
     },
-    db::Meme,
+    CONFIG,
 };
 
 pub use self::{
@@ -94,27 +103,45 @@ async fn send_meme(
         },
     };
 
-    // note: slight edge-case race condition here: there could have been something queued since we
-    //  checked whether anything was playing. not a significant negative impact and unlikely, so i'm
-    //  not worrying about it
     if let Some(audio) = audio {
         let audio = audio?;
 
-        {
-            let queue_lock = ctx.data.write().await.get::<PlayQueue>().cloned().unwrap();
-            let mut play_queue = queue_lock.write().unwrap();
+        let (_sb, call) = songbird(ctx, msg).await?;
+        let mut call = call.lock().await;
 
-            play_queue.meme_queue.push_back(PlayArgs {
-                initiator:      msg.author.name.clone(),
-                data:           ::either::Right(audio.data.clone()),
-                sender_channel: msg.channel_id,
-                start:          None,
-                end:            None,
-            });
+        if call.current_channel().is_none() {
+            call.join(CONFIG.discord.voice_channel()).await?;
         }
 
+        call.enqueue_input(Input::Lazy(Box::new(audio))).await;
+
         msg.react(ctx, ReactionType::Unicode("📣".to_owned())).await?;
     }
 
     Ok(())
 }
+
+#[async_trait]
+impl Compose for Audio {
+    fn create(&mut self) -> Result<AudioStream<Box<dyn MediaSource>>, AudioStreamError> {
+        let ms = std::io::Cursor::new(self.data.clone());
+        let ms: Box<dyn MediaSource> = Box::new(ms);
+
+        Ok(AudioStream {
+            input: ms,
+            hint:  None,
+        })
+    }
+
+    #[inline]
+    async fn create_async(
+        &mut self,
+    ) -> Result<AudioStream<Box<dyn MediaSource>>, AudioStreamError> {
+        self.create()
+    }
+
+    #[inline]
+    fn should_create_async(&self) -> bool {
+        false
+    }
+}