lmxopcua/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/AdminOperationsActorMailboxTests.cs

using Akka.Actor;
using Microsoft.EntityFrameworkCore;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Admin;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy;
using ZB.MOM.WW.OtOpcUa.Commons.Types;
using ZB.MOM.WW.OtOpcUa.Configuration;
using ZB.MOM.WW.OtOpcUa.ControlPlane.AdminOperations;
using ZB.MOM.WW.OtOpcUa.ControlPlane.Tests.Harness;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;

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

/// <summary>
///     R03:3.1 — the admin singleton's mailbox used to be suspended for the whole of every
///     <c>ReceiveAsync</c>: one Test Connect against a black-holed PLC (60 s clamp) or one large deploy
///     compose stalled every other admin operation fleet-wide, alarm acks included. Both now run off the
///     actor thread; only the "is one in flight" check stays serialised.
/// </summary>
public sealed class AdminOperationsActorMailboxTests : ControlPlaneActorTestBase
{
    private static readonly TimeSpan AckBudget = TimeSpan.FromSeconds(1);

    /// <summary>An alarm ack completes in under a second while a 60 s Test Connect is still pending.</summary>
    [Fact]
    public void Alarm_ack_completes_while_a_60s_test_connect_is_pending()
    {
        var dbFactory = NewInMemoryDbFactory();
        var coordinator = CreateTestProbe("coord");
        var probe = new BlockingProbe();
        var actor = Sys.ActorOf(AdminOperationsActor.Props(dbFactory, coordinator.Ref, new IDriverProbe[] { probe }));
        var connectClient = CreateTestProbe("connect-client");

        connectClient.Send(actor, new TestDriverConnect("Blocking", "{}", TimeoutSeconds: 60, Guid.NewGuid()));
        probe.Started.Wait(PresenceBudget).ShouldBeTrue("the probe must be in flight before the ack is sent");

        actor.Tell(new AcknowledgeAlarmCommand("alarm-1", "op", null, CorrelationId.NewId()));
        var ack = ExpectMsg<AcknowledgeAlarmResult>(AckBudget);
        ack.Ok.ShouldBeTrue();

        // Positive control: the probe reply still lands, with the right sender, once the probe returns.
        connectClient.ExpectNoMsg(TimeSpan.FromMilliseconds(200));
        probe.Release(new DriverProbeResult(true, null, TimeSpan.FromMilliseconds(5)));
        var result = connectClient.ExpectMsg<TestDriverConnectResult>(PresenceBudget);
        result.Ok.ShouldBeTrue();
        result.LatencyMs.ShouldNotBeNull();
    }

    /// <summary>A probe that times out still answers its own sender with the clamped timeout message.</summary>
    [Fact]
    public void Test_connect_timeout_is_reported_to_its_sender()
    {
        var dbFactory = NewInMemoryDbFactory();
        var coordinator = CreateTestProbe("coord");
        var probe = new BlockingProbe();   // never released: the CTS fires at the 1 s clamp floor
        var actor = Sys.ActorOf(AdminOperationsActor.Props(dbFactory, coordinator.Ref, new IDriverProbe[] { probe }));

        actor.Tell(new TestDriverConnect("Blocking", "{}", TimeoutSeconds: 1, Guid.NewGuid()));

        var result = ExpectMsg<TestDriverConnectResult>(PresenceBudget);
        result.Ok.ShouldBeFalse();
        result.Message.ShouldNotBeNull();
        result.Message.ShouldContain("timed out");
    }

    /// <summary>
    ///     The deploy compose (snapshot + validate + gzip + row write) runs off the actor: an alarm ack
    ///     completes while the compose is parked on a gated DB context, and the deploy still completes
    ///     (Accepted + dispatched) once the gate opens.
    /// </summary>
    [Fact]
    public void Alarm_ack_completes_while_a_deploy_compose_is_in_flight()
    {
        var gated = new GatedDbFactory(NewInMemoryDbFactory());
        var coordinator = CreateTestProbe("coord");
        var actor = Sys.ActorOf(AdminOperationsActor.Props(gated, coordinator.Ref, Enumerable.Empty<IDriverProbe>()));
        var deployer = CreateTestProbe("deployer");

        gated.Hold();
        deployer.Send(actor, new StartDeployment("joe", CorrelationId.NewId()));
        gated.Entered.Wait(PresenceBudget).ShouldBeTrue("the compose must be parked before the ack is sent");

        actor.Tell(new AcknowledgeAlarmCommand("alarm-2", "op", null, CorrelationId.NewId()));
        ExpectMsg<AcknowledgeAlarmResult>(AckBudget).Ok.ShouldBeTrue();

        // Positive control: the deploy completes once released.
        deployer.ExpectNoMsg(TimeSpan.FromMilliseconds(200));
        gated.Release();
        deployer.ExpectMsg<StartDeploymentResult>(PresenceBudget).Outcome.ShouldBe(StartDeploymentOutcome.Accepted);
        coordinator.ExpectMsg<DispatchDeployment>(PresenceBudget);
    }

    /// <summary>
    ///     Only the in-flight check is serialised — and it must still catch a second request that arrives
    ///     while the first compose has not yet written its row (the DB alone would let both through).
    /// </summary>
    [Fact]
    public void Second_StartDeployment_during_an_in_flight_compose_is_refused()
    {
        var gated = new GatedDbFactory(NewInMemoryDbFactory());
        var coordinator = CreateTestProbe("coord");
        var actor = Sys.ActorOf(AdminOperationsActor.Props(gated, coordinator.Ref, Enumerable.Empty<IDriverProbe>()));
        var first = CreateTestProbe("first");
        var second = CreateTestProbe("second");

        gated.Hold();
        first.Send(actor, new StartDeployment("joe", CorrelationId.NewId()));
        gated.Entered.Wait(PresenceBudget).ShouldBeTrue();

        second.Send(actor, new StartDeployment("kim", CorrelationId.NewId()));
        second.ExpectMsg<StartDeploymentResult>(AckBudget).Outcome.ShouldBe(StartDeploymentOutcome.AnotherDeploymentInFlight);

        gated.Release();
        first.ExpectMsg<StartDeploymentResult>(PresenceBudget).Outcome.ShouldBe(StartDeploymentOutcome.Accepted);
        coordinator.ExpectMsg<DispatchDeployment>(PresenceBudget);
    }

    private sealed class BlockingProbe : IDriverProbe
    {
        private readonly TaskCompletionSource<DriverProbeResult> _tcs =
            new(TaskCreationOptions.RunContinuationsAsynchronously);

        public ManualResetEventSlim Started { get; } = new();
        public string DriverType => "Blocking";

        public async Task<DriverProbeResult> ProbeAsync(string configJson, TimeSpan timeout, CancellationToken ct)
        {
            Started.Set();
            return await _tcs.Task.WaitAsync(ct);
        }

        public void Release(DriverProbeResult result) => _tcs.TrySetResult(result);
    }

    /// <summary>
    ///     Passes synchronous context creation through (the serialised in-flight check) and parks the
    ///     asynchronous one (the off-actor compose) on a gate while <see cref="Hold"/> is in effect.
    /// </summary>
    private sealed class GatedDbFactory(IDbContextFactory<OtOpcUaConfigDbContext> inner)
        : IDbContextFactory<OtOpcUaConfigDbContext>
    {
        private volatile TaskCompletionSource? _gate;

        public ManualResetEventSlim Entered { get; } = new();

        public void Hold() => _gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
        public void Release() => _gate?.TrySetResult();

        public OtOpcUaConfigDbContext CreateDbContext() => inner.CreateDbContext();

        public async Task<OtOpcUaConfigDbContext> CreateDbContextAsync(CancellationToken cancellationToken = default)
        {
            Entered.Set();
            if (_gate is { } gate) await gate.Task.WaitAsync(cancellationToken);
            return inner.CreateDbContext();
        }
    }
}