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

using System.Diagnostics;
using System.Dynamic;
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;
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;

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 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>();
        _workflowInstanceStore = _serviceProvider.GetRequiredService<IWorkflowInstanceStore>();
        _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;
            workflowModel.InstanceMode = "new";

            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 instances = await _workflowInstanceStore.FindManyAsync(
               new WorkflowInstanceFilter
               {
                   ExternalUserId = input.UserId,
                   DefinitionId = definition.DefinitionId,
                   WorkflowStatus = WorkflowStatus.Running
               },
               new WorkflowInstanceOrder<DateTimeOffset>(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<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 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<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, input.EventData, asynchronous: true);
            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;
            workflowModel.InstanceMode = "exist";

            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;
            workflowModel.InstanceMode = "exist";

            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

    #region GetActiveInstancesAsync
    public async Task<WorkflowResponse<List<WorkflowModel>>> 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<WorkflowModel>();

            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<List<WorkflowModel>>
            {
                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<List<WorkflowModel>>
            {
                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<WorkflowResponse<List<string>>> 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<List<string>>
            {
                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<List<string>>
            {
                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<WorkflowResponse<WorkflowModel>> 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<WorkflowModel>
                {
                    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<DateTimeOffset>(x => x.UpdatedAt!, OrderDirection.Descending)
            );

            var latest = instances.FirstOrDefault();

            if (latest == null)
            {
                return new WorkflowResponse<WorkflowModel>
                {
                    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<WorkflowModel>
            {
                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<WorkflowModel>
            {
                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<WorkflowResponse<dynamic>> 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<dynamic>
                {
                    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<DateTimeOffset>(x => x.UpdatedAt!, OrderDirection.Descending)
            );

            var latest = instances.FirstOrDefault();

            if (latest == null)
            {
                return new WorkflowResponse<dynamic>
                {
                    Status = 1,
                    Message = "No active workflow instance found",
                    Data = null
                };
            }

            var response = new WorkflowResponse<dynamic>
            {
                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<dynamic>
            {
                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<WorkflowResponse<dynamic>> 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<dynamic>
                {
                    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<dynamic>
            {
                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<dynamic>
            {
                Status = 100,
                Message = ex.Message,
                Data = false
            };

            sw.Stop();
            _logger.Log("Exception in CancelInstanceAsync", WFInstanceId: input.WorkflowInstanceId, response, sw.ElapsedMilliseconds, ex);
            return response;
        }
    }
    #endregion
}