db concurrency fixed

pavel-zhur · Sep 7, 2024 · cac316d · cac316d
1 parent f6e99b5
commit cac316d
Show file tree

Hide file tree

Showing 4 changed files with 31 additions and 9 deletions.
diff --git a/OneShelf.Videos/OneShelf.Videos.Telegram.Processor/OneShelf.Videos.Telegram.Processor.csproj b/OneShelf.Videos/OneShelf.Videos.Telegram.Processor/OneShelf.Videos.Telegram.Processor.csproj
@@ -7,6 +7,10 @@
 	  <IsPackable>false</IsPackable>
   </PropertyGroup>
 
+  <ItemGroup>
+    <PackageReference Include="Nito.AsyncEx.Coordination" Version="5.1.2" />
+  </ItemGroup>
+
 	<ItemGroup>
 		<ProjectReference Include="..\..\OneShelf.Telegram\OneShelf.Telegram\OneShelf.Telegram.csproj" />
 		<ProjectReference Include="..\OneShelf.Videos.BusinessLogic\OneShelf.Videos.BusinessLogic.csproj" />

diff --git a/OneShelf.Videos/OneShelf.Videos.Telegram.Processor/PipelineHandlers/VideosCollector.cs b/OneShelf.Videos/OneShelf.Videos.Telegram.Processor/PipelineHandlers/VideosCollector.cs
@@ -24,15 +24,17 @@ public class VideosCollector : PipelineHandler
     private readonly Scope _scope;
     private readonly HttpClient _httpClient;
     private readonly IOptions<VideosOptions> _videoOptions;
+    private readonly VideosCollectorMemory _videosCollectorMemory;
 
-    public VideosCollector(IScopedAbstractions scopedAbstractions, VideosDatabase videosDatabase, IOptions<TelegramOptions> telegramOptions, Scope scope, HttpClient httpClient, IOptions<VideosOptions> videoOptions) 
+    public VideosCollector(IScopedAbstractions scopedAbstractions, VideosDatabase videosDatabase, IOptions<TelegramOptions> telegramOptions, Scope scope, HttpClient httpClient, IOptions<VideosOptions> videoOptions, VideosCollectorMemory videosCollectorMemory) 
         : base(scopedAbstractions)
     {
         _videosDatabase = videosDatabase;
         _telegramOptions = telegramOptions;
         _scope = scope;
         _httpClient = httpClient;
         _videoOptions = videoOptions;
+        _videosCollectorMemory = videosCollectorMemory;
     }
 
     protected override async Task<bool> HandleSync(Update update)
@@ -124,12 +126,6 @@ protected override async Task<bool> HandleSync(Update update)
             return false;
         }
 
-        var alreadyResponded = false;
-        if (update.Message.MediaGroupId != null)
-        {
-            alreadyResponded = await _videosDatabase.TelegramMedia.AnyAsync(x => x.MediaGroupId == update.Message.MediaGroupId);
-        }
-
         var telegramMedia = new TelegramMedia
         {
             TelegramUpdateId = _scope.UpdateId,
@@ -157,9 +153,17 @@ protected override async Task<bool> HandleSync(Update update)
             ThumbnailHeight = thumbnail?.Height,
         };
 
-        _videosDatabase.TelegramMedia.Add(telegramMedia);
+        var alreadyResponded = false;
+        using (await _videosCollectorMemory.DatabaseLock.LockAsync())
+        {
+            if (update.Message.MediaGroupId != null)
+            {
+                alreadyResponded = await _videosDatabase.TelegramMedia.AnyAsync(x => x.MediaGroupId == update.Message.MediaGroupId);
+            }
 
-        await _videosDatabase.SaveChangesAsync();
+            _videosDatabase.TelegramMedia.Add(telegramMedia);
+            await _videosDatabase.SaveChangesAsync();
+        }
 
         QueueApi(null, api => Respond(api, update, telegramMedia, alreadyResponded));
 
@@ -170,6 +174,11 @@ private async Task Respond(TelegramBotClient api, Update update, TelegramMedia t
     {
         if (!alreadyResponded)
         {
+            await api.SetMessageReactionAsync(
+                update.Message!.Chat.Id,
+                update.Message.MessageId,
+                [new ReactionTypeEmoji("👀")]);
+
             await api.SendMessageAsync(
                 update.Message!.Chat.Id,
                 $"Got {telegramMedia.Id}.",

diff --git a/OneShelf.Videos/OneShelf.Videos.Telegram.Processor/ServiceCollectionExtensions.cs b/OneShelf.Videos/OneShelf.Videos.Telegram.Processor/ServiceCollectionExtensions.cs
@@ -35,6 +35,7 @@ public static IServiceCollection AddProcessor(this IServiceCollection services,
 
         services
             .AddVideosBusinessLogic(configuration)
+            .AddSingleton<VideosCollectorMemory>()
             .AddScoped<Scope>();
 
         return services;

diff --git a/OneShelf.Videos/OneShelf.Videos.Telegram.Processor/Services/VideosCollectorMemory.cs b/OneShelf.Videos/OneShelf.Videos.Telegram.Processor/Services/VideosCollectorMemory.cs
@@ -0,0 +1,8 @@
+using Nito.AsyncEx;
+
+namespace OneShelf.Videos.Telegram.Processor.Services;
+
+public class VideosCollectorMemory
+{
+    public AsyncLock DatabaseLock { get; } = new();
+}