tiku-backend.net/Tiku.IntegrationTests/MassTransitOutboxTests.cs

using MassTransit;
using MassTransit.EntityFrameworkCoreIntegration;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using StackExchange.Redis;
using System.Net;
using System.Net.Http.Headers;
using System.Text.Json;
using Tiku.Application;
using Tiku.Contracts;
using Tiku.Infrastructure;
using Tiku.Infrastructure.Messaging;
using Tiku.Infrastructure.Persistence;

namespace Tiku.IntegrationTests;

public sealed class MassTransitOutboxTests
{
    [Fact]
    public async Task Worker_consumer_uses_inbox_and_updates_non_authoritative_redis_version()
    {
        var rabbitMqHost = Environment.GetEnvironmentVariable("TIKU_TEST_RABBITMQ");
        var redisConnection = Environment.GetEnvironmentVariable("TIKU_TEST_REDIS");
        if (string.IsNullOrWhiteSpace(rabbitMqHost) || string.IsNullOrWhiteSpace(redisConnection))
        {
            return;
        }

        await using var factory = CreateRabbitFactory(rabbitMqHost);
        using var client = factory.CreateClient();
        Assert.True(await WaitForReadyAsync(client), "API dependencies did not become ready within 10 seconds.");

        var redisEnvironment = $"consumer-{Guid.NewGuid():N}";
        var workerBuilder = Host.CreateApplicationBuilder();
        workerBuilder.Services.AddApplication();
        workerBuilder.Services.AddInfrastructure(factory.DatabaseConnectionString);
        workerBuilder.Services.AddRedisSecurity(redisConnection, redisEnvironment);
        workerBuilder.Services.AddReliableMessaging(CreateRabbitOptions(rabbitMqHost, configureConsumers: true));
        using var worker = workerBuilder.Build();
        await worker.StartAsync();

        var tenantId = Guid.NewGuid();
        var userId = Guid.NewGuid();
        var messageId = Guid.NewGuid();
        const long version = 123456789;
        using (var scope = factory.CreateSystemScope("Publish duplicate inbox test message"))
        {
            var dbContext = scope.ServiceProvider.GetRequiredService<TikuDbContext>();
            var publishEndpoint = scope.ServiceProvider.GetRequiredService<IPublishEndpoint>();
            await using var transaction = await dbContext.Database.BeginTransactionAsync();
            var message = new AuthorizationStateChangedV1(
                Guid.NewGuid(), tenantId, userId, "consumer_test", version,
                DateTimeOffset.UtcNow, Guid.NewGuid().ToString("N"));
            await publishEndpoint.Publish(message, context => context.MessageId = messageId);
            await publishEndpoint.Publish(message, context => context.MessageId = messageId);
            await dbContext.SaveChangesAsync();
            await transaction.CommitAsync();
        }

        var redisKey = $"tiku:{redisEnvironment}:auth-inv:authorization:{tenantId:N}:{userId:N}";
        var consumed = false;
        for (var attempt = 0; attempt < 60; attempt++)
        {
            var multiplexer = worker.Services.GetRequiredService<IConnectionMultiplexer>();
            if (await multiplexer.GetDatabase().StringGetAsync(redisKey) == version)
            {
                consumed = true;
                break;
            }
            await Task.Delay(250);
        }
        Assert.True(consumed, "Worker did not consume the security event within 15 seconds.");

        using (var verification = factory.CreateSystemScope("Verify duplicate consumer inbox"))
        {
            var dbContext = verification.ServiceProvider.GetRequiredService<TikuDbContext>();
            Assert.Single(await dbContext.Set<InboxState>()
                .Where(item => item.MessageId == messageId)
                .ToArrayAsync());
        }

        var managementEndpoint = ResolveRabbitManagementEndpoint(rabbitMqHost);
        if (managementEndpoint is not null)
        {
            Assert.Equal(0, await GetQueueMessageCountAsync(
                managementEndpoint, "security-state-changed_error"));
        }

        var redis = worker.Services.GetRequiredService<IConnectionMultiplexer>();
        await redis.GetDatabase().KeyDeleteAsync(redisKey);
        await worker.StopAsync();
    }

    [Fact]
    public async Task RabbitMq_health_and_bus_outbox_follow_database_transaction()
    {
        var rabbitMqHost = Environment.GetEnvironmentVariable("TIKU_TEST_RABBITMQ");
        if (string.IsNullOrWhiteSpace(rabbitMqHost))
        {
            return;
        }

        await using var factory = CreateRabbitFactory(rabbitMqHost);
        using var client = factory.CreateClient();
        Assert.True(await WaitForReadyAsync(client), "API dependencies did not become ready within 10 seconds.");
        var ready = await client.GetAsync("/api/health/ready");
        using var readyJson = JsonDocument.Parse(await ready.Content.ReadAsStringAsync());

        Assert.Equal(HttpStatusCode.OK, ready.StatusCode);
        Assert.True(readyJson.RootElement.GetProperty("rabbitMq").GetProperty("configured").GetBoolean());
        Assert.True(readyJson.RootElement.GetProperty("rabbitMq").GetProperty("ready").GetBoolean());

        using (var rollbackScope = factory.CreateSystemScope("Verify rolled back bus outbox"))
        {
            var dbContext = rollbackScope.ServiceProvider.GetRequiredService<TikuDbContext>();
            var publisher = rollbackScope.ServiceProvider.GetRequiredService<ISecurityEventPublisher>();
            await using var transaction = await dbContext.Database.BeginTransactionAsync();
            await publisher.AuthorizationChangedAsync(
                null, null, "rollback_test", 1, Guid.NewGuid().ToString("N"));
            await dbContext.SaveChangesAsync();
            await transaction.RollbackAsync();
        }

        using (var verification = factory.CreateSystemScope("Verify rolled back outbox is empty"))
        {
            var dbContext = verification.ServiceProvider.GetRequiredService<TikuDbContext>();
            Assert.Empty(await dbContext.Set<OutboxMessage>().ToArrayAsync());
        }

        using (var commitScope = factory.CreateSystemScope("Verify committed bus outbox"))
        {
            var dbContext = commitScope.ServiceProvider.GetRequiredService<TikuDbContext>();
            var publisher = commitScope.ServiceProvider.GetRequiredService<ISecurityEventPublisher>();
            await using var transaction = await dbContext.Database.BeginTransactionAsync();
            await publisher.AuthorizationChangedAsync(
                null, null, "commit_test", 2, Guid.NewGuid().ToString("N"));
            await dbContext.SaveChangesAsync();
            await transaction.CommitAsync();
        }

        var drained = false;
        for (var attempt = 0; attempt < 40; attempt++)
        {
            using var verification = factory.CreateSystemScope("Wait for committed outbox delivery");
            var dbContext = verification.ServiceProvider.GetRequiredService<TikuDbContext>();
            if (!await dbContext.Set<OutboxMessage>().AnyAsync())
            {
                drained = true;
                break;
            }
            await Task.Delay(250);
        }
        if (!drained)
        {
            using var diagnostics = factory.CreateSystemScope("Inspect undelivered outbox");
            var dbContext = diagnostics.ServiceProvider.GetRequiredService<TikuDbContext>();
            var messages = await dbContext.Set<OutboxMessage>().CountAsync();
            var states = await dbContext.Set<OutboxState>().CountAsync();
            Assert.Fail($"Committed MassTransit outbox was not delivered within 10 seconds. messages={messages}, states={states}");
        }
    }

    private static Api.ApiTestFactory CreateRabbitFactory(string rabbitMqHost) =>
        new(configurationOverrides: new Dictionary<string, string?>
        {
            ["RabbitMq:Host"] = rabbitMqHost,
            ["RabbitMq:Username"] = Environment.GetEnvironmentVariable("TIKU_TEST_RABBITMQ_USERNAME") ?? "guest",
            ["RabbitMq:Password"] = Environment.GetEnvironmentVariable("TIKU_TEST_RABBITMQ_PASSWORD") ?? "guest"
        });

    private static MessagingOptions CreateRabbitOptions(string rabbitMqHost, bool configureConsumers) => new()
    {
        Host = rabbitMqHost,
        Username = Environment.GetEnvironmentVariable("TIKU_TEST_RABBITMQ_USERNAME") ?? "guest",
        Password = Environment.GetEnvironmentVariable("TIKU_TEST_RABBITMQ_PASSWORD") ?? "guest",
        ConfigureConsumers = configureConsumers
    };

    private static async Task<bool> WaitForReadyAsync(HttpClient client)
    {
        for (var attempt = 0; attempt < 40; attempt++)
        {
            if ((await client.GetAsync("/api/health/ready")).StatusCode == HttpStatusCode.OK)
            {
                return true;
            }
            await Task.Delay(250);
        }
        return false;
    }

    private static Uri? ResolveRabbitManagementEndpoint(string rabbitMqHost)
    {
        var configured = Environment.GetEnvironmentVariable("TIKU_TEST_RABBITMQ_MANAGEMENT");
        if (!string.IsNullOrWhiteSpace(configured))
        {
            return new Uri(configured);
        }

        var broker = new Uri(rabbitMqHost);
        return broker.IsLoopback ? new Uri($"http://{broker.Host}:15672") : null;
    }

    private static async Task<int> GetQueueMessageCountAsync(Uri managementEndpoint, string queueName)
    {
        using var client = new HttpClient { BaseAddress = managementEndpoint };
        var username = Environment.GetEnvironmentVariable("TIKU_TEST_RABBITMQ_USERNAME") ?? "guest";
        var password = Environment.GetEnvironmentVariable("TIKU_TEST_RABBITMQ_PASSWORD") ?? "guest";
        var credentials = Convert.ToBase64String(System.Text.Encoding.UTF8.GetBytes($"{username}:{password}"));
        client.DefaultRequestHeaders.Authorization = new AuthenticationHeaderValue("Basic", credentials);
        using var response = await client.GetAsync($"/api/queues/%2F/{Uri.EscapeDataString(queueName)}");
        response.EnsureSuccessStatusCode();
        using var document = JsonDocument.Parse(await response.Content.ReadAsStringAsync());
        return document.RootElement.GetProperty("messages").GetInt32();
    }
}