AC-487-get-triggered-workflows-by-payload-api-endpoint

src/WorkflowManager/Common/Interfaces/IWorkflowInstanceService.cs

@@ -31,7 +31,8 @@ public interface IWorkflowInstanceService : IPaginatedApi<WorkflowInstance>
         /// <param name="skip"></param>
         /// <param name="limit"></param>
         /// <param name="status"></param>
+        /// <param name="payloadId"></param>
         /// <returns></returns>
-        public Task<IList<WorkflowInstance>> GetAllAsync(int? skip = null, int? limit = null, Status? status = null);
+        public Task<IList<WorkflowInstance>> GetAllAsync(int? skip = null, int? limit = null, Status? status = null, string? payloadId = null);
     }
 }

src/WorkflowManager/Common/Services/WorkflowInstanceService.cs

@@ -39,10 +39,10 @@ public async Task<WorkflowInstance> GetByIdAsync(string id)
 
         public async Task<long> CountAsync() => await _workflowInstanceRepository.CountAsync();
 
-        public async Task<IList<WorkflowInstance>> GetAllAsync(int? skip = null, int? limit = null, Status? status = null)
-            => await _workflowInstanceRepository.GetAllAsync(skip, limit, status);
+        public async Task<IList<WorkflowInstance>> GetAllAsync(int? skip = null, int? limit = null, Status? status = null, string? payloadId = null)
+            => await _workflowInstanceRepository.GetAllAsync(skip, limit, status, payloadId);
 
         public async Task<IList<WorkflowInstance>> GetAllAsync(int? skip = null, int? limit = null)
-            => await _workflowInstanceRepository.GetAllAsync(skip, limit, null);
+            => await _workflowInstanceRepository.GetAllAsync(skip, limit, null, null);
     }
 }

src/WorkflowManager/Database/Interfaces/IWorkflowInstanceRepository.cs

@@ -58,7 +58,7 @@ public interface IWorkflowInstanceRepository
         /// <param name="skip"></param>
         /// <param name="limit"></param>
         /// <returns></returns>
-        Task<IList<WorkflowInstance>> GetAllAsync(int? skip, int? limit, Status? status);
+        Task<IList<WorkflowInstance>> GetAllAsync(int? skip, int? limit, Status? status, string? payloadId);
 
         /// <summary>
         /// Updates the Task Status for a given task within a workflow instance.

src/WorkflowManager/Database/Repositories/WorkflowInstanceRepository.cs

@@ -236,14 +236,18 @@ public async Task<WorkflowInstance> GetByWorkflowInstanceIdAsync(string workflow
 
         public Task<long> CountAsync() => CountAsync(_workflowInstanceCollection, null);
 
-        public async Task<IList<WorkflowInstance>> GetAllAsync(int? skip = null, int? limit = null, Status? status = null)
+        public async Task<IList<WorkflowInstance>> GetAllAsync(int? skip = null, int? limit = null, Status? status = null, string? payloadId = null)
         {
             var builder = Builders<WorkflowInstance>.Filter;
             var filter = builder.Empty;
             if (status is not null)
             {
                 filter &= builder.Eq(w => w.Status, status);
             }
+            if (string.IsNullOrEmpty(payloadId) is false)
+            {
+                filter &= builder.Eq(w => w.PayloadId, payloadId);
+            }
 
             return await GetAllAsync(_workflowInstanceCollection,
                                           filter,

src/WorkflowManager/WorkflowExecuter/Services/WorkflowExecuterService.cs

@@ -116,7 +116,7 @@ await _workflowRepository.GetWorkflowsByAeTitleAsync(message.CalledAeTitle) as L
             {
                 processed &= await _workflowInstanceRepository.CreateAsync(workflowInstances);
 
-                var workflowInstanceIds = workflowInstances.Select(workflowInstance => workflowInstance.WorkflowId);
+                var workflowInstanceIds = workflowInstances.Select(workflowInstance => workflowInstance.Id);
                 await _payloadService.UpdateWorkflowInstanceIdsAsync(payload.Id, workflowInstanceIds).ConfigureAwait(false);
             }
 

src/WorkflowManager/WorkflowManager/Controllers/WorkflowInstanceController.cs

@@ -67,24 +67,35 @@ public WorkflowInstanceController(
         /// </summary>
         /// <param name="filter">Filters.</param>
         /// <param name="status">Workflow instance status filter.</param>
+        /// <param name="payloadId">PayloadId.</param>
+        /// <param name="disablePagination">Disabled pagination.</param>
         /// <returns>A list of workflow instances.</returns>
         [HttpGet]
-        public async Task<IActionResult> GetListAsync([FromQuery] PaginationFilter filter, [FromQuery] string status = null)
+        public async Task<IActionResult> GetListAsync([FromQuery] PaginationFilter filter, [FromQuery] string status = null, [FromQuery] string? payloadId = null, [FromQuery] bool disablePagination = false)
         {
             try
             {
+                Status? parsedStatus = status == null ? null : Enum.Parse<Status>(status, true);
+
+                if (disablePagination is true)
+                {
+                    var unpagedData = await _workflowInstanceService.GetAllAsync(null, null, parsedStatus, payloadId);
+
+                    return Ok(unpagedData);
+                }
+
                 var route = Request?.Path.Value ?? string.Empty;
                 var pageSize = filter.PageSize ?? _options.Value.EndpointSettings.DefaultPageSize;
-                Status? parsedStatus = status == null ? null : Enum.Parse<Status>(status, true);
                 var validFilter = new PaginationFilter(filter.PageNumber, pageSize, _options.Value.EndpointSettings.MaxPageSize);
 
                 var pagedData = await _workflowInstanceService.GetAllAsync(
                     (validFilter.PageNumber - 1) * validFilter.PageSize,
                     validFilter.PageSize,
-                    parsedStatus);
+                    parsedStatus,
+                    payloadId);
 
                 var dataTotal = await _workflowInstanceService.CountAsync();
-                var pagedReponse = CreatePagedReponse<WorkflowInstance>(pagedData.ToList(), validFilter, dataTotal, _uriService, route);
+                var pagedReponse = CreatePagedReponse(pagedData.ToList(), validFilter, dataTotal, _uriService, route);
 
                 return Ok(pagedReponse);
             }

src/WorkflowManager/WorkflowManager/Services/UriService.cs

@@ -44,6 +44,11 @@ public UriService(Uri baseUri)
         /// <returns>Uri.</returns>
         public string GetPageUriString(PaginationFilter filter, string route)
         {
+            if (_baseUri.ToString().EndsWith('/') && route.StartsWith('/'))
+            {
+                route = route.TrimStart('/');
+            }
+
             var endpointUri = new Uri(string.Concat(_baseUri, route));
             var modifiedUri = QueryHelpers.AddQueryString(endpointUri.ToString(), "pageNumber", filter.PageNumber.ToString());
             modifiedUri = QueryHelpers.AddQueryString(modifiedUri, "pageSize", filter?.PageSize?.ToString() ?? string.Empty);

tests/IntegrationTests/WorkflowExecutor.IntegrationTests/Features/PayloadCollection.feature

@@ -41,3 +41,22 @@ Scenario: Payload collection will be populated with null patient details after r
     When I publish a Workflow Request Message Basic_AeTitle_Payload_Collection_Request_3 with artifacts no_patient_metadata in minio
     Then A payload collection is created with patient details Null_Patient
     And I can see 1 Workflow Instance is created
+
+@WorkflowInstanceDetails
+Scenario: Payload collection will be populated with workflow instance id after receiving a Workflow Request when workflow is matched
+    Given I have a clinical workflow Basic_Workflow_1
+    When I publish a Workflow Request Message Basic_AeTitle_Payload_Collection_Request_1 with artifacts full_patient_metadata in minio
+    Then A payload collection is created with 1 workflow instance id
+
+@WorkflowInstanceDetails
+Scenario: Payload collection will be populated with workflow instance ids after receiving a Workflow Request when multiple workflows are matched
+    Given I have a clinical workflow Basic_Workflow_1
+    Given I have a clinical workflow Basic_Workflow_2
+    When I publish a Workflow Request Message Basic_AeTitle_Payload_Collection_Request_1 with artifacts full_patient_metadata in minio
+    Then A payload collection is created with 2 workflow instance id
+
+@WorkflowInstanceDetails
+Scenario: Payload collection will not be populated with workflow instance ids after receiving a Workflow Request when no workflows are matched
+    Given I publish a Workflow Request Message Basic_AeTitle_Payload_Collection_Request_1 with artifacts full_patient_metadata in minio
+    Then A payload collection is created with 0 workflow instance id
+

tests/IntegrationTests/WorkflowExecutor.IntegrationTests/Features/WorkflowInstancesApi.feature

@@ -65,6 +65,20 @@ Scenario Outline: Get all workflow instances from API - Test pagination
     |                            | 0      | 0                |
     | ?pageNumber=1&pageSize=100 | 15     | 15               |
 
+@WorkflowInstancePagination
+Scenario Outline: Get all workflow instances from API with provided status or PayloadId - Test pagination
+    Given I have an endpoint /workflowinstances/<pagination_query>
+    And I have <amount> Workflow Instances
+    When I send a GET request
+    Then I will get a 200 response
+    And Pagination is working correctly for the <pagination_count> workflow instances
+    And All results have correct <expected_status> and <expected_payloadId>
+    Examples:
+    | pagination_query                                           | amount | pagination_count | expected_status | expected_payloadId                   |
+    | ?pageSize=1                                                | 15     | 15               |                 |                                      |
+    | ?pageSize=1&status=created                                 | 15     | 15               | 0               |                                      |
+    | ?pageSize=1&payloadId=5450c3a9-2b19-45b0-8b17-fb10f89d1b2d | 15     | 15               |                 | 5450c3a9-2b19-45b0-8b17-fb10f89d1b2d |
+
 @WorkflowInstancePagination
 Scenario Outline: Invalid pagination returns 400
     Given I have an endpoint /workflowinstances/<pagination_query>

tests/IntegrationTests/WorkflowExecutor.IntegrationTests/StepDefinitions/PayloadCollectionStepDefinitions.cs

@@ -17,6 +17,10 @@
 using BoDi;
 using Monai.Deploy.WorkflowManager.IntegrationTests.Support;
 using TechTalk.SpecFlow.Infrastructure;
+using Polly;
+using Polly.Retry;
+using NUnit.Framework;
+using Monai.Deploy.WorkflowManager.Contracts.Models;
 
 namespace Monai.Deploy.WorkflowManager.IntegrationTests.StepDefinitions
 {
@@ -29,6 +33,7 @@ public class PayloadCollectionStepDefinitions
         private Assertions Assertions { get; set; }
         private DataHelper DataHelper { get; set; }
         private readonly ISpecFlowOutputHelper _outputHelper;
+        private RetryPolicy RetryPolicy { get; set; }
 
 #pragma warning disable CS8618 // Non-nullable field must contain a non-null value when exiting constructor. Consider declaring as nullable.
         public PayloadCollectionStepDefinitions(ObjectContainer objectContainer, ISpecFlowOutputHelper outputHelper)
@@ -37,6 +42,7 @@ public PayloadCollectionStepDefinitions(ObjectContainer objectContainer, ISpecFl
             _outputHelper = outputHelper;
             Assertions = new Assertions(objectContainer);
             DataHelper = objectContainer.Resolve<DataHelper>();
+            RetryPolicy = Policy.Handle<Exception>().WaitAndRetry(retryCount: 5, sleepDurationProvider: _ => TimeSpan.FromMilliseconds(500));
         }
 
         [Then(@"A payload collection is created with patient details (.*)")]
@@ -64,5 +70,39 @@ public void ThenPayloadCollectionIsCreated(string patientDetailsName)
                 }
             }
         }
+        [Then(@"A payload collection is created with (.*) workflow instance id")]
+        public void ThenAPayloadCollectionIsCreatedWithWorkflowInstanceId(int count)
+        {
+            RetryPolicy.Execute(() =>
+            {
+                _outputHelper.WriteLine($"Retrieving payload collection using the payloadid={DataHelper.WorkflowRequestMessage.PayloadId}");
+                var payloadCollections = DataHelper.GetPayloadCollections(DataHelper.WorkflowRequestMessage.PayloadId.ToString());
+                _outputHelper.WriteLine($"Retrieved payload collection");
+
+                if (payloadCollections != null)
+                {
+                    foreach (var payloadCollection in payloadCollections)
+                    {
+
+                        var workflowInstances = DataHelper.GetWorkflowInstances(count, DataHelper.WorkflowRequestMessage.PayloadId.ToString());
+                        if (count != 0)
+                        {
+                            if (workflowInstances != null)
+                            {
+                                Assertions.AssertPayloadWorkflowInstanceId(payloadCollection, workflowInstances);
+                            }
+                            else
+                            {
+                                throw new Exception($"Workflow Instance not found");
+                            }
+                        }
+                        else
+                        {
+                            payloadCollection.WorkflowInstanceIds.Should().BeEmpty();
+                        }
+                    }
+                }
+            });
+        }
     }
 }

tests/IntegrationTests/WorkflowExecutor.IntegrationTests/StepDefinitions/WorkflowInstancesApiStepDefinitions.cs

@@ -56,10 +56,30 @@ public void ThenICanSeeExpectedWorkflowInstanceIsReturned()
         [Then(@"Pagination is working correctly for the (.*) workflow instances")]
         public void ThenPaginationIsWorkingCorrectlyForTheWorkflowInstance(int count)
         {
-            var request = ApiHelper.Request.RequestUri.Query;
+            var request = ApiHelper.Request.RequestUri?.Query;
             var result = ApiHelper.Response.Content.ReadAsStringAsync().Result;
             var deserializedResult = JsonConvert.DeserializeObject<PagedResponse<List<WorkflowInstance>>>(result);
             Assertions.AssertPagination(count, request, deserializedResult);
         }
+
+        [Then(@"All results have correct (.*) and (.*)")]
+        public void AllResultsHaveExpectedStatusOrPayloadId(int? expected_status, string? expected_payloadId)
+        {
+            var result = ApiHelper.Response.Content.ReadAsStringAsync().Result;
+            var deserializedResult = JsonConvert.DeserializeObject<PagedResponse<List<WorkflowInstance>>>(result);
+
+            Action<WorkflowInstance> func = wi => { };
+            if (string.IsNullOrWhiteSpace(expected_payloadId) is false)
+            {
+                func += wi => wi.PayloadId.Should().Be(expected_payloadId);
+            }
+            if (expected_status is not null)
+            {
+                func += wi => wi.Status.Should().Be((Status)expected_status);
+            }
+
+            deserializedResult.Should().NotBeNull();
+            deserializedResult?.Data.ForEach(func);
+        }
     }
 }

tests/IntegrationTests/WorkflowExecutor.IntegrationTests/StepDefinitions/WorkflowRequestStepDefinitions.cs

@@ -70,6 +70,7 @@ public void GivenIHaveClinicalWorkflows(int count)
             }
         }
 
+        [Given(@"I publish a Workflow Request Message (.*) with artifacts (.*) in minio")]
         [When(@"I publish a Workflow Request Message (.*) with artifacts (.*) in minio")]
         public async Task WhenIPublishAWorkflowRequestMessageWithObjects(string name, string folderName)
         {

tests/IntegrationTests/WorkflowExecutor.IntegrationTests/Support/Assertions.cs

@@ -283,48 +283,66 @@ public void AssertPayloadCollection(Payload payloadCollection, PatientDetails pa
             payloadCollection.Timestamp.Should().BeCloseTo(DateTime.UtcNow, precision: TimeSpan.FromMinutes(1));
             payloadCollection.PatientDetails.Should().BeEquivalentTo(patientDetails);
         }
+        public void AssertPayloadWorkflowInstanceId(Payload payloadCollection, List<WorkflowInstance> workflowInstances)
+        {
+            foreach (var workflowInstance in workflowInstances)
+            {
+                payloadCollection.WorkflowInstanceIds.Should().Contain(workflowInstance.Id.ToString());
+            }
+        }
 
         public void AssertWorkflowIstanceMatchesExpectedTaskStatusUpdate(WorkflowInstance updatedWorkflowInstance, TaskExecutionStatus taskExecutionStatus)
         {
             updatedWorkflowInstance.Tasks[0].Status.Should().Be(taskExecutionStatus);
         }
 
-        public static void AssertPagination<T>(int count, string queries, T? Response)
+        public static void AssertPagination<T>(int count, string? queries, T? Response)
         {
             var responseType = Response?.GetType();
-            var data = responseType?.GetProperty("Data")?.GetValue(Response, null) as ICollection;
-            var totalPages = responseType?.GetProperty("TotalPages")?.GetValue(Response, null);
-            var pageSize = responseType?.GetProperty("PageSize")?.GetValue(Response, null);
-            var totalRecords = responseType?.GetProperty("TotalRecords")?.GetValue(Response, null);
-            var pageNumber = responseType?.GetProperty("PageNumber")?.GetValue(Response, null);
-            int pageNumberQuery = 1;
-            int pageSizeQuery = 10;
-            var splitQuery = queries.Split("&").ToList();
-            if (queries != "")
+            GetPropertyValues(Response, responseType, out var data, out var totalPages, out var pageSize, out var totalRecords, out var pageNumber);
+            var pageNumberQuery = 1;
+            var pageSizeQuery = 10;
+
+            if (string.IsNullOrWhiteSpace(queries) is false)
             {
+                var splitQuery = queries?.Split("&") ?? Array.Empty<string>();
                 foreach (var query in splitQuery)
                 {
-                    if (query.Contains("pageNumber"))
+                    if (query.Contains("status=") || query.Contains("payloadId="))
+                    {
+                        continue;
+                    }
+                    else if (query.Contains("pageNumber") && int.TryParse(query.Split("=")[1], out var pageNumberResult))
                     {
-                        pageNumberQuery = Int32.Parse(query.Split("=")[1]);
+                        pageNumberQuery = pageNumberResult;
                     }
-                    else if (query.Contains("pageSize"))
+                    else if (query.Contains("pageSize") && int.TryParse(query.Split("=")[1], out var pageSizeResult))
                     {
-                        pageSizeQuery = Int32.Parse(query.Split("=")[1]);
+                        pageSizeQuery = pageSizeResult;
                     }
                     else
                     {
                         throw new Exception($"Pagination query {query} is not valid");
                     }
                 }
             }
+
             AssertDataCount(data, pageNumberQuery, pageSizeQuery, count);
             AssertTotalPages(totalPages, count, pageSizeQuery);
             totalRecords.Should().Be(count);
             pageNumber.Should().Be(pageNumberQuery);
             pageSize.Should().Be(pageSizeQuery);
         }
 
+        private static void GetPropertyValues<T>(T? Response, Type? responseType, out ICollection? data, out object? totalPages, out object? pageSize, out object? totalRecords, out object? pageNumber)
+        {
+            data = responseType?.GetProperty("Data")?.GetValue(Response, null) as ICollection;
+            totalPages = responseType?.GetProperty("TotalPages")?.GetValue(Response, null);
+            pageSize = responseType?.GetProperty("PageSize")?.GetValue(Response, null);
+            totalRecords = responseType?.GetProperty("TotalRecords")?.GetValue(Response, null);
+            pageNumber = responseType?.GetProperty("PageNumber")?.GetValue(Response, null);
+        }
+
         public void WorkflowInstanceIncludesTaskDetails(List<TaskDispatchEvent> taskDispatchEvents, WorkflowInstance workflowInstance, WorkflowRevision workflowRevision)
         {
             foreach (var taskDispatchEvent in taskDispatchEvents)