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