using System.Text.Json; using Health.Application.Notifications; using Health.Infrastructure.Data.Records; using Npgsql; namespace Health.Infrastructure.Notifications; public sealed class EfUserNotificationProducer(AppDbContext db) : IUserNotificationProducer { private readonly AppDbContext _db = db; public async Task EnqueueAsync( Guid userId, Guid sourceId, NotificationMessage message, CancellationToken ct) { var pushEnabled = await _db.NotificationPreferences.AsNoTracking() .Where(x => x.UserId == userId) .Select(x => (bool?)x.PushEnabled) .FirstOrDefaultAsync(ct); if (pushEnabled is false) return false; if (await _db.NotificationOutbox.AsNoTracking().AnyAsync(x => x.SourceTaskId == sourceId, ct)) return false; var now = DateTime.UtcNow; await _db.NotificationOutbox.AddAsync(new NotificationOutboxRecord { Id = Guid.NewGuid(), UserId = userId, SourceTaskId = sourceId, Type = message.Type, Payload = JsonSerializer.Serialize(message), Status = "Pending", AvailableAt = now, CreatedAt = now, UpdatedAt = now, }, ct); try { await _db.SaveChangesAsync(ct); return true; } catch (DbUpdateException ex) when ( ex.InnerException is PostgresException { SqlState: PostgresErrorCodes.UniqueViolation }) { _db.ChangeTracker.Clear(); return false; } } } public sealed class EfNotificationOutboxProcessor(AppDbContext db) : INotificationOutboxProcessor { private const int MaxAttempts = 5; private readonly AppDbContext _db = db; public async Task RecoverStaleAsync(CancellationToken ct) { var now = DateTime.UtcNow; await _db.NotificationOutbox .Where(x => x.Status == "Processing" && x.UpdatedAt < now.AddMinutes(-5) && x.Attempts < MaxAttempts) .ExecuteUpdateAsync(setters => setters .SetProperty(x => x.Status, "Pending") .SetProperty(x => x.AvailableAt, now) .SetProperty(x => x.UpdatedAt, now), ct); await _db.NotificationOutbox .Where(x => x.Status == "Processing" && x.UpdatedAt < now.AddMinutes(-5) && x.Attempts >= MaxAttempts) .ExecuteUpdateAsync(setters => setters .SetProperty(x => x.Status, "Failed") .SetProperty(x => x.LastError, "Notification task timed out and reached max retry count") .SetProperty(x => x.UpdatedAt, now), ct); } public async Task ProcessNextAsync(CancellationToken ct) { var now = DateTime.UtcNow; var candidate = await _db.NotificationOutbox.AsNoTracking() .Where(x => x.Status == "Pending" && x.AvailableAt <= now && x.Attempts < MaxAttempts) .OrderBy(x => x.CreatedAt) .FirstOrDefaultAsync(ct); if (candidate == null) return false; var task = await _db.NotificationOutbox .FirstOrDefaultAsync(x => x.Id == candidate.Id && x.Status == "Pending", ct); if (task == null) return true; task.Status = "Processing"; task.Attempts += 1; task.UpdatedAt = now; await _db.SaveChangesAsync(ct); try { var message = JsonSerializer.Deserialize(task.Payload) ?? throw new InvalidOperationException("Notification payload is empty"); var preference = await _db.NotificationPreferences.AsNoTracking() .FirstOrDefaultAsync(x => x.UserId == task.UserId, ct); var deliver = preference == null || ShouldDeliver(preference, message.Type, DateTime.UtcNow.AddHours(8)); if (deliver && !await _db.UserNotifications.AsNoTracking() .AnyAsync(x => x.SourceId == task.SourceTaskId, ct)) { await _db.UserNotifications.AddAsync(new UserNotification { Id = Guid.NewGuid(), UserId = task.UserId, SourceId = task.SourceTaskId, Type = message.Type, Title = message.Title, Message = message.Message, Severity = message.Severity, ActionType = message.ActionType, ActionTargetId = message.ActionTargetId, CreatedAt = task.CreatedAt, }, ct); await _db.SaveChangesAsync(ct); } task.Status = "Completed"; task.LastError = null; task.UpdatedAt = DateTime.UtcNow; await _db.SaveChangesAsync(ct); } catch (Exception ex) { task.Status = task.Attempts >= MaxAttempts ? "Failed" : "Pending"; task.AvailableAt = DateTime.UtcNow.AddSeconds(Math.Pow(2, task.Attempts) * 5); task.LastError = ex.Message.Length > 2000 ? ex.Message[..2000] : ex.Message; task.UpdatedAt = DateTime.UtcNow; await _db.SaveChangesAsync(ct); } return true; } private static bool ShouldDeliver(NotificationPreference preference, string type, DateTime nowCst) { if (!preference.PushEnabled) return false; var minuteOfDay = nowCst.Hour * 60 + nowCst.Minute; var inDnd = preference.DndEnabled && (preference.DndStartMinutes <= preference.DndEndMinutes ? minuteOfDay >= preference.DndStartMinutes && minuteOfDay < preference.DndEndMinutes : minuteOfDay >= preference.DndStartMinutes || minuteOfDay < preference.DndEndMinutes); if (inDnd) return false; return NormalizeType(type) switch { "medication_reminder" => preference.MedicationReminder, "exercise_reminder" => true, "follow_up_reminder" => preference.FollowUpReminder, "doctor_reply" => preference.DoctorReply, "abnormal_alert" => preference.AbnormalAlert, "health_record_reminder" => preference.HealthRecordReminder, "health_metric_alert" => preference.AbnormalAlert, _ => true, }; } private static string NormalizeType(string type) => type switch { "MedicationReminder" => "medication_reminder", "ExerciseReminder" => "exercise_reminder", "HealthMetricAlert" => "health_metric_alert", "FollowUpReminder" => "follow_up_reminder", _ => type.ToLowerInvariant(), }; }