lmxopcua/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/HistorianAdapterActor.cs

using System.Threading.Channels;
using Akka.Actor;
using Akka.Cluster.Tools.PublishSubscribe;
using Akka.Event;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Alerts;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Redundancy;
using ZB.MOM.WW.OtOpcUa.Commons.Types;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian;
using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa;
using ZB.MOM.WW.OtOpcUa.Runtime.ScriptedAlarms;

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

/// <summary>
/// Thin actor wrapper around <see cref="IAlarmHistorianSink"/>. Live alarm transitions arrive off the cluster
/// <c>alerts</c> topic; the Primary's adapter translates each to an <see cref="AlarmHistorianEvent"/> and queues it
/// for the sink. The queue is bounded and drained by ONE loop, so the sink sees one enqueue at a time in arrival order
/// (an alarm storm used to fan into one concurrent SQLite insert per transition, serialising on the file lock and
/// holding pool threads); a full queue holds the mailbox until room frees (backpressure, nothing dropped), and stopping
/// flushes the queue within the shared 5 s stop budget. Production deployments register
/// <see cref="LocalDbStoreAndForwardSink"/> against <c>IAlarmHistorianSink</c>; the sink owns the
/// durable queue + drain-to-HistorianGateway-SendEvent loop. The actor here owns nothing operational beyond
/// the message contract — its job is to keep the engine actors on Akka's mailbox without blocking
/// them on disk I/O or gateway round-trips.
///
/// Query queue depth + drain health via <see cref="GetStatus"/>.
/// </summary>
public sealed class HistorianAdapterActor : ReceiveActor
{
    public sealed record GetStatus
    {
        public static readonly GetStatus Instance = new();
    }

    private readonly IAlarmHistorianSink _sink;
    private readonly NodeId? _localNode;
    private readonly ILoggingAdapter _log = Context.GetLogger();
    private IActorRef? _mediator;

    /// <summary>The most events queued for the sink before a transition waits for room (backpressure).</summary>
    public const int QueueCapacity = 10_000;

    /// <summary>How long stopping waits for the queue to flush (the runtime's shared actor-stop budget).</summary>
    public static readonly TimeSpan FlushBudget = TimeSpan.FromSeconds(5);

    private readonly Channel<AlarmHistorianEvent> _queue = Channel.CreateBounded<AlarmHistorianEvent>(
        new BoundedChannelOptions(QueueCapacity) { SingleReader = true, SingleWriter = true, FullMode = BoundedChannelFullMode.Wait });

    private readonly CancellationTokenSource _lifetime = new();
    private Task _drain = Task.CompletedTask;

    /// <summary>Cached local <see cref="RedundancyRole"/> from the latest <see cref="RedundancyStateChanged"/>
    /// snapshot (null = unknown until the first snapshot arrives, or no <see cref="_localNode"/> wired). The
    /// durable sink enqueue in the <see cref="AlarmHistorianEvent"/> handler is gated on this: only the
    /// Primary historizes (default-write while unknown so single-node deploys + the boot window never drop
    /// historization).</summary>
    private RedundancyRole? _localRole;

    /// <summary>Creates the props for a HistorianAdapterActor instance.</summary>
    /// <param name="sink">The alarm historian sink implementation, or null to use a null sink.</param>
    /// <param name="localNode">The local cluster node id, used to read this node's <see cref="RedundancyRole"/>
    /// from the <c>redundancy-state</c> topic so only the Primary historizes to the durable sink. Null (the
    /// default) leaves the role unknown ⇒ default-write (single-node deploys + tests).</param>
    /// <returns>Props configured for creating a HistorianAdapterActor.</returns>
    public static Props Props(IAlarmHistorianSink? sink = null, NodeId? localNode = null) =>
        Akka.Actor.Props.Create(() => new HistorianAdapterActor(sink ?? NullAlarmHistorianSink.Instance, localNode));

    /// <summary>Initializes a new instance of the HistorianAdapterActor class.</summary>
    /// <param name="sink">The alarm historian sink to forward enqueued events to.</param>
    /// <param name="localNode">The local cluster node id, used to read this node's <see cref="RedundancyRole"/>
    /// from the <c>redundancy-state</c> topic so only the Primary historizes to the durable sink. Null leaves
    /// the role unknown ⇒ default-write (single-node deploys + tests).</param>
    public HistorianAdapterActor(IAlarmHistorianSink sink, NodeId? localNode = null)
    {
        _sink = sink;
        _localNode = localNode;

        // (The direct AlarmHistorianEvent receive is gone: it had no production sender, and a transition told directly
        // AND published on `alerts` would have been written twice.)

        // Live alarm transitions arrive off the cluster `alerts` DPS topic (subscribed in PreStart). The
        // Primary ScriptedAlarmHostActor publishes each transition ONCE, but DistributedPubSub fans that
        // single message to EVERY node's subscriber — including BOTH central nodes' historian adapters. The
        // ShouldHistorize gate keeps only the Primary writing ⇒ exactly-once across the warm pair.
        // NOTE: Translate is intentionally inside the gate so Secondary/Detached nodes never allocate a
        // discarded AlarmHistorianEvent.
        // t.HistorizeToAveva is not false: only explicit false suppresses the durable sink write. null
        // (CLR default for bool?) and true both historize. null is the rolling-restart / cross-version case:
        // an old-format message missing the field deserializes to null and is historized (default-on), so no
        // audit row is dropped at a handover — same posture as the AlarmTypeName null-coalesce in Translate.
        // The producer (ScriptedAlarmHostActor) always sets a concrete true/false.
        ReceiveAsync<AlarmTransitionEvent>(async t =>
        {
            if (!ShouldHistorize() || t.HistorizeToAveva is false) return;
            var evt = Translate(t);
            // Off the mailbox unless the queue is full: then the mailbox waits for room (bounded memory, nothing dropped).
            if (_queue.Writer.TryWrite(evt)) return;
            try
            {
                await _queue.Writer.WriteAsync(evt, _lifetime.Token).ConfigureAwait(false);
            }
            catch (Exception ex) when (ex is OperationCanceledException or ChannelClosedException)
            {
                // Stopping raced the enqueue.
            }
        });

        Receive<GetStatus>(_ => Sender.Tell(_sink.GetStatus()));

        // Cluster redundancy snapshots (published on the `redundancy-state` topic, subscribed in PreStart)
        // cache this node's role so the historize gate can scope the durable sink enqueue to the Primary.
        // The PubSub Subscribe acks (redundancy-state + alerts) are acked back to Self (no-op below).
        Receive<RedundancyStateChanged>(OnRedundancyStateChanged);
        // The redundancy-state topic also carries OpcUaProbeResult (OpcUaPublishActor peer-probes).
        // We don't consume it here — drop it so it doesn't dead-letter (matches PeerProbeSupervisor).
        Receive<ZB.MOM.WW.OtOpcUa.Runtime.Health.PeerOpcUaProbeActor.OpcUaProbeResult>(_ => { });
        Receive<SubscribeAck>(_ => { });
    }

    /// <summary>True iff this node should write to the durable sink: only the Primary historizes
    /// (default-write while the role is unknown so single-node deploys + the boot window never drop
    /// historization). DPS fans the Primary's single alerts publish to BOTH nodes' historians, so this
    /// gate is what keeps the durable write exactly-once across the warm-redundant pair.</summary>
    private bool ShouldHistorize() => _localRole is not (RedundancyRole.Secondary or RedundancyRole.Detached);

    /// <summary>Translates a live <see cref="AlarmTransitionEvent"/> (the alerts-topic shape) into the
    /// historian's <see cref="AlarmHistorianEvent"/>. <c>AlarmTypeName</c> is null-coalesced to
    /// "AlarmCondition": during a rolling restart Akka's JSON serializer applies the CLR default (null) to
    /// an old-format message's <c>AlarmTypeName</c> rather than the record's call-site default, and the
    /// historian must never store a null alarm type.</summary>
    /// <param name="t">The live transition published on the <c>alerts</c> topic.</param>
    /// <returns>The translated historian event.</returns>
    private static AlarmHistorianEvent Translate(AlarmTransitionEvent t) => new(
        AlarmId: t.AlarmId,
        EquipmentPath: t.EquipmentPath,
        AlarmName: t.AlarmName,
        AlarmTypeName: string.IsNullOrEmpty(t.AlarmTypeName) ? "AlarmCondition" : t.AlarmTypeName,
        Severity: ToSeverity(t.Severity),
        EventKind: t.TransitionKind,
        Message: t.Message,
        User: t.User,
        Comment: t.Comment,
        TimestampUtc: t.TimestampUtc);

    /// <summary>Maps the OPC UA 1–1000 numeric severity carried on a transition back to the coarse
    /// <see cref="AlarmSeverity"/>, inverting <c>ScriptedAlarmHostActor.SeverityToInt</c>'s bucket ceilings
    /// (Low=250, Medium=500, High=750, Critical=1000).</summary>
    /// <param name="severity">The OPC UA 1–1000 numeric severity.</param>
    /// <returns>The coarse alarm severity bucket.</returns>
    private static AlarmSeverity ToSeverity(int severity) => severity switch
    {
        <= 250 => AlarmSeverity.Low,
        <= 500 => AlarmSeverity.Medium,
        <= 750 => AlarmSeverity.High,
        _ => AlarmSeverity.Critical,
    };

    /// <inheritdoc />
    protected override void PreStart()
    {
        _mediator = DistributedPubSub.Get(Context.System).Mediator;
        _mediator.Tell(new Subscribe(OpcUaPublishActor.RedundancyStateTopic, Self));
        _mediator.Tell(new Subscribe(ScriptedAlarmHostActor.AlertsTopic, Self));
        _drain = Task.Run(DrainAsync);
        base.PreStart();
    }

    /// <inheritdoc />
    protected override void PostStop()
    {
        // Flush what the Primary already accepted from the alerts topic, bounded by the shared stop budget.
        _queue.Writer.TryComplete();
        if (!_drain.Wait(FlushBudget))
            _log.Warning("Historian adapter stopped with alarm events still queued for the sink; abandoned after {Seconds} s.",
                FlushBudget.TotalSeconds);
        _lifetime.Cancel();
        _lifetime.Dispose();
        base.PostStop();
    }

    /// <summary>The single loop feeding the sink, in arrival order. Never throws: a rejected event is logged.</summary>
    private async Task DrainAsync()
    {
        await foreach (var evt in _queue.Reader.ReadAllAsync(CancellationToken.None).ConfigureAwait(false))
            await EnqueueAsync(evt).ConfigureAwait(false);
    }

    /// <summary>Caches this node's <see cref="RedundancyRole"/> from a cluster redundancy snapshot so the
    /// <see cref="AlarmHistorianEvent"/> handler can gate the durable sink enqueue to the Primary. A snapshot
    /// that doesn't mention <see cref="_localNode"/> (or no local node wired) leaves the cached role unchanged
    /// ⇒ default-write. Mirrors <see cref="OpcUaPublishActor"/>'s handler.</summary>
    /// <param name="msg">The cluster redundancy snapshot.</param>
    private void OnRedundancyStateChanged(RedundancyStateChanged msg)
    {
        if (_localNode is null)
        {
            return;
        }

        var local = msg.Nodes.FirstOrDefault(n => n.NodeId == _localNode.Value);
        if (local is not null)
        {
            _localRole = local.Role;
        }
    }

    private async Task EnqueueAsync(AlarmHistorianEvent evt)
    {
        try
        {
            await _sink.EnqueueAsync(evt, _lifetime.Token).ConfigureAwait(false);
        }
        catch (OperationCanceledException) when (_lifetime.IsCancellationRequested)
        {
            // Stopped past the flush budget.
        }
        catch (Exception ex)
        {
            _log.Error(ex, "Historian sink rejected event for {AlarmId} at {Ts}",
                evt.AlarmId, evt.TimestampUtc);
        }
    }
}