vigilcare-clinical/VigilCareClinicalAPI.Tests/Helpers/RabbitMqTestHelper.cs

using FluentAssertions;
using Microsoft.Extensions.Options;
using RabbitMQ.Client;
using RabbitMQ.Client.Exceptions;
using System.Net.Http.Headers;
using System.Text;
using System.Text.Json;

public static class RabbitMqTestHelper
{
    private static readonly string[] NotificationQueues =
    [
        "alerts.paging.queue",
        "alerts.paging.dlq",
        "alerts.escalation.queue",
    ];

    /// <summary>
    /// Creates the configured virtual host and grants the test user full access.
    /// Integration tests use a dedicated vhost so a concurrently running dev API
    /// on the default "/" vhost cannot steal paging messages or recreate the DLQ TTL.
    /// </summary>
    public static async Task EnsureVirtualHostAsync(RabbitMqOptions o)
    {
        if (string.IsNullOrEmpty(o.VirtualHost) || o.VirtualHost == "/")
            return;

        using var http = new HttpClient();
        var credentials = Convert.ToBase64String(Encoding.UTF8.GetBytes($"{o.Username}:{o.Password}"));
        http.DefaultRequestHeaders.Authorization = new AuthenticationHeaderValue("Basic", credentials);

        var mgmtBase = $"http://{o.Host}:{o.Port + 10000}";
        var vhost = Uri.EscapeDataString(o.VirtualHost);

        using var create = await http.PutAsync($"{mgmtBase}/api/vhosts/{vhost}", null);
        create.EnsureSuccessStatusCode();

        var permissions = JsonSerializer.Serialize(new
        {
            configure = ".*",
            write = ".*",
            read = ".*",
        });
        using var grant = await http.PutAsync(
            $"{mgmtBase}/api/permissions/{vhost}/{Uri.EscapeDataString(o.Username)}",
            new StringContent(permissions, Encoding.UTF8, "application/json"));
        grant.EnsureSuccessStatusCode();
    }

    public static void PurgeNotificationQueues(IOptions<RabbitMqOptions> opts)
    {
        var o = opts.Value;
        using var connection = CreateConnection(o, "test-queue-purge");
        using var channel = connection.CreateModel();

        // x-message-ttl is immutable once the queue exists. A concurrently running dev API
        // (300000 ms) can recreate the DLQ after the test host provisions 5000 ms, leaving
        // escalation stuck until the 5-minute TTL expires — far beyond the test poll window.
        RecreatePagingDlq(channel, o.PagingAckTimeoutMs);

        foreach (var queue in new[] { "alerts.paging.queue", "alerts.escalation.queue" })
        {
            try
            {
                channel.QueuePurge(queue);
            }
            catch (OperationInterruptedException)
            {
                // Queue may not exist yet on a cold broker.
            }
        }
    }

    /// <summary>
    /// Waits for stale alert.generated Kafka messages to finish the paging → DLQ → escalation
    /// cycle. Purging alone is not enough: NotificationPublisher keeps republishing backlog
    /// after the purge, and prefetch=1 leaves one in-flight page ahead of the test alert.
    /// </summary>
    public static async Task WaitForNotificationPipelineIdleAsync(
        IOptions<RabbitMqOptions> opts,
        TimeSpan maxWait)
    {
        var o = opts.Value;
        // One full cycle: paging timeout + DLQ TTL + worker margin.
        var settle = TimeSpan.FromMilliseconds(o.PagingAckTimeoutMs * 2 + 10_000);
        var deadline = DateTimeOffset.UtcNow.Add(maxWait);
        DateTimeOffset? lastBusy = null;

        while (DateTimeOffset.UtcNow < deadline)
        {
            if (!AreReadyQueuesEmpty(o))
            {
                lastBusy = DateTimeOffset.UtcNow;
                await Task.Delay(500);
                continue;
            }

            lastBusy ??= DateTimeOffset.UtcNow;

            if (DateTimeOffset.UtcNow - lastBusy.Value >= settle)
                return;

            await Task.Delay(500);
        }
    }

    private static bool AreReadyQueuesEmpty(RabbitMqOptions o)
    {
        using var connection = CreateConnection(o, "test-queue-idle-check");
        using var channel = connection.CreateModel();

        foreach (var queue in NotificationQueues)
        {
            try
            {
                var stats = channel.QueueDeclarePassive(queue);
                if (stats.MessageCount > 0)
                    return false;
            }
            catch (OperationInterruptedException)
            {
                // Queue may not exist yet on a cold broker.
            }
        }

        return true;
    }

    private static ConnectionFactory CreateFactory(RabbitMqOptions o) =>
        RabbitMqConnectionFactory.Create(o);

    private static IConnection CreateConnection(RabbitMqOptions o, string name) =>
        CreateFactory(o).CreateConnection(name);

    private static void RecreatePagingDlq(IModel channel, int pagingAckTimeoutMs)
    {
        const string dlq = "alerts.paging.dlq";

        try
        {
            channel.QueueDelete(dlq, ifUnused: false, ifEmpty: false);
        }
        catch (OperationInterruptedException)
        {
            // Queue may not exist yet on a cold broker.
        }

        channel.ExchangeDeclare(
            RabbitMqTopologyProvisioner.Exchange,
            ExchangeType.Direct,
            durable: true);

        channel.QueueDeclare(
            queue: dlq,
            durable: true,
            exclusive: false,
            autoDelete: false,
            arguments: new Dictionary<string, object>
            {
                ["x-message-ttl"] = pagingAckTimeoutMs,
                ["x-dead-letter-exchange"] = RabbitMqTopologyProvisioner.Exchange,
                ["x-dead-letter-routing-key"] = RabbitMqTopologyProvisioner.EscalKey,
            });
    }
}