behzadalizadeh/Caspian.Elsa.Core
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

Add services

Browse Source
This commit is contained in:
Behzad Alizadeh
2026-01-07 14:12:18 +03:30
parent 1759d2f21f
commit 9ef806b196
7 changed files with 497 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/apps/Caspian.WorkflowFramework.Sample/Controllers/WorkflowController.cs
+84
View File
@@ -47,6 +47,20 @@ public class WorkflowController : Controller, ILoggable
}
#endregion
#region CreateOrGetInstance
[HttpPost("CreateOrGetInstance")]
public async Task<IActionResult> 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<IActionResult> Do([FromBody] PublishEventModel input)
@@ -76,4 +90,74 @@ public class WorkflowController : Controller, ILoggable
return Json(result);
}
#endregion
#region Cancel
[HttpPost("Cancel")]
public async Task<IActionResult> 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<IActionResult> 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<IActionResult> 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<IActionResult> 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<IActionResult> 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
}
src/apps/Caspian.WorkflowFramework/Contracts/IWorkflowService.cs
+38 -8
View File
@@ -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
/// <summary>
/// Create and executes a new workflow instance.
/// </summary>
/// <param name="input"></param>
/// <returns>A new <see cref="WorkflowResponse<WorkflowModel>"/> object.</returns>
/// <remarks>Create a new workflow instance and executes it and waits for it to complete or reach a suspend point.</remarks>
Task<WorkflowResponse<WorkflowModel>> CreateAndExecuteAsync(CreateAndExecuteWorkflowModel input);
/// <summary>
/// Publishes the specified event.
/// Create a new workflow instance if it does not exist or get the running instance.
/// </summary>
Task<WorkflowResponse<WorkflowModel>> CreateOrGetInstanceAsync(CreateAndExecuteWorkflowModel input);
/// <summary>
/// Publishes the specified event to a workflow instance.
/// </summary>
/// <param name="input"></param>
/// <returns>A new <see cref="WorkflowResponse<WorkflowModel>"/> object.</returns>
Task<WorkflowResponse<WorkflowModel>> PublishEventAsync(PublishEventModel input);
/// <summary>
/// Inquiry last state of workflow instance.
/// </summary>
/// <param name="input"></param>
/// <returns>A new <see cref="WorkflowResponse<WorkflowModel>"/> object.</returns>
Task<WorkflowResponse<WorkflowModel>> InquiryStateAsync(InquiryStateModel input);
/// <summary>
/// Gets all active workflow instances for a specific user.
/// </summary>
Task<WorkflowResponse<List<WorkflowModel>>> GetActiveInstancesAsync(GetActiveInstancesModel input);
/// <summary>
/// Gets all active workflow instance IDs for a specific user.
/// </summary>
Task<WorkflowResponse<List<string>>> GetActiveInstanceIdsAsync(GetActiveInstancesModel input);
/// <summary>
/// Gets the latest active workflow instance for a specific user and workflow name.
/// </summary>
Task<WorkflowResponse<WorkflowModel>> GetLatestActiveInstanceAsync(GetLatestActiveInstanceModel input);
/// <summary>
/// Gets the ID of the latest active workflow instance for a specific user and workflow name.
/// </summary>
Task<WorkflowResponse<dynamic>> GetLatestActiveInstanceIdAsync(GetLatestActiveInstanceModel input);
/// <summary>
/// Cancels the specified workflow instance.
/// </summary>
Task<WorkflowResponse<dynamic>> CancelInstanceAsync(CancelWorkflowInstanceModel input);
///// <summary>
///// Evicts a workflow definition from cache (optional, commented out in service).
///// </summary>
//Task<WorkflowResponse<WorkflowModel>> EvictDefFromCacheAsync(string definitionId);
}
src/apps/Caspian.WorkflowFramework/Models/CancelWorkflowInstanceModel.cs
+6
View File
@@ -0,0 +1,6 @@
namespace Caspian.WorkflowFramework.Models;
public class CancelWorkflowInstanceModel
{
public string WorkflowInstanceId { get; set; }
public long UserId { get; set; }
}
src/apps/Caspian.WorkflowFramework/Models/GetActiveInstancesModel.cs
+5
View File
@@ -0,0 +1,5 @@
namespace Caspian.WorkflowFramework.Models;
public class GetActiveInstancesModel
{
public long UserId { get; set; }
}
src/apps/Caspian.WorkflowFramework/Models/GetLatestActiveInstanceModel.cs
+6
View File
@@ -0,0 +1,6 @@
namespace Caspian.WorkflowFramework.Models;
public class GetLatestActiveInstanceModel
{
public long UserId { get; set; }
public string WorkflowName { get; set; }
}
src/apps/Caspian.WorkflowFramework/Services/WorkflowService.cs
+346 -22
View File
@@ -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<WorkflowService> _logger;
@@ -40,6 +41,7 @@ public class WorkflowService : IWorkflowService, ILoggable
_workflowDefinitionService = _serviceProvider.GetRequiredService<IWorkflowDefinitionService>();
_workflowInstanceVariableManager = _serviceProvider.GetRequiredService<IWorkflowInstanceVariableManager>();
_workflowInstanceManager = _serviceProvider.GetRequiredService<IWorkflowInstanceManager>();
_workflowInstanceStore = _serviceProvider.GetRequiredService<IWorkflowInstanceStore>();
_eventPublisher = _serviceProvider.GetRequiredService<IEventPublisher>();
_logger = logger;
}
@@ -132,14 +134,14 @@ public class WorkflowService : IWorkflowService, ILoggable
}
#endregion
#region InitAsync
#region CreateOrGetInstanceAsync
/// <inheritdoc />
public async Task<WorkflowResponse<WorkflowModel>> InitAsync(CreateAndExecuteWorkflowModel input)
public async Task<WorkflowResponse<WorkflowModel>> CreateOrGetInstanceAsync(CreateAndExecuteWorkflowModel input)
{
var sw = Stopwatch.StartNew();
try
{
_logger.Log("Start of InitAsync", input);
_logger.Log("Start of CreateOrGetInstanceAsync", input);
WorkflowResponse<WorkflowModel> response;
@@ -163,19 +165,38 @@ 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 = "jhzbsddvjkf",//definition.DefinitionId,
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
@@ -186,14 +207,18 @@ public class WorkflowService : IWorkflowService, ILoggable
});
var result = await workflowClient.RunInstanceAsync(RunWorkflowInstanceRequest.Empty);
workflowInstanceId = result.WorkflowInstanceId;
}
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<WorkflowModel>()
@@ -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<WorkflowModel> 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<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;
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;
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
//#region EvictDefFromCacheAsync
///// <inheritdoc />
//public async Task<WorkflowResponse<WorkflowModel>> EvictDefFromCacheAsync(string definitionId)
src/modules/Elsa.EntityFrameworkCore/Elsa.EntityFrameworkCore.csproj
+1 -1
View File
@@ -4,7 +4,7 @@
<PackageId>Caspian.Elsa.EntityFrameworkCore</PackageId>
<Company>Caspian</Company>
<Authors>Digital banking team</Authors>
<Version>1.1.0-preview.1</Version>
<Version>1.1.0-preview.2</Version>
<!--<GeneratePackageOnBuild>True</GeneratePackageOnBuild>-->
<Description>
Provides Entity Framework Core implementations of various abstractions from various modules.