lmxopcua/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Deployment/GrpcDeploymentArtifactFetcher.cs

using System.Collections.Concurrent;
using System.Security.Cryptography;
using Grpc.Core;
using Grpc.Net.Client;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using ZB.MOM.WW.OtOpcUa.Cluster;
using ZB.MOM.WW.OtOpcUa.Commons.Protos.DeploymentArtifact.V1;

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

/// <summary>
///     Streams the deployed-configuration artifact from central over gRPC and verifies it against the
///     revision hash the dispatch already carries (per-cluster mesh Phase 3, <c>FetchAndCache</c>).
/// </summary>
/// <remarks>
///     <para>
///         <b>One invariant.</b> Return the first endpoint's bytes that both reassemble AND verify
///         (<c>SHA-256(bytes)</c> lowercase-hex == the expected revision hash); otherwise
///         <see langword="null"/>. Every per-endpoint problem — an <see cref="RpcException"/>
///         (Unavailable / NotFound / deadline), a zero-length stream, or a hash mismatch — is treated
///         uniformly as "this endpoint did not supply valid bytes," logged, and the next endpoint is
///         tried. A hash mismatch is tried-next rather than fatal because a partitioned central could
///         legitimately serve stale bytes from one node and current bytes from another; when every
///         central shares one SQL store (the common case) the retry is merely cheap and harmless.
///     </para>
///     <para>
///         <b>Why <see langword="null"/>, not throw.</b> The caller treats <see langword="null"/>
///         identically to today's "reconcile returned null": an apply failure that keeps
///         last-known-good and does not advance the revision. A typed "no bytes" keeps the #485
///         contract explicit and cannot escape as an unhandled actor message.
///     </para>
/// </remarks>
public sealed class GrpcDeploymentArtifactFetcher : IDeploymentArtifactFetcher, IDisposable
{
    private readonly ConfigSourceOptions _options;
    private readonly ILogger<GrpcDeploymentArtifactFetcher> _log;
    private readonly Func<string, DeploymentArtifactService.DeploymentArtifactServiceClient> _clientFactory;
    private readonly ConcurrentDictionary<string, GrpcChannel> _channels = new(StringComparer.Ordinal);

    /// <summary>Initializes a new instance of the <see cref="GrpcDeploymentArtifactFetcher"/> class.</summary>
    /// <param name="options">Config-source options; supplies the endpoints, shared key, and deadline.</param>
    /// <param name="log">Logger.</param>
    /// <param name="clientFactory">
    ///     Seam for tests: builds the generated client for an endpoint. Defaults to a real h2c
    ///     <see cref="GrpcChannel"/> (one cached per endpoint).
    /// </param>
    public GrpcDeploymentArtifactFetcher(
        IOptions<ConfigSourceOptions> options,
        ILogger<GrpcDeploymentArtifactFetcher> log,
        Func<string, DeploymentArtifactService.DeploymentArtifactServiceClient>? clientFactory = null)
    {
        ArgumentNullException.ThrowIfNull(options);
        ArgumentNullException.ThrowIfNull(log);

        _options = options.Value;
        _log = log;
        _clientFactory = clientFactory ?? DefaultClientFactory;
    }

    /// <inheritdoc />
    public async Task<byte[]?> FetchAsync(string deploymentId, string expectedRevisionHash, CancellationToken ct)
    {
        ArgumentException.ThrowIfNullOrEmpty(deploymentId);
        ArgumentException.ThrowIfNullOrEmpty(expectedRevisionHash);

        var endpoints = _options.CentralFetchEndpoints;
        if (endpoints.Length == 0)
        {
            _log.LogWarning(
                "Artifact fetch for {DeploymentId} skipped: no ConfigSource:CentralFetchEndpoints configured.",
                deploymentId);
            return null;
        }

        foreach (var endpoint in endpoints)
        {
            try
            {
                var bytes = await FetchFromEndpointAsync(endpoint, deploymentId, ct);
                if (bytes.Length == 0)
                {
                    _log.LogWarning(
                        "Artifact fetch for {DeploymentId} from {Endpoint} returned zero bytes; trying next endpoint.",
                        deploymentId, endpoint);
                    continue;
                }

                var actualHash = Convert.ToHexStringLower(SHA256.HashData(bytes));
                if (!string.Equals(actualHash, expectedRevisionHash, StringComparison.OrdinalIgnoreCase))
                {
                    _log.LogWarning(
                        "Artifact fetch for {DeploymentId} from {Endpoint} failed SHA-256 verification " +
                        "(expected {Expected}, got {Actual}); trying next endpoint.",
                        deploymentId, endpoint, expectedRevisionHash, actualHash);
                    continue;
                }

                _log.LogInformation(
                    "Fetched + verified artifact for {DeploymentId} from {Endpoint} ({Bytes} bytes).",
                    deploymentId, endpoint, bytes.Length);
                return bytes;
            }
            catch (RpcException ex)
            {
                _log.LogWarning(
                    "Artifact fetch for {DeploymentId} from {Endpoint} failed ({Status}); trying next endpoint.",
                    deploymentId, endpoint, ex.StatusCode);
            }
        }

        _log.LogWarning(
            "Artifact fetch for {DeploymentId} failed on ALL {Count} endpoint(s); returning no bytes " +
            "(apply failure — last-known-good is kept).",
            deploymentId, endpoints.Length);
        return null;
    }

    private async Task<byte[]> FetchFromEndpointAsync(string endpoint, string deploymentId, CancellationToken ct)
    {
        var client = _clientFactory(endpoint);
        var headers = new Metadata { { "authorization", $"Bearer {_options.ApiKey}" } };

        using var deadlineCts = CancellationTokenSource.CreateLinkedTokenSource(ct);
        if (_options.FetchTimeoutSeconds > 0)
            deadlineCts.CancelAfter(TimeSpan.FromSeconds(_options.FetchTimeoutSeconds));

        using var call = client.Fetch(
            new FetchRequest { DeploymentId = deploymentId }, headers, cancellationToken: deadlineCts.Token);

        using var buffer = new MemoryStream();
        await foreach (var chunk in call.ResponseStream.ReadAllAsync(deadlineCts.Token))
            chunk.Data.WriteTo(buffer);

        return buffer.ToArray();
    }

    private DeploymentArtifactService.DeploymentArtifactServiceClient DefaultClientFactory(string endpoint)
    {
        // h2c: GrpcChannel over an http:// address negotiates prior-knowledge HTTP/2. One channel per
        // endpoint, cached and reused (creating a channel per call would leak sockets).
        var channel = _channels.GetOrAdd(endpoint, ep => GrpcChannel.ForAddress(ep));
        return new DeploymentArtifactService.DeploymentArtifactServiceClient(channel);
    }

    /// <inheritdoc />
    public void Dispose()
    {
        foreach (var channel in _channels.Values)
            channel.Dispose();
        _channels.Clear();
    }
}