using Akka.Actor;
using Akka.Event;
using ScadaLink.Commons.Messages.Communication;
using ScadaLink.Commons.Messages.Health;
namespace ScadaLink.Communication.Actors;
///
/// Central-side actor that routes messages from central to site clusters via Akka remoting.
/// Maintains a registry of known site actor paths (learned from heartbeats/connection events).
///
/// WP-4: All 8 message patterns routed through this actor.
/// WP-5: Ask timeout on connection drop (no central buffering). Debug streams killed on interruption.
///
public class CentralCommunicationActor : ReceiveActor
{
private readonly ILoggingAdapter _log = Context.GetLogger();
///
/// Maps SiteId → remote SiteCommunicationActor selection.
/// Updated when heartbeats arrive or connection state changes.
///
private readonly Dictionary _siteSelections = new();
///
/// Tracks active debug view subscriptions: correlationId → (siteId, subscriber).
/// Used to kill debug streams on site disconnection (WP-5).
///
private readonly Dictionary _debugSubscriptions = new();
///
/// Tracks in-progress deployments: deploymentId → siteId.
/// On central failover, in-progress deployments are treated as failed (WP-5).
///
private readonly Dictionary _inProgressDeployments = new();
public CentralCommunicationActor()
{
// Site registration via heartbeats
Receive(HandleHeartbeat);
// Connection state changes
Receive(HandleConnectionStateChanged);
// Site registration command (manual or from discovery)
Receive(HandleRegisterSite);
// Route enveloped messages to sites
Receive(HandleSiteEnvelope);
}
private void HandleHeartbeat(HeartbeatMessage heartbeat)
{
// Heartbeats arrive from sites — forward to any interested central actors
// The sender's path tells us the site's communication actor address
if (!_siteSelections.ContainsKey(heartbeat.SiteId))
{
var senderPath = Sender.Path.ToString();
_log.Info("Learned site {0} from heartbeat at path {1}", heartbeat.SiteId, senderPath);
}
// Forward heartbeat to parent/subscribers (central health monitoring)
Context.Parent.Tell(heartbeat);
}
private void HandleConnectionStateChanged(ConnectionStateChanged msg)
{
if (!msg.IsConnected)
{
_log.Warning("Site {0} disconnected at {1}", msg.SiteId, msg.Timestamp);
// WP-5: Kill active debug streams for the disconnected site
var toRemove = _debugSubscriptions
.Where(kvp => kvp.Value.SiteId == msg.SiteId)
.ToList();
foreach (var kvp in toRemove)
{
_log.Info("Killing debug stream {0} for disconnected site {1}", kvp.Key, msg.SiteId);
kvp.Value.Subscriber.Tell(new DebugStreamTerminated(msg.SiteId, kvp.Key));
_debugSubscriptions.Remove(kvp.Key);
}
// WP-5: Mark in-progress deployments as failed
var failedDeployments = _inProgressDeployments
.Where(kvp => kvp.Value == msg.SiteId)
.Select(kvp => kvp.Key)
.ToList();
foreach (var deploymentId in failedDeployments)
{
_log.Warning("Deployment {0} to site {1} treated as failed due to disconnection",
deploymentId, msg.SiteId);
_inProgressDeployments.Remove(deploymentId);
}
_siteSelections.Remove(msg.SiteId);
}
else
{
_log.Info("Site {0} connected at {1}", msg.SiteId, msg.Timestamp);
}
}
private void HandleRegisterSite(RegisterSite msg)
{
var selection = Context.ActorSelection(msg.RemoteActorPath);
_siteSelections[msg.SiteId] = selection;
_log.Info("Registered site {0} at path {1}", msg.SiteId, msg.RemoteActorPath);
}
private void HandleSiteEnvelope(SiteEnvelope envelope)
{
if (!_siteSelections.TryGetValue(envelope.SiteId, out var siteSelection))
{
_log.Warning("No known path for site {0}, cannot route message {1}",
envelope.SiteId, envelope.Message.GetType().Name);
// The Ask will timeout on the caller side — no central buffering (WP-5)
return;
}
// Track debug subscriptions for cleanup on disconnect
TrackMessageForCleanup(envelope);
// Forward the inner message to the site, preserving the original sender
// so the site can reply directly to the caller (completing the Ask pattern)
siteSelection.Tell(envelope.Message, Sender);
}
private void TrackMessageForCleanup(SiteEnvelope envelope)
{
switch (envelope.Message)
{
case Commons.Messages.DebugView.SubscribeDebugViewRequest sub:
_debugSubscriptions[sub.CorrelationId] = (envelope.SiteId, Sender);
break;
case Commons.Messages.DebugView.UnsubscribeDebugViewRequest unsub:
_debugSubscriptions.Remove(unsub.CorrelationId);
break;
case Commons.Messages.Deployment.DeployInstanceCommand deploy:
_inProgressDeployments[deploy.DeploymentId] = envelope.SiteId;
break;
}
}
protected override void PreStart()
{
_log.Info("CentralCommunicationActor started");
}
protected override void PostStop()
{
_log.Info("CentralCommunicationActor stopped. In-progress deployments treated as failed (WP-5).");
// On central failover, all in-progress deployments are failed
_inProgressDeployments.Clear();
_debugSubscriptions.Clear();
}
}
///
/// Command to register a site's remote communication actor path.
///
public record RegisterSite(string SiteId, string RemoteActorPath);
///
/// Notification sent to debug view subscribers when the stream is terminated
/// due to site disconnection (WP-5).
///
public record DebugStreamTerminated(string SiteId, string CorrelationId);