mxaccessgw/src/ZB.MOM.WW.MxGateway.Server/Dashboard/DashboardLiveDataService.cs

using ZB.MOM.WW.MxGateway.Contracts.Proto;
using ZB.MOM.WW.MxGateway.Server.Alarms;
using ZB.MOM.WW.MxGateway.Server.Sessions;

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

/// <summary>
///     Default <see cref="IDashboardLiveDataService"/>. Owns one shared gateway
///     session for the whole dashboard: it is opened lazily on first use and
///     re-opened transparently whenever it faults, is closed, or its lease
///     expires. All access is serialised through <see cref="_gate"/> so the
///     single backing worker only ever sees one in-flight command.
/// </summary>
public sealed class DashboardLiveDataService : IDashboardLiveDataService, IAsyncDisposable
{
    private const string BackendName = "Galaxy";
    private const string ClientName = "mxgateway-dashboard";

    // One browse page of tags plus headroom. Bounds the standing advise load the
    // single dashboard worker carries — and the event churn that advise set feeds —
    // however much of a galaxy an operator browses through in one sitting.
    //
    // The bound is per-read, not absolute: a read may never evict a tag it is itself
    // about to return, so a single read of more distinct tags than the cap leaves the
    // set that large. The invariant EvictForAsync actually maintains is
    //
    //     |advise set| after a read <= max(MaxSubscribedTags, distinct tags in that read)
    //
    // and any overshoot is squeezed back out by the next read that subscribes a tag
    // (see EvictForAsync). A browse page requests far fewer tags than the cap, so in
    // practice the set settles at MaxSubscribedTags.
    private const int MaxSubscribedTags = 256;

    private static readonly TimeSpan ReadTimeout = TimeSpan.FromSeconds(5);

    private readonly ISessionManager _sessionManager;
    private readonly IGatewayAlarmService _alarmService;
    private readonly ILogger<DashboardLiveDataService> _logger;
    private readonly SemaphoreSlim _gate = new(1, 1);

    // Least-recently-read-last advise set: the list holds every currently advised
    // tag ordered most- to least-recently read, the dictionary indexes into it.
    // Both are only ever touched under _gate, which already serialises all viewers.
    private readonly Dictionary<string, LinkedListNode<SubscribedTag>> _subscribed =
        new(StringComparer.OrdinalIgnoreCase);

    private readonly LinkedList<SubscribedTag> _recency = new();

    private GatewaySession? _session;
    private int _serverHandle;
    private bool _disposed;

    /// <summary>Initializes the live-data service.</summary>
    /// <param name="sessionManager">Gateway session manager.</param>
    /// <param name="alarmService">Gateway central alarm service.</param>
    /// <param name="logger">Diagnostic logger.</param>
    public DashboardLiveDataService(
        ISessionManager sessionManager,
        IGatewayAlarmService alarmService,
        ILogger<DashboardLiveDataService> logger)
    {
        _sessionManager = sessionManager ?? throw new ArgumentNullException(nameof(sessionManager));
        _alarmService = alarmService ?? throw new ArgumentNullException(nameof(alarmService));
        _logger = logger ?? throw new ArgumentNullException(nameof(logger));
    }

    /// <inheritdoc />
    public async Task<DashboardLiveReadResult> ReadAsync(
        IReadOnlyCollection<string> tagAddresses,
        CancellationToken cancellationToken)
    {
        ArgumentNullException.ThrowIfNull(tagAddresses);
        if (tagAddresses.Count == 0)
        {
            return DashboardLiveReadResult.Empty;
        }

        await _gate.WaitAsync(cancellationToken).ConfigureAwait(false);
        try
        {
            (GatewaySession session, int serverHandle) = await EnsureReadyAsync(cancellationToken)
                .ConfigureAwait(false);

            string[] toSubscribe = TouchAndCollectNewTags(tagAddresses, out int justReadCount);
            if (toSubscribe.Length > 0)
            {
                await EvictForAsync(session, serverHandle, toSubscribe.Length, justReadCount, cancellationToken)
                    .ConfigureAwait(false);
                IReadOnlyList<SubscribeResult> subscribeResults = await session
                    .SubscribeBulkAsync(serverHandle, toSubscribe, cancellationToken)
                    .ConfigureAwait(false);
                TrackSubscribed(toSubscribe, subscribeResults);
            }

            IReadOnlyList<BulkReadResult> results = await session
                .ReadBulkAsync(serverHandle, tagAddresses.ToArray(), ReadTimeout, cancellationToken)
                .ConfigureAwait(false);

            DashboardTagValue[] values = results
                .Select(DashboardTagValue.FromBulkReadResult)
                .ToArray();
            return new DashboardLiveReadResult(values, null, session.SessionId, session.WorkerProcessId);
        }
        catch (Exception exception) when (exception is not OperationCanceledException)
        {
            InvalidateSession();
            _logger.LogWarning(exception, "Dashboard live read failed; the dashboard session will be re-opened.");
            return new DashboardLiveReadResult([], exception.Message, null, null);
        }
        finally
        {
            _gate.Release();
        }
    }

    /// <inheritdoc />
    public Task<DashboardAlarmQueryResult> QueryAlarmsAsync(CancellationToken cancellationToken)
    {
        // Alarms come from the gateway's always-on central monitor; the
        // dashboard reads its in-process cache directly — no session needed.
        DashboardActiveAlarm[] alarms = _alarmService.CurrentAlarms
            .Select(DashboardActiveAlarm.FromSnapshot)
            .ToArray();

        string? error = _alarmService.State is GatewayAlarmMonitorState.Monitoring
            or GatewayAlarmMonitorState.Disabled
            ? null
            : _alarmService.LastError ?? $"Alarm monitor is {_alarmService.State}.";

        return Task.FromResult(new DashboardAlarmQueryResult(alarms, error, _alarmService.WorkerProcessId));
    }

    // Promotes every already-advised tag in this read to the front of the recency
    // list and returns the tags that still need subscribing (distinct, in request
    // order). `justReadCount` is how many distinct tags of this read were already
    // advised — they now occupy the front of the list and must never be evicted to
    // make room for the same read's new tags. Callers must hold _gate.
    //
    // Every tag of one read is equally recently read; the recency list needs a total
    // order anyway, so the whole service uses one tie-break: later in the request wins.
    // Promoting in request order gives that here, and TrackSubscribed inserts new tags
    // the same way.
    private string[] TouchAndCollectNewTags(IReadOnlyCollection<string> tagAddresses, out int justReadCount)
    {
        int touched = 0;
        List<string> toSubscribe = [];
        HashSet<string> seen = new(StringComparer.OrdinalIgnoreCase);

        foreach (string tag in tagAddresses)
        {
            if (_subscribed.TryGetValue(tag, out LinkedListNode<SubscribedTag>? node))
            {
                if (!ReferenceEquals(node, _recency.First))
                {
                    _recency.Remove(node);
                    _recency.AddFirst(node);
                }

                if (seen.Add(tag))
                {
                    touched++;
                }
            }
            else if (seen.Add(tag))
            {
                toSubscribe.Add(tag);
            }
        }

        justReadCount = touched;
        return [.. toSubscribe];
    }

    // Drops least-recently-read tags off the back of the advise set until the
    // incoming tags fit under MaxSubscribedTags, unadvising them on the worker in
    // one batch. A failed unadvise must not fail the read: the tags are dropped
    // from tracking regardless, and the session-invalidation path already handles
    // gateway/worker drift. Callers must hold _gate.
    //
    // Eviction stops at the tags this read just touched (`justReadCount`), so a read
    // whose own distinct tags outnumber the cap ends over it — see MaxSubscribedTags
    // for the exact invariant. That overshoot is not sticky: the next read that
    // subscribes anything computes `overflow` against the oversized set and evicts the
    // whole excess in one pass (a 300-tag set plus one new tag evicts 45 and lands
    // back at the cap). A read that subscribes nothing new evicts nothing, but it also
    // cannot grow the set.
    //
    // Cancellation mid-eviction follows this file's policy: OperationCanceledException
    // is deliberately not caught here or in ReadAsync, so it propagates with the tags
    // already dropped from tracking — the same end state as a failed unadvise.
    private async Task EvictForAsync(
        GatewaySession session,
        int serverHandle,
        int incomingCount,
        int justReadCount,
        CancellationToken cancellationToken)
    {
        int overflow = _subscribed.Count + incomingCount - MaxSubscribedTags;
        int evictable = _subscribed.Count - justReadCount;
        int evictCount = Math.Min(overflow, evictable);
        if (evictCount <= 0)
        {
            return;
        }

        List<int> evictedHandles = new(evictCount);
        for (int i = 0; i < evictCount && _recency.Last is { } oldest; i++)
        {
            _recency.RemoveLast();
            _subscribed.Remove(oldest.Value.TagAddress);
            if (oldest.Value.ItemHandle != 0)
            {
                evictedHandles.Add(oldest.Value.ItemHandle);
            }
        }

        _logger.LogDebug(
            "Dashboard advise set hit its cap of {Cap}; evicted {EvictedCount} least-recently-read tags.",
            MaxSubscribedTags,
            evictCount);

        if (evictedHandles.Count == 0)
        {
            return;
        }

        try
        {
            await session.UnsubscribeBulkAsync(serverHandle, evictedHandles, cancellationToken)
                .ConfigureAwait(false);
        }
        catch (Exception exception) when (exception is not OperationCanceledException)
        {
            _logger.LogDebug(
                exception,
                "Unadvising {EvictedCount} evicted dashboard tags failed; they stay dropped from tracking.",
                evictedHandles.Count);
        }
    }

    // Records the freshly advised tags as the most recently read, keeping each
    // tag's item handle so eviction can unadvise it. Tags the worker failed to
    // advise are still tracked (matching the pre-cap behaviour of not retrying
    // them on every read) but carry no handle, so eviction just forgets them.
    // Callers must hold _gate.
    private void TrackSubscribed(IReadOnlyList<string> tagAddresses, IReadOnlyList<SubscribeResult> results)
    {
        Dictionary<string, int> handles = new(results.Count, StringComparer.OrdinalIgnoreCase);
        foreach (SubscribeResult result in results)
        {
            if (result.WasSuccessful && !string.IsNullOrEmpty(result.TagAddress))
            {
                handles[result.TagAddress] = result.ItemHandle;
            }
        }

        // Request order, so the read's last tag ends up most recent — the same
        // tie-break TouchAndCollectNewTags applies to the tags it promotes.
        foreach (string tag in tagAddresses)
        {
            handles.TryGetValue(tag, out int itemHandle);
            _subscribed[tag] = _recency.AddFirst(new SubscribedTag(tag, itemHandle));
        }
    }

    // Forgets the whole advise set without unadvising: every call site is one where
    // the backing session (and with it every item handle) is already gone.
    // Callers must hold _gate.
    private void ClearSubscriptions()
    {
        _subscribed.Clear();
        _recency.Clear();
    }

    // Returns a Ready session + its Register server handle, opening a fresh
    // session when none exists or the current one is no longer usable. Callers
    // must hold _gate.
    private async Task<(GatewaySession Session, int ServerHandle)> EnsureReadyAsync(
        CancellationToken cancellationToken)
    {
        ObjectDisposedException.ThrowIf(_disposed, this);

        GatewaySession? existing = _session;
        if (existing is not null
            && existing.State == SessionState.Ready
            && _sessionManager.TryGetSession(existing.SessionId, out _))
        {
            return (existing, _serverHandle);
        }

        if (existing is not null)
        {
            _logger.LogInformation(
                "Dashboard session {SessionId} is no longer usable (state {State}); re-opening.",
                existing.SessionId,
                existing.State);
            await CloseQuietlyAsync(existing.SessionId).ConfigureAwait(false);
        }

        ClearSubscriptions();
        _session = null;

        GatewaySession session = await _sessionManager.OpenSessionAsync(
                new SessionOpenRequest(BackendName, ClientName, Guid.NewGuid().ToString("N"), CommandTimeout: null),
                ClientName,
                ownerKeyId: null,
                cancellationToken)
            .ConfigureAwait(false);

        WorkerCommandReply reply = await session.InvokeAsync(
                new WorkerCommand
                {
                    Command = new MxCommand
                    {
                        Kind = MxCommandKind.Register,
                        Register = new RegisterCommand { ClientName = ClientName },
                    },
                },
                cancellationToken)
            .ConfigureAwait(false);

        int? serverHandle = reply.Reply?.Register?.ServerHandle;
        if (serverHandle is null)
        {
            string diagnostic = reply.Reply?.ProtocolStatus?.Message
                ?? reply.Reply?.DiagnosticMessage
                ?? "Worker did not return a server handle for Register.";
            await CloseQuietlyAsync(session.SessionId).ConfigureAwait(false);
            throw new InvalidOperationException($"Dashboard session registration failed: {diagnostic}");
        }

        _session = session;
        _serverHandle = serverHandle.Value;
        _logger.LogInformation(
            "Dashboard session {SessionId} opened (worker pid {WorkerPid}).",
            session.SessionId,
            session.WorkerProcessId);
        return (session, _serverHandle);
    }

    // Drops the cached session so the next call re-opens. Callers must hold _gate.
    private void InvalidateSession()
    {
        _session = null;
        _serverHandle = 0;
        ClearSubscriptions();
    }

    private async Task CloseQuietlyAsync(string sessionId)
    {
        try
        {
            await _sessionManager.CloseSessionAsync(sessionId, CancellationToken.None).ConfigureAwait(false);
        }
        catch (Exception exception)
        {
            _logger.LogDebug(exception, "Closing stale dashboard session {SessionId} failed.", sessionId);
        }
    }

    /// <summary>Closes the underlying gateway session, if one is open.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    public async ValueTask DisposeAsync()
    {
        if (_disposed)
        {
            return;
        }

        _disposed = true;
        GatewaySession? session = _session;
        _session = null;
        if (session is not null)
        {
            await CloseQuietlyAsync(session.SessionId).ConfigureAwait(false);
        }

        _gate.Dispose();
    }

    // One entry of the advise set. ItemHandle is the handle the worker bound for
    // the tag, or 0 when the subscribe failed and there is nothing to unadvise.
    private readonly record struct SubscribedTag(string TagAddress, int ItemHandle);
}