using System.Diagnostics;
using Microsoft.Extensions.Logging.Abstractions;
using Opc.Ua;
using Opc.Ua.Server;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using HistorianRead = ZB.MOM.WW.OtOpcUa.Core.Abstractions.HistoryReadResult;
using SdkHistoryReadResult = Opc.Ua.HistoryReadResult;

namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests;

/// <summary>
///     Arch-review 03/S3 — the node-manager's HistoryRead arms must serve a multi-node batch with BOUNDED
///     per-node parallelism, a per-request deadline, and a process-wide concurrent-batch limiter, rather
///     than block-bridging the gateway per node sequentially on an SDK request thread. Boots the same real
///     <see cref="OtOpcUaSdkServer"/> harness as <see cref="NodeManagerHistoryReadTests"/> and drives the
///     public <c>HistoryRead</c> with a delaying fake <see cref="IHistorianDataSource"/> that records the
///     maximum observed concurrent reads (deterministic — no raw-timing dependence for the primary
///     assertions).
/// </summary>
public sealed class NodeManagerHistoryReadConcurrencyTests : IDisposable
{
    private static CancellationToken Ct => TestContext.Current.CancellationToken;

    private readonly string _pkiRoot = Path.Combine(
        Path.GetTempPath(),
        $"otopcua-historyread-conc-{Guid.NewGuid():N}");

    /// <summary>
    ///     RED repro (03/S3): a single Raw HistoryRead naming 8 historized nodes against a 150 ms-per-read
    ///     backend must serve at least two nodes concurrently. On the current sequential <c>foreach</c> the
    ///     max observed concurrency is pinned at 1, so this fails.
    /// </summary>
    [Fact]
    public async Task Raw_batch_is_served_with_bounded_parallelism()
    {
        var (host, server) = await BootAsync();
        var nm = server.NodeManager!;
        var fake = new ConcurrencyTrackingHistorianDataSource(TimeSpan.FromMilliseconds(150));
        nm.HistorianDataSource = fake;

        var nodeIds = MaterializeHistorizedNodes(nm, count: 8);

        var details = new ReadRawModifiedDetails
        {
            StartTime = DateTime.UtcNow.AddHours(-1),
            EndTime = DateTime.UtcNow,
            NumValuesPerNode = 10,
            IsReadModified = false,
        };

        var sw = Stopwatch.StartNew();
        await Task.Run(() => InvokeHistoryRead(server, nm, details, nodeIds), Ct)
            .WaitAsync(TimeSpan.FromSeconds(15), Ct);
        sw.Stop();

        // Primary (deterministic): the batch was served with >1 node in flight at once.
        fake.MaxObservedConcurrency.ShouldBeGreaterThanOrEqualTo(2);

        await host.DisposeAsync();
    }

    /// <summary>
    ///     Parallelism is bounded ABOVE by <c>HistoryReadBatchParallelism</c>: with the property set to 3 and
    ///     12 nodes in flight, no more than 3 reads are ever concurrent (03/S3).
    /// </summary>
    [Fact]
    public async Task Parallelism_is_bounded_above()
    {
        var (host, server) = await BootAsync();
        var nm = server.NodeManager!;
        nm.HistoryReadBatchParallelism = 3;
        var fake = new ConcurrencyTrackingHistorianDataSource(TimeSpan.FromMilliseconds(100));
        nm.HistorianDataSource = fake;

        var nodeIds = MaterializeHistorizedNodes(nm, count: 12);
        var details = new ReadRawModifiedDetails
        {
            StartTime = DateTime.UtcNow.AddHours(-1),
            EndTime = DateTime.UtcNow,
            NumValuesPerNode = 10,
            IsReadModified = false,
        };

        await Task.Run(() => InvokeHistoryRead(server, nm, details, nodeIds), Ct)
            .WaitAsync(TimeSpan.FromSeconds(15), Ct);

        fake.MaxObservedConcurrency.ShouldBeGreaterThanOrEqualTo(2);
        fake.MaxObservedConcurrency.ShouldBeLessThanOrEqualTo(3);

        await host.DisposeAsync();
    }

    /// <summary>
    ///     One node's backend throw does not poison a parallel batch: that node surfaces
    ///     <c>BadHistoryOperationUnsupported</c> while the other seven return Good (per-node isolation holds
    ///     under parallelism — 03/S3).
    /// </summary>
    [Fact]
    public async Task One_node_failure_does_not_poison_a_parallel_batch()
    {
        var (host, server) = await BootAsync();
        var nm = server.NodeManager!;

        var src = DateTime.UtcNow.AddSeconds(-5);
        var srv = DateTime.UtcNow;
        // Node index 3's tagname throws; all others return a single good sample.
        var fake = new PerTagnameFake(tagname => tagname == "WW.Tag3"
            ? throw new InvalidOperationException("backend boom for Tag3")
            : new HistorianRead(new[] { new DataValueSnapshot(1.0f, StatusCodes.Good, src, srv) }, null));
        nm.HistorianDataSource = fake;

        var nodeIds = MaterializeHistorizedNodes(nm, count: 8);
        var details = new ReadRawModifiedDetails
        {
            StartTime = DateTime.UtcNow.AddHours(-1),
            EndTime = DateTime.UtcNow,
            NumValuesPerNode = 10,
            IsReadModified = false,
        };

        var (results, errors) = await Task.Run(() => InvokeHistoryRead(server, nm, details, nodeIds), Ct)
            .WaitAsync(TimeSpan.FromSeconds(15), Ct);

        for (var i = 0; i < 8; i++)
        {
            if (i == 3)
            {
                StatusCode.IsBad(errors[i].StatusCode).ShouldBeTrue();
                errors[i].StatusCode.Code.ShouldBe(StatusCodes.BadHistoryOperationUnsupported);
            }
            else
            {
                errors[i].StatusCode.Code.ShouldBe(StatusCodes.Good);
                results[i].StatusCode.Code.ShouldBe(StatusCodes.Good);
            }
        }

        await host.DisposeAsync();
    }

    /// <summary>
    ///     A per-request deadline bounds a hung backend: reads that only complete on token cancellation
    ///     surface <c>BadTimeout</c> for every node and the call returns within the deadline (plus slack) —
    ///     it does not hang an SDK request thread indefinitely (03/S3).
    /// </summary>
    [Fact]
    public async Task Deadline_bounds_a_hung_backend()
    {
        var (host, server) = await BootAsync();
        var nm = server.NodeManager!;
        nm.HistoryReadDeadline = TimeSpan.FromMilliseconds(500);
        nm.HistorianDataSource = new HangingHistorianDataSource();

        var nodeIds = MaterializeHistorizedNodes(nm, count: 4);
        var details = new ReadRawModifiedDetails
        {
            StartTime = DateTime.UtcNow.AddHours(-1),
            EndTime = DateTime.UtcNow,
            NumValuesPerNode = 10,
            IsReadModified = false,
        };

        var (_, errors) = await Task.Run(() => InvokeHistoryRead(server, nm, details, nodeIds), Ct)
            .WaitAsync(TimeSpan.FromSeconds(10), Ct);

        for (var i = 0; i < 4; i++)
            errors[i].StatusCode.Code.ShouldBe(StatusCodes.BadTimeout);

        await host.DisposeAsync();
    }

    /// <summary>
    ///     A saturated process-wide batch limiter degrades a second concurrent HistoryRead to
    ///     <c>BadTooManyOperations</c> (rather than exhausting the SDK request-thread pool); once the parked
    ///     batch completes the slot frees and a later read succeeds (03/S3).
    /// </summary>
    [Fact]
    public async Task Saturated_limiter_rejects_with_BadTooManyOperations()
    {
        var (host, server) = await BootAsync();
        var nm = server.NodeManager!;
        nm.MaxConcurrentHistoryReadBatches = 1;
        nm.HistoryReadDeadline = TimeSpan.FromMilliseconds(200);
        var fake = new GatedRawFake();
        nm.HistorianDataSource = fake;

        var ids = MaterializeHistorizedNodes(nm, count: 2);
        var details = new ReadRawModifiedDetails
        {
            StartTime = DateTime.UtcNow.AddHours(-1),
            EndTime = DateTime.UtcNow,
            NumValuesPerNode = 10,
            IsReadModified = false,
        };

        // First batch parks in the gated fake, holding the single limiter permit.
        var first = Task.Run(() => InvokeHistoryRead(server, nm, details, ids[0]), Ct);

        var spin = Stopwatch.StartNew();
        while (fake.Entered == 0 && spin.Elapsed < TimeSpan.FromSeconds(5))
            await Task.Delay(10, Ct);
        fake.Entered.ShouldBe(1);

        // Second concurrent batch: the limiter is saturated → all handles BadTooManyOperations.
        var (_, errors2) = await Task.Run(() => InvokeHistoryRead(server, nm, details, ids[1]), Ct)
            .WaitAsync(TimeSpan.FromSeconds(5), Ct);
        errors2[0].StatusCode.Code.ShouldBe(StatusCodes.BadTooManyOperations);
        fake.Entered.ShouldBe(1, "the rejected batch must never reach the backend");

        // Release the parked batch; the slot frees and a later read succeeds.
        fake.Release();
        await first.WaitAsync(TimeSpan.FromSeconds(10), Ct);

        var (_, errors3) = await Task.Run(() => InvokeHistoryRead(server, nm, details, ids[1]), Ct)
            .WaitAsync(TimeSpan.FromSeconds(10), Ct);
        errors3[0].StatusCode.Code.ShouldBe(StatusCodes.Good);

        await host.DisposeAsync();
    }

    /// <summary>Materialises <paramref name="count"/> historized Float variables and returns their NodeIds.</summary>
    private static NodeId[] MaterializeHistorizedNodes(OtOpcUaNodeManager nm, int count)
    {
        var ids = new NodeId[count];
        for (var i = 0; i < count; i++)
        {
            var key = $"eq/tag{i}";
            nm.EnsureVariable(key, parentFolderNodeId: null, displayName: $"Tag{i}", dataType: "Float",
                writable: false, historianTagname: $"WW.Tag{i}");
            ids[i] = nm.TryGetVariable(key)!.NodeId;
        }

        return ids;
    }

    /// <summary>Issues one HistoryRead batch for all supplied node ids in order.</summary>
    private static (IList<SdkHistoryReadResult> Results, IList<ServiceResult> Errors) InvokeHistoryRead(
        OtOpcUaSdkServer server, OtOpcUaNodeManager nm, HistoryReadDetails details, params NodeId[] nodeIds)
    {
        var context = new OperationContext(
            new RequestHeader(), secureChannelContext: null, RequestType.HistoryRead, identity: null);

        var nodesToRead = nodeIds.Select(id => new HistoryReadValueId { NodeId = id }).ToList();
        var results = Enumerable.Repeat<SdkHistoryReadResult>(null!, nodeIds.Length).ToList();
        var errors = Enumerable.Repeat<ServiceResult>(null!, nodeIds.Length).ToList();

        nm.HistoryRead(
            context,
            details,
            TimestampsToReturn.Both,
            releaseContinuationPoints: false,
            nodesToRead,
            results,
            errors);

        return (results, errors);
    }

    /// <summary>
    ///     A delaying fake historian source that tracks the maximum number of reads in flight at once
    ///     (via interlocked live-counter) so a test can prove per-node parallelism deterministically.
    /// </summary>
    private sealed class ConcurrencyTrackingHistorianDataSource(TimeSpan delay) : IHistorianDataSource
    {
        private int _current;
        private int _max;

        /// <summary>The highest number of concurrently in-flight reads observed.</summary>
        public int MaxObservedConcurrency => Volatile.Read(ref _max);

        private async Task<HistorianRead> TrackAsync(CancellationToken ct)
        {
            var now = Interlocked.Increment(ref _current);
            UpdateMax(now);
            try
            {
                await Task.Delay(delay, ct).ConfigureAwait(false);
            }
            finally
            {
                Interlocked.Decrement(ref _current);
            }

            return new HistorianRead(
                new[] { new DataValueSnapshot(1.0f, StatusCodes.Good, DateTime.UtcNow, DateTime.UtcNow) }, null);
        }

        private void UpdateMax(int observed)
        {
            int cur;
            while ((cur = Volatile.Read(ref _max)) < observed &&
                   Interlocked.CompareExchange(ref _max, observed, cur) != cur)
            {
            }
        }

        public Task<HistorianRead> ReadRawAsync(
            string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode,
            CancellationToken cancellationToken) => TrackAsync(cancellationToken);

        public Task<HistorianRead> ReadProcessedAsync(
            string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval,
            HistoryAggregateType aggregate, CancellationToken cancellationToken) => TrackAsync(cancellationToken);

        public Task<HistorianRead> ReadAtTimeAsync(
            string fullReference, IReadOnlyList<DateTime> timestampsUtc, CancellationToken cancellationToken) =>
            TrackAsync(cancellationToken);

        public Task<HistoricalEventsResult> ReadEventsAsync(
            string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents,
            CancellationToken cancellationToken) =>
            Task.FromResult(new HistoricalEventsResult(Array.Empty<HistoricalEvent>(), null));

        public HistorianHealthSnapshot GetHealthSnapshot() => NullHistorianDataSource.Instance.GetHealthSnapshot();

        public void Dispose()
        {
        }
    }

    /// <summary>A fake whose reads complete ONLY on token cancellation — used to prove the per-request
    /// deadline surfaces <c>BadTimeout</c> rather than hanging.</summary>
    private sealed class HangingHistorianDataSource : IHistorianDataSource
    {
        private static async Task<HistorianRead> HangAsync(CancellationToken ct)
        {
            await Task.Delay(Timeout.Infinite, ct).ConfigureAwait(false);
            return new HistorianRead(Array.Empty<DataValueSnapshot>(), null);
        }

        public Task<HistorianRead> ReadRawAsync(
            string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode,
            CancellationToken cancellationToken) => HangAsync(cancellationToken);

        public Task<HistorianRead> ReadProcessedAsync(
            string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval,
            HistoryAggregateType aggregate, CancellationToken cancellationToken) => HangAsync(cancellationToken);

        public Task<HistorianRead> ReadAtTimeAsync(
            string fullReference, IReadOnlyList<DateTime> timestampsUtc, CancellationToken cancellationToken) =>
            HangAsync(cancellationToken);

        public async Task<HistoricalEventsResult> ReadEventsAsync(
            string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents,
            CancellationToken cancellationToken)
        {
            await Task.Delay(Timeout.Infinite, cancellationToken).ConfigureAwait(false);
            return new HistoricalEventsResult(Array.Empty<HistoricalEvent>(), null);
        }

        public HistorianHealthSnapshot GetHealthSnapshot() => NullHistorianDataSource.Instance.GetHealthSnapshot();

        public void Dispose() { }
    }

    /// <summary>A fake whose raw read parks (awaits a <see cref="TaskCompletionSource"/>) until the test
    /// releases it — used to hold the process-wide batch limiter permit while a second batch is issued.</summary>
    private sealed class GatedRawFake : IHistorianDataSource
    {
        private readonly TaskCompletionSource _gate = new(TaskCreationOptions.RunContinuationsAsynchronously);
        private int _entered;

        /// <summary>Number of raw reads that reached the backend body.</summary>
        public int Entered => Volatile.Read(ref _entered);

        /// <summary>Releases the parked read so its awaiting batch can complete.</summary>
        public void Release() => _gate.TrySetResult();

        public async Task<HistorianRead> ReadRawAsync(
            string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode,
            CancellationToken cancellationToken)
        {
            Interlocked.Increment(ref _entered);
            // Deliberately ignore the per-request deadline token: the first batch must keep HOLDING the
            // single limiter permit until the test releases it, so the second batch deterministically hits
            // a saturated limiter (rather than the first batch's own deadline freeing the slot early).
            await _gate.Task.ConfigureAwait(false);
            return new HistorianRead(
                new[] { new DataValueSnapshot(1.0f, StatusCodes.Good, DateTime.UtcNow, DateTime.UtcNow) }, null);
        }

        public Task<HistorianRead> ReadProcessedAsync(
            string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval,
            HistoryAggregateType aggregate, CancellationToken cancellationToken) =>
            NullHistorianDataSource.Instance.ReadProcessedAsync(fullReference, startUtc, endUtc, interval, aggregate, cancellationToken);

        public Task<HistorianRead> ReadAtTimeAsync(
            string fullReference, IReadOnlyList<DateTime> timestampsUtc, CancellationToken cancellationToken) =>
            NullHistorianDataSource.Instance.ReadAtTimeAsync(fullReference, timestampsUtc, cancellationToken);

        public Task<HistoricalEventsResult> ReadEventsAsync(
            string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents,
            CancellationToken cancellationToken) =>
            NullHistorianDataSource.Instance.ReadEventsAsync(sourceName, startUtc, endUtc, maxEvents, cancellationToken);

        public HistorianHealthSnapshot GetHealthSnapshot() => NullHistorianDataSource.Instance.GetHealthSnapshot();

        public void Dispose() { }
    }

    /// <summary>A per-tagname fake that delegates each raw read to a caller-supplied func (which may throw
    /// to simulate a per-node backend failure). Processed/AtTime/Events delegate to the null source.</summary>
    private sealed class PerTagnameFake(Func<string, HistorianRead> rawHandler) : IHistorianDataSource
    {
        public Task<HistorianRead> ReadRawAsync(
            string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode,
            CancellationToken cancellationToken) => Task.FromResult(rawHandler(fullReference));

        public Task<HistorianRead> ReadProcessedAsync(
            string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval,
            HistoryAggregateType aggregate, CancellationToken cancellationToken) =>
            NullHistorianDataSource.Instance.ReadProcessedAsync(fullReference, startUtc, endUtc, interval, aggregate, cancellationToken);

        public Task<HistorianRead> ReadAtTimeAsync(
            string fullReference, IReadOnlyList<DateTime> timestampsUtc, CancellationToken cancellationToken) =>
            NullHistorianDataSource.Instance.ReadAtTimeAsync(fullReference, timestampsUtc, cancellationToken);

        public Task<HistoricalEventsResult> ReadEventsAsync(
            string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents,
            CancellationToken cancellationToken) =>
            NullHistorianDataSource.Instance.ReadEventsAsync(sourceName, startUtc, endUtc, maxEvents, cancellationToken);

        public HistorianHealthSnapshot GetHealthSnapshot() => NullHistorianDataSource.Instance.GetHealthSnapshot();

        public void Dispose() { }
    }

    private async Task<(OpcUaApplicationHost Host, OtOpcUaSdkServer Server)> BootAsync()
    {
        var host = new OpcUaApplicationHost(
            new OpcUaApplicationHostOptions
            {
                ApplicationName = "OtOpcUa.HistoryReadConcTest",
                ApplicationUri = $"urn:OtOpcUa.HistoryReadConcTest:{Guid.NewGuid():N}",
                OpcUaPort = AllocateFreePort(),
                PublicHostname = "localhost",
                PkiStoreRoot = _pkiRoot,
            },
            NullLogger<OpcUaApplicationHost>.Instance);

        var server = new OtOpcUaSdkServer();
        await host.StartAsync(server, Ct);
        return (host, server);
    }

    private static int AllocateFreePort()
    {
        using var listener = new System.Net.Sockets.TcpListener(System.Net.IPAddress.Loopback, 0);
        listener.Start();
        var port = ((System.Net.IPEndPoint)listener.LocalEndpoint).Port;
        listener.Stop();
        return port;
    }

    /// <summary>Cleans up the PKI root directory.</summary>
    public void Dispose()
    {
        if (Directory.Exists(_pkiRoot))
        {
            try { Directory.Delete(_pkiRoot, recursive: true); }
            catch { /* best-effort cleanup */ }
        }
    }
}