ScadaBridge/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/GrpcCentralTransport.cs

using Akka.Actor;
using Grpc.Core;
using Microsoft.Extensions.Logging;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Audit;
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.Communication.Actors;
using AkkaStatus = Akka.Actor.Status;

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

/// <summary>
/// The <see cref="ICentralTransport"/> that carries the seven site→central sends over gRPC — the
/// migration target for the Akka <c>ClusterClient</c> path. Each method encodes the message with
/// <see cref="CentralControlDtoMapper"/>, dials <c>CentralControlService</c> through the sticky
/// <see cref="CentralChannelProvider"/>, and delivers the decoded reply (or a transient-failure
/// signal) to the waiting Ask.
/// </summary>
/// <remarks>
/// <para>
/// The actor handlers are synchronous; each method here kicks off the RPC as a detached task and
/// <c>Tell</c>s the result to <paramref name="replyTo"/> when it completes — <c>IActorRef.Tell</c>
/// is thread-safe, so the reply lands at the Ask exactly as central's ClusterClient reply did. On
/// any non-OK status the reply is <see cref="Status.Failure"/>, which the S&amp;F / audit / health
/// layers already treat as transient.
/// </para>
/// <para>
/// <b>Cross-node retry only on provably-unsent failures.</b> An <see cref="StatusCode.Unavailable"/>
/// (connection refused / node not ready) flips the channel pair and retries once on the peer.
/// A <see cref="StatusCode.DeadlineExceeded"/> is NEVER retried across nodes — a deploy / write /
/// failover may already have executed, and duplicating it is worse than surfacing a transient
/// failure the layer above tolerates.
/// </para>
/// <para>
/// <b>Per-call deadlines mirror today's Ask timeouts.</b> Notification submit/status →
/// <c>NotificationForwardTimeout</c> (30 s, the value the S&amp;F forwarder and the central service
/// both use); health → <c>HealthReportTimeout</c> (10 s); reconcile → <c>QueryTimeout</c> (30 s);
/// both ingest RPCs → <see cref="SiteStreamGrpcServer.AuditIngestAskTimeout"/> (the one shared 30 s
/// constant). The heartbeat, fire-and-forget with no server-side Ask, is merely bounded by
/// <c>HealthReportTimeout</c> and its failures are swallowed.
/// </para>
/// </remarks>
public sealed class GrpcCentralTransport : ICentralTransport
{
    private readonly CentralChannelProvider _channels;
    private readonly CommunicationOptions _options;
    private readonly ILogger<GrpcCentralTransport> _logger;

    /// <summary>Creates the transport over a channel pair.</summary>
    /// <param name="channels">The sticky central channel pair.</param>
    /// <param name="options">Communication options supplying the per-call deadlines.</param>
    /// <param name="logger">Logger for failover/fault diagnostics.</param>
    public GrpcCentralTransport(
        CentralChannelProvider channels,
        CommunicationOptions options,
        ILogger<GrpcCentralTransport> logger)
    {
        ArgumentNullException.ThrowIfNull(channels);
        ArgumentNullException.ThrowIfNull(options);
        ArgumentNullException.ThrowIfNull(logger);

        _channels = channels;
        _options = options;
        _logger = logger;
    }

    /// <inheritdoc />
    public void SubmitNotification(NotificationSubmit message, IActorRef replyTo)
    {
        var dto = CentralControlDtoMapper.ToDto(message);
        Dispatch(replyTo, _options.NotificationForwardTimeout,
            (c, o) => c.SubmitNotificationAsync(dto, o),
            ack => CentralControlDtoMapper.FromDto(ack));
    }

    /// <inheritdoc />
    public void QueryNotificationStatus(NotificationStatusQuery message, IActorRef replyTo)
    {
        var dto = CentralControlDtoMapper.ToDto(message);
        Dispatch(replyTo, _options.NotificationForwardTimeout,
            (c, o) => c.QueryNotificationStatusAsync(dto, o),
            response => CentralControlDtoMapper.FromDto(response));
    }

    /// <inheritdoc />
    public void IngestAuditEvents(IngestAuditEventsCommand message, IActorRef replyTo)
    {
        var batch = CentralControlDtoMapper.ToDto(message);
        Dispatch(replyTo, SiteStreamGrpcServer.AuditIngestAskTimeout,
            (c, o) => c.IngestAuditEventsAsync(batch, o),
            ack => new IngestAuditEventsReply(CentralControlDtoMapper.FromIngestAck(ack)));
    }

    /// <inheritdoc />
    public void IngestCachedTelemetry(IngestCachedTelemetryCommand message, IActorRef replyTo)
    {
        var batch = CentralControlDtoMapper.ToDto(message);
        Dispatch(replyTo, SiteStreamGrpcServer.AuditIngestAskTimeout,
            (c, o) => c.IngestCachedTelemetryAsync(batch, o),
            ack => new IngestCachedTelemetryReply(CentralControlDtoMapper.FromIngestAck(ack)));
    }

    /// <inheritdoc />
    public void ReconcileSite(ReconcileSiteRequest message, IActorRef replyTo)
    {
        var dto = CentralControlDtoMapper.ToDto(message);
        Dispatch(replyTo, _options.QueryTimeout,
            (c, o) => c.ReconcileSiteAsync(dto, o),
            response => CentralControlDtoMapper.FromDto(response));
    }

    /// <inheritdoc />
    public void ReportSiteHealth(SiteHealthReport message, IActorRef replyTo)
    {
        var dto = CentralControlDtoMapper.ToDto(message);
        Dispatch(replyTo, _options.HealthReportTimeout,
            (c, o) => c.ReportSiteHealthAsync(dto, o),
            ack => CentralControlDtoMapper.FromDto(ack));
    }

    /// <inheritdoc />
    public void SendHeartbeat(HeartbeatMessage message, IActorRef self)
    {
        _ = SendHeartbeatAsync(message);
    }

    private async Task SendHeartbeatAsync(HeartbeatMessage message)
    {
        var (index, client) = _channels.Current();
        var dto = CentralControlDtoMapper.ToDto(message);
        try
        {
            var options = new CallOptions(deadline: DateTime.UtcNow.Add(_options.HealthReportTimeout));
            using var call = client.HeartbeatAsync(dto, options);
            await call.ResponseAsync.ConfigureAwait(false);
        }
        catch (RpcException ex) when (IsConnectFailure(ex))
        {
            // Nudge the pair so the next call tries the peer, but never fault: a heartbeat
            // failure must not surface on the site's heartbeat timer path.
            _channels.ReportUnavailable(index);
            _logger.LogDebug(ex, "Heartbeat to central endpoint {Endpoint} was unavailable.", CurrentEndpointSafe());
        }
        catch (Exception ex)
        {
            _logger.LogDebug(ex, "Heartbeat to central failed (swallowed — heartbeats are fire-and-forget).");
        }
    }

    /// <summary>
    /// Runs a unary RPC on the current channel, delivers the decoded reply to
    /// <paramref name="replyTo"/>, and applies the sticky-failover / no-retry-on-deadline policy.
    /// </summary>
    private void Dispatch<TWire>(
        IActorRef replyTo,
        TimeSpan timeout,
        Func<CentralControlService.CentralControlServiceClient, CallOptions, AsyncUnaryCall<TWire>> call,
        Func<TWire, object> decode)
    {
        _ = DispatchAsync(replyTo, timeout, call, decode);
    }

    private async Task DispatchAsync<TWire>(
        IActorRef replyTo,
        TimeSpan timeout,
        Func<CentralControlService.CentralControlServiceClient, CallOptions, AsyncUnaryCall<TWire>> call,
        Func<TWire, object> decode)
    {
        var (index, client) = _channels.Current();
        try
        {
            var reply = await InvokeAsync(client, timeout, call).ConfigureAwait(false);
            replyTo.Tell(decode(reply));
        }
        catch (RpcException ex) when (IsConnectFailure(ex))
        {
            // Provably unsent: the connection was refused / the node was not ready. Fail over
            // to the peer and retry ONCE. This is the only status we retry across nodes.
            _channels.ReportUnavailable(index);
            var (retryIndex, retryClient) = _channels.Current();
            if (retryIndex != index)
            {
                try
                {
                    var reply = await InvokeAsync(retryClient, timeout, call).ConfigureAwait(false);
                    replyTo.Tell(decode(reply));
                    return;
                }
                catch (Exception retryEx)
                {
                    _logger.LogWarning(retryEx,
                        "Central control-plane call failed on both endpoints; surfacing as transient.");
                    replyTo.Tell(new AkkaStatus.Failure(retryEx));
                    return;
                }
            }

            replyTo.Tell(new AkkaStatus.Failure(ex));
        }
        catch (Exception ex)
        {
            // DeadlineExceeded / Internal / PermissionDenied / a PSK-resolution throw — do NOT
            // retry across nodes (the call may have run). Surface as the transient failure the
            // layer above already tolerates.
            replyTo.Tell(new AkkaStatus.Failure(ex));
        }
    }

    private static async Task<TWire> InvokeAsync<TWire>(
        CentralControlService.CentralControlServiceClient client,
        TimeSpan timeout,
        Func<CentralControlService.CentralControlServiceClient, CallOptions, AsyncUnaryCall<TWire>> call)
    {
        var options = new CallOptions(deadline: DateTime.UtcNow.Add(timeout));
        using var asyncCall = call(client, options);
        return await asyncCall.ResponseAsync.ConfigureAwait(false);
    }

    /// <summary>
    /// A failure that provably never reached a server — the only class safe to retry on the peer.
    /// <see cref="StatusCode.DeadlineExceeded"/> is deliberately excluded (the call may have run).
    /// </summary>
    /// <remarks>
    /// Two shapes qualify: a server-signalled <see cref="StatusCode.Unavailable"/> (e.g. a node
    /// that returns Unavailable while it is still starting), and a client-side failure to even
    /// start the call — Grpc.Net surfaces a refused/failed connection as
    /// <see cref="StatusCode.Internal"/> "Error starting gRPC call" with the transport exception
    /// attached, and there the request never left the client. Anything else — including a deadline,
    /// a permission denial, or a generic server-side Internal after the call reached the server —
    /// is NOT retried across nodes.
    /// </remarks>
    private static bool IsConnectFailure(RpcException ex)
    {
        if (ex.StatusCode == StatusCode.Unavailable)
        {
            return true;
        }

        // "Error starting gRPC call" is Grpc.Net's marker for a call that could not be sent — a
        // refused/failed connection carrying an HttpRequestException. Provably unsent.
        return ex.StatusCode == StatusCode.Internal
            && (ex.Status.DebugException is HttpRequestException
                || ex.Status.Detail.StartsWith("Error starting gRPC call", StringComparison.Ordinal));
    }

    private string CurrentEndpointSafe()
    {
        try
        {
            return _channels.CurrentEndpoint;
        }
        catch
        {
            return "(unknown)";
        }
    }
}