Caspian.WorkflowFramework/Caspian.WorkflowFramework.Package/Services/WorkflowService.cs

using Caspian.WorkflowFramework.Contracts;
using Caspian.WorkflowFramework.Models;
using Caspian.WorkflowFramework.Utility;
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.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 IEventPublisher _eventPublisher;
        private readonly IMdbLogger<WorkflowService> _logger;
        #endregion

        #region Ctor
        public WorkflowService(IServiceProvider serviceProvider,
            IMdbLogger<WorkflowService> logger)
        {
            _serviceProvider = serviceProvider;
            _workflowRuntime = _serviceProvider.GetRequiredService<IWorkflowRuntime>();
            _workflowDefinitionService = _serviceProvider.GetRequiredService<IWorkflowDefinitionService>();
            _workflowInstanceVariableManager = _serviceProvider.GetRequiredService<IWorkflowInstanceVariableManager>();
            _workflowInstanceManager = _serviceProvider.GetRequiredService<IWorkflowInstanceManager>();
            _eventPublisher = _serviceProvider.GetRequiredService<IEventPublisher>();
            _logger = logger;
        }
        #endregion

        #region CreateAndExecuteAsync
        /// <inheritdoc />
        public async Task<WorkflowResponse<WorkflowModel>> CreateAndExecuteAsync(CreateAndExecuteWorkflowModel input)
        {
            var sw = Stopwatch.StartNew();
            try
            {
                _logger.Log("Start of CreateAndExecuteAsync", input);

                WorkflowResponse<WorkflowModel> 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<WorkflowModel>()
                    {
                        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<string, object> { ["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;

                response = new WorkflowResponse<WorkflowModel>()
                {
                    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<WorkflowModel>()
                {
                    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
        /// <inheritdoc />
        public async Task<WorkflowResponse<WorkflowModel>> CreateOrGetInstanceAsync(CreateAndExecuteWorkflowModel input)
        {
            var sw = Stopwatch.StartNew();
            try
            {
                _logger.Log("Start of CreateOrGetInstanceAsync", input);

                WorkflowResponse<WorkflowModel> 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<WorkflowModel>()
                    {
                        Status = 1,
                        Message = msg,
                        Data = null
                    };

                    sw.Stop();

                    _logger.Log("End of CreateOrGetInstanceAsync", WFInstanceId: response.Data?.InstanceId, response, sw.ElapsedMilliseconds);

                    return response;
                }

                var runningWorkflowInstance = await _workflowInstanceManager
                    .FindAsync(new WorkflowInstanceFilter()
                    {
                        ExternalUserId = input.UserId,
                        DefinitionId = definition.DefinitionId,
                        WorkflowStatus = WorkflowStatus.Running
                    });

                var workflowInstanceId = string.Empty;
                if (runningWorkflowInstance != null)
                {
                    workflowInstanceId = runningWorkflowInstance.Id;
                }
                else
                {
                    var instanceInput = new Dictionary<string, object> { ["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 variables = await _workflowInstanceVariableManager
                    .GetVariablesAsync(workflowInstanceId);

                // تبدیل به دیکشنری
                var dictionary = variables.ToDictionary(a => a.Variable.Name, a => a.Value);

                var workflowModel = dictionary.ConvertToWorkflowModel();
                workflowModel.InstanceId = workflowClient.WorkflowInstanceId;
                workflowModel.UserId = input.UserId;

                response = new WorkflowResponse<WorkflowModel>()
                {
                    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<WorkflowModel>()
                {
                    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
        /// <inheritdoc />
        public async Task<WorkflowResponse<WorkflowModel>> PublishEventAsync(PublishEventModel input)
        {
            var sw = Stopwatch.StartNew();
            try
            {
                _logger.Log("Start of PublishEventAsync", WFInstanceId: input.WorkflowInstanceId, input);

                WorkflowResponse<WorkflowModel> response;

                var instance = await _workflowInstanceManager.FindByIdAsync(input.WorkflowInstanceId);

                if (instance == null)
                {
                    var msg = $"Workflow instance with id '{input.WorkflowInstanceId}' not found";
                    response = new WorkflowResponse<WorkflowModel>()
                    {
                        Status = 1,
                        Message = msg,
                        Data = null
                    };

                    sw.Stop();

                    _logger.Log("End of PublishEventAsync", WFInstanceId: input.WorkflowInstanceId, response, sw.ElapsedMilliseconds);

                    return response;
                }

                dynamic payload = null;

                if (input.EventData.ValueKind != JsonValueKind.Undefined &&
                    input.EventData.ValueKind != JsonValueKind.Null)
                {
                    var json = input.EventData.GetRawText();

                    var settings = new JsonSerializerSettings
                    {
                        Converters = { new ExpandoObjectConverter() }
                    };

                    // تبدیل JSON به object داینامیک
                    payload = JsonConvert.DeserializeObject<ExpandoObject>(json, settings);
                }

                await _eventPublisher.PublishAsync(input.EventName, null, input.WorkflowInstanceId, null, payload);

                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;

                response = new WorkflowResponse<WorkflowModel>()
                {
                    Status = 0,
                    Message = "Success",
                    Data = workflowModel
                };

                sw.Stop();

                _logger.Log("End of PublishEventAsync", WFInstanceId: input.WorkflowInstanceId, response, sw.ElapsedMilliseconds);

                return response;
            }
            catch (Exception ex)
            {
                var response = new WorkflowResponse<WorkflowModel>()
                {
                    Status = 100,
                    Message = ex.Message,
                    Data = null
                };

                sw.Stop();

                _logger.Log("Exception occurs in PublishEventAsync", WFInstanceId: input.WorkflowInstanceId, response, sw.ElapsedMilliseconds, ex);

                return response;
            }
        }
        #endregion

        #region InquiryStateAsync
        /// <inheritdoc />
        public async Task<WorkflowResponse<WorkflowModel>> InquiryStateAsync(InquiryStateModel input)
        {
            var sw = Stopwatch.StartNew();
            try
            {
                _logger.Log("Start of InquiryStateAsync", WFInstanceId: input.WorkflowInstanceId, input);

                WorkflowResponse<WorkflowModel> response;

                var instance = await _workflowInstanceManager.FindByIdAsync(input.WorkflowInstanceId);

                if (instance == null)
                {
                    var msg = $"Workflow instance with id '{input.WorkflowInstanceId}' not found";
                    response = new WorkflowResponse<WorkflowModel>()
                    {
                        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;

                response = new WorkflowResponse<WorkflowModel>()
                {
                    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<WorkflowModel>()
                {
                    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
    }
}