ScadaBridge/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/ReconciliationStatePruningTests.cs

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;

/// <summary>
/// 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.
/// </summary>
public class ReconciliationStatePruningTests : TestKit
{
    private sealed class MutableEnumerator : ISiteEnumerator
    {
        public volatile IReadOnlyList<SiteEntry> Sites = Array.Empty<SiteEntry>();
        public Task<IReadOnlyList<SiteEntry>> EnumerateAsync(CancellationToken ct = default) => Task.FromResult(Sites);
    }

    private sealed class PruneRecordingInvoker
        : GrpcPullAuditEventsClient.IPullAuditEventsInvoker, IGrpcChannelCachePruner
    {
        public List<IReadOnlyList<SiteEntry>> Retained { get; } = new();
        public Task<ProtoPullResponse> InvokeAsync(
            string siteId, string endpoint, ProtoPullRequest request, CancellationToken ct) =>
            Task.FromResult(new ProtoPullResponse());
        public void RetainOnly(IReadOnlyList<SiteEntry> liveSites) => Retained.Add(liveSites);
    }

    private sealed class AlwaysMoreClient : IPullAuditEventsClient
    {
        public Task<PullAuditEventsResponse> 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));
    }

    /// <summary>The shared pruning helper disposes channels whose (site, endpoint) is no longer enumerated.</summary>
    [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();
    }

    /// <summary>The pull client hands every enumeration it makes to a pruning-capable invoker.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [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<GrpcPullAuditEventsClient>.Instance);

        await sut.PullAsync("a", DateTime.MinValue, null, 10, CancellationToken.None);

        Assert.Same(sites.Sites, Assert.Single(invoker.Retained));
    }

    /// <summary>A site removed from the enumeration loses its latched stalled state (published false).</summary>
    [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<IAuditLogRepository>());
        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<SiteAuditReconciliationActor>.Instance)));

        Assert.True(probe.ExpectMsg<SiteAuditTelemetryStalledChanged>(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<SiteAuditTelemetryStalledChanged>(
            m => m.SiteId == "doomed", TimeSpan.FromSeconds(5));
        Assert.Equal("doomed", cleared.SiteId);
        Assert.False(cleared.Stalled);

        Watch(actor);
        Sys.Stop(actor);
        ExpectTerminated(actor);
    }

    /// <summary>The abandoned-EventId short-circuit set never grows past its bound.</summary>
    [Fact]
    public void AbandonedEventIdSet_IsBounded()
    {
        var actorRef = ActorOfAsTestActorRef<SiteAuditReconciliationActor>(Props.Create(() => new SiteAuditReconciliationActor(
            new MutableEnumerator(),
            new AlwaysMoreClient(),
            new ServiceCollection().BuildServiceProvider(),
            Options.Create(new SiteAuditReconciliationOptions { ReconciliationIntervalSeconds = 3600 }),
            NullLogger<SiteAuditReconciliationActor>.Instance)));
        var actor = actorRef.UnderlyingActor;

        for (var i = 0; i < SiteAuditReconciliationActor.MaxRememberedAbandonedEventIds + 5; i++)
        {
            actor.RememberAbandoned(Guid.NewGuid());
        }

        Assert.True(actor.AbandonedEventIdCountForTests <= SiteAuditReconciliationActor.MaxRememberedAbandonedEventIds);
    }
}