From 9ef806b196c2a0ce3c59f145577801aba459c2e9 Mon Sep 17 00:00:00 2001 From: Behzad Alizadeh Date: Wed, 7 Jan 2026 14:12:18 +0330 Subject: [PATCH] Add services --- .../Controllers/WorkflowController.cs | 84 ++++ .../Contracts/IWorkflowService.cs | 46 ++- .../Models/CancelWorkflowInstanceModel.cs | 6 + .../Models/GetActiveInstancesModel.cs | 5 + .../Models/GetLatestActiveInstanceModel.cs | 6 + .../Services/WorkflowService.cs | 390 ++++++++++++++++-- .../Elsa.EntityFrameworkCore.csproj | 2 +- 7 files changed, 497 insertions(+), 42 deletions(-) create mode 100644 src/apps/Caspian.WorkflowFramework/Models/CancelWorkflowInstanceModel.cs create mode 100644 src/apps/Caspian.WorkflowFramework/Models/GetActiveInstancesModel.cs create mode 100644 src/apps/Caspian.WorkflowFramework/Models/GetLatestActiveInstanceModel.cs diff --git a/src/apps/Caspian.WorkflowFramework.Sample/Controllers/WorkflowController.cs b/src/apps/Caspian.WorkflowFramework.Sample/Controllers/WorkflowController.cs index 8d66372..7e21184 100644 --- a/src/apps/Caspian.WorkflowFramework.Sample/Controllers/WorkflowController.cs +++ b/src/apps/Caspian.WorkflowFramework.Sample/Controllers/WorkflowController.cs @@ -47,6 +47,20 @@ public class WorkflowController : Controller, ILoggable } #endregion + #region CreateOrGetInstance + [HttpPost("CreateOrGetInstance")] + public async Task CreateOrGetInstance([FromBody] CreateAndExecuteWorkflowModel input) + { + _logger.Log("Start of CreateOrGetInstance action", JsonConvert.SerializeObject(input)); + + var result = await _workflowService.CreateOrGetInstanceAsync(input); + + _logger.Log("End of CreateOrGetInstance action", WFInstanceId: result.Data?.InstanceId, JsonConvert.SerializeObject(result)); + + return Json(result); + } + #endregion + #region Do [HttpPost("Do")] public async Task Do([FromBody] PublishEventModel input) @@ -76,4 +90,74 @@ public class WorkflowController : Controller, ILoggable return Json(result); } #endregion + + #region Cancel + [HttpPost("Cancel")] + public async Task Cancel([FromBody] CancelWorkflowInstanceModel input) + { + _logger.Log("Start of Cancel action", WFInstanceId: input.WorkflowInstanceId, JsonConvert.SerializeObject(input)); + + var result = await _workflowService.CancelInstanceAsync(input); + + _logger.Log("End of Cancel action", WFInstanceId: input.WorkflowInstanceId, JsonConvert.SerializeObject(result)); + + return Json(result); + } + #endregion + + #region GetLatestActive + [HttpPost("GetLatestActive")] + public async Task GetLatestActive([FromBody] GetLatestActiveInstanceModel input) + { + _logger.Log("Start of GetLatestActive action", JsonConvert.SerializeObject(input)); + + var result = await _workflowService.GetLatestActiveInstanceAsync(input); + + _logger.Log("End of GetLatestActive action", null, JsonConvert.SerializeObject(result)); + + return Json(result); + } + #endregion + + #region GetLatestActiveInstanceId + [HttpPost("GetLatestActiveInstanceId")] + public async Task GetLatestActiveInstanceId([FromBody] GetLatestActiveInstanceModel input) + { + _logger.Log("Start of GetLatestActiveInstanceId action", JsonConvert.SerializeObject(input)); + + var result = await _workflowService.GetLatestActiveInstanceIdAsync(input); + + _logger.Log("End of GetLatestActiveInstanceId action", null, JsonConvert.SerializeObject(result)); + + return Json(result); + } + #endregion + + #region GetActiveInstances + [HttpPost("GetActiveInstances")] + public async Task GetActiveInstances([FromBody] GetActiveInstancesModel input) + { + _logger.Log("Start of GetActiveInstances action", JsonConvert.SerializeObject(input)); + + var result = await _workflowService.GetActiveInstancesAsync(input); + + _logger.Log("End of GetActiveInstances action", null, JsonConvert.SerializeObject(result)); + + return Json(result); + } + #endregion + + #region GetActiveInstanceIds + [HttpPost("GetActiveInstanceIds")] + public async Task GetActiveInstanceIds([FromBody] GetActiveInstancesModel input) + { + _logger.Log("Start of GetActiveInstanceIds action", JsonConvert.SerializeObject(input)); + + var result = await _workflowService.GetActiveInstanceIdsAsync(input); + + _logger.Log("End of GetActiveInstanceIds action", null, JsonConvert.SerializeObject(result)); + + return Json(result); + } + #endregion } diff --git a/src/apps/Caspian.WorkflowFramework/Contracts/IWorkflowService.cs b/src/apps/Caspian.WorkflowFramework/Contracts/IWorkflowService.cs index 0b2db48..a8b63eb 100644 --- a/src/apps/Caspian.WorkflowFramework/Contracts/IWorkflowService.cs +++ b/src/apps/Caspian.WorkflowFramework/Contracts/IWorkflowService.cs @@ -1,4 +1,6 @@ using Caspian.WorkflowFramework.Models; +using System.Collections.Generic; +using System.Threading.Tasks; namespace Caspian.WorkflowFramework.Contracts; public interface IWorkflowService @@ -6,22 +8,50 @@ public interface IWorkflowService /// /// Create and executes a new workflow instance. /// - /// - /// A new object. - /// Create a new workflow instance and executes it and waits for it to complete or reach a suspend point. Task> CreateAndExecuteAsync(CreateAndExecuteWorkflowModel input); /// - /// Publishes the specified event. + /// Create a new workflow instance if it does not exist or get the running instance. + /// + Task> CreateOrGetInstanceAsync(CreateAndExecuteWorkflowModel input); + + /// + /// Publishes the specified event to a workflow instance. /// - /// - /// A new object. Task> PublishEventAsync(PublishEventModel input); /// /// Inquiry last state of workflow instance. /// - /// - /// A new object. Task> InquiryStateAsync(InquiryStateModel input); + + /// + /// Gets all active workflow instances for a specific user. + /// + Task>> GetActiveInstancesAsync(GetActiveInstancesModel input); + + /// + /// Gets all active workflow instance IDs for a specific user. + /// + Task>> GetActiveInstanceIdsAsync(GetActiveInstancesModel input); + + /// + /// Gets the latest active workflow instance for a specific user and workflow name. + /// + Task> GetLatestActiveInstanceAsync(GetLatestActiveInstanceModel input); + + /// + /// Gets the ID of the latest active workflow instance for a specific user and workflow name. + /// + Task> GetLatestActiveInstanceIdAsync(GetLatestActiveInstanceModel input); + + /// + /// Cancels the specified workflow instance. + /// + Task> CancelInstanceAsync(CancelWorkflowInstanceModel input); + + ///// + ///// Evicts a workflow definition from cache (optional, commented out in service). + ///// + //Task> EvictDefFromCacheAsync(string definitionId); } diff --git a/src/apps/Caspian.WorkflowFramework/Models/CancelWorkflowInstanceModel.cs b/src/apps/Caspian.WorkflowFramework/Models/CancelWorkflowInstanceModel.cs new file mode 100644 index 0000000..c878efc --- /dev/null +++ b/src/apps/Caspian.WorkflowFramework/Models/CancelWorkflowInstanceModel.cs @@ -0,0 +1,6 @@ +namespace Caspian.WorkflowFramework.Models; +public class CancelWorkflowInstanceModel +{ + public string WorkflowInstanceId { get; set; } + public long UserId { get; set; } +} diff --git a/src/apps/Caspian.WorkflowFramework/Models/GetActiveInstancesModel.cs b/src/apps/Caspian.WorkflowFramework/Models/GetActiveInstancesModel.cs new file mode 100644 index 0000000..c7e21a6 --- /dev/null +++ b/src/apps/Caspian.WorkflowFramework/Models/GetActiveInstancesModel.cs @@ -0,0 +1,5 @@ +namespace Caspian.WorkflowFramework.Models; +public class GetActiveInstancesModel +{ + public long UserId { get; set; } +} diff --git a/src/apps/Caspian.WorkflowFramework/Models/GetLatestActiveInstanceModel.cs b/src/apps/Caspian.WorkflowFramework/Models/GetLatestActiveInstanceModel.cs new file mode 100644 index 0000000..6d495fe --- /dev/null +++ b/src/apps/Caspian.WorkflowFramework/Models/GetLatestActiveInstanceModel.cs @@ -0,0 +1,6 @@ +namespace Caspian.WorkflowFramework.Models; +public class GetLatestActiveInstanceModel +{ + public long UserId { get; set; } + public string WorkflowName { get; set; } +} diff --git a/src/apps/Caspian.WorkflowFramework/Services/WorkflowService.cs b/src/apps/Caspian.WorkflowFramework/Services/WorkflowService.cs index 5a31e4d..fd34559 100644 --- a/src/apps/Caspian.WorkflowFramework/Services/WorkflowService.cs +++ b/src/apps/Caspian.WorkflowFramework/Services/WorkflowService.cs @@ -4,6 +4,7 @@ using System.Text.Json; using Caspian.WorkflowFramework.Contracts; using Caspian.WorkflowFramework.Models; using Caspian.WorkflowFramework.Utility; +using Elsa.Common.Entities; using Elsa.Common.Models; using Elsa.Workflows; using Elsa.Workflows.Management; @@ -14,7 +15,6 @@ using Elsa.Workflows.Runtime.Messages; using Microsoft.Extensions.DependencyInjection; using Newtonsoft.Json; using Newtonsoft.Json.Converters; -using Proto.Extensions; using San.MongoLogger; namespace Caspian.WorkflowFramework.Services; @@ -26,6 +26,7 @@ public class WorkflowService : IWorkflowService, ILoggable private readonly IWorkflowDefinitionService _workflowDefinitionService; private readonly IWorkflowInstanceVariableManager _workflowInstanceVariableManager; private readonly IWorkflowInstanceManager _workflowInstanceManager; + private readonly IWorkflowInstanceStore _workflowInstanceStore; private readonly IEventPublisher _eventPublisher; private readonly IWorkflowDefinitionCacheManager _workflowDefinitionCacheManager; private readonly IMdbLogger _logger; @@ -40,6 +41,7 @@ public class WorkflowService : IWorkflowService, ILoggable _workflowDefinitionService = _serviceProvider.GetRequiredService(); _workflowInstanceVariableManager = _serviceProvider.GetRequiredService(); _workflowInstanceManager = _serviceProvider.GetRequiredService(); + _workflowInstanceStore = _serviceProvider.GetRequiredService(); _eventPublisher = _serviceProvider.GetRequiredService(); _logger = logger; } @@ -132,14 +134,14 @@ public class WorkflowService : IWorkflowService, ILoggable } #endregion - #region InitAsync + #region CreateOrGetInstanceAsync /// - public async Task> InitAsync(CreateAndExecuteWorkflowModel input) + public async Task> CreateOrGetInstanceAsync(CreateAndExecuteWorkflowModel input) { var sw = Stopwatch.StartNew(); try { - _logger.Log("Start of InitAsync", input); + _logger.Log("Start of CreateOrGetInstanceAsync", input); WorkflowResponse response; @@ -163,37 +165,60 @@ public class WorkflowService : IWorkflowService, ILoggable sw.Stop(); - _logger.Log("End of InitAsync", WFInstanceId: response.Data?.InstanceId, response, sw.ElapsedMilliseconds); + _logger.Log("End of CreateOrGetInstanceAsync", WFInstanceId: response.Data?.InstanceId, response, sw.ElapsedMilliseconds); return response; } - var runningWorkflowInstance = await _workflowInstanceManager - .FindAsync(new WorkflowInstanceFilter() + var instances = await _workflowInstanceStore.FindManyAsync( + new WorkflowInstanceFilter + { + ExternalUserId = input.UserId, + DefinitionId = definition.DefinitionId, + WorkflowStatus = WorkflowStatus.Running + }, + new WorkflowInstanceOrder(x => x.UpdatedAt!, OrderDirection.Descending) + ); + + var runningWorkflowInstance = instances.FirstOrDefault(); + + //var runningWorkflowInstance = await _workflowInstanceManager + // .FindAsync(new WorkflowInstanceFilter() + // { + // ExternalUserId = input.UserId, + // DefinitionId = definition.DefinitionId, + // WorkflowStatus = WorkflowStatus.Running + // }); + + string workflowInstanceId = string.Empty; + if (runningWorkflowInstance != null) + { + workflowInstanceId = runningWorkflowInstance.Id; + } + else + { + var instanceInput = new Dictionary { ["UserId"] = input.UserId }; + input.Version = VersionOptions.Published; + var newInstance = await workflowClient.CreateInstanceAsync(new CreateWorkflowInstanceRequest { - ExternalUserId = input.UserId, - DefinitionId = "jhzbsddvjkf",//definition.DefinitionId, - WorkflowStatus = WorkflowStatus.Running + WorkflowDefinitionHandle = WorkflowDefinitionHandle.ByDefinitionId(definition.DefinitionId, input.Version), + Input = instanceInput, + ExternalUserId = input.UserId }); - var instanceInput = new Dictionary { ["UserId"] = input.UserId }; - input.Version = VersionOptions.Published; - var newInstance = await workflowClient.CreateInstanceAsync(new CreateWorkflowInstanceRequest - { - WorkflowDefinitionHandle = WorkflowDefinitionHandle.ByDefinitionId(definition.DefinitionId, input.Version), - Input = instanceInput, - ExternalUserId = input.UserId - }); + var result = await workflowClient.RunInstanceAsync(RunWorkflowInstanceRequest.Empty); + workflowInstanceId = result.WorkflowInstanceId; + } - var result = await workflowClient.RunInstanceAsync(RunWorkflowInstanceRequest.Empty); - - var variables = await _workflowInstanceVariableManager.GetVariablesAsync(result.WorkflowInstanceId); + var id = workflowInstanceId; + var variables = await _workflowInstanceVariableManager + .GetVariablesAsync(id); // تبدیل به دیکشنری var dictionary = variables.ToDictionary(a => a.Variable.Name, a => a.Value); var workflowModel = dictionary.ConvertToWorkflowModel(); - workflowModel.InstanceId = workflowClient.WorkflowInstanceId; + workflowModel.InstanceId = workflowInstanceId; workflowModel.UserId = input.UserId; response = new WorkflowResponse() @@ -205,7 +230,7 @@ public class WorkflowService : IWorkflowService, ILoggable sw.Stop(); - _logger.Log("End of InitAsync", WFInstanceId: response.Data?.InstanceId, response, sw.ElapsedMilliseconds); + _logger.Log("End of CreateOrGetInstanceAsync", WFInstanceId: response.Data?.InstanceId, response, sw.ElapsedMilliseconds); return response; } @@ -220,7 +245,7 @@ public class WorkflowService : IWorkflowService, ILoggable sw.Stop(); - _logger.Log("Exception occurs in InitAsync", WFInstanceId: response.Data?.InstanceId, response, sw.ElapsedMilliseconds, ex); + _logger.Log("Exception occurs in CreateOrGetInstanceAsync", WFInstanceId: response.Data?.InstanceId, response, sw.ElapsedMilliseconds, ex); return response; } @@ -327,14 +352,7 @@ public class WorkflowService : IWorkflowService, ILoggable WorkflowResponse response; - var instance = await _workflowInstanceManager.FindAsync(new WorkflowInstanceFilter() - { - ExternalUserId = 62181149, - DefinitionId = "8a44ce8bc5c71839", - WorkflowStatus = Elsa.Workflows.WorkflowStatus.Running - }); - - //var instance = await _workflowInstanceManager.FindByIdAsync(input.WorkflowInstanceId); + var instance = await _workflowInstanceManager.FindByIdAsync(input.WorkflowInstanceId); if (instance == null) { @@ -353,7 +371,7 @@ public class WorkflowService : IWorkflowService, ILoggable return response; } - var variables = await _workflowInstanceVariableManager.GetVariablesAsync(instance.Id);// input.WorkflowInstanceId); + var variables = await _workflowInstanceVariableManager.GetVariablesAsync(input.WorkflowInstanceId); // تبدیل به دیکشنری var dictionary = variables.ToDictionary(a => a.Variable.Name, a => a.Value); @@ -393,6 +411,312 @@ public class WorkflowService : IWorkflowService, ILoggable } #endregion + #region GetActiveInstancesAsync + public async Task>> GetActiveInstancesAsync(GetActiveInstancesModel input) + { + var sw = Stopwatch.StartNew(); + try + { + _logger.Log("Start of GetActiveInstancesAsync", input); + + var instances = await _workflowInstanceStore.FindManyAsync( + new WorkflowInstanceFilter + { + ExternalUserId = input.UserId, + WorkflowStatus = WorkflowStatus.Running + }); + + var result = new List(); + + foreach (var instance in instances) + { + var variables = await _workflowInstanceVariableManager.GetVariablesAsync(instance.Id); + var dict = variables.ToDictionary(x => x.Variable.Name, x => x.Value); + + var model = dict.ConvertToWorkflowModel(); + model.InstanceId = instance.Id; + model.UserId = instance.ExternalUserId; + + result.Add(model); + } + + var response = new WorkflowResponse> + { + Status = 0, + Message = "Success", + Data = result + }; + + sw.Stop(); + _logger.Log("End of GetActiveInstancesAsync", response, sw.ElapsedMilliseconds); + return response; + } + catch (Exception ex) + { + var response = new WorkflowResponse> + { + Status = 100, + Message = ex.Message, + Data = null + }; + + sw.Stop(); + _logger.Log("Exception in GetActiveInstancesAsync", response, sw.ElapsedMilliseconds, ex); + return response; + } + } + #endregion + + #region GetActiveInstanceIdsAsync + public async Task>> GetActiveInstanceIdsAsync(GetActiveInstancesModel input) + { + var sw = Stopwatch.StartNew(); + try + { + _logger.Log("Start of GetActiveInstanceIdsAsync", input); + + var instances = await _workflowInstanceStore.FindManyAsync( + new WorkflowInstanceFilter + { + ExternalUserId = input.UserId, + WorkflowStatus = WorkflowStatus.Running + }); + + var instanceIds = instances.Select(i => i.Id).ToList(); + + var response = new WorkflowResponse> + { + Status = 0, + Message = "Success", + Data = instanceIds + }; + + sw.Stop(); + _logger.Log("End of GetActiveInstanceIdsAsync", response, sw.ElapsedMilliseconds); + return response; + } + catch (Exception ex) + { + var response = new WorkflowResponse> + { + Status = 100, + Message = ex.Message, + Data = null + }; + + sw.Stop(); + _logger.Log("Exception in GetActiveInstanceIdsAsync", response, sw.ElapsedMilliseconds, ex); + return response; + } + } + #endregion + + #region GetLatestActiveInstanceAsync + public async Task> GetLatestActiveInstanceAsync(GetLatestActiveInstanceModel input) + { + var sw = Stopwatch.StartNew(); + try + { + _logger.Log("Start of GetLatestActiveInstanceAsync", input); + + var definition = await _workflowDefinitionService.FindWorkflowDefinitionAsync( + new WorkflowDefinitionFilter { Name = input.WorkflowName }); + + if (definition == null) + { + return new WorkflowResponse + { + Status = 1, + Message = $"Workflow '{input.WorkflowName}' not found", + Data = null + }; + } + + var instances = await _workflowInstanceStore.FindManyAsync( + new WorkflowInstanceFilter + { + ExternalUserId = input.UserId, + DefinitionId = definition.DefinitionId, + WorkflowStatus = WorkflowStatus.Running + }, + new WorkflowInstanceOrder(x => x.UpdatedAt!, OrderDirection.Descending) + ); + + var latest = instances.FirstOrDefault(); + + if (latest == null) + { + return new WorkflowResponse + { + Status = 1, + Message = "No active workflow instance found", + Data = null + }; + } + + var variables = await _workflowInstanceVariableManager.GetVariablesAsync(latest.Id); + var dict = variables.ToDictionary(x => x.Variable.Name, x => x.Value); + + var model = dict.ConvertToWorkflowModel(); + model.InstanceId = latest.Id; + model.UserId = latest.ExternalUserId; + + var response = new WorkflowResponse + { + Status = 0, + Message = "Success", + Data = model + }; + + sw.Stop(); + _logger.Log("End of GetLatestActiveInstanceAsync", response, sw.ElapsedMilliseconds); + return response; + } + catch (Exception ex) + { + var response = new WorkflowResponse + { + Status = 100, + Message = ex.Message, + Data = null + }; + + sw.Stop(); + _logger.Log("Exception in GetLatestActiveInstanceAsync", response, sw.ElapsedMilliseconds, ex); + return response; + } + } + #endregion + + #region GetLatestActiveInstanceIdAsync + public async Task> GetLatestActiveInstanceIdAsync(GetLatestActiveInstanceModel input) + { + var sw = Stopwatch.StartNew(); + try + { + _logger.Log("Start of GetLatestActiveInstanceIdAsync", input); + + var definition = await _workflowDefinitionService.FindWorkflowDefinitionAsync( + new WorkflowDefinitionFilter { Name = input.WorkflowName }); + + if (definition == null) + { + return new WorkflowResponse + { + Status = 1, + Message = $"Workflow '{input.WorkflowName}' not found", + Data = null + }; + } + + var instances = await _workflowInstanceStore.FindManyAsync( + new WorkflowInstanceFilter + { + ExternalUserId = input.UserId, + DefinitionId = definition.DefinitionId, + WorkflowStatus = WorkflowStatus.Running + }, + new WorkflowInstanceOrder(x => x.UpdatedAt!, OrderDirection.Descending) + ); + + var latest = instances.FirstOrDefault(); + + if (latest == null) + { + return new WorkflowResponse + { + Status = 1, + Message = "No active workflow instance found", + Data = null + }; + } + + var response = new WorkflowResponse + { + Status = 0, + Message = "Success", + Data = latest.Id + }; + + sw.Stop(); + _logger.Log("End of GetLatestActiveInstanceIdAsync", response, sw.ElapsedMilliseconds); + return response; + } + catch (Exception ex) + { + var response = new WorkflowResponse + { + Status = 100, + Message = ex.Message, + Data = null + }; + + sw.Stop(); + _logger.Log("Exception in GetLatestActiveInstanceIdAsync", response, sw.ElapsedMilliseconds, ex); + return response; + } + } + #endregion + + #region CancelInstanceAsync + public async Task> CancelInstanceAsync(CancelWorkflowInstanceModel input) + { + var sw = Stopwatch.StartNew(); + try + { + _logger.Log("Start of CancelInstanceAsync", WFInstanceId: input.WorkflowInstanceId, input); + + // پیدا کردن instance + var instance = await _workflowInstanceStore.FindAsync( + new WorkflowInstanceFilter + { + ExternalUserId = input.UserId, + WorkflowStatus = WorkflowStatus.Running + }); + + if (instance == null) + { + return new WorkflowResponse + { + Status = 1, + Message = "Workflow instance not found", + Data = false + }; + } + + // ایجاد client روی workflow instance موردنظر + var client = await _workflowRuntime.CreateClientAsync(instance.Id); + + // لغو workflow + await client.CancelAsync(); + + var response = new WorkflowResponse + { + Status = 0, + Message = "Success", + Data = true + }; + + sw.Stop(); + _logger.Log("End of CancelInstanceAsync", WFInstanceId: instance.Id, response, sw.ElapsedMilliseconds); + return response; + } + catch (Exception ex) + { + var response = new WorkflowResponse + { + Status = 100, + Message = ex.Message, + Data = false + }; + + sw.Stop(); + _logger.Log("Exception in CancelInstanceAsync", WFInstanceId: input.WorkflowInstanceId, response, sw.ElapsedMilliseconds, ex); + return response; + } + } + #endregion + //#region EvictDefFromCacheAsync ///// //public async Task> EvictDefFromCacheAsync(string definitionId) diff --git a/src/modules/Elsa.EntityFrameworkCore/Elsa.EntityFrameworkCore.csproj b/src/modules/Elsa.EntityFrameworkCore/Elsa.EntityFrameworkCore.csproj index d238105..93cb22e 100644 --- a/src/modules/Elsa.EntityFrameworkCore/Elsa.EntityFrameworkCore.csproj +++ b/src/modules/Elsa.EntityFrameworkCore/Elsa.EntityFrameworkCore.csproj @@ -4,7 +4,7 @@ Caspian.Elsa.EntityFrameworkCore Caspian Digital banking team - 1.1.0-preview.1 + 1.1.0-preview.2 Provides Entity Framework Core implementations of various abstractions from various modules.