using Akka.Actor; using Akka.TestKit.Xunit2; using Microsoft.Extensions.Logging.Abstractions; using NSubstitute; using NSubstitute.ExceptionExtensions; using ZB.MOM.WW.Audit; using ZB.MOM.WW.ScadaBridge.AuditLog.Central; using ZB.MOM.WW.ScadaBridge.AuditLog.Tests.TestSupport; using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories; using ZB.MOM.WW.ScadaBridge.Commons.Messages.Audit; using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit; using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; namespace ZB.MOM.WW.ScadaBridge.AuditLog.Tests.Central; /// /// Review 2026-10-21 06/IMPL-1 adversarial follow-up: central reports a row as REJECTED only for a /// permanent, row-intrinsic SQL failure; a database outage / timeout leaves the id in neither list, /// so the site never mistakes a maintenance window for a poison row. Also pins the classifier the /// reconciliation puller uses to decide what counts toward permanent abandonment (06/COR follow-up 3). /// public class AuditIngestRejectionTests : TestKit { private static AuditEvent NewEvent() => ScadaBridgeAuditEventFactory.Create( eventId: Guid.NewGuid(), occurredAtUtc: new DateTime(2026, 5, 20, 10, 0, 0, DateTimeKind.Utc), channel: AuditChannel.ApiOutbound, kind: AuditKind.ApiCall, status: AuditStatus.Delivered, sourceSiteId: "site-a"); /// The batch reply lists only the over-width row as rejected; the timed-out row is in neither list. [Fact] public void Ingest_ReportsOnlyPermanentRowFailuresAsRejected() { var ok = NewEvent(); var overWidth = NewEvent(); var timedOut = NewEvent(); var repo = Substitute.For(); repo.InsertManyIfNotExistsAsync(Arg.Any>(), Arg.Any(), Arg.Any()) .ThrowsAsync(SqlExceptionFactory.Create(2628)); repo.InsertIfNotExistsAsync(Arg.Is(e => e.EventId == overWidth.EventId), Arg.Any()) .ThrowsAsync(SqlExceptionFactory.Create(2628)); repo.InsertIfNotExistsAsync(Arg.Is(e => e.EventId == timedOut.EventId), Arg.Any()) .ThrowsAsync(SqlExceptionFactory.Create(-2)); var actor = Sys.ActorOf(Props.Create(() => new AuditLogIngestActor( StubAuditRepositoryProvider.Build(repo), NullLogger.Instance))); actor.Tell(new IngestAuditEventsCommand(new[] { ok, overWidth, timedOut }), TestActor); var reply = ExpectMsg(TimeSpan.FromSeconds(10)); Assert.Equal(new[] { ok.EventId }, reply.AcceptedEventIds); Assert.Equal(new[] { overWidth.EventId }, reply.RejectedEventIds); } /// Classifier: row-intrinsic SQL errors are permanent; timeouts, connection loss and non-SQL faults are not. [Fact] public void Classifier_SeparatesRowVerdictsFromDatabaseProblems() { Assert.True(AuditIngestErrorClassifier.IsPermanentRowRejection(SqlExceptionFactory.Create(2628))); Assert.True(AuditIngestErrorClassifier.IsPermanentRowRejection(SqlExceptionFactory.Create(515))); Assert.True(AuditIngestErrorClassifier.IsPermanentRowRejection( new InvalidOperationException("wrapped", SqlExceptionFactory.Create(8152)))); Assert.False(AuditIngestErrorClassifier.IsPermanentRowRejection(SqlExceptionFactory.Create(-2))); Assert.False(AuditIngestErrorClassifier.IsPermanentRowRejection(new TimeoutException())); Assert.False(AuditIngestErrorClassifier.IsPermanentRowRejection(new InvalidOperationException("bug"))); Assert.True(AuditIngestErrorClassifier.IsTransient(SqlExceptionFactory.Create(-2))); Assert.True(AuditIngestErrorClassifier.IsTransient(SqlExceptionFactory.Create(1205))); Assert.True(AuditIngestErrorClassifier.IsTransient(SqlExceptionFactory.Create(40613))); Assert.True(AuditIngestErrorClassifier.IsTransient(new TimeoutException())); Assert.True(AuditIngestErrorClassifier.IsTransient(new TaskCanceledException())); Assert.True(AuditIngestErrorClassifier.IsTransient( new InvalidOperationException("wrapped", new System.Net.Sockets.SocketException()))); Assert.False(AuditIngestErrorClassifier.IsTransient(SqlExceptionFactory.Create(2628))); Assert.False(AuditIngestErrorClassifier.IsTransient(new InvalidOperationException("bug"))); } private sealed class OneRowClient : IPullAuditEventsClient { private readonly AuditEvent[] _rows; public OneRowClient(params AuditEvent[] rows) => _rows = rows; public System.Collections.Concurrent.ConcurrentQueue Since { get; } = new(); public Task PullAsync( string siteId, DateTime sinceUtc, string? afterId, int batchSize, CancellationToken ct) { Since.Enqueue(sinceUtc); return Task.FromResult(new ZB.MOM.WW.ScadaBridge.Commons.Messages.Integration.PullAuditEventsResponse( _rows, MoreAvailable: false)); } } private sealed class OneSite : ISiteEnumerator { public Task> EnumerateAsync(CancellationToken ct = default) => Task.FromResult>(new[] { new SiteEntry("site-a", "http://a:8083") }); } /// /// A row that keeps hitting a TRANSIENT database fault is retried with the cursor held back for /// far longer than the permanent-abandon threshold, and is never abandoned. /// [Fact] public void Reconciliation_TransientFaults_NeverCountTowardAbandonment() { var row = NewEvent(); var repo = Substitute.For(); repo.InsertManyIfNotExistsAsync(Arg.Any>(), Arg.Any(), Arg.Any()) .ThrowsAsync(SqlExceptionFactory.Create(-2)); repo.InsertIfNotExistsAsync(Arg.Any(), Arg.Any()) .ThrowsAsync(SqlExceptionFactory.Create(-2)); var client = new OneRowClient(row); var actor = Sys.ActorOf(Props.Create(() => new SiteAuditReconciliationActor( new OneSite(), client, StubAuditRepositoryProvider.Build(repo), Microsoft.Extensions.Options.Options.Create(new SiteAuditReconciliationOptions { ReconciliationIntervalOverride = TimeSpan.FromMilliseconds(30), }), NullLogger.Instance))); AwaitCondition(() => client.Since.Count >= 12, TimeSpan.FromSeconds(10)); Watch(actor); Sys.Stop(actor); ExpectTerminated(actor); Assert.All(client.Since, s => Assert.Equal(DateTime.MinValue, s)); // cursor held every tick repo.DidNotReceive().InsertIfNotExistsAsync( Arg.Is(e => e.Action.EndsWith(".ReconciliationAbandoned", StringComparison.Ordinal)), Arg.Any()); } private SiteAuditReconciliationOptions FastOptions() => new() { ReconciliationIntervalOverride = TimeSpan.FromMilliseconds(30), }; private void RunTicks(IAuditLogRepository repo, OneRowClient client, int ticks) { var actor = Sys.ActorOf(Props.Create(() => new SiteAuditReconciliationActor( new OneSite(), client, StubAuditRepositoryProvider.Build(repo), Microsoft.Extensions.Options.Options.Create(FastOptions()), NullLogger.Instance))); AwaitCondition(() => client.Since.Count >= ticks, TimeSpan.FromSeconds(10)); Watch(actor); Sys.Stop(actor); ExpectTerminated(actor); } /// /// Follow-up M1: schema drift after a missed central migration (207 invalid column name on /// every insert) is not a ROW verdict — it never counts toward abandonment, so the site's /// rows are never flipped Reconciled and purged. /// [Fact] public void Reconciliation_SchemaDriftOnEveryRow_NeverAbandons() { var a = NewEvent(); var b = NewEvent(); var repo = Substitute.For(); repo.InsertManyIfNotExistsAsync(Arg.Any>(), Arg.Any(), Arg.Any()) .ThrowsAsync(SqlExceptionFactory.Create(207)); repo.InsertIfNotExistsAsync(Arg.Any(), Arg.Any()) .ThrowsAsync(SqlExceptionFactory.Create(207)); var client = new OneRowClient(a, b); RunTicks(repo, client, 12); Assert.All(client.Since, s => Assert.Equal(DateTime.MinValue, s)); repo.DidNotReceive().InsertIfNotExistsAsync( Arg.Is(e => e.Action.EndsWith(".ReconciliationAbandoned", StringComparison.Ordinal)), Arg.Any()); } /// /// Follow-up M1: a ROW-classified error (8114 conversion) that hits EVERY row of the page — /// e.g. a column whose type a migration changed — is a database problem, not one poison row. /// With no sibling inserting, nothing is abandoned however many ticks pass. /// [Fact] public void Reconciliation_PermanentErrorOnEveryRow_NoSiblingSucceeds_NeverAbandons() { var a = NewEvent(); var b = NewEvent(); var repo = Substitute.For(); repo.InsertManyIfNotExistsAsync(Arg.Any>(), Arg.Any(), Arg.Any()) .ThrowsAsync(SqlExceptionFactory.Create(8114)); repo.InsertIfNotExistsAsync(Arg.Is(e => e.EventId == a.EventId || e.EventId == b.EventId), Arg.Any()) .ThrowsAsync(SqlExceptionFactory.Create(8114)); var client = new OneRowClient(a, b); RunTicks(repo, client, 12); Assert.All(client.Since, s => Assert.Equal(DateTime.MinValue, s)); repo.DidNotReceive().InsertIfNotExistsAsync( Arg.Is(e => e.Action.EndsWith(".ReconciliationAbandoned", StringComparison.Ordinal)), Arg.Any()); } /// /// Follow-up M1, positive half: the same row-classified error on ONE row while a sibling in the /// page inserts IS a poison row — it is abandoned (synthetic row written) after the threshold. /// [Fact] public void Reconciliation_PermanentErrorOnOneRow_WithSiblingSuccess_Abandons() { var ok = NewEvent(); var poison = NewEvent(); var repo = Substitute.For(); repo.InsertManyIfNotExistsAsync(Arg.Any>(), Arg.Any(), Arg.Any()) .ThrowsAsync(SqlExceptionFactory.Create(8114)); repo.InsertIfNotExistsAsync(Arg.Is(e => e.EventId == poison.EventId), Arg.Any()) .ThrowsAsync(SqlExceptionFactory.Create(8114)); var client = new OneRowClient(ok, poison); RunTicks(repo, client, 8); repo.Received().InsertIfNotExistsAsync( Arg.Is(e => e.Action.EndsWith(".ReconciliationAbandoned", StringComparison.Ordinal)), Arg.Any()); } }