Root / src / Dpz.Core.MessageQueue.Test / Integration / MongoMessageOutboxStoreIntegrationTests.cs

using Dpz.Core.Entity.Base.MapperConfig;
using Dpz.Core.MessageQueue.Abstractions;
using Dpz.Core.MongodbAccess;
using Dpz.Core.Public.Entity;
using Dpz.Core.Service;
using Dpz.Core.Service.RepositoryService;
using Dpz.Core.Service.RepositoryServiceImpl;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging.Abstractions;
using MongoDB.Driver;
using MongoDB.Driver.Linq;

namespace Dpz.Core.MessageQueue.Test.Integration;

/// <summary>
/// 真实 MongoDB 集成测试，验证 <see cref="MongoMessageOutboxStore"/> 的状态机与查询语义。
/// 依赖 appsettings.Test.json 中 ConnectionStrings:mongodb 指向可用 Mongo（与其他 *.Test 项目共用）。
/// 每个用例使用独立 MessageId 前缀避免相互干扰，并在结束时清理。
/// </summary>
[Trait("Category", "Integration")]
[Collection(nameof(MongoOutboxTestCollection))]
public class MongoMessageOutboxStoreIntegrationTests(MongoOutboxFixture fixture) : IAsyncLifetime
{
    private readonly string _testRunId = $"test-{Guid.NewGuid():N}";

    public Task InitializeAsync() => Task.CompletedTask;

    public async Task DisposeAsync()
    {
        var filter = Builders<MessageOutboxRecord>.Filter.Where(x =>
            x.MessageId.StartsWith(_testRunId)
        );
        await fixture.Repository.Collection.DeleteManyAsync(filter);
    }

    [Fact]
    public async Task CreateAsync_ShouldPersistRecordWithPendingStatus()
    {
        var sut = fixture.BuildStore();
        var messageId = Id("create");

        await sut.CreateAsync(
            messageId,
            "Some.Test.Type",
            "ex.create",
            "rk.create",
            "{\"foo\":\"bar\"}",
            "src-create"
        );

        var record = await GetRecordAsync(messageId);
        Assert.NotNull(record);
        Assert.Equal("Some.Test.Type", record.MessageType);
        Assert.Equal("ex.create", record.Exchange);
        Assert.Equal("rk.create", record.RoutingKey);
        Assert.Equal("{\"foo\":\"bar\"}", record.Payload);
        Assert.Equal("src-create", record.Source);
        Assert.Equal(OutboxMessageStatus.Pending, record.Status);
        Assert.Equal(0, record.PublishAttempts);
        Assert.Null(record.SentAt);
        Assert.Null(record.LastPublishAttemptAt);
    }

    [Fact]
    public async Task MarkSentAsync_ShouldFlipStatusToSentAndStampSentAt()
    {
        var sut = fixture.BuildStore();
        var messageId = Id("sent");
        await sut.CreateAsync(messageId, "T", "ex", "rk", "{}", null);

        await sut.MarkSentAsync(messageId);

        var record = await GetRecordAsync(messageId);
        Assert.NotNull(record);
        Assert.Equal(OutboxMessageStatus.Sent, record.Status);
        Assert.Equal(1, record.PublishAttempts);
        Assert.NotNull(record.LastPublishAttemptAt);
        Assert.NotNull(record.SentAt);
    }

    [Fact]
    public async Task MarkPublishFailedAsync_ShouldRecordAttemptsAndScheduleNextRetry()
    {
        var sut = fixture.BuildStore();
        var messageId = Id("publish-failed");
        await sut.CreateAsync(messageId, "T", "ex", "rk", "{}", null);

        await sut.MarkPublishFailedAsync(messageId, "boom-1");

        var record = await GetRecordAsync(messageId);
        Assert.NotNull(record);
        Assert.Equal(OutboxMessageStatus.PublishFailed, record.Status);
        Assert.Equal(1, record.PublishAttempts);
        Assert.Equal("boom-1", record.LastPublishError);
        Assert.NotNull(record.LastPublishAttemptAt);
        Assert.NotNull(record.NextPublishRetryAt);

        // 第一次失败：退避 = 2^1 = 2 分钟
        var expectedDelay = TimeSpan.FromMinutes(2);
        var actualDelay = record.NextPublishRetryAt!.Value - record.LastPublishAttemptAt!.Value;
        Assert.InRange(
            actualDelay,
            expectedDelay - TimeSpan.FromSeconds(2),
            expectedDelay + TimeSpan.FromSeconds(2)
        );

        // 第二次失败：attempts=2，退避 = 2^2 = 4 分钟，PublishAttempts 累加
        await sut.MarkPublishFailedAsync(messageId, "boom-2");
        record = await GetRecordAsync(messageId);
        Assert.Equal(2, record!.PublishAttempts);
        Assert.Equal("boom-2", record.LastPublishError);
        Assert.Equal(OutboxMessageStatus.PublishFailed, record.Status);
    }

    [Fact]
    public async Task MarkPublishFailedAsync_ShouldNotThrow_WhenRecordMissing()
    {
        var sut = fixture.BuildStore();
        // 未先 Create 直接 MarkFail - 不应抛异常，仅警告日志（实现内会 return 早出）
        await sut.MarkPublishFailedAsync(Id("missing"), "no record");
    }

    [Fact]
    public async Task MarkConsumedAsync_ShouldFlipStatusToConsumed()
    {
        var sut = fixture.BuildStore();
        var messageId = Id("consumed");
        await sut.CreateAsync(messageId, "T", "ex", "rk", "{}", null);
        await sut.MarkSentAsync(messageId);

        await sut.MarkConsumedAsync(messageId);

        var record = await GetRecordAsync(messageId);
        Assert.Equal(OutboxMessageStatus.Consumed, record!.Status);
        Assert.Equal(1, record.ConsumeAttempts);
        Assert.NotNull(record.LastConsumeAttemptAt);
        Assert.NotNull(record.ConsumedAt);
    }

    [Fact]
    public async Task RecordConsumeAttemptFailedAsync_ShouldKeepStatusAndAccumulateAttempts()
    {
        var sut = fixture.BuildStore();
        var messageId = Id("consume-attempt-failed");
        await sut.CreateAsync(messageId, "T", "ex", "rk", "{}", null);
        await sut.MarkSentAsync(messageId);

        await sut.RecordConsumeAttemptFailedAsync(messageId, "consume err 1");

        var record = await GetRecordAsync(messageId);
        Assert.Equal(OutboxMessageStatus.Sent, record!.Status);
        Assert.Equal(1, record.ConsumeAttempts);
        Assert.Equal("consume err 1", record.LastConsumeError);
        Assert.NotNull(record.LastConsumeAttemptAt);
        Assert.Null(record.NextConsumeRetryAt);

        await sut.RecordConsumeAttemptFailedAsync(messageId, "consume err 2");
        record = await GetRecordAsync(messageId);
        Assert.Equal(OutboxMessageStatus.Sent, record!.Status);
        Assert.Equal(2, record.ConsumeAttempts);
        Assert.Equal("consume err 2", record.LastConsumeError);
    }

    [Fact]
    public async Task MarkConsumeFailedAsync_ShouldAccumulateAttemptsWithBackoff()
    {
        var sut = fixture.BuildStore();
        var messageId = Id("consume-failed");
        await sut.CreateAsync(messageId, "T", "ex", "rk", "{}", null);

        await sut.MarkConsumeFailedAsync(messageId, "consume err 1");
        var record = await GetRecordAsync(messageId);
        Assert.Equal(OutboxMessageStatus.ConsumeFailed, record!.Status);
        Assert.Equal(1, record.ConsumeAttempts);
        Assert.Equal("consume err 1", record.LastConsumeError);
        Assert.NotNull(record.NextConsumeRetryAt);

        await sut.MarkConsumeFailedAsync(messageId, "consume err 2");
        record = await GetRecordAsync(messageId);
        Assert.Equal(2, record!.ConsumeAttempts);
        Assert.Equal("consume err 2", record.LastConsumeError);
    }

    [Fact]
    public async Task GetPendingPublishRetryAsync_ShouldReturnDueFailedRecordsOnly()
    {
        var sut = fixture.BuildStore();

        var dueId = Id("pub-due");
        var notDueId = Id("pub-not-due");
        var sentId = Id("pub-sent");

        await sut.CreateAsync(dueId, "T", "ex.due", "rk.due", "{\"a\":1}", null);
        await sut.CreateAsync(notDueId, "T", "ex.nd", "rk.nd", "{}", null);
        await sut.CreateAsync(sentId, "T", "ex.sent", "rk.sent", "{}", null);

        // dueId: 标记为发布失败 + 强制让 NextPublishRetryAt 已到期
        await sut.MarkPublishFailedAsync(dueId, "err");
        await ForcePublishRetryAtAsync(dueId, DateTime.Now.AddMinutes(-1));

        // notDueId: 标记失败但保持默认 NextPublishRetryAt（未来）
        await sut.MarkPublishFailedAsync(notDueId, "err");

        // sentId: 已发送 - 不应被返回
        await sut.MarkSentAsync(sentId);

        var entries = await sut.GetPendingPublishRetryAsync(50);

        var entryIds = entries.Select(x => x.MessageId).ToList();
        Assert.Contains(dueId, entryIds);
        Assert.DoesNotContain(notDueId, entryIds);
        Assert.DoesNotContain(sentId, entryIds);

        var entry = entries.Single(x => x.MessageId == dueId);
        Assert.Equal("ex.due", entry.Exchange);
        Assert.Equal("rk.due", entry.RoutingKey);
        Assert.Equal("{\"a\":1}", entry.Payload);
        Assert.Equal(1, entry.PublishAttempts);
    }

    [Fact]
    public async Task GetPendingConsumeRetryAsync_ShouldReturnDueFailedRecordsOnly()
    {
        var sut = fixture.BuildStore();

        var dueId = Id("con-due");
        var notDueId = Id("con-not-due");

        await sut.CreateAsync(dueId, "T", "ex.cdue", "rk.cdue", "{}", null);
        await sut.CreateAsync(notDueId, "T", "ex.cnd", "rk.cnd", "{}", null);

        await sut.MarkConsumeFailedAsync(dueId, "err");
        await ForceConsumeRetryAtAsync(dueId, DateTime.Now.AddMinutes(-1));

        await sut.MarkConsumeFailedAsync(notDueId, "err");

        var entries = await sut.GetPendingConsumeRetryAsync(50);

        Assert.Contains(entries, x => x.MessageId == dueId);
        Assert.DoesNotContain(entries, x => x.MessageId == notDueId);
    }

    [Fact]
    public async Task GetPendingPublishRetryAsync_ShouldRespectBatchSize()
    {
        var sut = fixture.BuildStore();

        for (var i = 0; i < 5; i++)
        {
            var id = Id($"batch-{i}");
            await sut.CreateAsync(id, "T", "ex", "rk", "{}", null);
            await sut.MarkPublishFailedAsync(id, "err");
            await ForcePublishRetryAtAsync(id, DateTime.Now.AddMinutes(-1));
        }

        var top2 = await sut.GetPendingPublishRetryAsync(2);

        // batchSize 限制返回数量
        Assert.True(top2.Count <= 2);
    }

    private string Id(string suffix) => $"{_testRunId}-{suffix}";

    private Task<MessageOutboxRecord?> GetRecordAsync(string messageId) =>
        fixture.Repository.SearchFor(x => x.MessageId == messageId).FirstOrDefaultAsync()!;

    private Task ForcePublishRetryAtAsync(string messageId, DateTime when) =>
        fixture.Repository.UpdateAsync(
            x => x.MessageId == messageId,
            Builders<MessageOutboxRecord>.Update.Set(x => x.NextPublishRetryAt, when)
        );

    private Task ForceConsumeRetryAtAsync(string messageId, DateTime when) =>
        fixture.Repository.UpdateAsync(
            x => x.MessageId == messageId,
            Builders<MessageOutboxRecord>.Update.Set(x => x.NextConsumeRetryAt, when)
        );
}

[CollectionDefinition(nameof(MongoOutboxTestCollection))]
public class MongoOutboxTestCollection : ICollectionFixture<MongoOutboxFixture> { }

public class MongoOutboxFixture : IDisposable
{
    private readonly ServiceProvider _provider;
    private readonly IServiceScope _rootScope;

    public MongoOutboxFixture()
    {
        var configuration = new ConfigurationBuilder()
            .AddJsonFile("appsettings.Test.json", optional: true)
            .AddEnvironmentVariables()
            .Build();

        var connection = configuration.GetConnectionString("mongodb");
        if (string.IsNullOrWhiteSpace(connection))
        {
            throw new InvalidOperationException(
                "MongoOutboxFixture 初始化失败：appsettings.Test.json 缺少 ConnectionStrings:mongodb"
            );
        }

        var services = new ServiceCollection();
        services.AddSingleton<IConfiguration>(configuration);
        services.AddMemoryCache();
        services.AddFusionCache();
        services.AddSingleton(GlobalConfigMapper.GetTypeAdapterConfig());
        services.AddSingleton(GlobalConfigMapper.GetMapper());
        services.AddScoped(typeof(IRepository<>), typeof(Repository<>));
        services.AddDefaultServices();
        services.AddSingleton<Microsoft.Extensions.Logging.ILogger<MongoMessageOutboxStore>>(
            NullLogger<MongoMessageOutboxStore>.Instance
        );
        _provider = services.BuildServiceProvider();

        // 用根作用域获取一个仓储实例用于测试外部读写（清理 + 强制刷新 NextRetryAt）。
        // BuildStore() 走每次新作用域，模拟生产侧 ScopedOutboxStoreAdapter 的语义。
        _rootScope = _provider.CreateScope();
        Repository = _rootScope.ServiceProvider.GetRequiredService<
            IRepository<MessageOutboxRecord>
        >();
        // 主动触发一次 ping，及早暴露连接问题
        Repository.Database.RunCommand<MongoDB.Bson.BsonDocument>("{ping:1}");
    }

    public IRepository<MessageOutboxRecord> Repository { get; }

    public IMessageOutboxStore BuildStore()
    {
        // 每次构建都打开独立 scope，与生产 ScopedOutboxStoreAdapter 行为对齐
        var scope = _provider.CreateScope();
        // 注意：scope 在 fixture 销毁前不会显式释放，但 ServiceProvider.Dispose 会兜底清理
        return scope.ServiceProvider.GetRequiredService<IMongoMessageOutboxStore>();
    }

    public void Dispose()
    {
        _rootScope.Dispose();
        _provider.Dispose();
        GC.SuppressFinalize(this);
    }
}

⚠⚠ 以下内容为AI分析的结果，请根据实际情况进行判断。

代码解释

这是一个针对 MongoDB 消息发件箱（Message Outbox）存储 的集成测试类，用于验证 MongoMessageOutboxStore 的完整生命周期和状态机转换逻辑。

核心组件

1. `MongoMessageOutboxStoreIntegrationTests` （主测试类）

特性

[Trait("Category", "Integration")]：标记为集成测试（需真实 MongoDB 环境）
[Collection]：使用 xUnit 的共享 Fixture 机制，避免重复初始化数据库连接
IAsyncLifetime：支持异步初始化和清理

测试隔离机制

private readonly string _testRunId = $"test-{Guid.NewGuid():N}";

每次测试运行生成唯一 ID 前缀
所有消息 ID 使用该前缀（如 test-abc123-create）
测试结束时通过 DisposeAsync() 批量删除该批次的数据，避免污染数据库

核心测试场景

2. 创建消息（CreateAsync）

[Fact]
public async Task CreateAsync_ShouldPersistRecordWithPendingStatus()

验证：

消息记录正确持久化到 MongoDB
初始状态为 Pending
字段完整性（类型、交换机、路由键、负载、来源等）
发布尝试次数为 0
时间戳字段（SentAt、LastPublishAttemptAt）为 null

3. 标记已发送（MarkSentAsync）

[Fact]
public async Task MarkSentAsync_ShouldFlipStatusToSentAndStampSentAt()

验证：

状态从 Pending 转为 Sent
PublishAttempts 增加到 1
SentAt 和 LastPublishAttemptAt 被正确设置

4. 发布失败（MarkPublishFailedAsync）

[Fact]
public async Task MarkPublishFailedAsync_ShouldRecordAttemptsAndScheduleNextRetry()

验证重试机制：

状态变为 PublishFailed
指数退避算法：
- 第 1 次失败：退避 2^1 = 2 分钟
- 第 2 次失败：退避 2^2 = 4 分钟
记录错误信息和尝试次数
NextPublishRetryAt 被正确计算

5. 消费失败（MarkConsumeFailedAsync）

[Fact]
public async Task MarkConsumeFailedAsync_ShouldAccumulateAttemptsWithBackoff()

验证消费侧的重试逻辑（与发布侧对称）：

状态变为 ConsumeFailed
记录消费尝试次数和错误
NextConsumeRetryAt 使用退避算法

6. 查询待重试消息

[Fact]
public async Task GetPendingPublishRetryAsync_ShouldReturnDueFailedRecordsOnly()

验证查询语义：

只返回 PublishFailed 且 NextPublishRetryAt ≤ 当前时间 的记录
排除已发送（Sent）或未到期的记录
返回数据包含必要字段（Exchange、RoutingKey、Payload、Attempts）

批量限制：

[Fact]
public async Task GetPendingPublishRetryAsync_ShouldRespectBatchSize()

验证批量获取时 batchSize 参数生效（如只取前 2 条）

7. 边界情况

[Fact]
public async Task MarkPublishFailedAsync_ShouldNotThrow_WhenRecordMissing()

验证对不存在的消息 ID 调用方法时：

不抛出异常（优雅降级）
应该记录警告日志（代码注释提到）

测试基础设施

8. MongoOutboxFixture（共享测试上下文）

初始化流程

public MongoOutboxFixture()
{
    // 1. 加载配置（appsettings.Test.json）
    var configuration = new ConfigurationBuilder()
        .AddJsonFile("appsettings.Test.json", optional: true)
        .Build();

    // 2. 验证连接字符串
    var connection = configuration.GetConnectionString("mongodb");
    if (string.IsNullOrWhiteSpace(connection))
        throw new InvalidOperationException("缺少 MongoDB 连接配置");

    // 3. 注册依赖（仓储、缓存、日志等）
    services.AddScoped(typeof(IRepository<>), typeof(Repository<>));
    services.AddDefaultServices();

    // 4. 测试连接
    Repository.Database.RunCommand<BsonDocument>("{ping:1}");
}

关键方法

public IMessageOutboxStore BuildStore()
{
    // 每次测试创建独立作用域（Scope）
    var scope = _provider.CreateScope();
    return scope.ServiceProvider.GetRequiredService<IMongoMessageOutboxStore>();
}

为什么每次新建 Scope：模拟生产环境中 ScopedOutboxStoreAdapter 的行为（每个请求有独立 DbContext）

辅助方法

9. 直接操作数据库

private Task ForcePublishRetryAtAsync(string messageId, DateTime when) =>
    fixture.Repository.UpdateAsync(
        x => x.MessageId == messageId,
        Builders<MessageOutboxRecord>.Update.Set(x => x.NextPublishRetryAt, when)
    );

用于强制修改重试时间（模拟"已到期"场景），绕过业务逻辑直接操作数据。

设计亮点

真实环境测试
使用真实 MongoDB（而非 Mock），验证 LINQ 查询、索引性能和并发安全性
测试隔离
通过 GUID 前缀和自动清理，确保测试之间无副作用
状态机覆盖
完整测试消息从 Pending → Sent/PublishFailed → Consumed/ConsumeFailed 的所有转换路径

时间精度验证

Assert.InRange(actualDelay, 
    expectedDelay - TimeSpan.FromSeconds(2), 
    expectedDelay + TimeSpan.FromSeconds(2));

允许 ±2 秒误差（适应实际执行时的时间偏移）

生产对齐
通过 Scope 管理和共享 Fixture，模拟生产环境的依赖注入和资源生命周期

使用场景

CI/CD 流水线：在集成阶段验证与真实数据库的交互
回归测试：确保状态机逻辑变更不破坏现有行为
性能基准：通过批量测试验证查询效率和索引效果

评论加载中...