using Akka.Actor;
using Akka.Event;
using Microsoft.Extensions.Logging.Abstractions;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Protocol;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.DataConnection;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Management;
using ZB.MOM.WW.ScadaBridge.Commons.Serialization;
using ZB.MOM.WW.ScadaBridge.Commons.Types.DataConnections;
using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Adapters;
using ZB.MOM.WW.ScadaBridge.HealthMonitoring;
using ZB.MOM.WW.ScadaBridge.SiteEventLogging;
namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Actors;
///
/// Protocol extensibility — manages DataConnectionActor instances.
/// Routes messages to the correct connection actor based on connection name.
/// Adding a new protocol = implement IDataConnection + register with IDataConnectionFactory.
///
public class DataConnectionManagerActor : ReceiveActor
{
private readonly ILoggingAdapter _log = Context.GetLogger();
private readonly IDataConnectionFactory _factory;
private readonly DataConnectionOptions _options;
private readonly ISiteHealthCollector _healthCollector;
private readonly ISiteEventLogger? _siteEventLogger;
// Deployment-wide OPC UA application identity / cert-store paths — the same
// global options the DataConnectionFactory feeds to RealOpcUaClient when creating OPC
// UA connections. Needed by the verify-endpoint probe (VerifyEndpointCommand), which
// builds an ApplicationConfiguration directly rather than through a connection actor.
private readonly OpcUaGlobalOptions _opcUaGlobalOptions;
private readonly Dictionary _connectionActors = new();
///
/// Initializes a new with the required dependencies.
///
/// Factory used to create protocol-specific data connection adapters.
/// Configuration options for data connections.
/// Collector for site health metrics reported by connection actors.
/// Optional logger for site event entries; null disables site event logging.
///
/// Deployment-wide OPC UA application identity / cert-store paths used by the
/// verify-endpoint probe; null falls back to defaults (mirrors
/// 's default-options constructor).
///
public DataConnectionManagerActor(
IDataConnectionFactory factory,
DataConnectionOptions options,
ISiteHealthCollector healthCollector,
ISiteEventLogger? siteEventLogger = null,
OpcUaGlobalOptions? opcUaGlobalOptions = null)
{
_factory = factory;
_options = options;
_healthCollector = healthCollector;
_siteEventLogger = siteEventLogger;
_opcUaGlobalOptions = opcUaGlobalOptions ?? new OpcUaGlobalOptions();
Receive(HandleCreateConnection);
Receive(HandleRoute);
Receive(HandleRoute);
Receive(HandleRouteAlarms);
Receive(HandleRouteAlarms);
Receive(HandleRouteWrite);
Receive(HandleRouteWriteBatch);
Receive(HandleRemoveConnection);
Receive(HandleGetAllHealthReports);
Receive(HandleBrowse);
Receive(HandleSearch);
Receive(HandleReadTagValues);
Receive(HandleVerifyEndpoint);
}
private void HandleCreateConnection(CreateConnectionCommand command)
{
if (_connectionActors.ContainsKey(command.ConnectionName))
{
_log.Warning("Connection {0} already exists", command.ConnectionName);
return;
}
// Factory creates the correct adapter based on protocol type
var adapter = _factory.Create(command.ProtocolType, command.PrimaryConnectionDetails);
var props = Props.Create(() => new DataConnectionActor(
command.ConnectionName, adapter, _options, _healthCollector,
_factory, command.ProtocolType,
command.PrimaryConnectionDetails,
command.BackupConnectionDetails,
command.FailoverRetryCount,
_siteEventLogger));
// Sanitize name for Akka actor path (replace spaces and invalid chars)
var actorName = new string(command.ConnectionName
.Select(c => char.IsLetterOrDigit(c) || "-_.*$+:@&=,!~';()".Contains(c) ? c : '-')
.ToArray());
var actorRef = Context.ActorOf(props, actorName);
_connectionActors[command.ConnectionName] = actorRef;
_log.Info("Created DataConnectionActor for {0} (protocol={1})",
command.ConnectionName, command.ProtocolType);
}
private void HandleRoute(SubscribeTagsRequest request)
{
if (_connectionActors.TryGetValue(request.ConnectionName, out var actor))
actor.Forward(request);
else
{
_log.Warning("No connection actor for {0}", request.ConnectionName);
Sender.Tell(new SubscribeTagsResponse(
request.CorrelationId, request.InstanceUniqueName, false,
$"Unknown connection: {request.ConnectionName}", DateTimeOffset.UtcNow));
}
}
private void HandleRoute(UnsubscribeTagsRequest request)
{
if (_connectionActors.TryGetValue(request.ConnectionName, out var actor))
actor.Forward(request);
else
_log.Warning("No connection actor for {0} during unsubscribe", request.ConnectionName);
}
///
/// Routes a native alarm subscribe to the that owns
/// the named connection (the NativeAlarmActor sends here, not to the child directly).
///
private void HandleRouteAlarms(SubscribeAlarmsRequest request)
{
if (_connectionActors.TryGetValue(request.ConnectionName, out var actor))
actor.Forward(request);
else
{
_log.Warning("No connection actor for {0} during alarm subscribe", request.ConnectionName);
Sender.Tell(new SubscribeAlarmsResponse(
request.CorrelationId, request.InstanceUniqueName, false,
$"Unknown connection: {request.ConnectionName}", DateTimeOffset.UtcNow));
}
}
private void HandleRouteAlarms(UnsubscribeAlarmsRequest request)
{
if (_connectionActors.TryGetValue(request.ConnectionName, out var actor))
actor.Forward(request);
else
_log.Warning("No connection actor for {0} during alarm unsubscribe", request.ConnectionName);
}
private void HandleRouteWrite(WriteTagRequest request)
{
if (_connectionActors.TryGetValue(request.ConnectionName, out var actor))
actor.Forward(request);
else
{
_log.Warning("No connection actor for {0}", request.ConnectionName);
Sender.Tell(new WriteTagResponse(
request.CorrelationId, false,
$"Unknown connection: {request.ConnectionName}", DateTimeOffset.UtcNow));
}
}
///
/// Routes a to the child
/// that owns the named connection — the batch
/// counterpart of . The manager owns only the
/// unknown-connection failure (the same split as every other routed message);
/// the child resolves connected/not-connected and the per-write outcomes.
///
private void HandleRouteWriteBatch(WriteTagBatchRequest request)
{
if (_connectionActors.TryGetValue(request.ConnectionName, out var actor))
actor.Forward(request);
else
{
_log.Warning("No connection actor for {0}", request.ConnectionName);
Sender.Tell(new WriteTagBatchResponse(
request.CorrelationId, false,
$"Unknown connection: {request.ConnectionName}", DateTimeOffset.UtcNow));
}
}
///
/// Routes a from the central UI's OPC UA
/// Tag Browser to the child that owns the
/// named connection. The manager is the only actor that knows whether a
/// connection exists at this site — so it owns the
/// failure. Everything
/// else (capability check, session state, server errors) lives inside the
/// child where the adapter is held.
///
private void HandleBrowse(BrowseNodeCommand command)
{
if (_connectionActors.TryGetValue(command.ConnectionName, out var actor))
{
actor.Forward(command);
}
else
{
_log.Warning("No connection actor for {0} during browse", command.ConnectionName);
Sender.Tell(new BrowseNodeResult(
Array.Empty(),
Truncated: false,
new BrowseFailure(
BrowseFailureKind.ConnectionNotFound,
$"No data connection named '{command.ConnectionName}' at this site.")));
}
}
///
/// Routes a from the central UI's OPC
/// UA tag picker to the child that owns the
/// named connection — the address-space analogue of .
/// Same split: the manager owns
/// (only it knows the per-site connection set); the capability check and every
/// other failure live inside the child where the adapter is held.
///
private void HandleSearch(SearchAddressSpaceCommand command)
{
if (_connectionActors.TryGetValue(command.ConnectionName, out var actor))
{
actor.Forward(command);
}
else
{
_log.Warning("No connection actor for {0} during search", command.ConnectionName);
Sender.Tell(new SearchAddressSpaceResult(
Array.Empty(),
CapReached: false,
new BrowseFailure(
BrowseFailureKind.ConnectionNotFound,
$"No data connection named '{command.ConnectionName}' at this site.")));
}
}
///
/// Routes a from the CentralUI's Test
/// Bindings dialog to the child that
/// owns the named connection. Same split as —
/// the manager owns
/// because it is
/// the only actor with site-level visibility; every other failure
/// (not connected, server error, timeout) is resolved by the child where
/// the adapter is held.
///
private void HandleReadTagValues(ReadTagValuesCommand command)
{
if (_connectionActors.TryGetValue(command.ConnectionName, out var actor))
{
actor.Forward(command);
}
else
{
_log.Warning("No connection actor for {0} during test-bindings read", command.ConnectionName);
Sender.Tell(new ReadTagValuesResult(
Array.Empty(),
new ReadTagValuesFailure(
ReadTagValuesFailureKind.ConnectionNotFound,
$"No data connection named '{command.ConnectionName}' at this site.")));
}
}
///
/// Handles a from the Central UI's "Verify"
/// action — probes the endpoint config WITHOUT persisting it (connect → capture an
/// untrusted cert → disconnect) and pipes a structured
/// back to the sender. Verify does NOT require an existing connection (the config may be
/// brand-new and unsaved), so — unlike the routed browse/read handlers — it does not look
/// up a connection actor; it runs the probe directly. Only OPC UA is supported today.
///
private void HandleVerifyEndpoint(VerifyEndpointCommand cmd)
{
if (!string.Equals(cmd.Protocol, "OpcUa", StringComparison.OrdinalIgnoreCase))
{
Sender.Tell(new VerifyEndpointResult(
false, VerifyFailureKind.ServerError,
"Verify is only supported for OPC UA connections.", null));
return;
}
OpcUaEndpointConfig config;
try
{
(config, _) = OpcUaEndpointConfigSerializer.Deserialize(cmd.ConfigJson);
}
catch (Exception ex)
{
// Defensive: Deserialize is designed not to throw (it classifies Malformed), but
// a verify must never crash the manager — surface the parse failure as ServerError.
_log.Warning(ex, "Verify config for {0} could not be parsed", cmd.ConnectionName);
Sender.Tell(new VerifyEndpointResult(
false, VerifyFailureKind.ServerError,
"The endpoint configuration could not be parsed.", null));
return;
}
var probeLogger = NullLogger.Instance;
RealOpcUaClient
.VerifyEndpointAsync(config, _opcUaGlobalOptions, probeLogger, TimeSpan.FromSeconds(6), CancellationToken.None)
.PipeTo(Sender);
}
private void HandleRemoveConnection(RemoveConnectionCommand command)
{
if (_connectionActors.TryGetValue(command.ConnectionName, out var actor))
{
Context.Stop(actor);
_connectionActors.Remove(command.ConnectionName);
_healthCollector.RemoveConnection(command.ConnectionName);
_log.Info("Removed DataConnectionActor for {0}", command.ConnectionName);
}
}
private void HandleGetAllHealthReports(GetAllHealthReports _)
{
// Forward health report requests to all connection actors
foreach (var actor in _connectionActors.Values)
{
actor.Forward(new DataConnectionActor.GetHealthReport());
}
}
///
protected override SupervisorStrategy SupervisorStrategy()
{
return new OneForOneStrategy(
maxNrOfRetries: 10,
withinTimeRange: TimeSpan.FromMinutes(1),
decider: Decider.From(ex =>
{
_log.Warning(ex, "DataConnectionActor threw exception, resuming (subscription state preserved)");
return Directive.Resume;
}));
}
}
///
/// Command to remove a data connection actor.
///
public record RemoveConnectionCommand(string ConnectionName);
///
/// Request for health reports from all active connections.
///
public record GetAllHealthReports;