lmxopcua/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Grpc/DeploymentArtifactService.cs

using Google.Protobuf;
using Grpc.Core;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Logging;
using ZB.MOM.WW.OtOpcUa.Commons.Protos.DeploymentArtifact.V1;
using ZB.MOM.WW.OtOpcUa.Configuration;
// The generated service container is itself named DeploymentArtifactService; alias its nested server
// base so this impl can keep the natural name without colliding with the generated type.
using GeneratedServiceBase =
    ZB.MOM.WW.OtOpcUa.Commons.Protos.DeploymentArtifact.V1.DeploymentArtifactService.DeploymentArtifactServiceBase;

namespace ZB.MOM.WW.OtOpcUa.AdminUI.Grpc;

/// <summary>
///     Central-side artifact serve (per-cluster mesh Phase 3). Streams a deployment's
///     <c>ArtifactBlob</c> to a driver node fetching in <c>FetchAndCache</c> mode. Registered only on
///     admin-role nodes (which hold the central SQL connection) and only when
///     <c>ConfigServe:GrpcListenPort &gt; 0</c>.
/// </summary>
/// <remarks>
///     <para>
///         <b>Serves any deployment that has a non-empty blob — status is NOT gated.</b> A node
///         fetches the artifact <em>during apply</em>, before the deployment is sealed: central seals
///         only once every node has acked Applied, and a <c>FetchAndCache</c> node cannot ack until it
///         has fetched. Gating on <c>Sealed</c> would deadlock every such deploy (seal-needs-ack →
///         ack-needs-fetch → fetch-needs-sealed), and it matches Direct mode, which reads the same
///         <c>ArtifactBlob</c> from SQL while the row is still <c>AwaitingApplyAcks</c>. (Caught on the
///         Phase 3 live gate.) Access is already gated by the shared-key interceptor, so there is no
///         reason to hide a non-sealed deployment.
///     </para>
///     <para>
///         <b>NotFound is indistinguishable across two cases</b> — unknown id and an empty blob. The
///         first is existence-hiding; the second is the empty-artifact rule (a zero-length artifact is "no
///         answer," never "a configuration with no drivers"). Streaming empty bytes as if valid is
///         exactly what that defect class forbids.
///     </para>
/// </remarks>
public sealed class DeploymentArtifactGrpcService : GeneratedServiceBase
{
    /// <summary>Chunk size, matching <c>LocalDbDeploymentArtifactCache.ChunkSize</c> (128 KiB).</summary>
    private const int ChunkSize = 128 * 1024;

    private readonly IDbContextFactory<OtOpcUaConfigDbContext> _dbFactory;
    private readonly ILogger<DeploymentArtifactGrpcService> _log;

    /// <summary>Initializes a new instance of the <see cref="DeploymentArtifactGrpcService"/> class.</summary>
    /// <param name="dbFactory">Factory for the central config database holding the artifact bytes.</param>
    /// <param name="log">Logger.</param>
    public DeploymentArtifactGrpcService(
        IDbContextFactory<OtOpcUaConfigDbContext> dbFactory,
        ILogger<DeploymentArtifactGrpcService> log)
    {
        _dbFactory = dbFactory;
        _log = log;
    }

    /// <inheritdoc />
    public override async Task Fetch(
        FetchRequest request,
        IServerStreamWriter<ArtifactChunk> responseStream,
        ServerCallContext context)
    {
        ArgumentNullException.ThrowIfNull(request);
        ArgumentNullException.ThrowIfNull(responseStream);
        ArgumentNullException.ThrowIfNull(context);

        if (!Guid.TryParse(request.DeploymentId, out var id))
            throw new RpcException(new Status(StatusCode.NotFound, "unknown deployment"));

        await using var db = await _dbFactory.CreateDbContextAsync(context.CancellationToken);
        var row = await db.Deployments
            .AsNoTracking()
            .Where(d => d.DeploymentId == id)
            .Select(d => new { d.ArtifactBlob })
            .FirstOrDefaultAsync(context.CancellationToken);

        // Existence-hiding + the empty-artifact rule: unknown id / empty blob collapse to one NotFound. Status is NOT
        // gated — a node fetches during apply, before the deployment is sealed (see the class remarks).
        if (row is null || row.ArtifactBlob.Length == 0)
        {
            _log.LogDebug(
                "Artifact fetch for {DeploymentId} -> NotFound (found={Found}, bytes={Bytes})",
                request.DeploymentId, row is not null, row?.ArtifactBlob.Length ?? 0);
            throw new RpcException(new Status(StatusCode.NotFound, "unknown deployment"));
        }

        for (var offset = 0; offset < row.ArtifactBlob.Length; offset += ChunkSize)
        {
            var len = Math.Min(ChunkSize, row.ArtifactBlob.Length - offset);
            await responseStream.WriteAsync(
                new ArtifactChunk { Data = ByteString.CopyFrom(row.ArtifactBlob, offset, len) },
                context.CancellationToken);
        }

        _log.LogInformation(
            "Served artifact for {DeploymentId} ({Bytes} bytes, {Chunks} chunks)",
            request.DeploymentId, row.ArtifactBlob.Length,
            (row.ArtifactBlob.Length + ChunkSize - 1) / ChunkSize);
    }
}