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

using System;
using System.Collections.Generic;
using System.Threading;
using Google.Protobuf.Collections;
using ZB.MOM.WW.MxGateway.Contracts.Proto;

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

/// <summary>
///     Per-session cache of the most recent <c>OnWriteComplete</c> status rows
///     for each (server handle, item handle) pair. Written by the MXAccess
///     event sink as completion callbacks arrive; read by the write command
///     executor so a WriteSecured/WriteSecured2 reply can carry the correlated
///     MXAccess outcome instead of proving command acceptance only.
/// </summary>
/// <remarks>
///     Same threading posture as <see cref="MxAccessValueCache"/>: writers and
///     readers run on the worker's STA thread (COM dispatches events on the
///     apartment thread; commands also execute on the STA), so no internal
///     locking is required. A single sync root keeps it nominally thread-safe
///     for tests that drive it from a non-STA thread.
/// </remarks>
public sealed class MxAccessWriteCompletionCache
{
    private readonly Dictionary<long, CompletionEntry> entries = new();
    private readonly object syncRoot = new();

    /// <summary>Records the status rows of a fresh OnWriteComplete callback for the given handle pair.</summary>
    /// <param name="serverHandle">MXAccess server handle.</param>
    /// <param name="itemHandle">MXAccess item handle.</param>
    /// <param name="statuses">Status rows from the mapped OnWriteComplete event; cloned before storing.</param>
    public void Record(
        int serverHandle,
        int itemHandle,
        RepeatedField<MxStatusProxy> statuses)
    {
        if (statuses is null)
        {
            throw new ArgumentNullException(nameof(statuses));
        }

        lock (syncRoot)
        {
            long key = CreateItemKey(serverHandle, itemHandle);
            ulong version = entries.TryGetValue(key, out CompletionEntry existing)
                ? existing.Version + 1
                : 1UL;
            entries[key] = new CompletionEntry(version, statuses.Clone());
        }
    }

    /// <summary>Returns the current completion version for a handle pair, or 0 if none was recorded.</summary>
    /// <param name="serverHandle">MXAccess server handle.</param>
    /// <param name="itemHandle">MXAccess item handle.</param>
    /// <returns>The current completion version, or 0 if no completion was recorded.</returns>
    public ulong CurrentVersion(
        int serverHandle,
        int itemHandle)
    {
        lock (syncRoot)
        {
            return entries.TryGetValue(CreateItemKey(serverHandle, itemHandle), out CompletionEntry existing)
                ? existing.Version
                : 0UL;
        }
    }

    /// <summary>
    ///     Polls for a completion newer than <paramref name="sinceVersion"/> until it
    ///     arrives or the deadline elapses, calling <paramref name="pumpStep"/> on every
    ///     poll iteration so the worker's STA can dispatch the inbound MXAccess
    ///     OnWriteComplete message. Same loop shape as
    ///     <see cref="MxAccessValueCache.TryWaitForUpdate"/>.
    /// </summary>
    /// <param name="serverHandle">MXAccess server handle.</param>
    /// <param name="itemHandle">MXAccess item handle.</param>
    /// <param name="sinceVersion">Version snapshot captured before the write COM call.</param>
    /// <param name="deadlineUtc">Absolute UTC deadline.</param>
    /// <param name="pumpStep">Action that pumps any pending Windows messages.</param>
    /// <param name="statuses">The recorded status rows if a completion arrived before the deadline; empty otherwise.</param>
    /// <param name="pollIntervalMs">How long to sleep between pump cycles. Default 5 ms.</param>
    /// <returns><see langword="true"/> if a completion newer than <paramref name="sinceVersion"/> arrived before the deadline; otherwise <see langword="false"/>.</returns>
    public bool TryWaitForCompletion(
        int serverHandle,
        int itemHandle,
        ulong sinceVersion,
        DateTime deadlineUtc,
        Action pumpStep,
        out RepeatedField<MxStatusProxy> statuses,
        int pollIntervalMs = 5)
    {
        if (pumpStep is null)
        {
            throw new ArgumentNullException(nameof(pumpStep));
        }

        while (true)
        {
            pumpStep();

            lock (syncRoot)
            {
                if (entries.TryGetValue(CreateItemKey(serverHandle, itemHandle), out CompletionEntry entry)
                    && entry.Version > sinceVersion)
                {
                    statuses = entry.Statuses;
                    return true;
                }
            }

            if (DateTime.UtcNow >= deadlineUtc)
            {
                statuses = new RepeatedField<MxStatusProxy>();
                return false;
            }

            Thread.Sleep(pollIntervalMs);
        }
    }

    private static long CreateItemKey(
        int serverHandle,
        int itemHandle)
    {
        return ((long)serverHandle << 32) | (uint)itemHandle;
    }

    /// <summary>
    ///     Snapshot of the most recent OnWriteComplete status rows for a handle
    ///     pair. <see cref="Version"/> increments by one on every
    ///     <see cref="Record"/> call so the write executor can detect "a new
    ///     completion arrived since I captured my baseline".
    /// </summary>
    /// <remarks>
    ///     Plain readonly struct (not a record) so this compiles under the
    ///     worker's net48 target, which lacks <c>IsExternalInit</c>.
    /// </remarks>
    private readonly struct CompletionEntry
    {
        public CompletionEntry(
            ulong version,
            RepeatedField<MxStatusProxy> statuses)
        {
            Version = version;
            Statuses = statuses;
        }

        public ulong Version { get; }

        public RepeatedField<MxStatusProxy> Statuses { get; }
    }
}