ScadaBridge/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/CentralCommunicationActor.cs

using Akka.Actor;
using Akka.Cluster.Tools.PublishSubscribe;
using Akka.Event;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Audit;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Communication;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Deployment;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Health;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Notification;
using ZB.MOM.WW.ScadaBridge.HealthMonitoring;

namespace ZB.MOM.WW.ScadaBridge.Communication.Actors;

/// <summary>
/// Central-side actor that routes messages from central to site clusters over the gRPC
/// <c>SiteCommandService</c> command plane (<see cref="Grpc.GrpcSiteTransport"/>). Resolves site
/// addresses from the database on a periodic refresh cycle and reconciles the transport's per-site
/// channel pairs.
///
/// All 8 message patterns routed through this actor.
/// Ask timeout on connection drop (no central buffering). Debug streams killed on interruption.
/// </summary>
public class CentralCommunicationActor : ReceiveActor
{
    private readonly ILoggingAdapter _log = Context.GetLogger();
    private readonly IServiceProvider _serviceProvider;

    /// <summary>
    /// The central→site command transport (the gRPC <see cref="Grpc.GrpcSiteTransport"/>, built by
    /// the Host and injected). The <see cref="SiteEnvelope"/> handler delegates every send here, and
    /// each DB refresh tick reconciles its per-site channel pairs. Assigned in the constructor body
    /// before any message can arrive.
    /// </summary>
    private ISiteCommandTransport _transport = null!;

    // The previous _debugSubscriptions / _inProgressDeployments
    // dictionaries existed solely to support a documented "synchronous kill streams +
    // mark deployments failed on site disconnect" workflow triggered by
    // ConnectionStateChanged. No production code ever emitted that message — only
    // the unit test did — so the workflow was dead from end to end. Disconnect
    // detection is owned by the underlying transports: the gRPC keepalive PING
    // signals stream interruption in ~25s (handled by DebugStreamBridgeActor's own
    // reconnection logic), and an Ask round-trip for a deploy times out at the
    // CommunicationService layer (caller sees failure). The tracking dicts +
    // ConnectionStateChanged record + HandleConnectionStateChanged handler are
    // removed; see docs/requirements/Component-Communication.md "Connection
    // Failure Behavior" for the keepalive-based contract that survives.

    private ICancelable? _refreshSchedule;

    /// <summary>
    /// Per-actor lifecycle CTS threaded into the periodic
    /// <see cref="LoadSiteAddressesFromDb"/> repository call so a hung MS SQL
    /// connection is bounded by actor shutdown rather than holding piped tasks
    /// open indefinitely. Cancelled in <see cref="PostStop"/>; never reset.
    /// </summary>
    private readonly CancellationTokenSource _lifecycleCts = new();

    /// <summary>
    /// Proxy <see cref="IActorRef"/> for the central NotificationOutboxActor cluster singleton.
    /// Set via <see cref="RegisterNotificationOutbox"/> — the Host creates the singleton proxy
    /// after this actor and registers it (mirrors how the site-side actor receives its
    /// runtime <see cref="IActorRef"/>s). Null until registration completes; a notification
    /// arriving before then is rejected with a non-accepted ack so the site retries.
    /// </summary>
    private IActorRef? _notificationOutboxProxy;

    /// <summary>
    /// DistributedPubSub topic used to fan health reports out to the peer
    /// central node so both per-node aggregators stay in sync. See
    /// <see cref="SiteHealthReportReplica"/> for the protocol rationale.
    /// </summary>
    private const string HealthReportTopic = "site-health-replica";

    /// <summary>
    /// Constructs the actor over the given central→site command transport (production injects the
    /// gRPC <see cref="Grpc.GrpcSiteTransport"/>, built by the Host from the DI
    /// <see cref="Grpc.SitePairChannelProvider"/>; tests substitute a fake).
    /// </summary>
    /// <param name="serviceProvider">DI service provider for scoped repository and aggregator access.</param>
    /// <param name="transport">The central→site command transport to route every <see cref="SiteEnvelope"/> through.</param>
    public CentralCommunicationActor(
        IServiceProvider serviceProvider,
        ISiteCommandTransport transport)
        : this(serviceProvider)
    {
        _transport = transport ?? throw new ArgumentNullException(nameof(transport));
    }

    /// <summary>Shared wiring: sets scoped state and registers every message handler. The
    /// <see cref="_transport"/> is assigned by the delegating public constructor before any message
    /// is dispatched.</summary>
    /// <param name="serviceProvider">DI service provider.</param>
    private CentralCommunicationActor(IServiceProvider serviceProvider)
    {
        _serviceProvider = serviceProvider;

        // Site address cache loaded from database
        Receive<SiteAddressCacheLoaded>(HandleSiteAddressCacheLoaded);

        // Periodic refresh trigger
        Receive<RefreshSiteAddresses>(_ => LoadSiteAddressesFromDb());

        // A faulted LoadSiteAddressesFromDb task is piped here as a
        // Status.Failure. Without this handler the failure was an unhandled message
        // (debug-level only) and the refresh failed silently — operators could not
        // distinguish "no sites configured" from "database is down". Log at Warning.
        Receive<Status.Failure>(failure =>
            _log.Warning(failure.Cause,
                "Failed to load site addresses from the database; the per-site gRPC channel "
                + "cache was not refreshed and may be stale or empty"));

        // Health monitoring: heartbeats and health reports from sites
        Receive<HeartbeatMessage>(HandleHeartbeat);
        Receive<SiteHealthReport>(HandleSiteHealthReport);
        Receive<SiteHealthReportReplica>(r => ProcessLocally(r.Report));
        Receive<SiteHeartbeatReplica>(r => MarkHeartbeatLocally(r.Heartbeat));
        Receive<SubscribeAck>(_ => { /* DistributedPubSub subscribe confirmation */ });

        // Route enveloped messages to sites
        Receive<SiteEnvelope>(HandleSiteEnvelope);

        // Notification Outbox: the Host registers the outbox singleton proxy after this
        // actor is created (the proxy cannot exist before this actor's construction).
        Receive<RegisterNotificationOutbox>(msg =>
        {
            _notificationOutboxProxy = msg.OutboxProxy;
            _log.Info("Registered notification outbox proxy");
        });

        // Notification Outbox ingest: a site forwards a buffered NotificationSubmit to the
        // central cluster. Forward to the outbox proxy so the original
        // Sender (the site's transport path) is preserved and the NotificationSubmitAck
        // routes straight back to the site.
        Receive<NotificationSubmit>(HandleNotificationSubmit);

        // Notification Outbox status query: forward to the outbox proxy, preserving Sender
        // so the NotificationStatusResponse routes back to the querying site.
        Receive<NotificationStatusQuery>(HandleNotificationStatusQuery);

        // Audit Log site→central ingest is NOT relayed here. The central-hosted
        // CentralControlGrpcService Asks the audit-log-ingest singleton proxy directly
        // (as SiteStreamGrpcServer always has), so this actor no longer sits in that
        // path — the relay was a second hop with an identical 30 s Ask timeout that
        // could only add latency. Do not reintroduce it.

        // Startup reconciliation: a site node forwards its local deployed inventory on
        // startup. Resolve the scoped ReconcileService, diff the
        // inventory against central's expected set, and pipe the ReconcileSiteResponse
        // (gap fetch tokens + orphans) straight back to the site node.
        Receive<ReconcileSiteRequest>(HandleReconcileSiteRequest);
    }

    private void HandleNotificationSubmit(NotificationSubmit msg)
    {
        if (_notificationOutboxProxy == null)
        {
            // No outbox proxy registered yet. A non-accepted ack makes the site's
            // Store-and-Forward forwarder treat this as transient and retry later.
            _log.Warning(
                "Cannot route NotificationSubmit {0} — notification outbox not available",
                msg.NotificationId);
            Sender.Tell(new NotificationSubmitAck(
                msg.NotificationId, Accepted: false, Error: "notification outbox not available"));
            return;
        }

        _log.Debug("Routing NotificationSubmit {0} to the notification outbox", msg.NotificationId);
        _notificationOutboxProxy.Forward(msg);
    }

    private void HandleNotificationStatusQuery(NotificationStatusQuery msg)
    {
        if (_notificationOutboxProxy == null)
        {
            // No outbox proxy registered yet. Reply Found: false so the querying site
            // falls back to its local Store-and-Forward buffer to resolve the status.
            _log.Warning(
                "Cannot route NotificationStatusQuery {0} — notification outbox not available",
                msg.NotificationId);
            Sender.Tell(new NotificationStatusResponse(
                msg.CorrelationId, Found: false, Status: "Unknown",
                RetryCount: 0, LastError: null, DeliveredAt: null));
            return;
        }

        _log.Debug("Routing NotificationStatusQuery {0} to the notification outbox", msg.NotificationId);
        _notificationOutboxProxy.Forward(msg);
    }

    /// <summary>
    /// Startup reconciliation (site→central): resolve the scoped
    /// <see cref="ReconcileService"/> in a DI scope, diff the node's reported inventory
    /// against central's expected set, and pipe the <see cref="ReconcileSiteResponse"/>
    /// back to the site node's transport path. The actor stays thin — all the diff
    /// and staging logic lives in the service. Mirrors the DB-access pattern used by
    /// <see cref="LoadSiteAddressesFromDb"/> (Task.Run + CreateScope + PipeTo) and the
    /// Sender-preservation pattern of <see cref="HandleNotificationSubmit"/>.
    ///
    /// On a faulted task PipeTo delivers a <see cref="Status.Failure"/> to the node; its
    /// Ask faults and it simply retries reconcile on the next startup — reconcile is
    /// best-effort, so the fault is allowed to propagate rather than being swallowed.
    /// </summary>
    private void HandleReconcileSiteRequest(ReconcileSiteRequest msg)
    {
        // Capture Sender before the async/PipeTo — Akka resets Sender between dispatches.
        var replyTo = Sender;

        // Bound the DB work by the actor lifecycle. The CTS may have
        // been disposed by PostStop on a racing late message; treat that as "actor gone".
        CancellationToken ct;
        try
        {
            ct = _lifecycleCts.Token;
        }
        catch (ObjectDisposedException)
        {
            return;
        }

        _log.Debug(
            "Handling ReconcileSiteRequest from site {0} node {1} ({2} local instance(s))",
            msg.SiteIdentifier, msg.NodeId, msg.LocalNameToRevisionHash.Count);

        Task.Run(async () =>
        {
            using var scope = _serviceProvider.CreateScope();
            var service = scope.ServiceProvider.GetRequiredService<ReconcileService>();
            return await service.ReconcileAsync(msg, ct).ConfigureAwait(false);
        }).PipeTo(replyTo);
    }

    private void HandleHeartbeat(HeartbeatMessage heartbeat)
    {
        // Synthetic heartbeats carry no liveness meaning — today only
        // CentralChannelProvider's failback probe, which reuses this RPC to ask whether the
        // preferred central endpoint answers again. It is emitted by a site's TRANSPORT layer,
        // not by any node's heartbeat timer, so stamping it would let a site whose real
        // heartbeats had stopped keep looking alive on the health dashboard for as long as its
        // transport kept probing. Dropped before the local mark AND before the peer fan-out,
        // so neither central node's aggregator ever sees it.
        if (heartbeat.Synthetic)
        {
            _log.Debug(
                "Ignoring synthetic heartbeat from site {0} (node '{1}') — probe traffic, not liveness",
                heartbeat.SiteId, heartbeat.NodeHostname);
            return;
        }

        MarkHeartbeatLocally(heartbeat);

        // Fan the heartbeat out to the peer central node so BOTH aggregators mark
        // it, regardless of which central node the site delivered
        // to. Without this, a heartbeat that only ever reaches one node leaves the
        // other node's aggregator blind to that site's liveness after a failover
        // (arch review 02, Low). MarkHeartbeat is idempotent (timestamp overwrite),
        // so self-delivery via pub-sub is harmless.
        try
        {
            DistributedPubSub.Get(Context.System).Mediator.Tell(
                new Publish(HealthReportTopic, new SiteHeartbeatReplica(heartbeat)));
        }
        catch
        {
            // No-op in non-clustered hosts (TestKit).
        }
    }

    /// <summary>
    /// Marks a site heartbeat on the local aggregator without re-broadcasting.
    /// Used for both site-originated heartbeats and peer-replicated ones.
    /// A synthetic heartbeat is never marked — <see cref="HandleHeartbeat"/> already drops it
    /// before the fan-out, so a replica can only carry one if a peer node predates the flag;
    /// the guard here is belt-and-braces on the last hop before the aggregator.
    /// </summary>
    private void MarkHeartbeatLocally(HeartbeatMessage heartbeat)
    {
        if (heartbeat.Synthetic)
        {
            return;
        }

        var aggregator = _serviceProvider.GetService<ICentralHealthAggregator>();
        aggregator?.MarkHeartbeat(heartbeat.SiteId, heartbeat.Timestamp);
    }

    /// <summary>
    /// Handles a report delivered directly from a site:
    /// process locally, then fan out to the peer central node so its
    /// aggregator stays in sync.
    /// </summary>
    private void HandleSiteHealthReport(SiteHealthReport report)
    {
        ProcessLocally(report);

        try
        {
            DistributedPubSub.Get(Context.System).Mediator.Tell(
                new Publish(HealthReportTopic, new SiteHealthReportReplica(report)));
        }
        catch
        {
            // No-op in non-clustered hosts (TestKit).
        }

        // Ack the site so its AkkaHealthReportTransport Ask completes and the
        // report-loss counter-restore path can observe delivery (review 01
        // [Medium]). Guarded so the peer-replica path (SiteHealthReportReplica,
        // which arrives without an Ask sender) never dead-letters an ack.
        if (!Sender.IsNobody())
        {
            Sender.Tell(new SiteHealthReportAck(report.SiteId, report.SequenceNumber, Accepted: true));
        }
    }

    /// <summary>
    /// Applies a report to the local aggregator without re-broadcasting.
    /// Used for both site-originated reports and peer-replicated ones — the
    /// aggregator is idempotent via sequence-number comparison.
    /// </summary>
    private void ProcessLocally(SiteHealthReport report)
    {
        var aggregator = _serviceProvider.GetService<ICentralHealthAggregator>();
        if (aggregator != null)
        {
            aggregator.ProcessReport(report);
        }
        else
        {
            _log.Warning("ICentralHealthAggregator not available, dropping health report from site {0}", report.SiteId);
        }
    }

    // HandleConnectionStateChanged removed — no production
    // caller emitted ConnectionStateChanged, so the workflow ran only in tests.
    // Disconnect detection is owned by the transport layers (gRPC keepalive +
    // Ask timeout).

    private void HandleSiteEnvelope(SiteEnvelope envelope)
    {
        // Below-the-seam routing: the gRPC transport owns the "no route for this site ⇒ warn +
        // drop, caller's Ask times out" contract. Sender is the temporary Ask actor (or the
        // debug-bridge actor) and is preserved for reply routing.
        _transport.Send(envelope, Sender);
    }

    private void LoadSiteAddressesFromDb()
    {
        var self = Self;
        // Pass the actor's lifecycle CT into the repository
        // call so a hung database query is cancelled when the actor stops
        // rather than leaving the piped task to accumulate. Captured locally
        // because the lifecycle CTS may have been disposed by PostStop on a
        // racing late tick; treat that as "actor gone, give up".
        CancellationToken ct;
        try
        {
            ct = _lifecycleCts.Token;
        }
        catch (ObjectDisposedException)
        {
            return;
        }

        Task.Run(async () =>
        {
            using var scope = _serviceProvider.CreateScope();
            var repo = scope.ServiceProvider.GetRequiredService<ISiteRepository>();
            var sites = await repo.GetAllSitesAsync(ct).ConfigureAwait(false);

            var contacts = new Dictionary<string, List<string>>();
            // Parallel gRPC-endpoint cache fed by the SAME DB read (the streaming path's
            // GrpcNodeA/BAddress columns, NOT the Akka NodeA/BAddress ones). No second poll loop —
            // the gRPC transport rides this one, and the Akka transport simply ignores the field.
            var grpcContacts = new Dictionary<string, SiteGrpcEndpoints>();
            foreach (var site in sites)
            {
                var addrs = new List<string>();
                if (!string.IsNullOrWhiteSpace(site.NodeAAddress))
                {
                    var addr = site.NodeAAddress;
                    // Strip actor path suffix if present (legacy format)
                    var idx = addr.IndexOf("/user/");
                    if (idx > 0) addr = addr.Substring(0, idx);
                    addrs.Add(addr);
                }
                if (!string.IsNullOrWhiteSpace(site.NodeBAddress))
                {
                    var addr = site.NodeBAddress;
                    var idx = addr.IndexOf("/user/");
                    if (idx > 0) addr = addr.Substring(0, idx);
                    addrs.Add(addr);
                }
                if (addrs.Count > 0)
                    contacts[site.SiteIdentifier] = addrs;

                var grpcA = string.IsNullOrWhiteSpace(site.GrpcNodeAAddress) ? null : site.GrpcNodeAAddress;
                var grpcB = string.IsNullOrWhiteSpace(site.GrpcNodeBAddress) ? null : site.GrpcNodeBAddress;
                if (grpcA is not null || grpcB is not null)
                    grpcContacts[site.SiteIdentifier] = new SiteGrpcEndpoints(grpcA, grpcB);
            }

            // Freeze the cross-task payload before piping to
            // Self. The message record exposes read-only types (
            // IReadOnlyDictionary / IReadOnlyList) so the Akka.NET message-
            // immutability convention is enforced by type, not just convention.
            var frozen = contacts.ToDictionary(
                kvp => kvp.Key,
                kvp => (IReadOnlyList<string>)kvp.Value.AsReadOnly());

            // Carry the full configured-site identifier set (not just the
            // address-bearing subset in `frozen`) so the aggregator prunes only
            // genuinely-deleted sites and never an addressless-but-configured one.
            var knownSiteIds = sites.Select(s => s.SiteIdentifier).ToList();
            return new SiteAddressCacheLoaded(frozen, knownSiteIds, grpcContacts);
        }).PipeTo(self);
    }

    private void HandleSiteAddressCacheLoaded(SiteAddressCacheLoaded msg)
    {
        // Per-site transport resource reconciliation (build/drop gRPC channel
        // pairs). Runs on the actor thread each refresh tick.
        _transport.ReconcileSites(msg);

        // Self-healing eviction: a site deleted from configuration would otherwise
        // linger in the aggregator as a permanently-offline tile (and live KPI
        // sample source) forever. Prune on every refresh so it disappears within
        // one refresh interval without needing a dedicated deletion event. Transport-agnostic.
        _serviceProvider.GetService<ICentralHealthAggregator>()?.PruneUnknownSites(msg.KnownSiteIds);
    }

    // TrackMessageForCleanup removed — the dicts it fed
    // existed solely to support the dead ConnectionStateChanged workflow.

    /// <inheritdoc />
    protected override SupervisorStrategy SupervisorStrategy()
    {
        return new OneForOneStrategy(
            maxNrOfRetries: -1,
            withinTimeRange: Timeout.InfiniteTimeSpan,
            decider: Decider.From(ex =>
            {
                _log.Warning(ex, "Child actor of CentralCommunicationActor faulted, resuming (state preserved)");
                return Directive.Resume;
            }));
    }

    /// <inheritdoc />
    protected override void PreStart()
    {
        _log.Info("CentralCommunicationActor started");

        // Subscribe to the peer-replication topic so we receive health reports
        // delivered to the other central node and keep our local aggregator
        // in sync (a site may deliver its report to either central node).
        // Tolerant of non-clustered hosts (TestKit) where the extension is absent.
        try
        {
            DistributedPubSub.Get(Context.System).Mediator.Tell(
                new Subscribe(HealthReportTopic, Self));
        }
        catch (Exception ex)
        {
            _log.Debug("DistributedPubSub not available — peer health replication disabled: {0}", ex.Message);
        }

        // Schedule periodic refresh of site addresses from the database
        _refreshSchedule = Context.System.Scheduler.ScheduleTellRepeatedlyCancelable(
            TimeSpan.Zero,
            TimeSpan.FromSeconds(60),
            Self,
            new RefreshSiteAddresses(),
            ActorRefs.NoSender);
    }

    /// <inheritdoc />
    protected override void PostStop()
    {
        _log.Info("CentralCommunicationActor stopped");
        _refreshSchedule?.Cancel();
        // Cancel any in-flight LoadSiteAddressesFromDb so a
        // hung MS SQL query does not outlive the actor.
        try
        {
            _lifecycleCts.Cancel();
        }
        catch (ObjectDisposedException)
        {
            // Double-stop is benign.
        }
        _lifecycleCts.Dispose();
    }
}

/// <summary>
/// Command to trigger a refresh of site addresses from the database.
/// </summary>
public record RefreshSiteAddresses;

/// <summary>
/// Message carrying the loaded site contact data from the database. Per-transport resource
/// reconciliation happens on the actor thread in HandleSiteAddressCacheLoaded, which hands this
/// straight to <see cref="ISiteCommandTransport.ReconcileSites"/>.
///
/// The payload is exposed as <see cref="IReadOnlyDictionary{TKey,TValue}"/>
/// of <see cref="IReadOnlyList{T}"/> so the Akka.NET "messages are immutable"
/// convention is enforced at the type level rather than relying on producer
/// discipline. The producer wraps the constructed buckets with
/// <c>List&lt;T&gt;.AsReadOnly()</c> before piping to Self.
/// </summary>
/// <param name="SiteContacts">Akka remote addresses per site (from NodeA/NodeBAddress).</param>
/// <param name="KnownSiteIds">Every configured site id, address-bearing or not, for aggregator pruning.</param>
/// <param name="GrpcContacts">
/// gRPC endpoint pairs per site (from GrpcNodeA/GrpcNodeBAddress) — the streaming path's columns,
/// consumed by the gRPC transport and ignored by the Akka one.
/// </param>
public sealed record SiteAddressCacheLoaded(
    IReadOnlyDictionary<string, IReadOnlyList<string>> SiteContacts,
    IReadOnlyCollection<string> KnownSiteIds,
    IReadOnlyDictionary<string, SiteGrpcEndpoints> GrpcContacts);

/// <summary>
/// A site's gRPC node-pair endpoints, as loaded from <c>Site.GrpcNodeAAddress</c>/
/// <c>GrpcNodeBAddress</c>. Either may be null when only one node has a gRPC address configured.
/// </summary>
/// <param name="NodeA">NodeA gRPC base address (e.g. <c>http://scadabridge-site-a-node-a:8083</c>), or null.</param>
/// <param name="NodeB">NodeB gRPC base address, or null.</param>
public readonly record struct SiteGrpcEndpoints(string? NodeA, string? NodeB);

/// <summary>
/// Peer-replication envelope for a site heartbeat, fanned out over the same
/// DistributedPubSub topic as <see cref="SiteHealthReportReplica"/> so both
/// central aggregators mark heartbeats regardless of which node the site
/// delivered to. Only ever travels central↔central (same assembly
/// version, rolled together), so the default reflective serializer is safe.
/// </summary>
public sealed record SiteHeartbeatReplica(HeartbeatMessage Heartbeat);

/// <summary>
/// Notification sent to debug view subscribers when the stream is terminated
/// due to site disconnection.
/// </summary>
public record DebugStreamTerminated(string SiteId, string CorrelationId);

/// <summary>
/// Registers the central NotificationOutboxActor singleton proxy with the
/// <see cref="CentralCommunicationActor"/> so site-forwarded <see cref="NotificationSubmit"/>
/// and <see cref="NotificationStatusQuery"/> messages can be routed to it. Sent by the Host
/// after the outbox singleton proxy is created.
/// </summary>
public record RegisterNotificationOutbox(IActorRef OutboxProxy);

// NOTE: there is deliberately no RegisterAuditIngest counterpart. The audit-log-ingest
// singleton proxy goes STRAIGHT to the two gRPC servers that need it
// (CentralControlGrpcService.SetAuditIngestActor / SiteStreamGrpcServer.SetAuditIngestActor);
// routing audit batches through this actor was a redundant hop and was removed.