using Akka.Actor;
using Google.Protobuf.WellKnownTypes;
using Grpc.Core;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Audit;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Deployment;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Health;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Notification;
using GrpcStatus = Grpc.Core.Status;
namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc;
///
/// Central-hosted gRPC face of the seven site→central control messages
/// (Protos/central_control.proto). Decodes each request onto the SAME in-process
/// message type the removed Akka ClusterClient path used to carry, Asks
/// , and encodes the reply back.
///
///
///
/// Direction is inverted from . That server runs on a
/// site and central dials in; this one runs on CENTRAL and the site dials in. The two listen on
/// the same port number (8083) on their respective nodes, which is symmetry, not a collision —
/// a node is either central or a site, never both.
///
///
/// Zero handler logic lives here. Every RPC lands on a receive
/// CentralCommunicationActor already implements, so transport and handler cannot drift
/// in behaviour: the actor is the single implementation, and this class
/// is a codec plus an Ask. That is also why the service takes the actor through
/// rather than resolving anything from DI — the actor is created by the
/// host's Akka bootstrap, not by the container.
///
///
/// The two audit-ingest RPCs are the exception: they Ask the ingest singleton DIRECTLY.
/// CentralCommunicationActor used to relay them — an extra hop that re-Asked the
/// same audit-log-ingest proxy with the same 30 s
/// , so the inner Ask expired at the same
/// instant as the outer one and could only add latency plus a
/// repackaging. The proxy now arrives through
/// ; the redundant actor relay handlers were deleted.
///
///
/// Fault semantics. Failed central ingest returns a non-OK status, leaving the site's
/// rows Pending and making a lost batch visible to the next drain. A proxy not handed over
/// yet returns an empty ack. The legacy SiteStreamService ingest methods are retired and
/// always return Unimplemented; they are not an alternate ingestion path.
///
///
/// Status mapping, and why it is not uniform. A site transport may safely re-send a call
/// to the peer central node only when the call provably never ran. So:
///
///
/// - plus the server-owned not-started trailer — this
/// node is not ready; nothing was dispatched, so a cross-node retry is safe. Unavailable alone
/// can also be generated by a transport after application and does not prove retry safety.
/// - — the Ask timed out. The message WAS
/// delivered and may have been processed; retrying it on the other node would duplicate work.
/// - — the handler faulted (a piped
/// , e.g. a database error inside reconcile). Same
/// reasoning: it ran, so do not re-send it elsewhere.
///
///
public sealed class CentralControlGrpcService : CentralControlService.CentralControlServiceBase
{
private readonly ILogger _logger;
private readonly CommunicationOptions _options;
// Null until the host's Akka bootstrap hands the actor over. Doubles as the readiness
// flag: a call arriving before then cannot be served and is refused with Unavailable.
// Volatile because SetReady runs on the startup thread while calls are served on
// Kestrel's thread pool.
private volatile IActorRef? _central;
// Central-side audit-ingest singleton proxy (audit-log-ingest). Handed over by the host
// AFTER the singleton starts, so it arrives strictly later than _central. The two ingest
// RPCs Ask THIS proxy directly rather than relaying through CentralCommunicationActor —
// see the class remarks. Null until registration completes; an ingest arriving before then
// gets an empty ack (leave the site's rows Pending, retry next drain), which is exactly
// what the removed relay replied in the same window.
private volatile IActorRef? _auditIngest;
///
/// Creates the service. This must remain the only public constructor — see
/// for how the actor arrives, and the Host's
/// CentralControlAuthInterceptor for the interceptor-side version of the same rule.
///
/// Logger for readiness and fault diagnostics.
/// Communication options supplying the per-RPC Ask timeouts.
public CentralControlGrpcService(
ILogger logger,
IOptions options)
{
ArgumentNullException.ThrowIfNull(logger);
ArgumentNullException.ThrowIfNull(options);
_logger = logger;
_options = options.Value;
}
///
/// Hands the CentralCommunicationActor to the service and opens the gate. Mirrors
/// : the gRPC service is a DI singleton created
/// before the actor system exists, so the actor arrives post-construction.
///
///
/// The contract is deliberately narrow, exactly as on the site side: it asserts that the
/// actor exists and can receive, NOT that every downstream singleton proxy
/// (notification-outbox, audit-log-ingest) has registered itself yet. Those
/// register moments later in the same startup path, and the actor already answers a call
/// that beats them with a "not available, retry" reply
/// — so gating readiness on them would add nothing but a longer window in which sites
/// see .
///
/// The central communication actor.
public void SetReady(IActorRef centralCommunicationActor)
{
ArgumentNullException.ThrowIfNull(centralCommunicationActor);
_central = centralCommunicationActor;
}
///
/// Hands the central AuditLogIngestActor singleton proxy to the service so the two
/// ingest RPCs can Ask it directly. SiteStreamService retains only Unimplemented wire stubs.
///
///
/// Removing the CentralCommunicationActor relay removes an actor hop, a second Ask,
/// and a second timeout from every audit batch: the relay's own Ask was armed with the same
/// 30 s as this service's, so the two
/// expired together and the inner one could only ever add latency and a
/// repackaging on the way out.
///
/// The audit-log-ingest singleton proxy.
public void SetAuditIngestActor(IActorRef auditIngestProxy)
{
ArgumentNullException.ThrowIfNull(auditIngestProxy);
_auditIngest = auditIngestProxy;
}
/// Exposed for wiring assertions in tests.
internal bool IsReady => _central is not null;
/// Exposed for wiring assertions in tests.
internal bool IsAuditIngestBound => _auditIngest is not null;
///
public override async Task SubmitNotification(
NotificationSubmitDto request, ServerCallContext context)
{
var central = RequireReady(context);
var ack = await AskAsync(
central,
CentralControlDtoMapper.FromDto(request),
_options.NotificationForwardTimeout,
context).ConfigureAwait(false);
return CentralControlDtoMapper.ToDto(ack);
}
///
public override async Task QueryNotificationStatus(
NotificationStatusQueryDto request, ServerCallContext context)
{
var central = RequireReady(context);
var response = await AskAsync(
central,
CentralControlDtoMapper.FromDto(request) with { VerifiedSourceSiteId = CentralControlSiteIdentity.RequireVerified(context) },
_options.NotificationForwardTimeout,
context).ConfigureAwait(false);
return CentralControlDtoMapper.ToDto(response);
}
///
public override async Task IngestAuditEvents(
AuditEventBatch request, ServerCallContext context)
{
// An empty batch is a no-op the actor need never see; answering it here also means a
// site that drains an empty queue does not fail against a not-yet-ready central.
if (request.Events.Count == 0)
{
return new IngestAck();
}
// Node-readiness gate first (the actor system must be up), then the ingest proxy.
RequireReady(context);
var ingest = RequireAuditIngest(request.Events.Count, nameof(IngestAuditEvents));
if (ingest is null)
{
return new IngestAck();
}
var reply = await AskAsync(
ingest,
CentralControlDtoMapper.FromDto(request),
SiteStreamGrpcServer.AuditIngestAskTimeout,
context).ConfigureAwait(false);
return CentralControlDtoMapper.ToIngestAck(reply.AcceptedEventIds);
}
///
public override async Task IngestCachedTelemetry(
CachedTelemetryBatch request, ServerCallContext context)
{
if (request.Packets.Count == 0)
{
return new IngestAck();
}
RequireReady(context);
var ingest = RequireAuditIngest(request.Packets.Count, nameof(IngestCachedTelemetry));
if (ingest is null)
{
return new IngestAck();
}
var reply = await AskAsync(
ingest,
CentralControlDtoMapper.FromDto(request),
SiteStreamGrpcServer.AuditIngestAskTimeout,
context).ConfigureAwait(false);
return CentralControlDtoMapper.ToIngestAck(reply.AcceptedEventIds);
}
///
/// Returns the audit-ingest singleton proxy, or null when the host has not handed it
/// over yet (it starts moments after ). A null answer means the caller
/// must reply with an EMPTY — nothing accepted, rows stay
/// Pending on the site and drain on the next tick. That is deliberately NOT
/// : readiness here is intentionally narrow (see
/// ), and this is the exact reply the removed
/// CentralCommunicationActor relay produced in the same window.
///
private IActorRef? RequireAuditIngest(int itemCount, string rpc)
{
var ingest = _auditIngest;
if (ingest is not null)
{
return ingest;
}
_logger.LogWarning(
"{Rpc} received {Count} item(s) before SetAuditIngestActor was called; returning an "
+ "empty ack so the site keeps its rows Pending and retries.",
rpc, itemCount);
return null;
}
///
public override async Task ReconcileSite(
ReconcileSiteRequestDto request, ServerCallContext context)
{
var central = RequireReady(context);
var response = await AskAsync(
central,
CentralControlDtoMapper.FromDto(request),
_options.QueryTimeout,
context).ConfigureAwait(false);
return CentralControlDtoMapper.ToDto(response);
}
///
public override async Task ReportSiteHealth(
SiteHealthReportDto request, ServerCallContext context)
{
var central = RequireReady(context);
var ack = await AskAsync(
central,
CentralControlDtoMapper.FromDto(request),
_options.HealthReportTimeout,
context).ConfigureAwait(false);
return CentralControlDtoMapper.ToDto(ack);
}
// Application heartbeat — always answers OK, even when this node is not ready.
// The heartbeat is fire-and-forget on both sides: nothing on the site consumes the reply,
// and the site's heartbeat timer must never take a fault (a failing heartbeat that raised
// an error would be a self-inflicted outage on a purely informational signal). So this is
// the one RPC that does not go through RequireReady: a heartbeat arriving before the actor
// exists is logged and dropped, exactly as the actor itself drops one that arrives before
// ICentralHealthAggregator is resolvable. Liveness is still detected — the aggregator's
// offline timeout fires when the heartbeats stop landing.
///
public override Task Heartbeat(HeartbeatDto request, ServerCallContext context)
{
var central = _central;
if (central is null)
{
_logger.LogDebug(
"Dropped a heartbeat from site {SiteId}: the central communication actor is not "
+ "ready yet. Heartbeats are fire-and-forget, so the call still succeeds.",
request.SiteId);
return Task.FromResult(new Empty());
}
// Tell, never Ask: the actor's HandleHeartbeat sends no reply.
central.Tell(CentralControlDtoMapper.FromDto(request), ActorRefs.NoSender);
return Task.FromResult(new Empty());
}
///
/// Returns the central communication actor, or throws
/// when the host has not finished bringing the actor system up.
///
private IActorRef RequireReady(ServerCallContext context)
{
var central = _central;
if (central is not null)
{
return central;
}
_logger.LogWarning(
"Refused a control-plane call to {Method}: the central communication actor is not "
+ "ready yet. Nothing was dispatched, so the caller may retry (including against "
+ "the peer central node).",
context.Method);
throw GrpcRetrySafety.NotDispatched("Central control plane is not ready: the actor system is still starting.");
}
///
/// Asks the central actor and maps a fault onto the status code that tells the caller
/// whether a cross-node retry is safe. See the class remarks for the mapping rationale.
///
private async Task AskAsync(
IActorRef central, object message, TimeSpan timeout, ServerCallContext context)
{
try
{
return await central.Ask(message, timeout, context.CancellationToken)
.ConfigureAwait(false);
}
catch (AskTimeoutException ex)
{
_logger.LogWarning(ex,
"Control-plane call {Method} timed out after {Timeout} waiting for the central "
+ "communication actor.",
context.Method, timeout);
throw new RpcException(new GrpcStatus(
StatusCode.DeadlineExceeded,
$"Central did not answer within {timeout}."));
}
catch (OperationCanceledException)
{
// The client gave up or its deadline expired; there is no one left to answer.
throw new RpcException(new GrpcStatus(
StatusCode.Cancelled, "The call was cancelled."));
}
catch (Exception ex)
{
_logger.LogError(ex,
"Control-plane call {Method} faulted inside the central communication actor.",
context.Method);
throw new RpcException(new GrpcStatus(
StatusCode.Internal, "Central failed to process the call."));
}
}
}