diff --git a/Directory.Packages.props b/Directory.Packages.props index cafd4b063..f008be8d6 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -131,6 +131,9 @@ + + + diff --git a/Elsa.sln b/Elsa.sln index 15243aa3b..3d848fb0d 100644 --- a/Elsa.sln +++ b/Elsa.sln @@ -437,6 +437,12 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.AI.Persistence.EFCore. EndProject Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.AI.Persistence.EFCore.Oracle", "src\modules\Elsa.AI.Persistence.EFCore.Oracle\Elsa.AI.Persistence.EFCore.Oracle.csproj", "{743C1835-E41E-4E5C-9D12-585BAEE59EEE}" EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Platform.Integration", "src\modules\Elsa.Platform.Integration\Elsa.Platform.Integration.csproj", "{11F7C89F-835A-4D78-A846-9FCB463DA80C}" +EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Platform.Integration.UnitTests", "test\unit\Elsa.Platform.Integration.UnitTests\Elsa.Platform.Integration.UnitTests.csproj", "{84E70F39-2C24-4C93-A30F-6EC1330E2F49}" +EndProject +Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "platform", "platform", "{B284D6B0-E488-4062-ABC8-7AEB25784931}" +EndProject Global GlobalSection(SolutionConfigurationPlatforms) = preSolution Debug|Any CPU = Debug|Any CPU @@ -1957,6 +1963,30 @@ Global {743C1835-E41E-4E5C-9D12-585BAEE59EEE}.Release|x64.Build.0 = Release|Any CPU {743C1835-E41E-4E5C-9D12-585BAEE59EEE}.Release|x86.ActiveCfg = Release|Any CPU {743C1835-E41E-4E5C-9D12-585BAEE59EEE}.Release|x86.Build.0 = Release|Any CPU + {11F7C89F-835A-4D78-A846-9FCB463DA80C}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {11F7C89F-835A-4D78-A846-9FCB463DA80C}.Debug|Any CPU.Build.0 = Debug|Any CPU + {11F7C89F-835A-4D78-A846-9FCB463DA80C}.Debug|x64.ActiveCfg = Debug|Any CPU + {11F7C89F-835A-4D78-A846-9FCB463DA80C}.Debug|x64.Build.0 = Debug|Any CPU + {11F7C89F-835A-4D78-A846-9FCB463DA80C}.Debug|x86.ActiveCfg = Debug|Any CPU + {11F7C89F-835A-4D78-A846-9FCB463DA80C}.Debug|x86.Build.0 = Debug|Any CPU + {11F7C89F-835A-4D78-A846-9FCB463DA80C}.Release|Any CPU.ActiveCfg = Release|Any CPU + {11F7C89F-835A-4D78-A846-9FCB463DA80C}.Release|Any CPU.Build.0 = Release|Any CPU + {11F7C89F-835A-4D78-A846-9FCB463DA80C}.Release|x64.ActiveCfg = Release|Any CPU + {11F7C89F-835A-4D78-A846-9FCB463DA80C}.Release|x64.Build.0 = Release|Any CPU + {11F7C89F-835A-4D78-A846-9FCB463DA80C}.Release|x86.ActiveCfg = Release|Any CPU + {11F7C89F-835A-4D78-A846-9FCB463DA80C}.Release|x86.Build.0 = Release|Any CPU + {84E70F39-2C24-4C93-A30F-6EC1330E2F49}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {84E70F39-2C24-4C93-A30F-6EC1330E2F49}.Debug|Any CPU.Build.0 = Debug|Any CPU + {84E70F39-2C24-4C93-A30F-6EC1330E2F49}.Debug|x64.ActiveCfg = Debug|Any CPU + {84E70F39-2C24-4C93-A30F-6EC1330E2F49}.Debug|x64.Build.0 = Debug|Any CPU + {84E70F39-2C24-4C93-A30F-6EC1330E2F49}.Debug|x86.ActiveCfg = Debug|Any CPU + {84E70F39-2C24-4C93-A30F-6EC1330E2F49}.Debug|x86.Build.0 = Debug|Any CPU + {84E70F39-2C24-4C93-A30F-6EC1330E2F49}.Release|Any CPU.ActiveCfg = Release|Any CPU + {84E70F39-2C24-4C93-A30F-6EC1330E2F49}.Release|Any CPU.Build.0 = Release|Any CPU + {84E70F39-2C24-4C93-A30F-6EC1330E2F49}.Release|x64.ActiveCfg = Release|Any CPU + {84E70F39-2C24-4C93-A30F-6EC1330E2F49}.Release|x64.Build.0 = Release|Any CPU + {84E70F39-2C24-4C93-A30F-6EC1330E2F49}.Release|x86.ActiveCfg = Release|Any CPU + {84E70F39-2C24-4C93-A30F-6EC1330E2F49}.Release|x86.Build.0 = Release|Any CPU EndGlobalSection GlobalSection(SolutionProperties) = preSolution HideSolutionNode = FALSE @@ -2124,6 +2154,9 @@ Global {32F2EE33-8190-41D7-BD12-45AA02BEB930} = {F52DF980-C8F7-49D5-A8A1-249B44223600} {2D51FFE0-F114-48E1-BF58-FCED0FD1944F} = {F52DF980-C8F7-49D5-A8A1-249B44223600} {743C1835-E41E-4E5C-9D12-585BAEE59EEE} = {F52DF980-C8F7-49D5-A8A1-249B44223600} + {84E70F39-2C24-4C93-A30F-6EC1330E2F49} = {18453B51-25EB-4317-A4B3-B10518252E92} + {B284D6B0-E488-4062-ABC8-7AEB25784931} = {5BA4A8FA-F7F4-45B3-AEC8-8886D35AAC79} + {11F7C89F-835A-4D78-A846-9FCB463DA80C} = {B284D6B0-E488-4062-ABC8-7AEB25784931} EndGlobalSection GlobalSection(ExtensibilityGlobals) = postSolution SolutionGuid = {D4B5CEAA-7D70-4FCB-A68E-B03FBE5E0E5E} diff --git a/NuGet.Config b/NuGet.Config index f8618cb78..3a1e84216 100644 --- a/NuGet.Config +++ b/NuGet.Config @@ -7,6 +7,7 @@ + @@ -24,5 +25,9 @@ + + + + diff --git a/src/apps/Elsa.ModularServer.Web/Elsa.ModularServer.Web.csproj b/src/apps/Elsa.ModularServer.Web/Elsa.ModularServer.Web.csproj index 49c5a7920..a5e3635a9 100644 --- a/src/apps/Elsa.ModularServer.Web/Elsa.ModularServer.Web.csproj +++ b/src/apps/Elsa.ModularServer.Web/Elsa.ModularServer.Web.csproj @@ -25,6 +25,7 @@ + diff --git a/src/apps/Elsa.ModularServer.Web/Program.cs b/src/apps/Elsa.ModularServer.Web/Program.cs index e3cc5ff0f..c35f165f8 100644 --- a/src/apps/Elsa.ModularServer.Web/Program.cs +++ b/src/apps/Elsa.ModularServer.Web/Program.cs @@ -7,6 +7,7 @@ using Elsa.Diagnostics.ConsoleLogs.Dashboard.ShellFeatures; using Elsa.Diagnostics.StructuredLogs.Dashboard.ShellFeatures; using Elsa.ModularServer.Web; using Elsa.ModularServer.Web.Catalog; +using Elsa.Platform.Integration.ShellFeatures; using Elsa.ShellFeatures; using Elsa.Workflows.Api.ShellFeatures; using Elsa.Workflows.Management.ShellFeatures; @@ -29,6 +30,7 @@ ConsoleStreamHook.Install(); var builder = WebApplication.CreateBuilder(args); var services = builder.Services; var configuration = builder.Configuration; +configuration.AddJsonFile(configuration["Elsa:PlatformIntegration:ShellOverlayPath"] ?? "platform-shell-overrides.json", optional: true, reloadOnChange: false); var serviceVersion = typeof(Program).Assembly.GetName().Version?.ToString(); builder.Logging.AddOpenTelemetry(logging => @@ -77,6 +79,7 @@ builder.AddShells(shells => shells typeof(WorkflowRuntimeFeature), typeof(WorkflowsFeature), typeof(DistributedRuntimeFeature), + typeof(ElsaPlatformIntegrationFeature), typeof(DashboardApiFeature), typeof(WorkflowRuntimeDashboardFeature), typeof(ConsoleLogsDashboardFeature), diff --git a/src/apps/Elsa.ModularServer.Web/appsettings.Example.json b/src/apps/Elsa.ModularServer.Web/appsettings.Example.json index dfe40f9f7..f4ce0bfec 100644 --- a/src/apps/Elsa.ModularServer.Web/appsettings.Example.json +++ b/src/apps/Elsa.ModularServer.Web/appsettings.Example.json @@ -6,6 +6,18 @@ } }, "AllowedHosts": "*", + "Elsa": { + "PlatformIntegration": { + "Enabled": false, + "PlatformEndpoint": "https://localhost:7001", + "WorkspaceId": "00000000-0000-0000-0000-000000000000", + "EngineId": "00000000-0000-0000-0000-000000000000", + "EngineSecret": "", + "PollInterval": "00:00:05", + "ClaimLeaseDuration": "00:05:00", + "ShellOverlayPath": "platform-shell-overrides.json" + } + }, "CShells": { "Shells": [ { diff --git a/src/modules/Elsa.Platform.Integration/Elsa.Platform.Integration.csproj b/src/modules/Elsa.Platform.Integration/Elsa.Platform.Integration.csproj new file mode 100644 index 000000000..566a57559 --- /dev/null +++ b/src/modules/Elsa.Platform.Integration/Elsa.Platform.Integration.csproj @@ -0,0 +1,22 @@ + + + + + Integrates Elsa workflow runtimes with Elsa Platform deployment commands. + + elsa module platform integration deployment workflows + + + + + + + + + + + + + + + diff --git a/src/modules/Elsa.Platform.Integration/FodyWeavers.xml b/src/modules/Elsa.Platform.Integration/FodyWeavers.xml new file mode 100644 index 000000000..00e1d9a1c --- /dev/null +++ b/src/modules/Elsa.Platform.Integration/FodyWeavers.xml @@ -0,0 +1,3 @@ + + + \ No newline at end of file diff --git a/src/modules/Elsa.Platform.Integration/Models/PlatformRecipeArtifact.cs b/src/modules/Elsa.Platform.Integration/Models/PlatformRecipeArtifact.cs new file mode 100644 index 000000000..082a56398 --- /dev/null +++ b/src/modules/Elsa.Platform.Integration/Models/PlatformRecipeArtifact.cs @@ -0,0 +1,15 @@ +namespace Elsa.Platform.Integration.Models; + +public sealed class PlatformRecipeArtifact(IReadOnlyDictionary textEntries) +{ + private readonly IReadOnlyDictionary _textEntries = textEntries; + + public bool TryGetText(string path, out string content) + { + var normalizedPath = NormalizePath(path); + return _textEntries.TryGetValue(normalizedPath, out content!); + } + + private static string NormalizePath(string path) => + path.Replace('\\', '/').TrimStart('/'); +} diff --git a/src/modules/Elsa.Platform.Integration/Models/PlatformRuntimeContracts.cs b/src/modules/Elsa.Platform.Integration/Models/PlatformRuntimeContracts.cs new file mode 100644 index 000000000..1ce060357 --- /dev/null +++ b/src/modules/Elsa.Platform.Integration/Models/PlatformRuntimeContracts.cs @@ -0,0 +1,138 @@ +using System.Text.Json.Serialization; + +namespace Elsa.Platform.Integration.Models; + +[JsonConverter(typeof(JsonStringEnumConverter))] +public enum PlatformRuntimeCommandAction +{ + Unknown, + Deploy, + Rollback, + Validate, + RuntimeControl +} + +[JsonConverter(typeof(JsonStringEnumConverter))] +public enum PlatformRuntimeCommandStatus +{ + Unknown, + Pending, + Claimed, + Running, + Completed, + Failed, + Rejected, + Cancelled, + RecoveryRequired, + Expired +} + +[JsonConverter(typeof(JsonStringEnumConverter))] +public enum PlatformArtifactStatus +{ + Pending, + Downloading, + Validated, + Applying, + Applied, + Failed, + Rejected, + Skipped +} + +[JsonConverter(typeof(JsonStringEnumConverter))] +public enum PlatformDiagnosticSeverity +{ + Info, + Warning, + Error +} + +public sealed record PlatformRuntimeCommandListResponse(IReadOnlyList Commands); + +public sealed record PlatformRuntimeCommandClaimRequest(Guid EngineId, string WorkerId, int LeaseSeconds); + +public sealed record PlatformRuntimeCommandClaimResponse(PlatformRuntimeCommand Command, string LeaseToken); + +public sealed record PlatformRuntimeCommandHeartbeatRequest(string LeaseToken, string WorkerId); + +public sealed record PlatformRuntimeCommandProgressRequest(string LeaseToken, string Status, int? PercentComplete, string Message); + +public sealed record PlatformRuntimeCommandCompleteRequest( + string LeaseToken, + PlatformArtifactDigest? ObservedArtifactDigest, + string? RuntimeReference, + IReadOnlyList Diagnostics, + IReadOnlyList? Artifacts = null); + +public sealed record PlatformRuntimeCommandFailRequest( + string LeaseToken, + IReadOnlyList Diagnostics, + IReadOnlyList? Artifacts = null); + +public sealed record PlatformRuntimeCommandRejectRequest( + string LeaseToken, + IReadOnlyList Diagnostics, + IReadOnlyList? Artifacts = null); + +public sealed record PlatformRuntimeCommand( + Guid Id, + Guid WorkspaceId, + Guid RunId, + Guid EnvironmentId, + Guid EngineId, + PlatformRuntimeCommandAction Action, + PlatformRuntimeCommandStatus Status, + PlatformRuntimeCommandArtifactReference? Artifact, + PlatformRuntimeCommandRevisionReference? Revision, + string IdempotencyKey, + string? WorkerId, + DateTimeOffset? ClaimedAt, + DateTimeOffset? LeaseExpiresAt, + DateTimeOffset? HeartbeatAt, + int AttemptNumber, + int? PercentComplete, + string? ProgressMessage, + PlatformArtifactDigest? ObservedArtifactDigest, + string? RuntimeReference, + IReadOnlyList Diagnostics, + DateTimeOffset CreatedAt, + DateTimeOffset UpdatedAt, + DateTimeOffset? AvailableAt, + DateTimeOffset? ExpiresAt, + DateTimeOffset? CompletedAt, + IReadOnlyList? Artifacts = null); + +public sealed record PlatformRuntimeCommandArtifactReference( + Guid? ArtifactRecordId, + string? ArtifactId, + string? ArtifactTypeId, + PlatformArtifactDigest? ContentDigest); + +public sealed record PlatformRuntimeCommandRevisionReference(Guid? RevisionId); + +public sealed record PlatformArtifactItem( + Guid ArtifactRecordId, + string ArtifactId, + string ArtifactTypeId, + string? ArtifactSchemaVersion, + PlatformArtifactDigest ContentDigest, + string DisplayName, + string? DownloadUrl, + PlatformArtifactStatus Status, + PlatformArtifactDigest? ObservedDigest, + string? RuntimeReference, + IReadOnlyList? Diagnostics); + +public sealed record PlatformArtifactOutcome( + Guid ArtifactRecordId, + PlatformArtifactStatus Status, + PlatformArtifactDigest? ObservedDigest = null, + string? RuntimeReference = null, + IReadOnlyList? Diagnostics = null); + +public sealed record PlatformArtifactDigest(string Algorithm, string Value); + +public sealed record PlatformDiagnostic(string Code, PlatformDiagnosticSeverity Severity, string Message); + +public sealed record PlatformProblemDetails(string? Title, string? Detail); diff --git a/src/modules/Elsa.Platform.Integration/Options/ElsaPlatformIntegrationOptions.cs b/src/modules/Elsa.Platform.Integration/Options/ElsaPlatformIntegrationOptions.cs new file mode 100644 index 000000000..a798fe80d --- /dev/null +++ b/src/modules/Elsa.Platform.Integration/Options/ElsaPlatformIntegrationOptions.cs @@ -0,0 +1,53 @@ +namespace Elsa.Platform.Integration.Options; + +public class ElsaPlatformIntegrationOptions +{ + public const string ConfigurationSection = "Elsa:PlatformIntegration"; + + public bool Enabled { get; set; } + + public Uri? PlatformEndpoint { get; set; } + + public Guid WorkspaceId { get; set; } + + public Guid EngineId { get; set; } + + public string? EngineSecret { get; set; } + + public string WorkerId { get; set; } = $"{Environment.MachineName}:{Environment.ProcessId}"; + + public TimeSpan PollInterval { get; set; } = TimeSpan.FromSeconds(5); + + public TimeSpan ClaimLeaseDuration { get; set; } = TimeSpan.FromMinutes(5); + + public long MaxArtifactBytes { get; set; } = 4 * 1024 * 1024; + + public string ShellOverlayPath { get; set; } = "platform-shell-overrides.json"; + + public IReadOnlyList Capabilities { get; set; } = ["loom.recipe.apply"]; + + public void Validate() + { + if (!Enabled) + return; + + if (PlatformEndpoint is null) + throw new InvalidOperationException("Elsa Platform endpoint is required when Platform integration is enabled."); + if (WorkspaceId == Guid.Empty) + throw new InvalidOperationException("Elsa Platform workspace ID is required when Platform integration is enabled."); + if (EngineId == Guid.Empty) + throw new InvalidOperationException("Elsa Platform engine ID is required when Platform integration is enabled."); + if (string.IsNullOrWhiteSpace(EngineSecret)) + throw new InvalidOperationException("Elsa Platform engine secret is required when Platform integration is enabled."); + if (string.IsNullOrWhiteSpace(WorkerId)) + throw new InvalidOperationException("Elsa Platform worker ID is required when Platform integration is enabled."); + if (PollInterval <= TimeSpan.Zero) + throw new InvalidOperationException("Elsa Platform poll interval must be positive."); + if (ClaimLeaseDuration < TimeSpan.FromSeconds(1) || ClaimLeaseDuration.TotalSeconds > int.MaxValue) + throw new InvalidOperationException("Elsa Platform claim lease duration must be between 1 second and the maximum supported Platform lease."); + if (MaxArtifactBytes <= 0 || MaxArtifactBytes > Array.MaxLength) + throw new InvalidOperationException("Elsa Platform maximum artifact size must be between 1 byte and the maximum runtime buffer size."); + if (string.IsNullOrWhiteSpace(ShellOverlayPath)) + throw new InvalidOperationException("Elsa Platform shell overlay path is required when Platform integration is enabled."); + } +} diff --git a/src/modules/Elsa.Platform.Integration/Services/ElsaLoomRecipeArtifactApplier.cs b/src/modules/Elsa.Platform.Integration/Services/ElsaLoomRecipeArtifactApplier.cs new file mode 100644 index 000000000..147145930 --- /dev/null +++ b/src/modules/Elsa.Platform.Integration/Services/ElsaLoomRecipeArtifactApplier.cs @@ -0,0 +1,209 @@ +using System.IO.Compression; +using System.Security.Cryptography; +using CShells.Lifecycle; +using Elsa.Platform.Integration.Models; +using Elsa.Platform.Integration.Steps; +using Loom; + +namespace Elsa.Platform.Integration.Services; + +public class ElsaLoomRecipeArtifactApplier( + IServiceProvider serviceProvider, + IShellRegistry shellRegistry) : IPlatformRecipeArtifactApplier +{ + private static readonly JsonRecipeSerializer Serializer = new(); + + public async Task ApplyAsync( + PlatformRuntimeCommand command, + PlatformArtifactItem artifact, + Stream artifactZip, + CancellationToken cancellationToken = default) + { + var observedDigest = await ComputeDigestAsync(artifactZip, cancellationToken); + if (!DigestEquals(observedDigest, artifact.ContentDigest)) + { + return Rejected( + observedDigest, + "elsa-platform.artifact-digest-mismatch", + "Downloaded recipe artifact digest did not match the Platform command digest."); + } + + var textEntries = await ReadTextEntriesAsync(artifactZip, cancellationToken); + var recipeJson = FindRecipeJson(textEntries); + if (recipeJson is null) + return Rejected(observedDigest, "elsa-platform.recipe-payload-missing", "Loom recipe artifact ZIP did not contain a recipe JSON payload."); + + Recipe recipe; + try + { + recipe = Serializer.Deserialize(recipeJson); + } + catch (RecipeSerializationException ex) + { + return Rejected(observedDigest, "elsa-platform.recipe-payload-invalid", ex.Message); + } + + var reloadTracker = new PlatformShellReloadTracker(); + var recipeServices = new PlatformRecipeServiceProvider( + serviceProvider, + new Dictionary + { + [typeof(PlatformRecipeArtifact)] = new PlatformRecipeArtifact(textEntries), + [typeof(PlatformShellReloadTracker)] = reloadTracker + }); + + var engine = RecipeEngine.Create() + .RegisterStep() + .RegisterStep() + .RegisterStep() + .RegisterStep(); + + var runResult = await engine.RunAsync(recipe, new RecipeRunOptions + { + Services = recipeServices + }, cancellationToken); + + if (!runResult.Succeeded) + { + var status = runResult.Status == RecipeRunStatus.ValidationFailed + ? PlatformArtifactStatus.Rejected + : PlatformArtifactStatus.Failed; + return new PlatformRecipeArtifactApplyResult( + status, + observedDigest, + RuntimeReference(recipe), + ToPlatformDiagnostics(runResult.Diagnostics, runResult.Error)); + } + + var reloadFailure = await ReloadShellsAsync(reloadTracker, cancellationToken); + if (reloadFailure is not null) + { + return new PlatformRecipeArtifactApplyResult( + PlatformArtifactStatus.Failed, + observedDigest, + RuntimeReference(recipe), + [reloadFailure]); + } + + var diagnostics = ToPlatformDiagnostics(runResult.Diagnostics, null); + if (diagnostics.Count == 0) + diagnostics = [PlatformDiagnosticSanitizer.Info("elsa-platform.recipe-applied", "Loom recipe artifact was applied.")]; + + return new PlatformRecipeArtifactApplyResult( + PlatformArtifactStatus.Applied, + observedDigest, + RuntimeReference(recipe), + diagnostics); + } + + private async Task ReloadShellsAsync( + PlatformShellReloadTracker reloadTracker, + CancellationToken cancellationToken) + { + foreach (var shellId in reloadTracker.ShellIds) + { + var result = await shellRegistry.ReloadAsync(shellId, cancellationToken); + if (result.Error is not null) + { + return PlatformDiagnosticSanitizer.Error( + "elsa-platform.shell-reload-failed", + $"Shell '{shellId}' reload failed: {result.Error.Message}"); + } + } + + return null; + } + + private static async Task> ReadTextEntriesAsync( + Stream artifactZip, + CancellationToken cancellationToken) + { + artifactZip.Position = 0; + using var archive = new ZipArchive(artifactZip, ZipArchiveMode.Read, leaveOpen: true); + var entries = new Dictionary(StringComparer.OrdinalIgnoreCase); + foreach (var entry in archive.Entries.OrderBy(x => x.FullName, StringComparer.Ordinal)) + { + if (string.IsNullOrWhiteSpace(entry.Name) || !entry.FullName.EndsWith(".json", StringComparison.OrdinalIgnoreCase)) + continue; + + await using var stream = entry.Open(); + using var reader = new StreamReader(stream); + entries[NormalizePath(entry.FullName)] = await reader.ReadToEndAsync(cancellationToken); + } + + return entries; + } + + private static string? FindRecipeJson(IReadOnlyDictionary textEntries) + { + var recipePath = textEntries.Keys + .Where(x => x.StartsWith("payload/recipes/", StringComparison.OrdinalIgnoreCase)) + .Where(x => x.EndsWith(".json", StringComparison.OrdinalIgnoreCase)) + .OrderBy(x => x, StringComparer.Ordinal) + .FirstOrDefault(); + + if (recipePath is not null) + return textEntries[recipePath]; + + foreach (var candidate in new[] { "recipe.json", "loom.recipe.json" }) + { + if (textEntries.TryGetValue(candidate, out var recipeJson)) + return recipeJson; + } + + return null; + } + + private static async Task ComputeDigestAsync( + Stream stream, + CancellationToken cancellationToken) + { + stream.Position = 0; + using var sha = SHA256.Create(); + var hash = await sha.ComputeHashAsync(stream, cancellationToken); + stream.Position = 0; + return new PlatformArtifactDigest("sha256", Convert.ToHexString(hash).ToLowerInvariant()); + } + + private static IReadOnlyList ToPlatformDiagnostics( + IReadOnlyList diagnostics, + string? fallbackError) + { + var platformDiagnostics = diagnostics.Select(ToPlatformDiagnostic).ToList(); + if (platformDiagnostics.Count == 0 && !string.IsNullOrWhiteSpace(fallbackError)) + platformDiagnostics.Add(PlatformDiagnosticSanitizer.Error("elsa-platform.recipe-failed", fallbackError)); + + return platformDiagnostics; + } + + private static PlatformDiagnostic ToPlatformDiagnostic(RecipeDiagnostic diagnostic) + { + var message = diagnostic.ExceptionSummary is null + ? diagnostic.Message + : $"{diagnostic.Message} {diagnostic.ExceptionSummary}"; + return diagnostic.Severity switch + { + DiagnosticSeverity.Information => PlatformDiagnosticSanitizer.Info(diagnostic.Code, message), + DiagnosticSeverity.Warning => PlatformDiagnosticSanitizer.Warning(diagnostic.Code, message), + _ => PlatformDiagnosticSanitizer.Error(diagnostic.Code, message) + }; + } + + private static bool DigestEquals(PlatformArtifactDigest left, PlatformArtifactDigest right) => + left.Algorithm.Equals(right.Algorithm, StringComparison.OrdinalIgnoreCase) + && left.Value.Equals(right.Value, StringComparison.OrdinalIgnoreCase); + + private static PlatformRecipeArtifactApplyResult Rejected( + PlatformArtifactDigest observedDigest, + string code, + string message) => + new(PlatformArtifactStatus.Rejected, observedDigest, null, [PlatformDiagnosticSanitizer.Error(code, message)]); + + private static string RuntimeReference(Recipe recipe) => + recipe.Version is null + ? $"elsa://loom-recipes/{Uri.EscapeDataString(recipe.Name)}" + : $"elsa://loom-recipes/{Uri.EscapeDataString(recipe.Name)}@{Uri.EscapeDataString(recipe.Version)}"; + + private static string NormalizePath(string path) => + path.Replace('\\', '/').TrimStart('/'); +} diff --git a/src/modules/Elsa.Platform.Integration/Services/ElsaPlatformDeploymentWorker.cs b/src/modules/Elsa.Platform.Integration/Services/ElsaPlatformDeploymentWorker.cs new file mode 100644 index 000000000..64a2c951a --- /dev/null +++ b/src/modules/Elsa.Platform.Integration/Services/ElsaPlatformDeploymentWorker.cs @@ -0,0 +1,157 @@ +using Elsa.Platform.Integration.Models; +using Elsa.Platform.Integration.Options; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; + +namespace Elsa.Platform.Integration.Services; + +public class ElsaPlatformDeploymentWorker( + IServiceScopeFactory scopeFactory, + IOptions options, + ILogger logger) : BackgroundService +{ + private readonly ElsaPlatformIntegrationOptions _options = options.Value; + + protected override async Task ExecuteAsync(CancellationToken stoppingToken) + { + _options.Validate(); + if (!_options.Enabled) + return; + + while (!stoppingToken.IsCancellationRequested) + { + try + { + await ProcessAvailableCommandsAsync(stoppingToken); + } + catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) + { + return; + } + catch (Exception ex) + { + logger.LogError(ex, "Elsa Platform deployment worker failed while polling commands."); + } + + await Task.Delay(_options.PollInterval, stoppingToken); + } + } + + private async Task ProcessAvailableCommandsAsync(CancellationToken cancellationToken) + { + await using var scope = scopeFactory.CreateAsyncScope(); + var client = scope.ServiceProvider.GetRequiredService(); + var applier = scope.ServiceProvider.GetRequiredService(); + var commands = await client.PollAsync(cancellationToken); + + foreach (var command in commands) + { + var claim = await client.ClaimAsync(command.Id, cancellationToken); + if (claim is null) + continue; + + await ProcessClaimedCommandAsync(client, applier, claim, cancellationToken); + } + } + + private async Task ProcessClaimedCommandAsync( + IPlatformRuntimeCommandClient client, + IPlatformRecipeArtifactApplier applier, + PlatformRuntimeCommandClaimResponse claim, + CancellationToken cancellationToken) + { + var command = claim.Command; + if (command.Action is not PlatformRuntimeCommandAction.Deploy and not PlatformRuntimeCommandAction.Rollback) + { + await client.RejectAsync( + command.Id, + new PlatformRuntimeCommandRejectRequest( + claim.LeaseToken, + [PlatformDiagnosticSanitizer.Error("elsa-platform.command-unsupported", "Runtime command action is not supported by this Elsa runtime.")]), + cancellationToken); + return; + } + + if (command.Artifacts is not { Count: > 0 }) + { + await client.RejectAsync( + command.Id, + new PlatformRuntimeCommandRejectRequest( + claim.LeaseToken, + [PlatformDiagnosticSanitizer.Error("elsa-platform.artifact-missing", "Runtime command did not include deployment artifacts.")]), + cancellationToken); + return; + } + + var outcomes = new List(); + try + { + foreach (var artifact in command.Artifacts) + { + await client.ReportProgressAsync(command.Id, claim.LeaseToken, "downloading", 20, "Downloading recipe artifact.", cancellationToken); + await using var artifactZip = await client.DownloadArtifactAsync(command, artifact, claim.LeaseToken, cancellationToken); + await client.ReportProgressAsync(command.Id, claim.LeaseToken, "applying", 60, "Applying recipe artifact.", cancellationToken); + var result = await applier.ApplyAsync(command, artifact, artifactZip, cancellationToken); + outcomes.Add(new PlatformArtifactOutcome( + artifact.ArtifactRecordId, + result.Status, + result.ObservedDigest, + result.RuntimeReference, + result.Diagnostics)); + + if (!result.Succeeded) + break; + } + + var failed = outcomes.FirstOrDefault(x => x.Status == PlatformArtifactStatus.Failed); + if (failed is not null) + { + await client.FailAsync( + command.Id, + new PlatformRuntimeCommandFailRequest( + claim.LeaseToken, + failed.Diagnostics ?? [PlatformDiagnosticSanitizer.Error("elsa-platform.artifact-failed", "Recipe artifact apply failed.")], + outcomes), + cancellationToken); + return; + } + + var rejected = outcomes.FirstOrDefault(x => x.Status == PlatformArtifactStatus.Rejected); + if (rejected is not null) + { + await client.RejectAsync( + command.Id, + new PlatformRuntimeCommandRejectRequest( + claim.LeaseToken, + rejected.Diagnostics ?? [PlatformDiagnosticSanitizer.Error("elsa-platform.artifact-rejected", "Recipe artifact apply was rejected.")], + outcomes), + cancellationToken); + return; + } + + var first = outcomes.FirstOrDefault(); + await client.CompleteAsync( + command.Id, + new PlatformRuntimeCommandCompleteRequest( + claim.LeaseToken, + first?.ObservedDigest, + first?.RuntimeReference, + [PlatformDiagnosticSanitizer.Info("elsa-platform.command-completed", "Recipe deployment command completed.")], + outcomes), + cancellationToken); + } + catch (Exception ex) when (ex is not OperationCanceledException) + { + logger.LogError(ex, "Elsa Platform deployment worker failed to apply command {CommandId}.", command.Id); + await client.FailAsync( + command.Id, + new PlatformRuntimeCommandFailRequest( + claim.LeaseToken, + [PlatformDiagnosticSanitizer.Error("elsa-platform.command-failed", ex.Message)], + outcomes), + cancellationToken); + } + } +} diff --git a/src/modules/Elsa.Platform.Integration/Services/FileShellConfigurationOverlayStore.cs b/src/modules/Elsa.Platform.Integration/Services/FileShellConfigurationOverlayStore.cs new file mode 100644 index 000000000..40f03d2a3 --- /dev/null +++ b/src/modules/Elsa.Platform.Integration/Services/FileShellConfigurationOverlayStore.cs @@ -0,0 +1,148 @@ +using System.Text.Json; +using System.Text.Json.Nodes; +using Elsa.Platform.Integration.Options; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Options; + +namespace Elsa.Platform.Integration.Services; + +public sealed class FileShellConfigurationOverlayStore( + IOptions options, + IHostEnvironment environment) : IShellConfigurationOverlayStore +{ + private static readonly JsonSerializerOptions JsonOptions = new(JsonSerializerDefaults.Web) + { + WriteIndented = true + }; + private readonly SemaphoreSlim _semaphore = new(1, 1); + + public async Task ConfigureFeaturesAsync( + string shellId, + IReadOnlyDictionary? enabledFeatures, + IReadOnlyList? disabledFeatures, + CancellationToken cancellationToken = default) + { + await _semaphore.WaitAsync(cancellationToken); + try + { + var root = await LoadAsync(cancellationToken); + var before = Serialize(root); + var features = EnsureObject(root, "CShells", "Shells", shellId, "Features"); + + foreach (var feature in enabledFeatures ?? new Dictionary()) + features[feature.Key] = ToNode(feature.Value) ?? new JsonObject(); + + foreach (var featureId in disabledFeatures ?? []) + features.Remove(featureId); + + var state = EnsureObject(root, "Elsa", "PlatformIntegration", "Shells", shellId); + var disabledFeatureArray = new JsonArray(); + foreach (var featureId in disabledFeatures ?? []) + disabledFeatureArray.Add(featureId); + state["DisabledFeatures"] = disabledFeatureArray; + + return await SaveIfChangedAsync(root, before, cancellationToken); + } + finally + { + _semaphore.Release(); + } + } + + public async Task ConfigureSettingsAsync( + string shellId, + JsonElement settings, + CancellationToken cancellationToken = default) + { + await _semaphore.WaitAsync(cancellationToken); + try + { + var root = await LoadAsync(cancellationToken); + var before = Serialize(root); + var configuration = EnsureObject(root, "CShells", "Shells", shellId, "Configuration"); + if (ToNode(settings) is JsonObject settingsObject) + Merge(configuration, settingsObject); + + return await SaveIfChangedAsync(root, before, cancellationToken); + } + finally + { + _semaphore.Release(); + } + } + + private async Task LoadAsync(CancellationToken cancellationToken) + { + var path = GetPath(); + if (!File.Exists(path)) + return new JsonObject(); + + await using var stream = File.OpenRead(path); + return await JsonNode.ParseAsync(stream, cancellationToken: cancellationToken) as JsonObject + ?? throw new InvalidOperationException("The Platform shell overlay file must contain a JSON object."); + } + + private async Task SaveIfChangedAsync(JsonObject root, string before, CancellationToken cancellationToken) + { + var after = Serialize(root); + if (string.Equals(before, after, StringComparison.Ordinal)) + return false; + + var path = GetPath(); + var directory = Path.GetDirectoryName(path); + if (!string.IsNullOrWhiteSpace(directory)) + Directory.CreateDirectory(directory); + + var tempPath = $"{path}.{Guid.NewGuid():N}.tmp"; + await File.WriteAllTextAsync(tempPath, after, cancellationToken); + File.Move(tempPath, path, overwrite: true); + return true; + } + + private string GetPath() + { + var path = options.Value.ShellOverlayPath; + return Path.IsPathRooted(path) ? path : Path.Combine(environment.ContentRootPath, path); + } + + private static JsonObject EnsureObject(JsonObject root, params string[] path) + { + JsonObject current = root; + foreach (var segment in path) + { + if (current[segment] is not JsonObject child) + { + child = new JsonObject(); + current[segment] = child; + } + + current = child; + } + + return current; + } + + private static JsonNode? ToNode(JsonElement element) => + element.ValueKind switch + { + JsonValueKind.Undefined or JsonValueKind.Null => null, + _ => JsonNode.Parse(element.GetRawText()) + }; + + private static void Merge(JsonObject target, JsonObject source) + { + foreach (var property in source) + { + if (property.Value is JsonObject sourceObject && target[property.Key] is JsonObject targetObject) + { + Merge(targetObject, sourceObject); + continue; + } + + target[property.Key] = property.Value?.DeepClone(); + } + } + + private static string Serialize(JsonObject root) => + root.ToJsonString(JsonOptions); +} diff --git a/src/modules/Elsa.Platform.Integration/Services/IPlatformRecipeArtifactApplier.cs b/src/modules/Elsa.Platform.Integration/Services/IPlatformRecipeArtifactApplier.cs new file mode 100644 index 000000000..9438b522f --- /dev/null +++ b/src/modules/Elsa.Platform.Integration/Services/IPlatformRecipeArtifactApplier.cs @@ -0,0 +1,21 @@ +using Elsa.Platform.Integration.Models; + +namespace Elsa.Platform.Integration.Services; + +public interface IPlatformRecipeArtifactApplier +{ + Task ApplyAsync( + PlatformRuntimeCommand command, + PlatformArtifactItem artifact, + Stream artifactZip, + CancellationToken cancellationToken = default); +} + +public sealed record PlatformRecipeArtifactApplyResult( + PlatformArtifactStatus Status, + PlatformArtifactDigest? ObservedDigest, + string? RuntimeReference, + IReadOnlyList Diagnostics) +{ + public bool Succeeded => Status == PlatformArtifactStatus.Applied; +} diff --git a/src/modules/Elsa.Platform.Integration/Services/IPlatformRuntimeCommandClient.cs b/src/modules/Elsa.Platform.Integration/Services/IPlatformRuntimeCommandClient.cs new file mode 100644 index 000000000..7b24798c6 --- /dev/null +++ b/src/modules/Elsa.Platform.Integration/Services/IPlatformRuntimeCommandClient.cs @@ -0,0 +1,20 @@ +using Elsa.Platform.Integration.Models; + +namespace Elsa.Platform.Integration.Services; + +public interface IPlatformRuntimeCommandClient +{ + Task> PollAsync(CancellationToken cancellationToken = default); + + Task ClaimAsync(Guid commandId, CancellationToken cancellationToken = default); + + Task DownloadArtifactAsync(PlatformRuntimeCommand command, PlatformArtifactItem artifact, string leaseToken, CancellationToken cancellationToken = default); + + Task ReportProgressAsync(Guid commandId, string leaseToken, string status, int? percentComplete, string message, CancellationToken cancellationToken = default); + + Task CompleteAsync(Guid commandId, PlatformRuntimeCommandCompleteRequest request, CancellationToken cancellationToken = default); + + Task FailAsync(Guid commandId, PlatformRuntimeCommandFailRequest request, CancellationToken cancellationToken = default); + + Task RejectAsync(Guid commandId, PlatformRuntimeCommandRejectRequest request, CancellationToken cancellationToken = default); +} diff --git a/src/modules/Elsa.Platform.Integration/Services/IShellConfigurationOverlayStore.cs b/src/modules/Elsa.Platform.Integration/Services/IShellConfigurationOverlayStore.cs new file mode 100644 index 000000000..b567706fc --- /dev/null +++ b/src/modules/Elsa.Platform.Integration/Services/IShellConfigurationOverlayStore.cs @@ -0,0 +1,17 @@ +using System.Text.Json; + +namespace Elsa.Platform.Integration.Services; + +public interface IShellConfigurationOverlayStore +{ + Task ConfigureFeaturesAsync( + string shellId, + IReadOnlyDictionary? enabledFeatures, + IReadOnlyList? disabledFeatures, + CancellationToken cancellationToken = default); + + Task ConfigureSettingsAsync( + string shellId, + JsonElement settings, + CancellationToken cancellationToken = default); +} diff --git a/src/modules/Elsa.Platform.Integration/Services/PlatformDiagnosticSanitizer.cs b/src/modules/Elsa.Platform.Integration/Services/PlatformDiagnosticSanitizer.cs new file mode 100644 index 000000000..55ba3f2da --- /dev/null +++ b/src/modules/Elsa.Platform.Integration/Services/PlatformDiagnosticSanitizer.cs @@ -0,0 +1,28 @@ +using System.Text.RegularExpressions; +using Elsa.Platform.Integration.Models; + +namespace Elsa.Platform.Integration.Services; + +public static partial class PlatformDiagnosticSanitizer +{ + public static PlatformDiagnostic Info(string code, string message) => + new(code, PlatformDiagnosticSeverity.Info, SafeMessage(message)); + + public static PlatformDiagnostic Warning(string code, string message) => + new(code, PlatformDiagnosticSeverity.Warning, SafeMessage(message)); + + public static PlatformDiagnostic Error(string code, string message) => + new(code, PlatformDiagnosticSeverity.Error, SafeMessage(message)); + + public static string SafeMessage(string? message) + { + if (string.IsNullOrWhiteSpace(message)) + return "No diagnostic details were provided."; + + var safe = SensitiveWordsRegex().Replace(message, "[redacted]"); + return safe.Length <= 500 ? safe : safe[..500]; + } + + [GeneratedRegex("(api[-_ ]?key|bearer|client[-_ ]?secret|password|private[-_ ]?key|secret|token)", RegexOptions.IgnoreCase | RegexOptions.CultureInvariant)] + private static partial Regex SensitiveWordsRegex(); +} diff --git a/src/modules/Elsa.Platform.Integration/Services/PlatformRecipeServiceProvider.cs b/src/modules/Elsa.Platform.Integration/Services/PlatformRecipeServiceProvider.cs new file mode 100644 index 000000000..f1c12d241 --- /dev/null +++ b/src/modules/Elsa.Platform.Integration/Services/PlatformRecipeServiceProvider.cs @@ -0,0 +1,9 @@ +namespace Elsa.Platform.Integration.Services; + +internal sealed class PlatformRecipeServiceProvider( + IServiceProvider inner, + IReadOnlyDictionary services) : IServiceProvider +{ + public object? GetService(Type serviceType) => + services.TryGetValue(serviceType, out var service) ? service : inner.GetService(serviceType); +} diff --git a/src/modules/Elsa.Platform.Integration/Services/PlatformRuntimeCommandClient.cs b/src/modules/Elsa.Platform.Integration/Services/PlatformRuntimeCommandClient.cs new file mode 100644 index 000000000..77d351ef7 --- /dev/null +++ b/src/modules/Elsa.Platform.Integration/Services/PlatformRuntimeCommandClient.cs @@ -0,0 +1,149 @@ +using System.Net; +using System.Net.Http.Json; +using System.Text.Json; +using System.Text.Json.Serialization; +using Elsa.Platform.Integration.Models; +using Elsa.Platform.Integration.Options; +using Microsoft.Extensions.Options; + +namespace Elsa.Platform.Integration.Services; + +public class PlatformRuntimeCommandClient(HttpClient httpClient, IOptions options) : IPlatformRuntimeCommandClient +{ + private const string EngineSecretHeaderName = "X-Elsa-Engine-Secret"; + private const string LeaseHeaderName = "X-Elsa-Command-Lease"; + private const string WorkerHeaderName = "X-Elsa-Worker-Id"; + private static readonly JsonSerializerOptions JsonOptions = CreateJsonOptions(); + private readonly ElsaPlatformIntegrationOptions _options = options.Value; + + public async Task> PollAsync(CancellationToken cancellationToken = default) + { + using var request = CreateRequest(HttpMethod.Get, BuildUri($"/deployments/runtime/engines/{_options.EngineId:D}/commands")); + using var response = await httpClient.SendAsync(request, cancellationToken); + if (response.StatusCode is HttpStatusCode.Unauthorized or HttpStatusCode.Forbidden) + throw new InvalidOperationException("Elsa Platform runtime command poll was not authorized."); + + response.EnsureSuccessStatusCode(); + var body = await response.Content.ReadFromJsonAsync(JsonOptions, cancellationToken); + return body?.Commands ?? []; + } + + public async Task ClaimAsync(Guid commandId, CancellationToken cancellationToken = default) + { + using var response = await SendJsonAsync( + BuildUri($"/deployments/runtime/commands/{commandId:D}/claim"), + new PlatformRuntimeCommandClaimRequest(_options.EngineId, _options.WorkerId, (int)_options.ClaimLeaseDuration.TotalSeconds), + cancellationToken); + + if (response.StatusCode == HttpStatusCode.Conflict || response.StatusCode == HttpStatusCode.NotFound) + return null; + + response.EnsureSuccessStatusCode(); + return await response.Content.ReadFromJsonAsync(JsonOptions, cancellationToken); + } + + public async Task DownloadArtifactAsync( + PlatformRuntimeCommand command, + PlatformArtifactItem artifact, + string leaseToken, + CancellationToken cancellationToken = default) + { + if (string.IsNullOrWhiteSpace(artifact.DownloadUrl)) + throw new InvalidOperationException("Platform runtime command artifact does not include a download URL."); + + using var request = CreateRequest(HttpMethod.Get, BuildUri(artifact.DownloadUrl)); + request.Headers.Add(LeaseHeaderName, leaseToken); + request.Headers.Add(WorkerHeaderName, _options.WorkerId); + using var response = await httpClient.SendAsync(request, HttpCompletionOption.ResponseHeadersRead, cancellationToken); + response.EnsureSuccessStatusCode(); + var stream = await response.Content.ReadAsStreamAsync(cancellationToken); + var buffer = new MemoryStream(); + await CopyBoundedAsync(stream, buffer, cancellationToken); + buffer.Position = 0; + return buffer; + } + + public Task ReportProgressAsync( + Guid commandId, + string leaseToken, + string status, + int? percentComplete, + string message, + CancellationToken cancellationToken = default) => + SendMutationAsync( + commandId, + "progress", + new PlatformRuntimeCommandProgressRequest(leaseToken, status, percentComplete, message), + cancellationToken); + + public Task CompleteAsync(Guid commandId, PlatformRuntimeCommandCompleteRequest request, CancellationToken cancellationToken = default) => + SendMutationAsync(commandId, "complete", request, cancellationToken); + + public Task FailAsync(Guid commandId, PlatformRuntimeCommandFailRequest request, CancellationToken cancellationToken = default) => + SendMutationAsync(commandId, "fail", request, cancellationToken); + + public Task RejectAsync(Guid commandId, PlatformRuntimeCommandRejectRequest request, CancellationToken cancellationToken = default) => + SendMutationAsync(commandId, "reject", request, cancellationToken); + + private async Task SendMutationAsync( + Guid commandId, + string action, + TRequest body, + CancellationToken cancellationToken) + { + using var response = await SendJsonAsync(BuildUri($"/deployments/runtime/commands/{commandId:D}/{action}"), body, cancellationToken); + response.EnsureSuccessStatusCode(); + } + + private HttpRequestMessage CreateRequest(HttpMethod method, Uri uri) + { + var request = new HttpRequestMessage(method, uri); + if (!string.IsNullOrWhiteSpace(_options.EngineSecret)) + request.Headers.TryAddWithoutValidation(EngineSecretHeaderName, _options.EngineSecret); + return request; + } + + private async Task SendJsonAsync(Uri uri, TRequest body, CancellationToken cancellationToken) + { + var request = CreateRequest(HttpMethod.Post, uri); + request.Content = JsonContent.Create(body, options: JsonOptions); + return await httpClient.SendAsync(request, cancellationToken); + } + + private Uri BuildUri(string path) + { + var endpoint = _options.PlatformEndpoint ?? throw new InvalidOperationException("Elsa Platform endpoint is required."); + if (Uri.TryCreate(path, UriKind.Absolute, out var absoluteUri)) + return absoluteUri; + + var relative = path.StartsWith("/api/", StringComparison.OrdinalIgnoreCase) + ? path + : $"/api/workspaces/{_options.WorkspaceId:D}{path}"; + return new Uri($"{endpoint.AbsoluteUri.TrimEnd('/')}{relative}"); + } + + private async Task CopyBoundedAsync(Stream source, Stream destination, CancellationToken cancellationToken) + { + var buffer = new byte[81920]; + long total = 0; + while (true) + { + var read = await source.ReadAsync(buffer, cancellationToken); + if (read == 0) + return; + + total += read; + if (total > _options.MaxArtifactBytes) + throw new InvalidOperationException("Elsa Platform artifact exceeds the configured runtime size limit."); + + await destination.WriteAsync(buffer.AsMemory(0, read), cancellationToken); + } + } + + private static JsonSerializerOptions CreateJsonOptions() + { + var options = new JsonSerializerOptions(JsonSerializerDefaults.Web); + options.Converters.Add(new JsonStringEnumConverter(allowIntegerValues: false)); + return options; + } +} diff --git a/src/modules/Elsa.Platform.Integration/Services/PlatformShellReloadTracker.cs b/src/modules/Elsa.Platform.Integration/Services/PlatformShellReloadTracker.cs new file mode 100644 index 000000000..dbb40830c --- /dev/null +++ b/src/modules/Elsa.Platform.Integration/Services/PlatformShellReloadTracker.cs @@ -0,0 +1,14 @@ +namespace Elsa.Platform.Integration.Services; + +public sealed class PlatformShellReloadTracker +{ + private readonly HashSet _shellIds = new(StringComparer.OrdinalIgnoreCase); + + public IReadOnlyCollection ShellIds => _shellIds; + + public void MarkForReload(string shellId) + { + if (!string.IsNullOrWhiteSpace(shellId)) + _shellIds.Add(shellId); + } +} diff --git a/src/modules/Elsa.Platform.Integration/ShellFeatures/ElsaPlatformIntegrationFeature.cs b/src/modules/Elsa.Platform.Integration/ShellFeatures/ElsaPlatformIntegrationFeature.cs new file mode 100644 index 000000000..2cf419cf6 --- /dev/null +++ b/src/modules/Elsa.Platform.Integration/ShellFeatures/ElsaPlatformIntegrationFeature.cs @@ -0,0 +1,29 @@ +using CShells.Features; +using Elsa.Platform.Integration.Options; +using Elsa.Platform.Integration.Services; +using Elsa.Platform.PackageManifest.Generator.Hints; +using Elsa.Workflows.Management.ShellFeatures; +using JetBrains.Annotations; +using Microsoft.Extensions.DependencyInjection; + +namespace Elsa.Platform.Integration.ShellFeatures; + +[ManifestFeatureCategory("Integrations")] +[ShellFeature( + DisplayName = "Elsa Platform Integration", + Description = "Polls Elsa Platform deployment commands and applies Loom recipe artifacts to this Elsa runtime", + DependsOn = [typeof(WorkflowManagementFeature)])] +[UsedImplicitly] +public class ElsaPlatformIntegrationFeature : IShellFeature +{ + public void ConfigureServices(IServiceCollection services) + { + services.AddOptions() + .BindConfiguration(ElsaPlatformIntegrationOptions.ConfigurationSection); + + services.AddHttpClient(); + services.AddSingleton(); + services.AddScoped(); + services.AddHostedService(); + } +} diff --git a/src/modules/Elsa.Platform.Integration/Steps/ConfigureFeaturesStep.cs b/src/modules/Elsa.Platform.Integration/Steps/ConfigureFeaturesStep.cs new file mode 100644 index 000000000..865771fb2 --- /dev/null +++ b/src/modules/Elsa.Platform.Integration/Steps/ConfigureFeaturesStep.cs @@ -0,0 +1,48 @@ +using System.Text.Json; +using Elsa.Platform.Integration.Services; +using Loom; + +namespace Elsa.Platform.Integration.Steps; + +[Step("elsa.configure-features")] +public sealed class ConfigureFeaturesStep( + IShellConfigurationOverlayStore overlayStore, + PlatformShellReloadTracker reloadTracker) : IStep, IValidatingStep +{ + public string ShellId { get; init; } = "Default"; + + public Dictionary? Enable { get; init; } + + public List? Disable { get; init; } + + public ValueTask> ValidateAsync( + StepValidationContext context, + CancellationToken cancellationToken = default) + { + List diagnostics = []; + if (string.IsNullOrWhiteSpace(ShellId)) + diagnostics.Add(context.Error("ELSA_PLATFORM_SHELL_ID_REQUIRED", "Shell ID is required.", context.Target("input.shellId"))); + + if ((Enable is null || Enable.Count == 0) && (Disable is null || Disable.Count == 0)) + diagnostics.Add(context.Error("ELSA_PLATFORM_FEATURE_CHANGE_REQUIRED", "At least one feature enable or disable entry is required.", context.Target("input"))); + + foreach (var feature in Enable ?? []) + { + if (feature.Value.ValueKind is not JsonValueKind.Object and not JsonValueKind.Null and not JsonValueKind.Undefined) + diagnostics.Add(context.Error("ELSA_PLATFORM_FEATURE_CONFIG_INVALID", $"Feature '{feature.Key}' configuration must be a JSON object.", context.Target($"input.enable.{feature.Key}"))); + } + + var duplicate = Enable?.Keys.FirstOrDefault(feature => Disable?.Contains(feature, StringComparer.OrdinalIgnoreCase) == true); + if (duplicate is not null) + diagnostics.Add(context.Error("ELSA_PLATFORM_FEATURE_CHANGE_CONFLICT", $"Feature '{duplicate}' cannot be enabled and disabled in the same step.", context.Target("input"))); + + return ValueTask.FromResult>(diagnostics); + } + + public async ValueTask ExecuteAsync(StepContext context, CancellationToken cancellationToken = default) + { + var changed = await overlayStore.ConfigureFeaturesAsync(ShellId, Enable, Disable, cancellationToken); + if (changed) + reloadTracker.MarkForReload(ShellId); + } +} diff --git a/src/modules/Elsa.Platform.Integration/Steps/ConfigureSettingsStep.cs b/src/modules/Elsa.Platform.Integration/Steps/ConfigureSettingsStep.cs new file mode 100644 index 000000000..68ca9581e --- /dev/null +++ b/src/modules/Elsa.Platform.Integration/Steps/ConfigureSettingsStep.cs @@ -0,0 +1,36 @@ +using System.Text.Json; +using Elsa.Platform.Integration.Services; +using Loom; + +namespace Elsa.Platform.Integration.Steps; + +[Step("elsa.configure-settings")] +public sealed class ConfigureSettingsStep( + IShellConfigurationOverlayStore overlayStore, + PlatformShellReloadTracker reloadTracker) : IStep, IValidatingStep +{ + public string ShellId { get; init; } = "Default"; + + public JsonElement Settings { get; init; } + + public ValueTask> ValidateAsync( + StepValidationContext context, + CancellationToken cancellationToken = default) + { + List diagnostics = []; + if (string.IsNullOrWhiteSpace(ShellId)) + diagnostics.Add(context.Error("ELSA_PLATFORM_SHELL_ID_REQUIRED", "Shell ID is required.", context.Target("input.shellId"))); + + if (Settings.ValueKind != JsonValueKind.Object) + diagnostics.Add(context.Error("ELSA_PLATFORM_SETTINGS_REQUIRED", "Settings must be a JSON object.", context.Target("input.settings"))); + + return ValueTask.FromResult>(diagnostics); + } + + public async ValueTask ExecuteAsync(StepContext context, CancellationToken cancellationToken = default) + { + var changed = await overlayStore.ConfigureSettingsAsync(ShellId, Settings, cancellationToken); + if (changed) + reloadTracker.MarkForReload(ShellId); + } +} diff --git a/src/modules/Elsa.Platform.Integration/Steps/ImportWorkflowDefinitionStep.cs b/src/modules/Elsa.Platform.Integration/Steps/ImportWorkflowDefinitionStep.cs new file mode 100644 index 000000000..8d03a796b --- /dev/null +++ b/src/modules/Elsa.Platform.Integration/Steps/ImportWorkflowDefinitionStep.cs @@ -0,0 +1,97 @@ +using System.Text.Json; +using Elsa.Platform.Integration.Models; +using Elsa.Platform.Integration.Services; +using Elsa.Workflows; +using Elsa.Workflows.Management; +using Elsa.Workflows.Management.Models; +using Loom; + +namespace Elsa.Platform.Integration.Steps; + +[Step("elsa.import-workflow-definition")] +public sealed class ImportWorkflowDefinitionStep( + IWorkflowDefinitionImporter importer, + IApiSerializer apiSerializer, + PlatformRecipeArtifact artifact) : IStep, IValidatingStep +{ + public JsonElement WorkflowDefinition { get; init; } + + public string? Path { get; init; } + + public bool Publish { get; init; } = true; + + public ValueTask> ValidateAsync( + StepValidationContext context, + CancellationToken cancellationToken = default) + { + var hasInlineWorkflow = WorkflowDefinition.ValueKind == JsonValueKind.Object; + var hasPath = !string.IsNullOrWhiteSpace(Path); + if (!hasInlineWorkflow && !hasPath) + { + return ValueTask.FromResult>( + [ + context.Error( + "ELSA_PLATFORM_WORKFLOW_DEFINITION_MISSING", + "A workflow definition or artifact-relative path is required.", + context.Target("input.workflowDefinition")) + ]); + } + + if (hasPath && !artifact.TryGetText(Path!, out _)) + { + return ValueTask.FromResult>( + [ + context.Error( + "ELSA_PLATFORM_WORKFLOW_DEFINITION_FILE_MISSING", + $"Workflow definition file '{Path}' was not found in the recipe artifact.", + context.Target("input.path")) + ]); + } + + return ValueTask.FromResult>([]); + } + + public async ValueTask ExecuteAsync(StepContext context, CancellationToken cancellationToken = default) + { + var workflowJson = ResolveWorkflowJson(); + WorkflowDefinitionModel model; + try + { + model = apiSerializer.Deserialize(workflowJson); + } + catch (Exception ex) when (ex is JsonException or NotSupportedException or InvalidOperationException) + { + throw new InvalidOperationException("Workflow definition payload is invalid.", ex); + } + + var importResult = await importer.ImportAsync(new SaveWorkflowDefinitionRequest + { + Model = model, + Publish = Publish + }, cancellationToken); + + if (!importResult.Succeeded) + { + var errors = string.Join("; ", importResult.ValidationErrors.Select(x => x.Message)); + throw new InvalidOperationException($"Workflow definition validation failed: {errors}"); + } + + context.Log($"Workflow definition '{model.DefinitionId}' was imported."); + } + + private string ResolveWorkflowJson() + { + if (!string.IsNullOrWhiteSpace(Path)) + { + if (artifact.TryGetText(Path, out var content)) + return content; + + throw new InvalidOperationException($"Workflow definition file '{Path}' was not found in the recipe artifact."); + } + + if (WorkflowDefinition.ValueKind == JsonValueKind.Object) + return WorkflowDefinition.GetRawText(); + + throw new InvalidOperationException("A workflow definition or artifact-relative path is required."); + } +} diff --git a/src/modules/Elsa.Platform.Integration/Steps/VerifyCapabilitiesStep.cs b/src/modules/Elsa.Platform.Integration/Steps/VerifyCapabilitiesStep.cs new file mode 100644 index 000000000..fa653f70e --- /dev/null +++ b/src/modules/Elsa.Platform.Integration/Steps/VerifyCapabilitiesStep.cs @@ -0,0 +1,44 @@ +using Elsa.Features.Contracts; +using Elsa.Platform.Integration.Options; +using Loom; +using Microsoft.Extensions.Options; + +namespace Elsa.Platform.Integration.Steps; + +[Step("elsa.verify-capabilities")] +public sealed class VerifyCapabilitiesStep( + IInstalledFeatureProvider installedFeatureProvider, + IOptions options) : IStep, IValidatingStep +{ + public List? Features { get; init; } + + public List? Capabilities { get; init; } + + public ValueTask> ValidateAsync( + StepValidationContext context, + CancellationToken cancellationToken = default) + { + List diagnostics = []; + var installedFeatures = installedFeatureProvider.List().ToList(); + foreach (var feature in Features ?? []) + { + var exists = installedFeatures.Any(x => + string.Equals(x.Name, feature, StringComparison.OrdinalIgnoreCase) + || string.Equals(x.FullName, feature, StringComparison.OrdinalIgnoreCase)); + if (!exists) + diagnostics.Add(context.Error("ELSA_PLATFORM_FEATURE_MISSING", $"Required feature '{feature}' is not installed.", context.Target("input.features"))); + } + + var runtimeCapabilities = options.Value.Capabilities; + foreach (var capability in Capabilities ?? []) + { + if (!runtimeCapabilities.Contains(capability, StringComparer.OrdinalIgnoreCase)) + diagnostics.Add(context.Error("ELSA_PLATFORM_CAPABILITY_MISSING", $"Required runtime capability '{capability}' is not available.", context.Target("input.capabilities"))); + } + + return ValueTask.FromResult>(diagnostics); + } + + public ValueTask ExecuteAsync(StepContext context, CancellationToken cancellationToken = default) => + ValueTask.CompletedTask; +} diff --git a/test/unit/Elsa.Platform.Integration.UnitTests/Elsa.Platform.Integration.UnitTests.csproj b/test/unit/Elsa.Platform.Integration.UnitTests/Elsa.Platform.Integration.UnitTests.csproj new file mode 100644 index 000000000..09278896b --- /dev/null +++ b/test/unit/Elsa.Platform.Integration.UnitTests/Elsa.Platform.Integration.UnitTests.csproj @@ -0,0 +1,7 @@ + + + + + + + diff --git a/test/unit/Elsa.Platform.Integration.UnitTests/ElsaLoomRecipeArtifactApplierTests.cs b/test/unit/Elsa.Platform.Integration.UnitTests/ElsaLoomRecipeArtifactApplierTests.cs new file mode 100644 index 000000000..bbce8aab7 --- /dev/null +++ b/test/unit/Elsa.Platform.Integration.UnitTests/ElsaLoomRecipeArtifactApplierTests.cs @@ -0,0 +1,172 @@ +using System.IO.Compression; +using System.Security.Cryptography; +using System.Text; +using System.Text.Json; +using CShells.Lifecycle; +using Elsa.Platform.Integration.Models; +using Elsa.Platform.Integration.Options; +using Elsa.Platform.Integration.Services; +using Elsa.Workflows; +using Elsa.Workflows.Management; +using Elsa.Workflows.Management.Entities; +using Elsa.Workflows.Management.Models; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Options; +using NSubstitute; + +namespace Elsa.Platform.Integration.UnitTests; + +public class ElsaLoomRecipeArtifactApplierTests +{ + [Fact] + public async Task ApplyAsync_WithInlineImportWorkflowDefinitionRecipe_ImportsWorkflowDefinition() + { + var importer = Substitute.For(); + var apiSerializer = Substitute.For(); + var shellRegistry = Substitute.For(); + SaveWorkflowDefinitionRequest? importedRequest = null; + var model = new WorkflowDefinitionModel + { + DefinitionId = "workflow-1", + Name = "Workflow 1" + }; + apiSerializer.Deserialize(Arg.Any()).Returns(model); + importer.ImportAsync(Arg.Do(x => importedRequest = x), Arg.Any()) + .Returns(new ImportWorkflowResult(true, new WorkflowDefinition { DefinitionId = "workflow-1" }, [])); + + var services = new ServiceCollection() + .AddSingleton(importer) + .AddSingleton(apiSerializer) + .AddSingleton(Microsoft.Extensions.Options.Options.Create(new ElsaPlatformIntegrationOptions())) + .BuildServiceProvider(); + var sut = new ElsaLoomRecipeArtifactApplier(services, shellRegistry); + await using var artifactZip = CreateArtifact(""" + { + "name": "deploy-workflow", + "steps": [ + { + "id": "import-workflow", + "type": "elsa.import-workflow-definition", + "input": { + "workflowDefinition": { + "definitionId": "workflow-1", + "name": "Workflow 1" + } + } + } + ] + } + """); + var artifact = CreateArtifactItem(artifactZip); + + var result = await sut.ApplyAsync(CreateCommand(artifact), artifact, artifactZip); + + Assert.True(result.Succeeded); + Assert.NotNull(importedRequest); + Assert.True(importedRequest.Publish); + Assert.Same(model, importedRequest.Model); + await shellRegistry.DidNotReceiveWithAnyArgs().ReloadAsync(default!, default); + } + + [Fact] + public async Task ApplyAsync_WithMissingCapability_ReturnsRejected() + { + var services = new ServiceCollection() + .AddSingleton(Substitute.For()) + .AddSingleton(Substitute.For()) + .AddSingleton(Microsoft.Extensions.Options.Options.Create(new ElsaPlatformIntegrationOptions())) + .AddSingleton(Substitute.For()) + .BuildServiceProvider(); + var sut = new ElsaLoomRecipeArtifactApplier(services, Substitute.For()); + await using var artifactZip = CreateArtifact(""" + { + "name": "verify-capability", + "steps": [ + { + "id": "verify", + "type": "elsa.verify-capabilities", + "input": { + "capabilities": [ "missing.capability" ] + } + } + ] + } + """); + var artifact = CreateArtifactItem(artifactZip); + + var result = await sut.ApplyAsync(CreateCommand(artifact), artifact, artifactZip); + + Assert.Equal(PlatformArtifactStatus.Rejected, result.Status); + Assert.Contains(result.Diagnostics, x => x.Code == "ELSA_PLATFORM_CAPABILITY_MISSING"); + } + + private static MemoryStream CreateArtifact(string recipeJson) + { + var stream = new MemoryStream(); + using (var archive = new ZipArchive(stream, ZipArchiveMode.Create, leaveOpen: true)) + { + var entry = archive.CreateEntry("payload/recipes/recipe.json"); + using var entryStream = entry.Open(); + using var writer = new StreamWriter(entryStream, Encoding.UTF8); + writer.Write(recipeJson); + } + + stream.Position = 0; + return stream; + } + + private static PlatformArtifactItem CreateArtifactItem(Stream artifactZip) + { + var digest = ComputeDigest(artifactZip); + return new PlatformArtifactItem( + Guid.NewGuid(), + "recipe", + "loom.recipe", + "1.0", + digest, + "Recipe", + "https://example.com/recipe.zip", + PlatformArtifactStatus.Pending, + null, + null, + null); + } + + private static PlatformRuntimeCommand CreateCommand(PlatformArtifactItem artifact) => + new( + Guid.NewGuid(), + Guid.NewGuid(), + Guid.NewGuid(), + Guid.NewGuid(), + Guid.NewGuid(), + PlatformRuntimeCommandAction.Deploy, + PlatformRuntimeCommandStatus.Running, + null, + null, + "test", + null, + null, + null, + null, + 1, + null, + null, + null, + null, + [], + DateTimeOffset.UtcNow, + DateTimeOffset.UtcNow, + null, + null, + null, + [artifact]); + + private static PlatformArtifactDigest ComputeDigest(Stream stream) + { + stream.Position = 0; + using var sha = SHA256.Create(); + var hash = sha.ComputeHash(stream); + stream.Position = 0; + return new PlatformArtifactDigest("sha256", Convert.ToHexString(hash).ToLowerInvariant()); + } +} diff --git a/test/unit/Elsa.Platform.Integration.UnitTests/FileShellConfigurationOverlayStoreTests.cs b/test/unit/Elsa.Platform.Integration.UnitTests/FileShellConfigurationOverlayStoreTests.cs new file mode 100644 index 000000000..959a3448f --- /dev/null +++ b/test/unit/Elsa.Platform.Integration.UnitTests/FileShellConfigurationOverlayStoreTests.cs @@ -0,0 +1,76 @@ +using System.Text.Json.Nodes; +using Elsa.Platform.Integration.Options; +using Elsa.Platform.Integration.Services; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Options; +using NSubstitute; + +namespace Elsa.Platform.Integration.UnitTests; + +public class FileShellConfigurationOverlayStoreTests : IAsyncLifetime +{ + private readonly string _directory = Path.Combine(Path.GetTempPath(), $"elsa-platform-overlay-{Guid.NewGuid():N}"); + private readonly FileShellConfigurationOverlayStore _store; + private readonly string _overlayPath; + + public FileShellConfigurationOverlayStoreTests() + { + _overlayPath = Path.Combine(_directory, "platform-shell-overrides.json"); + var environment = Substitute.For(); + environment.ContentRootPath.Returns(_directory); + _store = new FileShellConfigurationOverlayStore( + Microsoft.Extensions.Options.Options.Create(new ElsaPlatformIntegrationOptions + { + ShellOverlayPath = "platform-shell-overrides.json" + }), + environment); + } + + [Fact] + public async Task ConfigureFeaturesAsync_WritesEnabledAndDisabledFeatureState() + { + var enabled = new Dictionary + { + ["Http"] = System.Text.Json.JsonDocument.Parse("""{ "baseUrl": "https://example.com" }""").RootElement.Clone() + }; + + var changed = await _store.ConfigureFeaturesAsync("Default", enabled, ["Python"]); + + Assert.True(changed); + var root = await ReadRootAsync(); + Assert.Equal("https://example.com", root["CShells"]!["Shells"]!["Default"]!["Features"]!["Http"]!["baseUrl"]!.GetValue()); + Assert.Equal("Python", root["Elsa"]!["PlatformIntegration"]!["Shells"]!["Default"]!["DisabledFeatures"]![0]!.GetValue()); + } + + [Fact] + public async Task ConfigureSettingsAsync_MergesShellConfiguration() + { + var settings = System.Text.Json.JsonDocument.Parse("""{ "WebRouting": { "Path": "tenant-a" } }""").RootElement.Clone(); + + var changed = await _store.ConfigureSettingsAsync("Default", settings); + + Assert.True(changed); + var root = await ReadRootAsync(); + Assert.Equal("tenant-a", root["CShells"]!["Shells"]!["Default"]!["Configuration"]!["WebRouting"]!["Path"]!.GetValue()); + } + + public Task InitializeAsync() + { + Directory.CreateDirectory(_directory); + return Task.CompletedTask; + } + + public Task DisposeAsync() + { + if (Directory.Exists(_directory)) + Directory.Delete(_directory, recursive: true); + + return Task.CompletedTask; + } + + private async Task ReadRootAsync() + { + await using var stream = File.OpenRead(_overlayPath); + return (await JsonNode.ParseAsync(stream))!; + } +}