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

using System.Text.Json;
using Akka.Actor;
using Microsoft.EntityFrameworkCore;
using Shouldly;
using Xunit;
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.Configuration.Entities;
using ZB.MOM.WW.OtOpcUa.Configuration.Enums;
using ZB.MOM.WW.OtOpcUa.Runtime.DeploymentCache;
using ZB.MOM.WW.OtOpcUa.Runtime.Drivers;
using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;

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

/// <summary>
///     LocalDb Phase 1 (Task 7) — the write half of the deployment-artifact cache.
/// </summary>
/// <remarks>
///     What matters here is not that a store happens, but the conditions under which it must NOT:
///     a failed apply, and — less obviously — a "successful" apply whose artifact never actually
///     loaded. See <c>ReconcileDrivers</c>: it swallows its own DB failures, so apply success does
///     not imply a real configuration was applied.
/// </remarks>
public sealed class DriverHostActorArtifactCacheTests : RuntimeActorTestBase
{
    private static readonly NodeId TestNode = NodeId.Parse("driver-test");
    private static readonly RevisionHash RevA = RevisionHash.Parse(new string('a', 64));

    [Fact]
    public void SuccessfulApply_StoresTheArtifactInTheCache()
    {
        var db = NewInMemoryDbFactory();
        var cache = new RecordingArtifactCache();
        var (deploymentId, artifact) = SeedDeployment(db, RevA, clusterId: null);

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

        actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId()));
        coordinator.ExpectMsg<ApplyAck>(TimeSpan.FromSeconds(5)).Outcome.ShouldBe(ApplyAckOutcome.Applied);

        AwaitAssert(() => cache.Stores.Count.ShouldBe(1), duration: TimeSpan.FromSeconds(3));

        var store = cache.Stores[0];
        store.DeploymentId.ShouldBe(deploymentId.ToString());
        store.RevisionHash.ShouldBe(RevA.Value);
        store.Artifact.ShouldBe(artifact);
    }

    [Fact]
    public void ApplyOfAClusterScopedArtifact_StoresUnderThatClusterId()
    {
        // The ClusterId is not config and not on IClusterRoleInfo — it is carried in the artifact's
        // Nodes[] rows. Getting this wrong keys the pair's two nodes to different cache rows, and
        // they would silently stop sharing a cache entry.
        var db = NewInMemoryDbFactory();
        var cache = new RecordingArtifactCache();
        var (deploymentId, _) = SeedDeployment(db, RevA, clusterId: "SITE-A");

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

        actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId()));
        coordinator.ExpectMsg<ApplyAck>(TimeSpan.FromSeconds(5));

        AwaitAssert(() => cache.Stores.Count.ShouldBe(1), duration: TimeSpan.FromSeconds(3));
        cache.Stores[0].ClusterId.ShouldBe("SITE-A");
    }

    [Fact]
    public void UnscopedArtifact_StoresUnderTheSingleClusterSentinel()
    {
        var db = NewInMemoryDbFactory();
        var cache = new RecordingArtifactCache();
        var (deploymentId, _) = SeedDeployment(db, RevA, clusterId: null);

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

        actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId()));
        coordinator.ExpectMsg<ApplyAck>(TimeSpan.FromSeconds(5));

        AwaitAssert(() => cache.Stores.Count.ShouldBe(1), duration: TimeSpan.FromSeconds(3));
        cache.Stores[0].ClusterId.ShouldBe("__single");
    }

    [Fact]
    public void EmptyArtifact_IsNotCached()
    {
        // THE case that makes this cache safe to boot from. ReconcileDrivers catches its own DB
        // failures, logs a warning and returns without rethrowing — so an apply can reach its
        // success path, ACK Applied, and have started zero drivers. Persisting that as
        // last-known-good would boot the node into an empty address space during the next outage:
        // strictly worse than having no cache at all. An empty blob is the observable form of it.
        var db = NewInMemoryDbFactory();
        var cache = new RecordingArtifactCache();

        var deploymentId = DeploymentId.NewId();
        using (var ctx = db.CreateDbContext())
        {
            ctx.Deployments.Add(new Deployment
            {
                DeploymentId = deploymentId.Value,
                RevisionHash = RevA.Value,
                Status = DeploymentStatus.Sealed,
                CreatedBy = "test",
                SealedAtUtc = DateTime.UtcNow,
                ArtifactBlob = Array.Empty<byte>(),
            });
            ctx.SaveChanges();
        }

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

        actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId()));

        // Since #486 the apply FAILS rather than reporting a success it did not achieve: empty bytes are
        // what an unreadable artifact looks like, not a legitimate no-op deployment. (A configuration that
        // genuinely declares nothing still serialises to a JSON document with empty arrays.)
        coordinator.ExpectMsg<ApplyAck>(TimeSpan.FromSeconds(5)).Outcome.ShouldBe(ApplyAckOutcome.Failed);

        // The point of this test is unchanged, and now holds for two independent reasons: nothing is cached.
        Thread.Sleep(500);
        cache.Stores.ShouldBeEmpty();
    }

    [Fact]
    public void AThrowingCache_DoesNotFailTheApply()
    {
        // By the time the cache runs, an Applied ACK has already gone to the coordinator. If a cache
        // failure escaped, ApplyAndAck's catch would send a SECOND, contradictory Failed ACK for a
        // deployment the fleet already believes is live.
        var db = NewInMemoryDbFactory();
        var (deploymentId, _) = SeedDeployment(db, RevA, clusterId: null);

        var coordinator = CreateTestProbe();
        var actor = Sys.ActorOf(DriverHostActor.Props(
            db, TestNode, coordinator.Ref,
            localRoles: new HashSet<string> { "driver" },
            deploymentArtifactCache: new ThrowingArtifactCache()));

        actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId()));

        coordinator.ExpectMsg<ApplyAck>(TimeSpan.FromSeconds(5)).Outcome.ShouldBe(ApplyAckOutcome.Applied);

        // Exactly one ACK, and it says Applied. A second Failed ACK arriving here would mean the
        // cache failure had corrupted the deployment protocol.
        coordinator.ExpectNoMsg(TimeSpan.FromMilliseconds(500));
    }

    [Fact]
    public void NoCacheConfigured_AppliesNormally()
    {
        // Admin-only graphs and most test harnesses pass null. Caching is skipped, nothing else changes.
        var db = NewInMemoryDbFactory();
        var (deploymentId, _) = SeedDeployment(db, RevA, clusterId: null);

        var coordinator = CreateTestProbe();
        var actor = Sys.ActorOf(DriverHostActor.Props(
            db, TestNode, coordinator.Ref,
            localRoles: new HashSet<string> { "driver" }));

        actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId()));

        coordinator.ExpectMsg<ApplyAck>(TimeSpan.FromSeconds(5)).Outcome.ShouldBe(ApplyAckOutcome.Applied);
    }

    /// <summary>
    ///     Seeds a sealed deployment whose artifact optionally scopes <see cref="TestNode"/> to a
    ///     cluster, and returns the exact bytes stored so a round-trip can be asserted.
    /// </summary>
    private static (DeploymentId Id, byte[] Artifact) SeedDeployment(
        IDbContextFactory<OtOpcUaConfigDbContext> db, RevisionHash rev, string? clusterId)
    {
        object payload = clusterId is null
            ? new
            {
                DriverInstances = Array.Empty<object>(),
            }
            : new
            {
                DriverInstances = Array.Empty<object>(),
                // ResolveClusterScope only filters when the artifact declares MORE than one cluster;
                // a single-cluster artifact is treated as unscoped.
                Clusters = new object[]
                {
                    new { ClusterId = clusterId },
                    new { ClusterId = "OTHER" },
                },
                Nodes = new object[]
                {
                    new { NodeId = TestNode.Value, ClusterId = clusterId },
                },
            };

        var artifact = JsonSerializer.SerializeToUtf8Bytes(payload);

        var id = DeploymentId.NewId();
        using var ctx = db.CreateDbContext();
        ctx.Deployments.Add(new Deployment
        {
            DeploymentId = id.Value,
            RevisionHash = rev.Value,
            Status = DeploymentStatus.Sealed,
            CreatedBy = "test",
            SealedAtUtc = DateTime.UtcNow,
            ArtifactBlob = artifact,
        });
        ctx.SaveChanges();
        return (id, artifact);
    }

    /// <summary>Records every store so the test can assert on what was cached.</summary>
    private sealed class RecordingArtifactCache : IDeploymentArtifactCache
    {
        private readonly Lock _gate = new();
        private readonly List<StoreCall> _stores = [];

        public IReadOnlyList<StoreCall> Stores
        {
            get { lock (_gate) { return _stores.ToArray(); } }
        }

        public Task StoreAsync(string clusterId, string deploymentId, string revisionHash,
                               byte[] artifact, CancellationToken ct = default)
        {
            lock (_gate)
            {
                _stores.Add(new StoreCall(clusterId, deploymentId, revisionHash, artifact));
            }

            return Task.CompletedTask;
        }

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

        public Task<CachedDeploymentArtifact?> GetCurrentUnkeyedAsync(CancellationToken ct = default)
            => Task.FromResult<CachedDeploymentArtifact?>(null);

        internal sealed record StoreCall(
            string ClusterId, string DeploymentId, string RevisionHash, byte[] Artifact);
    }

    /// <summary>Fails every store, to prove a cache fault cannot reach the deployment protocol.</summary>
    private sealed class ThrowingArtifactCache : IDeploymentArtifactCache
    {
        public Task StoreAsync(string clusterId, string deploymentId, string revisionHash,
                               byte[] artifact, CancellationToken ct = default)
            => throw new InvalidOperationException("disk full");

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

        public Task<CachedDeploymentArtifact?> GetCurrentUnkeyedAsync(CancellationToken ct = default)
            => Task.FromResult<CachedDeploymentArtifact?>(null);
    }
}