diff --git a/src/VolcanicCore.Api/Engine/ExecutionContext.cs b/src/VolcanicCore.Api/Engine/ExecutionContext.cs new file mode 100644 index 0000000..919a122 --- /dev/null +++ b/src/VolcanicCore.Api/Engine/ExecutionContext.cs @@ -0,0 +1,62 @@ +using System.Text.Json; +using Microsoft.Extensions.Logging; +using VolcanicCore.Core; + +namespace VolcanicCore.Api.Engine; + +// Interface pulled out so tests can substitute without spinning up real providers. +// Concrete impl lives in ConnectionInstanceRegistry (next task). +public interface IConnectionInstanceRegistry +{ + Task GetAsync(string connectionId, CancellationToken ct); +} + +public sealed class ExecutionContext : INodeContext +{ + private readonly IConnectionInstanceRegistry _instances; + private readonly string? _connectionId; + + public ExecutionContext( + string executionId, + string workflowId, + string nodeId, + JsonElement config, + Guid? answerChannelId, + ILogger logger, + string? connectionId, + IConnectionInstanceRegistry instances) + { + ExecutionId = executionId; + WorkflowId = workflowId; + NodeId = nodeId; + Config = config; + AnswerChannelId = answerChannelId; + Logger = logger; + _connectionId = connectionId; + _instances = instances; + } + + public string ExecutionId { get; } + public string WorkflowId { get; } + public string NodeId { get; } + public JsonElement Config { get; } + public Guid? AnswerChannelId { get; } + public ILogger Logger { get; } + + public T GetConnection() where T : class + { + if (_connectionId is null) + throw new InvalidOperationException( + $"Node {NodeId} has no configured connection but requested one of type {typeof(T).Name}."); + + // Async->sync inside a node call site is acceptable here — node methods + // are already Task-returning and the caller is fine to block briefly for a + // cached lookup. Real providers should complete synchronously for a cached hit. + var handle = _instances.GetAsync(_connectionId, CancellationToken.None) + .GetAwaiter().GetResult(); + if (handle is not T typed) + throw new InvalidCastException( + $"Connection {_connectionId} is not assignable to {typeof(T).Name}."); + return typed; + } +} diff --git a/src/VolcanicCore.Api/Engine/TriggerActivationSink.cs b/src/VolcanicCore.Api/Engine/TriggerActivationSink.cs new file mode 100644 index 0000000..38dc7c1 --- /dev/null +++ b/src/VolcanicCore.Api/Engine/TriggerActivationSink.cs @@ -0,0 +1,49 @@ +using System.Text.Json; +using VolcanicCore.Core; + +namespace VolcanicCore.Api.Engine; + +// Callback shape the TriggerHost hands to a trigger's StartAsync. Emitting +// via the sink kicks off one execution. +public delegate Task StartExecutionAsync( + string workflowId, + string triggerNodeId, + string triggerKind, // "nats.hook" | "http.webhook" | ... + IReadOnlyList items, + JsonElement? triggerData, + Guid? answerChannelId, + CancellationToken ct); + +public sealed class TriggerActivationSink : ITriggerSink +{ + private readonly string _workflowId; + private readonly string _triggerNodeId; + private readonly string _triggerKind; + private readonly StartExecutionAsync _start; + + public TriggerActivationSink( + string workflowId, string triggerNodeId, string triggerKind, StartExecutionAsync start) + { + _workflowId = workflowId; + _triggerNodeId = triggerNodeId; + _triggerKind = triggerKind; + _start = start; + } + + public Task EmitAsync( + IReadOnlyList items, + JsonElement? triggerData = null, + Guid? answerChannelId = null, + CancellationToken ct = default) + { + // Fire-and-forget: a NATS trigger's pump must not block per-event on + // execution completion. A Webhook trigger's /wh endpoint doesn't rely + // on this task either — it awaits WebhookRouter.Register for the reply. + _ = Task.Run(async () => + { + try { await _start(_workflowId, _triggerNodeId, _triggerKind, items, triggerData, answerChannelId, ct); } + catch { /* Executor already logged + persisted the failure */ } + }); + return Task.CompletedTask; + } +} diff --git a/tests/VolcanicCore.Tests/ExecutionContextTests.cs b/tests/VolcanicCore.Tests/ExecutionContextTests.cs new file mode 100644 index 0000000..c797414 --- /dev/null +++ b/tests/VolcanicCore.Tests/ExecutionContextTests.cs @@ -0,0 +1,59 @@ +using System.Text.Json; +using Microsoft.Extensions.Logging.Abstractions; +using VolcanicCore.Api.Engine; +using ExecutionContext = VolcanicCore.Api.Engine.ExecutionContext; + +namespace VolcanicCore.Tests; + +public class ExecutionContextTests +{ + private sealed class Handle { public string Value { get; set; } = ""; } + + [Fact] + public void GetConnection_returns_registered_handle_by_id() + { + var handle = new Handle { Value = "live" }; + var reg = FakeInstanceRegistry.With("conn-1", handle); + var ctx = new ExecutionContext( + executionId: "e", + workflowId: "w", + nodeId: "n", + config: JsonDocument.Parse("{}").RootElement, + answerChannelId: null, + logger: NullLogger.Instance, + connectionId: "conn-1", + instances: reg); + + Assert.Same(handle, ctx.GetConnection()); + } + + [Fact] + public void GetConnection_throws_when_node_has_no_connection_id() + { + var ctx = new ExecutionContext("e", "w", "n", + JsonDocument.Parse("{}").RootElement, null, + NullLogger.Instance, connectionId: null, instances: FakeInstanceRegistry.Empty); + + Assert.Throws(() => ctx.GetConnection()); + } + + [Fact] + public void GetConnection_throws_when_type_mismatches() + { + var reg = FakeInstanceRegistry.With("conn-1", new object()); + var ctx = new ExecutionContext("e", "w", "n", + JsonDocument.Parse("{}").RootElement, null, + NullLogger.Instance, connectionId: "conn-1", instances: reg); + + Assert.Throws(() => ctx.GetConnection()); + } + + private sealed class FakeInstanceRegistry : IConnectionInstanceRegistry + { + private readonly Dictionary _byId; + private FakeInstanceRegistry(Dictionary byId) => _byId = byId; + public static FakeInstanceRegistry Empty => new(new()); + public static FakeInstanceRegistry With(string id, object h) => new(new() { [id] = h }); + public Task GetAsync(string connectionId, CancellationToken ct) => Task.FromResult(_byId[connectionId]); + } +}