341 lines
12 KiB
C#
341 lines
12 KiB
C#
using System.Net;
|
|
using System.Net.Http.Json;
|
|
using System.Text.Json;
|
|
using FluentAssertions;
|
|
using Microsoft.EntityFrameworkCore;
|
|
using Microsoft.Extensions.DependencyInjection;
|
|
using Microsoft.Extensions.Logging;
|
|
using StackExchange.Redis;
|
|
|
|
[Collection("Integration")]
|
|
public class ConcurrencyTests : IAsyncLifetime
|
|
{
|
|
private readonly ApiFixture _fixture;
|
|
private readonly HttpClient _http;
|
|
|
|
public ConcurrencyTests(ApiFixture fixture)
|
|
{
|
|
_fixture = fixture;
|
|
_http = fixture.CreateClient();
|
|
_http.ClearAuth();
|
|
_http.AsAdmin();
|
|
}
|
|
|
|
public async Task InitializeAsync()
|
|
{
|
|
using var scope = _fixture.Services.CreateScope();
|
|
var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
|
|
await DbResetHelper.ResetAsync(db);
|
|
|
|
var redis = scope.ServiceProvider.GetRequiredService<IConnectionMultiplexer>();
|
|
var server = redis.GetServer(redis.GetEndPoints().First());
|
|
await server.FlushDatabaseAsync(1);
|
|
}
|
|
|
|
public Task DisposeAsync() => Task.CompletedTask;
|
|
|
|
// -----------------------------------------------------------------------
|
|
// Parallel patient registration — each must get a unique MRN
|
|
// -----------------------------------------------------------------------
|
|
|
|
[Fact]
|
|
public async Task ParallelPatientRegistration_AllGetUniqueMrns()
|
|
{
|
|
const int parallelCount = 10;
|
|
|
|
var tasks = Enumerable.Range(0, parallelCount).Select(i =>
|
|
{
|
|
var client = _fixture.CreateClient();
|
|
client.ClearAuth();
|
|
client.AsAdmin();
|
|
return client.PostAsJsonAsync(
|
|
"/api/v1/patients",
|
|
new RegisterPatientRequest(
|
|
$"Parallel{i}", "Patient",
|
|
new DateOnly(1980, 1, 1).AddDays(i), "Female"));
|
|
});
|
|
|
|
var responses = await Task.WhenAll(tasks);
|
|
|
|
var patientIds = new List<Guid>();
|
|
foreach (var resp in responses)
|
|
{
|
|
resp.StatusCode.Should().Be(HttpStatusCode.Created);
|
|
var body = await resp.Content.ReadFromJsonAsync<JsonDocument>();
|
|
patientIds.Add(body!.RootElement.GetProperty("data").GetProperty("id").GetGuid());
|
|
}
|
|
|
|
patientIds.Should().OnlyHaveUniqueItems("every parallel registration must receive a distinct ID");
|
|
|
|
using var scope = _fixture.Services.CreateScope();
|
|
var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
|
|
|
|
var mrns = await db.Patients
|
|
.Where(p => patientIds.Contains(p.Id))
|
|
.Select(p => p.Mrn)
|
|
.ToListAsync();
|
|
|
|
mrns.Should().HaveCount(parallelCount);
|
|
mrns.Should().OnlyHaveUniqueItems("every parallel registration must receive a distinct MRN");
|
|
}
|
|
|
|
// -----------------------------------------------------------------------
|
|
// Parallel SOFA_SEPSIS alerts for same encounter — only one bundle
|
|
// -----------------------------------------------------------------------
|
|
|
|
[Fact]
|
|
public async Task ParallelSepsisAlerts_SingleBundleCreated()
|
|
{
|
|
var (patientId, encounterId) = await SeedPatientAndEncounterAsync();
|
|
const int parallelCount = 5;
|
|
|
|
using var scope = _fixture.Services.CreateScope();
|
|
var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
|
|
|
|
var alertIds = new List<Guid>();
|
|
for (var i = 0; i < parallelCount; i++)
|
|
{
|
|
var alertId = Guid.NewGuid();
|
|
db.ClinicalAlerts.Add(new ClinicalAlert
|
|
{
|
|
Id = alertId,
|
|
EncounterId = encounterId,
|
|
PatientId = patientId,
|
|
AlertType = AlertType.SofaSepsis,
|
|
Severity = AlertSeverity.Critical,
|
|
Details = $"SOFA delta +{i + 2}",
|
|
Status = AlertStatus.Open,
|
|
TriggeredAt = DateTimeOffset.UtcNow
|
|
});
|
|
alertIds.Add(alertId);
|
|
}
|
|
await db.SaveChangesAsync();
|
|
|
|
var tasks = alertIds.Select(alertId =>
|
|
{
|
|
return Task.Run(async () =>
|
|
{
|
|
using var innerScope = _fixture.Services.CreateScope();
|
|
var handler = innerScope.ServiceProvider.GetRequiredService<SepsisAlertHandler>();
|
|
await handler.OnSepsisAlertCreatedAsync(
|
|
encounterId, alertId, AlertType.SofaSepsis, CancellationToken.None);
|
|
});
|
|
});
|
|
|
|
await Task.WhenAll(tasks);
|
|
|
|
var bundleCount = await db.SepsisBundles
|
|
.AsNoTracking()
|
|
.CountAsync(b => b.EncounterId == encounterId);
|
|
|
|
bundleCount.Should().Be(1, "concurrent SOFA_SEPSIS alerts must produce exactly one bundle");
|
|
|
|
var orderCount = await db.Orders
|
|
.AsNoTracking()
|
|
.CountAsync(o => o.EncounterId == encounterId);
|
|
|
|
orderCount.Should().Be(4, "only one bundle's orders (4) should exist");
|
|
}
|
|
|
|
// -----------------------------------------------------------------------
|
|
// Parallel observation ingest with same idempotency key — single record
|
|
// -----------------------------------------------------------------------
|
|
|
|
[Fact]
|
|
public async Task ParallelObservationIngest_SameIdempotencyKey_SingleRecord()
|
|
{
|
|
var (_, encounterId) = await SeedPatientAndEncounterAsync();
|
|
await SeedThresholdsAsync();
|
|
|
|
const int parallelCount = 5;
|
|
var idempotencyKey = $"device-parallel-{Guid.NewGuid():N}";
|
|
|
|
var tasks = Enumerable.Range(0, parallelCount).Select(_ =>
|
|
{
|
|
var client = _fixture.CreateClient();
|
|
client.ClearAuth();
|
|
client.AsAdmin();
|
|
return client.PostAsJsonAsync(
|
|
$"/api/v1/encounters/{encounterId}/observations",
|
|
new BatchIngestRequest(new List<IngestObservationRequest>
|
|
{
|
|
new("HEART_RATE", 78, "bpm", ObservationSource.Device,
|
|
DateTimeOffset.UtcNow, idempotencyKey)
|
|
}));
|
|
});
|
|
|
|
var responses = await Task.WhenAll(tasks);
|
|
|
|
foreach (var resp in responses)
|
|
resp.StatusCode.Should().Be(HttpStatusCode.Created);
|
|
|
|
using var scope = _fixture.Services.CreateScope();
|
|
var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
|
|
|
|
var obsCount = await db.Observations
|
|
.CountAsync(o => o.EncounterId == encounterId);
|
|
|
|
obsCount.Should().Be(1, "parallel ingests with the same idempotency key must produce exactly one record");
|
|
}
|
|
|
|
// -----------------------------------------------------------------------
|
|
// Parallel observation ingest without idempotency key — all succeed
|
|
// -----------------------------------------------------------------------
|
|
|
|
[Fact]
|
|
public async Task ParallelObservationIngest_NoIdempotencyKey_AllCreateRecords()
|
|
{
|
|
var (_, encounterId) = await SeedPatientAndEncounterAsync();
|
|
await SeedThresholdsAsync();
|
|
|
|
const int parallelCount = 5;
|
|
|
|
var tasks = Enumerable.Range(0, parallelCount).Select(i =>
|
|
{
|
|
var client = _fixture.CreateClient();
|
|
client.ClearAuth();
|
|
client.AsAdmin();
|
|
return client.PostAsJsonAsync(
|
|
$"/api/v1/encounters/{encounterId}/observations",
|
|
new BatchIngestRequest(new List<IngestObservationRequest>
|
|
{
|
|
new("HEART_RATE", 70 + i, "bpm", ObservationSource.Device,
|
|
DateTimeOffset.UtcNow, null)
|
|
}));
|
|
});
|
|
|
|
var responses = await Task.WhenAll(tasks);
|
|
|
|
foreach (var resp in responses)
|
|
resp.StatusCode.Should().Be(HttpStatusCode.Created);
|
|
|
|
using var scope = _fixture.Services.CreateScope();
|
|
var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
|
|
|
|
var obsCount = await db.Observations
|
|
.CountAsync(o => o.EncounterId == encounterId);
|
|
|
|
obsCount.Should().Be(parallelCount,
|
|
"each observation without an idempotency key is distinct and must be stored");
|
|
}
|
|
|
|
// -----------------------------------------------------------------------
|
|
// Parallel encounter open — duplicate active encounter rejected
|
|
// -----------------------------------------------------------------------
|
|
|
|
[Fact]
|
|
public async Task ParallelEncounterOpen_SameType_OnlyOneSucceeds()
|
|
{
|
|
var patientId = await CreatePatientViaApiAsync();
|
|
const int parallelCount = 5;
|
|
|
|
var tasks = Enumerable.Range(0, parallelCount).Select(_ =>
|
|
{
|
|
var client = _fixture.CreateClient();
|
|
client.ClearAuth();
|
|
client.AsAdmin();
|
|
return client.PostAsJsonAsync(
|
|
$"/api/v1/patients/{patientId}/encounters",
|
|
new OpenEncounterRequest(EncounterType.Inpatient, Department.Icu, "Dr. Race"));
|
|
});
|
|
|
|
var responses = await Task.WhenAll(tasks);
|
|
|
|
var created = responses.Count(r => r.StatusCode == HttpStatusCode.Created);
|
|
var conflicts = responses.Count(r => r.StatusCode == HttpStatusCode.Conflict);
|
|
|
|
created.Should().BeGreaterThanOrEqualTo(1, "at least one encounter must be created");
|
|
(created + conflicts).Should().Be(parallelCount,
|
|
"all responses must be either Created or Conflict");
|
|
|
|
using var scope = _fixture.Services.CreateScope();
|
|
var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
|
|
|
|
var activeCount = await db.Encounters
|
|
.CountAsync(e => e.PatientId == patientId
|
|
&& e.EncounterType == EncounterType.Inpatient
|
|
&& e.Status == EncounterStatus.Active);
|
|
|
|
activeCount.Should().Be(1,
|
|
"only one active encounter of the same type should exist per patient");
|
|
}
|
|
|
|
// -----------------------------------------------------------------------
|
|
// Helpers
|
|
// -----------------------------------------------------------------------
|
|
|
|
private async Task<(Guid PatientId, Guid EncounterId)> SeedPatientAndEncounterAsync()
|
|
{
|
|
using var scope = _fixture.Services.CreateScope();
|
|
var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
|
|
|
|
var patient = new Patient
|
|
{
|
|
Id = Guid.NewGuid(),
|
|
Mrn = $"MRN-CONC-{Guid.NewGuid():N}"[..20],
|
|
FirstName = "Concurrent",
|
|
LastName = "Test",
|
|
DateOfBirth = new DateOnly(1975, 6, 15),
|
|
Gender = "M",
|
|
CreatedAt = DateTimeOffset.UtcNow
|
|
};
|
|
var encounter = new Encounter
|
|
{
|
|
Id = Guid.NewGuid(),
|
|
PatientId = patient.Id,
|
|
EncounterType = EncounterType.Inpatient,
|
|
Status = EncounterStatus.Active,
|
|
Department = Department.Icu,
|
|
AttendingPhysician = "Dr. Concurrent",
|
|
AdmittedAt = DateTimeOffset.UtcNow,
|
|
CreatedAt = DateTimeOffset.UtcNow
|
|
};
|
|
|
|
db.Patients.Add(patient);
|
|
db.Encounters.Add(encounter);
|
|
await db.SaveChangesAsync();
|
|
|
|
return (patient.Id, encounter.Id);
|
|
}
|
|
|
|
private async Task SeedThresholdsAsync()
|
|
{
|
|
using var scope = _fixture.Services.CreateScope();
|
|
var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
|
|
|
|
if (!await db.AlertThresholds.AnyAsync(t => t.ObservationCode == "HEART_RATE"))
|
|
{
|
|
db.AlertThresholds.Add(new AlertThreshold
|
|
{
|
|
Id = Guid.NewGuid(),
|
|
ObservationCode = "HEART_RATE",
|
|
DisplayName = "Heart Rate",
|
|
Unit = "bpm",
|
|
CriticalLow = 30,
|
|
WarningLow = 50,
|
|
WarningHigh = 100,
|
|
CriticalHigh = 150,
|
|
CreatedAt = DateTimeOffset.UtcNow
|
|
});
|
|
await db.SaveChangesAsync();
|
|
}
|
|
|
|
var redis = scope.ServiceProvider.GetRequiredService<IConnectionMultiplexer>();
|
|
await redis.GetDatabase(1).StringSetAsync(
|
|
"threshold:HEART_RATE",
|
|
"""{"ObservationCode":"HEART_RATE","CriticalLow":30,"WarningLow":50,"WarningHigh":100,"CriticalHigh":150}""");
|
|
}
|
|
|
|
private async Task<Guid> CreatePatientViaApiAsync()
|
|
{
|
|
var resp = await _http.PostAsJsonAsync(
|
|
"/api/v1/patients",
|
|
new RegisterPatientRequest("Race", "Condition",
|
|
new DateOnly(1985, 3, 20), "Male"));
|
|
resp.EnsureSuccessStatusCode();
|
|
|
|
var body = await resp.Content.ReadFromJsonAsync<JsonDocument>();
|
|
return body!.RootElement.GetProperty("data").GetProperty("id").GetGuid();
|
|
}
|
|
}
|