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

using Akka.Actor;
using Grpc.Net.Client;
using Microsoft.Extensions.Logging;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Artifacts;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.DataConnection;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.DebugView;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Deployment;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.InboundApi;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Lifecycle;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Management;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.RemoteQuery;
using ZB.MOM.WW.ScadaBridge.Communication.Actors;

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

/// <summary>
/// The gRPC central→site command transport: encodes each <see cref="SiteEnvelope"/> command with
/// <see cref="SiteCommandDtoMapper"/>, dials the site's <c>SiteCommandService</c> through the sticky
/// <see cref="SitePairChannelProvider"/> (PSK + <c>x-scadabridge-site</c> already on the channel),
/// and routes the decoded reply back to the waiting <c>Ask</c> (or the debug-bridge actor).
/// </summary>
/// <remarks>
/// <para>
/// <b>Per-call deadlines match today's Ask timeouts exactly</b> — see <see cref="ResolveDeadline"/>.
/// Behaviour is otherwise unchanged from the Akka path: an unknown/unconfigured site is warned and
/// dropped (the caller's Ask times out), and a transport fault surfaces to the caller as a
/// <see cref="Status.Failure"/>, which the S&amp;F/audit layers already treat as transient.
/// </para>
/// <para>
/// <b>Cross-node retry is the channel provider's job</b> and happens only on
/// <c>Unavailable</c> — never on <c>DeadlineExceeded</c> (a write/deploy/failover may have run).
/// </para>
/// </remarks>
public sealed class GrpcSiteTransport : ISiteCommandTransport
{
    private readonly SitePairChannelProvider _channels;
    private readonly CommunicationOptions _options;
    private readonly ILogger<GrpcSiteTransport> _logger;

    // Sites this transport has pushed into the channel provider — used to diff removals per refresh,
    // the gRPC analogue of the Akka transport's _siteClients key set. Touched only on the actor thread.
    private readonly HashSet<string> _knownSites = new(StringComparer.Ordinal);

    /// <summary>Creates the gRPC transport and ensures the provider's failback loop is running.</summary>
    /// <param name="channels">The shared per-site channel-pair provider.</param>
    /// <param name="options">Communication options supplying the per-command deadlines.</param>
    /// <param name="logger">Logger for drop/fault diagnostics.</param>
    public GrpcSiteTransport(
        SitePairChannelProvider channels,
        CommunicationOptions options,
        ILogger<GrpcSiteTransport> logger)
    {
        _channels = channels ?? throw new ArgumentNullException(nameof(channels));
        _options = options ?? throw new ArgumentNullException(nameof(options));
        _logger = logger ?? throw new ArgumentNullException(nameof(logger));
        _channels.EnsureFailbackLoop();
    }

    /// <inheritdoc />
    public void Send(SiteEnvelope envelope, IActorRef replyTo)
    {
        ArgumentNullException.ThrowIfNull(envelope);

        // UnsubscribeDebugView is the one fire-and-forget command: the site never acks it over Akka
        // (the debug bridge Tells and expects nothing), so we run the RPC but do not deliver its
        // synthetic ack — mirroring today's no-reply behaviour and avoiding a dead-lettered ack.
        var fireAndForget = envelope.Message is UnsubscribeDebugViewRequest;

        _ = RunAsync(envelope, replyTo, fireAndForget);
    }

    private async Task RunAsync(SiteEnvelope envelope, IActorRef replyTo, bool fireAndForget)
    {
        try
        {
            var reply = await SendCoreAsync(envelope, CancellationToken.None).ConfigureAwait(false);
            if (!fireAndForget && !replyTo.IsNobody())
            {
                replyTo.Tell(reply, ActorRefs.NoSender);
            }
        }
        catch (SiteChannelUnavailableException)
        {
            // Parity with the Akka "no ClusterClient for site" path: warn and drop, so the caller's
            // Ask times out. Central never buffers.
            _logger.LogWarning(
                "No gRPC channel for site {SiteId}; dropping {Message} (caller's Ask will time out)",
                envelope.SiteId, envelope.Message.GetType().Name);
        }
        catch (Exception ex)
        {
            if (!fireAndForget && !replyTo.IsNobody())
            {
                // A timeout or non-OK status faults the caller's Ask exactly as an Akka Ask timeout
                // did — the S&F/audit layers already treat that as transient.
                replyTo.Tell(new Status.Failure(ex), ActorRefs.NoSender);
            }
            else
            {
                _logger.LogWarning(ex,
                    "Fire-and-forget {Message} to site {SiteId} faulted",
                    envelope.Message.GetType().Name, envelope.SiteId);
            }
        }
    }

    private Task<object> SendCoreAsync(SiteEnvelope envelope, CancellationToken ct)
    {
        var message = envelope.Message;

        // Compute the absolute deadline ONCE so a cross-node failover retry shares the overall
        // budget rather than restarting it.
        var deadline = DateTime.UtcNow + ResolveDeadline(message);
        var group = SiteCommandDtoMapper.GroupOf(message);

        return _channels.ExecuteAsync(
            envelope.SiteId,
            (channel, callCt) => InvokeAsync(channel, group, message, deadline, callCt),
            ct);
    }

    private static async Task<object> InvokeAsync(
        GrpcChannel channel, SiteCommandGroup group, object message, DateTime deadline, CancellationToken ct)
    {
        var client = new SiteCommandService.SiteCommandServiceClient(channel);

        switch (group)
        {
            case SiteCommandGroup.Lifecycle:
            {
                var reply = await client.ExecuteLifecycleAsync(
                    SiteCommandDtoMapper.ToLifecycleRequest(message),
                    deadline: deadline, cancellationToken: ct);
                return SiteCommandDtoMapper.FromLifecycleReply(reply);
            }

            case SiteCommandGroup.OpcUa:
            {
                var reply = await client.ExecuteOpcUaAsync(
                    SiteCommandDtoMapper.ToOpcUaRequest(message),
                    deadline: deadline, cancellationToken: ct);
                return SiteCommandDtoMapper.FromOpcUaReply(reply);
            }

            case SiteCommandGroup.Query:
            {
                var reply = await client.ExecuteQueryAsync(
                    SiteCommandDtoMapper.ToQueryRequest(message),
                    deadline: deadline, cancellationToken: ct);
                return SiteCommandDtoMapper.FromQueryReply(reply);
            }

            case SiteCommandGroup.Parked:
            {
                var reply = await client.ExecuteParkedAsync(
                    SiteCommandDtoMapper.ToParkedRequest(message),
                    deadline: deadline, cancellationToken: ct);
                return SiteCommandDtoMapper.FromParkedReply(reply);
            }

            case SiteCommandGroup.Route:
            {
                var reply = await client.ExecuteRouteAsync(
                    SiteCommandDtoMapper.ToRouteRequest(message),
                    deadline: deadline, cancellationToken: ct);
                return SiteCommandDtoMapper.FromRouteReply(reply);
            }

            case SiteCommandGroup.Failover:
            {
                var ack = await client.TriggerFailoverAsync(
                    SiteCommandDtoMapper.ToProto((TriggerSiteFailover)message),
                    deadline: deadline, cancellationToken: ct);
                return SiteCommandDtoMapper.FromProto(ack);
            }

            default:
                throw new ArgumentOutOfRangeException(nameof(group), group, "Unknown site command group.");
        }
    }

    /// <summary>
    /// The per-command deadline, set EQUAL to the Ask timeout <c>CommunicationService</c> uses for
    /// that command today, so flipping the transport changes nothing about how long a call waits.
    /// Note the group is NOT a uniform deadline class: within Lifecycle, <c>DeploymentStateQuery</c>
    /// uses <c>QueryTimeout</c> (not <c>LifecycleTimeout</c>), and <c>TriggerSiteFailover</c> uses
    /// <c>QueryTimeout</c> (not <c>LifecycleTimeout</c>) — matching the real Ask sites, not the
    /// plan's per-group table. The two parked relays (<c>RetryParkedOperation</c>/
    /// <c>DiscardParkedOperation</c>) map to <c>QueryTimeout</c> (30s), preserving the
    /// <c>SiteCallAuditActor</c> inner <c>RelayTimeout</c> (10s) &lt; 30s ordering. WaitForAttribute
    /// keeps its dynamic <c>request.Timeout + IntegrationTimeout</c> budget.
    /// </summary>
    /// <param name="message">The command being sent.</param>
    /// <returns>The deadline duration for that command.</returns>
    internal TimeSpan ResolveDeadline(object message) => message switch
    {
        RefreshDeploymentCommand => _options.DeploymentTimeout,
        EnableInstanceCommand or DisableInstanceCommand or DeleteInstanceCommand => _options.LifecycleTimeout,
        DeploymentStateQueryRequest => _options.QueryTimeout,
        DeployArtifactsCommand => _options.ArtifactDeploymentTimeout,

        BrowseNodeCommand or SearchAddressSpaceCommand or ReadTagValuesCommand or VerifyEndpointCommand
            or TrustServerCertCommand or ListServerCertsCommand or RemoveServerCertCommand or WriteTagRequest
            => _options.QueryTimeout,

        EventLogQueryRequest or DebugSnapshotRequest => _options.QueryTimeout,
        SubscribeDebugViewRequest or UnsubscribeDebugViewRequest => _options.DebugViewTimeout,

        ParkedMessageQueryRequest or ParkedMessageRetryRequest or ParkedMessageDiscardRequest
            or RetryParkedOperation or DiscardParkedOperation
            => _options.QueryTimeout,

        RouteToCallRequest or RouteToGetAttributesRequest or RouteToSetAttributesRequest
            => _options.IntegrationTimeout,
        RouteToWaitForAttributeRequest r => r.Timeout + _options.IntegrationTimeout,

        TriggerSiteFailover => _options.QueryTimeout,

        _ => throw new ArgumentException(
            $"'{message.GetType().Name}' is not a migrated site command.", nameof(message))
    };

    /// <inheritdoc />
    public void ReconcileSites(SiteAddressCacheLoaded cache)
    {
        ArgumentNullException.ThrowIfNull(cache);

        var desired = cache.GrpcContacts;

        // Drop sites that lost their gRPC endpoints (or were deleted).
        foreach (var removed in _knownSites.Where(s => !desired.ContainsKey(s)).ToList())
        {
            _channels.RemoveSite(removed);
            _knownSites.Remove(removed);
        }

        // Create/refresh the rest.
        foreach (var (siteId, endpoints) in desired)
        {
            _channels.UpdateSite(siteId, endpoints.NodeA, endpoints.NodeB);
            _knownSites.Add(siteId);
        }
    }
}