mirror of
https://github.com/Bubuni-Team/telegram-bot.git
synced 2026-07-31 00:29:24 +03:00
Use file-scoped namespaces
This commit is contained in:
Andriy
@@ -9,81 +9,80 @@ using Telegram.Bot;
|
||||
using Telegram.Bot.Exceptions;
|
||||
using Telegram.Bot.Types;
|
||||
|
||||
namespace Kruzya.TelegramBot.RichSiteSummary.Service
|
||||
namespace Kruzya.TelegramBot.RichSiteSummary.Service;
|
||||
|
||||
public class MessageSender : AbstractTimedHostedService
|
||||
{
|
||||
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, IBotInstance bot, ConcurrentQueue<UserMessage> userMessageQueue, ConcurrentQueue<UserUnsubscribe> userUnsubscribeQueue) : base(logger, bot)
|
||||
{
|
||||
protected override TimeSpan TimerPeriod => TimeSpan.FromSeconds(1);
|
||||
|
||||
protected readonly ConcurrentQueue<UserMessage> Queue;
|
||||
protected readonly ConcurrentQueue<UserUnsubscribe> UnsubscribeQueue;
|
||||
|
||||
public MessageSender(ILogger<MessageSender> logger, IBotInstance bot, ConcurrentQueue<UserMessage> userMessageQueue, ConcurrentQueue<UserUnsubscribe> userUnsubscribeQueue) : base(logger, bot)
|
||||
UnsubscribeQueue = userUnsubscribeQueue;
|
||||
Queue = userMessageQueue;
|
||||
}
|
||||
|
||||
protected override async Task OnRun()
|
||||
{
|
||||
if (Queue.IsEmpty || !Queue.TryDequeue(out var message))
|
||||
{
|
||||
UnsubscribeQueue = userUnsubscribeQueue;
|
||||
Queue = userMessageQueue;
|
||||
return;
|
||||
}
|
||||
|
||||
protected override async Task OnRun()
|
||||
_logger.LogDebug("Message for {ChatId} dequeued.", message.ChatId.Identifier);
|
||||
try
|
||||
{
|
||||
if (Queue.IsEmpty || !Queue.TryDequeue(out var message))
|
||||
if (string.IsNullOrWhiteSpace(message.ImageUrl))
|
||||
{
|
||||
await _bot.BotClient.SendTextMessageAsync(message.ChatId, message.Text, 0, message.ParseMode, null,
|
||||
message.DisableWebPagePreview);
|
||||
}
|
||||
|
||||
else
|
||||
{
|
||||
await _bot.BotClient.SendPhotoAsync(message.ChatId, new InputFileUrl(message.ImageUrl), 0, message.Text,
|
||||
message.ParseMode, null, false, message.DisableWebPagePreview);
|
||||
}
|
||||
}
|
||||
catch (ApiRequestException e)
|
||||
{
|
||||
_logger.LogDebug("Message for {ChatId} is failed: {error}.", message.ChatId.Identifier, e.Message);
|
||||
if (!e.Message.Contains("bot was blocked by user"))
|
||||
{
|
||||
ReEnqueue(message, e, message.ChatId); // looks like a network issue
|
||||
return;
|
||||
}
|
||||
|
||||
_logger.LogDebug("Message for {ChatId} dequeued.", message.ChatId.Identifier);
|
||||
try
|
||||
// User added bot to blacklist.
|
||||
// Unsubscribe him.
|
||||
UnsubscribeQueue.Enqueue(new UserUnsubscribe
|
||||
{
|
||||
if (string.IsNullOrWhiteSpace(message.ImageUrl))
|
||||
{
|
||||
await _bot.BotClient.SendTextMessageAsync(message.ChatId, message.Text, 0, message.ParseMode, null,
|
||||
message.DisableWebPagePreview);
|
||||
}
|
||||
|
||||
else
|
||||
{
|
||||
await _bot.BotClient.SendPhotoAsync(message.ChatId, new InputFileUrl(message.ImageUrl), 0, message.Text,
|
||||
message.ParseMode, null, false, message.DisableWebPagePreview);
|
||||
}
|
||||
}
|
||||
catch (ApiRequestException e)
|
||||
{
|
||||
_logger.LogDebug("Message for {ChatId} is failed: {error}.", message.ChatId.Identifier, e.Message);
|
||||
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)
|
||||
{
|
||||
_logger.LogDebug("Message for {ChatId} is failed: {error}.", message.ChatId.Identifier, e.Message);
|
||||
ReEnqueue(message, e);
|
||||
}
|
||||
ChatId = message.ChatId
|
||||
});
|
||||
}
|
||||
|
||||
private void ReEnqueue(UserMessage message, Exception e = null, ChatId chatId = null)
|
||||
catch (Exception e)
|
||||
{
|
||||
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);
|
||||
_logger.LogDebug("Message for {ChatId} is failed: {error}.", message.ChatId.Identifier, e.Message);
|
||||
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);
|
||||
}
|
||||
}
|
||||
@@ -20,242 +20,241 @@ using Feed = Kruzya.TelegramBot.RichSiteSummary.Data.Feed;
|
||||
|
||||
using CodeHollow.FeedReader.Feeds;
|
||||
|
||||
namespace Kruzya.TelegramBot.RichSiteSummary.Service
|
||||
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
|
||||
{
|
||||
/// <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 Timer _timer;
|
||||
|
||||
private readonly ILogger _logger;
|
||||
private readonly IServiceScopeFactory _scopeFactory;
|
||||
private readonly ConcurrentQueue<UserMessage> _queue;
|
||||
private readonly ILogger _logger;
|
||||
private readonly IServiceScopeFactory _scopeFactory;
|
||||
private readonly ConcurrentQueue<UserMessage> _queue;
|
||||
|
||||
private static TimeSpan TimerPeriod => TimeSpan.FromSeconds(45);
|
||||
private static TimeSpan TimerPeriod => TimeSpan.FromSeconds(45);
|
||||
|
||||
public RssFetch(ILogger<RssFetch> logger, ConcurrentQueue<UserMessage> queue, IServiceScopeFactory scopeFactory)
|
||||
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
|
||||
{
|
||||
_scopeFactory = scopeFactory;
|
||||
_logger = logger;
|
||||
_queue = queue;
|
||||
}
|
||||
using var scope = _scopeFactory.CreateScope();
|
||||
var dbContext = scope.ServiceProvider.GetRequiredService<RichSiteSummaryContext>();
|
||||
var period = scope.ServiceProvider.GetRequiredService<IConfiguration>()
|
||||
.GetValue<ushort>("rssFetchPeriod");
|
||||
|
||||
#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);
|
||||
var feeds = await dbContext.Feeds.ForFetching(period);
|
||||
_logger.LogDebug("Received {count} feeds for fetching", new {count = feeds.Length});
|
||||
|
||||
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
|
||||
foreach (var feed in feeds)
|
||||
{
|
||||
using var scope = _scopeFactory.CreateScope();
|
||||
var dbContext = scope.ServiceProvider.GetRequiredService<RichSiteSummaryContext>();
|
||||
var period = scope.ServiceProvider.GetRequiredService<IConfiguration>()
|
||||
.GetValue<ushort>("rssFetchPeriod");
|
||||
await ProcessFeed(feed, dbContext);
|
||||
|
||||
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.Update(feed);
|
||||
}
|
||||
|
||||
await dbContext.SaveChangesAsync();
|
||||
}
|
||||
finally
|
||||
{
|
||||
_timer.Change(TimerPeriod, TimerPeriod);
|
||||
feed.UpdatedAt = DateTime.Now;
|
||||
dbContext.Update(feed);
|
||||
}
|
||||
|
||||
await dbContext.SaveChangesAsync();
|
||||
}
|
||||
|
||||
#region Feeds
|
||||
|
||||
private async Task ProcessFeed(Feed feed, RichSiteSummaryContext dbContext)
|
||||
finally
|
||||
{
|
||||
var parsedFeed = await FetchFeed(feed);
|
||||
if (parsedFeed == null)
|
||||
{
|
||||
return;
|
||||
}
|
||||
_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>();
|
||||
var postsImages = new Dictionary<Post, string>();
|
||||
foreach (var post in parsedFeed.Items)
|
||||
Post feedPost;
|
||||
var newPosts = new List<Post>();
|
||||
var postsImages = new Dictionary<Post, string>();
|
||||
foreach (var post in parsedFeed.Items)
|
||||
{
|
||||
feedPost = await dbContext.Posts.ByFeedAndUrl(post.Link, feed);
|
||||
if (feedPost != null)
|
||||
{
|
||||
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);
|
||||
|
||||
postsImages.Add(feedPost, await FetchImageUrl(post));
|
||||
newPosts.Add(feedPost);
|
||||
// skip. This post already exists.
|
||||
continue;
|
||||
}
|
||||
|
||||
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, ImageUrl = postsImages.GetValueOrDefault(post)
|
||||
};
|
||||
feedPost = dbContext.Posts.Create();
|
||||
feedPost.Feed = feed;
|
||||
feedPost.Title = post.Title;
|
||||
feedPost.Url = post.Link;
|
||||
feedPost.PostedAt = post.PublishingDate.GetValueOrDefault(DateTime.Now);
|
||||
|
||||
_queue.Enqueue(message);
|
||||
_logger.LogDebug("Queued message for {SubscriberId}", subscriber.SubscriberId);
|
||||
}
|
||||
postsImages.Add(feedPost, await FetchImageUrl(post));
|
||||
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, ImageUrl = postsImages.GetValueOrDefault(post)
|
||||
};
|
||||
|
||||
_queue.Enqueue(message);
|
||||
_logger.LogDebug("Queued message for {SubscriberId}", subscriber.SubscriberId);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private async Task<CodeHollow.FeedReader.Feed> FetchFeed(Feed feed)
|
||||
private async Task<CodeHollow.FeedReader.Feed> FetchFeed(Feed feed)
|
||||
{
|
||||
try
|
||||
{
|
||||
return await FeedReader.ReadAsync(feed.Url);
|
||||
}
|
||||
catch (Exception e)
|
||||
{
|
||||
_logger.LogError("Feed {feed} can't be fetched: {error}", feed, e.Message);
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
#endregion
|
||||
|
||||
#region Image parsing
|
||||
|
||||
private async Task<string> FetchImageUrl(FeedItem item)
|
||||
{
|
||||
var feedItem = item.SpecificItem;
|
||||
if (feedItem is Rss20FeedItem rss20FeedItem)
|
||||
{
|
||||
var enclosure = rss20FeedItem.Enclosure;
|
||||
if (enclosure != null && IsValidImage(enclosure.MediaType))
|
||||
{
|
||||
return enclosure.Url;
|
||||
}
|
||||
}
|
||||
|
||||
return await FetchImageUrl(item.Description);
|
||||
}
|
||||
|
||||
private async Task<string> FetchImageUrl(string description)
|
||||
{
|
||||
if (string.IsNullOrWhiteSpace(description))
|
||||
{
|
||||
return "";
|
||||
}
|
||||
|
||||
using var httpClientHandler = new HttpClientHandler() { AllowAutoRedirect = true };
|
||||
using var httpClient = new HttpClient(httpClientHandler);
|
||||
|
||||
var imgRegEx = new Regex(@"<img[^>]+>");
|
||||
var srcRegEx = new Regex("src=\"([^\"]+)\"");
|
||||
var matches = imgRegEx.Matches(description);
|
||||
|
||||
var img = "";
|
||||
foreach (Match match in matches)
|
||||
{
|
||||
try
|
||||
{
|
||||
return await FeedReader.ReadAsync(feed.Url);
|
||||
var srcData = srcRegEx.Match(match.Value);
|
||||
var content = srcData.Groups[1].Value;
|
||||
|
||||
var response = await httpClient.SendAsync(new HttpRequestMessage(HttpMethod.Head, content));
|
||||
if (response.StatusCode == HttpStatusCode.OK &&
|
||||
IsValidImage(response.Content.Headers.ContentType.MediaType))
|
||||
{
|
||||
img = content;
|
||||
break;
|
||||
}
|
||||
}
|
||||
catch (Exception e)
|
||||
{
|
||||
_logger.LogError("Feed {feed} can't be fetched: {error}", feed, e.Message);
|
||||
return null;
|
||||
_logger.LogError("Caused error when fetching image for RSS post: {error}", e.Message);
|
||||
}
|
||||
}
|
||||
|
||||
#endregion
|
||||
|
||||
#region Image parsing
|
||||
|
||||
private async Task<string> FetchImageUrl(FeedItem item)
|
||||
{
|
||||
var feedItem = item.SpecificItem;
|
||||
if (feedItem is Rss20FeedItem rss20FeedItem)
|
||||
{
|
||||
var enclosure = rss20FeedItem.Enclosure;
|
||||
if (enclosure != null && IsValidImage(enclosure.MediaType))
|
||||
{
|
||||
return enclosure.Url;
|
||||
}
|
||||
}
|
||||
|
||||
return await FetchImageUrl(item.Description);
|
||||
}
|
||||
|
||||
private async Task<string> FetchImageUrl(string description)
|
||||
{
|
||||
if (string.IsNullOrWhiteSpace(description))
|
||||
{
|
||||
return "";
|
||||
}
|
||||
|
||||
using var httpClientHandler = new HttpClientHandler() { AllowAutoRedirect = true };
|
||||
using var httpClient = new HttpClient(httpClientHandler);
|
||||
|
||||
var imgRegEx = new Regex(@"<img[^>]+>");
|
||||
var srcRegEx = new Regex("src=\"([^\"]+)\"");
|
||||
var matches = imgRegEx.Matches(description);
|
||||
|
||||
var img = "";
|
||||
foreach (Match match in matches)
|
||||
{
|
||||
try
|
||||
{
|
||||
var srcData = srcRegEx.Match(match.Value);
|
||||
var content = srcData.Groups[1].Value;
|
||||
|
||||
var response = await httpClient.SendAsync(new HttpRequestMessage(HttpMethod.Head, content));
|
||||
if (response.StatusCode == HttpStatusCode.OK &&
|
||||
IsValidImage(response.Content.Headers.ContentType.MediaType))
|
||||
{
|
||||
img = content;
|
||||
break;
|
||||
}
|
||||
}
|
||||
catch (Exception e)
|
||||
{
|
||||
_logger.LogError("Caused error when fetching image for RSS post: {error}", e.Message);
|
||||
}
|
||||
}
|
||||
|
||||
return img;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Checks if image is valid for Telegram Bot API.
|
||||
/// </summary>
|
||||
/// <param name="contentType">Content-type for received content</param>
|
||||
/// <returns>True, if Telegram maybe can "use" this image, false if not.</returns>
|
||||
private static bool IsValidImage(string contentType)
|
||||
{
|
||||
return new[]
|
||||
{
|
||||
"image/jpeg",
|
||||
"image/bmp",
|
||||
"image/png"
|
||||
}.Contains(contentType);
|
||||
}
|
||||
|
||||
#endregion
|
||||
return img;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Checks if image is valid for Telegram Bot API.
|
||||
/// </summary>
|
||||
/// <param name="contentType">Content-type for received content</param>
|
||||
/// <returns>True, if Telegram maybe can "use" this image, false if not.</returns>
|
||||
private static bool IsValidImage(string contentType)
|
||||
{
|
||||
return new[]
|
||||
{
|
||||
"image/jpeg",
|
||||
"image/bmp",
|
||||
"image/png"
|
||||
}.Contains(contentType);
|
||||
}
|
||||
|
||||
#endregion
|
||||
}
|
||||
@@ -9,41 +9,40 @@ using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Telegram.Bot.Types;
|
||||
|
||||
namespace Kruzya.TelegramBot.RichSiteSummary.Service
|
||||
namespace Kruzya.TelegramBot.RichSiteSummary.Service;
|
||||
|
||||
public class Unsubscriber : AbstractTimedHostedService
|
||||
{
|
||||
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, IBotInstance bot, ConcurrentQueue<UserUnsubscribe> queue, IServiceScopeFactory scopeFactory) : base(logger, bot)
|
||||
{
|
||||
protected readonly ConcurrentQueue<UserUnsubscribe> Queue;
|
||||
protected readonly IServiceScopeFactory ScopeFactory;
|
||||
ScopeFactory = scopeFactory;
|
||||
Queue = queue;
|
||||
}
|
||||
|
||||
protected override TimeSpan TimerPeriod => TimeSpan.FromSeconds(1);
|
||||
|
||||
public Unsubscriber(ILogger<Unsubscriber> logger, IBotInstance bot, ConcurrentQueue<UserUnsubscribe> queue, IServiceScopeFactory scopeFactory) : base(logger, bot)
|
||||
protected override async Task OnRun()
|
||||
{
|
||||
if (Queue.IsEmpty || !Queue.TryDequeue(out var user))
|
||||
{
|
||||
ScopeFactory = scopeFactory;
|
||||
Queue = queue;
|
||||
return;
|
||||
}
|
||||
|
||||
protected override async Task OnRun()
|
||||
{
|
||||
if (Queue.IsEmpty || !Queue.TryDequeue(out var user))
|
||||
{
|
||||
return;
|
||||
}
|
||||
await UnsubscribeUser(user.ChatId);
|
||||
}
|
||||
|
||||
await UnsubscribeUser(user.ChatId);
|
||||
}
|
||||
private async Task UnsubscribeUser(ChatId chatId)
|
||||
{
|
||||
using var scope = ScopeFactory.CreateScope();
|
||||
var dbContext = scope.ServiceProvider.GetRequiredService<RichSiteSummaryContext>();
|
||||
|
||||
private async Task UnsubscribeUser(ChatId chatId)
|
||||
{
|
||||
using var scope = ScopeFactory.CreateScope();
|
||||
var dbContext = scope.ServiceProvider.GetRequiredService<RichSiteSummaryContext>();
|
||||
|
||||
var subscriptions = dbContext.Subscriptions;
|
||||
subscriptions.RemoveRange(
|
||||
subscriptions.Where(s => s.SubscriberId == chatId.Identifier)
|
||||
);
|
||||
await dbContext.SaveChangesAsync();
|
||||
}
|
||||
var subscriptions = dbContext.Subscriptions;
|
||||
subscriptions.RemoveRange(
|
||||
subscriptions.Where(s => s.SubscriberId == chatId.Identifier)
|
||||
);
|
||||
await dbContext.SaveChangesAsync();
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user