feat(nodes): WebhookNode — registers (workflow,path) route and emits body items

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;
 

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);
 

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<IAsyncDisposable> 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<IAsyncDisposable>(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; }
+    }
+}

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;
 

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<object> GetAsync(string id, CancellationToken ct) => throw new InvalidOperationException();
+    }
+    private sealed class FakeSink : ITriggerSink
+    {
+        public IReadOnlyList<Item>? LastItems { get; private set; }
+        public Guid? LastAnswer { get; private set; }
+        public Task EmitAsync(IReadOnlyList<Item> 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"));
+    }
+}