vigilcare-clinical/VigilCareClinicalAPI.Tests/NotificationPipelineTests.cs

using System.Net.Http.Json;
using System.Text.Json;
using FluentAssertions;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Options;
using Minio.DataModel.Args;
using Minio.Exceptions;
using StackExchange.Redis;

[Collection("Integration")]
public class NotificationPipelineTests : IAsyncLifetime
{
    private readonly ApiFixture _fixture;
    private readonly HttpClient _http;

    public NotificationPipelineTests(ApiFixture fixture)
    {
        _fixture = fixture;
        _http    = fixture.CreateClient();
        _http.AsNurse();
    }

    public async Task InitializeAsync()
    {
        using var scope = _fixture.Services.CreateScope();
        var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
        await DbResetHelper.ResetAsync(db);

        db.AlertThresholds.Add(new AlertThreshold
        {
            Id = Guid.NewGuid(),
            ObservationCode = "POTASSIUM_MEQ_L",
            DisplayName = "Serum Potassium",
            Unit = "mEq/L",
            CriticalLow = 2.5m,
            WarningLow = 3.5m,
            WarningHigh = 5.0m,
            CriticalHigh = 6.5m,
            CreatedAt = DateTimeOffset.UtcNow,
        });
        await db.SaveChangesAsync();

        var redis = scope.ServiceProvider.GetRequiredService<IConnectionMultiplexer>();
        await redis.GetDatabase(1).StringSetAsync(
            "threshold:POTASSIUM_MEQ_L",
            """{"ObservationCode":"POTASSIUM_MEQ_L","CriticalLow":2.5,"WarningLow":3.5,"WarningHigh":5.0,"CriticalHigh":6.5}""");

        RabbitMqTestHelper.PurgeNotificationQueues(
            scope.ServiceProvider.GetRequiredService<IOptions<RabbitMqOptions>>());
    }

    public Task DisposeAsync() => Task.CompletedTask;

    [Fact]
    public async Task DischargeEncounter_UploadsSummaryToMinIO()
    {
        var encounterId = await CreateActiveEncounterAsync();

        var resp = await _http.PatchAsJsonAsync(
            $"/api/v1/encounters/{encounterId}/status",
            new { status = "Discharged" });
        resp.EnsureSuccessStatusCode();

        // outbox relay → Kafka → notification-publisher → RabbitMQ → discharge worker → MinIO
        await Task.Delay(TimeSpan.FromSeconds(8));

        var mo     = _fixture.Services.GetRequiredService<IOptions<MinioOptions>>().Value;
        var client = MinioClientFactory.Build(mo);
        var objectKey = $"discharge-summaries/{encounterId}/summary.pdf";

        var exists = false;
        try
        {
            await client.StatObjectAsync(new StatObjectArgs()
                .WithBucket(mo.BucketName)
                .WithObject(objectKey));
            exists = true;
        }
        catch (ObjectNotFoundException) { }

        exists.Should().BeTrue($"MinIO object not found: {objectKey}");
    }

    [Fact]
    public async Task CriticalAlert_Unacknowledged_EscalatesAfterTimeout()
    {
        var encounterId = await CreateActiveEncounterAsync();
        var alertId     = await IngestCriticalPotassiumAsync(encounterId);

        // paging worker (≤6s) → DLQ (5s TTL) → escalation worker.
        // Poll instead of a fixed sleep: earlier tests may leave paging jobs queued
        // (prefetch=1), so wall-clock time varies across the full integration suite.
        var deadline = DateTimeOffset.UtcNow.AddSeconds(60);
        AlertStatus status;
        do
        {
            await Task.Delay(500);
            using var scope = _fixture.Services.CreateScope();
            var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
            status = (await db.ClinicalAlerts.FindAsync(alertId))!.Status;
        }
        while (status == AlertStatus.Open && DateTimeOffset.UtcNow < deadline);

        status.Should().Be(AlertStatus.Escalated);
    }

    [Fact]
    public async Task CriticalAlert_AcknowledgedBeforeTimeout_DoesNotEscalate()
    {
        var encounterId = await CreateActiveEncounterAsync();
        var alertId     = await IngestCriticalPotassiumAsync(encounterId);

        await Task.Delay(TimeSpan.FromSeconds(3));

        var ackResp = await _http.PostAsJsonAsync(
            $"/api/v1/alerts/{alertId}/acknowledge",
            new AcknowledgeAlertRequest("Reviewed — will adjust potassium replacement."));
        ackResp.EnsureSuccessStatusCode();

        await Task.Delay(TimeSpan.FromSeconds(8));

        using var scope = _fixture.Services.CreateScope();
        var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();

        var alert = await db.ClinicalAlerts.FindAsync(alertId);
        alert.Should().NotBeNull();
        alert!.Status.Should().Be(AlertStatus.Acknowledged);
    }

    private async Task<Guid> CreateActiveEncounterAsync()
    {
        var patientId = await CreatePatientAsync();

        var resp = await _http.PostAsJsonAsync(
            $"/api/v1/patients/{patientId}/encounters",
            new OpenEncounterRequest(EncounterType.Inpatient, Department.Icu, "Dr. Osei"));
        resp.EnsureSuccessStatusCode();

        var body = await resp.Content.ReadFromJsonAsync<JsonDocument>();
        return body!.RootElement.GetProperty("data").GetProperty("id").GetGuid();
    }

    private async Task<Guid> CreatePatientAsync()
    {
        var resp = await _http.PostAsJsonAsync(
            "/api/v1/patients",
            new RegisterPatientRequest("Eleanor", "Vance", new DateOnly(1962, 9, 14), "Female"));
        resp.EnsureSuccessStatusCode();

        var body = await resp.Content.ReadFromJsonAsync<JsonDocument>();
        return body!.RootElement.GetProperty("data").GetProperty("id").GetGuid();
    }

    private async Task<Guid> IngestCriticalPotassiumAsync(Guid encounterId)
    {
        var ingestResp = await _http.PostAsJsonAsync(
            $"/api/v1/encounters/{encounterId}/observations",
            new BatchIngestRequest(new List<IngestObservationRequest>
            {
                new("POTASSIUM_MEQ_L", 2.1m, "mEq/L", ObservationSource.Device, DateTimeOffset.UtcNow, null),
            }));
        ingestResp.EnsureSuccessStatusCode();

        var body = await ingestResp.Content.ReadFromJsonAsync<JsonDocument>();
        var alertId = body!.RootElement.GetProperty("data").GetProperty("alertId").GetGuid();
        alertId.Should().NotBe(Guid.Empty);
        return alertId;
    }
}