Implemented scheduling service and first two jobs

2026-10-28 17:01:28 +02:00
parent 6114d8f457
commit 7f686fa8c6
20 changed files with 304 additions and 112 deletions
@@ -0,0 +1,122 @@
+using PostFundManagement.Application.Services.Shared;
+using PostFundManagement.Domain.Abstractions;
+using PostFundManagement.Infrastructure.Database;
+
+namespace PostFundManagement.Application.Events;
+
+public sealed class ProcessEmailEvent : IEvent
+{
+    public Guid Id { get; set; } = Guid.CreateVersion7();
+
+    public string Name { get; set; } = nameof(ProcessEmailEvent);
+
+    public DateTimeOffset EnqueueAt { get; set; } = DateTimeOffset.UtcNow;
+
+    public string CorrelationId { get; set; } = "email-processor";
+
+    public static ProcessEmailEvent Create() => new();
+}
+
+public sealed class ProcessEmailEventHandler(IDbContextFactory<ApplicationDbContext> contextFactory,
+    IFeatureManager featureManager, EmailService emailService, ISkorubaUserService skorubaUserService) : INotificationHandler<ProcessEmailEvent>
+{
+    const int maxBatch = 100;
+
+    public async ValueTask Handle(ProcessEmailEvent notification, CancellationToken cancellationToken)
+    {
+        if (!await featureManager.IsEnabledAsync(FeatureEmailProcessor)) return;
+
+        await using var context = await contextFactory.CreateDbContextAsync(cancellationToken);
+
+        var outgoing = await context.Notifications.AsNoTracking()
+            .Include(i => i.Organisation)
+            .Include(i => i.RecipientUser)
+            .Include(i => i.Tenant)
+            .Where(n => n.Status == Domain.NotificationStatus.Queued)
+            .OrderByDescending(o => o.CreatedAt)
+            .Take(maxBatch)
+            .ToListAsync(cancellationToken);
+
+        if (outgoing?.Count == 0) return;
+
+        if (emailService.Status is not (Domain.EmailStatuses.Connected or Domain.EmailStatuses.Success))
+            await emailService.ConnectAsync(cancellationToken);
+
+        var failedTransmissions = new List<(long notificationId, string? error)>();
+
+        foreach (var enqueuedMessage in outgoing!)
+        {
+            var userClaimResult = await skorubaUserService.GetUserClaimsAsync(enqueuedMessage.RecipientUser!.Id, cancellationToken: cancellationToken);
+
+            if (userClaimResult.IsFailed)
+            {
+                failedTransmissions.Add((enqueuedMessage.Id, "Failed to retrieve user claims"));
+
+                continue;
+            }
+
+            var name = userClaimResult.Value.FirstOrDefault(c => c.ClaimType == "given_name")?.ClaimValue ?? enqueuedMessage.Organisation!.Name;
+            var surname = userClaimResult.Value.FirstOrDefault(c => c.ClaimType == "family_name")?.ClaimValue ?? string.Empty;
+            var email = enqueuedMessage.RecipientUser?.Email ?? enqueuedMessage.Organisation?.Email ?? enqueuedMessage.Destination ?? string.Empty;
+
+            if (string.IsNullOrWhiteSpace(email))
+            {
+                failedTransmissions.Add((enqueuedMessage.Id, "Could not resolve destination email address"));
+
+                continue;
+            }
+
+            var message = new Domain.Models.Email.Message
+            {
+                Sender = new Domain.Models.Email.Party
+                {
+                    Address = "noreply@ifritechnologies.co.za",
+                    Name = "Post Fund Management"
+                },
+                Recipient = new Domain.Models.Email.Party
+                {
+                    Address = email,
+                    Name = string.IsNullOrWhiteSpace(surname)
+                        ? name : $"{name} {surname}"
+                },
+                Subject = enqueuedMessage.Subject,
+                Body = new Domain.Models.Email.Body
+                {
+                    Properties = new Domain.Models.Email.BodyProperties
+                    {
+                        HasAttachments = false,
+                        IsHtml = true
+                    },
+                    Message = enqueuedMessage.Message
+                }
+            };
+
+            var response = await emailService.SendEmailAsync(message, cancellationToken);
+
+            if (response.IsFailed || response.Value.Status is not (Domain.EmailStatuses.Connected or Domain.EmailStatuses.Success))
+            {
+                failedTransmissions.Add((enqueuedMessage.Id, $"Error: {response.Value.Error}"));
+
+                continue;
+            }
+        }
+
+        foreach (var (notificationId, error) in failedTransmissions)
+            await context.Notifications.Where(n => n.Id == notificationId)
+                    .ExecuteUpdateAsync(p => p
+                        .SetProperty(u => u.Status, Domain.NotificationStatus.Failed)
+                        .SetProperty(u => u.Error, error), cancellationToken);
+
+        var failedIds = failedTransmissions.Select(f => f.notificationId).ToList();
+
+        foreach (var successMessage in outgoing)
+        {
+            if (failedIds.Contains(successMessage.Id)) continue;
+
+            await context.Notifications.Where(n => n.Id == successMessage.Id)
+                    .ExecuteUpdateAsync(p => p
+                        .SetProperty(u => u.Status, Domain.NotificationStatus.Sent)
+                        .SetProperty(u => u.Error, string.Empty), cancellationToken);
+        }
+    }
+}