using Akka.Actor;
using Akka.TestKit.Xunit2;
using NSubstitute;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Protocol;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.DataConnection;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Actors;
using ZB.MOM.WW.ScadaBridge.HealthMonitoring;

namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests.Actors;

/// <summary>
/// WP2.1b — the DataConnectionActor against a batch-capable adapter: chunked subscribe,
/// chunked/bounded re-subscribe, the seed deadline, batched tag-resolution probes, the
/// partial-failure surface and the coalesced quality push. The existing suite covers the
/// same actor against a NON batch-capable adapter, which keeps the per-tag fallback pinned.
/// </summary>
public class DataConnectionActorBatchTests : TestKit
{
    private readonly ISiteHealthCollector _health = Substitute.For<ISiteHealthCollector>();
    private readonly IDataConnectionFactory _factory = Substitute.For<IDataConnectionFactory>();

    private static DataConnectionOptions Options() => new()
    {
        ReconnectInterval = TimeSpan.FromMilliseconds(100),
        TagResolutionRetryInterval = TimeSpan.FromMilliseconds(150),
        TagResolutionRetryMaxInterval = TimeSpan.FromMilliseconds(600),
        WriteTimeout = TimeSpan.FromSeconds(5),
        SeedReadMaxAttempts = 1,
        SeedReadRetryDelay = TimeSpan.FromMilliseconds(10),
        SubscribeBatchSize = 5,
        SubscribeBatchDelay = TimeSpan.FromMilliseconds(50),
        SeedReadBatchSize = 4,
        SeedReadMaxParallelism = 2,
        SeedOverallTimeout = TimeSpan.FromSeconds(30),
        QualityFlushInterval = TimeSpan.FromMilliseconds(200)
    };

    private IActorRef CreateActor(FakeBatchDataConnection adapter, DataConnectionOptions options, string name) =>
        Sys.ActorOf(Props.Create(() => new DataConnectionActor(
            name, adapter, options, _health, _factory, "OpcUa")), name);

    private static string[] Tags(int count) => Enumerable.Range(1, count).Select(i => $"tag{i}").ToArray();

    [Fact]
    public void Subscribe_IssuesOneAdapterRoundTripPerChunk_NotPerTag()
    {
        // 12 tags at SubscribeBatchSize 5 → 3 batch calls (5/5/2) and ZERO per-tag
        // subscribes. This is finding #3: the old path was one adapter call (and one OPC UA
        // ApplyChanges) per tag.
        var adapter = new FakeBatchDataConnection();
        var actor = CreateActor(adapter, Options(), "batch-chunking");

        actor.Tell(new SubscribeTagsRequest("c1", "inst1", "batch-chunking", Tags(12), DateTimeOffset.UtcNow));
        ExpectMsg<SubscribeTagsResponse>(m => m.Success, TimeSpan.FromSeconds(5));

        var batches = adapter.SubscribeBatches.ToList();
        Assert.Equal(3, batches.Count);
        Assert.Equal([5, 5, 2], batches.Select(b => b.Count));
        Assert.Equal(0, Volatile.Read(ref adapter.SingleSubscribeCalls));
    }

    [Fact]
    public void Seed_UsesChunkedBulkReads_NotPerTagReads()
    {
        // Seeding 6 tags at SeedReadBatchSize 4 → 2 bulk reads, no single-tag reads.
        var adapter = new FakeBatchDataConnection();
        var actor = CreateActor(adapter, Options(), "batch-seed");

        actor.Tell(new SubscribeTagsRequest("c1", "inst1", "batch-seed", Tags(6), DateTimeOffset.UtcNow));
        ExpectMsg<SubscribeTagsResponse>(m => m.Success, TimeSpan.FromSeconds(5));

        var reads = adapter.ReadBatches.ToList();
        Assert.Equal(2, reads.Count);
        Assert.Equal(6, reads.Sum(r => r.Count));
        Assert.Equal(0, Volatile.Read(ref adapter.SingleReadCalls));
        // The seeded value still reaches the instance actor after registration.
        ExpectMsg<TagValueUpdate>(u => u.Quality == QualityCode.Good, TimeSpan.FromSeconds(5));
    }

    [Fact]
    public void Seed_HangingReads_ReturnAtTheOverallDeadline_AndTheSubscribeIsStillAcked()
    {
        // A device whose reads never answer must not hold SubscribeCompleted (and therefore
        // the instance's ack) open: the whole seed is bounded by SeedOverallTimeout.
        var options = Options();
        options.SeedOverallTimeout = TimeSpan.FromMilliseconds(600);
        options.SeedReadTimeout = TimeSpan.FromSeconds(30); // deliberately longer than the deadline
        options.SeedReadMaxAttempts = 3;

        var adapter = new FakeBatchDataConnection { HangReads = true };
        var actor = CreateActor(adapter, options, "batch-seed-deadline");

        var started = DateTimeOffset.UtcNow;
        actor.Tell(new SubscribeTagsRequest("c1", "inst1", "batch-seed-deadline", Tags(3), DateTimeOffset.UtcNow));
        ExpectMsg<SubscribeTagsResponse>(m => m.Success, TimeSpan.FromSeconds(10));
        var elapsed = DateTimeOffset.UtcNow - started;

        // Bounded by the overall deadline, not by SeedReadTimeout (30s) or the retry budget.
        Assert.True(elapsed < TimeSpan.FromSeconds(5), $"seed took {elapsed.TotalSeconds:F1}s");
        // No value was seeded — the tags stay Uncertain until a change notification.
        ExpectNoMsg(TimeSpan.FromMilliseconds(200));
    }

    [Fact]
    public void PartialFailure_PerTagRow_MarksTagBad_ButTheRequestStillSucceeds()
    {
        // A per-tag failure row is a tag-resolution problem: Bad quality is pushed for that
        // tag and the subscribe is still acked Success (the surviving tags are live).
        var adapter = new FakeBatchDataConnection();
        adapter.FailingTags.Add("tag2");
        var actor = CreateActor(adapter, Options(), "batch-partial");

        actor.Tell(new SubscribeTagsRequest("c1", "inst1", "batch-partial", Tags(3), DateTimeOffset.UtcNow));

        ExpectMsg<TagValueUpdate>(u => u.TagPath == "tag2" && u.Quality == QualityCode.Bad, TimeSpan.FromSeconds(5));
        ExpectMsg<SubscribeTagsResponse>(m => m.Success, TimeSpan.FromSeconds(5));
    }

    [Fact]
    public void ThrownBatch_AtConnectionLevel_FailsTheRequestAndDrivesReconnect()
    {
        // A THROWN batch means the whole chunk failed at connection level: the response must
        // say so, and the actor must enter Reconnecting (which pushes bad quality for the
        // connection).
        var adapter = new FakeBatchDataConnection
        {
            BatchSubscribeThrows = () => new InvalidOperationException("client is not connected")
        };
        var actor = CreateActor(adapter, Options(), "batch-connection-fault");

        actor.Tell(new SubscribeTagsRequest("c1", "inst1", "batch-connection-fault", Tags(3), DateTimeOffset.UtcNow));

        ExpectMsg<SubscribeTagsResponse>(m => !m.Success && m.ErrorMessage != null, TimeSpan.FromSeconds(5));
        ExpectMsg<ConnectionQualityChanged>(q => q.Quality == QualityCode.Bad, TimeSpan.FromSeconds(5));
    }

    [Fact]
    public void ResolutionProbes_AreBatched_NotOnePerTag()
    {
        // Every unresolved tag is probed in ONE batch call per chunk. Pre-fix the retry tick
        // fired one SubscribeAsync task per unresolved tag, every tick, forever.
        var options = Options();
        options.SubscribeBatchSize = 10;
        var adapter = new FakeBatchDataConnection();
        foreach (var tag in Tags(6))
            adapter.FailingTags.Add(tag);

        var actor = CreateActor(adapter, options, "batch-probe");
        actor.Tell(new SubscribeTagsRequest("c1", "inst1", "batch-probe", Tags(6), DateTimeOffset.UtcNow));

        for (var i = 0; i < 6; i++)
            ExpectMsg<TagValueUpdate>(u => u.Quality == QualityCode.Bad, TimeSpan.FromSeconds(5));
        ExpectMsg<SubscribeTagsResponse>(TimeSpan.FromSeconds(5));

        var initialCalls = adapter.SubscribeBatches.Count;
        AwaitCondition(() => adapter.SubscribeBatches.Count > initialCalls, TimeSpan.FromSeconds(5));

        // The probe round carried all six unresolved tags in a single call.
        var probe = adapter.SubscribeBatches.Skip(initialCalls).First();
        Assert.Equal(6, probe.Count);
        Assert.Equal(0, Volatile.Read(ref adapter.SingleSubscribeCalls));
    }

    [Fact]
    public void ResolutionProbes_BackOff_SoADeadDeviceIsNotProbedAtFullRate()
    {
        // Floor 150ms, ceiling 600ms: a fixed-interval retry would fire ~13 rounds in 2s;
        // the backoff sequence (150, 300, 600, 600 …) fires far fewer.
        var options = Options();
        options.SubscribeBatchSize = 10;
        var adapter = new FakeBatchDataConnection();
        adapter.FailingTags.Add("tag1");

        var actor = CreateActor(adapter, options, "batch-backoff");
        actor.Tell(new SubscribeTagsRequest("c1", "inst1", "batch-backoff", ["tag1"], DateTimeOffset.UtcNow));
        ExpectMsg<TagValueUpdate>(u => u.Quality == QualityCode.Bad, TimeSpan.FromSeconds(5));
        ExpectMsg<SubscribeTagsResponse>(TimeSpan.FromSeconds(5));

        var initialCalls = adapter.SubscribeBatches.Count;
        Thread.Sleep(2000);
        var rounds = adapter.SubscribeBatches.Count - initialCalls;

        Assert.InRange(rounds, 2, 8);
    }

    [Fact]
    public void QualityCounters_ArePushedOnTransitionOnly_AndCoalesced()
    {
        // Repeated values at UNCHANGED quality move no counter and must produce no collector
        // push at all; a genuine transition produces exactly one push per coalescing window.
        var adapter = new FakeBatchDataConnection();
        var actor = CreateActor(adapter, Options(), "batch-quality");

        actor.Tell(new SubscribeTagsRequest("c1", "inst1", "batch-quality", ["tag1"], DateTimeOffset.UtcNow));
        ExpectMsg<SubscribeTagsResponse>(m => m.Success, TimeSpan.FromSeconds(5));
        AwaitCondition(() => adapter.ValueCallback != null, TimeSpan.FromSeconds(5));

        // Let the seed's own transition flush, then start counting.
        Thread.Sleep(400);
        _health.ClearReceivedCalls();

        for (var i = 0; i < 20; i++)
            adapter.ValueCallback!("tag1", new TagValue(i, QualityCode.Good, DateTimeOffset.UtcNow));

        Thread.Sleep(400);
        Assert.DoesNotContain(_health.ReceivedCalls(), c => c.GetMethodInfo().Name == "UpdateTagQuality");

        // One genuine transition → exactly one coalesced push.
        adapter.ValueCallback!("tag1", new TagValue(1, QualityCode.Bad, DateTimeOffset.UtcNow));
        AwaitCondition(
            () => _health.ReceivedCalls().Count(c => c.GetMethodInfo().Name == "UpdateTagQuality") == 1,
            TimeSpan.FromSeconds(3));
        Thread.Sleep(300);
        Assert.Single(_health.ReceivedCalls(), c => c.GetMethodInfo().Name == "UpdateTagQuality");
    }

    [Fact]
    public void QualityCounters_FlushImmediatelyOnDisconnect()
    {
        // "Immediate bad quality on disconnect" must never wait for the coalescing timer.
        var options = Options();
        options.QualityFlushInterval = TimeSpan.FromSeconds(30); // would mask a deferred push
        var adapter = new FakeBatchDataConnection();
        var actor = CreateActor(adapter, options, "batch-quality-disconnect");

        actor.Tell(new SubscribeTagsRequest("c1", "inst1", "batch-quality-disconnect", ["tag1"], DateTimeOffset.UtcNow));
        ExpectMsg<SubscribeTagsResponse>(m => m.Success, TimeSpan.FromSeconds(5));
        ExpectMsg<TagValueUpdate>(TimeSpan.FromSeconds(5)); // the seed
        _health.ClearReceivedCalls();

        adapter.RaiseDisconnected();

        ExpectMsg<ConnectionQualityChanged>(q => q.Quality == QualityCode.Bad, TimeSpan.FromSeconds(5));
        AwaitCondition(
            () => _health.ReceivedCalls().Any(c => c.GetMethodInfo().Name == "UpdateTagQuality"),
            TimeSpan.FromSeconds(3));
    }

    [Fact]
    public void Unsubscribe_ReleasesEveryHandleInOneRoundTrip()
    {
        var adapter = new FakeBatchDataConnection();
        var actor = CreateActor(adapter, Options(), "batch-unsubscribe");

        actor.Tell(new SubscribeTagsRequest("c1", "inst1", "batch-unsubscribe", Tags(6), DateTimeOffset.UtcNow));
        ExpectMsg<SubscribeTagsResponse>(m => m.Success, TimeSpan.FromSeconds(5));

        actor.Tell(new UnsubscribeTagsRequest("c2", "inst1", "batch-unsubscribe", DateTimeOffset.UtcNow));

        AwaitCondition(() => adapter.UnsubscribeBatches.Count == 1, TimeSpan.FromSeconds(5));
        Assert.True(adapter.UnsubscribeBatches.TryDequeue(out var released));
        Assert.Equal(6, released!.Count);
    }

    [Fact]
    public void Reconnect_ReSubscribesInPacedChunks_AndRestoresTheResolvedCount()
    {
        // The reconnect re-subscribe is chunked and paced (SubscribeBatchDelay between
        // chunks) rather than firing one task per tag, and the per-tag results repopulate
        // _subscriptionIds so a later unsubscribe still releases every adapter handle.
        var options = Options();
        options.SubscribeBatchSize = 4;
        options.SubscribeBatchDelay = TimeSpan.FromMilliseconds(120);

        var adapter = new FakeBatchDataConnection();
        var actor = CreateActor(adapter, options, "batch-resubscribe");

        actor.Tell(new SubscribeTagsRequest("c1", "inst1", "batch-resubscribe", Tags(8), DateTimeOffset.UtcNow));
        ExpectMsg<SubscribeTagsResponse>(m => m.Success, TimeSpan.FromSeconds(5));

        var beforeReconnect = adapter.SubscribeBatches.Count;
        adapter.RaiseDisconnected();

        // Two re-subscribe chunks of 4.
        AwaitCondition(() => adapter.SubscribeBatches.Count >= beforeReconnect + 2, TimeSpan.FromSeconds(10));
        var reconnectChunks = adapter.SubscribeBatches.Skip(beforeReconnect).Take(2).ToList();
        Assert.Equal([4, 4], reconnectChunks.Select(c => c.Count));

        // Paced: the second chunk is not issued back-to-back with the first.
        var times = adapter.SubscribeBatchTimes.Skip(beforeReconnect).Take(2).ToList();
        Assert.True(times[1] - times[0] >= TimeSpan.FromMilliseconds(100),
            $"chunks were {(times[1] - times[0]).TotalMilliseconds:F0}ms apart");

        // Every tag is registered again, so unsubscribe releases all 8 handles.
        actor.Tell(new UnsubscribeTagsRequest("c2", "inst1", "batch-resubscribe", DateTimeOffset.UtcNow));
        AwaitCondition(
            () => adapter.UnsubscribeBatches.Any(b => b.Count == 8),
            TimeSpan.FromSeconds(10));
    }

    [Fact]
    public void ResolutionProbe_CompletingAfterUnsubscribe_IsDiscarded_AndItsHandleIsReleased()
    {
        // A tag unsubscribed while its resolution probe is in flight must NOT have the
        // probe's result applied: HandleUnsubscribe already dropped it from
        // _unresolvedTags and from TotalSubscribedTags, so storing the handle would push
        // ResolvedTags above TotalSubscribedTags forever (and drive TotalSubscribedTags
        // negative on the next redeploy) while leaking the adapter monitored item, which
        // no later unsubscribe could ever reference.
        var options = Options();
        options.SubscribeBatchSize = 10;
        // Long enough that the probe cannot fire before the gate below is armed.
        options.TagResolutionRetryInterval = TimeSpan.FromMilliseconds(800);
        options.TagResolutionRetryMaxInterval = TimeSpan.FromSeconds(2);

        var adapter = new FakeBatchDataConnection();
        adapter.FailingTags.Add("tag1");
        var actor = CreateActor(adapter, options, "batch-probe-unsubscribe-race");

        actor.Tell(new SubscribeTagsRequest(
            "c1", "inst1", "batch-probe-unsubscribe-race", ["tag1"], DateTimeOffset.UtcNow));
        ExpectMsg<TagValueUpdate>(u => u.Quality == QualityCode.Bad, TimeSpan.FromSeconds(5));
        ExpectMsg<SubscribeTagsResponse>(TimeSpan.FromSeconds(5));

        var afterSubscribe = Health(actor);
        Assert.Equal(1, afterSubscribe.TotalSubscribedTags);
        Assert.Equal(0, afterSubscribe.ResolvedTags);

        // Park the next probe inside the adapter, and let it succeed when released.
        var gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
        adapter.SubscribeGate = gate.Task;
        adapter.FailingTags.Clear();

        var beforeProbe = adapter.SubscribeBatches.Count;
        AwaitCondition(() => adapter.SubscribeBatches.Count > beforeProbe, TimeSpan.FromSeconds(10));

        // Unsubscribe while the probe is still in flight, and confirm it has been applied
        // before the probe result lands (same-sender ordering makes the report a barrier).
        actor.Tell(new UnsubscribeTagsRequest("c2", "inst1", "batch-probe-unsubscribe-race", DateTimeOffset.UtcNow));
        var afterUnsubscribe = Health(actor);
        Assert.Equal(0, afterUnsubscribe.TotalSubscribedTags);
        Assert.Equal(0, afterUnsubscribe.ResolvedTags);
        // The tag was still unresolved, so the unsubscribe itself had no handle to
        // release — every id released from here on is the raced probe's. (This also
        // rules out a vacuous run in which the probe resolved before the gate was armed.)
        Assert.Empty(adapter.UnsubscribeBatches);

        gate.SetResult();

        // The orphaned handle is released back to the adapter…
        AwaitCondition(
            () => adapter.UnsubscribeBatches.Any(b => b.Count == 1),
            TimeSpan.FromSeconds(10));

        // …and the counters are untouched by the discarded result.
        AwaitAssert(() =>
        {
            var report = Health(actor);
            Assert.Equal(0, report.TotalSubscribedTags);
            Assert.Equal(0, report.ResolvedTags);
        }, TimeSpan.FromSeconds(5));

        // A redeploy round trip still books exactly one tag in and one tag out — pre-fix
        // the stale handle made this read Total=0/Resolved=1 and then Total=-1.
        adapter.SubscribeGate = null;
        actor.Tell(new SubscribeTagsRequest(
            "c3", "inst1", "batch-probe-unsubscribe-race", ["tag1"], DateTimeOffset.UtcNow));
        ExpectMsg<SubscribeTagsResponse>(m => m.Success, TimeSpan.FromSeconds(5));

        var afterRedeploy = Health(actor);
        Assert.Equal(1, afterRedeploy.TotalSubscribedTags);
        Assert.Equal(1, afterRedeploy.ResolvedTags);

        actor.Tell(new UnsubscribeTagsRequest("c4", "inst1", "batch-probe-unsubscribe-race", DateTimeOffset.UtcNow));
        var afterFinalUnsubscribe = Health(actor);
        Assert.Equal(0, afterFinalUnsubscribe.TotalSubscribedTags);
        Assert.Equal(0, afterFinalUnsubscribe.ResolvedTags);
    }

    [Fact]
    public void ReconnectResubscribe_CompletingAfterUnsubscribe_IsDiscarded_AndItsHandleIsReleased()
    {
        // Same race on the OTHER batch source: the reconnect re-subscribe. ReSubscribeAll
        // has already zeroed ResolvedTags, so applying the late result would report a
        // resolved tag for an instance that no longer exists and leak its handle.
        var options = Options();
        options.SubscribeBatchSize = 10;

        var adapter = new FakeBatchDataConnection();
        var actor = CreateActor(adapter, options, "batch-resubscribe-unsubscribe-race");

        actor.Tell(new SubscribeTagsRequest(
            "c1", "inst1", "batch-resubscribe-unsubscribe-race", ["tag1"], DateTimeOffset.UtcNow));
        ExpectMsg<SubscribeTagsResponse>(m => m.Success, TimeSpan.FromSeconds(5));

        var gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
        adapter.SubscribeGate = gate.Task;

        var beforeReconnect = adapter.SubscribeBatches.Count;
        adapter.RaiseDisconnected();
        AwaitCondition(() => adapter.SubscribeBatches.Count > beforeReconnect, TimeSpan.FromSeconds(10));

        actor.Tell(new UnsubscribeTagsRequest(
            "c2", "inst1", "batch-resubscribe-unsubscribe-race", DateTimeOffset.UtcNow));
        // The unsubscribe found no handle to release (ReSubscribeAll cleared
        // _subscriptionIds), so no release round trip has happened yet.
        Health(actor);
        Assert.Empty(adapter.UnsubscribeBatches);

        gate.SetResult();

        // The handle minted by the re-subscribe is released rather than leaked, and the
        // discarded row books no resolved tag. (TotalSubscribedTags across a reconnect
        // window is a separate, pre-existing accounting gap — ReSubscribeAll clears the
        // maps HandleUnsubscribe decrements from — so this pins the invariant that
        // matters here: ResolvedTags never exceeds it.)
        AwaitCondition(
            () => adapter.UnsubscribeBatches.Any(b => b.Count == 1),
            TimeSpan.FromSeconds(10));

        AwaitAssert(() =>
        {
            var report = Health(actor);
            Assert.Equal(0, report.ResolvedTags);
            Assert.True(report.ResolvedTags <= report.TotalSubscribedTags);
        }, TimeSpan.FromSeconds(5));
    }

    /// <summary>Point-in-time counters straight off the actor, fished past any pending pushes.</summary>
    private DataConnectionHealthReport Health(IActorRef actor)
    {
        actor.Tell(new DataConnectionActor.GetHealthReport());
        return FishForMessage<DataConnectionHealthReport>(_ => true, TimeSpan.FromSeconds(5));
    }
}