mxaccessgw/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessCommandExecutor.cs

using System;
using System.Collections.Generic;
using ZB.MOM.WW.MxGateway.Contracts.Proto;
using ZB.MOM.WW.MxGateway.Worker.Conversion;
using ZB.MOM.WW.MxGateway.Worker.Sta;

namespace ZB.MOM.WW.MxGateway.Worker.MxAccess;

/// <summary>
/// Executes MXAccess commands on an STA session.
/// </summary>
public sealed class MxAccessCommandExecutor : IStaCommandExecutor
{
    /// <summary>Default per-tag timeout used when <c>ReadBulkCommand.timeout_ms</c> is zero.</summary>
    internal static readonly TimeSpan DefaultReadBulkTimeout = TimeSpan.FromMilliseconds(1000);

    /// <summary>
    ///     Default bounded wait for the OnWriteComplete callback after a
    ///     WriteSecured/WriteSecured2 COM call. 1.5 s keeps the unary reply
    ///     inside the OtOpcUa driver's 2 s Tier A write-resilience budget (a
    ///     longer gateway wait must raise that consumer timeout in step) while
    ///     covering the common fast-commit case; on expiry the reply returns
    ///     with empty statuses — unconfirmed, not failed.
    /// </summary>
    internal static readonly TimeSpan DefaultWriteCompletionTimeout = TimeSpan.FromMilliseconds(1500);

    private readonly MxAccessSession session;
    private readonly VariantConverter variantConverter;
    private readonly MxStatusProxyConverter statusProxyConverter;
    private readonly IAlarmCommandHandler? alarmCommandHandler;
    private readonly Action pumpStep;
    private readonly TimeSpan writeCompletionTimeout;

    /// <summary>
    /// Initializes a command executor with an MXAccess session.
    /// </summary>
    /// <param name="session">MXAccess session on the STA thread.</param>
    public MxAccessCommandExecutor(MxAccessSession session)
        : this(session, new VariantConverter(), alarmCommandHandler: null, pumpStep: null)
    {
    }

    /// <summary>
    /// Initializes a command executor with an MXAccess session and a variant converter.
    /// </summary>
    /// <param name="session">MXAccess session on the STA thread.</param>
    /// <param name="variantConverter">Converter for MXAccess variant values to MxValue protobuf messages.</param>
    public MxAccessCommandExecutor(
        MxAccessSession session,
        VariantConverter variantConverter)
        : this(session, variantConverter, alarmCommandHandler: null, pumpStep: null)
    {
    }

    /// <summary>
    /// Initializes a command executor with an MXAccess session, variant
    /// converter, and an alarm command handler. The alarm handler is
    /// optional — when null, alarm-side commands return an
    /// "alarm consumer not configured" diagnostic.
    /// </summary>
    /// <param name="session">MXAccess session on the STA thread.</param>
    /// <param name="variantConverter">Converter for MXAccess variant values to MxValue protobuf messages.</param>
    /// <param name="alarmCommandHandler">Optional handler for alarm-side commands.</param>
    public MxAccessCommandExecutor(
        MxAccessSession session,
        VariantConverter variantConverter,
        IAlarmCommandHandler? alarmCommandHandler)
        : this(session, variantConverter, alarmCommandHandler, pumpStep: null)
    {
    }

    /// <summary>
    /// Initializes a command executor with an MXAccess session, variant
    /// converter, alarm command handler, and a Windows-message pump action.
    /// The pump action is invoked from inside <c>ReadBulk</c>'s wait loop so
    /// MXAccess COM events queued for this STA can be dispatched while the
    /// executor is still holding the thread. Pass <c>null</c> in tests where
    /// ReadBulk is exercised against a fake worker that pre-populates the
    /// value cache — the executor falls back to a no-op pump step.
    /// </summary>
    /// <param name="session">MXAccess session on the STA thread.</param>
    /// <param name="variantConverter">Converter for MXAccess variant values to MxValue protobuf messages.</param>
    /// <param name="alarmCommandHandler">Optional handler for alarm-side commands.</param>
    /// <param name="pumpStep">Action to pump Windows messages, or null for tests.</param>
    /// <param name="writeCompletionTimeout">
    ///     Bounded wait for the OnWriteComplete callback after a
    ///     WriteSecured/WriteSecured2 COM call, or null for
    ///     <see cref="DefaultWriteCompletionTimeout"/>. Zero (or negative)
    ///     disables the wait entirely — replies keep the pure fire-and-forget
    ///     shape.
    /// </param>
    public MxAccessCommandExecutor(
        MxAccessSession session,
        VariantConverter variantConverter,
        IAlarmCommandHandler? alarmCommandHandler,
        Action? pumpStep,
        TimeSpan? writeCompletionTimeout = null)
    {
        this.session = session ?? throw new ArgumentNullException(nameof(session));
        this.variantConverter = variantConverter ?? throw new ArgumentNullException(nameof(variantConverter));
        this.statusProxyConverter = new MxStatusProxyConverter();
        this.alarmCommandHandler = alarmCommandHandler;

        // The no-op fallback is a TEST affordance only — never pass null on a
        // real STA. The cache wait loops this pump step feeds
        // (MxAccessValueCache.TryWaitForUpdate,
        // MxAccessWriteCompletionCache.TryWaitForCompletion) block in
        // StaWaitHelper.WaitForSignalOrMessages, whose contract is that the pump
        // step drains the message queue to empty on every call: the wait wakes on
        // input that is merely present, so a pump step that removes nothing makes
        // every wait return instantly and turns the loop into a busy spin at 100%
        // CPU until its deadline (deadline and reply shape still hold — it is a
        // CPU fault, not a correctness one). It is harmless here only because the
        // fakes that pass null pre-populate the caches and run on threads with no
        // windows and no pending input, so there is never a message left behind
        // for the wait to keep waking on.
        this.pumpStep = pumpStep ?? (static () => { });
        this.writeCompletionTimeout = writeCompletionTimeout ?? DefaultWriteCompletionTimeout;
    }

    /// <inheritdoc />
    public MxCommandReply Execute(StaCommand command)
    {
        if (command is null)
        {
            throw new ArgumentNullException(nameof(command));
        }

        return command.Kind switch
        {
            MxCommandKind.Register => ExecuteRegister(command),
            MxCommandKind.Unregister => ExecuteUnregister(command),
            MxCommandKind.AddItem => ExecuteAddItem(command),
            MxCommandKind.AddItem2 => ExecuteAddItem2(command),
            MxCommandKind.RemoveItem => ExecuteRemoveItem(command),
            MxCommandKind.Advise => ExecuteAdvise(command),
            MxCommandKind.UnAdvise => ExecuteUnAdvise(command),
            MxCommandKind.AdviseSupervisory => ExecuteAdviseSupervisory(command),
            MxCommandKind.Suspend => ExecuteSuspend(command),
            MxCommandKind.Activate => ExecuteActivate(command),
            MxCommandKind.AuthenticateUser => ExecuteAuthenticateUser(command),
            MxCommandKind.ArchestraUserToId => ExecuteArchestrAUserToId(command),
            MxCommandKind.AddBufferedItem => ExecuteAddBufferedItem(command),
            MxCommandKind.SetBufferedUpdateInterval => ExecuteSetBufferedUpdateInterval(command),
            MxCommandKind.Write => ExecuteWrite(command),
            MxCommandKind.Write2 => ExecuteWrite2(command),
            MxCommandKind.WriteSecured => ExecuteWriteSecured(command),
            MxCommandKind.WriteSecured2 => ExecuteWriteSecured2(command),
            MxCommandKind.AddItemBulk => ExecuteAddItemBulk(command),
            MxCommandKind.AdviseItemBulk => ExecuteAdviseItemBulk(command),
            MxCommandKind.RemoveItemBulk => ExecuteRemoveItemBulk(command),
            MxCommandKind.UnAdviseItemBulk => ExecuteUnAdviseItemBulk(command),
            MxCommandKind.SubscribeBulk => ExecuteSubscribeBulk(command),
            MxCommandKind.UnsubscribeBulk => ExecuteUnsubscribeBulk(command),
            MxCommandKind.WriteBulk => ExecuteWriteBulk(command),
            MxCommandKind.Write2Bulk => ExecuteWrite2Bulk(command),
            MxCommandKind.WriteSecuredBulk => ExecuteWriteSecuredBulk(command),
            MxCommandKind.WriteSecured2Bulk => ExecuteWriteSecured2Bulk(command),
            MxCommandKind.ReadBulk => ExecuteReadBulk(command),
            MxCommandKind.SubscribeAlarms => ExecuteSubscribeAlarms(command),
            MxCommandKind.UnsubscribeAlarms => ExecuteUnsubscribeAlarms(command),
            MxCommandKind.AcknowledgeAlarm => ExecuteAcknowledgeAlarm(command),
            MxCommandKind.AcknowledgeAlarmByName => ExecuteAcknowledgeAlarmByName(command),
            MxCommandKind.QueryActiveAlarms => ExecuteQueryActiveAlarms(command),
            _ => CreateInvalidRequestReply(command, $"Unsupported MXAccess command kind {command.Kind}."),
        };
    }

    private MxCommandReply ExecuteRegister(StaCommand command)
    {
        if (command.Command.PayloadCase != MxCommand.PayloadOneofCase.Register)
        {
            return CreateInvalidRequestReply(command, "Register command payload is required.");
        }

        int serverHandle = session.Register(command.Command.Register.ClientName);
        MxCommandReply reply = CreateOkReply(command);
        reply.ReturnValue = variantConverter.Convert(serverHandle);
        reply.Register = new RegisterReply
        {
            ServerHandle = serverHandle,
        };

        return reply;
    }

    private MxCommandReply ExecuteUnregister(StaCommand command)
    {
        if (command.Command.PayloadCase != MxCommand.PayloadOneofCase.Unregister)
        {
            return CreateInvalidRequestReply(command, "Unregister command payload is required.");
        }

        session.Unregister(command.Command.Unregister.ServerHandle);
        return CreateOkReply(command);
    }

    private MxCommandReply ExecuteAddItem(StaCommand command)
    {
        if (command.Command.PayloadCase != MxCommand.PayloadOneofCase.AddItem)
        {
            return CreateInvalidRequestReply(command, "AddItem command payload is required.");
        }

        AddItemCommand addItemCommand = command.Command.AddItem;
        int itemHandle = session.AddItem(
            addItemCommand.ServerHandle,
            addItemCommand.ItemDefinition);

        MxCommandReply reply = CreateOkReply(command);
        reply.ReturnValue = variantConverter.Convert(itemHandle);
        reply.AddItem = new AddItemReply
        {
            ItemHandle = itemHandle,
        };

        return reply;
    }

    private MxCommandReply ExecuteAddItem2(StaCommand command)
    {
        if (command.Command.PayloadCase != MxCommand.PayloadOneofCase.AddItem2)
        {
            return CreateInvalidRequestReply(command, "AddItem2 command payload is required.");
        }

        AddItem2Command addItem2Command = command.Command.AddItem2;
        int itemHandle = session.AddItem2(
            addItem2Command.ServerHandle,
            addItem2Command.ItemDefinition,
            addItem2Command.ItemContext);

        MxCommandReply reply = CreateOkReply(command);
        reply.ReturnValue = variantConverter.Convert(itemHandle);
        reply.AddItem2 = new AddItem2Reply
        {
            ItemHandle = itemHandle,
        };

        return reply;
    }

    private MxCommandReply ExecuteRemoveItem(StaCommand command)
    {
        if (command.Command.PayloadCase != MxCommand.PayloadOneofCase.RemoveItem)
        {
            return CreateInvalidRequestReply(command, "RemoveItem command payload is required.");
        }

        RemoveItemCommand removeItemCommand = command.Command.RemoveItem;
        session.RemoveItem(
            removeItemCommand.ServerHandle,
            removeItemCommand.ItemHandle);

        return CreateOkReply(command);
    }

    private MxCommandReply ExecuteAdvise(StaCommand command)
    {
        if (command.Command.PayloadCase != MxCommand.PayloadOneofCase.Advise)
        {
            return CreateInvalidRequestReply(command, "Advise command payload is required.");
        }

        AdviseCommand adviseCommand = command.Command.Advise;
        session.Advise(
            adviseCommand.ServerHandle,
            adviseCommand.ItemHandle);

        return CreateOkReply(command);
    }

    private MxCommandReply ExecuteUnAdvise(StaCommand command)
    {
        if (command.Command.PayloadCase != MxCommand.PayloadOneofCase.UnAdvise)
        {
            return CreateInvalidRequestReply(command, "UnAdvise command payload is required.");
        }

        UnAdviseCommand unAdviseCommand = command.Command.UnAdvise;
        session.UnAdvise(
            unAdviseCommand.ServerHandle,
            unAdviseCommand.ItemHandle);

        return CreateOkReply(command);
    }

    private MxCommandReply ExecuteAdviseSupervisory(StaCommand command)
    {
        if (command.Command.PayloadCase != MxCommand.PayloadOneofCase.AdviseSupervisory)
        {
            return CreateInvalidRequestReply(command, "AdviseSupervisory command payload is required.");
        }

        AdviseSupervisoryCommand adviseSupervisoryCommand = command.Command.AdviseSupervisory;
        session.AdviseSupervisory(
            adviseSupervisoryCommand.ServerHandle,
            adviseSupervisoryCommand.ItemHandle);

        return CreateOkReply(command);
    }

    private MxCommandReply ExecuteSuspend(StaCommand command)
    {
        if (command.Command.PayloadCase != MxCommand.PayloadOneofCase.Suspend)
        {
            return CreateInvalidRequestReply(command, "Suspend command payload is required.");
        }

        SuspendCommand suspendCommand = command.Command.Suspend;
        object nativeStatus = session.Suspend(
            suspendCommand.ServerHandle,
            suspendCommand.ItemHandle);

        MxCommandReply reply = CreateOkReply(command);
        reply.Suspend = new SuspendReply
        {
            Status = statusProxyConverter.Convert(nativeStatus),
        };

        return reply;
    }

    private MxCommandReply ExecuteActivate(StaCommand command)
    {
        if (command.Command.PayloadCase != MxCommand.PayloadOneofCase.Activate)
        {
            return CreateInvalidRequestReply(command, "Activate command payload is required.");
        }

        ActivateCommand activateCommand = command.Command.Activate;
        object nativeStatus = session.Activate(
            activateCommand.ServerHandle,
            activateCommand.ItemHandle);

        MxCommandReply reply = CreateOkReply(command);
        reply.Activate = new ActivateReply
        {
            Status = statusProxyConverter.Convert(nativeStatus),
        };

        return reply;
    }

    private MxCommandReply ExecuteAuthenticateUser(StaCommand command)
    {
        if (command.Command.PayloadCase != MxCommand.PayloadOneofCase.AuthenticateUser)
        {
            return CreateInvalidRequestReply(command, "AuthenticateUser command payload is required.");
        }

        AuthenticateUserCommand authenticateUserCommand = command.Command.AuthenticateUser;

        // The credential (verify_user_password) is passed straight to MXAccess
        // and is never written to logs, diagnostics, or the reply. MXAccess is
        // allowed to fail authentication; the native HResult is surfaced by the
        // dispatcher's exception path.
        int userId = session.AuthenticateUser(
            authenticateUserCommand.ServerHandle,
            authenticateUserCommand.VerifyUser,
            authenticateUserCommand.VerifyUserPassword);

        MxCommandReply reply = CreateOkReply(command);
        reply.AuthenticateUser = new AuthenticateUserReply
        {
            UserId = userId,
        };

        return reply;
    }

    private MxCommandReply ExecuteArchestrAUserToId(StaCommand command)
    {
        if (command.Command.PayloadCase != MxCommand.PayloadOneofCase.ArchestraUserToId)
        {
            return CreateInvalidRequestReply(command, "ArchestrAUserToId command payload is required.");
        }

        ArchestrAUserToIdCommand archestrAUserToIdCommand = command.Command.ArchestraUserToId;
        int userId = session.ArchestrAUserToId(
            archestrAUserToIdCommand.ServerHandle,
            archestrAUserToIdCommand.UserIdGuid);

        MxCommandReply reply = CreateOkReply(command);
        reply.ArchestraUserToId = new ArchestrAUserToIdReply
        {
            UserId = userId,
        };

        return reply;
    }

    private MxCommandReply ExecuteAddBufferedItem(StaCommand command)
    {
        if (command.Command.PayloadCase != MxCommand.PayloadOneofCase.AddBufferedItem)
        {
            return CreateInvalidRequestReply(command, "AddBufferedItem command payload is required.");
        }

        AddBufferedItemCommand addBufferedItemCommand = command.Command.AddBufferedItem;
        int itemHandle = session.AddBufferedItem(
            addBufferedItemCommand.ServerHandle,
            addBufferedItemCommand.ItemDefinition,
            addBufferedItemCommand.ItemContext);

        MxCommandReply reply = CreateOkReply(command);
        reply.ReturnValue = variantConverter.Convert(itemHandle);
        reply.AddBufferedItem = new AddBufferedItemReply
        {
            ItemHandle = itemHandle,
        };

        return reply;
    }

    private MxCommandReply ExecuteSetBufferedUpdateInterval(StaCommand command)
    {
        if (command.Command.PayloadCase != MxCommand.PayloadOneofCase.SetBufferedUpdateInterval)
        {
            return CreateInvalidRequestReply(command, "SetBufferedUpdateInterval command payload is required.");
        }

        SetBufferedUpdateIntervalCommand setBufferedUpdateIntervalCommand = command.Command.SetBufferedUpdateInterval;
        session.SetBufferedUpdateInterval(
            setBufferedUpdateIntervalCommand.ServerHandle,
            setBufferedUpdateIntervalCommand.UpdateIntervalMilliseconds);

        return CreateOkReply(command);
    }

    private MxCommandReply ExecuteWrite(StaCommand command)
    {
        if (command.Command.PayloadCase != MxCommand.PayloadOneofCase.Write)
        {
            return CreateInvalidRequestReply(command, "Write command payload is required.");
        }

        WriteCommand writeCommand = command.Command.Write;
        if (writeCommand.Value is null)
        {
            return CreateInvalidRequestReply(command, "Write command value is required.");
        }

        // Same pre-call baseline rule as ExecuteWriteSecured: plain Write is
        // also fire-and-forget in MXAccess, and its commit outcome only exists
        // in the later OnWriteComplete callback.
        MxAccessWriteCompletionCache completionCache = session.WriteCompletionCache;
        ulong completionBaseline = CaptureCompletionBaseline(
            completionCache,
            writeCommand.ServerHandle,
            writeCommand.ItemHandle);

        session.Write(
            writeCommand.ServerHandle,
            writeCommand.ItemHandle,
            variantConverter.ConvertToComValue(writeCommand.Value),
            writeCommand.UserId);

        MxCommandReply reply = CreateOkReply(command);
        AwaitWriteCompletion(
            reply,
            completionCache,
            writeCommand.ServerHandle,
            writeCommand.ItemHandle,
            completionBaseline);
        return reply;
    }

    private MxCommandReply ExecuteWrite2(StaCommand command)
    {
        if (command.Command.PayloadCase != MxCommand.PayloadOneofCase.Write2)
        {
            return CreateInvalidRequestReply(command, "Write2 command payload is required.");
        }

        Write2Command write2Command = command.Command.Write2;
        if (write2Command.Value is null)
        {
            return CreateInvalidRequestReply(command, "Write2 command value is required.");
        }

        if (write2Command.TimestampValue is null)
        {
            return CreateInvalidRequestReply(command, "Write2 command timestamp value is required.");
        }

        // Same pre-call baseline rule as ExecuteWriteSecured.
        MxAccessWriteCompletionCache completionCache = session.WriteCompletionCache;
        ulong completionBaseline = CaptureCompletionBaseline(
            completionCache,
            write2Command.ServerHandle,
            write2Command.ItemHandle);

        session.Write2(
            write2Command.ServerHandle,
            write2Command.ItemHandle,
            variantConverter.ConvertToComValue(write2Command.Value),
            variantConverter.ConvertToComValue(write2Command.TimestampValue),
            write2Command.UserId);

        MxCommandReply reply = CreateOkReply(command);
        AwaitWriteCompletion(
            reply,
            completionCache,
            write2Command.ServerHandle,
            write2Command.ItemHandle,
            completionBaseline);
        return reply;
    }

    private MxCommandReply ExecuteWriteSecured(StaCommand command)
    {
        if (command.Command.PayloadCase != MxCommand.PayloadOneofCase.WriteSecured)
        {
            return CreateInvalidRequestReply(command, "WriteSecured command payload is required.");
        }

        WriteSecuredCommand writeSecuredCommand = command.Command.WriteSecured;
        if (writeSecuredCommand.Value is null)
        {
            return CreateInvalidRequestReply(command, "WriteSecured command value is required.");
        }

        // Baseline BEFORE the COM call: a completion that dispatches during or
        // immediately after WriteSecured bumps the version past this snapshot,
        // so a fast commit still correlates (no missed-callback window).
        MxAccessWriteCompletionCache completionCache = session.WriteCompletionCache;
        ulong completionBaseline = CaptureCompletionBaseline(
            completionCache,
            writeSecuredCommand.ServerHandle,
            writeSecuredCommand.ItemHandle);

        session.WriteSecured(
            writeSecuredCommand.ServerHandle,
            writeSecuredCommand.ItemHandle,
            writeSecuredCommand.CurrentUserId,
            writeSecuredCommand.VerifierUserId,
            variantConverter.ConvertToComValue(writeSecuredCommand.Value));

        MxCommandReply reply = CreateOkReply(command);
        AwaitWriteCompletion(
            reply,
            completionCache,
            writeSecuredCommand.ServerHandle,
            writeSecuredCommand.ItemHandle,
            completionBaseline);
        return reply;
    }

    private MxCommandReply ExecuteWriteSecured2(StaCommand command)
    {
        if (command.Command.PayloadCase != MxCommand.PayloadOneofCase.WriteSecured2)
        {
            return CreateInvalidRequestReply(command, "WriteSecured2 command payload is required.");
        }

        WriteSecured2Command writeSecured2Command = command.Command.WriteSecured2;
        if (writeSecured2Command.Value is null)
        {
            return CreateInvalidRequestReply(command, "WriteSecured2 command value is required.");
        }

        if (writeSecured2Command.TimestampValue is null)
        {
            return CreateInvalidRequestReply(command, "WriteSecured2 command timestamp value is required.");
        }

        // Same pre-call baseline rule as ExecuteWriteSecured.
        MxAccessWriteCompletionCache completionCache = session.WriteCompletionCache;
        ulong completionBaseline = CaptureCompletionBaseline(
            completionCache,
            writeSecured2Command.ServerHandle,
            writeSecured2Command.ItemHandle);

        session.WriteSecured2(
            writeSecured2Command.ServerHandle,
            writeSecured2Command.ItemHandle,
            writeSecured2Command.CurrentUserId,
            writeSecured2Command.VerifierUserId,
            variantConverter.ConvertToComValue(writeSecured2Command.Value),
            variantConverter.ConvertToComValue(writeSecured2Command.TimestampValue));

        MxCommandReply reply = CreateOkReply(command);
        AwaitWriteCompletion(
            reply,
            completionCache,
            writeSecured2Command.ServerHandle,
            writeSecured2Command.ItemHandle,
            completionBaseline);
        return reply;
    }

    private MxCommandReply ExecuteAddItemBulk(StaCommand command)
    {
        if (command.Command.PayloadCase != MxCommand.PayloadOneofCase.AddItemBulk)
        {
            return CreateInvalidRequestReply(command, "AddItemBulk command payload is required.");
        }

        AddItemBulkCommand addItemBulkCommand = command.Command.AddItemBulk;
        return CreateBulkReply(
            command,
            session.AddItemBulk(addItemBulkCommand.ServerHandle, addItemBulkCommand.TagAddresses));
    }

    private MxCommandReply ExecuteAdviseItemBulk(StaCommand command)
    {
        if (command.Command.PayloadCase != MxCommand.PayloadOneofCase.AdviseItemBulk)
        {
            return CreateInvalidRequestReply(command, "AdviseItemBulk command payload is required.");
        }

        AdviseItemBulkCommand adviseItemBulkCommand = command.Command.AdviseItemBulk;
        return CreateBulkReply(
            command,
            session.AdviseItemBulk(adviseItemBulkCommand.ServerHandle, adviseItemBulkCommand.ItemHandles));
    }

    private MxCommandReply ExecuteRemoveItemBulk(StaCommand command)
    {
        if (command.Command.PayloadCase != MxCommand.PayloadOneofCase.RemoveItemBulk)
        {
            return CreateInvalidRequestReply(command, "RemoveItemBulk command payload is required.");
        }

        RemoveItemBulkCommand removeItemBulkCommand = command.Command.RemoveItemBulk;
        return CreateBulkReply(
            command,
            session.RemoveItemBulk(removeItemBulkCommand.ServerHandle, removeItemBulkCommand.ItemHandles));
    }

    private MxCommandReply ExecuteUnAdviseItemBulk(StaCommand command)
    {
        if (command.Command.PayloadCase != MxCommand.PayloadOneofCase.UnAdviseItemBulk)
        {
            return CreateInvalidRequestReply(command, "UnAdviseItemBulk command payload is required.");
        }

        UnAdviseItemBulkCommand unAdviseItemBulkCommand = command.Command.UnAdviseItemBulk;
        return CreateBulkReply(
            command,
            session.UnAdviseItemBulk(unAdviseItemBulkCommand.ServerHandle, unAdviseItemBulkCommand.ItemHandles));
    }

    private MxCommandReply ExecuteSubscribeBulk(StaCommand command)
    {
        if (command.Command.PayloadCase != MxCommand.PayloadOneofCase.SubscribeBulk)
        {
            return CreateInvalidRequestReply(command, "SubscribeBulk command payload is required.");
        }

        SubscribeBulkCommand subscribeBulkCommand = command.Command.SubscribeBulk;
        return CreateBulkReply(
            command,
            session.SubscribeBulk(subscribeBulkCommand.ServerHandle, subscribeBulkCommand.TagAddresses));
    }

    private MxCommandReply ExecuteUnsubscribeBulk(StaCommand command)
    {
        if (command.Command.PayloadCase != MxCommand.PayloadOneofCase.UnsubscribeBulk)
        {
            return CreateInvalidRequestReply(command, "UnsubscribeBulk command payload is required.");
        }

        UnsubscribeBulkCommand unsubscribeBulkCommand = command.Command.UnsubscribeBulk;
        return CreateBulkReply(
            command,
            session.UnsubscribeBulk(unsubscribeBulkCommand.ServerHandle, unsubscribeBulkCommand.ItemHandles));
    }

    private MxCommandReply ExecuteWriteBulk(StaCommand command)
    {
        if (command.Command.PayloadCase != MxCommand.PayloadOneofCase.WriteBulk)
        {
            return CreateInvalidRequestReply(command, "WriteBulk command payload is required.");
        }

        WriteBulkCommand writeBulkCommand = command.Command.WriteBulk;
        foreach (WriteBulkEntry entry in writeBulkCommand.Entries)
        {
            if (entry.Value is null)
            {
                return CreateInvalidRequestReply(
                    command,
                    $"WriteBulk entry for item handle {entry.ItemHandle} is missing its value.");
            }
        }

        return CreateBulkWriteReply(
            command,
            session.WriteBulk(
                writeBulkCommand.ServerHandle,
                writeBulkCommand.Entries,
                value => variantConverter.ConvertToComValue(value)));
    }

    private MxCommandReply ExecuteWrite2Bulk(StaCommand command)
    {
        if (command.Command.PayloadCase != MxCommand.PayloadOneofCase.Write2Bulk)
        {
            return CreateInvalidRequestReply(command, "Write2Bulk command payload is required.");
        }

        Write2BulkCommand write2BulkCommand = command.Command.Write2Bulk;
        foreach (Write2BulkEntry entry in write2BulkCommand.Entries)
        {
            if (entry.Value is null)
            {
                return CreateInvalidRequestReply(
                    command,
                    $"Write2Bulk entry for item handle {entry.ItemHandle} is missing its value.");
            }

            if (entry.TimestampValue is null)
            {
                return CreateInvalidRequestReply(
                    command,
                    $"Write2Bulk entry for item handle {entry.ItemHandle} is missing its timestamp value.");
            }
        }

        return CreateBulkWriteReply(
            command,
            session.Write2Bulk(
                write2BulkCommand.ServerHandle,
                write2BulkCommand.Entries,
                value => variantConverter.ConvertToComValue(value)));
    }

    private MxCommandReply ExecuteWriteSecuredBulk(StaCommand command)
    {
        if (command.Command.PayloadCase != MxCommand.PayloadOneofCase.WriteSecuredBulk)
        {
            return CreateInvalidRequestReply(command, "WriteSecuredBulk command payload is required.");
        }

        WriteSecuredBulkCommand writeSecuredBulkCommand = command.Command.WriteSecuredBulk;
        foreach (WriteSecuredBulkEntry entry in writeSecuredBulkCommand.Entries)
        {
            if (entry.Value is null)
            {
                return CreateInvalidRequestReply(
                    command,
                    $"WriteSecuredBulk entry for item handle {entry.ItemHandle} is missing its value.");
            }
        }

        return CreateBulkWriteReply(
            command,
            session.WriteSecuredBulk(
                writeSecuredBulkCommand.ServerHandle,
                writeSecuredBulkCommand.Entries,
                value => variantConverter.ConvertToComValue(value)));
    }

    private MxCommandReply ExecuteWriteSecured2Bulk(StaCommand command)
    {
        if (command.Command.PayloadCase != MxCommand.PayloadOneofCase.WriteSecured2Bulk)
        {
            return CreateInvalidRequestReply(command, "WriteSecured2Bulk command payload is required.");
        }

        WriteSecured2BulkCommand writeSecured2BulkCommand = command.Command.WriteSecured2Bulk;
        foreach (WriteSecured2BulkEntry entry in writeSecured2BulkCommand.Entries)
        {
            if (entry.Value is null)
            {
                return CreateInvalidRequestReply(
                    command,
                    $"WriteSecured2Bulk entry for item handle {entry.ItemHandle} is missing its value.");
            }

            if (entry.TimestampValue is null)
            {
                return CreateInvalidRequestReply(
                    command,
                    $"WriteSecured2Bulk entry for item handle {entry.ItemHandle} is missing its timestamp value.");
            }
        }

        return CreateBulkWriteReply(
            command,
            session.WriteSecured2Bulk(
                writeSecured2BulkCommand.ServerHandle,
                writeSecured2BulkCommand.Entries,
                value => variantConverter.ConvertToComValue(value)));
    }

    private MxCommandReply ExecuteReadBulk(StaCommand command)
    {
        if (command.Command.PayloadCase != MxCommand.PayloadOneofCase.ReadBulk)
        {
            return CreateInvalidRequestReply(command, "ReadBulk command payload is required.");
        }

        ReadBulkCommand readBulkCommand = command.Command.ReadBulk;
        TimeSpan timeout = readBulkCommand.TimeoutMs == 0
            ? DefaultReadBulkTimeout
            : TimeSpan.FromMilliseconds(readBulkCommand.TimeoutMs);

        IReadOnlyList<BulkReadResult> results = session.ReadBulk(
            readBulkCommand.ServerHandle,
            readBulkCommand.TagAddresses,
            timeout,
            pumpStep);

        MxCommandReply reply = CreateOkReply(command);
        BulkReadReply bulkReply = new();
        bulkReply.Results.Add(results);
        reply.ReadBulk = bulkReply;
        return reply;
    }

    private MxCommandReply ExecuteSubscribeAlarms(StaCommand command)
    {
        if (command.Command.PayloadCase != MxCommand.PayloadOneofCase.SubscribeAlarms)
        {
            return CreateInvalidRequestReply(command, "SubscribeAlarms command payload is required.");
        }
        if (alarmCommandHandler is null)
        {
            return CreateInvalidRequestReply(
                command,
                "SubscribeAlarms requires an alarm command handler; the worker was constructed without one.");
        }

        SubscribeAlarmsCommand subscribeCommand = command.Command.SubscribeAlarms;
        string subscription = subscribeCommand.SubscriptionExpression ?? string.Empty;
        if (string.IsNullOrWhiteSpace(subscription))
        {
            return CreateInvalidRequestReply(command, "SubscribeAlarms.subscription_expression is required.");
        }

        try
        {
            alarmCommandHandler.Subscribe(subscribeCommand, command.SessionId);
            return CreateOkReply(command);
        }
        catch (Exception ex)
        {
            return CreateAlarmFailureReply(command, ex);
        }
    }

    private MxCommandReply ExecuteUnsubscribeAlarms(StaCommand command)
    {
        if (command.Command.PayloadCase != MxCommand.PayloadOneofCase.UnsubscribeAlarms)
        {
            return CreateInvalidRequestReply(command, "UnsubscribeAlarms command payload is required.");
        }
        if (alarmCommandHandler is null)
        {
            // No handler configured — Unsubscribe is a no-op in that case;
            // it can't be in a subscribed state to begin with.
            return CreateOkReply(command);
        }

        try
        {
            alarmCommandHandler.Unsubscribe();
            return CreateOkReply(command);
        }
        catch (Exception ex)
        {
            return CreateAlarmFailureReply(command, ex);
        }
    }

    private MxCommandReply ExecuteAcknowledgeAlarm(StaCommand command)
    {
        if (command.Command.PayloadCase != MxCommand.PayloadOneofCase.AcknowledgeAlarmCommand)
        {
            return CreateInvalidRequestReply(command, "AcknowledgeAlarm command payload is required.");
        }
        if (alarmCommandHandler is null)
        {
            return CreateInvalidRequestReply(
                command,
                "AcknowledgeAlarm requires an alarm command handler; the worker was constructed without one.");
        }

        AcknowledgeAlarmCommand payload = command.Command.AcknowledgeAlarmCommand;
        if (!Guid.TryParse(payload.AlarmGuid, out Guid alarmGuid))
        {
            return CreateInvalidRequestReply(
                command,
                $"AcknowledgeAlarm.alarm_guid is not a valid canonical GUID: '{payload.AlarmGuid}'.");
        }

        try
        {
            int rc = alarmCommandHandler.Acknowledge(
                alarmGuid,
                payload.Comment,
                payload.OperatorUser,
                payload.OperatorNode,
                payload.OperatorDomain,
                payload.OperatorFullName);
            MxCommandReply reply = CreateOkReply(command);
            reply.Hresult = rc;
            reply.AcknowledgeAlarm = new AcknowledgeAlarmReplyPayload
            {
                NativeStatus = rc,
            };
            if (rc != 0)
            {
                reply.DiagnosticMessage = $"AVEVA AlarmAckByGUID returned non-zero status {rc}.";
            }
            return reply;
        }
        catch (Exception ex)
        {
            return CreateAlarmFailureReply(command, ex);
        }
    }

    private MxCommandReply ExecuteAcknowledgeAlarmByName(StaCommand command)
    {
        if (command.Command.PayloadCase != MxCommand.PayloadOneofCase.AcknowledgeAlarmByNameCommand)
        {
            return CreateInvalidRequestReply(command, "AcknowledgeAlarmByName command payload is required.");
        }
        if (alarmCommandHandler is null)
        {
            return CreateInvalidRequestReply(
                command,
                "AcknowledgeAlarmByName requires an alarm command handler; the worker was constructed without one.");
        }

        AcknowledgeAlarmByNameCommand payload = command.Command.AcknowledgeAlarmByNameCommand;
        if (string.IsNullOrWhiteSpace(payload.AlarmName))
        {
            return CreateInvalidRequestReply(command, "AcknowledgeAlarmByName.alarm_name is required.");
        }

        try
        {
            int rc = alarmCommandHandler.AcknowledgeByName(
                payload.AlarmName,
                payload.ProviderName,
                payload.GroupName,
                payload.Comment,
                payload.OperatorUser,
                payload.OperatorNode,
                payload.OperatorDomain,
                payload.OperatorFullName);
            MxCommandReply reply = CreateOkReply(command);
            reply.Hresult = rc;
            reply.AcknowledgeAlarm = new AcknowledgeAlarmReplyPayload
            {
                NativeStatus = rc,
            };
            if (rc != 0)
            {
                reply.DiagnosticMessage = $"AVEVA AlarmAckByName returned non-zero status {rc}.";
            }
            return reply;
        }
        catch (Exception ex)
        {
            return CreateAlarmFailureReply(command, ex);
        }
    }

    private MxCommandReply ExecuteQueryActiveAlarms(StaCommand command)
    {
        if (command.Command.PayloadCase != MxCommand.PayloadOneofCase.QueryActiveAlarmsCommand)
        {
            return CreateInvalidRequestReply(command, "QueryActiveAlarms command payload is required.");
        }
        if (alarmCommandHandler is null)
        {
            return CreateInvalidRequestReply(
                command,
                "QueryActiveAlarms requires an alarm command handler; the worker was constructed without one.");
        }

        try
        {
            IReadOnlyList<ActiveAlarmSnapshot> snapshots = alarmCommandHandler.QueryActive(
                command.Command.QueryActiveAlarmsCommand.AlarmFilterPrefix,
                out bool snapshotTruncated);
            QueryActiveAlarmsReplyPayload payload = new QueryActiveAlarmsReplyPayload();
            payload.Snapshots.AddRange(snapshots);
            // Set-level degraded status: the snapshot may omit actives because
            // the provider fetch hit its cap. The records carry the same flag,
            // but a prefix filter can leave none, so the payload states it too.
            // Same value the records were stamped with — one read, not a second.
            payload.SnapshotTruncated = snapshotTruncated;
            MxCommandReply reply = CreateOkReply(command);
            reply.QueryActiveAlarms = payload;
            return reply;
        }
        catch (Exception ex)
        {
            return CreateAlarmFailureReply(command, ex);
        }
    }

    /// <summary>
    ///     Captures the write-completion baseline for <paramref name="serverHandle"/> /
    ///     <paramref name="itemHandle"/> immediately before a Write/Write2/WriteSecured/
    ///     WriteSecured2 COM call. Pumps once first (WRK-32) so an <c>OnWriteComplete</c>
    ///     already queued for this handle — most commonly the still-undelivered callback of an
    ///     earlier write to the same tag that this worker gave up waiting on — folds into the
    ///     baseline instead of being available to ride this write's reply.
    /// </summary>
    /// <remarks>
    ///     This narrows, but does not close, the correlation gap documented on
    ///     <see cref="AwaitWriteCompletion"/>: a completion that MXAccess queues after this pump
    ///     step but before the COM call returns is still indistinguishable from this write's own
    ///     completion, because <c>OnWriteComplete</c> carries no per-call token — only the
    ///     server/item handle pair.
    /// </remarks>
    private ulong CaptureCompletionBaseline(
        MxAccessWriteCompletionCache completionCache,
        int serverHandle,
        int itemHandle)
    {
        pumpStep();
        return completionCache.CurrentVersion(serverHandle, itemHandle);
    }

    /// <summary>
    ///     Bounded pump-wait for the OnWriteComplete row matching a Write/Write2/
    ///     WriteSecured/WriteSecured2 call, copied onto the reply when it
    ///     arrives in time. The executor holds the STA thread but pumps
    ///     Windows messages each poll (ReadBulk precedent) so the COM callback
    ///     can dispatch re-entrantly; on expiry the reply keeps its empty
    ///     statuses — the consumer's unconfirmed path, never a synthesized
    ///     failure. Protocol status/hresult stay acceptance-only either way.
    /// </summary>
    /// <remarks>
    ///     Correlation is recency-based per handle, not per call: MXAccess's
    ///     <c>OnWriteComplete</c> callback carries no correlation token, only the server/item
    ///     handle pair, so this method reports the first completion whose version exceeds
    ///     <paramref name="completionBaseline"/> as this write's outcome. If an earlier write to
    ///     the same handle times out unconfirmed and its callback is still undelivered, that
    ///     stale completion can dispatch during this write's pump-wait and ride this reply
    ///     instead — <see cref="CaptureCompletionBaseline"/>'s pre-call pump only folds in a
    ///     completion already queued before this write started; it cannot fold in one queued
    ///     during the COM call or the wait itself. The <c>Hresult</c>/<c>ProtocolStatus</c>
    ///     fields stay acceptance-only regardless, so the exposure is limited to the enrichment
    ///     <c>statuses</c> field, and it is inherent to the COM surface — a native client
    ///     correlating by handle has the identical ambiguity. Full closure needs an
    ///     MXAccess-side correlation token, which does not exist.
    /// </remarks>
    private void AwaitWriteCompletion(
        MxCommandReply reply,
        MxAccessWriteCompletionCache completionCache,
        int serverHandle,
        int itemHandle,
        ulong completionBaseline)
    {
        if (writeCompletionTimeout <= TimeSpan.Zero)
        {
            return;
        }

        if (completionCache.TryWaitForCompletion(
                serverHandle,
                itemHandle,
                completionBaseline,
                DateTime.UtcNow + writeCompletionTimeout,
                pumpStep,
                out Google.Protobuf.Collections.RepeatedField<MxStatusProxy> statuses))
        {
            reply.Statuses.Add(statuses);
        }
    }

    private static MxCommandReply CreateAlarmFailureReply(StaCommand command, Exception exception)
    {
        return new MxCommandReply
        {
            SessionId = command.SessionId,
            CorrelationId = command.CorrelationId,
            Kind = command.Kind,
            ProtocolStatus = new ProtocolStatus
            {
                Code = ProtocolStatusCode.MxaccessFailure,
                Message = exception.Message,
            },
            DiagnosticMessage = $"{exception.GetType().FullName}: {exception.Message}",
        };
    }

    private static MxCommandReply CreateOkReply(StaCommand command)
    {
        return new MxCommandReply
        {
            SessionId = command.SessionId,
            CorrelationId = command.CorrelationId,
            Kind = command.Kind,
            Hresult = 0,
            ProtocolStatus = new ProtocolStatus
            {
                Code = ProtocolStatusCode.Ok,
                Message = "OK",
            },
        };
    }

    private static MxCommandReply CreateBulkReply(
        StaCommand command,
        IEnumerable<SubscribeResult> results)
    {
        MxCommandReply reply = CreateOkReply(command);
        BulkSubscribeReply bulkReply = new();
        bulkReply.Results.Add(results);

        switch (command.Kind)
        {
            case MxCommandKind.AddItemBulk:
                reply.AddItemBulk = bulkReply;
                break;
            case MxCommandKind.AdviseItemBulk:
                reply.AdviseItemBulk = bulkReply;
                break;
            case MxCommandKind.RemoveItemBulk:
                reply.RemoveItemBulk = bulkReply;
                break;
            case MxCommandKind.UnAdviseItemBulk:
                reply.UnAdviseItemBulk = bulkReply;
                break;
            case MxCommandKind.SubscribeBulk:
                reply.SubscribeBulk = bulkReply;
                break;
            case MxCommandKind.UnsubscribeBulk:
                reply.UnsubscribeBulk = bulkReply;
                break;
            default:
                throw new InvalidOperationException($"Unsupported bulk command kind {command.Kind}.");
        }

        return reply;
    }

    private static MxCommandReply CreateBulkWriteReply(
        StaCommand command,
        IEnumerable<BulkWriteResult> results)
    {
        MxCommandReply reply = CreateOkReply(command);
        BulkWriteReply bulkReply = new();
        bulkReply.Results.Add(results);

        switch (command.Kind)
        {
            case MxCommandKind.WriteBulk:
                reply.WriteBulk = bulkReply;
                break;
            case MxCommandKind.Write2Bulk:
                reply.Write2Bulk = bulkReply;
                break;
            case MxCommandKind.WriteSecuredBulk:
                reply.WriteSecuredBulk = bulkReply;
                break;
            case MxCommandKind.WriteSecured2Bulk:
                reply.WriteSecured2Bulk = bulkReply;
                break;
            default:
                throw new InvalidOperationException($"Unsupported bulk write command kind {command.Kind}.");
        }

        return reply;
    }

    private static MxCommandReply CreateInvalidRequestReply(
        StaCommand command,
        string message)
    {
        return new MxCommandReply
        {
            SessionId = command.SessionId,
            CorrelationId = command.CorrelationId,
            Kind = command.Kind,
            ProtocolStatus = new ProtocolStatus
            {
                Code = ProtocolStatusCode.InvalidRequest,
                Message = message,
            },
            DiagnosticMessage = message,
        };
    }
}