ScadaBridge/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Cluster/CentralCacheInvalidationTwoNodeTests.cs

using Akka.Actor;
using Akka.Cluster.Tools.PublishSubscribe;
using Microsoft.Extensions.Logging.Abstractions;
using Xunit;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Communication;
using ZB.MOM.WW.ScadaBridge.Host;

namespace ZB.MOM.WW.ScadaBridge.IntegrationTests.Cluster;

/// <summary>
/// ScadaBridge#100 item 3 / #83 / #103 across a REAL two-node central cluster built from the
/// production HOCON: an invalidation published on one node's bus is delivered, through
/// <c>DistributedPubSub</c> and the production serializer, to the other node's handlers — and
/// not back to the publishing node's own handlers.
/// </summary>
public class CentralCacheInvalidationTwoNodeTests
{
    private sealed class RecordingHandler : ICentralCacheInvalidationHandler
    {
        private readonly List<ICentralCacheInvalidation> _applied = new();

        public IReadOnlyList<ICentralCacheInvalidation> Applied
        {
            get { lock (_applied) return _applied.ToList(); }
        }

        public void Apply(ICentralCacheInvalidation invalidation)
        {
            lock (_applied) _applied.Add(invalidation);
        }
    }

    private static (CentralCacheInvalidationBus Bus, RecordingHandler Handler) Wire(ActorSystem node)
    {
        var bus = new CentralCacheInvalidationBus(NullLogger<CentralCacheInvalidationBus>.Instance);
        var handler = new RecordingHandler();
        node.ActorOf(Props.Create(() => new CentralCacheInvalidationSubscriber(
            bus, new ICentralCacheInvalidationHandler[] { handler })), "central-cache-invalidation");
        bus.SetMediator(DistributedPubSub.Get(node).Mediator);
        return (bus, handler);
    }

    /// <summary>
    /// Every typed invalidation crosses to the peer intact; the publisher never applies its own.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task Invalidations_ReachThePeerNode_AndNotThePublisher()
    {
        await using var cluster = await TwoNodeClusterFixture.StartAsync();
        var a = Wire(cluster.NodeA);
        var b = Wire(cluster.NodeB);

        // Subscriptions spread by gossip, so probe until the peer has registered its subscriber.
        var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(20);
        while (b.Handler.Applied.Count == 0)
        {
            Assert.True(DateTime.UtcNow < deadline, "node B never received node A's invalidation");
            a.Bus.Publish(new RoleMappingsInvalidated());
            await Task.Delay(250);
        }

        a.Bus.Publish(new ExternalDefinitionsInvalidated());
        a.Bus.Publish(new InboundApiMethodInvalidated("MesMoveIn"));
        a.Bus.Publish(new SitePskInvalidated("site-a"));

        deadline = DateTime.UtcNow + TimeSpan.FromSeconds(10);
        while (!b.Handler.Applied.Any(i => i is SitePskInvalidated))
        {
            Assert.True(DateTime.UtcNow < deadline, "node B did not receive the later invalidations");
            await Task.Delay(100);
        }

        var applied = b.Handler.Applied;
        Assert.Contains(applied, i => i is RoleMappingsInvalidated);
        Assert.Contains(new ExternalDefinitionsInvalidated(), applied);
        Assert.Contains(new InboundApiMethodInvalidated("MesMoveIn"), applied);
        // One topic only since ScadaBridge#105: the PSK invalidation arrives exactly once.
        await Task.Delay(500);
        Assert.Single(b.Handler.Applied, i => i is SitePskInvalidated);
        Assert.Contains(new SitePskInvalidated("site-a"), b.Handler.Applied);

        // The publisher skips its own envelopes, and with the legacy bare PSK topic gone nothing
        // comes back to it at all.
        Assert.Empty(a.Handler.Applied);
    }
}