using System.Text.Json;
using Akka.Actor;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.DataConnection;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Streaming;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Alarms;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Flattening;
using ZB.MOM.WW.ScadaBridge.SiteEventLogging;
using ZB.MOM.WW.ScadaBridge.SiteRuntime.Messages;
using ZB.MOM.WW.ScadaBridge.SiteRuntime.Persistence;
namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Actors;
///
/// Native Alarm Actor — child of Instance Actor, peer to the computed
/// . Mirrors a single source binding's native alarms
/// read-only: subscribes through the DCL, applies snapshot/live transitions,
/// retains active+unacked conditions, persists to site SQLite, and emits an
/// enriched upward for every transition.
///
/// State is keyed by (the
/// stable per-condition key). A snapshot replay (Snapshot* sentinels) atomically
/// swaps the mirrored set so conditions that cleared while disconnected emit a
/// return-to-normal. Persistence is best-effort/fire-and-forget — a failed write
/// never blocks the actor or aborts the emit.
///
/// Supervision: parent Instance Actor uses Resume for coordinator children.
///
public class NativeAlarmActor : ReceiveActor
{
private readonly ResolvedNativeAlarmSource _source;
private readonly string _instanceName;
private readonly IActorRef _instanceActor;
private readonly IActorRef _dclManager;
private readonly SiteStorageService _storage;
private readonly SiteRuntimeOptions _options;
private readonly ILogger _logger;
private readonly AlarmKind _nativeKind;
private readonly IServiceProvider? _serviceProvider;
///
/// Severity at or above which a native-alarm raise is logged as
/// Error to the site event log; below it, raises log as Warning.
/// Mirrors the 0–1000 condition-severity scale.
///
private const int ErrorSeverityThreshold = 700;
/// Current mirrored conditions, keyed by source reference.
private readonly Dictionary _alarms = new();
/// Buffer accumulating a snapshot replay until .
private readonly Dictionary _snapshotBuffer = new();
///
/// Pending mirror upserts coalesced by source reference (P4): under an alarm storm each
/// source reference keeps only its LATEST transition, flushed as one batched SQLite write on
/// a short single-shot timer rather than a write per transition. A crash loses at most one
/// flush interval of mirror writes; the (re)subscribe snapshot swap reconciles the mirror, so
/// the durability trade is safe. A removes the ref here so a
/// deleted condition is not resurrected by a still-pending upsert.
///
private readonly Dictionary _dirtyUpserts = new();
private ICancelable? _retryTimer;
private ICancelable? _flushTimer;
/// How long transitions coalesce before a batched persist flush (P4).
private readonly TimeSpan _persistFlushInterval;
/// Initializes a new for a single resolved source binding.
/// The resolved native alarm source (connection + source reference + filter).
/// The owning instance's unique name.
/// Parent instance actor; receives emitted .
/// DCL target the subscribe request is sent to.
/// Site-local SQLite store for rehydration + persistence.
/// Site runtime options (alarm cap, retry interval).
/// Logger for diagnostics.
/// Alarm kind to stamp on emitted events (OPC UA vs MxAccess); set by the
/// Instance Actor from the connection protocol. Defaults to .
/// Optional DI service provider used to resolve the optional
/// for alarm operational events. Fire-and-forget;
/// a logging failure never affects the mirror.
/// How long native-alarm transitions coalesce before a
/// batched SQLite persist flush (P4). Defaults to 100 ms.
public NativeAlarmActor(
ResolvedNativeAlarmSource source,
string instanceName,
IActorRef instanceActor,
IActorRef dclManager,
SiteStorageService storage,
SiteRuntimeOptions options,
ILogger logger,
AlarmKind nativeKind = AlarmKind.NativeOpcUa,
IServiceProvider? serviceProvider = null,
TimeSpan? persistFlushInterval = null)
{
_source = source;
_instanceName = instanceName;
_instanceActor = instanceActor;
_dclManager = dclManager;
_storage = storage;
_options = options;
_logger = logger;
_nativeKind = nativeKind;
_serviceProvider = serviceProvider;
_persistFlushInterval = persistFlushInterval ?? TimeSpan.FromMilliseconds(100);
Receive(HandleRehydration);
Receive(HandleTransition);
Receive(HandleSourceUnavailable);
Receive(HandleSubscribeResponse);
Receive(_ => SendSubscribe());
Receive(_ => FlushDirtyUpserts());
}
///
protected override void PreStart()
{
base.PreStart();
// Rehydrate last-known state from SQLite (non-blocking), then subscribe.
// A fresh source snapshot will reconcile/replace this once it arrives.
_storage.GetNativeAlarmsAsync(_instanceName, _source.CanonicalName)
.PipeTo(Self, Self, rows => new RehydrationCompleted(rows));
SendSubscribe();
_logger.LogInformation(
"NativeAlarmActor started for {Source} ({Connection}:{Ref}) on {Instance}",
_source.CanonicalName, _source.ConnectionName, _source.SourceReference, _instanceName);
}
///
protected override void PostStop()
{
_retryTimer?.Cancel();
_flushTimer?.Cancel();
// Best-effort final flush of any coalesced upserts so a graceful stop does not lose the
// last ≤ flush-interval of mirror writes (P4). Fire-and-forget — a failed final write is
// reconciled by the next (re)subscribe snapshot swap.
FlushDirtyUpserts();
base.PostStop();
}
private void SendSubscribe()
{
_dclManager.Tell(new SubscribeAlarmsRequest(
CorrelationId: Guid.NewGuid().ToString("N"),
InstanceUniqueName: _instanceName,
ConnectionName: _source.ConnectionName,
SourceReference: _source.SourceReference,
ConditionFilter: _source.ConditionFilter,
Timestamp: DateTimeOffset.UtcNow));
// Arm a response-timeout retry so a LOST subscribe response is re-sent, not
// only a failure reply (S4). Cancelled on a Success reply; a failure reply
// re-arms it at 1× (overriding this 2× response-timeout arm). 2× interval so
// a slow-but-arriving response is not pre-empted by a redundant re-send.
_retryTimer?.Cancel();
_retryTimer = Context.System.Scheduler.ScheduleTellOnceCancelable(
TimeSpan.FromMilliseconds(_options.NativeAlarmRetryIntervalMs * 2),
Self, RetrySubscribe.Instance, Self);
}
private void HandleRehydration(RehydrationCompleted msg)
{
foreach (var row in msg.Rows)
{
// A live transition that arrived before the read completed wins.
if (_alarms.ContainsKey(row.SourceReference))
{
continue;
}
AlarmConditionState? condition;
try
{
condition = JsonSerializer.Deserialize(row.ConditionJson);
}
catch (JsonException ex)
{
_logger.LogWarning(ex, "Discarding unreadable persisted native alarm {Ref} on {Instance}",
row.SourceReference, _instanceName);
continue;
}
if (condition is null)
{
continue;
}
// Restore the persisted display metadata (UA4) so the rehydrated condition renders
// fully (type/category/message/values) until the first source snapshot replaces it.
// Null/malformed metadata falls back to empty strings (pre-UA4 rows behave as before).
NativeAlarmMetadata meta = EmptyMetadata;
if (!string.IsNullOrEmpty(row.MetadataJson))
{
try
{
meta = JsonSerializer.Deserialize(row.MetadataJson) ?? EmptyMetadata;
}
catch (JsonException)
{
meta = EmptyMetadata;
}
}
var t = new NativeAlarmTransition(
row.SourceReference, string.Empty, meta.AlarmTypeName, AlarmTransitionKind.Snapshot,
condition, meta.Category, string.Empty, meta.Message, string.Empty, string.Empty,
null, row.LastTransitionAt, meta.CurrentValue, meta.LimitValue);
_alarms[row.SourceReference] = t;
// Rehydration replays last-known state on (re)start — surface it
// upward for the DebugView but do NOT re-log it as a fresh operational
// event (it is not a live transition).
Emit(t, t.Condition, logSiteEvent: false);
}
}
private void HandleTransition(NativeAlarmTransitionUpdate update)
{
var t = update.Transition;
switch (t.Kind)
{
case AlarmTransitionKind.Snapshot:
_snapshotBuffer[t.SourceReference] = t;
break;
case AlarmTransitionKind.SnapshotComplete:
ApplySnapshotSwap();
break;
default:
ApplyLiveTransition(t);
break;
}
}
///
/// Atomically replaces the mirrored set with the buffered snapshot: conditions
/// no longer present emit a return-to-normal (and drop out); present conditions
/// are upserted, persisted, and emitted.
///
private void ApplySnapshotSwap()
{
foreach (var (sourceRef, prior) in _alarms)
{
if (!_snapshotBuffer.ContainsKey(sourceRef))
{
Emit(prior, prior.Condition with { Active = false });
PersistDelete(sourceRef);
// This condition is gone for good — tell the parent
// to evict its _latestAlarmEvents key so it does not retain a stale
// (Normal) entry forever.
NotifyParentDropped(sourceRef);
}
}
_alarms.Clear();
foreach (var (sourceRef, t) in _snapshotBuffer)
{
_alarms[sourceRef] = t;
MarkDirtyUpsert(t);
// A snapshot replay is a re-sync of the source's current
// active set on (re)subscribe, NOT a live transition — surface it
// upward for the DebugView but do NOT re-log an `alarm` operational
// event. Otherwise every DCL reconnect would re-emit an `alarm`
// event for every already-active native condition (the
// synthesised return-to-normal above IS a real state change and
// keeps logSiteEvent: true).
Emit(t, t.Condition, logSiteEvent: false);
}
_snapshotBuffer.Clear();
EnforceCap();
}
private void ApplyLiveTransition(NativeAlarmTransition t)
{
// Ignore stale (out-of-order) transitions for a condition we already hold.
if (_alarms.TryGetValue(t.SourceReference, out var existing) && t.TransitionTime < existing.TransitionTime)
{
return;
}
_alarms[t.SourceReference] = t;
MarkDirtyUpsert(t);
Emit(t, t.Condition);
// Retention: a resolved condition (inactive AND acknowledged) drops out of
// the mirror — the return-to-normal has already been emitted above.
if (!t.Condition.Active && t.Condition.Acknowledged)
{
_alarms.Remove(t.SourceReference);
PersistDelete(t.SourceReference);
// Evict the parent's _latestAlarmEvents key for the
// now-resolved condition so it does not leak.
NotifyParentDropped(t.SourceReference);
}
EnforceCap();
}
private void HandleSourceUnavailable(NativeAlarmSourceUnavailable msg)
{
// Keep last-known conditions (uncertain) rather than clearing them; the
// reconnect snapshot reconciles state. No emit — avoids a flap.
_logger.LogWarning(
"Native alarm feed unavailable for {Source} ({Connection}) on {Instance}; retaining {Count} mirrored alarms pending reconnect snapshot",
_source.CanonicalName, msg.ConnectionName, _instanceName, _alarms.Count);
}
private void HandleSubscribeResponse(SubscribeAlarmsResponse resp)
{
if (resp.Success)
{
// Success — cancel the response-timeout retry armed in SendSubscribe.
_retryTimer?.Cancel();
_logger.LogInformation("Native alarm subscription established for {Source} on {Instance}",
_source.CanonicalName, _instanceName);
return;
}
_logger.LogWarning(
"Native alarm subscription failed for {Source} on {Instance}: {Error}; retrying in {RetryMs}ms",
_source.CanonicalName, _instanceName, resp.ErrorMessage, _options.NativeAlarmRetryIntervalMs);
_retryTimer?.Cancel();
_retryTimer = Context.System.Scheduler.ScheduleTellOnceCancelable(
TimeSpan.FromMilliseconds(_options.NativeAlarmRetryIntervalMs),
Self, RetrySubscribe.Instance, Self);
}
/// Caps the mirrored set, dropping the oldest conditions (and logging — no silent truncation).
private void EnforceCap()
{
var cap = _options.MirroredAlarmCapPerSource;
if (_alarms.Count <= cap)
{
return;
}
var overflow = _alarms.Values
.OrderBy(a => a.TransitionTime)
.Take(_alarms.Count - cap)
.ToList();
foreach (var evicted in overflow)
{
var sourceRef = evicted.SourceReference;
// The sibling drop paths (ApplySnapshotSwap, the
// ApplyLiveTransition retention drop) always emit a return-to-normal
// before the condition leaves the mirror. EnforceCap previously dropped
// a condition whose last-emitted state could still be Active, with no
// compensating emit — so the Instance Actor (and central's stream / the
// operator Alarm Summary) kept showing it Active forever, a phantom
// stuck alarm the mirror could never clear. Emit the return-to-normal
// for any still-active evicted condition (mirroring ApplySnapshotSwap)
// before removing it.
if (evicted.Condition.Active)
{
Emit(evicted, evicted.Condition with { Active = false });
}
_alarms.Remove(sourceRef);
PersistDelete(sourceRef);
// This condition is gone for good — evict the parent's
// _latestAlarmEvents key so it does not retain a stale entry.
NotifyParentDropped(sourceRef);
_logger.LogWarning(
"Native alarm cap {Cap} exceeded for {Source} on {Instance}; dropped oldest mirrored alarm {Ref}",
cap, _source.CanonicalName, _instanceName, sourceRef);
}
}
///
/// Signals the parent Instance Actor that a native condition has
/// left the mirror for good so it can evict the matching _latestAlarmEvents
/// key. Always sent AFTER the condition's final return-to-normal
/// emit, so the stream still sees the clear.
///
private void NotifyParentDropped(string sourceReference)
{
_instanceActor.Tell(new NativeAlarmDropped(sourceReference));
}
///
/// Builds and tells the parent an enriched for a condition.
///
/// The mirrored transition.
/// The condition state to surface (may differ from 's
/// own condition, e.g. a synthesised return-to-normal on snapshot swap).
/// When true (live + snapshot transitions), emit an
/// alarm operational event. Suppressed for SQLite rehydration so a node restart does not
/// re-log every last-known condition.
private void Emit(NativeAlarmTransition t, AlarmConditionState condition, bool logSiteEvent = true)
{
var change = new AlarmStateChanged(
_instanceName,
t.SourceReference,
condition.Active ? AlarmState.Active : AlarmState.Normal,
condition.Severity,
t.TransitionTime)
{
Kind = _nativeKind,
Condition = condition,
SourceReference = t.SourceReference,
AlarmTypeName = t.AlarmTypeName,
Category = t.Category,
Message = t.Message,
OperatorUser = t.OperatorUser,
OperatorComment = t.OperatorComment,
OriginalRaiseTime = t.OriginalRaiseTime,
CurrentValue = t.CurrentValue,
LimitValue = t.LimitValue,
NativeSourceCanonicalName = _source.CanonicalName,
};
_instanceActor.Tell(change);
if (logSiteEvent)
{
LogAlarmEvent(t, condition);
}
}
///
/// Fire-and-forget an alarm operational event mirroring a native
/// condition transition. An active condition is a raise (severity by the
/// condition's severity); an inactive condition is a return-to-normal; an
/// acknowledge transition is informational. Resolved optionally and never
/// awaited so a logging failure cannot affect the mirror (matching the
/// established ScriptActor/ScriptExecutionActor pattern).
///
private void LogAlarmEvent(NativeAlarmTransition t, AlarmConditionState condition)
{
var logger = _serviceProvider?.GetService();
if (logger == null)
{
return;
}
string severity;
string message;
if (t.Kind == AlarmTransitionKind.Acknowledge)
{
severity = "Info";
message = $"Native alarm {t.SourceReference} acknowledged";
}
else if (condition.Active)
{
severity = condition.Severity >= ErrorSeverityThreshold ? "Error" : "Warning";
message = $"Native alarm {t.SourceReference} active (severity {condition.Severity})";
}
else
{
severity = "Info";
message = $"Native alarm {t.SourceReference} returned to normal";
}
_ = logger.LogEventAsync(
"alarm", severity, _instanceName, $"NativeAlarmActor:{_source.CanonicalName}", message);
}
///
/// Records the latest transition per source reference and arms the coalesced flush timer (P4).
/// The actual SQLite write is deferred to so an alarm storm
/// collapses into one batched write instead of a write per transition.
///
private void MarkDirtyUpsert(NativeAlarmTransition t)
{
_dirtyUpserts[t.SourceReference] = t;
if (_flushTimer == null)
{
_flushTimer = Context.System.Scheduler.ScheduleTellOnceCancelable(
_persistFlushInterval, Self, FlushPersistence.Instance, Self);
}
}
///
/// Persists all coalesced upserts in one batched transaction (P4), then disarms the flush
/// timer. Fire-and-forget with OnlyOnFaulted logging — a failed write never blocks the actor;
/// the mirror reconciles on the next (re)subscribe snapshot.
///
private void FlushDirtyUpserts()
{
_flushTimer = null;
if (_dirtyUpserts.Count == 0)
return;
var batch = _dirtyUpserts.Values
.Select(t => (
t.SourceReference,
JsonSerializer.Serialize(t.Condition),
(string?)JsonSerializer.Serialize(new NativeAlarmMetadata(
t.AlarmTypeName, t.Category, t.Message, t.CurrentValue, t.LimitValue)),
t.TransitionTime))
.ToList();
_dirtyUpserts.Clear();
_storage.UpsertNativeAlarmsAsync(_instanceName, _source.CanonicalName, batch)
.ContinueWith(
task => _logger.LogWarning(task.Exception,
"Failed to persist {Count} native alarms on {Instance}", batch.Count, _instanceName),
TaskContinuationOptions.OnlyOnFaulted);
}
private void PersistDelete(string sourceReference)
{
// Drop any pending coalesced upsert for this ref so the delete is not resurrected by a
// still-buffered upsert on the next flush (P4).
_dirtyUpserts.Remove(sourceReference);
_storage.DeleteNativeAlarmAsync(_instanceName, _source.CanonicalName, sourceReference)
.ContinueWith(
task => _logger.LogWarning(task.Exception,
"Failed to delete native alarm {Ref} on {Instance}", sourceReference, _instanceName),
TaskContinuationOptions.OnlyOnFaulted);
}
// ── Internal messages ──
private sealed record RehydrationCompleted(IReadOnlyList Rows);
private sealed class RetrySubscribe
{
public static readonly RetrySubscribe Instance = new();
private RetrySubscribe() { }
}
/// Self-tell that fires the coalesced native-alarm persist flush (P4).
private sealed class FlushPersistence
{
public static readonly FlushPersistence Instance = new();
private FlushPersistence() { }
}
/// Empty metadata fallback for pre-UA4 rows or malformed metadata_json.
private static readonly NativeAlarmMetadata EmptyMetadata =
new(string.Empty, string.Empty, string.Empty, string.Empty, string.Empty);
///
/// Persisted display metadata for a native alarm condition (UA4). Serialized into the
/// metadata_json column so a rehydrated condition renders fully (type/category/message/
/// current+limit values) before the first source snapshot re-supplies it.
///
private sealed record NativeAlarmMetadata(
string AlarmTypeName, string Category, string Message, string CurrentValue, string LimitValue);
}