diff --git a/src/VolcanicCore.Api/Endpoints/WebhookEndpoint.cs b/src/VolcanicCore.Api/Endpoints/WebhookEndpoint.cs index fe75c76..a32e40d 100644 --- a/src/VolcanicCore.Api/Endpoints/WebhookEndpoint.cs +++ b/src/VolcanicCore.Api/Endpoints/WebhookEndpoint.cs @@ -1,6 +1,7 @@ using System.Text.Json; using Microsoft.AspNetCore.Http; using VolcanicCore.Api.Engine; +using VolcanicCore.Core; namespace VolcanicCore.Api.Endpoints; diff --git a/src/VolcanicCore.Api/Engine/WebhookRegistry.cs b/src/VolcanicCore.Core/WebhookRegistry.cs similarity index 97% rename from src/VolcanicCore.Api/Engine/WebhookRegistry.cs rename to src/VolcanicCore.Core/WebhookRegistry.cs index 00564f8..817dcc1 100644 --- a/src/VolcanicCore.Api/Engine/WebhookRegistry.cs +++ b/src/VolcanicCore.Core/WebhookRegistry.cs @@ -1,7 +1,7 @@ using System.Collections.Concurrent; using System.Text.Json; -namespace VolcanicCore.Api.Engine; +namespace VolcanicCore.Core; public delegate Task WebhookInvocation(JsonElement body, Guid answerChannelId, CancellationToken ct); diff --git a/src/VolcanicCore.Nodes/WebhookNode.cs b/src/VolcanicCore.Nodes/WebhookNode.cs new file mode 100644 index 0000000..4040ecb --- /dev/null +++ b/src/VolcanicCore.Nodes/WebhookNode.cs @@ -0,0 +1,43 @@ +using System.Text.Json; +using VolcanicCore.Core; + +namespace VolcanicCore.Nodes; + +public sealed class WebhookNode : ITriggerNode +{ + private readonly WebhookRegistry _routes; + public WebhookNode(WebhookRegistry routes) => _routes = routes; + + public string Id => "http.webhook"; + public string DisplayName => "Webhook"; + public NodeCategory Category => NodeCategory.Trigger; + public int InputCount => 0; + public int OutputCount => 1; + public string? ConnectionProviderId => null; + public JsonSchema ConfigSchema { get; } = new(JsonDocument.Parse(""" + { + "type": "object", + "required": ["path"], + "properties": { + "path": { "type": "string", "title": "URL path (after /wh/{workflow-id}/)" } + } + } + """).RootElement); + + public Task StartAsync(INodeContext ctx, ITriggerSink sink, CancellationToken ct) + { + var path = ctx.Config.GetProperty("path").GetString()!; + var reg = _routes.Register(ctx.WorkflowId, path, async (body, channel, cbCt) => + { + await sink.EmitAsync(new[] { new Item(body) }, answerChannelId: channel, ct: cbCt); + }); + return Task.FromResult(new Handle(reg)); + } + + private sealed class Handle : IAsyncDisposable + { + private readonly IDisposable _reg; + public Handle(IDisposable reg) => _reg = reg; + public ValueTask DisposeAsync() { _reg.Dispose(); return default; } + } +} diff --git a/tests/VolcanicCore.Tests/WebhookEndpointTests.cs b/tests/VolcanicCore.Tests/WebhookEndpointTests.cs index 2087720..f08d0e5 100644 --- a/tests/VolcanicCore.Tests/WebhookEndpointTests.cs +++ b/tests/VolcanicCore.Tests/WebhookEndpointTests.cs @@ -3,6 +3,7 @@ using System.Net.Http.Json; using System.Text.Json; using Microsoft.Extensions.DependencyInjection; using VolcanicCore.Api.Engine; +using VolcanicCore.Core; namespace VolcanicCore.Tests; diff --git a/tests/VolcanicCore.Tests/WebhookNodeTests.cs b/tests/VolcanicCore.Tests/WebhookNodeTests.cs new file mode 100644 index 0000000..d2ba4c6 --- /dev/null +++ b/tests/VolcanicCore.Tests/WebhookNodeTests.cs @@ -0,0 +1,49 @@ +using System.Text.Json; +using Microsoft.Extensions.Logging.Abstractions; +using VolcanicCore.Api.Engine; +using VolcanicCore.Core; +using VolcanicCore.Nodes; + +// ImplicitUsings pulls System.Threading → collision with our ExecutionContext. +using ExecutionContext = VolcanicCore.Api.Engine.ExecutionContext; + +namespace VolcanicCore.Tests; + +public class WebhookNodeTests +{ + private sealed class NullReg : IConnectionInstanceRegistry + { + public Task GetAsync(string id, CancellationToken ct) => throw new InvalidOperationException(); + } + private sealed class FakeSink : ITriggerSink + { + public IReadOnlyList? LastItems { get; private set; } + public Guid? LastAnswer { get; private set; } + public Task EmitAsync(IReadOnlyList items, JsonElement? td, Guid? ans, CancellationToken ct) + { LastItems = items; LastAnswer = ans; return Task.CompletedTask; } + } + + [Fact] + public async Task StartAsync_registers_path_and_dispose_removes_it() + { + var registry = new WebhookRegistry(); + var node = new WebhookNode(registry); + var sink = new FakeSink(); + var ctx = new ExecutionContext("", "wf-a", "n1", + JsonDocument.Parse("""{"path":"submit"}""").RootElement, + null, NullLogger.Instance, null, new NullReg()); + + var handle = await node.StartAsync(ctx, sink, CancellationToken.None); + var invoke = registry.Lookup("wf-a", "submit"); + Assert.NotNull(invoke); + + var body = JsonDocument.Parse("""{"greeting":"hi"}""").RootElement; + var channel = Guid.NewGuid(); + await invoke!(body, channel, CancellationToken.None); + Assert.Equal(channel, sink.LastAnswer); + Assert.Equal("hi", sink.LastItems![0].Json.GetProperty("greeting").GetString()); + + await handle.DisposeAsync(); + Assert.Null(registry.Lookup("wf-a", "submit")); + } +}