lmxopcua/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorApplyFailureTests.cs

using System.Diagnostics.Metrics;
using System.Text.Json;
using Akka.Actor;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Logging.Abstractions;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.Observability;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
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.OpcUaServer;
using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa;
using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;

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

/// <summary>
///     archreview 01/S-1: when the applier reports a degraded apply (a rebuild that threw, or per-node
///     materialise failures), <see cref="OpcUaPublishActor.HandleRebuild"/> must surface it — increment the
///     <c>otopcua.opcua.apply.failed</c> meter and log at Error — instead of the optimistic Info line. The
///     happy path must stay Info-only (zero increments).
/// </summary>
public sealed class OpcUaPublishActorApplyFailureTests : RuntimeActorTestBase
{
    /// <summary>A rebuild whose sink throws increments otopcua.opcua.apply.failed (kind=rebuild). R2-07 — a
    /// pure-add no longer rebuilds, so the degraded-rebuild path is exercised via a rebuild-forcing change:
    /// a first PureAdd deploy establishes the applied composition, then a rename (ChangedEquipment ⇒ Rebuild)
    /// drives the throwing SafeRebuild.</summary>
    [Fact]
    public void Rebuild_when_sink_throws_increments_apply_failed_meter()
    {
        using var recorder = new MeterRecorder("otopcua.opcua.apply.failed");
        var db = NewInMemoryDbFactory();
        var sink = new ThrowOnRebuildSink();
        var applier = new AddressSpaceApplier(sink, NullLogger<AddressSpaceApplier>.Instance);
        var dep1 = SeedEquipmentDeployment(db, ("eq-1", "eq-1"));

        var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink, dbFactory: db, applier: applier));
        // First deploy: PureAdd — no rebuild, so the throwing sink is not hit yet.
        actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId(), new DeploymentId(dep1)));
        Thread.Sleep(200);
        recorder.Total.ShouldBe(0);

        // Second deploy: eq-1 RENAMED ⇒ ChangedEquipment ⇒ Rebuild ⇒ the sink's RebuildAddressSpace throws.
        var dep2 = SeedEquipmentDeployment(db, ("eq-1", "eq-1-renamed"));
        actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId(), new DeploymentId(dep2)));

        // A PRESENCE budget, so it is an upper bound before giving up rather than a wait: the helper polls
        // and returns the instant the meter fires, so a generous value costs nothing on the happy path and
        // can never make a genuinely failing assertion pass. Two seconds covered an idle machine but not a
        // fully parallel assembly run, where this deploy→rebuild→throw chain (two DB-backed applies) has to
        // share 14 cores with 40-odd other Akka test classes; it failed there roughly 1 run in 30.
        AwaitAssert(() =>
        {
            recorder.Total.ShouldBeGreaterThanOrEqualTo(1);
            recorder.WithTag("kind", "rebuild").ShouldBeGreaterThanOrEqualTo(1);
        }, duration: TimeSpan.FromSeconds(15));
    }

    /// <summary>A clean rebuild does NOT increment the apply-failed meter (Info-only happy path).</summary>
    [Fact]
    public void Rebuild_happy_path_does_not_increment_apply_failed_meter()
    {
        using var recorder = new MeterRecorder("otopcua.opcua.apply.failed");
        var db = NewInMemoryDbFactory();
        var sink = new NoopSink();
        var applier = new AddressSpaceApplier(sink, NullLogger<AddressSpaceApplier>.Instance);
        SeedEquipmentDeployment(db, ("eq-1", "eq-1"));

        var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(sink: sink, dbFactory: db, applier: applier));
        actor.Tell(new OpcUaPublishActor.RebuildAddressSpace(CorrelationId.NewId()));

        // Give the actor time to process, then confirm the failed meter never fired.
        Thread.Sleep(400);
        recorder.Total.ShouldBe(0);
    }

    private void AwaitAssert(Action assertion, TimeSpan duration)
    {
        var deadline = DateTime.UtcNow + duration;
        Exception? last = null;
        while (DateTime.UtcNow < deadline)
        {
            try { assertion(); return; }
            catch (Exception ex) { last = ex; Thread.Sleep(25); }
        }
        if (last is not null) throw last;
    }

    private static Guid SeedEquipmentDeployment(IDbContextFactory<OtOpcUaConfigDbContext> dbFactory, params (string Id, string Name)[] equipment)
    {
        var artifact = JsonSerializer.SerializeToUtf8Bytes(new
        {
            Equipment = equipment.Select(e => new
            {
                EquipmentId = e.Id,
                MachineCode = e.Id.ToUpperInvariant(),
                UnsLineId = "line-1",
                Name = e.Name,
            }).ToArray(),
            DriverInstances = Array.Empty<object>(),
            ScriptedAlarms = Array.Empty<object>(),
        });

        var id = Guid.NewGuid();
        using var ctx = dbFactory.CreateDbContext();
        ctx.Deployments.Add(new Deployment
        {
            DeploymentId = id,
            RevisionHash = new string('a', 64),
            Status = DeploymentStatus.Sealed,
            CreatedBy = "test",
            SealedAtUtc = DateTime.UtcNow,
            ArtifactBlob = artifact,
        });
        ctx.SaveChanges();
        return id;
    }

    /// <summary>A sink whose RebuildAddressSpace throws (drives the applier's SafeRebuild catch → RebuildFailed).</summary>
    private sealed class ThrowOnRebuildSink : IOpcUaAddressSpaceSink
    {
        public void WriteValue(string nodeId, object? value, OpcUaQuality quality, DateTime sourceTimestampUtc, AddressSpaceRealm realm = AddressSpaceRealm.Uns) { }
        public void WriteAlarmQuality(string alarmNodeId, OpcUaQuality quality, DateTime sourceTimestampUtc, AddressSpaceRealm realm) { }
        public void WriteAlarmCondition(string alarmNodeId, AlarmConditionSnapshot state, DateTime occurredUtc, AddressSpaceRealm realm = AddressSpaceRealm.Uns) { }
        public void MaterialiseAlarmCondition(string alarmNodeId, string equipmentNodeId, string displayName, string alarmType, int severity, AddressSpaceRealm realm, bool isNative = false) { }
        public void EnsureFolder(string folderNodeId, string? parentNodeId, string displayName, AddressSpaceRealm realm = AddressSpaceRealm.Uns) { }
        public void EnsureVariable(string variableNodeId, string? parentFolderNodeId, string displayName, string dataType, bool writable, AddressSpaceRealm realm, string? historianTagname = null, bool isArray = false, uint? arrayLength = null) { }
        public void RebuildAddressSpace() => throw new InvalidOperationException("simulated rebuild fault");
        public void RaiseNodesAddedModelChange(string affectedNodeId, AddressSpaceRealm realm = AddressSpaceRealm.Uns) { }
        public void WireAlarmNotifiers(string alarmNodeId, AddressSpaceRealm alarmRealm, IReadOnlyList<string> notifierFolderNodeIds, AddressSpaceRealm notifierFolderRealm) { }
        public void AddReference(string sourceNodeId, AddressSpaceRealm sourceRealm, string targetNodeId, AddressSpaceRealm targetRealm, string referenceType = "Organizes") { }
    }

    /// <summary>A no-op sink — a clean apply (no failures).</summary>
    private sealed class NoopSink : IOpcUaAddressSpaceSink
    {
        public void WriteValue(string nodeId, object? value, OpcUaQuality quality, DateTime sourceTimestampUtc, AddressSpaceRealm realm = AddressSpaceRealm.Uns) { }
        public void WriteAlarmQuality(string alarmNodeId, OpcUaQuality quality, DateTime sourceTimestampUtc, AddressSpaceRealm realm) { }
        public void WriteAlarmCondition(string alarmNodeId, AlarmConditionSnapshot state, DateTime occurredUtc, AddressSpaceRealm realm = AddressSpaceRealm.Uns) { }
        public void MaterialiseAlarmCondition(string alarmNodeId, string equipmentNodeId, string displayName, string alarmType, int severity, AddressSpaceRealm realm, bool isNative = false) { }
        public void EnsureFolder(string folderNodeId, string? parentNodeId, string displayName, AddressSpaceRealm realm = AddressSpaceRealm.Uns) { }
        public void EnsureVariable(string variableNodeId, string? parentFolderNodeId, string displayName, string dataType, bool writable, AddressSpaceRealm realm, string? historianTagname = null, bool isArray = false, uint? arrayLength = null) { }
        public void RebuildAddressSpace() { }
        public void RaiseNodesAddedModelChange(string affectedNodeId, AddressSpaceRealm realm = AddressSpaceRealm.Uns) { }
        public void WireAlarmNotifiers(string alarmNodeId, AddressSpaceRealm alarmRealm, IReadOnlyList<string> notifierFolderNodeIds, AddressSpaceRealm notifierFolderRealm) { }
        public void AddReference(string sourceNodeId, AddressSpaceRealm sourceRealm, string targetNodeId, AddressSpaceRealm targetRealm, string referenceType = "Organizes") { }
    }

    /// <summary>Listens to a single instrument by name on the central meter and tallies value + tags.</summary>
    private sealed class MeterRecorder : IDisposable
    {
        private readonly string _name;
        private readonly MeterListener _listener;
        private long _total;
        private readonly List<KeyValuePair<string, object?>[]> _tagSets = new();
        private readonly object _gate = new();

        public MeterRecorder(string instrumentName)
        {
            _name = instrumentName;
            _listener = new MeterListener
            {
                InstrumentPublished = (instrument, listener) =>
                {
                    if (instrument.Meter.Name == OtOpcUaTelemetry.MeterName && instrument.Name == _name)
                        listener.EnableMeasurementEvents(instrument);
                }
            };
            _listener.SetMeasurementEventCallback<long>((_, value, tags, _) =>
            {
                lock (_gate)
                {
                    _total += value;
                    _tagSets.Add(tags.ToArray());
                }
            });
            _listener.Start();
        }

        public long Total { get { lock (_gate) return _total; } }

        public int WithTag(string key, string value)
        {
            lock (_gate) return _tagSets.Count(set => set.Any(t => t.Key == key && Equals(t.Value, value)));
        }

        public void Dispose() => _listener.Dispose();
    }
}