mod event subscriptions added

migrations/0001_events/down.sql

@@ -0,0 +1 @@
+DROP TABLE subscriptions;

migrations/0001_events/up.sql

@@ -0,0 +1,6 @@
+CREATE TABLE subscriptions (
+    game INTEGER NOT NULL,
+    channel INTEGER NOT NULL,
+    guild INTEGER,
+    PRIMARY KEY (game, channel)
+);

src/commands.rs

@@ -28,6 +28,7 @@ pub mod prelude {
 pub mod basic;
 mod game;
 mod mods;
+pub mod subs;
 
 pub use game::{Game, ListGames};
 pub use mods::{ListMods, ModInfo, Popular};

src/commands/subs.rs

@@ -0,0 +1,141 @@
+use std::time::Duration;
+
+use futures::{Future, Stream};
+use log::{debug, warn};
+use modio::filter::{Operator, Order};
+use modio::games::GamesListOptions;
+use modio::EventListOptions;
+use modio::Modio;
+use serenity::prelude::*;
+use tokio::runtime::TaskExecutor;
+use tokio::timer::Interval;
+
+use crate::db::Subscriptions;
+use crate::util;
+
+command!(
+    Subscribe(self, ctx, msg, args) {
+        let mut ctx2 = ctx.clone();
+        let channel_id = msg.channel_id;
+        let guild_id = msg.guild_id.clone();
+
+        let mut opts = GamesListOptions::new();
+        match args.single::<u32>() {
+            Ok(id) => opts.id(Operator::Equals, id),
+            Err(_) => opts.fulltext(args.rest().to_string()),
+        };
+        let task = self
+            .modio
+            .games()
+            .list(&opts)
+            .and_then(|mut list| Ok(list.shift()))
+            .and_then(move |game| {
+                if let Some(g) = game {
+                    let ret = Subscriptions::add(&mut ctx2, g.id, channel_id, guild_id);
+                    match ret {
+                        Ok(_) => {
+                            let _ = channel_id.say(format!("Subscribed to '{}'", g.name));
+                        }
+                        Err(e) => eprintln!("{}", e),
+                    }
+                }
+                Ok(())
+            })
+            .map_err(|e| {
+                eprintln!("{}", e);
+            });
+
+        self.executor.spawn(task);
+    }
+
+    options(opts) {
+        opts.min_args = Some(1);
+    }
+);
+
+command!(
+    Unsubscribe(self, ctx, msg, args) {
+        let mut ctx2 = ctx.clone();
+        let channel_id = msg.channel_id;
+        let guild_id = msg.guild_id.clone();
+
+        let mut opts = GamesListOptions::new();
+        match args.single::<u32>() {
+            Ok(id) => opts.id(Operator::Equals, id),
+            Err(_) => opts.fulltext(args.rest().to_string()),
+        };
+        let task = self
+            .modio
+            .games()
+            .list(&opts)
+            .and_then(|mut list| Ok(list.shift()))
+            .and_then(move |game| {
+                if let Some(g) = game {
+                    let ret = Subscriptions::remove(&mut ctx2, g.id, channel_id, guild_id);
+                    match ret {
+                        Ok(_) => {
+                            let _ = channel_id.say(format!("Unsubscribed to '{}'", g.name));
+                        }
+                        Err(e) => eprintln!("{}", e),
+                    }
+                }
+                Ok(())
+            })
+            .map_err(|e| {
+                eprintln!("{}", e);
+            });
+
+        self.executor.spawn(task);
+    }
+);
+
+pub fn task(
+    client: &Client,
+    modio: Modio,
+    exec: TaskExecutor,
+) -> impl Future<Item = (), Error = ()> {
+    let data = client.data.clone();
+
+    Interval::new_interval(Duration::from_secs(3 * 60))
+        .for_each(move |_| {
+            let tstamp = util::current_timestamp() - 3 * 30;
+            let mut opts = EventListOptions::new();
+            opts.date_added(Operator::GreaterThan, tstamp);
+            opts.sort_by(EventListOptions::ID, Order::Asc);
+
+            let data = data.lock();
+            let Subscriptions(subs) = data
+                .get::<Subscriptions>()
+                .expect("failed to get subscriptions");
+
+            for (game, channels) in subs.clone() {
+                if channels.is_empty() {
+                    continue;
+                }
+                debug!("polling events for game={} channels: {:?}", game, channels);
+                let task = modio
+                    .game(game)
+                    .mods()
+                    .events(&opts)
+                    .collect()
+                    .and_then(move |events| {
+                        for e in events {
+                            for (channel, _) in &channels {
+                                let _ = channel.say(format!(
+                                    "[{}] {:?}",
+                                    tstamp,
+                                    e,
+                                ));
+                            }
+                        }
+                        Ok(())
+                    })
+                    .map_err(|_| ());
+
+                exec.spawn(task);
+            }
+
+            Ok(())
+        })
+        .map_err(|e| warn!("interval errored: {}", e))
+}

src/db.rs

@@ -1,11 +1,12 @@
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 
 use diesel::prelude::*;
 use diesel::r2d2::{ConnectionManager, Pool};
 use diesel::sqlite::SqliteConnection;
 use log::info;
 use serenity::client::Context;
 use serenity::model::channel::Message;
+use serenity::model::id::ChannelId;
 use serenity::model::id::GuildId;
 
 use crate::error::Error;
@@ -108,6 +109,82 @@ impl Settings {
     }
 }
 
+#[derive(Default)]
+pub struct Subscriptions(pub HashMap<u32, HashSet<(ChannelId, Option<GuildId>)>>);
+
+impl Subscriptions {
+    pub fn add(
+        ctx: &mut Context,
+        game_id: u32,
+        channel_id: ChannelId,
+        guild_id: Option<GuildId>,
+    ) -> Result<()> {
+        use crate::schema::subscriptions::dsl::*;
+
+        {
+            let mut data = ctx.data.lock();
+            data.get_mut::<Subscriptions>()
+                .expect("failed to get settings map")
+                .0
+                .entry(game_id)
+                .or_insert_with(Default::default)
+                .insert((channel_id, guild_id));
+        }
+
+        let data = ctx.data.lock();
+        let pool = data
+            .get::<PoolKey>()
+            .expect("failed to get connection pool");
+
+        pool.get()
+            .map_err(Error::from)
+            .and_then(|conn| {
+                diesel::replace_into(subscriptions)
+                    .values((
+                        game.eq(game_id as i32),
+                        channel.eq(channel_id.0 as i64),
+                        guild.eq(guild_id.map(|g| g.0 as i64)),
+                    ))
+                    .execute(&conn)
+                    .map_err(Error::from)
+            })
+            .map(|_| ())
+    }
+
+    pub fn remove(
+        ctx: &mut Context,
+        game_id: u32,
+        channel_id: ChannelId,
+        guild_id: Option<GuildId>,
+    ) -> Result<()> {
+        use crate::schema::subscriptions::dsl::*;
+
+        {
+            let mut data = ctx.data.lock();
+            data.get_mut::<Subscriptions>()
+                .expect("failed to get settings map")
+                .0
+                .entry(game_id)
+                .or_insert_with(Default::default)
+                .remove(&(channel_id, guild_id));
+        }
+
+        let data = ctx.data.lock();
+        let pool = data
+            .get::<PoolKey>()
+            .expect("failed to get connection pool");
+
+        pool.get()
+            .map_err(Error::from)
+            .and_then(|conn| {
+                let pred = game.eq(game_id as i32).and(channel.eq(channel_id.0 as i64));
+                let filter = subscriptions.filter(pred);
+                diesel::delete(filter).execute(&conn).map_err(Error::from)
+            })
+            .map(|_| ())
+    }
+}
+
 pub fn init_db(database_url: String) -> Result<DbPool> {
     let mgr = ConnectionManager::new(database_url);
     let pool = Pool::new(mgr)?;
@@ -150,6 +227,29 @@ pub fn load_settings(pool: &DbPool, guilds: &[GuildId]) -> Result<HashMap<GuildI
         })
 }
 
+pub fn load_subscriptions(pool: &DbPool) -> Result<Subscriptions> {
+    use crate::schema::subscriptions::dsl::*;
+    pool.get()
+        .map_err(Error::from)
+        .and_then(|conn| {
+            subscriptions
+                .load::<(i32, i64, Option<i64>)>(&conn)
+                .map_err(Error::from)
+        })
+        .and_then(|list| {
+            Ok(Subscriptions(list.into_iter().fold(
+                Default::default(),
+                |mut map, (game_id, channel_id, guild_id)| {
+                    let guild_id = guild_id.map(|id| GuildId(id as u64));
+                    map.entry(game_id as u32)
+                        .or_insert_with(Default::default)
+                        .insert((ChannelId(channel_id as u64), guild_id));
+                    map
+                },
+            )))
+        })
+}
+
 impl From<(GuildId, u32)> for ChangeSettings {
     fn from(c: (GuildId, u32)) -> Self {
         Self {

src/main.rs

@@ -29,9 +29,11 @@ mod macros;
 mod commands;
 mod db;
 mod error;
+#[rustfmt::skip]
 mod schema;
 mod util;
 
+use commands::subs;
 use commands::{Game, ListGames, ListMods, ModInfo, Popular};
 use util::*;
 
@@ -54,13 +56,17 @@ fn try_main() -> CliResult {
     dotenv().ok();
     env_logger::init();
 
-    let (mut client, modio, rt) = util::initialize()?;
+    let (mut client, modio, mut rt) = util::initialize()?;
 
     let games_cmd = ListGames::new(modio.clone(), rt.executor());
     let game_cmd = Game::new(modio.clone(), rt.executor());
     let mods_cmd = ListMods::new(modio.clone(), rt.executor());
     let mod_cmd = ModInfo::new(modio.clone(), rt.executor());
     let popular_cmd = Popular::new(modio.clone(), rt.executor());
+    let subscribe_cmd = subs::Subscribe::new(modio.clone(), rt.executor());
+    let unsubscribe_cmd = subs::Unsubscribe::new(modio.clone(), rt.executor());
+
+    rt.spawn(subs::task(&client, modio.clone(), rt.executor()));
 
     client.with_framework(
         StandardFramework::new()
@@ -81,6 +87,8 @@ fn try_main() -> CliResult {
                     .cmd("mods", mods_cmd)
                     .cmd("mod", mod_cmd)
                     .cmd("popular", popular_cmd)
+                    .cmd("subscribe", subscribe_cmd)
+                    .cmd("unsubscribe", unsubscribe_cmd)
             })
             .help(help_commands::with_embeds),
     );

src/schema.patch

@@ -8,3 +8,16 @@
          game -> Nullable<Integer>,
          prefix -> Nullable<Text>,
      }
+--- src/schema.rs.orig	2019-02-24 22:39:33.641530261 +0100
++++ src/schema.rs	2019-02-24 22:40:13.184879005 +0100
+@@ -9,8 +9,8 @@
+ table! {
+     subscriptions (game, channel) {
+         game -> Integer,
+-        channel -> Integer,
+-        guild -> Nullable<Integer>,
++        channel -> BigInt,
++        guild -> Nullable<BigInt>,
+     }
+ }
+

src/schema.rs

@@ -5,3 +5,16 @@ table! {
         prefix -> Nullable<Text>,
     }
 }
+
+table! {
+    subscriptions (game, channel) {
+        game -> Integer,
+        channel -> BigInt,
+        guild -> Nullable<BigInt>,
+    }
+}
+
+allow_tables_to_appear_in_same_query!(
+    settings,
+    subscriptions,
+);