using IM.Application.Chat.DTOs; using IM.Application.Chat.Queries; using IM.Domain.Common; using IM.Infrastructure.Persistence; using MediatR; namespace IM.Application.Chat.Handlers; /// /// 未读计数查询。读扩散热路径:Redis 优先,未命中回退 PostgreSQL 冷备。 /// /// /// 兼容策略:Redis 命中的会话用 Redis 值(权威);Redis 未命中的会话(老数据/降级期间写入)回退 PG 的 UnreadCount。 /// 二者合并去重,最终结果保证不丢未读。 /// public class GetUnreadCountQueryHandler(ImDbContext db, IUnreadCounter unreadCounter) : IRequestHandler { public async Task Handle(GetUnreadCountQuery request, CancellationToken ct) { // 先从 PG 取用户参与的所有会话及其冷备未读数(也作为 Redis 全故障时的兜底数据源)。 // 注意不能用 p.UnreadCount > 0 预过滤:发送链路已把未读累加移到 Redis,PG 冷备可能仍为 0。 var participants = await db.ConversationParticipants .Where(p => p.UserId == request.UserId) .Select(p => new { p.ConversationId, p.UnreadCount }) .ToListAsync(ct); if (participants.Count == 0) { return new UnreadCountsDto([], 0); } var pgUnread = participants.ToDictionary(p => p.ConversationId, p => p.UnreadCount); // 批量查 Redis 热数据 var redisUnread = await unreadCounter.GetByUserAsync( request.UserId, pgUnread.Keys, ct); // 合并:Redis 命中用 Redis 值,未命中回退 PG var items = pgUnread.Keys .Select(convId => new UnreadCountDto( convId, redisUnread.TryGetValue(convId, out var redisValue) ? redisValue : pgUnread[convId])) .Where(i => i.UnreadCount > 0) .ToList(); return new UnreadCountsDto(items, items.Sum(i => i.UnreadCount)); } }