using Grpc.Core;
using Microsoft.Extensions.Logging;
using ZB.MOM.WW.HistorianGateway.Client;
using ZB.MOM.WW.HistorianGateway.Contracts.Grpc;
using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian;
using ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Mapping;
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway;
///
/// backed by the HistorianGateway SendEvent path. The
/// drain worker behind SqliteStoreAndForwardSink calls
/// and uses the returned per-event
/// to decide retry vs. dead-letter, so this writer maps every
/// gateway result — success ack, the published client's typed exception hierarchy, raw
/// (defensive), and any unexpected error — onto exactly one outcome per
/// event and never throws.
///
///
///
/// Each event is sent individually so one poison event cannot fail the whole batch: a permanent
/// failure on event N is dead-lettered while its siblings continue. Outcomes are returned in
/// input order, one per event; an empty batch yields an empty list with no gateway call.
///
///
/// Outcome mapping. Success (or store-forward-queued) ack ⇒ .
/// Transient gRPC codes (Unavailable, DeadlineExceeded, ResourceExhausted,
/// Aborted, Internal) and the auth codes (Unauthenticated,
/// PermissionDenied) ⇒ — an auth fix
/// re-enables the batch, so an auth blip never dead-letters. Permanent codes
/// (InvalidArgument, FailedPrecondition, OutOfRange, Unimplemented) ⇒
/// (dead-letter poison — mirrors the Wonderware
/// PerEventStatus==2 boundary). The typed client exceptions are classified by type, or by
/// the they wrap; any other or unclassifiable error defaults to
/// so the drain worker cannot loop a poison
/// event forever.
///
///
public sealed class GatewayAlarmHistorianWriter : IAlarmHistorianWriter, IDisposable
{
private readonly IHistorianGatewayClient _client;
private readonly ILogger _logger;
/// Creates the writer over a gateway client seam.
/// The gateway client used for the SendEvent write path.
/// Logger for per-event outcome diagnostics (never logs event content).
public GatewayAlarmHistorianWriter(IHistorianGatewayClient client, ILogger logger)
{
_client = client ?? throw new ArgumentNullException(nameof(client));
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
}
///
public async Task> WriteBatchAsync(
IReadOnlyList batch, CancellationToken cancellationToken)
{
ArgumentNullException.ThrowIfNull(batch);
if (batch.Count == 0)
{
return Array.Empty();
}
var outcomes = new HistorianWriteOutcome[batch.Count];
for (var i = 0; i < batch.Count; i++)
{
if (cancellationToken.IsCancellationRequested)
{
// Shutdown mid-drain: short-circuit the remaining events to RetryPlease rather than
// calling the gateway with a cancelled token. They stay queued for retry next startup
// — a cancellation must NEVER dead-letter an in-flight event (silent data loss).
outcomes[i] = HistorianWriteOutcome.RetryPlease;
continue;
}
outcomes[i] = await SendOneAsync(batch[i], cancellationToken).ConfigureAwait(false);
}
return outcomes;
}
private async Task SendOneAsync(AlarmHistorianEvent evt, CancellationToken cancellationToken)
{
try
{
var ack = await _client.SendEventAsync(AlarmEventMapper.ToHistorianEvent(evt), cancellationToken)
.ConfigureAwait(false);
return MapAck(ack);
}
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
{
// Cancellation mid-send at shutdown is NOT a poison event. Map to RetryPlease so the
// event stays queued for next startup rather than being dead-lettered (data loss).
_logger.LogDebug("Alarm SendEvent cancelled at shutdown; will retry.");
return HistorianWriteOutcome.RetryPlease;
}
catch (Exception exception)
{
// NEVER throw out of the writer — the drain worker expects a per-event outcome. Classify
// and log only the failure category (no event content, hostnames, or credentials).
var outcome = Classify(exception);
if (outcome == HistorianWriteOutcome.PermanentFail)
{
_logger.LogWarning(
"Alarm SendEvent permanently failed ({Exception}); dead-lettering the event.",
exception.GetType().Name);
}
else
{
_logger.LogDebug(
"Alarm SendEvent transiently failed ({Exception}); will retry.",
exception.GetType().Name);
}
return outcome;
}
}
// A non-success ack that the gateway durably queued (store-forward) is still accepted — do not
// re-drain it. A non-success, non-queued ack is a soft failure: retry rather than dead-letter.
private static HistorianWriteOutcome MapAck(WriteAck ack) =>
ack.Success || ack.Queued ? HistorianWriteOutcome.Ack : HistorianWriteOutcome.RetryPlease;
private static HistorianWriteOutcome Classify(Exception exception) => exception switch
{
// Published client's typed hierarchy (production reality). Unavailable + both auth kinds retry.
HistorianGatewayUnavailableException => HistorianWriteOutcome.RetryPlease,
HistorianGatewayAuthenticationException => HistorianWriteOutcome.RetryPlease,
HistorianGatewayAuthorizationException => HistorianWriteOutcome.RetryPlease,
// A base client exception wrapping a permanent/transient RpcException → classify by inner status.
HistorianGatewayException { InnerException: RpcException inner } => ClassifyStatus(inner.StatusCode),
// Defensive raw RpcException path (the seam type signature permits it).
RpcException rpc => ClassifyStatus(rpc.StatusCode),
// Anything else (incl. a bare base client exception we cannot classify) → dead-letter to avoid
// an infinite drain loop on a poison event.
_ => HistorianWriteOutcome.PermanentFail,
};
private static HistorianWriteOutcome ClassifyStatus(StatusCode code) => code switch
{
StatusCode.Unavailable
or StatusCode.DeadlineExceeded
or StatusCode.ResourceExhausted
or StatusCode.Aborted
or StatusCode.Internal
// An auth fix re-enables the whole batch — never dead-letter on an auth blip.
or StatusCode.Unauthenticated
or StatusCode.PermissionDenied => HistorianWriteOutcome.RetryPlease,
StatusCode.InvalidArgument
or StatusCode.FailedPrecondition
or StatusCode.OutOfRange
or StatusCode.Unimplemented => HistorianWriteOutcome.PermanentFail,
// Unknown/unclassified gRPC code → dead-letter to avoid an infinite drain loop.
_ => HistorianWriteOutcome.PermanentFail,
};
///
/// Disposes the underlying gateway client and its gRPC channel. The concrete
/// implements ; test doubles
/// that only implement are safely no-opped by the cast guard.
///
public void Dispose() => (_client as IDisposable)?.Dispose();
}