Replace polling with push-based Telegram notification queue

RazorPagesMovie/Controllers/Api/V1/ContactController.cs

@@ -3,12 +3,13 @@ using Microsoft.AspNetCore.Mvc;
 using RazorPagesMovie.Api.V1.Contact;
 using RazorPagesMovie.Data;
 using RazorPagesMovie.Models;
+using RazorPagesMovie.Services;
 
 namespace RazorPagesMovie.Controllers.Api.V1;
 
 [ApiController]
 [Route("api/v1/[controller]")]
-public class ContactController(RazorPagesMovieContext context) : ControllerBase
+public class ContactController(RazorPagesMovieContext context, IContactMessageNotificationQueue notificationQueue) : ControllerBase
 {
     [HttpPost]
     public async Task<IActionResult> Post(ContactDto request)
@@ -31,6 +32,8 @@ public class ContactController(RazorPagesMovieContext context) : ControllerBase
         context.ContactMessage.Add(contactMessage);
         await context.SaveChangesAsync();
 
+        notificationQueue.Enqueue(contactMessage.Id);
+
         return Ok(new { success = true });
     }
 }

RazorPagesMovie/Program.cs

@@ -29,6 +29,7 @@ builder.Services.AddRazorPages(options =>
 builder.Services.AddControllers();
 
 builder.Services.AddHttpClient();
+builder.Services.AddSingleton<IContactMessageNotificationQueue, ContactMessageNotificationQueue>();
 builder.Services.AddHostedService<ContactMessageNotificationService>();
 
 var app = builder.Build();

RazorPagesMovie/Services/ContactMessageNotificationService.cs

@@ -6,35 +6,86 @@ namespace RazorPagesMovie.Services;
 
 public class ContactMessageNotificationService : BackgroundService
 {
-    private static readonly TimeSpan CheckInterval = TimeSpan.FromMinutes(5);
+    private static readonly TimeSpan FallbackSweepInterval = TimeSpan.FromMinutes(2);
 
     private readonly IServiceScopeFactory _scopeFactory;
     private readonly IHttpClientFactory _httpClientFactory;
     private readonly IConfiguration _configuration;
+    private readonly IContactMessageNotificationQueue _queue;
     private readonly ILogger<ContactMessageNotificationService> _logger;
 
     public ContactMessageNotificationService(
         IServiceScopeFactory scopeFactory,
         IHttpClientFactory httpClientFactory,
         IConfiguration configuration,
+        IContactMessageNotificationQueue queue,
         ILogger<ContactMessageNotificationService> logger)
     {
         _scopeFactory = scopeFactory;
         _httpClientFactory = httpClientFactory;
         _configuration = configuration;
+        _queue = queue;
         _logger = logger;
     }
 
     protected override async Task ExecuteAsync(CancellationToken stoppingToken)
     {
-        while (!stoppingToken.IsCancellationRequested)
+        var consumerTask = ConsumeQueueAsync(stoppingToken);
+        var fallbackSweepTask = RunFallbackSweepAsync(stoppingToken);
+
+        await Task.WhenAll(consumerTask, fallbackSweepTask);
+    }
+
+    private async Task ConsumeQueueAsync(CancellationToken stoppingToken)
+    {
+        await foreach (var contactMessageId in _queue.DequeueAllAsync(stoppingToken))
         {
-            await CheckForNewContactMessagesAsync(stoppingToken);
+            try
-            await Task.Delay(CheckInterval, stoppingToken);
+            {
+                await SendNotificationAsync(contactMessageId, stoppingToken);
+            }
+            finally
+            {
+                _queue.Release(contactMessageId);
+            }
         }
     }
 
-    private async Task CheckForNewContactMessagesAsync(CancellationToken stoppingToken)
+    private async Task RunFallbackSweepAsync(CancellationToken stoppingToken)
+    {
+        while (!stoppingToken.IsCancellationRequested)
+        {
+            await EnqueueUnsentFromDatabaseAsync(stoppingToken);
+
+            try
+            {
+                await Task.Delay(FallbackSweepInterval, stoppingToken);
+            }
+            catch (OperationCanceledException)
+            {
+                // Dienst wird beendet.
+            }
+        }
+    }
+
+    private async Task EnqueueUnsentFromDatabaseAsync(CancellationToken stoppingToken)
+    {
+        using var scope = _scopeFactory.CreateScope();
+        var context = scope.ServiceProvider.GetRequiredService<RazorPagesMovieContext>();
+
+        var offeneIds = await context.ContactMessage
+            .Where(c => !c.NotificationSent)
+            .OrderBy(c => c.SentAt)
+            .Select(c => c.Id)
+            .ToListAsync(stoppingToken);
+
+        foreach (var id in offeneIds)
+        {
+            _queue.Enqueue(id);
+        }
+    }
+
+    private async Task SendNotificationAsync(Guid contactMessageId, CancellationToken stoppingToken)
     {
         var botToken = _configuration["Telegram:BotToken"];
         var chatId = _configuration["Telegram:ChatId"];
@@ -48,48 +99,41 @@ public class ContactMessageNotificationService : BackgroundService
         using var scope = _scopeFactory.CreateScope();
         var context = scope.ServiceProvider.GetRequiredService<RazorPagesMovieContext>();
 
-        var neueEintraege = await context.ContactMessage
+        var eintrag = await context.ContactMessage.FindAsync(new object?[] { contactMessageId }, stoppingToken);
-            .Where(c => !c.NotificationSent)
+        if (eintrag is null || eintrag.NotificationSent)
-            .OrderBy(c => c.SentAt)
-            .ToListAsync(stoppingToken);
-
-        if (neueEintraege.Count == 0)
         {
             return;
         }
 
+        var text = $"Neue Kontaktanfrage von {eintrag.Name}\n"
+            + $"E-Mail: {eintrag.Email}\n"
+            + (string.IsNullOrWhiteSpace(eintrag.Phone) ? "" : $"Telefon: {eintrag.Phone}\n")
+            + (string.IsNullOrWhiteSpace(eintrag.Subject) ? "" : $"Betreff: {eintrag.Subject}\n")
+            + $"\n{eintrag.Message}";
+
         var httpClient = _httpClientFactory.CreateClient();
         var sendMessageUrl = $"https://api.telegram.org/bot{botToken}/sendMessage";
 
-        foreach (var eintrag in neueEintraege)
+        try
         {
-            var text = $"Neue Kontaktanfrage von {eintrag.Name}\n"
+            var response = await httpClient.PostAsJsonAsync(
-                + $"E-Mail: {eintrag.Email}\n"
+                sendMessageUrl,
-                + (string.IsNullOrWhiteSpace(eintrag.Phone) ? "" : $"Telefon: {eintrag.Phone}\n")
+                new { chat_id = chatId, text },
-                + (string.IsNullOrWhiteSpace(eintrag.Subject) ? "" : $"Betreff: {eintrag.Subject}\n")
+                stoppingToken);
-                + $"\n{eintrag.Message}";
 
-            try
+            if (response.IsSuccessStatusCode)
             {
-                var response = await httpClient.PostAsJsonAsync(
+                eintrag.NotificationSent = true;
-                    sendMessageUrl,
+                await context.SaveChangesAsync(stoppingToken);
-                    new { chat_id = chatId, text },
-                    stoppingToken);
-
-                if (response.IsSuccessStatusCode)
-                {
-                    eintrag.NotificationSent = true;
-                    await context.SaveChangesAsync(stoppingToken);
-                }
-                else
-                {
-                    _logger.LogWarning("Telegram-Versand fehlgeschlagen für ContactMessage {Id}: {StatusCode}", eintrag.Id, response.StatusCode);
-                }
             }
-            catch (Exception ex)
+            else
             {
-                _logger.LogError(ex, "Telegram-Versand fehlgeschlagen für ContactMessage {Id}", eintrag.Id);
+                _logger.LogWarning("Telegram-Versand fehlgeschlagen für ContactMessage {Id}: {StatusCode}", eintrag.Id, response.StatusCode);
             }
         }
+        catch (Exception ex)
+        {
+            _logger.LogError(ex, "Telegram-Versand fehlgeschlagen für ContactMessage {Id}", eintrag.Id);
+        }
     }
-}
+}

RazorPagesMovie/Services/IContactMessageNotificationQueue.cs

@@ -0,0 +1,32 @@
+using System.Collections.Concurrent;
+using System.Threading.Channels;
+
+namespace RazorPagesMovie.Services;
+
+public interface IContactMessageNotificationQueue
+{
+    void Enqueue(Guid contactMessageId);
+
+    IAsyncEnumerable<Guid> DequeueAllAsync(CancellationToken cancellationToken);
+
+    void Release(Guid contactMessageId);
+}
+
+public class ContactMessageNotificationQueue : IContactMessageNotificationQueue
+{
+    private readonly Channel<Guid> _channel = Channel.CreateUnbounded<Guid>();
+    private readonly ConcurrentDictionary<Guid, byte> _inFlight = new();
+
+    public void Enqueue(Guid contactMessageId)
+    {
+        if (_inFlight.TryAdd(contactMessageId, 0))
+        {
+            _channel.Writer.TryWrite(contactMessageId);
+        }
+    }
+
+    public IAsyncEnumerable<Guid> DequeueAllAsync(CancellationToken cancellationToken) =>
+        _channel.Reader.ReadAllAsync(cancellationToken);
+
+    public void Release(Guid contactMessageId) => _inFlight.TryRemove(contactMessageId, out _);
+}