junction/src/Junction.Protocols.OpcUa/OpcuaDriver.cs

using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using Junction.Domain;
using Junction.Domain.Models;
using Junction.Domain.Protocols;

namespace Junction.Protocols.OpcUa
{
    /// <summary>
    /// Per-machine OPC UA protocol driver. Reads the configured node values and browses the address
    /// space through an injected <see cref="IUaClient"/> (the transport seam, mirroring the HttpClient
    /// injection in <c>MtconnectDriver</c>). SDK types stay behind <see cref="IUaClient"/> so this
    /// driver is unit-tested against a mock with no live server.
    /// <para>
    /// Never throws for expected transport failures: maps them to <see cref="Result{T}.Fail(OperationError)"/>
    /// and honours cancellation via <see cref="Result{T}.Cancelled"/>.
    /// </para>
    /// </summary>
    public sealed class OpcuaDriver : IProtocolDriver
    {
        private const string Source = "OpcuaDriver";

        private readonly IUaClient _client;
        private readonly Guid _machineId;
        private readonly IReadOnlyList<string> _nodeIds;
        private readonly string _browseRoot;
        private readonly TimeSpan _timeout;

        // Selected data item ids to keep in ReadCurrentAsync snapshots. Empty => opt-in "monitor nothing".
        private readonly HashSet<string> _monitoredItemIds;

        /// <summary>Protocol identifier this driver serves.</summary>
        public string ProtocolId => "opcua";

        /// <param name="client">Transport seam; injected for testability, owned by the factory.</param>
        /// <param name="machineId">Machine this driver reads for; stamped onto the snapshot.</param>
        /// <param name="nodeIds">Configured node ids to read in <see cref="ReadCurrentAsync"/>.</param>
        /// <param name="browseRoot">Address-space browse start node for <see cref="ProbeAsync"/>.</param>
        /// <param name="timeout">Per-operation timeout enforced via a linked <see cref="CancellationTokenSource"/>.</param>
        /// <param name="monitoredItemIds">
        /// DataItem ids the user selected to monitor. <see cref="ReadCurrentAsync"/> keeps only these
        /// (opt-in); empty/null => "monitor nothing".
        /// </param>
        public OpcuaDriver(
            IUaClient client,
            Guid machineId,
            IReadOnlyList<string>? nodeIds,
            string browseRoot,
            TimeSpan timeout,
            IReadOnlyCollection<string>? monitoredItemIds = null)
        {
            _client = client ?? throw new ArgumentNullException(nameof(client));
            _machineId = machineId;
            _nodeIds = nodeIds ?? Array.Empty<string>();
            _browseRoot = browseRoot ?? "";
            if (timeout <= TimeSpan.Zero && timeout != Timeout.InfiniteTimeSpan)
            {
                throw new ArgumentOutOfRangeException(nameof(timeout), timeout, "Timeout must be positive or Timeout.InfiniteTimeSpan.");
            }
            _timeout = timeout;

            _monitoredItemIds = new HashSet<string>(StringComparer.Ordinal);
            if (monitoredItemIds != null)
            {
                foreach (var id in monitoredItemIds)
                {
                    if (!string.IsNullOrEmpty(id))
                    {
                        _monitoredItemIds.Add(id);
                    }
                }
            }
        }

        /// <inheritdoc />
        public async Task<Result<MachineSnapshot>> ReadCurrentAsync(CancellationToken cancellationToken)
        {
            if (cancellationToken.IsCancellationRequested)
            {
                return Result<MachineSnapshot>.Cancelled();
            }

            using var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
            cts.CancelAfter(_timeout);

            IReadOnlyList<UaReadResult> reads;
            try
            {
                await _client.ConnectAsync(cts.Token).ConfigureAwait(false);
                reads = await _client.ReadAsync(_nodeIds, cts.Token).ConfigureAwait(false);
            }
            catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
            {
                return Result<MachineSnapshot>.Cancelled();
            }
            catch (OperationCanceledException ex)
            {
                return Fail("CURRENT_TIMEOUT", "OPC UA read timed out: " + ex.Message);
            }
            catch (Exception ex)
            {
                return Fail("CURRENT_READ_ERROR", "OPC UA read failed: " + ex.Message);
            }

            var items = new List<DataItem>(reads.Count);
            foreach (UaReadResult r in reads)
            {
                items.Add(new DataItem(r.NodeId, r.NodeId, r.Value, "VARIABLE", r.SourceTimestamp));
            }

            var snapshot = new MachineSnapshot(_machineId, DateTimeOffset.UtcNow, ConnectionState.Connected, items);
            return Result<MachineSnapshot>.Ok(FilterToMonitored(snapshot));
        }

        /// <inheritdoc />
        public async Task<Result<IReadOnlyList<DataItemDescriptor>>> ProbeAsync(CancellationToken cancellationToken)
        {
            if (cancellationToken.IsCancellationRequested)
            {
                return Result<IReadOnlyList<DataItemDescriptor>>.Cancelled();
            }

            using var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
            cts.CancelAfter(_timeout);

            IReadOnlyList<DataItemDescriptor> descriptors;
            try
            {
                await _client.ConnectAsync(cts.Token).ConfigureAwait(false);
                descriptors = await _client.BrowseAsync(_browseRoot, cts.Token).ConfigureAwait(false);
            }
            catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
            {
                return Result<IReadOnlyList<DataItemDescriptor>>.Cancelled();
            }
            catch (OperationCanceledException ex)
            {
                return ProbeFail("PROBE_TIMEOUT", "OPC UA browse timed out: " + ex.Message);
            }
            catch (Exception ex)
            {
                return ProbeFail("PROBE_BROWSE_ERROR", "OPC UA browse failed: " + ex.Message);
            }

            return Result<IReadOnlyList<DataItemDescriptor>>.Ok(descriptors);
        }

        /// <summary>
        /// Returns a copy of <paramref name="snapshot"/> keeping only items whose id is in the monitored
        /// selection. Empty selection yields an empty item set (opt-in). Connection state and capture
        /// instant are preserved. Exact semantics as MtconnectDriver.
        /// </summary>
        private MachineSnapshot FilterToMonitored(MachineSnapshot snapshot)
        {
            if (_monitoredItemIds.Count == 0)
            {
                return new MachineSnapshot(
                    snapshot.MachineId, snapshot.CapturedAt, snapshot.ConnectionState, Array.Empty<DataItem>());
            }

            var kept = new List<DataItem>();
            for (int i = 0; i < snapshot.Items.Count; i++)
            {
                DataItem item = snapshot.Items[i];
                if (_monitoredItemIds.Contains(item.Id))
                {
                    kept.Add(item);
                }
            }

            return new MachineSnapshot(snapshot.MachineId, snapshot.CapturedAt, snapshot.ConnectionState, kept);
        }

        private static Result<MachineSnapshot> Fail(string code, string message) =>
            Result<MachineSnapshot>.Fail(new OperationError(code, Source, message));

        private static Result<IReadOnlyList<DataItemDescriptor>> ProbeFail(string code, string message) =>
            Result<IReadOnlyList<DataItemDescriptor>>.Fail(new OperationError(code, Source, message));
    }
}