ScadaBridge/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Adapters/RealMxGatewayClient.cs

using System.Collections.Concurrent;
using System.Globalization;
using Grpc.Core;
using Microsoft.Extensions.Logging;
using ZB.MOM.WW.MxGateway.Client;
using ZB.MOM.WW.MxGateway.Contracts.Proto;
using ZB.MOM.WW.MxGateway.Contracts.Proto.Galaxy;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Protocol;

namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Adapters;

/// <summary>
/// Production <see cref="IMxGatewayClient"/> implementation over the
/// <c>ZB.MOM.WW.MxGateway.Client</c> NuGet package. This is the only type in the
/// Data Connection Layer that references the generated gRPC/protobuf contracts;
/// the adapter and its tests run entirely against the neutral seam.
/// </summary>
public sealed class RealMxGatewayClient : IMxGatewayClient
{
    private readonly ILogger<RealMxGatewayClient> _logger;
    private readonly ILoggerFactory? _loggerFactory;

    private MxGatewayClient? _client;
    private GalaxyRepositoryClient? _galaxy;
    private MxGatewaySession? _session;
    private int _serverHandle;
    private int _writeUserId;
    private int _readTimeoutMs;
    private int _supervisoryAdviseParallelism = 16;
    private ulong _lastSeq;

    // tag ↔ MXAccess item handle, maintained across subscribe/write.
    private readonly ConcurrentDictionary<string, int> _tagToHandle = new();
    private readonly ConcurrentDictionary<int, string> _handleToTag = new();

    // Item handles already advised in SUPERVISORY mode (advise-once tracking).
    // MXAccess only accepts a write on an item that holds a supervisory advise; a
    // plain Advise is not sufficient and the COM Write blocks until the gateway
    // command timeout. When the connection has no write-user context
    // (WriteUserId == 0) we therefore advise supervisory by DEFAULT — for both the
    // monitoring subscription and any write — so the one connection can read and
    // write. A supervisory advise still delivers OnDataChange (monitoring is
    // unaffected), and the worker's UnAdvise tears down either advice kind. The
    // Lazy<Task> per handle guarantees the advise runs at most once even under a
    // concurrent first-time subscribe+write on the same new handle: every racing
    // caller awaits the same Task, so none reaches WriteBulk before the advise has
    // completed. A faulted advise is evicted so the next write retries.
    private readonly ConcurrentDictionary<int, Lazy<Task>> _supervisoryAdvised = new();

    // With no configured MXAccess write-user this connection behaves as a
    // supervisory client: every advise defaults to AdviseSupervisory so writes are
    // accepted. A configured write-user keeps the prior plain-advise behaviour.
    private bool UseSupervisoryAdvise => _writeUserId == 0;

    /// <summary>Initializes a new instance of <see cref="RealMxGatewayClient"/>.</summary>
    /// <param name="loggerFactory">Logger factory shared with the gateway client.</param>
    public RealMxGatewayClient(ILoggerFactory? loggerFactory)
    {
        _loggerFactory = loggerFactory;
        _logger = (loggerFactory ?? Microsoft.Extensions.Logging.Abstractions.NullLoggerFactory.Instance)
            .CreateLogger<RealMxGatewayClient>();
    }

    /// <inheritdoc />
    public async Task ConnectAsync(MxGatewayConnectionOptions options, CancellationToken ct = default)
    {
        _writeUserId = options.WriteUserId;
        _readTimeoutMs = options.ReadTimeoutMs;
        _supervisoryAdviseParallelism = Math.Max(1, options.SupervisoryAdviseParallelism);

        var clientOptions = new MxGatewayClientOptions
        {
            Endpoint = new Uri(options.Endpoint),
            ApiKey = options.ApiKey,
            UseTls = options.UseTls,
            CaCertificatePath = options.CaFile,
            ServerNameOverride = options.ServerName,
            LoggerFactory = _loggerFactory,
        };

        _client = MxGatewayClient.Create(clientOptions);
        _galaxy = GalaxyRepositoryClient.Create(clientOptions);
        _session = await _client.OpenSessionAsync(cancellationToken: ct).ConfigureAwait(false);
        _serverHandle = await _session.RegisterAsync(options.ClientName, ct).ConfigureAwait(false);
    }

    /// <inheritdoc />
    public async Task DisconnectAsync(CancellationToken ct = default)
    {
        if (_session is not null)
            await _session.CloseAsync(ct).ConfigureAwait(false);
    }

    /// <inheritdoc />
    public async Task<string> SubscribeAsync(string tagPath, CancellationToken ct = default)
    {
        var handle = await GetOrAddItemHandleAsync(tagPath, ct).ConfigureAwait(false);
        // With no write-user context, advise supervisory by default so the same
        // subscription also makes the item writable; otherwise plain monitor-advise.
        if (UseSupervisoryAdvise)
            await EnsureSupervisoryAdvisedAsync(handle, ct).ConfigureAwait(false);
        else
            await _session!.AdviseAsync(_serverHandle, handle, ct).ConfigureAwait(false);
        return handle.ToString(CultureInfo.InvariantCulture);
    }

    /// <inheritdoc />
    public async Task<IReadOnlyList<MxSubscribeOutcome>> SubscribeBulkAsync(
        IReadOnlyList<string> tagPaths, CancellationToken ct = default)
    {
        if (tagPaths.Count == 0)
            return [];

        // Plain-advise mode (a configured write user): the worker's SubscribeBulk does
        // AddItem + Advise for every tag in ONE gateway command, collapsing the historical
        // 2 RPCs per tag to 1 RPC per chunk.
        if (!UseSupervisoryAdvise)
        {
            var replies = await _session!.SubscribeBulkAsync(_serverHandle, tagPaths, ct).ConfigureAwait(false);
            return MapSubscribeResults(tagPaths, replies);
        }

        // Supervisory mode (WriteUserId == 0): the worker has no BULK supervisory advise
        // (SubscribeBulk issues plain Advise), so bulk-add the items in one command and
        // pipeline the per-item AdviseSupervisory commands with a bounded window instead
        // of issuing them serially. Cross-repo follow-up: an additive `supervisory` flag on
        // SubscribeBulkCommand in mxaccessgw would collapse this to one RPC as well.
        var added = await _session!.AddItemBulkAsync(_serverHandle, tagPaths, ct).ConfigureAwait(false);
        var outcomes = MapSubscribeResults(tagPaths, added);

        var advisable = outcomes
            .Select((o, i) => (Outcome: o, Index: i))
            .Where(x => x.Outcome.Success)
            .ToList();

        var advised = await BulkPipeline.RunAsync(
            advisable,
            _supervisoryAdviseParallelism,
            async (x, token) =>
            {
                var handle = int.Parse(x.Outcome.SubscriptionId!, NumberStyles.Integer, CultureInfo.InvariantCulture);
                await EnsureSupervisoryAdvisedAsync(handle, token).ConfigureAwait(false);
                return x.Outcome;
            },
            (x, ex) => x.Outcome with { Success = false, SubscriptionId = null, Error = ex.Message },
            ct).ConfigureAwait(false);

        var result = outcomes.ToArray();
        for (var i = 0; i < advisable.Count; i++)
            result[advisable[i].Index] = advised[i];

        // A failed advise leaves an added-but-unadvised item; drop our handle mapping so a
        // later retry re-adds cleanly rather than reusing a half-subscribed handle.
        foreach (var failed in result.Where(r => !r.Success))
        {
            if (_tagToHandle.TryRemove(failed.TagPath, out var staleHandle))
                _handleToTag.TryRemove(staleHandle, out _);
        }

        return result;
    }

    /// <summary>
    /// Maps the gateway's per-item bulk results back onto the requested tag paths. Results
    /// are returned in request order; the tag address on the reply is preferred when the
    /// gateway echoes it. Successful rows populate the tag ↔ item-handle maps that the
    /// event loop and the write path key off.
    /// </summary>
    private MxSubscribeOutcome[] MapSubscribeResults(
        IReadOnlyList<string> tagPaths, IReadOnlyList<SubscribeResult> replies)
    {
        var outcomes = new MxSubscribeOutcome[tagPaths.Count];
        for (var i = 0; i < tagPaths.Count; i++)
        {
            var tag = tagPaths[i];
            if (i >= replies.Count)
            {
                outcomes[i] = new MxSubscribeOutcome(tag, false, null, "Gateway returned no result for this tag.");
                continue;
            }

            var reply = replies[i];
            if (!reply.WasSuccessful)
            {
                outcomes[i] = new MxSubscribeOutcome(tag, false, null,
                    string.IsNullOrEmpty(reply.ErrorMessage) ? "Bulk subscribe failed." : reply.ErrorMessage);
                continue;
            }

            _tagToHandle[tag] = reply.ItemHandle;
            _handleToTag[reply.ItemHandle] = tag;
            outcomes[i] = new MxSubscribeOutcome(
                tag, true, reply.ItemHandle.ToString(CultureInfo.InvariantCulture), null);
        }

        return outcomes;
    }

    /// <inheritdoc />
    public async Task UnsubscribeBulkAsync(IReadOnlyList<string> subscriptionIds, CancellationToken ct = default)
    {
        var handles = new List<int>(subscriptionIds.Count);
        foreach (var id in subscriptionIds)
        {
            if (int.TryParse(id, NumberStyles.Integer, CultureInfo.InvariantCulture, out var handle))
                handles.Add(handle);
        }

        if (handles.Count == 0)
            return;

        await _session!.UnsubscribeBulkAsync(_serverHandle, handles, ct).ConfigureAwait(false);

        foreach (var handle in handles)
        {
            // UnsubscribeBulk unadvises AND removes the item, releasing every advice kind.
            _supervisoryAdvised.TryRemove(handle, out _);
            if (_handleToTag.TryRemove(handle, out var tag))
                ((ICollection<KeyValuePair<string, int>>)_tagToHandle).Remove(new(tag, handle));
        }
    }

    /// <inheritdoc />
    public async Task UnsubscribeAsync(string subscriptionId, CancellationToken ct = default)
    {
        if (!int.TryParse(subscriptionId, NumberStyles.Integer, CultureInfo.InvariantCulture, out var handle))
            return;

        await _session!.UnAdviseAsync(_serverHandle, handle, ct).ConfigureAwait(false);
        await _session.RemoveItemAsync(_serverHandle, handle, ct).ConfigureAwait(false);

        // RemoveItem releases the item and all its advices (plain + supervisory),
        // so forget any supervisory-advise state for this handle.
        _supervisoryAdvised.TryRemove(handle, out _);
        if (_handleToTag.TryRemove(handle, out var tag))
            ((ICollection<KeyValuePair<string, int>>)_tagToHandle).Remove(new(tag, handle));
    }

    /// <inheritdoc />
    public async Task<IReadOnlyList<MxReadOutcome>> ReadAsync(IReadOnlyList<string> tagPaths, CancellationToken ct = default)
    {
        var results = await _session!
            .ReadBulkAsync(_serverHandle, tagPaths, TimeSpan.FromMilliseconds(_readTimeoutMs), ct)
            .ConfigureAwait(false);

        return results.Select(r => new MxReadOutcome(
            r.TagAddress,
            r.WasSuccessful,
            r.WasSuccessful ? r.Value?.ToClrValue() : null,
            MapQuality(r.Quality, r.Statuses),
            r.SourceTimestamp?.ToDateTimeOffset() ?? DateTimeOffset.UtcNow,
            r.WasSuccessful ? null : r.ErrorMessage)).ToList();
    }

    /// <inheritdoc />
    public async Task<IReadOnlyList<MxWriteOutcome>> WriteAsync(IReadOnlyList<(string TagPath, object? Value)> writes, CancellationToken ct = default)
    {
        // Build entries in request order; remember the tag for each handle so the
        // per-handle BulkWriteResult can be mapped back to its tag.
        var entries = new List<WriteBulkEntry>(writes.Count);
        var orderedTags = new List<string>(writes.Count);
        foreach (var (tag, value) in writes)
        {
            // MXAccess addresses a WHOLE array attribute with a trailing "[]" on the
            // item reference. A write to the bare reference is silently dropped — the
            // COM Write returns success but the value never commits (reads work
            // either way). This mirrors the AVEVA MES Camstar API, which registers
            // array tags as "<object>.<attr>[]". Scalars keep the bare reference.
            var writeRef = IsArrayValue(value) ? tag + "[]" : tag;
            var handle = await GetOrAddItemHandleAsync(writeRef, ct).ConfigureAwait(false);
            // MXAccess requires a supervisory advise on the item before it will
            // accept a write; without it the worker's synchronous COM Write blocks.
            // With no write-user context we advise supervisory by default (a
            // supervisory subscription may already cover this handle — advise-once).
            if (UseSupervisoryAdvise)
                await EnsureSupervisoryAdvisedAsync(handle, ct).ConfigureAwait(false);
            // MXAccess SAFEARRAY writes must supply the attribute's FULL declared
            // length; a short array (only the N elements the caller provided) is
            // rejected and the COM write blocks. Pad list values out to the live
            // array's slot count before encoding (callers signal the valid count
            // via a separate scalar, e.g. MoveInNumberWorkOrders).
            var toWrite = await PadArrayToDeclaredSizeAsync(writeRef, value, ct).ConfigureAwait(false);
            entries.Add(new WriteBulkEntry
            {
                ItemHandle = handle,
                Value = ToMxValue(toWrite),
                UserId = _writeUserId,
            });
            // Map the per-handle result back to the ORIGINAL caller tag (not writeRef).
            orderedTags.Add(tag);
        }

        var results = await _session!.WriteBulkAsync(_serverHandle, entries, ct).ConfigureAwait(false);

        // Results are returned in request order; pair by index back to the tags.
        return results.Select((r, i) => new MxWriteOutcome(
            i < orderedTags.Count ? orderedTags[i] : (_handleToTag.TryGetValue(r.ItemHandle, out var t) ? t : ""),
            r.WasSuccessful,
            r.WasSuccessful ? null : r.ErrorMessage)).ToList();
    }

    /// <inheritdoc />
    public async Task<(IReadOnlyList<MxBrowseChild> Children, bool Truncated)> BrowseChildrenAsync(string? parentNodeId, CancellationToken ct = default)
    {
        // Navigation browse returns child OBJECTS only (IncludeAttributes = false).
        // An object's own attributes are loaded lazily — via DiscoverHierarchy below,
        // scoped to that single object — only when the object is expanded. This keeps
        // each browse level's payload to a few KB. Inlining every child's full
        // attribute set (the previous behaviour) produced replies that exceeded
        // Akka's remote frame size (e.g. ~152 KB for one attribute-heavy area), and
        // remoting silently discarded the oversized reply, hanging the picker.
        var request = new BrowseChildrenRequest { IncludeAttributes = false };

        // Object NodeIds are the Galaxy gobject id (encoded as a string); attribute
        // NodeIds are FullTagReference leaves and never arrive here as a parent.
        var parentGobjectId = 0;
        var haveParentObject = !string.IsNullOrEmpty(parentNodeId)
            && int.TryParse(parentNodeId, NumberStyles.Integer, CultureInfo.InvariantCulture, out parentGobjectId);
        if (haveParentObject)
        {
            request.ParentGobjectId = parentGobjectId;
        }

        BrowseChildrenReply reply;
        GalaxyObject? parentObject = null;
        try
        {
            reply = await _galaxy!.BrowseChildrenRawAsync(request, ct).ConfigureAwait(false);

            // When expanding a concrete object, fetch just that object (MaxDepth = 0)
            // with its attributes so they can be listed as selectable leaves. The root
            // level has no parent object, hence no attributes of its own.
            if (haveParentObject)
            {
                var attrRequest = new DiscoverHierarchyRequest
                {
                    RootGobjectId = parentGobjectId,
                    MaxDepth = 0,
                    IncludeAttributes = true,
                };
                var attrReply = await _galaxy!.DiscoverHierarchyRawAsync(attrRequest, ct).ConfigureAwait(false);
                parentObject = attrReply.Objects.Count > 0 ? attrReply.Objects[0] : null;
            }
        }
        catch (RpcException ex) when (ex.StatusCode == StatusCode.Unavailable)
        {
            throw new ConnectionNotConnectedException($"MxGateway repository unavailable: {ex.Status.Detail}");
        }
        catch (RpcException ex) when (ex.StatusCode == StatusCode.Unimplemented)
        {
            // The data pipe (read/subscribe/write) works against every gateway
            // build, but Galaxy hierarchy browsing (the BrowseChildren RPC) was
            // added later. An older gateway answers Unimplemented. Surface a
            // clear, actionable reason instead of a raw gRPC fault — the actor
            // maps NotSupportedException to BrowseFailureKind.NotBrowsable and
            // carries this message through to the picker.
            throw new NotSupportedException(
                "The connected MxGateway build does not support hierarchy browsing. "
                + "Update the gateway to a build that implements BrowseChildren, "
                + "or enter the tag reference manually.");
        }

        var children = new List<MxBrowseChild>();

        // Navigable child objects, keyed by gobject id. Always marked expandable —
        // every Galaxy object carries attributes (and may host sub-objects), both
        // resolved on demand when the node is expanded.
        foreach (var obj in reply.Children)
        {
            children.Add(new MxBrowseChild(
                obj.GobjectId.ToString(CultureInfo.InvariantCulture),
                string.IsNullOrEmpty(obj.TagName) ? obj.ContainedName : obj.TagName,
                BrowseNodeClass.Object,
                true));
        }

        // The expanded object's own attributes, as selectable leaves keyed by their
        // full tag reference.
        if (parentObject is not null)
        {
            foreach (var attr in parentObject.Attributes)
            {
                children.Add(new MxBrowseChild(
                    attr.FullTagReference,
                    attr.AttributeName,
                    BrowseNodeClass.Variable,
                    false,
                    string.IsNullOrEmpty(attr.DataTypeName) ? null : attr.DataTypeName));
            }
        }

        return (children, !string.IsNullOrEmpty(reply.NextPageToken));
    }

    /// <inheritdoc />
    public async Task RunEventLoopAsync(Action<MxValueUpdate> onUpdate, CancellationToken ct = default)
    {
        await foreach (var ev in _session!.StreamEventsAsync(_lastSeq, ct).ConfigureAwait(false))
        {
            _lastSeq = ev.WorkerSequence;
            if (ev.Family != MxEventFamily.OnDataChange)
                continue;
            if (!_handleToTag.TryGetValue(ev.ItemHandle, out var tag))
                continue;

            onUpdate(new MxValueUpdate(
                tag,
                ev.Value?.ToClrValue(),
                MapQuality(ev.Quality, ev.Statuses),
                ev.SourceTimestamp?.ToDateTimeOffset() ?? DateTimeOffset.UtcNow));
        }
    }

    /// <inheritdoc />
    public async Task RunAlarmStreamAsync(
        string? alarmFilterPrefix, Action<Commons.Types.Alarms.NativeAlarmTransition> onTransition,
        CancellationToken ct = default)
    {
        var reconnectDelay = TimeSpan.FromSeconds(5);
        while (!ct.IsCancellationRequested)
        {
            try
            {
                var request = new StreamAlarmsRequest
                {
                    ClientCorrelationId = Guid.NewGuid().ToString("N"),
                    AlarmFilterPrefix = alarmFilterPrefix ?? string.Empty,
                };

                // Every transport reopen produces a fresh authoritative frame,
                // including an empty snapshot after the previous stream faulted.
                onTransition(MxGatewayAlarmMapper.SnapshotComplete() with { Kind = Commons.Types.Enums.AlarmTransitionKind.SnapshotStart });
                await foreach (var message in _client!.StreamAlarmsAsync(request, ct)
                    .WithCancellation(ct).ConfigureAwait(false))
                {
                    if (ct.IsCancellationRequested) break;
                    switch (message.PayloadCase)
                    {
                        case AlarmFeedMessage.PayloadOneofCase.ActiveAlarm:
                            onTransition(MxGatewayAlarmMapper.MapSnapshot(message.ActiveAlarm));
                            break;
                        case AlarmFeedMessage.PayloadOneofCase.Transition:
                            onTransition(MxGatewayAlarmMapper.MapTransition(message.Transition));
                            break;
                        case AlarmFeedMessage.PayloadOneofCase.SnapshotComplete:
                            onTransition(MxGatewayAlarmMapper.SnapshotComplete());
                            break;
                    }
                }
            }
            catch (OperationCanceledException) when (ct.IsCancellationRequested)
            {
                return; // clean shutdown
            }
            catch (Exception ex)
            {
                _logger.LogWarning(ex,
                    "MxGateway alarm stream faulted; reopening in {DelaySeconds}s", reconnectDelay.TotalSeconds);
            }

            try
            {
                await Task.Delay(reconnectDelay, ct).ConfigureAwait(false);
            }
            catch (OperationCanceledException) when (ct.IsCancellationRequested)
            {
                return;
            }
        }
    }

    /// <summary>
    /// Asynchronously disposes the active MxAccess session, client, and galaxy resources.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    public async ValueTask DisposeAsync()
    {
        if (_session is not null) await _session.DisposeAsync().ConfigureAwait(false);
        if (_client is not null) await _client.DisposeAsync().ConfigureAwait(false);
        if (_galaxy is not null) await _galaxy.DisposeAsync().ConfigureAwait(false);
    }

    private async Task<int> GetOrAddItemHandleAsync(string tagPath, CancellationToken ct)
    {
        if (_tagToHandle.TryGetValue(tagPath, out var existing))
            return existing;

        var handle = await _session!.AddItemAsync(_serverHandle, tagPath, ct).ConfigureAwait(false);
        _tagToHandle[tagPath] = handle;
        _handleToTag[handle] = tagPath;
        return handle;
    }

    /// <summary>
    /// MXAccess fixed-size SAFEARRAY attributes only accept a write that supplies
    /// the whole array; a short list (just the caller's N elements) is rejected and
    /// the COM write blocks. For a list value, read the tag's current array to learn
    /// its slot count and pad the value to that length with element-type defaults
    /// (empty string / 0 / false / default) — the caller's values fill slots 0..N-1,
    /// the remainder are cleared. Scalars (and a non-array / unreadable tag) pass
    /// through unchanged.
    /// </summary>
    private async Task<object?> PadArrayToDeclaredSizeAsync(string tag, object? value, CancellationToken ct)
    {
        switch (value)
        {
            case IReadOnlyList<string> v: return await PadAsync(tag, v, string.Empty, ct).ConfigureAwait(false);
            case IReadOnlyList<int> v: return await PadAsync(tag, v, 0, ct).ConfigureAwait(false);
            case IReadOnlyList<long> v: return await PadAsync(tag, v, 0L, ct).ConfigureAwait(false);
            case IReadOnlyList<float> v: return await PadAsync(tag, v, 0f, ct).ConfigureAwait(false);
            case IReadOnlyList<double> v: return await PadAsync(tag, v, 0d, ct).ConfigureAwait(false);
            case IReadOnlyList<bool> v: return await PadAsync(tag, v, false, ct).ConfigureAwait(false);
            case IReadOnlyList<DateTimeOffset> v: return await PadAsync(tag, v, default(DateTimeOffset), ct).ConfigureAwait(false);
            case IReadOnlyList<DateTime> v: return await PadAsync(tag, v, default(DateTime), ct).ConfigureAwait(false);
            default: return value; // scalar or unsupported — leave as-is
        }
    }

    /// <summary>
    /// Whether a write value is a multi-element (array) attribute value — i.e. one
    /// <see cref="ToMxValue"/> encodes as an MXAccess array. Such attributes must be
    /// addressed with a trailing "[]" on the item reference (see <c>WriteAsync</c>).
    /// The set matches the array cases in <see cref="ToMxValue"/> /
    /// <see cref="PadArrayToDeclaredSizeAsync"/>; scalars (string, int, bool, …) are false.
    /// </summary>
    private static bool IsArrayValue(object? value) => value switch
    {
        IReadOnlyList<bool> => true,
        IReadOnlyList<int> => true,
        IReadOnlyList<long> => true,
        IReadOnlyList<float> => true,
        IReadOnlyList<double> => true,
        IReadOnlyList<string> => true,
        IReadOnlyList<DateTimeOffset> => true,
        IReadOnlyList<DateTime> => true,
        _ => false,
    };

    private async Task<IReadOnlyList<T>> PadAsync<T>(string tag, IReadOnlyList<T> values, T pad, CancellationToken ct)
    {
        var size = await GetArraySlotCountAsync(tag, ct).ConfigureAwait(false);
        if (size <= 0)
        {
            // Couldn't determine the declared size (read failed or tag is not an
            // array). Write what we have rather than blocking the caller; if the
            // node really is a short SAFEARRAY the gateway will surface the error.
            _logger.LogWarning("Could not determine array size for '{Tag}'; writing {Count} element(s) unpadded", tag, values.Count);
            return values;
        }

        if (values.Count > size)
            _logger.LogWarning("Array write for '{Tag}' has {Count} elements but the node holds {Size}; truncating", tag, values.Count, size);

        var result = new List<T>(size);
        for (var i = 0; i < size; i++)
            result.Add(i < values.Count ? values[i] : pad);
        return result;
    }

    /// <summary>Reads the tag's current array element count (its declared SAFEARRAY length), or 0 if not an array / unreadable.</summary>
    private async Task<int> GetArraySlotCountAsync(string tag, CancellationToken ct)
    {
        var reads = await _session!
            .ReadBulkAsync(_serverHandle, new[] { tag }, TimeSpan.FromMilliseconds(_readTimeoutMs), ct)
            .ConfigureAwait(false);
        var v = reads.Count > 0 ? reads[0].Value : null;
        if (v is null || v.KindCase != MxValue.KindOneofCase.ArrayValue)
            return 0;

        var a = v.ArrayValue;
        return a.ValuesCase switch
        {
            MxArray.ValuesOneofCase.StringValues => a.StringValues.Values.Count,
            MxArray.ValuesOneofCase.Int32Values => a.Int32Values.Values.Count,
            MxArray.ValuesOneofCase.Int64Values => a.Int64Values.Values.Count,
            MxArray.ValuesOneofCase.FloatValues => a.FloatValues.Values.Count,
            MxArray.ValuesOneofCase.DoubleValues => a.DoubleValues.Values.Count,
            MxArray.ValuesOneofCase.BoolValues => a.BoolValues.Values.Count,
            MxArray.ValuesOneofCase.TimestampValues => a.TimestampValues.Values.Count,
            _ => 0,
        };
    }

    /// <summary>
    /// Ensures the item is advised in MXAccess SUPERVISORY mode (advise-once). A
    /// supervisory advise is what lets the gateway accept a write and it still
    /// delivers OnDataChange, so it doubles as the monitoring subscription when this
    /// connection has no write-user context. The client package exposes only a plain
    /// <c>AdviseAsync</c>, so the supervisory advise is issued as a raw
    /// <see cref="MxCommandKind.AdviseSupervisory"/> command (mirroring how the session
    /// builds every other command). The result is remembered per handle; a failure is
    /// not cached so the next attempt retries rather than wedging the item permanently.
    /// </summary>
    private async Task EnsureSupervisoryAdvisedAsync(int handle, CancellationToken ct)
    {
        // GetOrAdd + Lazy runs the advise at most once per handle; concurrent
        // callers await the same Task and so never reach WriteBulk before it
        // completes (default Lazy is thread-safe, factory runs once).
        var advise = _supervisoryAdvised.GetOrAdd(
            handle,
            h => new Lazy<Task>(() => AdviseSupervisoryCoreAsync(h, ct)));
        try
        {
            await advise.Value.ConfigureAwait(false);
        }
        catch
        {
            // Evict the faulted entry (only if it's still the one we awaited) so the
            // next write retries rather than caching a transient advise failure.
            _supervisoryAdvised.TryRemove(new KeyValuePair<int, Lazy<Task>>(handle, advise));
            throw;
        }
    }

    private async Task AdviseSupervisoryCoreAsync(int handle, CancellationToken ct)
    {
        var reply = await _session!.InvokeAsync(
            new MxCommandRequest
            {
                SessionId = _session.SessionId,
                ClientCorrelationId = Guid.NewGuid().ToString("N"),
                Command = new MxCommand
                {
                    Kind = MxCommandKind.AdviseSupervisory,
                    AdviseSupervisory = new AdviseSupervisoryCommand
                    {
                        ServerHandle = _serverHandle,
                        ItemHandle = handle,
                    },
                },
            },
            ct).ConfigureAwait(false);
        reply.EnsureProtocolSuccess().EnsureMxAccessSuccess();
    }

    /// <summary>
    /// Maps MXAccess quality. A failing status proxy is authoritative bad; otherwise
    /// the OPC-style quality byte: ≥192 Good, ≥64 Uncertain, else Bad.
    /// </summary>
    private static QualityCode MapQuality(int quality, IEnumerable<MxStatusProxy> statuses)
    {
        if (statuses.Any(s => !s.IsSuccess()))
            return QualityCode.Bad;
        return quality switch
        {
            >= 192 => QualityCode.Good,
            >= 64 => QualityCode.Uncertain,
            _ => QualityCode.Bad,
        };
    }

    private static MxValue ToMxValue(object? value) => value switch
    {
        null => new MxValue { IsNull = true },
        bool b => b.ToMxValue(),
        int i => i.ToMxValue(),
        long l => l.ToMxValue(),
        float f => f.ToMxValue(),
        double d => d.ToMxValue(),
        string s => s.ToMxValue(),
        DateTimeOffset dto => dto.ToMxValue(),
        DateTime dt => dt.ToMxValue(),
        // List/array attribute values. The InstanceActor decodes a canonical JSON
        // List attribute into a typed List<T> (AttributeValueCodec produces
        // List<string|int|float|double|bool|DateTime>) before the write so the DCL
        // can push a real MXAccess array. Map each to the client's typed array
        // encoder. Without these a List<T> would fall through to Convert.ToString
        // and write a garbage scalar string to an array node — which the gateway's
        // COM write then rejects/blocks (the array-write hang).
        IReadOnlyList<bool> lb => lb.ToMxValue(),
        IReadOnlyList<int> li => li.ToMxValue(),
        IReadOnlyList<long> ll => ll.ToMxValue(),
        IReadOnlyList<float> lf => lf.ToMxValue(),
        IReadOnlyList<double> ld => ld.ToMxValue(),
        IReadOnlyList<string> ls => ls.ToMxValue(),
        IReadOnlyList<DateTimeOffset> ldto => ldto.ToMxValue(),
        IReadOnlyList<DateTime> ldt => ldt.Select(d => (DateTimeOffset)d).ToList().ToMxValue(),
        // Fall back to invariant string for any other CLR type.
        _ => Convert.ToString(value, CultureInfo.InvariantCulture)!.ToMxValue(),
    };
}

/// <summary>Builds <see cref="RealMxGatewayClient"/> instances.</summary>
public sealed class RealMxGatewayClientFactory : IMxGatewayClientFactory
{
    private readonly ILoggerFactory? _loggerFactory;

    /// <summary>Initializes a new factory.</summary>
    /// <param name="loggerFactory">Logger factory passed to each created client.</param>
    public RealMxGatewayClientFactory(ILoggerFactory? loggerFactory) => _loggerFactory = loggerFactory;

    /// <inheritdoc />
    public IMxGatewayClient Create() => new RealMxGatewayClient(_loggerFactory);
}