Skip to content

Commit ca177e1

Browse files
seanmcilvennaseanmcilvenna
authored
Merge branch 'dev' into LNK-4436
2 parents 2aad629 + 14808c8 commit ca177e1

68 files changed

Lines changed: 2183 additions & 737 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎DotNet/Admin.BFF/Application/Commands/Integration/CreatePatientListAcquired/CreatePatientListAcquired.cs‎

Lines changed: 84 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,84 @@
1+
using Confluent.Kafka;
2+
using Hl7.Fhir.Model;
3+
using LantanaGroup.Link.LinkAdmin.BFF.Application.Models.Integration;
4+
using LantanaGroup.Link.LinkAdmin.BFF.Infrastructure;
5+
using LantanaGroup.Link.LinkAdmin.BFF.Infrastructure.Logging;
6+
using LantanaGroup.Link.Shared.Application.Models;
7+
using LantanaGroup.Link.Shared.Application.Models.Kafka;
8+
using Newtonsoft.Json;
9+
using OpenTelemetry.Trace;
10+
using System.Diagnostics;
11+
using System.Text.Json;
12+
13+
namespace LantanaGroup.Link.LinkAdmin.BFF.Application.Commands.Integration
14+
{
15+
public class CreatePatientListAcquired : ICreatePatientListAcquired
16+
{
17+
private readonly ILogger<CreatePatientListAcquired> _logger;
18+
private readonly IProducer<string, PatientListMessage> _producer;
19+
20+
public CreatePatientListAcquired(ILogger<CreatePatientListAcquired> logger, IProducer<string, PatientListMessage> producer)
21+
{
22+
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
23+
_producer = producer ?? throw new ArgumentNullException(nameof(producer));
24+
}
25+
26+
public async Task<string> Execute(PatientListAcquired model, string? userId = null, CancellationToken cancellationToken = default)
27+
{
28+
using Activity? activity = ServiceActivitySource.Instance.StartActivity("Producing Patient Acquired Event");
29+
string correlationId = Guid.NewGuid().ToString();
30+
31+
try
32+
{
33+
var headers = new Headers
34+
{
35+
{ "X-Correlation-Id", System.Text.Encoding.ASCII.GetBytes(correlationId) }
36+
};
37+
38+
// create a list
39+
// var patientList = new List { };
40+
41+
// add entries to the patientList for each patient in model.PatientIds
42+
/* for (int i = 0; i < model.PatientIds.Count; i++)
43+
{
44+
var entry = new List.EntryComponent
45+
{
46+
Item = new ResourceReference("Patient/" + model.PatientIds[i]),
47+
};
48+
patientList.Entry.Add(entry);
49+
}*/
50+
51+
52+
// Output the FHIR List in JSON format
53+
//var json = JsonConvert.SerializeObject(model.PatientLists, Formatting.Indented);
54+
55+
// Convert the object to a nicely formatted JSON string
56+
/*var jsonValue = JsonSerializer.Serialize(model.PatientLists, new JsonSerializerOptions
57+
{
58+
WriteIndented = true // optional: pretty formatting
59+
});
60+
*/
61+
62+
var message = new Message<string, PatientListMessage>
63+
{
64+
Key = model.FacilityId,
65+
Value = new PatientListMessage { PatientLists = model.PatientLists, ReportTrackingId = model.ReportTrackingId},
66+
Headers = headers
67+
};
68+
69+
await _producer.ProduceAsync(nameof(KafkaTopic.PatientListsAcquired), message);
70+
71+
return correlationId;
72+
73+
}
74+
catch (Exception ex)
75+
{
76+
Activity.Current?.SetStatus(ActivityStatusCode.Error);
77+
Activity.Current?.RecordException(ex);
78+
_logger.LogKafkaProducerException(nameof(KafkaTopic.PatientListsAcquired), ex.Message);
79+
throw;
80+
}
81+
82+
}
83+
}
84+
}

‎DotNet/Admin.BFF/Application/Commands/Integration/CreatePatientListAcquired/ICreatePatientListAcquired.cs‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,10 @@
1+
using LantanaGroup.Link.LinkAdmin.BFF.Application.Models.Integration;
2+
using System.ComponentModel.DataAnnotations;
3+
4+
namespace LantanaGroup.Link.LinkAdmin.BFF.Application.Commands.Integration
5+
{
6+
public interface ICreatePatientListAcquired
7+
{
8+
Task<string> Execute([Required] PatientListAcquired model, string? userId = null, CancellationToken cancellationToken = default);
9+
}
10+
}

‎DotNet/Admin.BFF/Application/Interfaces/Models/IPatientListsAcquired.cs‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,20 @@
1+
using LantanaGroup.Link.Shared.Application.Models.Kafka;
2+
using System.ComponentModel.DataAnnotations;
3+
4+
namespace LantanaGroup.Link.LinkAdmin.BFF.Application.Interfaces.Models
5+
{
6+
public interface IPatientListsAcquired
7+
{
8+
/// <summary>
9+
/// The unique identifier of the facility
10+
/// </summary>
11+
[Required]
12+
string FacilityId { get; set; }
13+
14+
/// <summary>
15+
/// List of patient identifiers acquired from the facility
16+
/// </summary>
17+
[Required]
18+
List<PatientListItem> PatientLists { get; set; }
19+
}
20+
}

‎DotNet/Admin.BFF/Application/Models/Integration/PatientListAcquired.cs‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,31 @@
1+
using Hl7.Fhir.Model;
2+
using LantanaGroup.Link.LinkAdmin.BFF.Application.Interfaces.Models;
3+
using LantanaGroup.Link.Shared.Application.Models.Kafka;
4+
5+
namespace LantanaGroup.Link.LinkAdmin.BFF.Application.Models.Integration
6+
{
7+
public class PatientListAcquired : IPatientListsAcquired
8+
{
9+
/// <summary>
10+
/// Key for the patient event (FacilityId)
11+
/// </summary>
12+
/// <example>TestFacility01</example>
13+
public string FacilityId { get; set; } = string.Empty;
14+
15+
/// <summary>
16+
/// The id of the patient subject to the event
17+
/// </summary>
18+
/// <example>TestPatient01</example>
19+
public List<PatientListItem> PatientLists { get; set; } = new List<PatientListItem>();
20+
21+
public string ReportTrackingId { get; set; } = string.Empty;
22+
23+
}
24+
25+
26+
public class PatientListAcquiredMessage
27+
{
28+
public List<PatientListItem> PatientListItems { get; set; } = new List<PatientListItem>();
29+
public string ReportTrackingId { get; set; } = string.Empty;
30+
}
31+
}

‎DotNet/Admin.BFF/Presentation/Endpoints/IntegrationTestingEndpoints.cs‎

Lines changed: 27 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@
66
using LantanaGroup.Link.LinkAdmin.BFF.Application.Models.Integration;
77
using LantanaGroup.Link.LinkAdmin.BFF.Application.Models.Responses;
88
using LantanaGroup.Link.LinkAdmin.BFF.Infrastructure.Logging;
9+
using LantanaGroup.Link.Shared.Application.Models.Kafka;
910
using Link.Authorization.Infrastructure;
1011
using Link.Authorization.Policies;
1112
using Microsoft.Extensions.Options;
@@ -19,18 +20,20 @@ public class IntegrationTestingEndpoints : IApi
1920
private readonly ILogger<IntegrationTestingEndpoints> _logger;
2021
private readonly ICreatePatientEvent _createPatientEvent;
2122
private readonly ICreatePatientAcquired _createPatientAcquired;
23+
private readonly ICreatePatientListAcquired _createPatientListAcquired;
2224
private readonly ICreateReportScheduled _createReportScheduled;
2325
private readonly ICreateDataAcquisitionRequested _createDataAcquisitionRequested;
2426
private readonly KafkaConsumerManager _kafkaConsumerManager;
2527
private readonly IOptions<AuthenticationSchemaConfig> _authenticationSchemaConfig;
2628

27-
public IntegrationTestingEndpoints(ILogger<IntegrationTestingEndpoints> logger, IOptions<AuthenticationSchemaConfig> authenticationSchemaConfig, ICreatePatientEvent createPatientEvent, KafkaConsumerManager kafkaConsumerManager, ICreateReportScheduled createReportScheduled, ICreateDataAcquisitionRequested createDataAcquisitionRequested, ICreatePatientAcquired createPatientAcquired)
29+
public IntegrationTestingEndpoints(ILogger<IntegrationTestingEndpoints> logger, IOptions<AuthenticationSchemaConfig> authenticationSchemaConfig, ICreatePatientEvent createPatientEvent, KafkaConsumerManager kafkaConsumerManager, ICreateReportScheduled createReportScheduled, ICreateDataAcquisitionRequested createDataAcquisitionRequested, ICreatePatientAcquired createPatientAcquired, ICreatePatientListAcquired createPatientListAcquired)
2830
{
2931
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
3032
_createPatientEvent = createPatientEvent ?? throw new ArgumentNullException(nameof(createPatientEvent));
3133
_createReportScheduled = createReportScheduled ?? throw new ArgumentNullException(nameof(createReportScheduled));
3234
_createDataAcquisitionRequested = createDataAcquisitionRequested ?? throw new ArgumentNullException(nameof(createDataAcquisitionRequested));
3335
_createPatientAcquired = createPatientAcquired ?? throw new ArgumentNullException(nameof(createPatientAcquired));
36+
_createPatientListAcquired = createPatientListAcquired ?? throw new ArgumentNullException(nameof(createPatientListAcquired));
3437
_kafkaConsumerManager = kafkaConsumerManager ?? throw new ArgumentNullException(nameof(kafkaConsumerManager));
3538
_authenticationSchemaConfig = authenticationSchemaConfig ?? throw new ArgumentNullException(nameof(authenticationSchemaConfig));
3639
}
@@ -99,6 +102,17 @@ public void RegisterEndpoints(WebApplication app)
99102
Description = "Produces a new data acquisition requested event that will be sent to the broker. Allows for testing processes outside of scheduled events."
100103
});
101104

105+
integrationEndpoints.MapPost("/patient-list-acquired", CreatePatientListAcquired)
106+
.Produces<EventProducerResponse>(StatusCodes.Status200OK)
107+
.Produces<ValidationFailureResponse>(StatusCodes.Status400BadRequest)
108+
.Produces(StatusCodes.Status401Unauthorized)
109+
.ProducesProblem(StatusCodes.Status500InternalServerError)
110+
.WithOpenApi(x => new OpenApiOperation(x)
111+
{
112+
Summary = "Integration Testing - Produce Data Acquisition Requested Event",
113+
Description = "Produces a new data acquisition requested event that will be sent to the broker. Allows for testing processes outside of scheduled events."
114+
});
115+
102116

103117
integrationEndpoints.MapPost("/start-consumers", CreateConsumersRequested)
104118
.Produces<EventProducerResponse>(StatusCodes.Status200OK)
@@ -174,6 +188,18 @@ public async Task<IResult> CreatePatientAcquired(HttpContext context, PatientAcq
174188
});
175189
}
176190

191+
public async Task<IResult> CreatePatientListAcquired(HttpContext context,PatientListAcquired model)
192+
{
193+
var user = context.User;
194+
195+
var correlationId = await _createPatientListAcquired.Execute(model, user?.FindFirst(ClaimTypes.Email)?.Value);
196+
return Results.Ok(new EventProducerResponse
197+
{
198+
Id = correlationId,
199+
Message = $"The patient acquired was created succcessfully with a correlation id of '{correlationId}'."
200+
});
201+
}
202+
177203

178204
public async Task<IResult> CreatePatientEvent(HttpContext context, PatientEvent model)
179205
{

‎DotNet/Admin.BFF/Program.cs‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,7 @@
3838
using LantanaGroup.Link.Shared.Application.Extensions.Caching;
3939
using LantanaGroup.Link.Shared.Application.Health;
4040
using LantanaGroup.Link.Shared.Application.Models;
41+
using LantanaGroup.Link.Shared.Application.Models.Kafka;
4142

4243
var builder = WebApplication.CreateBuilder(args);
4344

@@ -99,6 +100,8 @@ static void RegisterServices(WebApplicationBuilder builder)
99100
// Add Kafka Producer Factories
100101
builder.Services.RegisterKafkaProducer<string, object>(kafkaConnection, new Confluent.Kafka.ProducerConfig { CompressionType = Confluent.Kafka.CompressionType.Zstd });
101102

103+
builder.Services.RegisterKafkaProducer<string, PatientListMessage>(kafkaConnection, new Confluent.Kafka.ProducerConfig { CompressionType = Confluent.Kafka.CompressionType.Zstd });
104+
102105
// Add fluent validation
103106
builder.Services.AddValidatorsFromAssemblyContaining(typeof(PatientEventValidator));
104107

@@ -111,8 +114,9 @@ static void RegisterServices(WebApplicationBuilder builder)
111114
//TODO: https://learn.microsoft.com/en-us/aspnet/core/security/data-protection/configuration/overview?view=aspnetcore-8.0
112115

113116
// Add commands
114-
builder.Services.AddTransient<ICreatePatientEvent, CreatePatientEvent>();
117+
builder.Services.AddTransient<ICreatePatientListAcquired, CreatePatientListAcquired>();
115118
builder.Services.AddTransient<ICreatePatientAcquired, CreatePatientAcquired>();
119+
builder.Services.AddTransient<ICreatePatientEvent, CreatePatientEvent>();
116120
builder.Services.AddTransient<ICreateReportScheduled, CreateReportScheduled>();
117121
builder.Services.AddTransient<ICreateDataAcquisitionRequested, CreateDataAcquisitionRequested>();
118122
builder.Services.AddTransient<IGetLinkAccount, GetLinkAccount>();

‎DotNet/Census/Application/Services/PatientListService.cs‎

Lines changed: 27 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -82,7 +82,6 @@ public async Task<List<IBaseResponse>> ProcessList(string facilityId, PatientLis
8282
var existingEvent =
8383
await _patientEventQueries.GetLatestEventByFacilityAndPatientId(facilityId, patientId,
8484
cancellationToken);
85-
string sharedCorrelationId = existingEvent?.CorrelationId ?? Guid.NewGuid().ToString();
8685

8786
bool shouldSkip = false;
8887
if (existingEvent != null)
@@ -91,11 +90,13 @@ await _patientEventQueries.GetLatestEventByFacilityAndPatientId(facilityId, pati
9190
if (skipProcessing.result)
9291
{
9392
_logger.LogInformation(
94-
"Skipping processing for patient {PatientId} in facility {FacilityId}. " +
95-
"Reason: Admit event processing is planned for future implementation. " +
96-
"Event type: {EventType}, List type: {ListType}",
97-
patientId, facilityId, "Admit", list.ListType);
98-
93+
"{SkipMessage} PatientId: {PatientId}, FacilityId: {FacilityId}, EventType: {EventType}, ListType: {ListType}",
94+
skipProcessing.message,
95+
patientId,
96+
facilityId,
97+
"Admit",
98+
list.ListType);
99+
99100
shouldSkip = true; // Mark for skipping but don't continue yet
100101
}
101102
}
@@ -105,6 +106,11 @@ await _patientEventQueries.GetLatestEventByFacilityAndPatientId(facilityId, pati
105106
continue; // Skip processing for this patient
106107
}
107108

109+
var sharedCorrelationId = existingEvent != null
110+
&& existingEvent.EventType == EventType.FHIRListDischarge
111+
&& list.ListType == ListType.Admit
112+
? Guid.NewGuid().ToString() : (existingEvent?.CorrelationId ?? Guid.NewGuid().ToString());
113+
108114
await EnsureAdmitEventExists(facilityId, patientId, sharedCorrelationId, list.ListType,
109115
existingEvent,
110116
cancellationToken);
@@ -185,13 +191,25 @@ private async Task EnsureAdmitEventExists(string facilityId, string patientId, s
185191
if (existingEvent == null && listType == ListType.Discharge)
186192
{
187193
//create and add an admit event
188-
var admitEvent =
189-
new FHIRListAdmitPayload(patientId, DateTime.UtcNow).CreatePatientEvent(facilityId, correlationId);
194+
var payload = new FHIRListAdmitPayload(patientId, DateTime.UtcNow);
195+
var admitEvent = payload.CreatePatientEvent(facilityId, correlationId);
196+
190197
try
191198
{
192199
await _patientEventManager.AddPatientEvent(admitEvent, cancellationToken);
193200
_logger.LogInformation("Added admit event for patient {patientId} in facility {facilityId}", patientId,
194201
facilityId);
202+
203+
PatientEncounter encounter =
204+
await _patientEncounterQueries.GetPatientEncounterByCorrelationIdAsync(correlationId,
205+
cancellationToken);
206+
207+
if (encounter == null)
208+
{
209+
var patientEncounter = payload.CreatePatientEncounter(facilityId, correlationId);
210+
encounter = await _patientEncounterManager.AddPatientEncounterAsync(patientEncounter,
211+
cancellationToken);
212+
}
195213
}
196214
catch (Exception ex)
197215
{
@@ -238,7 +256,7 @@ public async Task<List<IBaseResponse>> ProcessLists(string facilityId, List<Pati
238256
}
239257

240258
List<IBaseResponse> messages = new List<IBaseResponse>();
241-
foreach (var list in lists)
259+
foreach (var list in lists.OrderBy(x => x.ListType))
242260
{
243261
messages.AddRange(await ProcessList(facilityId, list, cancellationToken));
244262
}

0 commit comments

Comments
 (0)