lmxopcua/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs

using Akka.Actor;
using Akka.Cluster;
using Akka.Cluster.Tools.PublishSubscribe;
using Akka.Event;
using Microsoft.EntityFrameworkCore;
using ZB.MOM.WW.OtOpcUa.Cluster.Redundancy;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Redundancy;
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.OpcUaServer;
using ZB.MOM.WW.OtOpcUa.Runtime.Drivers;
using ZB.MOM.WW.OtOpcUa.Runtime.Health;

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

/// <summary>
///     Single-threaded bridge between Akka messages and the OPC UA SDK address space. Hosted on
///     the pinned <c>opcua-synchronized-dispatcher</c> (HOCON) so the OPC UA SDK sees
///     only one thread per actor instance — its session/subscription locks expect strict
///     single-threaded access.
///
///     Address-space writes route through <see cref="IOpcUaAddressSpaceSink"/>; ServiceLevel
///     writes route through <see cref="IServiceLevelPublisher"/>. Production binds SDK-backed
///     implementations; dev/Mac/tests bind the Null* defaults so the actor stays decoupled from
///     <c>Opc.Ua.Server</c>. The remaining piece is wiring those bindings to a real
///     <c>StandardServer</c> address space — tracked as F10b.
/// </summary>
public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
{
    public const string DispatcherId = "opcua-synchronized-dispatcher";
    public const string RedundancyStateTopic = "redundancy-state";

    /// <summary>Gets or sets the timer scheduler for the periodic DB-health refresh tick.</summary>
    public ITimerScheduler Timers { get; set; } = null!;

    /// <summary>Self-tick that drives the periodic Ask of the local <see cref="DbHealthProbeActor"/>
    /// for its cached status. Private singleton — the actor pumps this for itself, no external sender.</summary>
    private sealed class HealthTick { public static readonly HealthTick Instance = new(); private HealthTick() { } }

    /// <summary>A driver/vtag value update for a single materialised Variable node. v3 Batch 4: carries the
    /// node's <see cref="AddressSpaceRealm"/> so the sink resolves the right namespace — the driver fan-out
    /// posts one of these per registered NodeId (the raw node in <see cref="AddressSpaceRealm.Raw"/> and each
    /// referencing UNS node in <see cref="AddressSpaceRealm.Uns"/>) with identical value/quality/timestamp.
    /// <see cref="Realm"/> defaults to <see cref="AddressSpaceRealm.Uns"/> so pre-v3 callers (VirtualTag
    /// equipment nodes) keep compiling; the driver fan-out passes it EXPLICITLY per node.</summary>
    public sealed record AttributeValueUpdate(string NodeId, object? Value, OpcUaQuality Quality, DateTime TimestampUtc, AddressSpaceRealm Realm);

    /// <summary>Carries the full Part 9 condition state for a scripted alarm to the sink. The
    /// <paramref name="State"/> snapshot is the Commons projection the Runtime host maps from the engine's
    /// Core <c>AlarmConditionState</c> + severity/message — the actor stays decoupled from
    /// <c>Core.ScriptedAlarms</c>.</summary>
    /// <param name="AlarmNodeId">The alarm node id (== ScriptedAlarmId for scripted conditions; == RawPath for
    /// v3 native raw conditions).</param>
    /// <param name="State">The full condition state to project onto the node.</param>
    /// <param name="TimestampUtc">The source timestamp of the transition in UTC.</param>
    /// <param name="Realm">The namespace realm the condition lives in — <see cref="AddressSpaceRealm.Uns"/> for
    /// scripted alarms (default), <see cref="AddressSpaceRealm.Raw"/> for v3 native raw conditions.</param>
    public sealed record AlarmStateUpdate(string AlarmNodeId, AlarmConditionSnapshot State, DateTime TimestampUtc, AddressSpaceRealm Realm);
    /// <summary>#477 — annotate a materialised condition's source-data Quality out of band from any alarm
    /// transition (the driver-connectivity path: comms lost → <see cref="OpcUaQuality.Bad"/>, restored →
    /// <see cref="OpcUaQuality.Good"/>). Routed to <see cref="IOpcUaAddressSpaceSink.WriteAlarmQuality"/>,
    /// which sets ONLY Quality and fires one Part 9 event on a quality-bucket change.</summary>
    /// <param name="AlarmNodeId">The condition node id (RawPath for a native alarm).</param>
    /// <param name="Quality">The source-data quality to annotate.</param>
    /// <param name="TimestampUtc">The connectivity transition timestamp in UTC.</param>
    /// <param name="Realm">The namespace realm the condition lives in (<see cref="AddressSpaceRealm.Raw"/> for native).</param>
    public sealed record AlarmQualityUpdate(string AlarmNodeId, OpcUaQuality Quality, DateTime TimestampUtc, AddressSpaceRealm Realm);
    /// <summary>
    ///     Triggers an address-space rebuild. <paramref name="DeploymentId"/> is the deployment
    ///     just applied by the host; the rebuild loads THAT artifact so materialisation matches the
    ///     applied config + the SubscribeBulk pass. It is null only for legacy/dev callers, which
    ///     fall back to the latest sealed deployment (lags a not-yet-sealed apply by one revision).
    /// </summary>
    public sealed record RebuildAddressSpace(CorrelationId Correlation, DeploymentId? DeploymentId = null);

    /// <summary>Inject driver-discovered nodes (FixedTree) under an equipment at runtime (post-connect).</summary>
    /// <param name="EquipmentRootNodeId">The OPC UA NodeId of the equipment root folder to inject the
    /// discovered nodes under (e.g. "EQ-3686c0272279"); also the node the NodeAdded model-change is
    /// announced under.</param>
    public sealed record MaterialiseDiscoveredNodes(
        string EquipmentRootNodeId,
        IReadOnlyList<DiscoveredFolder> Folders,
        IReadOnlyList<DiscoveredVariable> Variables);

    public sealed record ServiceLevelChanged(byte ServiceLevel);

    private readonly IOpcUaAddressSpaceSink _sink;
    private readonly IServiceLevelPublisher _serviceLevel;
    private readonly bool _subscribeRedundancyTopic;
    private readonly NodeId? _localNode;
    private readonly IDbContextFactory<OtOpcUaConfigDbContext>? _dbFactory;
    private readonly AddressSpaceApplier? _applier;
    private readonly IActorRef? _dbHealthProbe;
    private readonly TimeSpan _staleWindow;
    private readonly TimeSpan _probeFreshnessWindow;
    private readonly TimeSpan _healthTickInterval;
    private readonly Akka.Cluster.Cluster _cluster = Akka.Cluster.Cluster.Get(Context.System);
    private readonly ILoggingAdapter _log = Context.GetLogger();

    private int _writes;
    private byte _lastServiceLevel;
    private bool _publishedAtLeastOnce;
    private DbHealthProbeActor.DbHealthStatus? _lastDbHealth;
    private RedundancyStateChanged? _lastSnapshot;
    private (bool Ok, DateTime At)? _probeAboutMe;
    private AddressSpaceComposition _lastApplied = new(
        Array.Empty<UnsAreaProjection>(),
        Array.Empty<UnsLineProjection>(),
        Array.Empty<EquipmentNode>(),
        Array.Empty<DriverInstancePlan>(),
        Array.Empty<ScriptedAlarmPlan>());

    /// <summary>Gets the number of writes performed.</summary>
    public int WriteCount => _writes;
    /// <summary>Gets the last published service level.</summary>
    public byte LastServiceLevel => _lastServiceLevel;

    /// <summary>Production Props — pins the OPC UA dispatcher + subscribes to the
    /// <c>redundancy-state</c> DPS topic so cluster transitions drive the local ServiceLevel
    /// publish path. When <paramref name="dbFactory"/> + <paramref name="applier"/> are supplied,
    /// <see cref="RebuildAddressSpace"/> reads the latest deployment artifact + drives the
    /// applier through the sink. When <paramref name="dbHealthProbe"/> is supplied the local
    /// ServiceLevel is computed via <see cref="ServiceLevelCalculator"/> from real DB-health +
    /// staleness + role-leader inputs; otherwise the legacy role-only switch is used.</summary>
    /// <param name="sink">The OPC UA address space sink.</param>
    /// <param name="serviceLevel">The service level publisher.</param>
    /// <param name="localNode">The local cluster node ID.</param>
    /// <param name="dbFactory">The optional database context factory.</param>
    /// <param name="applier">The optional Phase 7 applier.</param>
    /// <param name="dbHealthProbe">The optional <see cref="DbHealthProbeActor"/> ref; when null the
    /// legacy role-only ServiceLevel seam is used until a <see cref="DbHealthProbeActor.DbHealthStatus"/> arrives.</param>
    /// <param name="staleWindow">The window beyond which a DB-health sample or redundancy snapshot is
    /// considered stale; defaults to 30 seconds.</param>
    /// <param name="probeFreshnessWindow">The window beyond which a peer's OPC UA probe verdict about
    /// this node is considered stale (and thus given the benefit of the doubt rather than demoting);
    /// defaults to 30 seconds.</param>
    /// <param name="healthTickInterval">The period between self-driven DB-health refresh ticks (each
    /// Asks <paramref name="dbHealthProbe"/> for its cached status); defaults to 5 seconds. No timer is
    /// started when <paramref name="dbHealthProbe"/> is null.</param>
    /// <returns>The configured <see cref="Props"/> for creating this actor, pinned to the
    /// <see cref="DispatcherId"/> dispatcher.</returns>
    public static Props Props(
        IOpcUaAddressSpaceSink? sink = null,
        IServiceLevelPublisher? serviceLevel = null,
        NodeId? localNode = null,
        IDbContextFactory<OtOpcUaConfigDbContext>? dbFactory = null,
        AddressSpaceApplier? applier = null,
        IActorRef? dbHealthProbe = null,
        TimeSpan? staleWindow = null,
        TimeSpan? probeFreshnessWindow = null,
        TimeSpan? healthTickInterval = null) =>
        Akka.Actor.Props.Create(() => new OpcUaPublishActor(
            sink ?? NullOpcUaAddressSpaceSink.Instance,
            serviceLevel ?? NullServiceLevelPublisher.Instance,
            subscribeRedundancyTopic: true,
            localNode,
            dbFactory,
            applier,
            dbHealthProbe,
            staleWindow,
            probeFreshnessWindow,
            healthTickInterval)).WithDispatcher(DispatcherId);

    /// <summary>Test-only Props that omits the pinned-dispatcher requirement and skips the
    /// DPS subscribe so unit tests can spin up the actor on a vanilla TestKit cluster.</summary>
    /// <param name="sink">The OPC UA address space sink.</param>
    /// <param name="serviceLevel">The service level publisher.</param>
    /// <param name="subscribeRedundancyTopic">Whether to subscribe to the redundancy topic.</param>
    /// <param name="localNode">The local cluster node ID.</param>
    /// <param name="dbFactory">The optional database context factory.</param>
    /// <param name="applier">The optional Phase 7 applier.</param>
    /// <param name="dbHealthProbe">The optional <see cref="DbHealthProbeActor"/> ref; when null the
    /// legacy role-only ServiceLevel seam is used until a <see cref="DbHealthProbeActor.DbHealthStatus"/> arrives.</param>
    /// <param name="staleWindow">The window beyond which a DB-health sample or redundancy snapshot is
    /// considered stale; defaults to 30 seconds.</param>
    /// <param name="probeFreshnessWindow">The window beyond which a peer's OPC UA probe verdict about
    /// this node is considered stale (and thus given the benefit of the doubt rather than demoting);
    /// defaults to 30 seconds.</param>
    /// <param name="healthTickInterval">The period between self-driven DB-health refresh ticks (each
    /// Asks <paramref name="dbHealthProbe"/> for its cached status); defaults to 5 seconds. No timer is
    /// started when <paramref name="dbHealthProbe"/> is null.</param>
    /// <returns>The configured <see cref="Props"/> for creating this actor in tests, without dispatcher
    /// pinning or the DPS subscribe.</returns>
    public static Props PropsForTests(
        IOpcUaAddressSpaceSink? sink = null,
        IServiceLevelPublisher? serviceLevel = null,
        bool subscribeRedundancyTopic = false,
        NodeId? localNode = null,
        IDbContextFactory<OtOpcUaConfigDbContext>? dbFactory = null,
        AddressSpaceApplier? applier = null,
        IActorRef? dbHealthProbe = null,
        TimeSpan? staleWindow = null,
        TimeSpan? probeFreshnessWindow = null,
        TimeSpan? healthTickInterval = null) =>
        Akka.Actor.Props.Create(() => new OpcUaPublishActor(
            sink ?? NullOpcUaAddressSpaceSink.Instance,
            serviceLevel ?? NullServiceLevelPublisher.Instance,
            subscribeRedundancyTopic,
            localNode,
            dbFactory,
            applier,
            dbHealthProbe,
            staleWindow,
            probeFreshnessWindow,
            healthTickInterval));

    /// <summary>Initializes a new instance of the <see cref="OpcUaPublishActor"/> class.</summary>
    /// <param name="sink">The OPC UA address space sink.</param>
    /// <param name="serviceLevel">The service level publisher.</param>
    /// <param name="subscribeRedundancyTopic">Whether to subscribe to the redundancy topic.</param>
    /// <param name="localNode">The local cluster node ID.</param>
    /// <param name="dbFactory">The optional database context factory.</param>
    /// <param name="applier">The optional Phase 7 applier.</param>
    /// <param name="dbHealthProbe">The optional <see cref="DbHealthProbeActor"/> ref; when null the
    /// legacy role-only ServiceLevel seam is used until a <see cref="DbHealthProbeActor.DbHealthStatus"/> arrives.</param>
    /// <param name="staleWindow">The window beyond which a DB-health sample or redundancy snapshot is
    /// considered stale; defaults to 30 seconds.</param>
    /// <param name="probeFreshnessWindow">The window beyond which a peer's OPC UA probe verdict about
    /// this node is considered stale (and thus given the benefit of the doubt rather than demoting);
    /// defaults to 30 seconds.</param>
    /// <param name="healthTickInterval">The period between self-driven DB-health refresh ticks (each
    /// Asks <paramref name="dbHealthProbe"/> for its cached status); defaults to 5 seconds. No timer is
    /// started when <paramref name="dbHealthProbe"/> is null.</param>
    public OpcUaPublishActor(
        IOpcUaAddressSpaceSink sink,
        IServiceLevelPublisher serviceLevel,
        bool subscribeRedundancyTopic,
        NodeId? localNode,
        IDbContextFactory<OtOpcUaConfigDbContext>? dbFactory = null,
        AddressSpaceApplier? applier = null,
        IActorRef? dbHealthProbe = null,
        TimeSpan? staleWindow = null,
        TimeSpan? probeFreshnessWindow = null,
        TimeSpan? healthTickInterval = null)
    {
        _sink = sink;
        _serviceLevel = serviceLevel;
        _subscribeRedundancyTopic = subscribeRedundancyTopic;
        _localNode = localNode;
        _dbFactory = dbFactory;
        _applier = applier;
        _dbHealthProbe = dbHealthProbe;
        _staleWindow = staleWindow ?? TimeSpan.FromSeconds(30);
        _probeFreshnessWindow = probeFreshnessWindow ?? TimeSpan.FromSeconds(30);
        _healthTickInterval = healthTickInterval ?? TimeSpan.FromSeconds(5);

        Receive<AttributeValueUpdate>(HandleAttributeUpdate);
        Receive<AlarmStateUpdate>(HandleAlarmUpdate);
        Receive<AlarmQualityUpdate>(HandleAlarmQualityUpdate);
        Receive<RebuildAddressSpace>(HandleRebuild);
        Receive<MaterialiseDiscoveredNodes>(HandleMaterialiseDiscovered);
        Receive<ServiceLevelChanged>(HandleServiceLevelChanged);
        Receive<RedundancyStateChanged>(HandleRedundancyStateChanged);
        Receive<DbHealthProbeActor.DbHealthStatus>(HandleDbHealthStatus);
        Receive<HealthTick>(_ => OnHealthTick());
        Receive<PeerOpcUaProbeActor.OpcUaProbeResult>(HandlePeerProbe);
        Receive<SubscribeAck>(_ => { /* PubSub ack */ });
    }

    /// <inheritdoc />
    protected override void PreStart()
    {
        if (_subscribeRedundancyTopic)
        {
            DistributedPubSub.Get(Context.System).Mediator.Tell(new Subscribe(RedundancyStateTopic, Self));
        }

        // Production delivery of DB-health: when a probe is wired, kick an immediate refresh + start the
        // periodic tick so _lastDbHealth gets populated and kept fresh without any external pump. Gated
        // ONLY on the probe being present (independent of the DPS subscribe) so the calculator path works
        // in tests too. No probe → stay on the legacy role-only seam (no timer).
        if (_dbHealthProbe is not null)
        {
            Self.Tell(HealthTick.Instance);                                        // immediate first refresh
            Timers.StartPeriodicTimer("db-health", HealthTick.Instance, _healthTickInterval);
        }
    }

    private void HandleAttributeUpdate(AttributeValueUpdate msg)
    {
        try
        {
            _sink.WriteValue(msg.NodeId, msg.Value, msg.Quality, msg.TimestampUtc, msg.Realm);
            Interlocked.Increment(ref _writes);
            OtOpcUaTelemetry.OpcUaSinkWrite.Add(1, new KeyValuePair<string, object?>("kind", "value"));
        }
        catch (Exception ex)
        {
            _log.Warning(ex, "OpcUaPublish: sink.WriteValue threw for {Node}", msg.NodeId);
        }
    }

    private void HandleAlarmUpdate(AlarmStateUpdate msg)
    {
        try
        {
            _sink.WriteAlarmCondition(msg.AlarmNodeId, msg.State, msg.TimestampUtc, msg.Realm);
            Interlocked.Increment(ref _writes);
            OtOpcUaTelemetry.OpcUaSinkWrite.Add(1, new KeyValuePair<string, object?>("kind", "alarm"));
        }
        catch (Exception ex)
        {
            _log.Warning(ex, "OpcUaPublish: sink.WriteAlarmCondition threw for {Node}", msg.AlarmNodeId);
        }
    }

    private void HandleAlarmQualityUpdate(AlarmQualityUpdate msg)
    {
        try
        {
            _sink.WriteAlarmQuality(msg.AlarmNodeId, msg.Quality, msg.TimestampUtc, msg.Realm);
            Interlocked.Increment(ref _writes);
            OtOpcUaTelemetry.OpcUaSinkWrite.Add(1, new KeyValuePair<string, object?>("kind", "alarm-quality"));
        }
        catch (Exception ex)
        {
            _log.Warning(ex, "OpcUaPublish: sink.WriteAlarmQuality threw for {Node}", msg.AlarmNodeId);
        }
    }

    private void HandleRebuild(RebuildAddressSpace msg)
    {
        using var span = OtOpcUaTelemetry.StartAddressSpaceRebuildSpan();
        span?.SetTag("otopcua.correlation_id", msg.Correlation.ToString());

        // Two modes: when dbFactory + applier are wired, do a real diff-and-apply pass against
        // the latest deployment artifact. Without them, fall back to a raw sink rebuild — the
        // F10b/dev path before the integration completes.
        if (_dbFactory is null || _applier is null)
        {
            try
            {
                _sink.RebuildAddressSpace();
                OtOpcUaTelemetry.OpcUaSinkWrite.Add(1, new KeyValuePair<string, object?>("kind", "rebuild"));
            }
            catch (Exception ex)
            {
                _log.Error(ex, "OpcUaPublish: sink.RebuildAddressSpace threw (correlation={Correlation})",
                    msg.Correlation);
            }
            return;
        }

        try
        {
            // Prefer the artifact of the deployment the host just applied — at apply time it is not
            // yet Sealed, so LoadLatestArtifact would return the PREVIOUS revision and materialise a
            // stale composition (variables that don't match the SubscribeBulk refs). Fall back to
            // latest-sealed only for legacy callers that don't carry a DeploymentId.
            var artifact = msg.DeploymentId is { } depId
                ? LoadArtifact(depId)
                : LoadLatestArtifact();
            var composition = _localNode is { } ln
                ? DeploymentArtifact.ParseComposition(artifact, ln.Value,
                    inconsistency => _log.Warning("OpcUaPublish {Node}: cross-cluster binding — {Message}", ln, inconsistency))
                : DeploymentArtifact.ParseComposition(artifact);
            var plan = AddressSpacePlanner.Compute(_lastApplied, composition);

            if (plan.IsEmpty)
            {
                _log.Debug("OpcUaPublish: rebuild requested but plan is empty (correlation={Correlation})",
                    msg.Correlation);
                return;
            }

            var outcome = _applier.Apply(plan);
            _lastApplied = composition;

            // Sum swallowed per-node materialise failures across every pass together with Apply's own
            // removal-pass tally (archreview 01/S-1). A degraded apply used to vanish into per-node
            // Warnings + an optimistic Info line; now it surfaces at Error + a dedicated meter.
            var failedNodes = outcome.FailedNodes;
            failedNodes += _applier.MaterialiseHierarchy(composition);
            // v3 Batch 4 — the Raw device subtree (ns=Raw): containers (Folder→Driver→Device→TagGroup) then
            // tag Variables keyed by RawPath. Materialised BEFORE the UNS references so each UNS Organizes→Raw
            // edge finds its raw target. The driver binds live values to these raw NodeIds.
            failedNodes += _applier.MaterialiseRawSubtree(composition);
            // v3 Batch 4 — the UNS reference Variables (ns=UNS): each projects a raw tag under its equipment
            // folder (created by MaterialiseHierarchy) with an Organizes→Raw edge; values fan out from the raw
            // node. Runs AFTER both the equipment folders AND the raw subtree exist.
            failedNodes += _applier.MaterialiseUnsReferences(composition);
            // T14 — scripted alarms get their own pass right after the hierarchy so the equipment
            // folders they parent under already exist. Materialises real Part 9 AlarmConditionState
            // nodes (keyed by ScriptedAlarmId so AlarmStateUpdate writes target them); disabled
            // alarms are skipped.
            failedNodes += _applier.MaterialiseScriptedAlarms(composition);
            // Equipment-namespace tags get their own pass: ensures each signal's Variable (and any
            // FolderPath sub-folder) exists under its already-materialised equipment folder so
            // clients can browse them. Live values are pushed by DriverHostActor.ForwardToMux after
            // each subscription cycle; variables show BadWaitingForInitialData only until the first
            // publish interval fires.
            failedNodes += _applier.MaterialiseEquipmentTags(composition);
            // Equipment-namespace VirtualTags get their own pass right after the equipment tags:
            // ensures each computed signal's Variable (and any FolderPath sub-folder) exists under its
            // equipment folder with a folder-scoped NodeId. VirtualTagHostActor.OnResult pushes live
            // values once the first dependency update arrives; until then variables show BadWaitingForInitialData.
            failedNodes += _applier.MaterialiseEquipmentVirtualTags(composition);

            // R2-07 T4b — on a NON-rebuild apply (PureAdd / AttributeOnly), the Materialise passes above
            // just created exactly the added nodes WITHOUT tearing anything down, so announce them with a
            // Part 3 GeneralModelChangeEvent(NodeAdded) per affected parent — model-aware clients then
            // re-browse and pick up the new nodes while every existing MonitoredItem stays alive. Ordering
            // is correct by construction: Apply returned before the passes ran, and this announce runs after
            // them, so the nodes exist when clients re-browse. After a full rebuild the announcement is moot
            // (subscriptions are dead anyway) — the guard skips it. AnnounceAddedNodes is Safe-wrapped
            // internally, so a faulting announce can never break the deploy.
            if (!outcome.RebuildCalled) _applier.AnnounceAddedNodes(plan);

            OtOpcUaTelemetry.OpcUaSinkWrite.Add(1, new KeyValuePair<string, object?>("kind", "rebuild"));
            if (outcome.RebuildFailed || failedNodes > 0)
            {
                OtOpcUaTelemetry.OpcUaApplyFailed.Add(1,
                    new KeyValuePair<string, object?>("kind", outcome.RebuildFailed ? "rebuild" : "nodes"));
                _log.Error(
                    "OpcUaPublish: address-space apply DEGRADED (correlation={Correlation}, rebuildFailed={RebuildFailed}, failedNodes={FailedNodes}, added={Added}, removed={Removed}, changed={Changed}) — the running address space may be stale or partial",
                    msg.Correlation, outcome.RebuildFailed, failedNodes,
                    outcome.AddedNodes, outcome.RemovedNodes, outcome.ChangedNodes);
            }
            else
            {
                _log.Info("OpcUaPublish: applied rebuild (correlation={Correlation}, added={Added}, removed={Removed}, changed={Changed}, rebuild={Rebuild})",
                    msg.Correlation, outcome.AddedNodes, outcome.RemovedNodes, outcome.ChangedNodes, outcome.RebuildCalled);
            }
        }
        catch (Exception ex)
        {
            _log.Error(ex, "OpcUaPublish: rebuild pipeline threw (correlation={Correlation})", msg.Correlation);
        }
    }

    /// <summary>Read a specific deployment's artifact blob from ConfigDb (the one just applied,
    /// which may not be Sealed yet). Empty array on any failure — parser treats it as "no composition".</summary>
    private byte[] LoadArtifact(DeploymentId deploymentId)
    {
        try
        {
            using var db = _dbFactory!.CreateDbContext();
            return db.Deployments.AsNoTracking()
                .Where(d => d.DeploymentId == deploymentId.Value)
                .Select(d => d.ArtifactBlob)
                .FirstOrDefault() ?? Array.Empty<byte>();
        }
        catch (Exception ex)
        {
            _log.Warning(ex, "OpcUaPublish: failed to load artifact for deployment {Id}; rebuild becomes no-op", deploymentId);
            return Array.Empty<byte>();
        }
    }

    /// <summary>Read the most recent <c>Sealed</c> deployment's artifact blob from ConfigDb.
    /// Empty array on any failure — the parser treats empty blob as "no composition".</summary>
    private byte[] LoadLatestArtifact()
    {
        try
        {
            using var db = _dbFactory!.CreateDbContext();
            return db.Deployments.AsNoTracking()
                .Where(d => d.Status == Configuration.Enums.DeploymentStatus.Sealed)
                .OrderByDescending(d => d.SealedAtUtc)
                .Select(d => d.ArtifactBlob)
                .FirstOrDefault() ?? Array.Empty<byte>();
        }
        catch (Exception ex)
        {
            _log.Warning(ex, "OpcUaPublish: failed to load latest deployment artifact; rebuild becomes no-op");
            return Array.Empty<byte>();
        }
    }

    /// <summary>Forwards driver-discovered (FixedTree) nodes to the applier so they are injected under
    /// the equipment at runtime. No-op (logged) when no applier is wired (dev/Mac/legacy seam), matching the
    /// optional-applier tolerance of <see cref="HandleRebuild"/>.</summary>
    private void HandleMaterialiseDiscovered(MaterialiseDiscoveredNodes msg)
    {
        if (_applier is null)
        {
            _log.Debug("OpcUaPublish: no applier wired — discarding MaterialiseDiscoveredNodes for {Equipment}", msg.EquipmentRootNodeId);
            return;
        }
        var failedNodes = _applier.MaterialiseDiscoveredNodes(msg.EquipmentRootNodeId, msg.Folders, msg.Variables);
        if (failedNodes > 0)
        {
            // archreview 01/S-1: a swallowed discovered-node injection failure surfaces at Error + the
            // dedicated meter instead of vanishing into per-node Warnings.
            OtOpcUaTelemetry.OpcUaApplyFailed.Add(1, new KeyValuePair<string, object?>("kind", "nodes"));
            _log.Error(
                "OpcUaPublish: discovered-node injection DEGRADED for {Equipment} (failedNodes={FailedNodes}) — some discovered nodes may be missing",
                msg.EquipmentRootNodeId, failedNodes);
        }
    }

    private void HandleServiceLevelChanged(ServiceLevelChanged msg)
    {
        // Always publish the FIRST computed level, even if it equals the byte-default 0. Otherwise a
        // node starting Detached/role-less (first level = 0) would be dedup'd away, leaving the SDK's
        // built-in default (255 = full service) standing — a degraded node wrongly advertising 255.
        if (_publishedAtLeastOnce && msg.ServiceLevel == _lastServiceLevel) return;
        _lastServiceLevel = msg.ServiceLevel;
        try
        {
            _serviceLevel.Publish(msg.ServiceLevel);
            _publishedAtLeastOnce = true;
            OtOpcUaTelemetry.ServiceLevelChange.Add(1,
                new KeyValuePair<string, object?>("level", msg.ServiceLevel));
            _log.Debug("OpcUaPublish: ServiceLevel={Level}", msg.ServiceLevel);
        }
        catch (Exception ex)
        {
            _log.Warning(ex, "OpcUaPublish: ServiceLevel publisher threw at level {Level}", msg.ServiceLevel);
        }
    }

    /// <summary>Caches the latest redundancy snapshot and recomputes the local ServiceLevel.
    /// The actual byte is produced by <see cref="RecomputeServiceLevel"/> — either via the
    /// health-aware <see cref="ServiceLevelCalculator"/> (once a DB-health probe + sample are
    /// wired) or via the legacy role-only seam (back-compat / bootstrap).</summary>
    private void HandleRedundancyStateChanged(RedundancyStateChanged msg)
    {
        _lastSnapshot = msg;
        RecomputeServiceLevel();
    }

    /// <summary>Caches the latest DB-health sample and recomputes the local ServiceLevel. The
    /// probe pushes these (or the actor Asks for them); either way the freshest sample feeds the
    /// calculator's <c>DbReachable</c>/<c>Stale</c> inputs.</summary>
    private void HandleDbHealthStatus(DbHealthProbeActor.DbHealthStatus msg)
    {
        _lastDbHealth = msg;
        RecomputeServiceLevel();
    }

    /// <summary>Periodic self-tick: Asks the local <see cref="DbHealthProbeActor"/> for its cached
    /// status and pipes the reply back to <see cref="Self"/>, where <see cref="HandleDbHealthStatus"/>
    /// caches it + recomputes. A hung/late probe (Ask timeout) pipes a <c>Reachable=false</c> status so
    /// the node FAIL-SAFE-DEMOTES rather than freezing the last-known-good. The continuation only
    /// CONSTRUCTS a record off the actor thread — the actual state mutation happens on the actor thread
    /// when the piped <see cref="DbHealthProbeActor.DbHealthStatus"/> is received, so no actor field is
    /// touched here.</summary>
    private void OnHealthTick()
    {
        _dbHealthProbe!.Ask<DbHealthProbeActor.DbHealthStatus>(
                DbHealthProbeActor.GetStatus.Instance, TimeSpan.FromSeconds(1))
            .ContinueWith(t => t.IsCompletedSuccessfully
                ? t.Result
                : new DbHealthProbeActor.DbHealthStatus(false, DateTime.UtcNow, "db-health ask timeout"))
            .PipeTo(Self);
    }

    /// <summary>Records a peer's OPC UA probe verdict about THIS node and recomputes the local
    /// ServiceLevel. The probe's <see cref="PeerOpcUaProbeActor.OpcUaProbeResult.NodeId"/> is the
    /// target that was probed, so a result whose <c>NodeId</c> is not this node is about a peer and
    /// is ignored. A matching result is stamped with the receive time so <see cref="OpcUaProbeOk"/>
    /// can debounce stale verdicts.</summary>
    private void HandlePeerProbe(PeerOpcUaProbeActor.OpcUaProbeResult r)
    {
        // The result targets the probed node. If it isn't me, it's about a peer — ignore it.
        if (_localNode is null || r.NodeId != _localNode.Value) return;
        _probeAboutMe = (r.Ok, DateTime.UtcNow);
        RecomputeServiceLevel();
    }

    /// <summary>The OPC UA self-probe input for the calculator: "did a peer recently observe MY OPC UA
    /// endpoint as reachable?" Returns <c>true</c> (benefit of the doubt) when no peer verdict has
    /// arrived yet (single-node / no peer) or when the latest verdict is older than
    /// <see cref="_probeFreshnessWindow"/> (the peer went away — don't penalise this node for that).
    /// Only an actively-observed, RECENT <c>Ok==false</c> demotes.</summary>
    private bool OpcUaProbeOk()
    {
        if (_probeAboutMe is not { } verdict) return true;
        if (DateTime.UtcNow - verdict.At > _probeFreshnessWindow) return true;
        return verdict.Ok;
    }

    /// <summary>
    ///     Computes the local OPC UA ServiceLevel and routes it through <see cref="ServiceLevelChanged"/>
    ///     (the dedup/publish/metric handler). The full <see cref="ServiceLevelCalculator"/> path is
    ///     used once a DB-health probe is wired AND a sample has arrived; until then (and when no probe
    ///     is supplied at all) a legacy role-only seam keeps the historical "primary-leader → 240,
    ///     secondary → 100, detached → 0" behaviour. The calculator does not model Detached, so a
    ///     detached local node is guarded to 0 before either path runs.
    /// </summary>
    private void RecomputeServiceLevel()
    {
        if (_localNode is null || _lastSnapshot is null) return;

        var entry = _lastSnapshot.Nodes.FirstOrDefault(n => n.NodeId == _localNode.Value);

        // The calculator does NOT model Detached — a healthy detached node would wrongly compute
        // 240, so guard it (and the missing-entry case) to 0 here.
        if (entry is null || entry.Role == RedundancyRole.Detached)
        {
            Self.Tell(new ServiceLevelChanged(0));
            return;
        }

        // Legacy / back-compat seam: with no DB-health probe wired (or before the first sample
        // arrives) fall back to the old role-only switch. This preserves historical behaviour and
        // is the bootstrap value until the first DbHealthStatus lands.
        if (_dbHealthProbe is null || _lastDbHealth is null)
        {
            Self.Tell(new ServiceLevelChanged(LegacyRoleOnly(entry)));
            return;
        }

        var now = DateTime.UtcNow;
        var inputs = new NodeHealthInputs(
            MemberState: SafeSelfStatus(),
            DbReachable: _lastDbHealth.Reachable,
            OpcUaProbeOk: OpcUaProbeOk(),
            Stale: !_lastDbHealth.Reachable
                   || (now - _lastDbHealth.AsOfUtc) > _staleWindow
                   || (now - entry.AsOfUtc) > _staleWindow,
            IsDriverRoleLeader: entry.IsRoleLeaderForDriver);

        Self.Tell(new ServiceLevelChanged(ServiceLevelCalculator.Compute(inputs)));
    }

    /// <summary>The legacy role-only ServiceLevel switch (primary-leader → 240, primary → 200,
    /// secondary → 100, _ → 0). Preserved as the back-compat / bootstrap seam.</summary>
    private static byte LegacyRoleOnly(NodeRedundancyState entry) => entry.Role switch
    {
        RedundancyRole.Primary when entry.IsRoleLeaderForDriver => 240,
        RedundancyRole.Primary => 200,
        RedundancyRole.Secondary => 100,
        _ => 0,
    };

    /// <summary>Reads this node's cluster <see cref="MemberStatus"/>, returning
    /// <see cref="MemberStatus.Removed"/> if the cluster is unavailable (so the calculator treats it
    /// as untrusted → 0 rather than throwing).</summary>
    private MemberStatus SafeSelfStatus()
    {
        try
        {
            return _cluster.SelfMember.Status;
        }
        catch (Exception ex)
        {
            _log.Debug(ex, "OpcUaPublish: SelfMember status unavailable; treating as Removed (ServiceLevel→0)");
            return MemberStatus.Removed;
        }
    }
}