ScadaBridge/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/GrpcPullSiteCallsClient.cs

using System.Collections.Concurrent;
using Google.Protobuf.WellKnownTypes;
using Grpc.Core;
using Grpc.Net.Client;
using Microsoft.Extensions.Logging;
using ZB.MOM.WW.ScadaBridge.Commons.Entities.Audit;
using ZB.MOM.WW.ScadaBridge.Communication;
using ZB.MOM.WW.ScadaBridge.Communication.Grpc;
using ProtoPullRequest = ZB.MOM.WW.ScadaBridge.Communication.Grpc.PullSiteCallsRequest;
using ProtoPullResponse = ZB.MOM.WW.ScadaBridge.Communication.Grpc.PullSiteCallsResponse;
using PullSiteCallsResponse = ZB.MOM.WW.ScadaBridge.Commons.Messages.Integration.PullSiteCallsResponse;

namespace ZB.MOM.WW.ScadaBridge.AuditLog.Central;

/// <summary>
/// Production <see cref="IPullSiteCallsClient"/> (Site Call Audit) that the
/// central reconciliation tick (a separate follow-up component) uses to pull the
/// next batch of cached-call operational rows from a site over the
/// <c>PullSiteCalls</c> unary gRPC RPC served by <c>SiteStreamGrpcServer</c>.
/// A near-exact sibling of <see cref="GrpcPullAuditEventsClient"/>.
/// </summary>
/// <remarks>
/// <para>
/// <b>Endpoint resolution.</b> The caller passes only a <c>siteId</c>; this
/// client resolves it to a gRPC authority via <see cref="ISiteEnumerator"/>
/// (<see cref="SiteEntry.GrpcEndpoint"/>) on every call so a NodeA→NodeB
/// failover flip or an edited site address takes effect on the next tick. A site
/// with no registered endpoint yields an empty response (no dial).
/// </para>
/// <para>
/// <b>SourceSite re-stamp.</b> The site leaves
/// <c>SiteCallOperationalDto.SourceSite</c> empty (the tracking store has no
/// site-id column). This client is the authority that knows which site it
/// dialed, so it re-stamps the mapped <see cref="SiteCall.SourceSite"/> from
/// <c>siteId</c> — the same "re-stamp from the forwarder's own id" pattern the
/// site push path uses.
/// </para>
/// <para>
/// <b>Fault tolerance.</b> Per the <see cref="IPullSiteCallsClient"/> contract,
/// tolerable transport faults (<see cref="StatusCode.Unavailable"/>,
/// <see cref="StatusCode.DeadlineExceeded"/>, <see cref="StatusCode.Cancelled"/>,
/// bare <see cref="HttpRequestException"/> / <c>SocketException</c>) are caught
/// and collapsed to an empty response so one offline site never sinks the rest
/// of the reconciliation tick. Any other transport/protocol fault is also
/// swallowed to empty: reconciliation is best-effort. Per-row DTO mapping faults
/// (e.g. a single unparseable <c>TrackedOperationId</c>) are narrower still —
/// the offending row is skipped+logged and the rest of the batch is returned.
/// </para>
/// <para>
/// <b>Testability.</b> The unary call is reached through the
/// <see cref="IPullSiteCallsInvoker"/> seam. Production binds
/// <see cref="GrpcPullSiteCallsInvoker"/> (one cached <see cref="GrpcChannel"/>
/// per endpoint, keepalive from <see cref="CommunicationOptions"/>); unit tests
/// inject a fake invoker so no real HTTP/2 endpoint is required.
/// </para>
/// </remarks>
public sealed class GrpcPullSiteCallsClient : IPullSiteCallsClient
{
    private readonly ISiteEnumerator _sites;
    private readonly IPullSiteCallsInvoker _invoker;
    private readonly ILogger<GrpcPullSiteCallsClient> _logger;

    /// <summary>
    /// Creates the client over the given site enumerator and unary-call invoker.
    /// </summary>
    /// <param name="sites">Resolves a <c>siteId</c> to its gRPC endpoint.</param>
    /// <param name="invoker">Seam that issues the <c>PullSiteCalls</c> unary RPC against a resolved endpoint.</param>
    /// <param name="logger">Logger for transport-fault diagnostics.</param>
    public GrpcPullSiteCallsClient(
        ISiteEnumerator sites,
        IPullSiteCallsInvoker invoker,
        ILogger<GrpcPullSiteCallsClient> logger)
    {
        _sites = sites ?? throw new ArgumentNullException(nameof(sites));
        _invoker = invoker ?? throw new ArgumentNullException(nameof(invoker));
        _logger = logger ?? throw new ArgumentNullException(nameof(logger));
    }

    /// <inheritdoc />
    public async Task<PullSiteCallsResponse> PullAsync(
        string siteId,
        DateTime sinceUtc,
        string? afterId,
        int batchSize,
        CancellationToken ct)
    {
        var (endpoint, fallback) = await ResolveEndpointsAsync(siteId, ct).ConfigureAwait(false);
        if (endpoint is null)
        {
            // No gRPC address registered for the site — a configuration decision
            // (mirrors ISiteEnumerator's own contract), not a runtime error, so
            // there is simply nothing to pull.
            _logger.LogDebug(
                "PullSiteCalls skipped: no gRPC endpoint registered for site {SiteId}.", siteId);
            return Empty;
        }

        var request = new ProtoPullRequest
        {
            // ReadChangedSinceAsync treats DateTime.MinValue as "from the start";
            // EnsureUtc keeps Timestamp.FromDateTime happy (it requires UTC kind).
            SinceUtc = Timestamp.FromDateTime(EnsureUtc(sinceUtc)),
            BatchSize = batchSize,
            // Composite-keyset tiebreak. proto3 has no nullable string —
            // an unset/empty AfterId is the site's signal to keep the legacy
            // inclusive-timestamp contract (also what a first pull sends).
            AfterId = afterId ?? string.Empty,
        };

        var (reply, transportFault) = await TryInvokeAsync(endpoint, request, siteId, ct)
            .ConfigureAwait(false);

        // NodeB failover: a transport fault against the primary (NodeA)
        // dials the fallback (NodeB) ONCE before collapsing to empty, so a NodeA
        // outage doesn't take the reconciliation loss-recovery net offline. Only
        // the transport-fault set triggers this (not a mapping/unexpected fault),
        // and not when the caller's token is already cancelled (host shutdown).
        if (reply is null && transportFault
            && !string.IsNullOrWhiteSpace(fallback)
            && !string.Equals(fallback, endpoint, StringComparison.Ordinal)
            && !ct.IsCancellationRequested)
        {
            _logger.LogInformation(
                "PullSiteCalls failing over from primary {Primary} to NodeB {Fallback} for site {SiteId}.",
                endpoint, fallback, siteId);
            (reply, _) = await TryInvokeAsync(fallback, request, siteId, ct).ConfigureAwait(false);
        }

        if (reply is null)
        {
            return Empty;
        }

        // Map proto DTOs to central SiteCall entities PER-ROW so one malformed
        // operational (e.g. an unparseable TrackedOperationId) is skipped+logged
        // rather than sinking the whole batch through the outer catch-all. Each
        // survivor is re-stamped with SourceSite from the dialed siteId (the site
        // leaves it empty).
        var siteCalls = new List<SiteCall>(reply.Operationals.Count);
        foreach (var dto in reply.Operationals)
        {
            try
            {
                var sc = SiteCallDtoMapper.FromDto(dto) with { SourceSite = siteId };
                siteCalls.Add(sc);
            }
            catch (Exception ex)
            {
                _logger.LogWarning(ex,
                    "PullSiteCalls dropped a malformed operational row from site {SiteId} (id='{Id}'); continuing with the rest of the batch.",
                    siteId, dto.TrackedOperationId);
            }
        }

        // Order oldest-first by UpdatedAtUtc (the wire is already ordered by the
        // site read, but the contract is explicit, so sort defensively).
        siteCalls.Sort((a, b) => a.UpdatedAtUtc.CompareTo(b.UpdatedAtUtc));

        return new PullSiteCallsResponse(siteCalls, reply.MoreAvailable);
    }

    /// <summary>
    /// Issues one pull against <paramref name="endpoint"/>, classifying faults.
    /// Returns <c>(reply, false)</c> on success; <c>(null, true)</c> on a
    /// tolerable TRANSPORT fault (the failover-eligible set:
    /// <see cref="StatusCode.Unavailable"/> / <see cref="StatusCode.DeadlineExceeded"/> /
    /// <see cref="StatusCode.Cancelled"/> / <see cref="HttpRequestException"/> /
    /// <c>SocketException</c> / <see cref="OperationCanceledException"/>); and
    /// <c>(null, false)</c> on any other (mapping/unexpected) fault, which
    /// collapses to empty WITHOUT a fallback dial.
    /// </summary>
    private async Task<(ProtoPullResponse? Reply, bool TransportFault)> TryInvokeAsync(
        string endpoint, ProtoPullRequest request, string siteId, CancellationToken ct)
    {
        try
        {
            var reply = await _invoker.InvokeAsync(siteId, endpoint, request, ct).ConfigureAwait(false);
            return (reply, false);
        }
        catch (RpcException ex) when (IsTolerable(ex.StatusCode))
        {
            _logger.LogDebug(ex,
                "PullSiteCalls tolerable transport fault for site {SiteId} ({Endpoint}): {Status}.",
                siteId, endpoint, ex.StatusCode);
            return (null, true);
        }
        catch (Exception ex) when (ex is HttpRequestException or System.Net.Sockets.SocketException)
        {
            _logger.LogDebug(ex,
                "PullSiteCalls connection-layer fault for site {SiteId} ({Endpoint}).",
                siteId, endpoint);
            return (null, true);
        }
        catch (OperationCanceledException)
        {
            // Reconciliation tick cancelled — caller token (host shutdown) or an
            // internal gRPC deadline / linked-CTS cancellation. Tolerable for a
            // best-effort pull.
            return (null, true);
        }
        catch (Exception ex)
        {
            // Any other fault. Reconciliation is best-effort; swallow to empty
            // rather than throw — and do NOT fail over, since a second node would
            // hit the same non-transport fault.
            _logger.LogWarning(ex,
                "PullSiteCalls unexpected fault for site {SiteId} ({Endpoint}). Returning empty batch.",
                siteId, endpoint);
            return (null, false);
        }
    }

    private async Task<(string? Primary, string? Fallback)> ResolveEndpointsAsync(
        string siteId, CancellationToken ct)
    {
        var sites = await _sites.EnumerateAsync(ct).ConfigureAwait(false);
        foreach (var site in sites)
        {
            if (string.Equals(site.SiteId, siteId, StringComparison.Ordinal) &&
                !string.IsNullOrWhiteSpace(site.GrpcEndpoint))
            {
                return (site.GrpcEndpoint, site.FallbackGrpcEndpoint);
            }
        }
        return (null, null);
    }

    private static readonly PullSiteCallsResponse Empty =
        new(Array.Empty<SiteCall>(), MoreAvailable: false);

    private static bool IsTolerable(StatusCode code) => code is
        StatusCode.Unavailable or
        StatusCode.DeadlineExceeded or
        StatusCode.Cancelled;

    // All ScadaBridge timestamps are UTC by invariant. A non-UTC cursor (the
    // reconciliation cursor starts at DateTime.MinValue, Kind=Unspecified) is
    // treated AS UTC — never ToUniversalTime()-converted: on a host with a
    // positive UTC offset MinValue.ToUniversalTime() underflows and
    // Timestamp.FromDateTime throws, crashing the first pull for every site.
    private static DateTime EnsureUtc(DateTime value) =>
        value.Kind == DateTimeKind.Utc ? value : DateTime.SpecifyKind(value, DateTimeKind.Utc);

    /// <summary>
    /// Seam over the <c>PullSiteCalls</c> unary gRPC call against a resolved site
    /// endpoint. Extracted so <see cref="GrpcPullSiteCallsClient"/> can be
    /// unit-tested without a real <see cref="GrpcChannel"/>. Production binds
    /// <see cref="GrpcPullSiteCallsInvoker"/>.
    /// </summary>
    public interface IPullSiteCallsInvoker
    {
        /// <summary>
        /// Issues the <c>PullSiteCalls</c> unary RPC against <paramref name="endpoint"/>.
        /// May throw <see cref="RpcException"/> / <see cref="HttpRequestException"/>
        /// on transport faults — the caller classifies and swallows tolerable ones.
        /// </summary>
        /// <param name="siteId">
        /// The site being pulled from. Selects which preshared key the call presents —
        /// <c>PullSiteCalls</c> is gated by the site's <c>ControlPlaneAuthInterceptor</c>, and
        /// keys are per-site, so the endpoint alone is not enough to authenticate.
        /// </param>
        /// <param name="endpoint">The site gRPC authority (e.g. <c>http://site-a:8083</c>).</param>
        /// <param name="request">The wire-format pull request.</param>
        /// <param name="ct">Cancellation token.</param>
        /// <returns>The wire-format pull response.</returns>
        Task<ProtoPullResponse> InvokeAsync(
            string siteId, string endpoint, ProtoPullRequest request, CancellationToken ct);
    }
}

/// <summary>
/// Production <see cref="GrpcPullSiteCallsClient.IPullSiteCallsInvoker"/>: caches
/// one <see cref="GrpcChannel"/> per endpoint (keepalive from
/// <see cref="CommunicationOptions"/>, mirroring <c>SiteStreamGrpcClient</c>) and
/// issues the unary <c>PullSiteCallsAsync</c> call. The cache is keyed by
/// endpoint string, so a changed site address (NodeA→NodeB failover flip / an
/// edited gRPC address) is reached as soon as the resolver hands the new endpoint
/// to <see cref="InvokeAsync"/>. The channel for a previous address lingers idle
/// until <see cref="Dispose"/> (idle channels hold no streams — a minor cache
/// footprint cost, not a correctness or liveness gap). Sibling of
/// <see cref="GrpcPullAuditEventsInvoker"/>.
/// </summary>
public sealed class GrpcPullSiteCallsInvoker
    : GrpcPullSiteCallsClient.IPullSiteCallsInvoker, IDisposable
{
    private readonly ConcurrentDictionary<(string Site, string Endpoint), GrpcChannel> _channels = new();
    private readonly CommunicationOptions _options;
    private readonly ISitePskProvider? _pskProvider;

    /// <summary>Creates the invoker using default <see cref="CommunicationOptions"/>.</summary>
    public GrpcPullSiteCallsInvoker()
        : this(new CommunicationOptions())
    {
    }

    /// <summary>
    /// Creates the invoker, applying the configured gRPC keepalive settings to
    /// every channel it opens.
    /// </summary>
    /// <param name="options">Communication options supplying gRPC keepalive timings.</param>
    public GrpcPullSiteCallsInvoker(CommunicationOptions options)
        : this(options, pskProvider: null)
    {
    }

    /// <summary>
    /// Creates the invoker with per-site call credentials, the production shape: the site's
    /// <c>ControlPlaneAuthInterceptor</c> refuses an unauthenticated <c>PullSiteCalls</c>.
    /// </summary>
    /// <param name="options">Communication options supplying gRPC keepalive timings.</param>
    /// <param name="pskProvider">Resolves each site's preshared key; null dials unauthenticated.</param>
    public GrpcPullSiteCallsInvoker(CommunicationOptions options, ISitePskProvider? pskProvider)
    {
        _options = options ?? throw new ArgumentNullException(nameof(options));
        _pskProvider = pskProvider;
    }

    /// <inheritdoc />
    public async Task<ProtoPullResponse> InvokeAsync(
        string siteId, string endpoint, ProtoPullRequest request, CancellationToken ct)
    {
        var channel = GetOrCreateChannel(siteId, endpoint);
        var client = new SiteStreamService.SiteStreamServiceClient(channel);
        using var call = client.PullSiteCallsAsync(request, cancellationToken: ct);
        return await call.ResponseAsync.ConfigureAwait(false);
    }

    // Race-safe channel cache (create-then-GetOrAdd-then-dispose-if-lost): two
    // concurrent first dials of the same endpoint can both build a GrpcChannel;
    // only the channel actually installed survives, the loser is disposed.
    // Mirrors SiteStreamGrpcClientFactory / GrpcPullAuditEventsInvoker.
    private GrpcChannel GetOrCreateChannel(string siteId, string endpoint)
    {
        var key = (siteId, endpoint);
        if (!_channels.TryGetValue(key, out var channel))
        {
            var created = CreateChannel(siteId, endpoint);
            channel = _channels.GetOrAdd(key, created);
            if (!ReferenceEquals(channel, created))
            {
                created.Dispose();
            }
        }
        return channel;
    }

    // Keyed by (site, endpoint) rather than endpoint alone: the call credentials are bound to
    // the channel, and they are per-site, so two sites sharing an endpoint string would
    // otherwise share one channel carrying the first site's key.
    private GrpcChannel CreateChannel(string siteId, string endpoint) =>
        GrpcChannel.ForAddress(endpoint, new GrpcChannelOptions
        {
            HttpHandler = new SocketsHttpHandler
            {
                KeepAlivePingDelay = _options.GrpcKeepAlivePingDelay,
                KeepAlivePingTimeout = _options.GrpcKeepAlivePingTimeout,
                KeepAlivePingPolicy = HttpKeepAlivePingPolicy.Always,
            },
        }.WithSiteCredentials(_pskProvider, siteId));

    /// <summary>Disposes all cached channels.</summary>
    public void Dispose()
    {
        foreach (var channel in _channels.Values)
        {
            channel.Dispose();
        }
        _channels.Clear();
    }
}