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

using System;
using ArchestrA.MxAccess;
using Proto = ZB.MOM.WW.MxGateway.Contracts.Proto;

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

/// <summary>Sink for MXAccess COM events that converts them to protobuf format.</summary>
public sealed class MxAccessBaseEventSink : IMxAccessEventSink
{
    private readonly MxAccessEventMapper eventMapper;
    private readonly MxAccessEventQueue eventQueue;
    private readonly MxAccessValueCache valueCache;
    private LMXProxyServerClass? server;
    private string sessionId = string.Empty;

    /// <summary>Initializes a new instance of the MxAccessBaseEventSink class with a default queue.</summary>
    public MxAccessBaseEventSink()
        : this(new MxAccessEventQueue())
    {
    }

    /// <summary>Initializes a new instance of the MxAccessBaseEventSink class with a provided queue.</summary>
    /// <param name="eventQueue">Queue for buffering converted MXAccess events.</param>
    public MxAccessBaseEventSink(MxAccessEventQueue eventQueue)
        : this(eventQueue, new MxAccessEventMapper(), new MxAccessValueCache())
    {
    }

    /// <summary>Initializes a new instance of the MxAccessBaseEventSink class with provided queue and mapper.</summary>
    /// <param name="eventQueue">Queue for buffering converted MXAccess events.</param>
    /// <param name="eventMapper">Converter for MXAccess events to protobuf format.</param>
    public MxAccessBaseEventSink(
        MxAccessEventQueue eventQueue,
        MxAccessEventMapper eventMapper)
        : this(eventQueue, eventMapper, new MxAccessValueCache())
    {
    }

    /// <summary>
    ///     Initializes a new instance of the MxAccessBaseEventSink class with
    ///     provided queue, mapper, and a shared value cache. The cache is
    ///     populated from every successful <c>OnDataChange</c> dispatch so the
    ///     worker's ReadBulk executor can satisfy a "current value" request
    ///     from an already-advised tag without touching the subscription.
    /// </summary>
    /// <param name="eventQueue">Queue for buffering converted MXAccess events.</param>
    /// <param name="eventMapper">Converter for MXAccess events to protobuf format.</param>
    /// <param name="valueCache">Per-session last-value cache shared with the MxAccessSession.</param>
    public MxAccessBaseEventSink(
        MxAccessEventQueue eventQueue,
        MxAccessEventMapper eventMapper,
        MxAccessValueCache valueCache)
    {
        this.eventQueue = eventQueue ?? throw new ArgumentNullException(nameof(eventQueue));
        this.eventMapper = eventMapper ?? throw new ArgumentNullException(nameof(eventMapper));
        this.valueCache = valueCache ?? throw new ArgumentNullException(nameof(valueCache));
    }

    /// <summary>
    ///     The last-value cache populated by this sink. Exposed so the
    ///     MxAccessSession can share the same instance for ReadBulk lookups.
    /// </summary>
    public MxAccessValueCache ValueCache => valueCache;

    /// <inheritdoc />
    public void Attach(
        object mxAccessComObject,
        string sessionId)
    {
        this.sessionId = sessionId ?? string.Empty;
        server = (LMXProxyServerClass)mxAccessComObject;
        server.OnDataChange += OnDataChange;
        server.OnWriteComplete += OnWriteComplete;
        server.OperationComplete += OperationComplete;
        server.OnBufferedDataChange += OnBufferedDataChange;
    }

    /// <inheritdoc />
    public void Detach()
    {
        if (server is null)
        {
            return;
        }

        server.OnDataChange -= OnDataChange;
        server.OnWriteComplete -= OnWriteComplete;
        server.OperationComplete -= OperationComplete;
        server.OnBufferedDataChange -= OnBufferedDataChange;
        server = null;
        sessionId = string.Empty;
    }

    /// <summary>
    ///     Handles the MXAccess <c>OnDataChange</c> COM event: converts the
    ///     event arguments to a protobuf <see cref="Proto.MxEvent"/> and enqueues
    ///     it. Subscribed to the COM object's event in <see cref="Attach"/>.
    ///     Exposed <c>internal</c> so unit tests can drive the integrated
    ///     sink → mapper → queue path without a live MXAccess COM event source.
    /// </summary>
    /// <param name="hLMXServerHandle">The MXAccess server handle.</param>
    /// <param name="phItemHandle">The MXAccess item handle.</param>
    /// <param name="pvItemValue">The new item value.</param>
    /// <param name="pwItemQuality">The item quality.</param>
    /// <param name="pftItemTimeStamp">The item timestamp.</param>
    /// <param name="pVars">Status array from MXAccess event.</param>
    internal void OnDataChange(
        int hLMXServerHandle,
        int phItemHandle,
        object pvItemValue,
        int pwItemQuality,
        object pftItemTimeStamp,
        ref MXSTATUS_PROXY[] pVars)
    {
        MXSTATUS_PROXY[] statuses = pVars;
        // Build the protobuf event once, enqueue it for the outbound stream, and
        // also publish it into the per-session value cache so ReadBulk can serve
        // it as a "current value" without re-advising. The cache update is the
        // ONLY new side effect — fail-fast on conversion still drops the event
        // through the same EnqueueEvent path as before.
        EnqueueEvent(
            () => eventMapper.CreateOnDataChange(
                sessionId,
                hLMXServerHandle,
                phItemHandle,
                pvItemValue,
                pwItemQuality,
                pftItemTimeStamp,
                statuses),
            mxEvent => valueCache.Set(hLMXServerHandle, phItemHandle, mxEvent));
    }

    /// <summary>
    ///     Handles the MXAccess <c>OnWriteComplete</c> COM event. Exposed
    ///     <c>internal</c> as a unit-test seam; see <see cref="OnDataChange"/>.
    /// </summary>
    /// <param name="hLMXServerHandle">The MXAccess server handle.</param>
    /// <param name="phItemHandle">The MXAccess item handle.</param>
    /// <param name="pVars">Status array from MXAccess event.</param>
    internal void OnWriteComplete(
        int hLMXServerHandle,
        int phItemHandle,
        ref MXSTATUS_PROXY[] pVars)
    {
        MXSTATUS_PROXY[] statuses = pVars;
        EnqueueEvent(() => eventMapper.CreateOnWriteComplete(
            sessionId,
            hLMXServerHandle,
            phItemHandle,
            statuses));
    }

    /// <summary>
    ///     Handles the MXAccess <c>OperationComplete</c> COM event. Exposed
    ///     <c>internal</c> as a unit-test seam; see <see cref="OnDataChange"/>.
    /// </summary>
    /// <param name="hLMXServerHandle">The MXAccess server handle.</param>
    /// <param name="phItemHandle">The MXAccess item handle.</param>
    /// <param name="pVars">Status array from MXAccess event.</param>
    internal void OperationComplete(
        int hLMXServerHandle,
        int phItemHandle,
        ref MXSTATUS_PROXY[] pVars)
    {
        MXSTATUS_PROXY[] statuses = pVars;
        EnqueueEvent(() => eventMapper.CreateOperationComplete(
            sessionId,
            hLMXServerHandle,
            phItemHandle,
            statuses));
    }

    /// <summary>
    ///     Handles the MXAccess <c>OnBufferedDataChange</c> COM event. Exposed
    ///     <c>internal</c> as a unit-test seam; see <see cref="OnDataChange"/>.
    /// </summary>
    /// <param name="hLMXServerHandle">The MXAccess server handle.</param>
    /// <param name="phItemHandle">The MXAccess item handle.</param>
    /// <param name="dtDataType">The data type of the buffered value.</param>
    /// <param name="pvItemValue">The new item value.</param>
    /// <param name="pwItemQuality">The item quality.</param>
    /// <param name="pftItemTimeStamp">The item timestamp.</param>
    /// <param name="pVars">Status array from MXAccess event.</param>
    internal void OnBufferedDataChange(
        int hLMXServerHandle,
        int phItemHandle,
        MxDataType dtDataType,
        object pvItemValue,
        object pwItemQuality,
        object pftItemTimeStamp,
        ref MXSTATUS_PROXY[] pVars)
    {
        MXSTATUS_PROXY[] statuses = pVars;
        EnqueueEvent(() => eventMapper.CreateOnBufferedDataChange(
            sessionId,
            hLMXServerHandle,
            phItemHandle,
            (int)dtDataType,
            pvItemValue,
            pwItemQuality,
            pftItemTimeStamp,
            statuses));
    }

    private void EnqueueEvent(Func<Proto.MxEvent> createEvent)
    {
        EnqueueEvent(createEvent, postPublish: null);
    }

    private void EnqueueEvent(Func<Proto.MxEvent> createEvent, Action<Proto.MxEvent>? postPublish)
    {
        Proto.MxEvent mxEvent;
        try
        {
            mxEvent = createEvent();
        }
        catch (Exception exception)
        {
            eventQueue.RecordFault(CreateEventConversionFault(exception));
            return;
        }

        try
        {
            eventQueue.Enqueue(mxEvent);
        }
        catch (Exception exception)
        {
            // Two distinct failures land here, both intentionally fail-fast:
            //  - A conversion failure from createEvent() — recorded here as an
            //    MxaccessEventConversionFailed fault.
            //  - An MxAccessEventQueueOverflowException from Enqueue when the
            //    queue is at capacity. Per the fail-fast backpressure design
            //    (docs/DesignDecisions.md) the event is dropped and the queue
            //    has *already* self-recorded a QueueOverflow fault. Because
            //    MxAccessEventQueue.RecordFault keeps only the first fault,
            //    this catch's RecordFault call is then a deliberate near
            //    no-op rather than a second, conflicting fault.
            eventQueue.RecordFault(CreateEventConversionFault(exception));
            return;
        }

        // Only publish to caches/observers after the event has cleared the
        // queue, so a queue overflow does not leak a "fresher" cached value
        // than what was actually shipped to the gateway.
        if (postPublish is not null)
        {
            try
            {
                postPublish(mxEvent);
            }
            catch (Exception exception)
            {
                eventQueue.RecordFault(CreateEventConversionFault(exception));
            }
        }
    }

    private Proto.WorkerFault CreateEventConversionFault(Exception exception)
    {
        return new Proto.WorkerFault
        {
            Category = Proto.WorkerFaultCategory.MxaccessEventConversionFailed,
            ExceptionType = exception.GetType().FullName ?? string.Empty,
            DiagnosticMessage = $"{exception.GetType().FullName}: HRESULT 0x{unchecked((uint)exception.HResult):X8}",
            ProtocolStatus = new Proto.ProtocolStatus
            {
                Code = Proto.ProtocolStatusCode.MxaccessFailure,
                Message = "MXAccess event conversion failed.",
            },
        };
    }
}