lmxopcua/src/ZB.MOM.WW.OtOpcUa.Host/Historian/HistoryContinuationPoint.cs

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using Opc.Ua;
using Serilog;

namespace ZB.MOM.WW.OtOpcUa.Host.Historian
{
    /// <summary>
    ///     Manages continuation points for OPC UA HistoryRead requests that return
    ///     more data than the per-request limit allows.
    /// </summary>
    internal sealed class HistoryContinuationPointManager
    {
        private static readonly ILogger Log = Serilog.Log.ForContext<HistoryContinuationPointManager>();

        private readonly ConcurrentDictionary<Guid, StoredContinuation> _store = new();
        private readonly TimeSpan _timeout;

        public HistoryContinuationPointManager() : this(TimeSpan.FromMinutes(5)) { }

        internal HistoryContinuationPointManager(TimeSpan timeout)
        {
            _timeout = timeout;
        }

        /// <summary>
        ///     Stores remaining data values and returns a continuation point identifier.
        /// </summary>
        public byte[] Store(List<DataValue> remaining)
        {
            PurgeExpired();
            var id = Guid.NewGuid();
            _store[id] = new StoredContinuation(remaining, DateTime.UtcNow);
            Log.Debug("Stored history continuation point {Id} with {Count} remaining values", id, remaining.Count);
            return id.ToByteArray();
        }

        /// <summary>
        ///     Retrieves and removes the remaining data values for a continuation point.
        ///     Returns null if the continuation point is invalid or expired.
        /// </summary>
        public List<DataValue>? Retrieve(byte[] continuationPoint)
        {
            PurgeExpired();
            if (continuationPoint == null || continuationPoint.Length != 16)
                return null;

            var id = new Guid(continuationPoint);
            if (!_store.TryRemove(id, out var stored))
                return null;

            if (DateTime.UtcNow - stored.CreatedAt > _timeout)
            {
                Log.Debug("History continuation point {Id} expired", id);
                return null;
            }

            return stored.Values;
        }

        /// <summary>
        ///     Releases a continuation point without retrieving its data.
        /// </summary>
        public void Release(byte[] continuationPoint)
        {
            PurgeExpired();
            if (continuationPoint == null || continuationPoint.Length != 16)
                return;

            var id = new Guid(continuationPoint);
            _store.TryRemove(id, out _);
        }

        private void PurgeExpired()
        {
            var cutoff = DateTime.UtcNow - _timeout;
            foreach (var kvp in _store)
            {
                if (kvp.Value.CreatedAt < cutoff)
                    _store.TryRemove(kvp.Key, out _);
            }
        }

        private sealed class StoredContinuation
        {
            public StoredContinuation(List<DataValue> values, DateTime createdAt)
            {
                Values = values;
                CreatedAt = createdAt;
            }

            public List<DataValue> Values { get; }
            public DateTime CreatedAt { get; }
        }
    }
}