feat(nodes): NatsHookNode — trigger emits one execution per message

src/VolcanicCore.Nodes/NatsHookNode.cs

@@ -0,0 +1,86 @@
+using System.Text;
+using System.Text.Json;
+using NATS.Client.Core;
+using VolcanicCore.Core;
+
+namespace VolcanicCore.Nodes;
+
+public sealed class NatsHookNode : ITriggerNode
+{
+    // Testable seam over the tiny slice of INatsConnection we actually use.
+    public interface INatsSubscriber
+    {
+        IAsyncEnumerable<byte[]> SubscribeAsync(string subject, CancellationToken ct);
+    }
+
+    private sealed class NatsSubscriberAdapter : INatsSubscriber
+    {
+        private readonly INatsConnection _conn;
+        public NatsSubscriberAdapter(INatsConnection c) => _conn = c;
+        public async IAsyncEnumerable<byte[]> SubscribeAsync(string subject,
+            [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken ct)
+        {
+            await foreach (var msg in _conn.SubscribeAsync<byte[]>(subject, cancellationToken: ct))
+                if (msg.Data is not null) yield return msg.Data;
+        }
+    }
+
+    public string Id => "nats.hook";
+    public string DisplayName => "NATS Hook";
+    public NodeCategory Category => NodeCategory.Trigger;
+    public int InputCount => 0;
+    public int OutputCount => 1;
+    public string? ConnectionProviderId => "nats";
+    public JsonSchema ConfigSchema { get; } = new(JsonDocument.Parse("""
+        {
+          "type": "object",
+          "required": ["subject"],
+          "properties": { "subject": { "type": "string", "title": "Subject" } }
+        }
+        """).RootElement);
+
+    public Task<IAsyncDisposable> StartAsync(INodeContext ctx, ITriggerSink sink, CancellationToken ct)
+    {
+        var subject = ctx.Config.GetProperty("subject").GetString()!;
+        var raw = ctx.GetConnection<object>();
+        var sub = raw as INatsSubscriber ?? new NatsSubscriberAdapter((INatsConnection)raw);
+
+        var pumpCts = CancellationTokenSource.CreateLinkedTokenSource(ct);
+        var pump = Task.Run(async () =>
+        {
+            try
+            {
+                await foreach (var payload in sub.SubscribeAsync(subject, pumpCts.Token))
+                {
+                    JsonElement json;
+                    try
+                    {
+                        using var doc = JsonDocument.Parse(payload);
+                        json = doc.RootElement.Clone();
+                    }
+                    catch (JsonException)
+                    {
+                        json = JsonSerializer.SerializeToElement(new { raw = Encoding.UTF8.GetString(payload) });
+                    }
+                    await sink.EmitAsync(new[] { new Item(json) }, ct: pumpCts.Token);
+                }
+            }
+            catch (OperationCanceledException) { /* stop */ }
+        }, pumpCts.Token);
+
+        return Task.FromResult<IAsyncDisposable>(new Handle(pumpCts, pump));
+    }
+
+    private sealed class Handle : IAsyncDisposable
+    {
+        private readonly CancellationTokenSource _cts;
+        private readonly Task _pump;
+        public Handle(CancellationTokenSource c, Task p) { _cts = c; _pump = p; }
+        public async ValueTask DisposeAsync()
+        {
+            _cts.Cancel();
+            try { await _pump; } catch { }
+            _cts.Dispose();
+        }
+    }
+}

tests/VolcanicCore.Tests/NatsHookNodeTests.cs

@@ -0,0 +1,69 @@
+using System.Text;
+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 NatsHookNodeTests
+{
+    private sealed class FakeSubscriber : NatsHookNode.INatsSubscriber
+    {
+        private readonly List<byte[]> _payloads;
+        public FakeSubscriber(params byte[][] payloads) => _payloads = payloads.ToList();
+        public async IAsyncEnumerable<byte[]> SubscribeAsync(string subject,
+            [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken ct)
+        {
+            foreach (var p in _payloads)
+            {
+                yield return p;
+                await Task.Yield();
+            }
+        }
+    }
+
+    private sealed class FakeInstanceRegistry : IConnectionInstanceRegistry
+    {
+        private readonly object _handle;
+        public FakeInstanceRegistry(object h) => _handle = h;
+        public Task<object> GetAsync(string id, CancellationToken ct) => Task.FromResult(_handle);
+    }
+
+    private sealed class FakeSink : ITriggerSink
+    {
+        public List<IReadOnlyList<Item>> Batches { get; } = new();
+        public Task EmitAsync(IReadOnlyList<Item> items, JsonElement? td, Guid? ans, CancellationToken ct)
+        { lock (Batches) Batches.Add(items); return Task.CompletedTask; }
+    }
+
+    [Fact]
+    public async Task Subscribes_and_emits_one_execution_per_message()
+    {
+        var node = new NatsHookNode();
+        var sub = new FakeSubscriber(
+            Encoding.UTF8.GetBytes("{\"n\":1}"),
+            Encoding.UTF8.GetBytes("{\"n\":2}"));
+        var reg = new FakeInstanceRegistry(sub);
+        var sink = new FakeSink();
+
+        var ctx = new ExecutionContext("", "wf", "n1",
+            JsonDocument.Parse("""{"subject":"topic"}""").RootElement,
+            null, NullLogger.Instance, "conn-1", reg);
+
+        var handle = await node.StartAsync(ctx, sink, CancellationToken.None);
+
+        // Give the pump a moment to drain the fake enumerable.
+        for (var i = 0; i < 50 && sink.Batches.Count < 2; i++) await Task.Delay(10);
+
+        Assert.Equal(2, sink.Batches.Count);
+        Assert.Equal(1, sink.Batches[0][0].Json.GetProperty("n").GetInt32());
+        Assert.Equal(2, sink.Batches[1][0].Json.GetProperty("n").GetInt32());
+
+        await handle.DisposeAsync();
+    }
+}