using System.Collections.Concurrent;
using Microsoft.Extensions.Logging;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Drivers;
namespace ZB.MOM.WW.OtOpcUa.AdminUI.Hubs;
///
/// Thread-safe in-memory implementation of .
/// Keyed by (, );
/// last write wins per key.
///
///
/// Keyed per reporting node since lmxopcua#847: both nodes of a redundant pair run the same driver instance,
/// so a driver-id-only key made a driver Faulted on central-1 and Healthy on central-2 flip between the two on
/// every publish. A snapshot from a node that predates ReportingNodeId keys under the empty node.
///
public sealed class InMemoryDriverStatusSnapshotStore : IDriverStatusSnapshotStore
{
// Each entry carries the LOCAL receive time (lmxopcua#857) so a view primed from the store can age a snapshot
// from when this process actually got it, not from when the view happened to read it.
private readonly ConcurrentDictionary<(string DriverInstanceId, string NodeId), StoredDriverHealth> _byInstance = new();
private readonly ILogger? _logger;
private readonly TimeProvider _time;
private readonly FailureLogThrottle _failureLog = new();
/// Creates the store.
/// Optional logger for a subscriber that throws during a raise.
/// Clock that stamps each snapshot's local receive time;
/// when null.
public InMemoryDriverStatusSnapshotStore(
ILogger? logger = null, TimeProvider? time = null)
{
_logger = logger;
_time = time ?? TimeProvider.System;
}
/// Raised whenever a driver status snapshot is stored or updated.
public event Action? SnapshotChanged;
private static (string, string) KeyOf(DriverHealthChanged s) => (s.DriverInstanceId, s.ReportingNodeId ?? "");
///
public void Upsert(DriverHealthChanged snapshot)
{
_byInstance[KeyOf(snapshot)] = new StoredDriverHealth(snapshot, _time.GetUtcNow().UtcDateTime);
// Capture-then-invoke so a concurrent unsubscribe can't null the delegate mid-raise; one
// subscriber at a time so a throwing one cannot starve the rest (or escape onto the bridge actor).
SafeMulticast.Invoke(SnapshotChanged, snapshot,
ex => _logger?.Log(_failureLog.NextLevel(), ex, "Driver status subscriber threw for {DriverInstanceId}", snapshot.DriverInstanceId));
}
///
public int EvictUnconfiguredStale(IReadOnlySet configuredInstanceIds, DateTime olderThanUtc)
{
var removed = 0;
foreach (var (key, entry) in _byInstance)
{
if (configuredInstanceIds.Contains(key.DriverInstanceId) || entry.Snapshot.PublishedUtc >= olderThanUtc) continue;
// Remove only the exact entry inspected: a fresh publish that raced in is kept.
if (_byInstance.TryRemove(new KeyValuePair<(string, string), StoredDriverHealth>(key, entry))) removed++;
}
return removed;
}
///
public bool TryGet(string driverInstanceId, out DriverHealthChanged snapshot)
{
DriverHealthChanged? latest = null;
foreach (var (key, entry) in _byInstance)
{
var s = entry.Snapshot;
if (!string.Equals(key.DriverInstanceId, driverInstanceId, StringComparison.Ordinal)) continue;
if (latest is null || s.PublishedUtc > latest.PublishedUtc) latest = s;
}
snapshot = latest!;
return latest is not null;
}
///
public IReadOnlyCollection GetAll() => _byInstance.Values.Select(e => e.Snapshot).ToArray();
///
public IReadOnlyCollection GetAllWithReceiveTime() => _byInstance.Values.ToArray();
}