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; using Elsa.Workflows.Management.Filters; using Elsa.Workflows.Models; using Elsa.Workflows.Runtime; using Elsa.Workflows.Runtime.Filters; using Elsa.Workflows.Runtime.Messages; using Microsoft.Extensions.DependencyInjection; using Newtonsoft.Json; using Newtonsoft.Json.Converters; using San.MongoLogger; using System.Diagnostics; using System.Dynamic; using System.Text.Json; namespace Caspian.WorkflowFramework.Services; public class WorkflowService : IWorkflowService, ILoggable { #region Fields private readonly IServiceProvider _serviceProvider; private readonly IWorkflowRuntime _workflowRuntime; 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 IBookmarkStore _bookmarkStore; private readonly IActivityExecutionStore _activityExecutionStore; private readonly IMdbLogger _logger; #endregion #region Ctor public WorkflowService(IServiceProvider serviceProvider, IMdbLogger logger) { _serviceProvider = serviceProvider; _workflowRuntime = _serviceProvider.GetRequiredService(); _workflowDefinitionService = _serviceProvider.GetRequiredService(); _workflowInstanceVariableManager = _serviceProvider.GetRequiredService(); _workflowInstanceManager = _serviceProvider.GetRequiredService(); _workflowInstanceStore = _serviceProvider.GetRequiredService(); _workflowDefinitionCacheManager = _serviceProvider.GetRequiredService(); _eventPublisher = _serviceProvider.GetRequiredService(); _bookmarkStore = _serviceProvider.GetRequiredService(); _activityExecutionStore = _serviceProvider.GetRequiredService(); _logger = logger; } #endregion #region CreateAndExecuteAsync /// public async Task> CreateAndExecuteAsync(CreateAndExecuteWorkflowModel input) { var sw = Stopwatch.StartNew(); try { _logger.Log("Start of CreateAndExecuteAsync", input); WorkflowResponse response; var workflowClient = await _workflowRuntime.CreateClientAsync(); var definition = await _workflowDefinitionService .FindWorkflowDefinitionAsync(new WorkflowDefinitionFilter() { Name = input.Name }); if (definition == null) { var msg = $"Workflow '{input.Name}' defintion not found"; response = new WorkflowResponse() { Status = 1, Message = msg, Data = null }; sw.Stop(); _logger.Log("End of CreateAndExecuteAsync", WFInstanceId: response.Data?.InstanceId, response, sw.ElapsedMilliseconds); return response; } 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); var variables = await _workflowInstanceVariableManager.GetVariablesAsync(result.WorkflowInstanceId); // تبدیل به دیکشنری var dictionary = variables.ToDictionary(a => a.Variable.Name, a => a.Value); var workflowModel = dictionary.ConvertToWorkflowModel(); workflowModel.InstanceId = workflowClient.WorkflowInstanceId; workflowModel.UserId = input.UserId; workflowModel.InstanceMode = "new"; response = new WorkflowResponse() { Status = 0, Message = "Success", Data = workflowModel }; sw.Stop(); _logger.Log("End of CreateAndExecuteAsync", WFInstanceId: response.Data?.InstanceId, response, sw.ElapsedMilliseconds); return response; } catch (Exception ex) { var response = new WorkflowResponse() { Status = 100, Message = ex.Message, Data = null }; sw.Stop(); _logger.Log("Exception occurs in CreateAndExecuteAsync", WFInstanceId: response.Data?.InstanceId, response, sw.ElapsedMilliseconds, ex); return response; } } #endregion #region CreateOrGetInstanceAsync /// public async Task> CreateOrGetInstanceAsync(CreateAndExecuteWorkflowModel input) { var sw = Stopwatch.StartNew(); try { _logger.Log("Start of CreateOrGetInstanceAsync", input); WorkflowResponse response; var workflowClient = await _workflowRuntime.CreateClientAsync(); var definition = await _workflowDefinitionService .FindWorkflowDefinitionAsync(new WorkflowDefinitionFilter() { Name = input.Name }); if (definition == null) { var msg = $"Workflow '{input.Name}' defintion not found"; response = new WorkflowResponse() { Status = 1, Message = msg, Data = null }; sw.Stop(); _logger.Log("End of CreateOrGetInstanceAsync", WFInstanceId: response.Data?.InstanceId, response, sw.ElapsedMilliseconds); return response; } 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 { WorkflowDefinitionHandle = WorkflowDefinitionHandle.ByDefinitionId(definition.DefinitionId, input.Version), Input = instanceInput, ExternalUserId = input.UserId }); var result = await workflowClient.RunInstanceAsync(RunWorkflowInstanceRequest.Empty); workflowInstanceId = 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 = workflowInstanceId; workflowModel.UserId = input.UserId; workflowModel.InstanceMode = runningWorkflowInstance != null ? "exist" : "new"; response = new WorkflowResponse() { Status = 0, Message = "Success", Data = workflowModel }; sw.Stop(); _logger.Log("End of CreateOrGetInstanceAsync", WFInstanceId: response.Data?.InstanceId, response, sw.ElapsedMilliseconds); return response; } catch (Exception ex) { var response = new WorkflowResponse() { Status = 100, Message = ex.Message, Data = null }; sw.Stop(); _logger.Log("Exception occurs in CreateOrGetInstanceAsync", WFInstanceId: response.Data?.InstanceId, response, sw.ElapsedMilliseconds, ex); return response; } } #endregion #region PublishEventAsync /// public async Task> PublishEventAsync( PublishEventModel input) { var sw = Stopwatch.StartNew(); try { _logger.Log( "Start of PublishEventAsync", WFInstanceId: input.WorkflowInstanceId, input); var instance = await _workflowInstanceManager.FindByIdAsync( input.WorkflowInstanceId); if (instance == null) { var notFoundResponse = new WorkflowResponse { Status = 1, Message = $"Workflow instance with id " + $"'{input.WorkflowInstanceId}' not found", Data = null }; sw.Stop(); _logger.Log( "End of PublishEventAsync", WFInstanceId: input.WorkflowInstanceId, notFoundResponse, sw.ElapsedMilliseconds); return notFoundResponse; } /* * JsonElement * ↓ * ExpandoObject * ↓ * اضافه‌شدن SessionId * ↓ * PublishEvent */ ExpandoObject payload = new(); if (input.EventData.ValueKind is not JsonValueKind.Undefined and not JsonValueKind.Null) { if (input.EventData.ValueKind != JsonValueKind.Object) { throw new InvalidOperationException( "EventData must be a JSON object."); } var json = input.EventData.GetRawText(); var serializerSettings = new JsonSerializerSettings { Converters = { new ExpandoObjectConverter() } }; payload = JsonConvert.DeserializeObject( json, serializerSettings) ?? new ExpandoObject(); } /* * ExpandoObject پیاده‌سازی IDictionary است؛ * بنابراین بدون Serialize/Deserialize مجدد می‌توانیم * SessionId را به Payload اضافه کنیم. */ if (!string.IsNullOrWhiteSpace(input.SessionId)) { var payloadDictionary = (IDictionary)payload; payloadDictionary["SessionId"] = input.SessionId; } await _eventPublisher.PublishAsync( eventName: input.EventName, correlationId: null, workflowInstanceId: input.WorkflowInstanceId, activityInstanceId: null, payload: payload, asynchronous: false); var variables = await _workflowInstanceVariableManager.GetVariablesAsync( input.WorkflowInstanceId); var dictionary = variables.ToDictionary( x => x.Variable.Name, x => x.Value); var workflowModel = dictionary.ConvertToWorkflowModel(); workflowModel.InstanceId = instance.Id; workflowModel.UserId = instance.ExternalUserId; workflowModel.InstanceMode = "exist"; var response = new WorkflowResponse { Status = 0, Message = "Success", Data = workflowModel }; sw.Stop(); _logger.Log( "End of PublishEventAsync", WFInstanceId: input.WorkflowInstanceId, response, sw.ElapsedMilliseconds); return response; } catch (Exception ex) { sw.Stop(); var response = new WorkflowResponse { Status = 100, Message = ex.Message, Data = null }; _logger.Log( "Exception occurs in PublishEventAsync", WFInstanceId: input.WorkflowInstanceId, response, sw.ElapsedMilliseconds, ex); return response; } } #endregion #region InquiryStateAsync /// public async Task> InquiryStateAsync(InquiryStateModel input) { var sw = Stopwatch.StartNew(); try { _logger.Log("Start of InquiryStateAsync", WFInstanceId: input.WorkflowInstanceId, input); WorkflowResponse response; var instance = await _workflowInstanceManager.FindByIdAsync(input.WorkflowInstanceId); if (instance == null) { var msg = $"Workflow instance with id '{input.WorkflowInstanceId}' not found"; response = new WorkflowResponse() { Status = 1, Message = msg, Data = null }; sw.Stop(); _logger.Log("End of InquiryStateAsync", WFInstanceId: input.WorkflowInstanceId, response, sw.ElapsedMilliseconds); return response; } var variables = await _workflowInstanceVariableManager.GetVariablesAsync(input.WorkflowInstanceId); // تبدیل به دیکشنری var dictionary = variables.ToDictionary(a => a.Variable.Name, a => a.Value); var workflowModel = dictionary.ConvertToWorkflowModel(); workflowModel.InstanceId = instance.Id; workflowModel.UserId = instance.ExternalUserId; workflowModel.InstanceMode = "exist"; response = new WorkflowResponse() { Status = 0, Message = "Success", Data = workflowModel }; sw.Stop(); _logger.Log("End of InquiryStateAsync", WFInstanceId: input.WorkflowInstanceId, response, sw.ElapsedMilliseconds); return response; } catch (Exception ex) { var response = new WorkflowResponse() { Status = 100, Message = ex.Message, Data = null }; sw.Stop(); _logger.Log("Exception occurs in InquiryStateAsync", WFInstanceId: input.WorkflowInstanceId, response, sw.ElapsedMilliseconds, ex); return response; } } #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; model.InstanceMode = "exist"; 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; model.InstanceMode = "exist"; 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, Id = input.WorkflowInstanceId, 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 ForceCancelInstanceAsync public async Task> ForceCancelInstanceAsync(CancelWorkflowInstanceModel input) { var sw = Stopwatch.StartNew(); try { _logger.Log("Start ForceCancelInstanceAsync", WFInstanceId: input.WorkflowInstanceId, input); var instance = await _workflowInstanceManager.FindByIdAsync(input.WorkflowInstanceId); if (instance == null) return new WorkflowResponse { Status = 1, Message = "Instance not found", Data = false }; // جلوگری از Resume شدن دوباره instance.Status = WorkflowStatus.Finished; instance.SubStatus = WorkflowSubStatus.Cancelled; // پاک‌سازی bookmarks از دیتابیس واقعی var bookmarkFilter = new BookmarkFilter() { WorkflowInstanceId = input.WorkflowInstanceId }; await _bookmarkStore.DeleteAsync(bookmarkFilter); // پاک‌سازی scheduled activities (در صورتی که runtime distributed باشد) var activityFilter = new ActivityExecutionRecordFilter() { WorkflowInstanceId = input.WorkflowInstanceId }; await _activityExecutionStore.DeleteManyAsync(activityFilter); //// پاک‌سازی incidents (در پروژه‌های بزرگ ضروری است) //await _incidentStore.DeleteManyAsync( // x => x.WorkflowInstanceId == input.WorkflowInstanceId //); // ذخیره instance تغییر یافته await _workflowInstanceManager.SaveAsync(instance); sw.Stop(); var response = new WorkflowResponse { Status = 0, Message = "Force cancel completed", Data = true }; _logger.Log("End ForceCancelInstanceAsync", WFInstanceId: instance.Id, response, sw.ElapsedMilliseconds); return response; } catch (Exception ex) { sw.Stop(); var response = new WorkflowResponse { Status = 100, Message = ex.Message, Data = false }; _logger.Log("Exception in ForceCancelInstanceAsync", WFInstanceId: input.WorkflowInstanceId, response, sw.ElapsedMilliseconds, ex); return response; } } #endregion #region EvictDefFromCacheAsync /// public async Task EvictDefFromCacheAsync(string name, CancellationToken cancellationToken = default) { var sw = Stopwatch.StartNew(); try { _logger.Log("Start of EvictDefFromCacheAsync", name); var defFilter = new WorkflowDefinitionFilter() { Name = name, VersionOptions = VersionOptions.Published }; var definition = await _workflowDefinitionService .FindWorkflowDefinitionAsync(defFilter); await _workflowDefinitionCacheManager.EvictWorkflowDefinitionAsync(definition.DefinitionId); var workflowDefinitionFilter = new WorkflowDefinitionFilter() { DefinitionHandle = WorkflowDefinitionHandle.ByDefinitionId(definition.DefinitionId, VersionOptions.Published) }; var createWorkflowDefinitionFilterCacheKey = _workflowDefinitionCacheManager.CreateWorkflowDefinitionFilterCacheKey(workflowDefinitionFilter); _workflowDefinitionCacheManager.Cache.Remove(createWorkflowDefinitionFilterCacheKey); var createWorkflowFilterCacheKey = _workflowDefinitionCacheManager.CreateWorkflowFilterCacheKey(workflowDefinitionFilter); _workflowDefinitionCacheManager.Cache.Remove(createWorkflowFilterCacheKey); } finally { sw.Stop(); _logger.Log("End of EvictDefFromCacheAsync", name, sw.ElapsedMilliseconds, null); } } #endregion }