Merge pull request #37 from ZenonEl/feat/download-queue

feat(queue): add download queue with configurable concurrency limit

Author: ZenonEl

Config/Config.cs

@@ -44,6 +44,9 @@ class Config
         public static int torControlPort = 9051;
         public static int torChangingChainInterval = 5; // Minutes
 
+        // Download Queue
+        public static int maxConcurrentDownloads = 3;
+
         // Delays
         public static int videoGetDelay = 1000;
         public static int contactSendDelay = 1000;
@@ -87,6 +90,7 @@ public static void LoadConfig()
             dbType = configuration.GetValue("AppSettings:DatabaseType", "sqlite");
             isUseGalleryDl = configuration.GetValue("AppSettings:UseGalleryDl", false);
             accessDeniedMessageContact = configuration.GetValue("AppSettings:AccessDeniedMessageContact", " ");
+            maxConcurrentDownloads = configuration.GetValue("AppSettings:MaxConcurrentDownloads", 3);
 
             videoGetDelay = configuration.GetValue("MessageDelaySettings:VideoGetDelay", 1000);
             contactSendDelay = configuration.GetValue("MessageDelaySettings:ContactSendDelay", 1000);

Program.cs

@@ -67,6 +67,7 @@ static async Task Main(string[] args)
                 Scheduler scheduler = app.Services.GetRequiredService<Scheduler>();
 
                 Log.Information($"Log level: {Config.logLevel}");
+                DownloadQueue.Initialize(Config.maxConcurrentDownloads);
                 scheduler.Init();
 
                 await tgBot.Start();

TelegramBot/MediaDownloader.cs

@@ -152,7 +152,28 @@ private async Task UpdateHandler(ITelegramBotClient botClient, Update update, Ca
     public async Task HandleMediaRequest(ITelegramBotClient botClient, string contentUrl, long chatId, Message statusMessage,
                                                 List<long>? targetUserIds = null, bool groupChat = false, string caption = "")
     {
-        List<byte[]>? mediaFiles = await MediaGet.DownloadMedia(botClient, contentUrl, statusMessage, cancellationToken);
+        List<byte[]>? mediaFiles = await DownloadQueue.EnqueueAsync(
+            async () => await MediaGet.DownloadMedia(botClient, contentUrl, statusMessage, cancellationToken),
+            position =>
+            {
+                _ = Task.Run(async () =>
+                {
+                    try
+                    {
+                        await botClient.EditMessageText(
+                            statusMessage.Chat.Id,
+                            statusMessage.MessageId,
+                            $"⏳ Queued (#{position})",
+                            cancellationToken: cancellationToken);
+                    }
+                    catch (Exception ex)
+                    {
+                        Log.Debug(ex, "Error editing queue status message.");
+                    }
+                });
+            },
+            cancellationToken);
+
         if (mediaFiles?.Count > 0)
         {
             Log.Debug($"Downloaded {mediaFiles.Count} files");

TelegramBot/Services/DownloadQueue.cs

@@ -0,0 +1,49 @@
+// Copyright (C) 2024-2025 ZenonEl
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published
+// by the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// Эта программа является свободным программным обеспечением: вы можете распространять и/или изменять
+// её на условиях Стандартной общественной лицензии GNU Affero, опубликованной
+// Фондом свободного программного обеспечения, либо версии 3 лицензии, либо
+// (по вашему выбору) любой более поздней версии.
+
+namespace TelegramMediaRelayBot
+{
+    public static class DownloadQueue
+    {
+        private static SemaphoreSlim _semaphore = null!;
+        private static int _queuedCount = 0;
+
+        public static void Initialize(int maxConcurrent)
+        {
+            _semaphore = new SemaphoreSlim(maxConcurrent, maxConcurrent);
+        }
+
+        public static async Task<T?> EnqueueAsync<T>(
+            Func<Task<T?>> downloadFunc,
+            Action<int>? onQueued = null,
+            CancellationToken ct = default)
+        {
+            int position = Interlocked.Increment(ref _queuedCount);
+            try
+            {
+                if (_semaphore.CurrentCount == 0)
+                    onQueued?.Invoke(position);
+
+                await _semaphore.WaitAsync(ct);
+                Interlocked.Decrement(ref _queuedCount);
+
+                return await downloadFunc();
+            }
+            finally
+            {
+                _semaphore.Release();
+            }
+        }
+
+        public static int QueuedCount => _queuedCount;
+        public static int ActiveCount => Config.maxConcurrentDownloads - _semaphore.CurrentCount;
+    }
+}

appsettings.json.example

@@ -14,7 +14,10 @@
 
         "_comment_telegram_api": "TelegramApiBaseUrl: custom Bot API server URL (null = default). TelegramApiProxy: proxy for Telegram API requests, useful in regions where Telegram is blocked.",
         "TelegramApiBaseUrl": null,
-        "TelegramApiProxy": null
+        "TelegramApiProxy": null,
+
+        "_comment_queue": "Maximum number of simultaneous media downloads. Prevents system overload when many links arrive at once.",
+        "MaxConcurrentDownloads": 3
     },
     "_comment_tor": "Tor SOCKS proxy for media downloads and automatic circuit rotation.",
     "Tor": {