feat: 应用内通知系统 + 结构化手术史/用药等相关改动

- 新增用户通知 outbox 流水线(EfUserNotificationPipeline)与后台投递 worker - 通知中心页面及前端通知服务接入 - 健康指标异常、用药/运动提醒等事件统一产出站内通知 - 健康档案结构化手术史、用药提醒扫描、医生/用户端点等配套调整 - AppDbContext 注册通知相关实体
2026-06-21 21:06:29 +08:00
parent b57d0d16f4
commit 13714d9ed8
34 changed files with 1541 additions and 200 deletions
--- a/backend/src/Health.Infrastructure/Notifications/EfUserNotificationPipeline.cs
+++ b/backend/src/Health.Infrastructure/Notifications/EfUserNotificationPipeline.cs
@@ -0,0 +1,139 @@
+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)
+    {
+        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, "通知任务执行超时且已达到最大重试次数")
+                .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 claimed = await _db.NotificationOutbox
+            .Where(x => x.Id == candidate.Id && x.Status == "Pending")
+            .ExecuteUpdateAsync(setters => setters
+                .SetProperty(x => x.Status, "Processing")
+                .SetProperty(x => x.Attempts, x => x.Attempts + 1)
+                .SetProperty(x => x.UpdatedAt, now), ct);
+        if (claimed == 0) return true;
+
+        try
+        {
+            var message = JsonSerializer.Deserialize<NotificationMessage>(candidate.Payload)
+                ?? throw new InvalidOperationException("通知消息内容为空");
+
+            if (!await _db.UserNotifications.AsNoTracking()
+                    .AnyAsync(x => x.SourceId == candidate.SourceTaskId, ct))
+            {
+                await _db.UserNotifications.AddAsync(new UserNotification
+                {
+                    Id = Guid.NewGuid(),
+                    UserId = candidate.UserId,
+                    SourceId = candidate.SourceTaskId,
+                    Type = message.Type,
+                    Title = message.Title,
+                    Message = message.Message,
+                    Severity = message.Severity,
+                    ActionType = message.ActionType,
+                    ActionTargetId = message.ActionTargetId,
+                    CreatedAt = candidate.CreatedAt,
+                }, ct);
+                // Persist the inbox item first. If completion fails, stale-task
+                // recovery can safely retry because SourceId is idempotent.
+                await _db.SaveChangesAsync(ct);
+            }
+
+            await _db.NotificationOutbox.Where(x => x.Id == candidate.Id)
+                .ExecuteUpdateAsync(setters => setters
+                    .SetProperty(x => x.Status, "Completed")
+                    .SetProperty(x => x.LastError, (string?)null)
+                    .SetProperty(x => x.UpdatedAt, DateTime.UtcNow), ct);
+        }
+        catch (Exception ex)
+        {
+            var attempts = candidate.Attempts + 1;
+            await _db.NotificationOutbox.Where(x => x.Id == candidate.Id)
+                .ExecuteUpdateAsync(setters => setters
+                    .SetProperty(x => x.Status, attempts >= MaxAttempts ? "Failed" : "Pending")
+                    .SetProperty(x => x.AvailableAt,
+                        DateTime.UtcNow.AddSeconds(Math.Pow(2, attempts) * 5))
+                    .SetProperty(x => x.LastError,
+                        ex.Message.Length > 2000 ? ex.Message[..2000] : ex.Message)
+                    .SetProperty(x => x.UpdatedAt, DateTime.UtcNow), ct);
+        }
+
+        return true;
+    }
+}