🚧 RSS as a module

modules/RichSiteSummary/Service/MessageSender.cs

@@ -0,0 +1,82 @@
+using System;
+using System.Collections.Concurrent;
+using System.Text;
+using System.Threading.Tasks;
+using BotFramework;
+using Kruzya.TelegramBot.Core.Service;
+using Microsoft.Extensions.Logging;
+using Telegram.Bot.Exceptions;
+using Telegram.Bot.Types;
+
+namespace Kruzya.TelegramBot.RichSiteSummary.Service
+{
+    public class MessageSender : AbstractTimedHostedService
+    {
+        protected override TimeSpan TimerPeriod => TimeSpan.FromSeconds(1);
+        
+        protected readonly ConcurrentQueue<UserMessage> _queue;
+        protected readonly ConcurrentQueue<UserUnsubscribe> _unsubscribeQueue;
+        
+        public MessageSender(ILogger<MessageSender> logger, ITelegramBot bot, ConcurrentQueue<UserMessage> userMessageQueue, ConcurrentQueue<UserUnsubscribe> userUnsubscribeQueue) : base(logger, bot)
+        {
+            _unsubscribeQueue = userUnsubscribeQueue;
+            _queue = userMessageQueue;
+        }
+
+        protected override async Task OnRun()
+        {
+            if (_queue.Count == 0)
+            {
+                return;
+            }
+
+            UserMessage message;
+            if (!_queue.TryDequeue(out message))
+            {
+                return;
+            }
+
+            try
+            {
+                await _bot.BotClient.SendTextMessageAsync(message.ChatId, message.Text, message.ParseMode,
+                    message.DisableWebPagePreview);
+            }
+            catch (ApiRequestException e)
+            {
+                if (!e.Message.Contains("bot was blocked by user"))
+                {
+                    ReEnqueue(message, e, message.ChatId); // looks like a network issue
+                    return;
+                }
+
+                // User added bot to blacklist.
+                // Unsubscribe him.
+                _unsubscribeQueue.Enqueue(new UserUnsubscribe()
+                {
+                    ChatId = message.ChatId
+                });
+            }
+            catch (Exception e)
+            {
+                ReEnqueue(message, e);
+            }
+        }
+        
+        private void ReEnqueue(UserMessage message, Exception e = null, ChatId chatId = null)
+        {
+            if (e != null)
+            {
+                var eMessage = new StringBuilder();
+                eMessage.Append(e.Message);
+                if (chatId != null)
+                {
+                    eMessage.Append($" ({chatId.Identifier})");
+                }
+
+                _logger.LogError(eMessage.ToString());
+            }
+
+            _queue.Enqueue(message);
+        }
+    }
+}

modules/RichSiteSummary/Service/RssFetch.cs

@@ -0,0 +1,179 @@
+using System;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+using CodeHollow.FeedReader;
+using Kruzya.TelegramBot.Core.Extensions;
+using Kruzya.TelegramBot.RichSiteSummary.Data;
+using Microsoft.Extensions.Configuration;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+using Telegram.Bot.Types;
+using Telegram.Bot.Types.Enums;
+using Feed = Kruzya.TelegramBot.RichSiteSummary.Data.Feed;
+
+namespace Kruzya.TelegramBot.RichSiteSummary.Service
+{
+    /// <summary>
+    /// Performs a RSS feed parsing.
+    /// Grabs the all feeds from database.
+    ///
+    /// TODO: move entity grabbing to another service.
+    /// </summary>
+    public class RssFetch : IHostedService, IDisposable
+    {
+        private Timer _timer;
+        
+        private readonly ILogger _logger;
+        private readonly IServiceScopeFactory _scopeFactory;
+        private readonly ConcurrentQueue<UserMessage> _queue;
+
+        private TimeSpan timerPeriod => TimeSpan.FromSeconds(45);
+
+        public RssFetch(ILogger<RssFetch> logger, ConcurrentQueue<UserMessage> queue, IServiceScopeFactory scopeFactory)
+        {
+            _scopeFactory = scopeFactory;
+            _logger = logger;
+            _queue = queue;
+        }
+
+        #region IHostedService
+        /// <summary>
+        /// Initializes the RSS fetcher timer.
+        /// </summary>
+        /// <param name="cancellationToken"></param>
+        /// <returns></returns>
+        public Task StartAsync(CancellationToken cancellationToken)
+        {
+            _logger.LogInformation("RSS fetcher service is starting.");
+            _timer = new Timer(DoFetch, null, TimeSpan.Zero, timerPeriod);
+
+            return Task.CompletedTask;
+        }
+
+        /// <summary>
+        /// Stops the RSS fetcher timer.
+        /// </summary>
+        /// <param name="cancellationToken"></param>
+        /// <returns></returns>
+        public Task StopAsync(CancellationToken cancellationToken)
+        {
+            _logger.LogInformation("RSS fetcher service is stopping.");
+            _timer?.Change(Timeout.Infinite, 0);
+
+            return Task.CompletedTask;
+        }
+        #endregion
+        #region IDisposable
+        /// <summary>
+        /// Disposes the timer.
+        /// </summary>
+        public void Dispose()
+        {
+            _timer?.Dispose();
+        }
+        #endregion
+
+        /// <summary>
+        /// Performs the job of fetching RSS data.
+        /// </summary>
+        /// <param name="state"></param>
+        private async void DoFetch(object state)
+        {
+            _timer.Change(Timeout.Infinite, 0);
+            _logger.LogDebug("RSS fetcher service is triggered.");
+
+            try
+            {
+                using var scope = _scopeFactory.CreateScope();
+                var dbContext = scope.ServiceProvider.GetRequiredService<RichSiteSummaryContext>();
+                var period = scope.ServiceProvider.GetRequiredService<IConfiguration>()
+                    .GetValue<UInt16>("rssFetchPeriod");
+
+                var feeds = await dbContext.Feeds.ForFetching(period);
+                _logger.LogDebug("Received {count} feeds for fetching", new {count = feeds.Length});
+
+                foreach (var feed in feeds)
+                {
+                    await ProcessFeed(feed, dbContext);
+
+                    feed.UpdatedAt = DateTime.Now;
+                    dbContext.MarkAsModified(feed);
+                }
+
+                await dbContext.SaveChangesAsync();
+            }
+            finally
+            {
+                _timer.Change(timerPeriod, timerPeriod);
+            }
+        }
+
+        #region Feeds
+
+        private async Task ProcessFeed(Feed feed, RichSiteSummaryContext dbContext)
+        {
+            var parsedFeed = await FetchFeed(feed);
+            if (parsedFeed == null)
+            {
+                return;
+            }
+            
+            Post feedPost;
+            var newPosts = new List<Post>();
+            foreach (var post in parsedFeed.Items)
+            {
+                feedPost = await dbContext.Posts.ByFeedAndUrl(post.Link, feed);
+                if (feedPost != null)
+                {
+                    // skip. This post already exists.
+                    continue;
+                }
+
+                feedPost = dbContext.Posts.Create();
+                feedPost.Feed = feed;
+                feedPost.Title = post.Title;
+                feedPost.Url = post.Link;
+                feedPost.PostedAt = post.PublishingDate.GetValueOrDefault(DateTime.Now);
+                
+                newPosts.Add(feedPost);
+            }
+
+            if (newPosts.Count > 0)
+            {
+                var subscribers = await dbContext.Subscriptions.ByFeed(feed);
+                foreach (var post in newPosts)
+                {
+                    var text = post.MessageText;
+                    foreach (var subscriber in subscribers)
+                    {
+                        var message = new UserMessage()
+                        {
+                            ChatId = new ChatId(subscriber.SubscriberId), DisableWebPagePreview = true,
+                            ParseMode = ParseMode.Html, Text = text
+                        };
+
+                        _queue.Enqueue(message);
+                    }
+                }
+            }
+        }
+
+        private async Task<CodeHollow.FeedReader.Feed> FetchFeed(Feed feed)
+        {
+            try
+            {
+                return await FeedReader.ReadAsync(feed.Url.ToString());
+            }
+            catch (Exception e)
+            {
+                _logger.LogError($"Feed {feed} can't be fetched: {e.Message}");
+                return null;
+            }
+        }
+        
+        #endregion
+    }
+}

modules/RichSiteSummary/Service/Unsubscriber.cs

@@ -0,0 +1,52 @@
+using System;
+using System.Collections.Concurrent;
+using System.Linq;
+using System.Threading.Tasks;
+using BotFramework;
+using Kruzya.TelegramBot.Core.Service;
+using Kruzya.TelegramBot.RichSiteSummary.Data;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Logging;
+using Telegram.Bot.Types;
+
+namespace Kruzya.TelegramBot.RichSiteSummary.Service
+{
+    public class Unsubscriber : AbstractTimedHostedService
+    {
+        protected readonly ConcurrentQueue<UserUnsubscribe> _queue;
+        protected readonly IServiceScopeFactory _scopeFactory;
+
+        protected override TimeSpan TimerPeriod => TimeSpan.FromSeconds(1);
+
+        public Unsubscriber(ILogger<Unsubscriber> logger, ITelegramBot bot, ConcurrentQueue<UserUnsubscribe> queue, IServiceScopeFactory scopeFactory) : base(logger, bot)
+        {
+            _scopeFactory = scopeFactory;
+            _queue = queue;
+        }
+
+        protected override async Task OnRun()
+        {
+            if (_queue.Count == 0)
+            {
+                return;
+            }
+
+            UserUnsubscribe user;
+            if (!_queue.TryDequeue(out user))
+            {
+                return;
+            }
+
+            await UnsubscribeUser(user.ChatId);
+        }
+
+        private async Task UnsubscribeUser(ChatId chatId)
+        {
+            using var scope = _scopeFactory.CreateScope();
+            var dbContext = scope.ServiceProvider.GetRequiredService<RichSiteSummaryContext>();
+            
+            dbContext.Subscriptions.RemoveRange(dbContext.Subscriptions.Where(s => s.SubscriberId == chatId.Identifier));
+            await dbContext.SaveChangesAsync();
+        }
+    }
+}