using System.Text.Json;
using Microsoft.Extensions.Logging.Abstractions;
using VolcanicCore.Api.Engine;
using VolcanicCore.Api.Registries;
using VolcanicCore.Core;
using VolcanicCore.Persistence;
using VolcanicCore.Persistence.Encryption;
using VolcanicCore.Persistence.Migrations;
using VolcanicCore.Persistence.Repositories;
using System.Security.Cryptography;

namespace VolcanicCore.Tests;

public class ExecutorTests
{
    // --- test node types ---------------------------------------------------
    private sealed class TestTrigger : ITriggerNode
    {
        public string Id { get; init; } = "test.trigger";
        public string DisplayName => Id;
        public NodeCategory Category => NodeCategory.Trigger;
        public int InputCount => 0;
        public int OutputCount => 1;
        public string? ConnectionProviderId => null;
        public JsonSchema ConfigSchema => new(JsonDocument.Parse("{}").RootElement);
        public Task<IAsyncDisposable> StartAsync(INodeContext ctx, ITriggerSink sink, CancellationToken ct)
            => throw new NotSupportedException("triggered externally in tests");
    }

    private sealed class EchoAction : IActionNode
    {
        public string Id { get; init; } = "test.echo";
        public string DisplayName => Id;
        public NodeCategory Category => NodeCategory.Action;
        public int InputCount => 1;
        public int OutputCount => 1;
        public string? ConnectionProviderId => null;
        public JsonSchema ConfigSchema => new(JsonDocument.Parse("{}").RootElement);
        public Task<IReadOnlyList<IReadOnlyList<Item>>> ExecuteAsync(
            INodeContext ctx, IReadOnlyList<IReadOnlyList<Item>> inputsByPort, CancellationToken ct)
            => Task.FromResult<IReadOnlyList<IReadOnlyList<Item>>>(new[] { inputsByPort[0] });
    }

    private sealed class TerminalSink : IActionNode
    {
        public string Id { get; init; } = "test.sink";
        public string DisplayName => Id;
        public NodeCategory Category => NodeCategory.Terminal;
        public int InputCount => 1;
        public int OutputCount => 0;
        public string? ConnectionProviderId => null;
        public JsonSchema ConfigSchema => new(JsonDocument.Parse("{}").RootElement);
        public Task<IReadOnlyList<IReadOnlyList<Item>>> ExecuteAsync(
            INodeContext ctx, IReadOnlyList<IReadOnlyList<Item>> inputsByPort, CancellationToken ct)
            => Task.FromResult<IReadOnlyList<IReadOnlyList<Item>>>(Array.Empty<IReadOnlyList<Item>>());
    }

    // --- setup helpers -----------------------------------------------------
    private sealed class NullInstanceRegistry : IConnectionInstanceRegistry
    {
        public Task<object> GetAsync(string id, CancellationToken ct)
            => throw new InvalidOperationException("no connections in this test");
    }

    private static (ExecutionRepository ex, NodeExecutionRepository ne, Executor exec, NodeRegistry nr)
        Build(params INodeType[] nodes)
    {
        var path = Path.Combine(Path.GetTempPath(), $"vc-{Guid.NewGuid():N}.db");
        var f = new SqliteConnectionFactory(path);
        using (var cn = f.Open())
            new Migrator(new IMigration[]
            {
                new Migration001Initial(), new Migration002Connections(), new Migration003Workflows(),
                new Migration004Executions(), new Migration005NodeExecutions(),
            }).Apply(cn);
        var er = new ExecutionRepository(f);
        var nr = new NodeExecutionRepository(f);
        var registry = new NodeRegistry(nodes);
        var exec = new Executor(er, nr, registry, new NullInstanceRegistry(), NullLoggerFactory.Instance);
        return (er, nr, exec, registry);
    }

    private static JsonElement Item(int v)
        => JsonSerializer.SerializeToElement(new { v });

    // --- tests -------------------------------------------------------------

    [Fact]
    public async Task Linear_trigger_action_terminal_records_all_three_node_executions_and_completes()
    {
        var (er, nr, exec, _) = Build(new TestTrigger(), new EchoAction(), new TerminalSink());

        var graph = new WorkflowGraph
        {
            Nodes = new[] {
                new WorkflowNode("t", "test.trigger", null, "{}"),
                new WorkflowNode("a", "test.echo",    null, "{}"),
                new WorkflowNode("s", "test.sink",    null, "{}"),
            },
            Edges = new[] {
                new WorkflowEdge("t", 0, "a", 0),
                new WorkflowEdge("a", 0, "s", 0),
            },
        };

        var eid = await exec.StartExecutionAsync(
            workflowId: "wf1",
            graph: graph,
            triggerNodeId: "t",
            triggerKind: "manual",
            triggerItems: new[] { new Core.Item(Item(1)) },
            triggerData: null,
            answerChannelId: null,
            ct: CancellationToken.None);

        var run = er.Get(eid)!;
        Assert.Equal("completed", run.Status);
        var rows = nr.ListByExecution(eid);
        Assert.Equal(3, rows.Count);
        Assert.All(rows, r => Assert.Equal("completed", r.Status));
        // trigger has empty inputs and one output list
        var trig = rows.Single(r => r.NodeId == "t");
        Assert.Equal("[]", trig.InputsJson);
    }

    private sealed class Throwing : IActionNode
    {
        public string Id { get; init; } = "test.throw";
        public string DisplayName => Id;
        public NodeCategory Category => NodeCategory.Action;
        public int InputCount => 1;
        public int OutputCount => 1;
        public string? ConnectionProviderId => null;
        public JsonSchema ConfigSchema => new(JsonDocument.Parse("{}").RootElement);
        public Task<IReadOnlyList<IReadOnlyList<Item>>> ExecuteAsync(
            INodeContext ctx, IReadOnlyList<IReadOnlyList<Item>> inputsByPort, CancellationToken ct)
            => throw new InvalidOperationException("boom");
    }

    private sealed class Recorder : IActionNode
    {
        public List<INodeContext> Calls { get; } = new();
        public string Id { get; init; } = "test.record";
        public string DisplayName => Id;
        public NodeCategory Category => NodeCategory.Terminal;
        public int InputCount => 1;
        public int OutputCount => 0;
        public string? ConnectionProviderId => null;
        public JsonSchema ConfigSchema => new(JsonDocument.Parse("{}").RootElement);
        public Task<IReadOnlyList<IReadOnlyList<Item>>> ExecuteAsync(
            INodeContext ctx, IReadOnlyList<IReadOnlyList<Item>> inputsByPort, CancellationToken ct)
        {
            lock (Calls) Calls.Add(ctx);
            return Task.FromResult<IReadOnlyList<IReadOnlyList<Item>>>(Array.Empty<IReadOnlyList<Item>>());
        }
    }

    [Fact]
    public async Task Fanout_two_downstream_terminals_both_receive_and_complete()
    {
        var recA = new Recorder { Id = "sink.a" };
        var recB = new Recorder { Id = "sink.b" };
        var (er2, nr2, exec2, _) = Build(new TestTrigger(), recA, recB);

        var graph = new WorkflowGraph
        {
            Nodes = new[] {
                new WorkflowNode("t",  "test.trigger", null, "{}"),
                new WorkflowNode("a",  "sink.a",       null, "{}"),
                new WorkflowNode("b",  "sink.b",       null, "{}"),
            },
            Edges = new[] {
                new WorkflowEdge("t", 0, "a", 0),
                new WorkflowEdge("t", 0, "b", 0),
            },
        };

        var eid = await exec2.StartExecutionAsync("wf", graph, "t", "manual",
            new[] { new Core.Item(Item(1)) }, null, null, CancellationToken.None);

        Assert.Equal("completed", er2.Get(eid)!.Status);
        Assert.Single(recA.Calls);
        Assert.Single(recB.Calls);
    }

    [Fact]
    public async Task Failure_propagates_downstream_as_skipped_and_rolls_up_to_partial_or_failed()
    {
        var recDownstreamOfFailure = new Recorder { Id = "sink.after_throw" };
        var recDownstreamOfSuccess = new Recorder { Id = "sink.after_echo" };
        var (er2, nr2, exec2, _) = Build(new TestTrigger(), new Throwing(), new EchoAction(),
            recDownstreamOfFailure, recDownstreamOfSuccess);

        var graph = new WorkflowGraph
        {
            Nodes = new[] {
                new WorkflowNode("t",  "test.trigger",       null, "{}"),
                new WorkflowNode("tf", "test.throw",         null, "{}"),
                new WorkflowNode("te", "test.echo",          null, "{}"),
                new WorkflowNode("sf", "sink.after_throw",   null, "{}"),
                new WorkflowNode("se", "sink.after_echo",    null, "{}"),
            },
            Edges = new[] {
                new WorkflowEdge("t",  0, "tf", 0),
                new WorkflowEdge("t",  0, "te", 0),
                new WorkflowEdge("tf", 0, "sf", 0),
                new WorkflowEdge("te", 0, "se", 0),
            },
        };

        var eid = await exec2.StartExecutionAsync("wf", graph, "t", "manual",
            new[] { new Core.Item(Item(1)) }, null, null, CancellationToken.None);

        var rows = nr2.ListByExecution(eid);
        Assert.Equal("failed", rows.Single(r => r.NodeId == "tf").Status);
        Assert.Equal("skipped_due_to_failure", rows.Single(r => r.NodeId == "sf").Status);
        Assert.Equal("completed", rows.Single(r => r.NodeId == "te").Status);
        Assert.Equal("completed", rows.Single(r => r.NodeId == "se").Status);
        // Trigger succeeded + one branch succeeded + one branch failed → partial.
        Assert.Equal("partial", er2.Get(eid)!.Status);
        Assert.Empty(recDownstreamOfFailure.Calls);
        Assert.Single(recDownstreamOfSuccess.Calls);
    }

    [Fact]
    public async Task Config_string_fields_are_templated_from_first_input_item_before_node_runs()
    {
        var rec = new Recorder { Id = "sink.template" };
        var (er, nr, exec, _) = Build(new TestTrigger(), rec);

        var graph = new WorkflowGraph
        {
            Nodes = new[] {
                new WorkflowNode("t", "test.trigger",  null, "{}"),
                new WorkflowNode("s", "sink.template", null,
                    """{"greeting":"hi {input.name}","meta":{"deep":"v={input.tags[0]}"},"list":["a","b {input.name}"]}"""),
            },
            Edges = new[] { new WorkflowEdge("t", 0, "s", 0) },
        };

        var trig = new Core.Item(JsonSerializer.SerializeToElement(new { name = "world", tags = new[] { "x", "y" } }));
        var eid = await exec.StartExecutionAsync("wf", graph, "t", "manual",
            new[] { trig }, null, null, CancellationToken.None);

        Assert.Equal("completed", er.Get(eid)!.Status);
        var cfg = rec.Calls.Single().Config;
        Assert.Equal("hi world", cfg.GetProperty("greeting").GetString());
        Assert.Equal("v=x", cfg.GetProperty("meta").GetProperty("deep").GetString());
        var list = cfg.GetProperty("list");
        Assert.Equal("a", list[0].GetString());
        Assert.Equal("b world", list[1].GetString());
    }

    [Fact]
    public async Task Missing_input_property_renders_as_empty_in_templated_config()
    {
        var rec = new Recorder { Id = "sink.template2" };
        var (er, _, exec, _) = Build(new TestTrigger(), rec);

        var graph = new WorkflowGraph
        {
            Nodes = new[] {
                new WorkflowNode("t", "test.trigger",   null, "{}"),
                new WorkflowNode("s", "sink.template2", null, """{"msg":"value: {input.missing}"}"""),
            },
            Edges = new[] { new WorkflowEdge("t", 0, "s", 0) },
        };

        var trig = new Core.Item(JsonSerializer.SerializeToElement(new { other = 1 }));
        var eid = await exec.StartExecutionAsync("wf", graph, "t", "manual",
            new[] { trig }, null, null, CancellationToken.None);

        Assert.Equal("completed", er.Get(eid)!.Status);
        Assert.Equal("value: ", rec.Calls.Single().Config.GetProperty("msg").GetString());
    }

    [Fact]
    public async Task All_downstream_failed_rolls_up_to_failed()
    {
        var (er, nr, exec, _) = Build(new TestTrigger(), new Throwing());
        var graph = new WorkflowGraph
        {
            Nodes = new[] {
                new WorkflowNode("t",  "test.trigger", null, "{}"),
                new WorkflowNode("tf", "test.throw",   null, "{}"),
            },
            Edges = new[] { new WorkflowEdge("t", 0, "tf", 0) },
        };
        var eid = await exec.StartExecutionAsync("wf", graph, "t", "manual",
            new[] { new Core.Item(Item(1)) }, null, null, CancellationToken.None);
        Assert.Equal("failed", er.Get(eid)!.Status);
    }

    // --- coercion of templated primitives ---------------------------------

    // A test node that requires status to be a JSON number, mirroring
    // WebResponseNode's real check pattern. We assert on ctx.Config here.
    private sealed class NumberFieldNode : IActionNode
    {
        public List<INodeContext> Calls { get; } = new();
        public string Id { get; init; } = "test.numberfield";
        public string DisplayName => Id;
        public NodeCategory Category => NodeCategory.Terminal;
        public int InputCount => 1;
        public int OutputCount => 0;
        public string? ConnectionProviderId => null;
        public JsonSchema ConfigSchema => new(JsonDocument.Parse("""
            {
              "type": "object",
              "properties": { "status": { "type": "integer" } }
            }
        """).RootElement);
        public Task<IReadOnlyList<IReadOnlyList<Item>>> ExecuteAsync(
            INodeContext ctx, IReadOnlyList<IReadOnlyList<Item>> _, CancellationToken __)
        {
            // Strict JsonValueKind.Number check — the coercion pass must produce
            // a number here, not a string.
            var s = ctx.Config.GetProperty("status");
            if (s.ValueKind != JsonValueKind.Number)
                throw new InvalidOperationException($"expected number, got {s.ValueKind}");
            lock (Calls) Calls.Add(ctx);
            return Task.FromResult<IReadOnlyList<IReadOnlyList<Item>>>(Array.Empty<IReadOnlyList<Item>>());
        }
    }

    [Fact]
    public async Task Template_into_integer_config_field_is_coerced_to_number()
    {
        var rec = new NumberFieldNode();
        var (er, nr, exec, _) = Build(new TestTrigger(), rec);

        var graph = new WorkflowGraph
        {
            Nodes = new[]
            {
                new WorkflowNode("t", "test.trigger",     null, "{}"),
                new WorkflowNode("s", "test.numberfield", null, """{"status":"{input.code}"}"""),
            },
            Edges = new[] { new WorkflowEdge("t", 0, "s", 0) },
        };

        var trig = new Core.Item(JsonSerializer.SerializeToElement(new { code = 201 }));
        var eid = await exec.StartExecutionAsync("wf", graph, "t", "manual",
            new[] { trig }, null, null, CancellationToken.None);

        Assert.Equal("completed", er.Get(eid)!.Status);
        var got = rec.Calls.Single().Config.GetProperty("status");
        Assert.Equal(JsonValueKind.Number, got.ValueKind);
        Assert.Equal(201, got.GetInt32());
    }

    [Fact]
    public async Task Non_parseable_integer_config_fails_node_with_clear_message()
    {
        var rec = new NumberFieldNode();
        var (er, nr, exec, _) = Build(new TestTrigger(), rec);

        var graph = new WorkflowGraph
        {
            Nodes = new[]
            {
                new WorkflowNode("t", "test.trigger",     null, "{}"),
                new WorkflowNode("s", "test.numberfield", null, """{"status":"{input.code}"}"""),
            },
            Edges = new[] { new WorkflowEdge("t", 0, "s", 0) },
        };

        var trig = new Core.Item(JsonSerializer.SerializeToElement(new { code = "abc" }));
        var eid = await exec.StartExecutionAsync("wf", graph, "t", "manual",
            new[] { trig }, null, null, CancellationToken.None);

        Assert.Equal("failed", er.Get(eid)!.Status);
        var row = nr.ListByExecution(eid).Single(r => r.NodeId == "s");
        Assert.Equal("failed", row.Status);
        Assert.NotNull(row.Error);
        Assert.Contains("expected integer", row.Error!);
        Assert.Contains("status", row.Error!);
        Assert.Empty(rec.Calls);
    }

    // --- WrapPin helper --------------------------------------------------

    [Fact]
    public void WrapPin_object_produces_a_single_item_wrapping_the_object()
    {
        var pin = JsonDocument.Parse("""{"a":1}""").RootElement;
        var items = Executor.WrapPin(pin);
        Assert.Single(items);
        Assert.Equal("""{"a":1}""", items[0].Json.GetRawText());
        Assert.Null(items[0].Binary);
    }

    [Fact]
    public void WrapPin_array_of_objects_produces_one_item_per_element()
    {
        var pin = JsonDocument.Parse("""[{"a":1},{"a":2}]""").RootElement;
        var items = Executor.WrapPin(pin);
        Assert.Equal(2, items.Count);
        Assert.Equal("""{"a":1}""", items[0].Json.GetRawText());
        Assert.Equal("""{"a":2}""", items[1].Json.GetRawText());
    }

    [Fact]
    public void WrapPin_empty_array_produces_zero_items()
    {
        var pin = JsonDocument.Parse("[]").RootElement;
        var items = Executor.WrapPin(pin);
        Assert.Empty(items);
    }

    [Fact]
    public void WrapPin_scalar_produces_a_single_item_with_the_scalar()
    {
        var pin = JsonDocument.Parse("42").RootElement;
        var items = Executor.WrapPin(pin);
        Assert.Single(items);
        Assert.Equal("42", items[0].Json.GetRawText());
    }

    // --- trigger pinned_output on manual runs ------------------------------

    [Fact]
    public async Task Trigger_pinned_output_overrides_trigger_items_on_manual_run()
    {
        var rec = new Recorder { Id = "sink.trig_pin_manual" };
        var (er, nr, exec, _) = Build(new TestTrigger(), rec);

        var graph = new WorkflowGraph
        {
            Nodes = new[]
            {
                new WorkflowNode("t", "test.trigger", null, "{}",
                    PinnedOutputJson: """{"pinned":true}"""),
                new WorkflowNode("s", "sink.trig_pin_manual", null, "{}"),
            },
            Edges = new[] { new WorkflowEdge("t", 0, "s", 0) },
        };

        var passedIn = new Core.Item(JsonSerializer.SerializeToElement(new { pinned = false }));
        var eid = await exec.StartExecutionAsync("wf", graph, "t", "manual",
            new[] { passedIn }, null, null, CancellationToken.None);

        Assert.Equal("completed", er.Get(eid)!.Status);
        var sinkRow = nr.ListByExecution(eid).Single(r => r.NodeId == "s");
        using var inputs = JsonDocument.Parse(sinkRow.InputsJson);
        var port0Item0Json = inputs.RootElement[0][0].GetProperty("json");
        Assert.True(port0Item0Json.GetProperty("pinned").GetBoolean());
    }

    [Fact]
    public async Task Trigger_pinned_output_is_ignored_on_non_manual_run()
    {
        var rec = new Recorder { Id = "sink.trig_pin_webhook" };
        var (er, nr, exec, _) = Build(new TestTrigger(), rec);

        var graph = new WorkflowGraph
        {
            Nodes = new[]
            {
                new WorkflowNode("t", "test.trigger", null, "{}",
                    PinnedOutputJson: """{"pinned":true}"""),
                new WorkflowNode("s", "sink.trig_pin_webhook", null, "{}"),
            },
            Edges = new[] { new WorkflowEdge("t", 0, "s", 0) },
        };

        var passedIn = new Core.Item(JsonSerializer.SerializeToElement(new { pinned = false }));
        var eid = await exec.StartExecutionAsync("wf", graph, "t", "http.webhook",
            new[] { passedIn }, null, null, CancellationToken.None);

        Assert.Equal("completed", er.Get(eid)!.Status);
        var sinkRow = nr.ListByExecution(eid).Single(r => r.NodeId == "s");
        using var inputs = JsonDocument.Parse(sinkRow.InputsJson);
        var port0Item0Json = inputs.RootElement[0][0].GetProperty("json");
        Assert.False(port0Item0Json.GetProperty("pinned").GetBoolean());
    }

    // --- node pinned_input on manual runs ---------------------------------

    [Fact]
    public async Task Node_pinned_input_overrides_upstream_output_on_manual_run()
    {
        var rec = new Recorder { Id = "sink.node_pin_manual" };
        var (er, nr, exec, _) = Build(new TestTrigger(), new EchoAction(), rec);

        var graph = new WorkflowGraph
        {
            Nodes = new[]
            {
                new WorkflowNode("t", "test.trigger", null, "{}"),
                new WorkflowNode("a", "test.echo",    null, "{}"),
                new WorkflowNode("s", "sink.node_pin_manual", null, "{}",
                    PinnedInputJson: """{"pinned":true}"""),
            },
            Edges = new[]
            {
                new WorkflowEdge("t", 0, "a", 0),
                new WorkflowEdge("a", 0, "s", 0),
            },
        };

        var passedIn = new Core.Item(JsonSerializer.SerializeToElement(new { pinned = false }));
        var eid = await exec.StartExecutionAsync("wf", graph, "t", "manual",
            new[] { passedIn }, null, null, CancellationToken.None);

        Assert.Equal("completed", er.Get(eid)!.Status);
        // Sink's persisted inputs must be the pin, not what `a` emitted.
        var sinkRow = nr.ListByExecution(eid).Single(r => r.NodeId == "s");
        using var inputs = JsonDocument.Parse(sinkRow.InputsJson);
        var port0Item0Json = inputs.RootElement[0][0].GetProperty("json");
        Assert.True(port0Item0Json.GetProperty("pinned").GetBoolean());
        // `a` still ran and its own output row exists — confirming we don't skip upstream.
        Assert.Contains(nr.ListByExecution(eid), r => r.NodeId == "a" && r.Status == "completed");
    }

    [Fact]
    public async Task Node_pinned_input_is_ignored_on_non_manual_run()
    {
        var rec = new Recorder { Id = "sink.node_pin_webhook" };
        var (er, nr, exec, _) = Build(new TestTrigger(), new EchoAction(), rec);

        var graph = new WorkflowGraph
        {
            Nodes = new[]
            {
                new WorkflowNode("t", "test.trigger", null, "{}"),
                new WorkflowNode("a", "test.echo",    null, "{}"),
                new WorkflowNode("s", "sink.node_pin_webhook", null, "{}",
                    PinnedInputJson: """{"pinned":true}"""),
            },
            Edges = new[]
            {
                new WorkflowEdge("t", 0, "a", 0),
                new WorkflowEdge("a", 0, "s", 0),
            },
        };

        var passedIn = new Core.Item(JsonSerializer.SerializeToElement(new { pinned = false }));
        var eid = await exec.StartExecutionAsync("wf", graph, "t", "http.webhook",
            new[] { passedIn }, null, null, CancellationToken.None);

        Assert.Equal("completed", er.Get(eid)!.Status);
        var sinkRow = nr.ListByExecution(eid).Single(r => r.NodeId == "s");
        using var inputs = JsonDocument.Parse(sinkRow.InputsJson);
        var port0Item0Json = inputs.RootElement[0][0].GetProperty("json");
        // Real upstream (echo of the trigger's `pinned:false`) reaches the sink.
        Assert.False(port0Item0Json.GetProperty("pinned").GetBoolean());
    }
}