lmxopcua/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Telemetry/ITelemetryLocalHub.cs

using System.Threading.Channels;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Alerts;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Drivers;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Logging;

namespace ZB.MOM.WW.OtOpcUa.Runtime.Telemetry;

/// <summary>
///     A single live-telemetry item flowing through the node-local hub. A closed union over the four
///     wire-facing telemetry records — carried as the domain records themselves, NOT as proto (the
///     gRPC projection happens at the streaming-service seam, one layer out).
/// </summary>
/// <remarks>
///     Two of the four are <b>snapshot-style</b> (a last-value-per-key view: <see cref="Health"/> keyed
///     by <c>DriverInstanceId</c>, <see cref="Resilience"/> keyed by <c>(DriverInstanceId, HostName)</c>)
///     and are cached so a newly-attaching subscriber is primed immediately. The other two
///     (<see cref="Alarm"/>, <see cref="Script"/>) are <b>append-style</b> logs and are forwarded live-only.
/// </remarks>
public abstract record TelemetryItem
{
    private TelemetryItem() { }

    /// <summary>An append-style alarm transition. Live-forwarded only; never cached/replayed.</summary>
    public sealed record Alarm(AlarmTransitionEvent E) : TelemetryItem;

    /// <summary>An append-style script-log line. Live-forwarded only; never cached/replayed.</summary>
    public sealed record Script(ScriptLogEntry E) : TelemetryItem;

    /// <summary>A snapshot-style driver-health change, cached last-value per <c>DriverInstanceId</c>.</summary>
    public sealed record Health(DriverHealthChanged E) : TelemetryItem;

    /// <summary>A snapshot-style resilience-status change, cached last-value per <c>(DriverInstanceId, HostName)</c>.</summary>
    public sealed record Resilience(DriverResilienceStatusChanged E) : TelemetryItem;
}

/// <summary>
///     One attached reader on the <see cref="ITelemetryLocalHub"/>. Disposing detaches it from the hub
///     and completes its channel; the reader first yields the cached snapshots, then live deltas.
/// </summary>
public interface ITelemetrySubscription : IDisposable
{
    /// <summary>The bounded reader this subscription drains. Completed on <see cref="IDisposable.Dispose"/>.</summary>
    ChannelReader<TelemetryItem> Reader { get; }
}

/// <summary>
///     Process-wide, node-local fan-out hub sitting between this node's own telemetry producers and the
///     (Phase-5) gRPC streaming service central dials into.
/// </summary>
/// <remarks>
///     <para>
///         <b>CRITICAL INVARIANT — this hub carries ONLY this node's own telemetry.</b> It is fed
///         directly, in-process, by the node's producers and MUST NEVER subscribe to cluster
///         DistributedPubSub. On the single mesh, DPS delivers every node's events to every node; if the
///         hub read DPS it would stream peers' events and central — which dials each node individually —
///         would double-count.
///     </para>
///     <para>
///         Fan-out is lossy under backpressure by design: each subscriber owns a bounded channel with
///         <see cref="BoundedChannelFullMode.DropOldest"/>, so a slow consumer sheds its own oldest items
///         and can never block the producer / node-actor threads.
///     </para>
/// </remarks>
public interface ITelemetryLocalHub
{
    /// <summary>
    ///     Fans <paramref name="item"/> to every currently-attached subscriber (non-blocking, drop-oldest
    ///     on a full channel) and, for the two snapshot-style items, updates the last-value cache that
    ///     primes future subscribers.
    /// </summary>
    void Emit(TelemetryItem item);

    /// <summary>
    ///     Attaches a new subscriber. Its reader first yields the cached snapshots (all cached
    ///     <see cref="TelemetryItem.Health"/>, then all cached <see cref="TelemetryItem.Resilience"/>),
    ///     then live deltas — with no delta lost across the attach boundary. Dispose to detach.
    /// </summary>
    /// <param name="boundedCapacity">
    ///     Per-subscriber channel capacity (must be positive). The snapshot prelude is written into this
    ///     same bounded/DropOldest channel before <see cref="Subscribe"/> returns, so
    ///     <paramref name="boundedCapacity"/> should comfortably exceed this node's live driver-instance +
    ///     (instance, host) count — otherwise the earliest cached snapshots are silently evicted during
    ///     priming before the reader ever drains them. The node-local cache holds only THIS node's
    ///     instances (a handful), so a normal large capacity leaves ample headroom.
    /// </param>
    ITelemetrySubscription Subscribe(int boundedCapacity);
}