lmxopcua/src/Client/ZB.MOM.WW.OtOpcUa.Client.Shared/Adapters/DefaultSessionAdapter.cs

using Opc.Ua;
using Opc.Ua.Client;
using Serilog;
using ZB.MOM.WW.OtOpcUa.Client.Shared.Helpers;

namespace ZB.MOM.WW.OtOpcUa.Client.Shared.Adapters;

/// <summary>
///     Production session adapter wrapping a real OPC UA Session.
/// </summary>
internal sealed class DefaultSessionAdapter : ISessionAdapter
{
    private static readonly ILogger Logger = Log.ForContext<DefaultSessionAdapter>();
    private readonly Session _session;

    /// <summary>
    /// Wraps a live OPC UA session so the shared client can issue runtime operations through a testable adapter surface.
    /// </summary>
    /// <param name="session">The connected OPC UA session used for browsing, reads, writes, history, and subscriptions.</param>
    public DefaultSessionAdapter(Session session)
    {
        _session = session;
    }

    /// <inheritdoc />
    public bool Connected => _session.Connected;

    /// <inheritdoc />
    public string SessionId => _session.SessionId?.ToString() ?? string.Empty;

    /// <inheritdoc />
    public string SessionName => _session.SessionName ?? string.Empty;

    /// <inheritdoc />
    public string EndpointUrl => _session.Endpoint?.EndpointUrl ?? string.Empty;

    /// <inheritdoc />
    public string ServerName => _session.Endpoint?.Server?.ApplicationName?.Text ?? string.Empty;

    /// <inheritdoc />
    public string SecurityMode => _session.Endpoint?.SecurityMode.ToString() ?? string.Empty;

    /// <inheritdoc />
    public string SecurityPolicyUri => _session.Endpoint?.SecurityPolicyUri ?? string.Empty;

    /// <inheritdoc />
    public NamespaceTable NamespaceUris => _session.NamespaceUris;

    /// <inheritdoc />
    public void RegisterKeepAliveHandler(Action<bool> callback)
    {
        _session.KeepAlive += (_, e) =>
        {
            var isGood = e.Status == null || ServiceResult.IsGood(e.Status);
            callback(isGood);
        };
    }

    /// <inheritdoc />
    public async Task<DataValue> ReadValueAsync(NodeId nodeId, CancellationToken ct)
    {
        return await _session.ReadValueAsync(nodeId, ct);
    }

    /// <inheritdoc />
    public async Task<StatusCode> WriteValueAsync(NodeId nodeId, DataValue value, CancellationToken ct)
    {
        var writeValue = new WriteValue
        {
            NodeId = nodeId,
            AttributeId = Attributes.Value,
            Value = value
        };

        var writeCollection = new WriteValueCollection { writeValue };
        var response = await _session.WriteAsync(null, writeCollection, ct);
        // A malformed or service-level-faulted response can come back with an empty
        // Results collection alongside a service fault. Surface the service result
        // (or BadUnexpectedError) rather than letting Results[0] throw
        // IndexOutOfRangeException upstream.
        if (response.Results == null || response.Results.Count == 0)
        {
            var serviceResult = response.ResponseHeader?.ServiceResult.Code ?? StatusCodes.BadUnexpectedError;
            throw new ServiceResultException(serviceResult,
                $"Write response contained no results for node {nodeId}.");
        }

        return response.Results[0];
    }

    /// <inheritdoc />
    public async Task<(byte[]? ContinuationPoint, ReferenceDescriptionCollection References)> BrowseAsync(
        NodeId nodeId, uint nodeClassMask, CancellationToken ct)
    {
        var (_, continuationPoint, references) = await _session.BrowseAsync(
            null,
            null,
            nodeId,
            0u,
            BrowseDirection.Forward,
            ReferenceTypeIds.HierarchicalReferences,
            true,
            nodeClassMask);

        return (continuationPoint, references ?? []);
    }

    /// <inheritdoc />
    public async Task<(byte[]? ContinuationPoint, ReferenceDescriptionCollection References)> BrowseNextAsync(
        byte[] continuationPoint, CancellationToken ct)
    {
        // Pass the caller's token so a cancelled browse does not block on the continuation
        // round-trip.
        var (_, nextCp, nextRefs) = await _session.BrowseNextAsync(null, false, continuationPoint, ct);
        return (nextCp, nextRefs ?? []);
    }

    /// <inheritdoc />
    public async Task<Models.HistoryUpdateOutcome> HistoryUpdateAsync(HistoryUpdateDetails details, CancellationToken ct)
    {
        var response = await _session.HistoryUpdateAsync(
            null,
            new ExtensionObjectCollection { new ExtensionObject(details) },
            ct).ConfigureAwait(false);

        var statuses = new List<StatusCode>();
        var ops = new List<IReadOnlyList<StatusCode>>();
        if (response.Results != null)
            foreach (var r in response.Results)
            {
                statuses.Add(r.StatusCode);
                ops.Add(r.OperationResults?.ToArray() ?? Array.Empty<StatusCode>());
            }

        return new Models.HistoryUpdateOutcome(response.ResponseHeader.ServiceResult, statuses, ops);
    }

    /// <inheritdoc />
    public async Task<IReadOnlyList<DataValue>> HistoryReadRawAsync(
        NodeId nodeId, DateTime startTime, DateTime endTime, int maxValues, CancellationToken ct)
    {
        var details = new ReadRawModifiedDetails
        {
            StartTime = startTime,
            EndTime = endTime,
            NumValuesPerNode = (uint)maxValues,
            IsReadModified = false,
            ReturnBounds = false
        };

        // Continuation pages are READ, never released, until the server finishes or the cap is reached; a read stopped
        // by the cap releases the server's cursor (HistoryReadPaging). This path used to pass
        // `releaseContinuationPoints: continuationPoint != null`, which RELEASED the second page instead of reading it,
        // silently truncating every raw read to its first server page.
        return await HistoryReadPaging.ReadAllAsync(
            (cp, release, token) => ReadHistoryPageAsync(details, nodeId, cp, release, TimestampsToReturn.Source, token),
            result => result.HistoryData is ExtensionObject { Body: HistoryData historyData }
                ? historyData.DataValues
                : [],
            maxValues, nodeId, ct).ConfigureAwait(false);
    }

    /// <inheritdoc />
    public async Task<IReadOnlyList<Models.HistoryModifiedEntry>> HistoryReadModifiedAsync(
        NodeId nodeId, DateTime startTime, DateTime endTime, int maxValues, CancellationToken ct)
    {
        var details = new ReadRawModifiedDetails
        {
            StartTime = startTime,
            EndTime = endTime,
            NumValuesPerNode = (uint)maxValues,
            IsReadModified = true,
            ReturnBounds = false
        };

        return await HistoryReadPaging.ReadAllAsync(
            (cp, release, token) => ReadHistoryPageAsync(details, nodeId, cp, release, TimestampsToReturn.Both, token),
            ToModifiedEntries, maxValues, nodeId, ct).ConfigureAwait(false);
    }

    private static IEnumerable<Models.HistoryModifiedEntry> ToModifiedEntries(HistoryReadResult result)
    {
        if (result.HistoryData is not ExtensionObject { Body: HistoryModifiedData modified }) yield break;
        for (var i = 0; i < modified.DataValues.Count; i++)
        {
            var info = i < modified.ModificationInfos.Count ? modified.ModificationInfos[i] : null;
            yield return new Models.HistoryModifiedEntry(
                modified.DataValues[i],
                info?.ModificationTime ?? DateTime.MinValue,
                info?.UpdateType ?? default,
                info?.UserName);
        }
    }

    /// <summary>Issues one single-node HistoryRead (or a release of its continuation point).</summary>
    private async Task<HistoryReadResult?> ReadHistoryPageAsync(
        ReadRawModifiedDetails details, NodeId nodeId, byte[]? continuationPoint, bool release,
        TimestampsToReturn timestamps, CancellationToken ct)
    {
        var nodesToRead = new HistoryReadValueIdCollection
        {
            new HistoryReadValueId { NodeId = nodeId, ContinuationPoint = continuationPoint }
        };

        // The async overload honours the token and does not block the caller's thread (the UI dispatcher for
        // client.ui consumers).
        var response = await _session.HistoryReadAsync(
            null, new ExtensionObject(details), timestamps, release, nodesToRead, ct).ConfigureAwait(false);

        return response.Results is { Count: > 0 } results ? results[0] : null;
    }

    /// <inheritdoc />
    public async Task<IReadOnlyList<DataValue>> HistoryReadAggregateAsync(
        NodeId nodeId, DateTime startTime, DateTime endTime, NodeId aggregateId, double intervalMs,
        CancellationToken ct)
    {
        var details = new ReadProcessedDetails
        {
            StartTime = startTime,
            EndTime = endTime,
            ProcessingInterval = intervalMs,
            AggregateType = [aggregateId]
        };

        var nodesToRead = new HistoryReadValueIdCollection
        {
            new HistoryReadValueId { NodeId = nodeId }
        };

        // Use the async overload so the method honors the cancellation token and
        // does not block on a synchronous service round-trip.
        var response = await _session.HistoryReadAsync(
            null,
            new ExtensionObject(details),
            TimestampsToReturn.Source,
            false,
            nodesToRead,
            ct).ConfigureAwait(false);

        var results = response.Results;
        var allValues = new List<DataValue>();

        if (results != null && results.Count > 0)
        {
            var result = results[0];
            HistoryReadStatus.ThrowIfRefused(result.StatusCode, nodeId);
            if (result.HistoryData is ExtensionObject ext &&
                ext.Body is HistoryData historyData)
                allValues.AddRange(historyData.DataValues);
        }

        return allValues;
    }

    /// <inheritdoc />
    public async Task<ISubscriptionAdapter> CreateSubscriptionAsync(int publishingIntervalMs, CancellationToken ct)
    {
        var subscription = new Subscription(_session.DefaultSubscription)
        {
            PublishingInterval = publishingIntervalMs,
            DisplayName = "ClientShared_Subscription"
        };

        _session.AddSubscription(subscription);
        try
        {
            await subscription.CreateAsync(ct);
        }
        catch
        {
            // A subscription the server never created must not linger in Session.Subscriptions either.
            try { await _session.RemoveSubscriptionAsync(subscription, CancellationToken.None); }
            catch (Exception ex) { Logger.Warning(ex, "Error removing a subscription that failed to create"); }
            throw;
        }

        // Deleting the adapter also removes the subscription from the SDK session, so a re-targeted alarm/data
        // subscription does not accumulate dead Subscription objects in Session.Subscriptions.
        return new DefaultSubscriptionAdapter(subscription,
            async (s, token) => await _session.RemoveSubscriptionAsync(s, token).ConfigureAwait(false));
    }

    /// <inheritdoc />
    public async Task CloseAsync(CancellationToken ct)
    {
        try
        {
            // Use the async overload so the caller does not block on the close
            // service round-trip and the cancellation token is honored.
            if (_session.Connected) await _session.CloseAsync(ct).ConfigureAwait(false);
        }
        catch (Exception ex)
        {
            Logger.Warning(ex, "Error closing session");
        }
    }

    /// <summary>
    /// Releases the wrapped OPC UA session when the shared client shuts down or swaps endpoints during failover.
    /// </summary>
    public void Dispose()
    {
        try
        {
            if (_session.Connected) DisposeBridge.Run(ct => _session.CloseAsync(ct));
        }
        catch
        {
        }

        _session.Dispose();
    }

    /// <inheritdoc />
    public async Task<IList<object>?> CallMethodAsync(NodeId objectId, NodeId methodId, object[] inputArguments,
        CancellationToken ct = default)
    {
        var result = await _session.CallAsync(
            null,
            new CallMethodRequestCollection
            {
                new()
                {
                    ObjectId = objectId,
                    MethodId = methodId,
                    InputArguments = new VariantCollection(inputArguments.Select(a => new Variant(a)))
                }
            },
            ct);

        // An empty Results collection paired with a service fault must surface as
        // a ServiceResultException, not an IndexOutOfRangeException from Results[0].
        if (result.Results == null || result.Results.Count == 0)
        {
            var serviceResult = result.ResponseHeader?.ServiceResult.Code ?? StatusCodes.BadUnexpectedError;
            throw new ServiceResultException(serviceResult,
                $"Call response contained no results for method {methodId} on {objectId}.");
        }

        var callResult = result.Results[0];
        if (StatusCode.IsBad(callResult.StatusCode))
            throw new ServiceResultException(callResult.StatusCode);

        return callResult.OutputArguments?.Select(v => v.Value).ToList();
    }
}