lmxopcua/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorFetchAndCacheBootstrapTests.cs

using System.Text.Json;
using Akka.Actor;
using Microsoft.EntityFrameworkCore;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.Interfaces;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Fleet;
using ZB.MOM.WW.OtOpcUa.Commons.Types;
using ZB.MOM.WW.OtOpcUa.Configuration;
using ZB.MOM.WW.OtOpcUa.Runtime.DeploymentCache;
using ZB.MOM.WW.OtOpcUa.Runtime.Drivers;
using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa;
using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;

namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers;

/// <summary>
///     Per-cluster mesh Phase 3 (Task 6) — <c>FetchAndCache</c> bootstrap. A driver node in this mode
///     boots its served state from the LocalDb pointer, reading NO config from central SQL; an empty
///     or unreadable cache lands Steady-with-no-revision (the first dispatch fetches), NOT Stale.
/// </summary>
/// <remarks>
///     Every test injects a <see cref="ThrowingDbFactory"/> for central SQL — reaching Steady (and, in
///     two of them, applying a dispatch) proves the boot never touched it. This is the FetchAndCache
///     counterpart to <see cref="DriverHostActorBootFromCacheTests"/> (which covers the Direct-mode
///     SQL-unreachable fallback, where <c>RunningFromCache</c> is TRUE — here it is normal operation,
///     so it stays FALSE).
/// </remarks>
public sealed class DriverHostActorFetchAndCacheBootstrapTests : RuntimeActorTestBase
{
    private static readonly NodeId TestNode = NodeId.Parse("driver-test");
    private static readonly RevisionHash CachedRev = RevisionHash.Parse(new string('c', 64));

    private static byte[] Artifact() =>
        JsonSerializer.SerializeToUtf8Bytes(new { DriverInstances = Array.Empty<object>() });

    [Fact]
    public void WithCachedArtifact_RestoresServedState_WithoutReadingCentralSql()
    {
        var blob = Artifact();
        var cached = new CachedDeploymentArtifact(
            DeploymentId.NewId().ToString(), CachedRev.Value, blob, DateTimeOffset.UtcNow.AddHours(-1));
        var cache = new StubArtifactCache(cached);
        var publishProbe = this.CreateCleanOutcomePublishProbe();

        var actor = Sys.ActorOf(DriverHostActor.Props(
            new ThrowingDbFactory(), TestNode, ackRouter: null,
            localRoles: new HashSet<string> { "driver" },
            opcUaPublishActor: publishProbe.Ref,
            deploymentArtifactCache: cache,
            fetchAndCacheMode: true,
            artifactFetcher: new RecordingFetcher(Artifact())));

        // Served state rebuilt from the CACHED bytes (no ConfigDb read — the factory throws).
        var rebuild = publishProbe.ExpectMsg<OpcUaPublishActor.RebuildAddressSpace>(TimeSpan.FromSeconds(5));
        rebuild.Artifact.ShouldBe(blob);

        var snapshot = AskDiagnostics(actor);
        snapshot.CurrentRevision.ShouldBe(CachedRev);
        // NORMAL operation, not the degraded Direct-mode fallback — a FetchAndCache node can still fetch.
        snapshot.RunningFromCache.ShouldBeFalse();
        cache.UnkeyedReads.ShouldBe(1);
    }

    [Fact]
    public void WithEmptyCache_EntersSteadyNoRevision_AndTheNextDispatchFetches()
    {
        var cache = new StubArtifactCache(current: null);
        var fetcher = new RecordingFetcher(Artifact());
        var coordinator = CreateTestProbe();

        var actor = Sys.ActorOf(DriverHostActor.Props(
            new ThrowingDbFactory(), TestNode, coordinator.Ref,
            localRoles: new HashSet<string> { "driver" },
            deploymentArtifactCache: cache,
            fetchAndCacheMode: true,
            artifactFetcher: fetcher));

        var snapshot = AskDiagnostics(actor);
        snapshot.CurrentRevision.ShouldBeNull();
        snapshot.RunningFromCache.ShouldBeFalse();

        // Steady, not Stale: a dispatch is serviced (Stale would ignore it, never acking) and fetches.
        actor.Tell(new DispatchDeployment(DeploymentId.NewId(), CachedRev, CorrelationId.NewId()));
        coordinator.ExpectMsg<ApplyAck>(TimeSpan.FromSeconds(5)).Outcome.ShouldBe(ApplyAckOutcome.Applied);
        fetcher.Calls.ShouldBe(1);
    }

    [Fact]
    public void WithAnUnreadableCache_EntersSteadyNoRevision_NotStale()
    {
        var fetcher = new RecordingFetcher(Artifact());
        var coordinator = CreateTestProbe();

        var actor = Sys.ActorOf(DriverHostActor.Props(
            new ThrowingDbFactory(), TestNode, coordinator.Ref,
            localRoles: new HashSet<string> { "driver" },
            deploymentArtifactCache: new ThrowingReadCache(),
            fetchAndCacheMode: true,
            artifactFetcher: fetcher));

        var snapshot = AskDiagnostics(actor);
        snapshot.CurrentRevision.ShouldBeNull();
        snapshot.RunningFromCache.ShouldBeFalse();

        // A cache-read fault is NOT Stale (Stale means "central SQL down, retry it" — there is no
        // config SQL read to recover here). The node is Steady and the next dispatch fetches.
        actor.Tell(new DispatchDeployment(DeploymentId.NewId(), CachedRev, CorrelationId.NewId()));
        coordinator.ExpectMsg<ApplyAck>(TimeSpan.FromSeconds(5)).Outcome.ShouldBe(ApplyAckOutcome.Applied);
        fetcher.Calls.ShouldBe(1);
    }

    /// <summary>
    ///     #601 / review R1 m4 — a cache row whose bytes are malformed applied nothing, so its revision must not
    ///     become current: the boot lands Steady-with-no-revision (as for an empty cache), and a re-dispatch of
    ///     the SAME revision really applies (fetches) instead of being acked by the "already at this rev"
    ///     short-circuit. <see cref="WithCachedArtifact_RestoresServedState_WithoutReadingCentralSql"/> is the
    ///     control: valid bytes do adopt the revision.
    /// </summary>
    [Fact]
    public void WithMalformedCachedBytes_TheRevisionIsNotAdopted_AndARedispatchOfItActuallyApplies()
    {
        var cached = new CachedDeploymentArtifact(
            DeploymentId.NewId().ToString(), CachedRev.Value,
            System.Text.Encoding.UTF8.GetBytes("{\"DriverInstances\": ["), DateTimeOffset.UtcNow.AddHours(-1));
        var fetcher = new RecordingFetcher(Artifact());
        var coordinator = CreateTestProbe();

        var actor = Sys.ActorOf(DriverHostActor.Props(
            new ThrowingDbFactory(), TestNode, coordinator.Ref,
            localRoles: new HashSet<string> { "driver" },
            deploymentArtifactCache: new StubArtifactCache(cached),
            fetchAndCacheMode: true,
            artifactFetcher: fetcher));

        var snapshot = AskDiagnostics(actor);
        snapshot.CurrentRevision.ShouldBeNull();
        snapshot.RunningFromCache.ShouldBeFalse();

        actor.Tell(new DispatchDeployment(DeploymentId.Parse(cached.DeploymentId), CachedRev, CorrelationId.NewId()));
        coordinator.ExpectMsg<ApplyAck>(TimeSpan.FromSeconds(5)).Outcome.ShouldBe(ApplyAckOutcome.Applied);
        fetcher.Calls.ShouldBe(1);
        AskDiagnostics(actor).CurrentRevision.ShouldBe(CachedRev);
    }

    private NodeDiagnosticsSnapshot AskDiagnostics(IActorRef actor)
    {
        var probe = CreateTestProbe();
        AwaitAssert(
            () =>
            {
                actor.Tell(new GetDiagnostics(CorrelationId.NewId()), probe.Ref);
                probe.ExpectMsg<NodeDiagnosticsSnapshot>(TimeSpan.FromSeconds(2));
            },
            duration: TimeSpan.FromSeconds(5));

        actor.Tell(new GetDiagnostics(CorrelationId.NewId()), probe.Ref);
        return probe.ExpectMsg<NodeDiagnosticsSnapshot>(TimeSpan.FromSeconds(5));
    }

    private sealed class RecordingFetcher(byte[]? bytes) : IDeploymentArtifactFetcher
    {
        private int _calls;
        public int Calls => Volatile.Read(ref _calls);

        public Task<byte[]?> FetchAsync(string deploymentId, string expectedRevisionHash, CancellationToken ct)
        {
            Interlocked.Increment(ref _calls);
            return Task.FromResult(bytes);
        }
    }

    /// <summary>Stands in for an unreachable central SQL Server.</summary>
    private sealed class ThrowingDbFactory : IDbContextFactory<OtOpcUaConfigDbContext>
    {
        public OtOpcUaConfigDbContext CreateDbContext()
            => throw new InvalidOperationException("ConfigDb unreachable");
    }

    private sealed class StubArtifactCache(CachedDeploymentArtifact? current) : IDeploymentArtifactCache
    {
        private int _unkeyedReads;
        public int UnkeyedReads => Volatile.Read(ref _unkeyedReads);

        public Task StoreAsync(string clusterId, string deploymentId, string revisionHash,
                               byte[] artifact, CancellationToken ct = default)
            => Task.CompletedTask;

        public Task<CachedDeploymentArtifact?> GetCurrentAsync(string clusterId, CancellationToken ct = default)
            => Task.FromResult(current);

        public Task<CachedDeploymentArtifact?> GetCurrentUnkeyedAsync(CancellationToken ct = default)
        {
            Interlocked.Increment(ref _unkeyedReads);
            return Task.FromResult(current);
        }
    }

    private sealed class ThrowingReadCache : IDeploymentArtifactCache
    {
        public Task StoreAsync(string clusterId, string deploymentId, string revisionHash,
                               byte[] artifact, CancellationToken ct = default)
            => Task.CompletedTask;

        public Task<CachedDeploymentArtifact?> GetCurrentAsync(string clusterId, CancellationToken ct = default)
            => throw new InvalidOperationException("cache unreadable");

        public Task<CachedDeploymentArtifact?> GetCurrentUnkeyedAsync(CancellationToken ct = default)
            => throw new InvalidOperationException("cache unreadable");
    }
}