using System.Diagnostics;
using Grpc.Core;
using Google.Protobuf.WellKnownTypes;
using ZB.MOM.WW.MxGateway.Contracts;
using ZB.MOM.WW.MxGateway.Contracts.Proto;
using ZB.MOM.WW.MxGateway.Server.Alarms;
using ZB.MOM.WW.MxGateway.Server.Metrics;
using ZB.MOM.WW.MxGateway.Server.Security.Authentication;
using ZB.MOM.WW.MxGateway.Server.Security.Authorization;
using ZB.MOM.WW.MxGateway.Server.Sessions;
using ZB.MOM.WW.MxGateway.Server.Workers;

namespace ZB.MOM.WW.MxGateway.Server.Grpc;

/// <summary>gRPC service implementation for MXAccess Gateway operations.</summary>
public sealed class MxAccessGatewayService(
    ISessionManager sessionManager,
    IGatewayRequestIdentityAccessor identityAccessor,
    IConstraintEnforcer constraintEnforcer,
    MxAccessGrpcRequestValidator requestValidator,
    MxAccessGrpcMapper mapper,
    IEventStreamService eventStreamService,
    GatewayMetrics metrics,
    ILogger<MxAccessGatewayService> logger,
    IGatewayAlarmService alarmService) : MxAccessGateway.MxAccessGatewayBase
{
    /// <inheritdoc />
    public override async Task<OpenSessionReply> OpenSession(
        OpenSessionRequest request,
        ServerCallContext context)
    {
        try
        {
            requestValidator.ValidateOpenSession(request);

            // The session's owner id and its dashboard-visibility tags both come from the resolved
            // API key identity, never from the request: the key is the tenant principal, so a
            // client cannot label its own session with another tenant's tag.
            ApiKeyIdentity? owner = identityAccessor.Current;
            GatewaySession session = await sessionManager
                .OpenSessionAsync(
                    SessionOpenRequest.FromContract(request),
                    ResolveClientIdentity(),
                    owner?.KeyId,
                    owner?.EffectiveConstraints.DashboardTags,
                    context.CancellationToken)
                .ConfigureAwait(false);

            OpenSessionReply reply = new()
            {
                SessionId = session.SessionId,
                BackendName = session.BackendName,
                WorkerProcessId = session.WorkerProcessId ?? 0,
                WorkerProtocolVersion = GatewayContractInfo.WorkerProtocolVersion,
                GatewayProtocolVersion = GatewayContractInfo.GatewayProtocolVersion,
                DefaultCommandTimeout = Google.Protobuf.WellKnownTypes.Duration.FromTimeSpan(session.CommandTimeout),
                ProtocolStatus = MxAccessGrpcMapper.Ok(),
            };
            reply.Capabilities.Add("unary-open-session");
            reply.Capabilities.Add("unary-close-session");
            reply.Capabilities.Add("unary-invoke");
            reply.Capabilities.Add("server-stream-events");
            reply.Capabilities.Add("bulk-subscribe-commands");
            reply.Capabilities.Add("bulk-read-commands");
            reply.Capabilities.Add("bulk-write-commands");
            reply.Capabilities.Add("unary-acknowledge-alarm");
            reply.Capabilities.Add("server-stream-active-alarms");

            return reply;
        }
        catch (Exception exception) when (exception is not RpcException)
        {
            throw MapException(exception);
        }
    }

    /// <inheritdoc />
    public override async Task<CloseSessionReply> CloseSession(
        CloseSessionRequest request,
        ServerCallContext context)
    {
        try
        {
            requestValidator.ValidateCloseSession(request);
            SessionCloseResult result = await sessionManager
                .CloseSessionAsync(request.SessionId, context.CancellationToken)
                .ConfigureAwait(false);

            return new CloseSessionReply
            {
                SessionId = result.SessionId,
                FinalState = result.FinalState,
                ProtocolStatus = MxAccessGrpcMapper.Ok(result.AlreadyClosed ? "Session was already closed." : "Session closed."),
            };
        }
        catch (Exception exception) when (exception is not RpcException)
        {
            throw MapException(exception);
        }
    }

    /// <inheritdoc />
    public override async Task<MxCommandReply> Invoke(
        MxCommandRequest request,
        ServerCallContext context)
    {
        try
        {
            requestValidator.ValidateInvoke(request);

            // PERF(followup): this resolve and the sessionManager.InvokeAsync below look the same
            // session up twice (a dictionary hit each, so measured cost is negligible). Collapsing
            // them needs a SessionManager overload taking an already-resolved GatewaySession, which
            // would duplicate InvokeAsync's fault mapping (SessionNotFound / state checks / metrics)
            // at a second entry point — deliberately not worth it until a profile says otherwise.
            GatewaySession session = ResolveSession(request.SessionId);
            MxCommand command = request.Command;
            BulkConstraintPlan? bulkConstraintPlan = await ApplyConstraintsAsync(
                    session,
                    command,
                    request.ClientCorrelationId,
                    context.CancellationToken)
                .ConfigureAwait(false);

            MxCommand commandToInvoke = bulkConstraintPlan?.Command ?? command;
            if (bulkConstraintPlan is { HasAllowedItems: false })
            {
                return bulkConstraintPlan.CreateDeniedReply(request);
            }

            // Map from the command alone: cloning the whole request only to overwrite its command with
            // commandToInvoke deep-cloned the (potentially large) original payload for nothing, since
            // MapCommand reads nothing but the command. The one clone that matters still
            // happens inside MapCommand, which is what keeps the worker-bound graph unaliased from
            // commandToInvoke — the caller still reads it below via TrackCommandReply.
            WorkerCommand workerCommand = mapper.MapCommand(commandToInvoke);
            WorkerCommandReply workerReply = await sessionManager
                .InvokeAsync(request.SessionId, workerCommand, context.CancellationToken)
                .ConfigureAwait(false);

            MxCommandReply publicReply = mapper.MapCommandReply(workerReply);
            if (bulkConstraintPlan is not null)
            {
                publicReply = bulkConstraintPlan.MergeDeniedInto(publicReply);
            }

            session.TrackCommandReply(commandToInvoke, publicReply);
            return publicReply;
        }
        catch (Exception exception) when (exception is not RpcException)
        {
            throw MapException(exception);
        }
    }

    /// <inheritdoc />
    public override async Task StreamEvents(
        StreamEventsRequest request,
        IServerStreamWriter<MxEvent> responseStream,
        ServerCallContext context)
    {
        try
        {
            requestValidator.ValidateStreamEvents(request);
            await foreach (MxEvent publicEvent in eventStreamService
                .StreamEventsAsync(request, identityAccessor.Current?.KeyId, context.CancellationToken)
                .WithCancellation(context.CancellationToken)
                .ConfigureAwait(false))
            {
                // Measure send latency with the allocation-free timestamp API rather
                // than allocating a Stopwatch object per event per subscriber on the highest-
                // volume gateway path. Stopwatch.GetTimestamp/GetElapsedTime measure the exact
                // same wall-clock span the former Stopwatch.StartNew()/.Elapsed did.
                long sendStartTimestamp = Stopwatch.GetTimestamp();
                await responseStream.WriteAsync(publicEvent).ConfigureAwait(false);
                metrics.RecordEventStreamSend(
                    publicEvent.Family.ToString(),
                    Stopwatch.GetElapsedTime(sendStartTimestamp));
            }
        }
        catch (Exception exception) when (exception is not RpcException)
        {
            throw MapException(exception);
        }
    }

    /// <inheritdoc />
    public override async Task<AcknowledgeAlarmReply> AcknowledgeAlarm(
        AcknowledgeAlarmRequest request,
        ServerCallContext context)
    {
        try
        {
            ArgumentNullException.ThrowIfNull(request);
            if (string.IsNullOrEmpty(request.AlarmFullReference))
            {
                throw new RpcException(new Status(StatusCode.InvalidArgument, "alarm_full_reference is required."));
            }

            if (string.IsNullOrWhiteSpace(request.OperatorUser))
            {
                throw new RpcException(new Status(StatusCode.InvalidArgument, "operator_user is required."));
            }

            return await alarmService.AcknowledgeAsync(request, context.CancellationToken)
                .ConfigureAwait(false);
        }
        catch (Exception exception) when (exception is not RpcException)
        {
            throw MapException(exception);
        }
    }

    /// <inheritdoc />
    public override async Task StreamAlarms(
        StreamAlarmsRequest request,
        IServerStreamWriter<AlarmFeedMessage> responseStream,
        ServerCallContext context)
    {
        try
        {
            ArgumentNullException.ThrowIfNull(request);
            await foreach (AlarmFeedMessage message in alarmService
                .StreamAsync(request.AlarmFilterPrefix, context.CancellationToken)
                .WithCancellation(context.CancellationToken)
                .ConfigureAwait(false))
            {
                await responseStream.WriteAsync(message).ConfigureAwait(false);
            }
        }
        catch (Exception exception) when (exception is not RpcException)
        {
            throw MapException(exception);
        }
    }

    /// <inheritdoc />
    public override async Task QueryActiveAlarms(
        QueryActiveAlarmsRequest request,
        IServerStreamWriter<ActiveAlarmSnapshot> responseStream,
        ServerCallContext context)
    {
        try
        {
            ArgumentNullException.ThrowIfNull(request);
            string? filter = string.IsNullOrEmpty(request.AlarmFilterPrefix)
                ? null
                : request.AlarmFilterPrefix;
            foreach (ActiveAlarmSnapshot snapshot in alarmService.CurrentAlarms)
            {
                context.CancellationToken.ThrowIfCancellationRequested();
                if (filter is not null
                    && !snapshot.AlarmFullReference.StartsWith(filter, StringComparison.Ordinal))
                {
                    continue;
                }

                await responseStream.WriteAsync(snapshot).ConfigureAwait(false);
            }
        }
        catch (Exception exception) when (exception is not RpcException)
        {
            throw MapException(exception);
        }
    }

    private string? ResolveClientIdentity()
    {
        return identityAccessor.Current?.DisplayName ?? identityAccessor.Current?.KeyId;
    }

    private GatewaySession ResolveSession(string sessionId)
    {
        if (!sessionManager.TryGetSession(sessionId, out GatewaySession? session) || session is null)
        {
            throw new SessionManagerException(
                SessionManagerErrorCode.SessionNotFound,
                $"Session {sessionId} was not found.");
        }

        return session;
    }

    private async Task<BulkConstraintPlan?> ApplyConstraintsAsync(
        GatewaySession session,
        MxCommand command,
        string? correlationId,
        CancellationToken cancellationToken)
    {
        ApiKeyIdentity? identity = identityAccessor.Current;
        switch (command.Kind)
        {
            case MxCommandKind.AddItem:
                await EnforceReadTagAsync(identity, command.Kind, command.AddItem.ItemDefinition, correlationId, cancellationToken)
                    .ConfigureAwait(false);
                return null;
            case MxCommandKind.AddItem2:
                await EnforceReadTagAsync(identity, command.Kind, command.AddItem2.ItemDefinition, correlationId, cancellationToken)
                    .ConfigureAwait(false);
                return null;
            case MxCommandKind.AddItemBulk:
                return await FilterTagBulkAsync(
                        identity,
                        command,
                        command.AddItemBulk.ServerHandle,
                        command.AddItemBulk.TagAddresses,
                        correlationId,
                        cancellationToken)
                    .ConfigureAwait(false);
            case MxCommandKind.SubscribeBulk:
                return await FilterTagBulkAsync(
                        identity,
                        command,
                        command.SubscribeBulk.ServerHandle,
                        command.SubscribeBulk.TagAddresses,
                        correlationId,
                        cancellationToken)
                    .ConfigureAwait(false);
            case MxCommandKind.AdviseItemBulk:
                return await FilterHandleBulkAsync(
                        identity,
                        session,
                        command,
                        command.AdviseItemBulk.ServerHandle,
                        command.AdviseItemBulk.ItemHandles,
                        correlationId,
                        cancellationToken)
                    .ConfigureAwait(false);
            case MxCommandKind.ReadBulk:
                return await FilterReadBulkAsync(
                        identity,
                        command,
                        command.ReadBulk.ServerHandle,
                        command.ReadBulk.TagAddresses,
                        correlationId,
                        cancellationToken)
                    .ConfigureAwait(false);
            case MxCommandKind.WriteBulk:
                return await FilterWriteBulkAsync(
                        identity,
                        session,
                        command,
                        command.WriteBulk.ServerHandle,
                        command.WriteBulk.Entries,
                        entry => entry.ItemHandle,
                        correlationId,
                        cancellationToken)
                    .ConfigureAwait(false);
            case MxCommandKind.Write2Bulk:
                return await FilterWriteBulkAsync(
                        identity,
                        session,
                        command,
                        command.Write2Bulk.ServerHandle,
                        command.Write2Bulk.Entries,
                        entry => entry.ItemHandle,
                        correlationId,
                        cancellationToken)
                    .ConfigureAwait(false);
            case MxCommandKind.WriteSecuredBulk:
                return await FilterWriteBulkAsync(
                        identity,
                        session,
                        command,
                        command.WriteSecuredBulk.ServerHandle,
                        command.WriteSecuredBulk.Entries,
                        entry => entry.ItemHandle,
                        correlationId,
                        cancellationToken)
                    .ConfigureAwait(false);
            case MxCommandKind.WriteSecured2Bulk:
                return await FilterWriteBulkAsync(
                        identity,
                        session,
                        command,
                        command.WriteSecured2Bulk.ServerHandle,
                        command.WriteSecured2Bulk.Entries,
                        entry => entry.ItemHandle,
                        correlationId,
                        cancellationToken)
                    .ConfigureAwait(false);
            case MxCommandKind.Write:
                await EnforceWriteHandleAsync(
                        identity,
                        session,
                        command.Kind,
                        command.Write.ServerHandle,
                        command.Write.ItemHandle,
                        correlationId,
                        cancellationToken)
                    .ConfigureAwait(false);
                return null;
            case MxCommandKind.Write2:
                await EnforceWriteHandleAsync(
                        identity,
                        session,
                        command.Kind,
                        command.Write2.ServerHandle,
                        command.Write2.ItemHandle,
                        correlationId,
                        cancellationToken)
                    .ConfigureAwait(false);
                return null;
            case MxCommandKind.WriteSecured:
                await EnforceWriteHandleAsync(
                        identity,
                        session,
                        command.Kind,
                        command.WriteSecured.ServerHandle,
                        command.WriteSecured.ItemHandle,
                        correlationId,
                        cancellationToken)
                    .ConfigureAwait(false);
                return null;
            case MxCommandKind.WriteSecured2:
                await EnforceWriteHandleAsync(
                        identity,
                        session,
                        command.Kind,
                        command.WriteSecured2.ServerHandle,
                        command.WriteSecured2.ItemHandle,
                        correlationId,
                        cancellationToken)
                    .ConfigureAwait(false);
                return null;
            default:
                return null;
        }
    }

    private async Task EnforceReadTagAsync(
        ApiKeyIdentity? identity,
        MxCommandKind commandKind,
        string tagAddress,
        string? correlationId,
        CancellationToken cancellationToken)
    {
        ConstraintFailure? failure = await constraintEnforcer
            .CheckReadTagAsync(identity, tagAddress, cancellationToken)
            .ConfigureAwait(false);
        if (failure is null)
        {
            return;
        }

        await constraintEnforcer.RecordDenialAsync(identity, commandKind.ToString(), tagAddress, failure, correlationId, cancellationToken)
            .ConfigureAwait(false);
        throw new RpcException(new Status(StatusCode.PermissionDenied, failure.Message));
    }

    private async Task EnforceWriteHandleAsync(
        ApiKeyIdentity? identity,
        GatewaySession session,
        MxCommandKind commandKind,
        int serverHandle,
        int itemHandle,
        string? correlationId,
        CancellationToken cancellationToken)
    {
        ConstraintFailure? failure = await constraintEnforcer
            .CheckWriteHandleAsync(identity, session, serverHandle, itemHandle, cancellationToken)
            .ConfigureAwait(false);
        if (failure is null)
        {
            return;
        }

        await constraintEnforcer.RecordDenialAsync(identity, commandKind.ToString(), itemHandle.ToString(System.Globalization.CultureInfo.InvariantCulture), failure, correlationId, cancellationToken)
            .ConfigureAwait(false);
        throw new RpcException(new Status(StatusCode.PermissionDenied, failure.Message));
    }

    private async Task<BulkConstraintPlan?> FilterTagBulkAsync(
        ApiKeyIdentity? identity,
        MxCommand command,
        int serverHandle,
        IReadOnlyList<string> tagAddresses,
        string? correlationId,
        CancellationToken cancellationToken)
    {
        // An identity with no read constraints allows every tag, so the per-item enforcer call below
        // can only answer "allowed" — the whole loop (and the plan it would build) is dead work.
        // Returning null is exactly what the denied.Count == 0 exit below returns.
        if (!constraintEnforcer.HasReadConstraints(identity))
        {
            return null;
        }

        Dictionary<int, SubscribeResult> denied = [];
        List<string> allowed = [];
        for (int index = 0; index < tagAddresses.Count; index++)
        {
            string tagAddress = tagAddresses[index];
            ConstraintFailure? failure = await constraintEnforcer
                .CheckReadTagAsync(identity, tagAddress, cancellationToken)
                .ConfigureAwait(false);
            if (failure is null)
            {
                allowed.Add(tagAddress);
                continue;
            }

            await constraintEnforcer.RecordDenialAsync(identity, command.Kind.ToString(), tagAddress, failure, correlationId, cancellationToken)
                .ConfigureAwait(false);
            denied[index] = new SubscribeResult
            {
                ServerHandle = serverHandle,
                TagAddress = tagAddress,
                WasSuccessful = false,
                ErrorMessage = failure.Message,
            };
        }

        if (denied.Count == 0)
        {
            return null;
        }

        // Build the filtered command directly instead of cloning the original and clearing it:
        // the clone deep-copied every denied address only to drop it. The payload's other fields
        // (server_handle) are copied across explicitly. Nothing aliases the request here — these
        // bulk payloads carry only strings — and the worker-bound graph is still the unaliased copy
        // MapCommand makes.
        MxCommand filtered = new() { Kind = command.Kind };
        if (command.Kind == MxCommandKind.AddItemBulk)
        {
            AddItemBulkCommand payload = new() { ServerHandle = command.AddItemBulk.ServerHandle };
            payload.TagAddresses.Add(allowed);
            filtered.AddItemBulk = payload;
        }
        else
        {
            SubscribeBulkCommand payload = new() { ServerHandle = command.SubscribeBulk.ServerHandle };
            payload.TagAddresses.Add(allowed);
            filtered.SubscribeBulk = payload;
        }

        return new SubscribeBulkConstraintPlan(filtered, tagAddresses.Count, denied, allowed.Count > 0);
    }

    private async Task<BulkConstraintPlan?> FilterReadBulkAsync(
        ApiKeyIdentity? identity,
        MxCommand command,
        int serverHandle,
        IReadOnlyList<string> tagAddresses,
        string? correlationId,
        CancellationToken cancellationToken)
    {
        // Mirrors FilterTagBulkAsync but produces BulkReadResult denial entries
        // so the reply payload merges into BulkReadReply.Results, not
        // BulkSubscribeReply.Results.
        if (!constraintEnforcer.HasReadConstraints(identity))
        {
            return null;
        }

        Dictionary<int, BulkReadResult> denied = [];
        List<string> allowed = [];
        for (int index = 0; index < tagAddresses.Count; index++)
        {
            string tagAddress = tagAddresses[index];
            ConstraintFailure? failure = await constraintEnforcer
                .CheckReadTagAsync(identity, tagAddress, cancellationToken)
                .ConfigureAwait(false);
            if (failure is null)
            {
                allowed.Add(tagAddress);
                continue;
            }

            await constraintEnforcer.RecordDenialAsync(identity, command.Kind.ToString(), tagAddress, failure, correlationId, cancellationToken)
                .ConfigureAwait(false);
            denied[index] = new BulkReadResult
            {
                ServerHandle = serverHandle,
                TagAddress = tagAddress,
                WasSuccessful = false,
                WasCached = false,
                ErrorMessage = failure.Message,
            };
        }

        if (denied.Count == 0)
        {
            return null;
        }

        MxCommand filtered = new() { Kind = command.Kind };
        ReadBulkCommand payload = new()
        {
            ServerHandle = command.ReadBulk.ServerHandle,
            TimeoutMs = command.ReadBulk.TimeoutMs,
        };
        payload.TagAddresses.Add(allowed);
        filtered.ReadBulk = payload;

        return new ReadBulkConstraintPlan(filtered, tagAddresses.Count, denied, allowed.Count > 0);
    }

    private async Task<BulkConstraintPlan?> FilterWriteBulkAsync<TEntry>(
        ApiKeyIdentity? identity,
        GatewaySession session,
        MxCommand command,
        int serverHandle,
        Google.Protobuf.Collections.RepeatedField<TEntry> entries,
        Func<TEntry, int> getItemHandle,
        string? correlationId,
        CancellationToken cancellationToken) where TEntry : class
    {
        // The four bulk-write families each carry a different per-entry message
        // shape (WriteBulkEntry / Write2BulkEntry / WriteSecuredBulkEntry /
        // WriteSecured2BulkEntry), but the constraint check itself is identical
        // — "is this caller allowed to write to this server+item handle?".
        // Parameterising on TEntry + getItemHandle keeps a single filter
        // routine for all four and avoids duplicating CheckWriteHandleAsync
        // calls.
        if (!constraintEnforcer.HasWriteConstraints(identity))
        {
            return null;
        }

        Dictionary<int, BulkWriteResult> denied = [];
        List<TEntry> allowed = [];
        for (int index = 0; index < entries.Count; index++)
        {
            TEntry entry = entries[index];
            int itemHandle = getItemHandle(entry);
            ConstraintFailure? failure = await constraintEnforcer
                .CheckWriteHandleAsync(identity, session, serverHandle, itemHandle, cancellationToken)
                .ConfigureAwait(false);
            if (failure is null)
            {
                allowed.Add(entry);
                continue;
            }

            await constraintEnforcer.RecordDenialAsync(
                    identity,
                    command.Kind.ToString(),
                    itemHandle.ToString(System.Globalization.CultureInfo.InvariantCulture),
                    failure,
                    correlationId,
                    cancellationToken)
                .ConfigureAwait(false);
            denied[index] = new BulkWriteResult
            {
                ServerHandle = serverHandle,
                ItemHandle = itemHandle,
                WasSuccessful = false,
                ErrorMessage = failure.Message,
            };
        }

        if (denied.Count == 0)
        {
            return null;
        }

        return new WriteBulkConstraintPlan(
            BuildFilteredWriteBulkCommand(command, allowed),
            entries.Count,
            denied,
            allowed.Count > 0);
    }

    /// <summary>
    ///     Builds the allowed-only bulk-write command. The allowed entries are carried over by
    ///     reference rather than deep-cloned: the caller only reads this command (TrackCommandReply),
    ///     and the copy the worker mutates and owns is the one <c>MapCommand</c> clones — the same
    ///     no-aliasing boundary as before. Cloning the whole command here and clearing it copied
    ///     every denied entry's payload (including <c>WriteSecured</c> values) for nothing.
    /// </summary>
    /// <typeparam name="TEntry">The per-family bulk-write entry message type.</typeparam>
    /// <param name="command">The original command, read for its kind and payload scalars.</param>
    /// <param name="allowed">The entries that survived constraint filtering, in original order.</param>
    /// <returns>A command of the same kind carrying only the allowed entries.</returns>
    private static MxCommand BuildFilteredWriteBulkCommand<TEntry>(MxCommand command, IReadOnlyList<TEntry> allowed)
        where TEntry : class
    {
        MxCommand filtered = new() { Kind = command.Kind };
        switch (command.Kind)
        {
            case MxCommandKind.WriteBulk:
            {
                WriteBulkCommand payload = new() { ServerHandle = command.WriteBulk.ServerHandle };
                payload.Entries.Add((IEnumerable<WriteBulkEntry>)allowed);
                filtered.WriteBulk = payload;
                break;
            }

            case MxCommandKind.Write2Bulk:
            {
                Write2BulkCommand payload = new() { ServerHandle = command.Write2Bulk.ServerHandle };
                payload.Entries.Add((IEnumerable<Write2BulkEntry>)allowed);
                filtered.Write2Bulk = payload;
                break;
            }

            case MxCommandKind.WriteSecuredBulk:
            {
                WriteSecuredBulkCommand payload = new() { ServerHandle = command.WriteSecuredBulk.ServerHandle };
                payload.Entries.Add((IEnumerable<WriteSecuredBulkEntry>)allowed);
                filtered.WriteSecuredBulk = payload;
                break;
            }

            case MxCommandKind.WriteSecured2Bulk:
            {
                WriteSecured2BulkCommand payload = new() { ServerHandle = command.WriteSecured2Bulk.ServerHandle };
                payload.Entries.Add((IEnumerable<WriteSecured2BulkEntry>)allowed);
                filtered.WriteSecured2Bulk = payload;
                break;
            }

            default:
                // Only the four bulk-write kinds above reach FilterWriteBulkAsync, so this is
                // unreachable. It throws rather than falling back to the unmodified command,
                // because that fallback failed OPEN: a fifth bulk-write kind added upstream
                // without a case here would silently ship the DENIED entries to the worker while
                // still reporting them denied to the caller. Failing loud on a kind nobody can
                // reach today is strictly safer than a constraint bypass nobody would notice.
                throw new UnreachableException(
                    $"Command kind {command.Kind} reached bulk-write constraint filtering without a filter case.");
        }

        return filtered;
    }

    private async Task<BulkConstraintPlan?> FilterHandleBulkAsync(
        ApiKeyIdentity? identity,
        GatewaySession session,
        MxCommand command,
        int serverHandle,
        IReadOnlyList<int> itemHandles,
        string? correlationId,
        CancellationToken cancellationToken)
    {
        if (!constraintEnforcer.HasReadConstraints(identity))
        {
            return null;
        }

        Dictionary<int, SubscribeResult> denied = [];
        List<int> allowed = [];
        for (int index = 0; index < itemHandles.Count; index++)
        {
            int itemHandle = itemHandles[index];
            ConstraintFailure? failure = await constraintEnforcer
                .CheckReadHandleAsync(identity, session, serverHandle, itemHandle, cancellationToken)
                .ConfigureAwait(false);
            if (failure is null)
            {
                allowed.Add(itemHandle);
                continue;
            }

            await constraintEnforcer.RecordDenialAsync(identity, command.Kind.ToString(), itemHandle.ToString(System.Globalization.CultureInfo.InvariantCulture), failure, correlationId, cancellationToken)
                .ConfigureAwait(false);
            denied[index] = new SubscribeResult
            {
                ServerHandle = serverHandle,
                ItemHandle = itemHandle,
                WasSuccessful = false,
                ErrorMessage = failure.Message,
            };
        }

        if (denied.Count == 0)
        {
            return null;
        }

        MxCommand filtered = new() { Kind = command.Kind };
        AdviseItemBulkCommand payload = new() { ServerHandle = command.AdviseItemBulk.ServerHandle };
        payload.ItemHandles.Add(allowed);
        filtered.AdviseItemBulk = payload;

        return new SubscribeBulkConstraintPlan(filtered, itemHandles.Count, denied, allowed.Count > 0);
    }

    /// <summary>
    ///     Polymorphic constraint plan returned from <see cref="ApplyConstraintsAsync"/>.
    ///     Each concrete subtype is keyed to a specific bulk-reply shape — the
    ///     SubscribeResult-based AddItem/Advise/Subscribe family, the
    ///     BulkWriteResult-based Write* bulk family, and the BulkReadResult-based
    ///     ReadBulk command. Subtypes own their own merge / denied-reply build
    ///     logic so the Invoke dispatch site never branches on reply shape.
    /// </summary>
    private abstract record BulkConstraintPlan(
        MxCommand Command,
        int OriginalCount,
        bool HasAllowedItems)
    {
        /// <summary>Builds a reply containing only the denied entries (used when no items survived filtering).</summary>
        /// <param name="request">The original command request.</param>
        /// <returns>A reply with denied results only.</returns>
        public abstract MxCommandReply CreateDeniedReply(MxCommandRequest request);

        /// <summary>Splices denied entries back into the worker's allowed-only reply in original-index order.</summary>
        /// <param name="reply">The worker's reply containing only allowed results.</param>
        /// <returns>A merged reply in the original order.</returns>
        public abstract MxCommandReply MergeDeniedInto(MxCommandReply reply);
    }

    private sealed record SubscribeBulkConstraintPlan(
        MxCommand Command,
        int OriginalCount,
        IReadOnlyDictionary<int, SubscribeResult> DeniedResults,
        bool HasAllowedItems)
        : BulkConstraintPlan(Command, OriginalCount, HasAllowedItems)
    {
        /// <inheritdoc />
        public override MxCommandReply CreateDeniedReply(MxCommandRequest request)
        {
            MxCommandReply reply = new()
            {
                SessionId = request.SessionId,
                CorrelationId = request.ClientCorrelationId,
                Kind = request.Command.Kind,
                ProtocolStatus = MxAccessGrpcMapper.Ok(),
            };
            SetPayload(reply, BuildMerged(new BulkSubscribeReply()));
            return reply;
        }

        /// <inheritdoc />
        public override MxCommandReply MergeDeniedInto(MxCommandReply reply)
        {
            BulkSubscribeReply allowed = GetPayload(reply) ?? new BulkSubscribeReply();
            SetPayload(reply, BuildMerged(allowed));
            return reply;
        }

        private BulkSubscribeReply BuildMerged(BulkSubscribeReply allowed)
        {
            Queue<SubscribeResult> allowedResults = new(allowed.Results);
            BulkSubscribeReply merged = new();
            for (int index = 0; index < OriginalCount; index++)
            {
                if (DeniedResults.TryGetValue(index, out SubscribeResult? denied))
                {
                    merged.Results.Add(denied);
                }
                else if (allowedResults.TryDequeue(out SubscribeResult? allowedResult))
                {
                    merged.Results.Add(allowedResult);
                }
            }

            return merged;
        }

        private BulkSubscribeReply? GetPayload(MxCommandReply reply) => Command.Kind switch
        {
            MxCommandKind.AddItemBulk => reply.AddItemBulk,
            MxCommandKind.AdviseItemBulk => reply.AdviseItemBulk,
            MxCommandKind.SubscribeBulk => reply.SubscribeBulk,
            _ => null,
        };

        private void SetPayload(MxCommandReply reply, BulkSubscribeReply payload)
        {
            switch (Command.Kind)
            {
                case MxCommandKind.AddItemBulk:
                    reply.AddItemBulk = payload;
                    break;
                case MxCommandKind.AdviseItemBulk:
                    reply.AdviseItemBulk = payload;
                    break;
                case MxCommandKind.SubscribeBulk:
                    reply.SubscribeBulk = payload;
                    break;
            }
        }
    }

    private sealed record WriteBulkConstraintPlan(
        MxCommand Command,
        int OriginalCount,
        IReadOnlyDictionary<int, BulkWriteResult> DeniedResults,
        bool HasAllowedItems)
        : BulkConstraintPlan(Command, OriginalCount, HasAllowedItems)
    {
        /// <inheritdoc />
        public override MxCommandReply CreateDeniedReply(MxCommandRequest request)
        {
            MxCommandReply reply = new()
            {
                SessionId = request.SessionId,
                CorrelationId = request.ClientCorrelationId,
                Kind = request.Command.Kind,
                ProtocolStatus = MxAccessGrpcMapper.Ok(),
            };
            SetPayload(reply, BuildMerged(new BulkWriteReply()));
            return reply;
        }

        /// <inheritdoc />
        public override MxCommandReply MergeDeniedInto(MxCommandReply reply)
        {
            BulkWriteReply allowed = GetPayload(reply) ?? new BulkWriteReply();
            SetPayload(reply, BuildMerged(allowed));
            return reply;
        }

        private BulkWriteReply BuildMerged(BulkWriteReply allowed)
        {
            Queue<BulkWriteResult> allowedResults = new(allowed.Results);
            BulkWriteReply merged = new();
            for (int index = 0; index < OriginalCount; index++)
            {
                if (DeniedResults.TryGetValue(index, out BulkWriteResult? denied))
                {
                    merged.Results.Add(denied);
                }
                else if (allowedResults.TryDequeue(out BulkWriteResult? allowedResult))
                {
                    merged.Results.Add(allowedResult);
                }
            }

            return merged;
        }

        private BulkWriteReply? GetPayload(MxCommandReply reply) => Command.Kind switch
        {
            MxCommandKind.WriteBulk => reply.WriteBulk,
            MxCommandKind.Write2Bulk => reply.Write2Bulk,
            MxCommandKind.WriteSecuredBulk => reply.WriteSecuredBulk,
            MxCommandKind.WriteSecured2Bulk => reply.WriteSecured2Bulk,
            _ => null,
        };

        private void SetPayload(MxCommandReply reply, BulkWriteReply payload)
        {
            switch (Command.Kind)
            {
                case MxCommandKind.WriteBulk:
                    reply.WriteBulk = payload;
                    break;
                case MxCommandKind.Write2Bulk:
                    reply.Write2Bulk = payload;
                    break;
                case MxCommandKind.WriteSecuredBulk:
                    reply.WriteSecuredBulk = payload;
                    break;
                case MxCommandKind.WriteSecured2Bulk:
                    reply.WriteSecured2Bulk = payload;
                    break;
            }
        }
    }

    private sealed record ReadBulkConstraintPlan(
        MxCommand Command,
        int OriginalCount,
        IReadOnlyDictionary<int, BulkReadResult> DeniedResults,
        bool HasAllowedItems)
        : BulkConstraintPlan(Command, OriginalCount, HasAllowedItems)
    {
        /// <inheritdoc />
        public override MxCommandReply CreateDeniedReply(MxCommandRequest request)
        {
            MxCommandReply reply = new()
            {
                SessionId = request.SessionId,
                CorrelationId = request.ClientCorrelationId,
                Kind = request.Command.Kind,
                ProtocolStatus = MxAccessGrpcMapper.Ok(),
            };
            reply.ReadBulk = BuildMerged(new BulkReadReply());
            return reply;
        }

        /// <inheritdoc />
        public override MxCommandReply MergeDeniedInto(MxCommandReply reply)
        {
            BulkReadReply allowed = reply.ReadBulk ?? new BulkReadReply();
            reply.ReadBulk = BuildMerged(allowed);
            return reply;
        }

        private BulkReadReply BuildMerged(BulkReadReply allowed)
        {
            Queue<BulkReadResult> allowedResults = new(allowed.Results);
            BulkReadReply merged = new();
            for (int index = 0; index < OriginalCount; index++)
            {
                if (DeniedResults.TryGetValue(index, out BulkReadResult? denied))
                {
                    merged.Results.Add(denied);
                }
                else if (allowedResults.TryDequeue(out BulkReadResult? allowedResult))
                {
                    merged.Results.Add(allowedResult);
                }
            }

            return merged;
        }
    }

    private RpcException MapException(Exception exception)
    {
        if (exception is OperationCanceledException)
        {
            return new RpcException(new Status(StatusCode.Cancelled, "gRPC request was canceled."));
        }

        if (exception is SessionManagerException sessionException)
        {
            return MapSessionException(sessionException);
        }

        if (exception is WorkerClientException workerClientException)
        {
            return MapWorkerClientException(workerClientException);
        }

        logger.LogWarning(exception, "Public gRPC request failed.");
        return new RpcException(new Status(StatusCode.Unavailable, "Gateway request failed before an MXAccess reply was available."));
    }

    private static RpcException MapSessionException(SessionManagerException exception)
    {
        StatusCode statusCode = exception.ErrorCode switch
        {
            SessionManagerErrorCode.SessionNotFound => StatusCode.NotFound,
            SessionManagerErrorCode.SessionNotReady => StatusCode.FailedPrecondition,
            SessionManagerErrorCode.EventSubscriberAlreadyActive => StatusCode.ResourceExhausted,
            SessionManagerErrorCode.EventSubscriberLimitReached => StatusCode.ResourceExhausted,
            SessionManagerErrorCode.EventQueueOverflow => StatusCode.ResourceExhausted,
            SessionManagerErrorCode.SessionLimitExceeded => StatusCode.ResourceExhausted,
            SessionManagerErrorCode.OpenFailed => StatusCode.Unavailable,
            SessionManagerErrorCode.CloseFailed => StatusCode.Unavailable,
            SessionManagerErrorCode.PermissionDenied => StatusCode.PermissionDenied,
            _ => StatusCode.Unavailable,
        };

        return new RpcException(new Status(statusCode, exception.Message));
    }

    private static RpcException MapWorkerClientException(WorkerClientException exception)
    {
        StatusCode statusCode = exception.ErrorCode switch
        {
            WorkerClientErrorCode.CommandTimeout => StatusCode.DeadlineExceeded,
            WorkerClientErrorCode.GatewayShutdown => StatusCode.Cancelled,
            WorkerClientErrorCode.CallerCancelled => StatusCode.Cancelled,
            WorkerClientErrorCode.InvalidState => StatusCode.FailedPrecondition,
            WorkerClientErrorCode.ProtocolViolation => StatusCode.Internal,
            WorkerClientErrorCode.CommandTooLarge => StatusCode.ResourceExhausted,
            _ => StatusCode.Unavailable,
        };

        return new RpcException(new Status(statusCode, exception.Message));
    }
}