lmxopcua/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptLogTopicSink.cs

using Serilog.Core;
using Serilog.Events;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Logging;

namespace ZB.MOM.WW.OtOpcUa.Core.Scripting;

/// <summary>
///     Serilog sink that converts each script <see cref="LogEvent"/> into a
///     <see cref="ScriptLogEntry"/> and forwards it to an <see cref="IScriptLogPublisher"/>.
///     The production publisher (<c>DpsScriptLogPublisher</c>) routes the entry onto the
///     Akka DPS <c>script-logs</c> topic so the live Script-log Admin UI page can display it.
/// </summary>
/// <remarks>
///     <para>
///         Expected to be registered in the root script-logger pipeline alongside the
///         rolling <c>scripts-*.log</c> file sink and the
///         <see cref="ScriptLogCompanionSink"/>. Events below the configured minimum level
///         are silently discarded so Debug/Verbose noise does not saturate the cluster bus
///         in production deployments.
///     </para>
///     <para>
///         Identity properties (<c>ScriptId</c>, <c>VirtualTagId</c>, <c>AlarmId</c>,
///         <c>EquipmentId</c>) are lifted from the event's structured-property bag. If
///         <c>ScriptId</c> is absent the sink falls back to the legacy <c>ScriptName</c>
///         property so pipelines that pre-date this sink continue to work correctly. If
///         neither property is present <c>ScriptId</c> defaults to <c>"unknown"</c>.
///     </para>
///     <para>
///         <b>Rate-limited per source.</b> A script that logs at Information on every evaluation of a 1 Hz
///         input put one rendered string per second per script onto the cluster bus; 200 such scripts shed
///         everyone's entries out of the bounded telemetry channels. A <see cref="ScriptLogRateLimiter"/>
///         keyed by the entry's identity (script + virtual tag + alarm) admits 20 entries per minute per
///         source and publishes one summary entry naming the dropped count when the window rolls.
///     </para>
/// </remarks>
public sealed class ScriptLogTopicSink : ILogEventSink
{
    private readonly IScriptLogPublisher _publisher;
    private readonly LogEventLevel _min;
    private readonly ScriptLogRateLimiter _limiter;

    /// <summary>Initializes a new instance of the <see cref="ScriptLogTopicSink"/> class.</summary>
    /// <param name="publisher">The publisher that routes entries to the cluster bus. Must not be <c>null</c>.</param>
    /// <param name="min">Minimum log level to forward. Events below this level are discarded (default: <see cref="LogEventLevel.Information"/>).</param>
    /// <param name="limiter">Per-source rate limiter; null ⇒ the default 20 per minute.</param>
    /// <exception cref="ArgumentNullException">Thrown when <paramref name="publisher"/> is <c>null</c>.</exception>
    public ScriptLogTopicSink(IScriptLogPublisher publisher, LogEventLevel min = LogEventLevel.Information,
        ScriptLogRateLimiter? limiter = null)
    {
        _publisher = publisher ?? throw new ArgumentNullException(nameof(publisher));
        _min = min;
        _limiter = limiter ?? new ScriptLogRateLimiter();
    }

    /// <summary>
    ///     Converts the <paramref name="logEvent"/> to a <see cref="ScriptLogEntry"/> and
    ///     publishes it, unless the event is below the configured minimum level or the source is over its
    ///     per-minute budget.
    /// </summary>
    /// <param name="logEvent">The Serilog event to process. Silently ignored when <c>null</c>.</param>
    public void Emit(LogEvent logEvent)
    {
        if (logEvent is null || logEvent.Level < _min) return;

        // Extract a string scalar property by key, returning null when absent or non-string.
        string? P(string key) =>
            logEvent.Properties.TryGetValue(key, out var v) && v is ScalarValue { Value: string s }
                ? s
                : null;

        var scriptId = P(ScriptLoggerFactory.ScriptIdProperty)
                       ?? P(ScriptLoggerFactory.ScriptNameProperty)
                       ?? "unknown";
        var virtualTagId = P(ScriptLoggerFactory.VirtualTagIdProperty);
        var alarmId = P(ScriptLoggerFactory.AlarmIdProperty);
        var equipmentId = P(ScriptLoggerFactory.EquipmentIdProperty);

        var admission = _limiter.Admit(string.Concat(scriptId, "|", virtualTagId, "|", alarmId));
        if (admission.SuppressedInPreviousWindow > 0)
        {
            _publisher.Publish(new ScriptLogEntry(
                ScriptId: scriptId,
                Level: nameof(LogEventLevel.Warning),
                Message: $"{admission.SuppressedInPreviousWindow} script-log entr{(admission.SuppressedInPreviousWindow == 1 ? "y" : "ies")} " +
                         $"from this source were suppressed in the previous {_limiter.Window.TotalSeconds:F0} s " +
                         $"(limit {_limiter.LimitPerWindow}/window); see scripts-*.log for the full stream",
                TimestampUtc: logEvent.Timestamp.UtcDateTime,
                VirtualTagId: virtualTagId,
                AlarmId: alarmId,
                EquipmentId: equipmentId));
        }
        if (!admission.Admitted) return;

        _publisher.Publish(new ScriptLogEntry(
            ScriptId: scriptId,
            Level: logEvent.Level.ToString(),
            Message: logEvent.RenderMessage(),
            TimestampUtc: logEvent.Timestamp.UtcDateTime,
            VirtualTagId: virtualTagId,
            AlarmId: alarmId,
            EquipmentId: equipmentId));
    }
}