Skip to content

Commit 6a4be58

Browse files
edward-miller-lcgedward-miller-lcg
committed
Merge branch 'dev' into LNK-4303-add-event-date
2 parents bed9c5a + 53bdfa6 commit 6a4be58

64 files changed

Lines changed: 2125 additions & 728 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

@@ -108,6 +109,8 @@ static void RegisterServices(WebApplicationBuilder builder)
108109
// Add Kafka Producer Factories
109110
builder.Services.RegisterKafkaProducer<string, object>(kafkaConnection, new Confluent.Kafka.ProducerConfig { CompressionType = Confluent.Kafka.CompressionType.Zstd });
110111

112+
builder.Services.RegisterKafkaProducer<string, PatientListMessage>(kafkaConnection, new Confluent.Kafka.ProducerConfig { CompressionType = Confluent.Kafka.CompressionType.Zstd });
113+
111114
// Add fluent validation
112115
builder.Services.AddValidatorsFromAssemblyContaining(typeof(PatientEventValidator));
113116

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

122125
// Add commands
123-
builder.Services.AddTransient<ICreatePatientEvent, CreatePatientEvent>();
126+
builder.Services.AddTransient<ICreatePatientListAcquired, CreatePatientListAcquired>();
124127
builder.Services.AddTransient<ICreatePatientAcquired, CreatePatientAcquired>();
128+
builder.Services.AddTransient<ICreatePatientEvent, CreatePatientEvent>();
125129
builder.Services.AddTransient<ICreateReportScheduled, CreateReportScheduled>();
126130
builder.Services.AddTransient<ICreateDataAcquisitionRequested, CreateDataAcquisitionRequested>();
127131
builder.Services.AddTransient<IGetLinkAccount, GetLinkAccount>();

‎DotNet/Census/Listeners/PatientListsAcquiredListener.cs‎

Lines changed: 20 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -17,19 +17,19 @@ namespace LantanaGroup.Link.Census.Listeners;
1717

1818
public class PatientListsAcquiredListener : BackgroundService
1919
{
20-
private readonly IKafkaConsumerFactory<string, List<PatientListItem>> _kafkaConsumerFactory;
20+
private readonly IKafkaConsumerFactory<string, PatientListMessage> _kafkaConsumerFactory;
2121
private readonly ILogger<PatientListsAcquiredListener> _logger;
22-
private readonly IDeadLetterExceptionHandler<string, List<PatientListItem>> _nonTransientExceptionHandler;
23-
private readonly ITransientExceptionHandler<string, List<PatientListItem>> _transientExceptionHandler;
22+
private readonly IDeadLetterExceptionHandler<string, PatientListMessage> _nonTransientExceptionHandler;
23+
private readonly ITransientExceptionHandler<string, PatientListMessage> _transientExceptionHandler;
2424
private readonly IServiceScopeFactory _scopeFactory;
2525
private readonly IEventProducerService<PatientEvent> _eventProducerService;
2626

2727
public PatientListsAcquiredListener(
2828
ILogger<PatientListsAcquiredListener> logger,
29-
IKafkaConsumerFactory<string, List<PatientListItem>> kafkaConsumerFactory,
29+
IKafkaConsumerFactory<string, PatientListMessage> kafkaConsumerFactory,
3030
IProducer<string, object> kafkaProducer,
31-
IDeadLetterExceptionHandler<string, List<PatientListItem>> nonTransientExceptionHandler,
32-
ITransientExceptionHandler<string, List<PatientListItem>> transientExceptionHandler,
31+
IDeadLetterExceptionHandler<string, PatientListMessage> nonTransientExceptionHandler,
32+
ITransientExceptionHandler<string, PatientListMessage> transientExceptionHandler,
3333
IServiceScopeFactory scopeFactory,
3434
IEventProducerService<PatientEvent> eventProducerService
3535
)
@@ -70,7 +70,7 @@ private async Task StartConsumerLoop(CancellationToken cancellationToken)
7070

7171
IEnumerable<IBaseResponse>? responseMessages = null;
7272
kafkaConsumer.Subscribe(KafkaTopic.PatientListsAcquired.ToString());
73-
ConsumeResult<string, List<PatientListItem>>? rawmessage = null;
73+
ConsumeResult<string, PatientListMessage>? rawmessage = null;
7474

7575
using var scope = _scopeFactory.CreateScope();
7676

@@ -80,7 +80,7 @@ private async Task StartConsumerLoop(CancellationToken cancellationToken)
8080
{
8181
try
8282
{
83-
await kafkaConsumer.ConsumeWithInstrumentation((Func<ConsumeResult<string, List<PatientListItem>>?, CancellationToken, Task>)(async (result, CancellationToken) =>
83+
await kafkaConsumer.ConsumeWithInstrumentation((Func<ConsumeResult<string, PatientListMessage>?, CancellationToken, Task>)(async (result, CancellationToken) =>
8484
{
8585
rawmessage = result;
8686

@@ -111,7 +111,18 @@ await kafkaConsumer.ConsumeWithInstrumentation((Func<ConsumeResult<string, List<
111111
try
112112
{
113113
var patientListService = scope.ServiceProvider.GetRequiredService<IPatientListService>();
114-
responseMessages = await patientListService.ProcessLists(facilityId, rawmessage.Message.Value, cancellationToken);
114+
responseMessages = await patientListService.ProcessLists(facilityId, rawmessage.Message.Value.PatientLists, cancellationToken);
115+
116+
// Inject reportTrackingId into each PatientEvent
117+
responseMessages = responseMessages.Select(resp =>
118+
{
119+
if (resp is PatientEventResponse per && per.PatientEvent != null)
120+
{
121+
per.PatientEvent.ReportTrackingId = rawmessage.Message.Value.ReportTrackingId;
122+
}
123+
return resp;
124+
}).ToList();
125+
115126

116127
if (responseMessages == null || !responseMessages.Any())
117128
{

‎DotNet/Census/Program.cs‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -143,9 +143,9 @@ static void RegisterServices(WebApplicationBuilder builder)
143143

144144
// Add Kafka consumers and producers
145145
builder.Services.AddTransient<IKafkaConsumerFactory<string, string>, KafkaConsumerFactory<string, string>>();
146-
builder.Services.AddTransient<IKafkaConsumerFactory<string, List<PatientListItem>>, KafkaConsumerFactory<string, List<PatientListItem>>>();
146+
builder.Services.AddTransient<IKafkaConsumerFactory<string, PatientListMessage>, KafkaConsumerFactory<string, PatientListMessage>>();
147147
builder.Services.AddTransient<IKafkaProducerFactory<string, string>, KafkaProducerFactory<string, string>>();
148-
builder.Services.AddTransient<IKafkaProducerFactory<string, List<PatientListItem>>, KafkaProducerFactory<string, List<PatientListItem>>>();
148+
builder.Services.AddTransient<IKafkaProducerFactory<string, PatientListMessage>, KafkaProducerFactory<string, PatientListMessage>>();
149149
builder.Services.AddTransient<IKafkaProducerFactory<string, object>, KafkaProducerFactory<string, object>>();
150150
builder.Services.AddTransient<IKafkaProducerFactory<string, AuditEventMessage>, KafkaProducerFactory<string, AuditEventMessage>>();
151151
builder.Services.AddTransient<IKafkaProducerFactory<string, LantanaGroup.Link.Census.Application.Models.Messages.PatientEvent>, KafkaProducerFactory<string, LantanaGroup.Link.Census.Application.Models.Messages.PatientEvent>>();
@@ -179,9 +179,9 @@ static void RegisterServices(WebApplicationBuilder builder)
179179

180180
// Add exception handlers
181181
builder.Services.AddTransient<IDeadLetterExceptionHandler<string, string>, DeadLetterExceptionHandler<string, string>>();
182-
builder.Services.AddTransient<IDeadLetterExceptionHandler<string, List<PatientListItem>>, DeadLetterExceptionHandler<string, List<PatientListItem>>>();
182+
builder.Services.AddTransient<IDeadLetterExceptionHandler<string, PatientListMessage>, DeadLetterExceptionHandler<string, PatientListMessage>>();
183183
builder.Services.AddTransient<ITransientExceptionHandler<string, string>, TransientExceptionHandler<string, string>>();
184-
builder.Services.AddTransient<ITransientExceptionHandler<string, List<PatientListItem>>, TransientExceptionHandler<string, List<PatientListItem>>>();
184+
builder.Services.AddTransient<ITransientExceptionHandler<string, PatientListMessage>, TransientExceptionHandler<string, PatientListMessage>>();
185185

186186
// Quartz
187187
var quartzProps = new NameValueCollection

‎DotNet/DataAcquisition.Domain/Application/Managers/DataAcquisitionLogManager.cs‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -41,6 +41,8 @@ public async Task<DataAcquisitionLogModel> CreateAsync(CreateDataAcquisitionLogM
4141
FacilityId = model.FacilityId,
4242
QueryPhase = model.QueryPhase,
4343
FhirVersion = model.FhirVersion,
44+
QueryType = model.QueryType,
45+
ResourceId = model.ResourceId,
4446
FhirQueries = model.FhirQuery.Select(q => new FhirQuery
4547
{
4648
FacilityId = model.FacilityId,

‎DotNet/DataAcquisition.Domain/Application/Managers/FhirListQueryConfigurationManager.cs‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -107,6 +107,8 @@ public async Task<FhirListConfigurationModel> CreateAsync(CreateFhirListConfigur
107107
FhirBaseServerUrl = model.FhirBaseServerUrl,
108108
FacilityId = model.FacilityId,
109109
Authentication = model.Authentication?.ToDomain(),
110+
CreateDate = DateTime.UtcNow,
111+
ModifyDate = DateTime.UtcNow,
110112
};
111113

112114
var newEntity = await _database.FhirListConfigurationRepository.AddAsync(entity);

0 commit comments

Comments
 (0)