build-flow.api/tests/BuildFlow.IM.Integration.Tests/ChatHubReadAndRecallTests.cs

using IM.Application.Chat.DTOs;
using IM.Domain.Common;
using IM.Domain.Enums;
using IM.Infrastructure.Mongo;
using IM.Infrastructure.Persistence;
using MongoDB.Entities;
using Xunit;
using Microsoft.AspNetCore.SignalR.Client;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.DependencyInjection;

namespace BuildFlow.IM.Integration.Tests;

/// <summary>
/// 已读回执与消息撤回的端到端用例。
/// 覆盖：已读双写清零（PG+Redis）并广播、撤回标记 Mongo 并广播、撤回会话摘要更新。
/// </summary>
[Xunit.Collection(ImServerCollection.Name)]
public class ChatHubReadAndRecallTests : IAsyncLifetime
{
    private readonly ImServerFixture _fixture;
    private readonly List<HubConnection> _connections = new();

    public ChatHubReadAndRecallTests(ImServerFixture fixture) => _fixture = fixture;

    public Task InitializeAsync() => _fixture.CleanRedisAsync();

    public async Task DisposeAsync()
    {
        foreach (var conn in _connections)
        {
            try { await conn.DisposeAsync(); } catch { }
        }
    }

    private async Task<HubConnection> ConnectAsync(Guid userId, string userName)
    {
        var conn = HubConnectionFactory.Create(_fixture, userId, userName);
        await conn.StartAsync();
        _connections.Add(conn);
        return conn;
    }

    [Fact]
    public async Task MarkAsRead_resets_unread_in_both_pg_and_redis_and_broadcasts()
    {
        // Arrange：私聊会话，Alice 先发消息使 Bob 产生未读
        _fixture.UserInfoService.SetUser(ImServerFixture.AliceId, "Alice");
        _fixture.UserInfoService.SetUser(ImServerFixture.BobId, "Bob");
        var convId = await ImTestSeeder.SeedPrivateConversationAsync(_fixture, ImServerFixture.AliceId, ImServerFixture.BobId);

        var alice = await ConnectAsync(ImServerFixture.AliceId, "Alice");
        var bob = await ConnectAsync(ImServerFixture.BobId, "Bob");

        string? lastMessageId = null;
        bob.On<MessageDto>("OnMessageReceived", m => lastMessageId = m.Id);

        await alice.InvokeAsync("SendMessage", new SendMessageRequest(
            convId, MessageContentTypeEnum.Text, "read me", null, null));
        await WaitForAsync(() => lastMessageId is not null);

        // 此时 Bob 有 1 条未读（Redis）
        var unreadCounter = _fixture.Services.GetRequiredService<IUnreadCounter>();
        Assert.Equal(1, await unreadCounter.GetAsync(convId, ImServerFixture.BobId));

        // 监听已读广播（Alice 侧应收到 Bob 已读该消息的事件）
        // 用强类型载荷解析，避免 dynamic 取属性受 JSON 命名/类型差异影响
        string? readMessageId = null;
        Guid? readUserId = null;
        alice.On<HubEvents.MessageReadEvent>("OnMessageRead", payload =>
        {
            readMessageId = payload.MessageId;
            readUserId = payload.UserId;
        });

        // Act：Bob 标记已读
        await bob.InvokeAsync("MarkAsRead", new MarkAsReadRequest(convId, lastMessageId!));

        // Assert ①：Redis 未读清零
        await WaitForAsync(async () => await unreadCounter.GetAsync(convId, ImServerFixture.BobId) == 0);
        Assert.Equal(0, await unreadCounter.GetAsync(convId, ImServerFixture.BobId));

        // Assert ②：PG 未读清零、读位点更新
        var pgUnread = await ImTestSeeder.GetPgUnreadAsync(_fixture, convId, ImServerFixture.BobId);
        Assert.Equal(0, pgUnread);

        // Assert ③：已读事件广播给会话参与者，携带正确的消息 ID 与用户
        await WaitForAsync(() => readMessageId == lastMessageId);
        Assert.Equal(lastMessageId, readMessageId);
        Assert.Equal(ImServerFixture.BobId, readUserId);
    }

    [Fact]
    public async Task RecallMessage_marks_mongo_recalled_and_updates_conversation_summary()
    {
        // Arrange：私聊会话，Alice 发一条消息（成为最后一条），随后本人撤回
        _fixture.UserInfoService.SetUser(ImServerFixture.AliceId, "Alice");
        var convId = await ImTestSeeder.SeedPrivateConversationAsync(_fixture, ImServerFixture.AliceId, ImServerFixture.BobId);

        var alice = await ConnectAsync(ImServerFixture.AliceId, "Alice");

        string? messageId = null;
        alice.On<MessageDto>("OnMessageReceived", m => messageId = m.Id);

        await alice.InvokeAsync("SendMessage", new SendMessageRequest(
            convId, MessageContentTypeEnum.Text, "oops, recall me", null, null));
        await WaitForAsync(() => messageId is not null);

        // 监听撤回广播
        string? recalledId = null;
        alice.On<HubEvents.MessageRecalledEvent>("OnMessageRecalled", payload => recalledId = payload.MessageId);

        // Act：Alice 撤回自己的消息（在 2 分钟时间窗内）
        await alice.InvokeAsync("RecallMessage", new RecallMessageRequest(convId, messageId!));

        // Assert ①：Mongo 标记撤回（MongoDB.Entities 静态 DB.Default）
        var docs = await DB.Default.Find<MessageDocument>()
            .Match(m => m.ID == messageId!)
            .ExecuteAsync();
        var doc = docs.First();
        Assert.True(doc.IsRecalled);
        Assert.Equal(ImServerFixture.AliceId, doc.RecalledBy);
        Assert.NotNull(doc.RecalledAt);

        // Assert ②：会话列表摘要更新为撤回提示（因为撤回的恰是最后一条消息）
        await using (var scope2 = _fixture.Services.CreateAsyncScope())
        {
            var db = scope2.ServiceProvider.GetRequiredService<ImDbContext>();
            var conv = await db.Conversations.AsNoTracking().FirstAsync(c => c.Id == convId);
            Assert.Equal("[消息已撤回]", conv.LastMessageContent);
        }

        // Assert ③：撤回事件广播
        await WaitForAsync(() => recalledId == messageId);
        Assert.Equal(messageId, recalledId);
    }

    private static async Task WaitForAsync(Func<bool> condition, TimeSpan? timeout = null)
    {
        var deadline = DateTime.UtcNow + (timeout ?? TimeSpan.FromSeconds(5));
        while (DateTime.UtcNow < deadline)
        {
            if (condition()) return;
            await Task.Delay(25);
        }
        throw new TimeoutException("等待条件在超时内未满足");
    }

    /// <summary>异步条件重载：谓词返回 Task&lt;bool&gt;，用于轮询 Redis 等 IO 状态。</summary>
    private static async Task WaitForAsync(Func<Task<bool>> condition, TimeSpan? timeout = null)
    {
        var deadline = DateTime.UtcNow + (timeout ?? TimeSpan.FromSeconds(5));
        while (DateTime.UtcNow < deadline)
        {
            if (await condition()) return;
            await Task.Delay(25);
        }
        throw new TimeoutException("等待条件在超时内未满足");
    }
}