Stream notification fixes, no more spam?

src/NadekoBot/Modules/Searches/Commands/StreamNotificationCommands.cs

@@ -12,6 +12,11 @@ using NadekoBot.Services.Database.Models;
 using System.Net.Http;
 using Discord.WebSocket;
 using NadekoBot.Attributes;
+using Microsoft.EntityFrameworkCore;
+using Newtonsoft.Json;
+using NLog;
+using NadekoBot.Services.Database;
+using NadekoBot.Extensions;
 
 namespace NadekoBot.Modules.Searches
 {
@@ -19,87 +24,111 @@ namespace NadekoBot.Modules.Searches
     {
         public class StreamStatus
         {
-            public StreamStatus(string link, bool isLive, string views)
-            {
-                Link = link;
-                IsLive = isLive;
-                Views = views;
-            }
-
             public bool IsLive { get; set; }
-            public string Link { get; set; }
+            public string ApiLink { get; set; }
             public string Views { get; set; }
         }
+
+        public class HitboxResponse {
+            public bool Success { get; set; } = true;
+            [JsonProperty("media_is_live")]
+            public string MediaIsLive { get; set; }
+            public bool IsLive  => MediaIsLive == "1";
+            [JsonProperty("media_views")]
+            public string Views { get; set; }
+        }
+
+        public class TwitchResponse
+        {
+            public string Error { get; set; } = null;
+            public bool IsLive => Stream != null;
+            public StreamInfo Stream { get; set; }
+
+            public class StreamInfo
+            {
+                public int Viewers { get; set; }
+            }
+        }
+
+        public class BeamResponse
+        {
+            public string Error { get; set; } = null;
+
+            [JsonProperty("online")]
+            public bool IsLive { get; set; }
+            public int ViewersCurrent { get; set; }
+        }
+
+        public class StreamNotFoundException : Exception
+        {
+            public StreamNotFoundException(string message) : base("Stream '" + message + "' not found.")
+            {
+            }
+        }
+
         [Group]
         public class StreamNotificationCommands
         {
             private Timer checkTimer { get; }
             private ConcurrentDictionary<string, StreamStatus> oldCachedStatuses = new ConcurrentDictionary<string, StreamStatus>();
             private ConcurrentDictionary<string, StreamStatus> cachedStatuses = new ConcurrentDictionary<string, StreamStatus>();
+            private Logger _log { get; }
+
             private bool FirstPass { get; set; } = true;
 
             public StreamNotificationCommands()
             {
+
+                _log = NLog.LogManager.GetCurrentClassLogger();
                 checkTimer = new Timer(async (state) =>
                 {
                     oldCachedStatuses = new ConcurrentDictionary<string, StreamStatus>(cachedStatuses);
-                    cachedStatuses = new ConcurrentDictionary<string, StreamStatus>();
+                    cachedStatuses.Clear();
-                    try
+                    IEnumerable<FollowedStream> streams;
+                    using (var uow = DbHandler.UnitOfWork())
                     {
-                        IEnumerable<FollowedStream> streams;
+                        streams = uow.GuildConfigs.GetAllFollowedStreams();
-                        using (var uow = DbHandler.UnitOfWork())
-                        {
-                            streams = uow.GuildConfigs.GetAllFollowedStreams();
-                        }
-                        foreach (var stream in streams)
-                        {
-                            StreamStatus data;
-                            try
-                            {
-                                data = await GetStreamStatus(stream).ConfigureAwait(false);
-                                if (data == null)
-                                    return;
-                            }
-                            catch
-                            {
-                                continue;
-                            }
-
-                            StreamStatus oldData;
-                            oldCachedStatuses.TryGetValue(data.Link, out oldData);
-
-                            if (oldData == null || data.IsLive != oldData.IsLive)
-                            {
-                                if (FirstPass)
-                                    continue;
-                                var server = NadekoBot.Client.GetGuild(stream.GuildId);
-                                var channel = server?.GetTextChannel(stream.ChannelId);
-                                if (channel == null)
-                                    continue;
-                                var msg = $"`{stream.Username}`'s stream is now " +
-                                          $"**{(data.IsLive ? "ONLINE" : "OFFLINE")}** with " +
-                                          $"**{data.Views}** viewers.";
-                                if (data.IsLive)
-                                    if (stream.Type == FollowedStream.FollowedStreamType.Hitbox)
-                                        msg += $"\n`Here is the Link:`【 http://www.hitbox.tv/{stream.Username}/ 】";
-                                    else if (stream.Type == FollowedStream.FollowedStreamType.Twitch)
-                                        msg += $"\n`Here is the Link:`【 http://www.twitch.tv/{stream.Username}/ 】";
-                                    else if (stream.Type == FollowedStream.FollowedStreamType.Beam)
-                                        msg += $"\n`Here is the Link:`【 http://www.beam.pro/{stream.Username}/ 】";
-                                try { await channel.SendMessageAsync(msg).ConfigureAwait(false); } catch { }
-                            }
-                        }
-                        FirstPass = false;
                     }
-                    catch { }
+
-                }, null, TimeSpan.Zero, TimeSpan.FromSeconds(60));
+                    await Task.WhenAll(streams.Select(async fs =>
+                                    {
+                                        try
+                                        {
+                                            var newStatus = await GetStreamStatus(fs).ConfigureAwait(false);
+                                            if (FirstPass)
+                                            {
+                                                return;
+                                            }
+
+                                            StreamStatus oldStatus;
+                                            if (oldCachedStatuses.TryGetValue(newStatus.ApiLink, out oldStatus) &&
+                                                oldStatus.IsLive != newStatus.IsLive)
+                                            {
+                                                var msg = $"`{fs.Username}`'s stream is now " +
+                                                    $"**{(newStatus.IsLive ? "ONLINE" : "OFFLINE")}** with " +
+                                                    $"**{newStatus.Views}** viewers.";
+
+                                                var server = NadekoBot.Client.GetGuild(fs.GuildId);
+                                                var channel = server?.GetTextChannel(fs.ChannelId);
+                                                if (channel == null)
+                                                    return;
+
+                                                try { await channel.SendMessageAsync(msg).ConfigureAwait(false); } catch { }
+                                            }
+                                        }
+                                        catch (Exception ex)
+                                        {
+                                            _log.Error(ex);
+                                        }
+                                    }));
+
+                    FirstPass = false;
+                }, null, TimeSpan.Zero, TimeSpan.FromSeconds(5));
             }
 
             private async Task<StreamStatus> GetStreamStatus(FollowedStream stream, bool checkCache = true)
             {
-                bool isLive;
                 string response;
-                JObject data;
                 StreamStatus result;
                 switch (stream.Type)
                 {
@@ -111,10 +140,16 @@ namespace NadekoBot.Modules.Searches
                         {
                             response = await http.GetStringAsync(hitboxUrl).ConfigureAwait(false);
                         }
-                        data = JObject.Parse(response);
+                        var hbData = JsonConvert.DeserializeObject<HitboxResponse>(response);
-                        isLive = data["media_is_live"].ToString() == "1";
+                        if (!hbData.Success)
-                        result = new StreamStatus(hitboxUrl, isLive, data["media_views"].ToString());
+                            throw new StreamNotFoundException($"{stream.Username} [{stream.Type}]");
-                        cachedStatuses.TryAdd(hitboxUrl, result);
+                        result = new StreamStatus()
+                        {
+                            IsLive = hbData.IsLive,
+                            ApiLink = hitboxUrl,
+                            Views = hbData.Views
+                        };
+                        cachedStatuses.AddOrUpdate(hitboxUrl, result, (key, old) => result);
                         return result;
                     case FollowedStream.FollowedStreamType.Twitch:
                         var twitchUrl = $"https://api.twitch.tv/kraken/streams/{Uri.EscapeUriString(stream.Username)}?client_id=67w6z9i09xv2uoojdm9l0wsyph4hxo6";
@@ -122,12 +157,22 @@ namespace NadekoBot.Modules.Searches
                             return result;
                         using (var http = new HttpClient())
                         {
+                            _log.Info(twitchUrl);
                             response = await http.GetStringAsync(twitchUrl).ConfigureAwait(false);
                         }
-                        data = JObject.Parse(response);
+                        var twData = JsonConvert.DeserializeObject<TwitchResponse>(response);
-                        isLive = !string.IsNullOrWhiteSpace(data["stream"].ToString());
+                        if (twData.Error != null)
-                        result = new StreamStatus(twitchUrl, isLive, isLive ? data["stream"]["viewers"].ToString() : "0");
+                        {
-                        cachedStatuses.TryAdd(twitchUrl, result);
+                            _log.Error(twData.Error);
+                            throw new StreamNotFoundException($"{stream.Username} [{stream.Type}]");
+                        }
+                        result = new StreamStatus()
+                        {
+                            IsLive = twData.IsLive,
+                            ApiLink = twitchUrl,
+                            Views = twData.Stream?.Viewers.ToString() ?? "0"
+                        };
+                        cachedStatuses.AddOrUpdate(twitchUrl, result, (key, old) => result);
                         return result;
                     case FollowedStream.FollowedStreamType.Beam:
                         var beamUrl = $"https://beam.pro/api/v1/channels/{stream.Username}";
@@ -137,10 +182,17 @@ namespace NadekoBot.Modules.Searches
                         {
                             response = await http.GetStringAsync(beamUrl).ConfigureAwait(false);
                         }
-                        data = JObject.Parse(response);
+
-                        isLive = data["online"].ToObject<bool>() == true;
+                        var bmData = JsonConvert.DeserializeObject<BeamResponse>(response);
-                        result = new StreamStatus(beamUrl, isLive, data["viewersCurrent"].ToString());
+                        if (bmData.Error != null)
-                        cachedStatuses.TryAdd(beamUrl, result);
+                            throw new StreamNotFoundException($"{stream.Username} [{stream.Type}]");
+                        result = new StreamStatus()
+                        {
+                            IsLive = bmData.IsLive,
+                            ApiLink = beamUrl,
+                            Views = bmData.ViewersCurrent.ToString()
+                        };
+                        cachedStatuses.AddOrUpdate(beamUrl, result, (key, old) => result);
                         return result;
                     default:
                         break;
@@ -178,7 +230,10 @@ namespace NadekoBot.Modules.Searches
                 IEnumerable<FollowedStream> streams;
                 using (var uow = DbHandler.UnitOfWork())
                 {
-                    streams = uow.GuildConfigs.For(channel.Guild.Id).FollowedStreams;
+                    streams = uow.GuildConfigs
+                                 .For(channel.Guild.Id, 
+                                      set => set.Include(gc => gc.FollowedStreams))
+                                 .FollowedStreams;
                 }
 
                 if (!streams.Any())
@@ -198,30 +253,33 @@ namespace NadekoBot.Modules.Searches
             [NadekoCommand, Usage, Description, Aliases]
             [RequireContext(ContextType.Guild)]
             [RequirePermission(GuildPermission.ManageMessages)]
-            public async Task RemoveStream(IUserMessage msg, [Remainder] string username)
+            public async Task RemoveStream(IUserMessage msg, FollowedStream.FollowedStreamType type, [Remainder] string username)
             {
                 var channel = (ITextChannel)msg.Channel;
 
                 username = username.ToLowerInvariant().Trim();
 
-                FollowedStream toRemove;
+                var fs = new FollowedStream()
+                {
+                    ChannelId = channel.Id,
+                    Username = username,
+                    Type = type
+                };
+
+                bool removed;
                 using (var uow = DbHandler.UnitOfWork())
                 {
-                    var config = uow.GuildConfigs.For(channel.Guild.Id);
+                    var config = uow.GuildConfigs.For(channel.Guild.Id, set => set.Include(gc => gc.FollowedStreams));
-                    var streams = config.FollowedStreams;
+                    removed = config.FollowedStreams.Remove(fs);
-                    toRemove = streams.Where(fs => fs.ChannelId == channel.Id && fs.Username.ToLowerInvariant() == username).FirstOrDefault();
+                    if (removed)
-                    if (toRemove != null)
+                        await uow.CompleteAsync().ConfigureAwait(false);
-                    {
-                        config.FollowedStreams = new HashSet<FollowedStream>(streams.Except(new[] { toRemove }));
-                        await uow.CompleteAsync();
-                    }
                 }
-                if (toRemove == null)
+                if (!removed)
                 {
                     await channel.SendMessageAsync(":anger: No such stream.").ConfigureAwait(false);
                     return;
                 }
-                await channel.SendMessageAsync($":ok: Removed `{toRemove.Username}`'s stream ({toRemove.Type}) from notifications.").ConfigureAwait(false);
+                await channel.SendMessageAsync($":ok: Removed `{username}`'s stream ({type}) from notifications.").ConfigureAwait(false);
             }
 
             [NadekoCommand, Usage, Description, Aliases]
@@ -238,7 +296,7 @@ namespace NadekoBot.Modules.Searches
                     var streamStatus = (await GetStreamStatus(new FollowedStream
                     {
                         Username = stream,
-                        Type = platform
+                        Type = platform,
                     }));
                     if (streamStatus.IsLive)
                     {
@@ -265,16 +323,7 @@ namespace NadekoBot.Modules.Searches
                     Username = username,
                     Type = type,
                 };
-                bool exists;
+
-                using (var uow = DbHandler.UnitOfWork())
-                {
-                    exists = uow.GuildConfigs.For(channel.Guild.Id).FollowedStreams.Where(fs => fs.ChannelId == channel.Id && fs.Username.ToLowerInvariant().Trim()  == username).Any();
-                }
-                if (exists)
-                {
-                    await channel.SendMessageAsync($":anger: I am already following `{username}` ({type}) stream on this channel.").ConfigureAwait(false);
-                    return;
-                }
                 StreamStatus data;
                 try
                 {
@@ -285,22 +334,35 @@ namespace NadekoBot.Modules.Searches
                     await channel.SendMessageAsync(":anger: Stream probably doesn't exist.").ConfigureAwait(false);
                     return;
                 }
-                var msg = $"Stream is currently **{(data.IsLive ? "ONLINE" : "OFFLINE")}** with **{data.Views}** viewers";
+
-                if (data.IsLive)
-                    if (type == FollowedStream.FollowedStreamType.Hitbox)
-                        msg += $"\n`Here is the Link:`【 http://www.hitbox.tv/{stream.Username}/ 】";
-                    else if (type == FollowedStream.FollowedStreamType.Twitch)
-                        msg += $"\n`Here is the Link:`【 http://www.twitch.tv/{stream.Username}/ 】";
-                    else if (type == FollowedStream.FollowedStreamType.Beam)
-                        msg += $"\n`Here is the Link:`【 https://beam.pro/{stream.Username}/ 】";
                 using (var uow = DbHandler.UnitOfWork())
                 {
-                    uow.GuildConfigs.For(channel.Guild.Id).FollowedStreams.Add(stream);
+                    uow.GuildConfigs.For(channel.Guild.Id, set => set.Include(gc => gc.FollowedStreams))
-                    await uow.CompleteAsync();
+                                    .FollowedStreams
+                                    .Add(stream);
+                    await uow.CompleteAsync().ConfigureAwait(false);
                 }
+                var msg = $"Stream is currently **{(data.IsLive ? "ONLINE" : "OFFLINE")}** with **{data.Views}** viewers";
+                if (data.IsLive)
+                    msg += stream.GetLink();
                 msg = $":ok: I will notify this channel when status changes.\n{msg}";
                 await channel.SendMessageAsync(msg).ConfigureAwait(false);
             }
         }
     }
+
+    public static class FollowedStreamExtensions
+    {
+        public static string GetLink(this FollowedStream fs)
+        {
+            //todo C#7
+            if (fs.Type == FollowedStream.FollowedStreamType.Hitbox)
+                return $"\n`Here is the Link:`【 http://www.hitbox.tv/{fs.Username}/ 】";
+            else if (fs.Type == FollowedStream.FollowedStreamType.Twitch)
+                return $"\n`Here is the Link:`【 http://www.twitch.tv/{fs.Username}/ 】";
+            else if (fs.Type == FollowedStream.FollowedStreamType.Beam)
+                return $"\n`Here is the Link:`【 https://beam.pro/{fs.Username}/ 】";
+            return "???";
+        }
+    }
 }

src/NadekoBot/Resources/CommandStrings.Designer.cs

@@ -5280,7 +5280,7 @@ namespace NadekoBot.Resources {
         }
         
         /// <summary>
-        ///    Looks up a localized string similar to Removes notifications of a certain streamer on this channel..
+        ///    Looks up a localized string similar to Removes notifications of a certain streamer from a certain platform on this channel..
         /// </summary>
         public static string removestream_desc {
             get {
@@ -5289,7 +5289,7 @@ namespace NadekoBot.Resources {
         }
         
         /// <summary>
-        ///    Looks up a localized string similar to `{0}rms SomeGuy`.
+        ///    Looks up a localized string similar to `{0}rms Twitch SomeGuy` or `{0}rms Beam SomeOtherGuy`.
         /// </summary>
         public static string removestream_usage {
             get {

src/NadekoBot/Resources/CommandStrings.resx

@@ -1759,10 +1759,10 @@
     <value>removestream rms</value>
   </data>
   <data name="removestream_desc" xml:space="preserve">
-    <value>Removes notifications of a certain streamer on this channel.</value>
+    <value>Removes notifications of a certain streamer from a certain platform on this channel.</value>
   </data>
   <data name="removestream_usage" xml:space="preserve">
-    <value>`{0}rms SomeGuy`</value>
+    <value>`{0}rms Twitch SomeGuy` or `{0}rms Beam SomeOtherGuy`</value>
   </data>
   <data name="liststreams_cmd" xml:space="preserve">
     <value>liststreams ls</value>

src/NadekoBot/Services/Database/Models/FollowedStream.cs

@@ -11,5 +11,17 @@
         {
             Twitch, Hitbox, Beam
         }
+
+        public override int GetHashCode() => 
+            ChannelId.GetHashCode() ^ Username.GetHashCode();
+
+        public override bool Equals(object obj)
+        {
+            var fs = obj as FollowedStream;
+            if (fs == null)
+                return false;
+
+            return fs.ChannelId == ChannelId && fs.Username.ToLowerInvariant().Trim() == Username.ToLowerInvariant().Trim();
+        }
     }
 }

src/NadekoBot/Services/Database/Repositories/IGuildConfigRepository.cs

@@ -1,11 +1,14 @@
-using NadekoBot.Services.Database.Models;
+using Microsoft.EntityFrameworkCore;
+using NadekoBot.Services.Database.Models;
+using System;
 using System.Collections.Generic;
+using System.Linq;
 
 namespace NadekoBot.Services.Database.Repositories
 {
     public interface IGuildConfigRepository : IRepository<GuildConfig>
     {
-        GuildConfig For(ulong guildId);
+        GuildConfig For(ulong guildId, Func<DbSet<GuildConfig>, IQueryable<GuildConfig>> includes = null);
         GuildConfig PermissionsFor(ulong guildId);
         IEnumerable<GuildConfig> PermissionsForAll();
         GuildConfig SetNewRootPermission(ulong guildId, Permission p);

src/NadekoBot/Services/Database/Repositories/Impl/GuildConfigRepository.cs

@@ -3,6 +3,7 @@ using System.Collections.Generic;
 using System.Linq;
 using Microsoft.EntityFrameworkCore;
 using NadekoBot.Modules.Permissions;
+using System;
 
 namespace NadekoBot.Services.Database.Repositories.Impl
 {
@@ -33,20 +34,30 @@ namespace NadekoBot.Services.Database.Repositories.Impl
         /// </summary>
         /// <param name="guildId"></param>
         /// <returns></returns>
-        public GuildConfig For(ulong guildId)
+        public GuildConfig For(ulong guildId, Func<DbSet<GuildConfig>, IQueryable<GuildConfig>> includes = null)
         {
-            var config = _set
+            GuildConfig config;
-                            .Include(gc => gc.FollowedStreams)
+
-                             .Include(gc => gc.LogSetting)
+            if (includes == null)
-                                .ThenInclude(ls => ls.IgnoredChannels)
+            {
-                            .Include(gc => gc.LogSetting)
+                config = _set
-                                .ThenInclude(ls => ls.IgnoredVoicePresenceChannelIds)
+                                .Include(gc => gc.FollowedStreams)
-                            .Include(gc => gc.FilterInvitesChannelIds)
+                                 .Include(gc => gc.LogSetting)
-                            .Include(gc => gc.FilterWordsChannelIds)
+                                    .ThenInclude(ls => ls.IgnoredChannels)
-                            .Include(gc => gc.FilteredWords)
+                                .Include(gc => gc.LogSetting)
-                            .Include(gc => gc.GenerateCurrencyChannelIds)
+                                    .ThenInclude(ls => ls.IgnoredVoicePresenceChannelIds)
-                            .Include(gc => gc.CommandCooldowns)
+                                .Include(gc => gc.FilterInvitesChannelIds)
-                            .FirstOrDefault(c => c.GuildId == guildId);
+                                .Include(gc => gc.FilterWordsChannelIds)
+                                .Include(gc => gc.FilteredWords)
+                                .Include(gc => gc.GenerateCurrencyChannelIds)
+                                .Include(gc => gc.CommandCooldowns)
+                                .FirstOrDefault(c => c.GuildId == guildId);
+            }
+            else
+            {
+                var set = includes(_set);
+                config = set.FirstOrDefault(c => c.GuildId == guildId);
+            }
 
             if (config == null)
             {