using System.Collections.Concurrent; using Akka.Actor; using Akka.TestKit.Xunit2; using Grpc.Net.Client; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging.Abstractions; using Microsoft.Extensions.Options; using NSubstitute; using ZB.MOM.WW.Audit; using ZB.MOM.WW.ScadaBridge.AuditLog.Central; using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories; using ZB.MOM.WW.ScadaBridge.Commons.Messages.Integration; using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit; using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; using ProtoPullRequest = ZB.MOM.WW.ScadaBridge.Communication.Grpc.PullAuditEventsRequest; using ProtoPullResponse = ZB.MOM.WW.ScadaBridge.Communication.Grpc.PullAuditEventsResponse; namespace ZB.MOM.WW.ScadaBridge.AuditLog.Tests.Central; /// /// Review 2026-10-21 06/MEM-2: per-site reconciliation state and the per-(site, endpoint) gRPC /// pull-channel caches are pruned for sites absent from the enumeration, and the /// abandoned-EventId short-circuit set is bounded. /// public class ReconciliationStatePruningTests : TestKit { private sealed class MutableEnumerator : ISiteEnumerator { public volatile IReadOnlyList Sites = Array.Empty(); public Task> EnumerateAsync(CancellationToken ct = default) => Task.FromResult(Sites); } private sealed class PruneRecordingInvoker : GrpcPullAuditEventsClient.IPullAuditEventsInvoker, IGrpcChannelCachePruner { public List> Retained { get; } = new(); public Task InvokeAsync( string siteId, string endpoint, ProtoPullRequest request, CancellationToken ct) => Task.FromResult(new ProtoPullResponse()); public void RetainOnly(IReadOnlyList liveSites) => Retained.Add(liveSites); } private sealed class AlwaysMoreClient : IPullAuditEventsClient { public Task PullAsync( string siteId, DateTime sinceUtc, string? afterId, int batchSize, CancellationToken ct) => Task.FromResult(new PullAuditEventsResponse( new[] { ScadaBridgeAuditEventFactory.Create( eventId: Guid.NewGuid(), occurredAtUtc: DateTime.UtcNow, channel: AuditChannel.ApiOutbound, kind: AuditKind.ApiCall, status: AuditStatus.Delivered, sourceSiteId: siteId), }, MoreAvailable: true)); } /// The shared pruning helper disposes channels whose (site, endpoint) is no longer enumerated. [Fact] public void RetainOnly_DropsChannelsOfDeletedSitesAndStaleAddresses() { var channels = new ConcurrentDictionary<(string Site, string Endpoint), GrpcChannel>(); channels[("a", "http://a1:8083")] = GrpcChannel.ForAddress("http://a1:8083"); channels[("a", "http://a-old:8083")] = GrpcChannel.ForAddress("http://a-old:8083"); channels[("a", "http://a2:8083")] = GrpcChannel.ForAddress("http://a2:8083"); channels[("gone", "http://g:8083")] = GrpcChannel.ForAddress("http://g:8083"); var pruned = GrpcChannelCachePruning.RetainOnly(channels, new[] { new SiteEntry("a", "http://a1:8083", "http://a2:8083") }); Assert.Equal(2, pruned); Assert.Equal( new[] { ("a", "http://a1:8083"), ("a", "http://a2:8083") }.OrderBy(k => k.Item2), channels.Keys.Select(k => (k.Site, k.Endpoint)).OrderBy(k => k.Item2)); foreach (var c in channels.Values) c.Dispose(); } /// The pull client hands every enumeration it makes to a pruning-capable invoker. /// A task that represents the asynchronous operation. [Fact] public async Task PullClient_PrunesTheInvokerCache_WithEachEnumeration() { var sites = new MutableEnumerator { Sites = new[] { new SiteEntry("a", "http://a:8083") } }; var invoker = new PruneRecordingInvoker(); var sut = new GrpcPullAuditEventsClient(sites, invoker, NullLogger.Instance); await sut.PullAsync("a", DateTime.MinValue, null, 10, CancellationToken.None); Assert.Same(sites.Sites, Assert.Single(invoker.Retained)); } /// A site removed from the enumeration loses its latched stalled state (published false). [Fact] public void RemovedSite_IsPruned_AndItsStalledLatchCleared() { var sites = new MutableEnumerator { Sites = new[] { new SiteEntry("doomed", "http://d:8083") } }; var services = new ServiceCollection(); services.AddScoped(_ => Substitute.For()); var probe = CreateTestProbe(); Sys.EventStream.Subscribe(probe.Ref, typeof(SiteAuditTelemetryStalledChanged)); var actor = Sys.ActorOf(Props.Create(() => new SiteAuditReconciliationActor( sites, new AlwaysMoreClient(), services.BuildServiceProvider(), Options.Create(new SiteAuditReconciliationOptions { ReconciliationIntervalOverride = TimeSpan.FromMilliseconds(100), MaxPagesPerTick = 1, StalledAfterNonDrainingCycles = 1, }), NullLogger.Instance))); Assert.True(probe.ExpectMsg(TimeSpan.FromSeconds(5)).Stalled); // Not an empty list: an empty enumeration deliberately prunes nothing. sites.Sites = new[] { new SiteEntry("survivor", "http://s:8083") }; var cleared = probe.FishForMessage( m => m.SiteId == "doomed", TimeSpan.FromSeconds(5)); Assert.Equal("doomed", cleared.SiteId); Assert.False(cleared.Stalled); Watch(actor); Sys.Stop(actor); ExpectTerminated(actor); } /// The abandoned-EventId short-circuit set never grows past its bound. [Fact] public void AbandonedEventIdSet_IsBounded() { var actorRef = ActorOfAsTestActorRef(Props.Create(() => new SiteAuditReconciliationActor( new MutableEnumerator(), new AlwaysMoreClient(), new ServiceCollection().BuildServiceProvider(), Options.Create(new SiteAuditReconciliationOptions { ReconciliationIntervalSeconds = 3600 }), NullLogger.Instance))); var actor = actorRef.UnderlyingActor; for (var i = 0; i < SiteAuditReconciliationActor.MaxRememberedAbandonedEventIds + 5; i++) { actor.RememberAbandoned(Guid.NewGuid()); } Assert.True(actor.AbandonedEventIdCountForTests <= SiteAuditReconciliationActor.MaxRememberedAbandonedEventIds); } }