Merge pull request #982 from Project-MONAI/nds-payload-add-workflownames

adding triggered workflow names to payload

Author: neildsouth

src/TaskManager/Plug-ins/Argo/ArgoClient.cs

@@ -17,7 +17,6 @@
 using System.Globalization;
 using System.Text;
 using Argo;
-using Ardalis.GuardClauses;
 using Microsoft.Extensions.Logging;
 using Monai.Deploy.WorkflowManager.TaskManager.Argo.Logging;
 using System.Net;

src/TaskManager/Plug-ins/Email/EmailPlugin.cs

@@ -16,7 +16,6 @@
 
 
 using System.Net.Mail;
-using Ardalis.GuardClauses;
 using FellowOakDicom;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;

src/WorkflowManager/Common/Interfaces/IPayloadService.cs

@@ -48,19 +48,20 @@ Task<IList<PayloadDto>> GetAllAsync(int? skip = null,
         /// <param name="payloadId">payload id to delete.</param>
         Task<bool> DeletePayloadFromStorageAsync(string payloadId);
 
-        /// <summary>
-        /// Updates a payload
-        /// </summary>
-        /// <param name="payload"></param>
-        /// <returns></returns>
-        Task<bool> UpdateWorkflowInstanceIdsAsync(string payloadId, IEnumerable<string> workflowInstances);
-
         /// <summary>
         /// Gets the expiry date for a payload.
         /// </summary>
         /// <param name="now"></param>
         /// <param name="workflowInstanceId"></param>
         /// <returns>date of expiry or null</returns>
         Task<DateTime?> GetExpiry(DateTime now, string? workflowInstanceId);
+
+        /// <summary>
+        /// Updates a payload
+        /// </summary>
+        /// <param name="payloadId">payload id to update.</param>
+        /// <param name="payload">updated payload.</param>
+        /// <returns>true if the update is successful, false otherwise.</returns>
+        Task<bool> UpdateAsyncWorkflowIds(Payload payload);
     }
 }

src/WorkflowManager/Common/Services/PayloadService.cs

@@ -213,7 +213,7 @@ public async Task<bool> DeletePayloadFromStorageAsync(string payloadId)
 
             // update the payload to in progress before we request deletion from storage
             payload.PayloadDeleted = PayloadDeleted.InProgress;
-            await _payloadRepository.UpdateAsync(payload);
+            await _payloadRepository.UpdateAsyncWorkflowIds(payload);
 
             // run deletion in alternative thread so the user isn't held up
 #pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
@@ -238,26 +238,19 @@ public async Task<bool> DeletePayloadFromStorageAsync(string payloadId)
                 }
                 finally
                 {
-                    await _payloadRepository.UpdateAsync(payload);
+                    await _payloadRepository.UpdateAsyncWorkflowIds(payload);
                 }
             });
 #pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
 
             return true;
         }
 
-        public async Task<bool> UpdateWorkflowInstanceIdsAsync(string payloadId, IEnumerable<string> workflowInstances)
+        public Task<bool> UpdateAsyncWorkflowIds(Payload payload)
         {
-            if (await _payloadRepository.UpdateAssociatedWorkflowInstancesAsync(payloadId, workflowInstances))
-            {
-                _logger.PayloadUpdated(payloadId);
-                return true;
-            }
-            else
-            {
-                _logger.PayloadUpdateFailed(payloadId);
-                return false;
-            }
+            ArgumentNullException.ThrowIfNull(payload, nameof(payload));
+
+            return _payloadRepository.UpdateAsyncWorkflowIds(payload);
         }
     }
 }

src/WorkflowManager/Common/Services/WorkflowService.cs

@@ -14,7 +14,6 @@
  * limitations under the License.
  */
 
-using Ardalis.GuardClauses;
 using Microsoft.Extensions.Logging;
 using Monai.Deploy.WorkflowManager.Common.Miscellaneous.Interfaces;
 using Monai.Deploy.WorkflowManager.Common.Contracts.Models;

src/WorkflowManager/Contracts/Migrations/M006_Payload_triggeredWorkflows.cs

@@ -0,0 +1,42 @@
+// 
+// Copyright 2023 Guy’s and St Thomas’ NHS Foundation Trust
+// 
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+// 
+// http://www.apache.org/licenses/LICENSE-2.0
+// 
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+using Monai.Deploy.WorkflowManager.Common.Contracts.Models;
+using Mongo.Migration.Migrations.Document;
+using MongoDB.Bson;
+
+namespace Monai.Deploy.WorkflowManager.Common.Contracts.Migrations
+{
+    public class M006_Payload_triggeredWorkflows : DocumentMigration<Payload>
+    {
+        public M006_Payload_triggeredWorkflows() : base("1.0.6") { }
+
+        public override void Up(BsonDocument document)
+        {
+            document.Add("TriggeredWorkflowNames", BsonNull.Create(null).ToJson(), true);
+        }
+
+        public override void Down(BsonDocument document)
+        {
+            try
+            {
+                document.Remove("TriggeredWorkflowNames");
+            }
+            catch
+            {  // can ignore we dont want failures stopping startup !
+            }
+        }
+    }
+}

src/WorkflowManager/Contracts/Models/Payload.cs

@@ -27,7 +27,7 @@
 
 namespace Monai.Deploy.WorkflowManager.Common.Contracts.Models
 {
-    [CollectionLocation("Payloads"), RuntimeVersion("1.0.5")]
+    [CollectionLocation("Payloads"), RuntimeVersion("1.0.6")]
     public class Payload : IDocument
     {
         [JsonConverter(typeof(DocumentVersionConvert)), BsonSerializer(typeof(DocumentVersionConverBson))]
@@ -40,10 +40,13 @@ public class Payload : IDocument
         public string PayloadId { get; set; } = string.Empty;
 
         [JsonProperty(PropertyName = "workflows")]
-        public IEnumerable<string> Workflows { get; set; } = new List<string>();
+        public IEnumerable<string> Workflows { get; set; } = [];
+
+        [JsonProperty(PropertyName = "workflow_names")]
+        public List<string> TriggeredWorkflowNames { get; set; } = [];
 
         [JsonProperty(PropertyName = "workflow_instance_ids")]
-        public IEnumerable<string> WorkflowInstanceIds { get; set; } = new List<string>();
+        public IEnumerable<string> WorkflowInstanceIds { get; set; } = [];
 
         [JsonProperty(PropertyName = "file_count")]
         public int FileCount { get; set; }
@@ -61,10 +64,10 @@ public class Payload : IDocument
         public PayloadDeleted PayloadDeleted { get; set; } = PayloadDeleted.No;
 
         [JsonProperty(PropertyName = "files")]
-        public IList<BlockStorageInfo> Files { get; set; } = new List<BlockStorageInfo>();
+        public IList<BlockStorageInfo> Files { get; set; } = [];
 
         [JsonProperty(PropertyName = "patient_details")]
-        public PatientDetails PatientDetails { get; set; } = new PatientDetails();
+        public PatientDetails PatientDetails { get; set; } = new();
 
         public DataOrigin DataTrigger { get; set; } = new DataOrigin { DataService = DataService.DIMSE };
 

src/WorkflowManager/Database/Interfaces/IPayloadRepository.cs

@@ -51,15 +51,7 @@ public interface IPayloadRepository
         /// </summary>
         /// <param name="payload">The payload to update.</param>
         /// <returns>The updated payload.</returns>
-        Task<bool> UpdateAsync(Payload payload);
-
-        /// <summary>
-        /// Updates a payload in the database.
-        /// </summary>
-        /// <param name="payloadId"></param>
-        /// <param name="workflowInstances"></param>
-        /// <returns></returns>
-        Task<bool> UpdateAssociatedWorkflowInstancesAsync(string payloadId, IEnumerable<string> workflowInstances);
+        Task<bool> UpdateAsyncWorkflowIds(Payload payload);
 
         /// <summary>
         /// Gets all the payloads that might need deleted

src/WorkflowManager/Database/Repositories/PayloadRepository.cs

@@ -18,7 +18,6 @@
 using System.Collections.Generic;
 using System.Linq;
 using System.Threading.Tasks;
-using Ardalis.GuardClauses;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
 using Monai.Deploy.WorkflowManager.Common.Contracts.Models;
@@ -123,7 +122,7 @@ public async Task<Payload> GetByIdAsync(string payloadId)
             return payload;
         }
 
-        public async Task<bool> UpdateAsync(Payload payload)
+        public async Task<bool> UpdateAsyncWorkflowIds(Payload payload)
         {
             ArgumentNullException.ThrowIfNull(payload, nameof(payload));
 
@@ -132,31 +131,17 @@ public async Task<bool> UpdateAsync(Payload payload)
                 var filter = Builders<Payload>.Filter.Eq(p => p.PayloadId, payload.PayloadId);
                 await _payloadCollection.ReplaceOneAsync(filter, payload);
 
-                return true;
-            }
-            catch (Exception ex)
-            {
-                _logger.DbUpdatePayloadError(payload.PayloadId, ex);
-                return false;
-            }
-        }
-
-        public async Task<bool> UpdateAssociatedWorkflowInstancesAsync(string payloadId, IEnumerable<string> workflowInstances)
-        {
-            Guard.Against.NullOrEmpty(workflowInstances, nameof(workflowInstances));
-            ArgumentNullException.ThrowIfNullOrWhiteSpace(payloadId, nameof(payloadId));
-
-            try
-            {
                 await _payloadCollection.FindOneAndUpdateAsync(
-                    i => i.Id == payloadId,
-                    Builders<Payload>.Update.Set(p => p.WorkflowInstanceIds, workflowInstances));
+                    i => i.Id == payload.Id,
+                    Builders<Payload>.Update
+                        .Set(p => p.TriggeredWorkflowNames, payload.TriggeredWorkflowNames)
+                        .Set(p => p.WorkflowInstanceIds, payload.WorkflowInstanceIds));
 
                 return true;
             }
             catch (Exception ex)
             {
-                _logger.DbUpdateWorkflowInstanceError(ex);
+                _logger.DbUpdatePayloadError(payload.PayloadId, ex);
                 return false;
             }
         }

src/WorkflowManager/PayloadListener/Services/EventPayloadRecieverService.cs

@@ -92,6 +92,12 @@ public async Task ReceiveWorkflowPayload(MessageReceivedEventArgs message)
 
                         return;
                     }
+
+                    if (string.IsNullOrWhiteSpace(string.Join("", payload.TriggeredWorkflowNames)) is false)
+                    {
+                        await PayloadService.UpdateAsyncWorkflowIds(payload);
+                    }
+
                 }
                 else
                 {