AI-Health/backend/src/Health.Infrastructure/Notifications/EfUserNotificationPipeline.cs

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<bool> 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<bool> 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<NotificationMessage>(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(),
    };
}