ScadaBridge/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/SiteReplicationActorTests.cs

using System.Collections.Concurrent;
using System.Diagnostics.Metrics;
using Akka.Actor;
using Akka.TestKit.Xunit2;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using ZB.MOM.WW.ScadaBridge.SiteRuntime;
using ZB.MOM.WW.ScadaBridge.SiteRuntime.Actors;
using ZB.MOM.WW.ScadaBridge.SiteRuntime.Deployment;
using ZB.MOM.WW.ScadaBridge.SiteRuntime.Messages;
using ZB.MOM.WW.ScadaBridge.SiteRuntime.Persistence;
using ZB.MOM.WW.ScadaBridge.StoreAndForward;
using ZB.MOM.WW.ScadaBridge.Commons.Observability;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
using ZB.MOM.WW.ScadaBridge.TestSupport;

namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests.Actors;

/// <summary>
/// Tests for <see cref="SiteReplicationActor"/>'s notify-and-fetch config replication:
/// the active node now replicates an id-only <see cref="ReplicateConfigDeploy"/> (no inline
/// config JSON — killing the intra-site 128 KB frame trap), and the standby fetches the
/// config from central over HTTP and writes it with the older-write guard.
/// </summary>
public class SiteReplicationActorTests : TestKit, IDisposable
{
    // Cluster provider is required because SiteReplicationActor calls Cluster.Get in its ctor
    // and subscribes to cluster events in PreStart. We use the in-memory TestTransport (not
    // dot-netty) so no real socket is bound and no DNS lookup happens — the actor only needs
    // the cluster extension to load; these tests never form a real two-node cluster.
    private const string ClusterConfig = @"
akka {
    actor { provider = cluster }
    remote {
        enabled-transports = [""akka.remote.test""]
        test {
            transport-class = ""Akka.Remote.Transport.TestTransport, Akka.Remote""
            applied-adapters = []
            registry-key = site-repl-test
            local-address = ""test://site-repl@localhost:1""
            maximum-payload-bytes = 128000b
            scheme-identifier = test
        }
    }
    cluster { roles = [""site-test""] }
    loglevel = WARNING
}";

    private const string SiteRole = "site-test";

    private readonly SiteStorageService _storage;
    private readonly TestLocalDb _siteLocalDb;
    private readonly TestLocalDb _sfLocalDb;
    private readonly StoreAndForwardStorage _sfStorage;
    private readonly ReplicationService _replicationService;
    private readonly string _sfDbFile;

    public SiteReplicationActorTests() : base(ClusterConfig, "site-repl")
    {
        _sfDbFile = Path.Combine(Path.GetTempPath(), $"site-repl-sf-{Guid.NewGuid():N}.db");

        // SiteStorageService takes an ILocalDb now; LocalDb has no in-memory mode, so the
        // site store gets its own temp-file database alongside the S&F one.
        _siteLocalDb = TestLocalDb.CreateTemp("site-repl-test");
        _storage = new SiteStorageService(
            _siteLocalDb.Db, NullLogger<SiteStorageService>.Instance);
        _storage.InitializeAsync().GetAwaiter().GetResult();

        _sfLocalDb = TestLocalDb.Create(_sfDbFile);
        _sfStorage = new StoreAndForwardStorage(
            _sfLocalDb.Db, NullLogger<StoreAndForwardStorage>.Instance);
        _sfStorage.InitializeAsync().GetAwaiter().GetResult();

        _replicationService = new ReplicationService(
            new StoreAndForwardOptions(), NullLogger<ReplicationService>.Instance);
    }

    void IDisposable.Dispose()
    {
        Shutdown();
        // The master connection anchors the WAL — dispose before deleting.
        var siteDbPath = _siteLocalDb.Path;
        _siteLocalDb.Dispose();
        _sfLocalDb.Dispose();
        TestLocalDb.DeleteFiles(siteDbPath);
        TestLocalDb.DeleteFiles(_sfDbFile);
    }

    private IActorRef CreateReplicationActor(IDeploymentConfigFetcher fetcher) =>
        ActorOf(Props.Create(() => new SiteReplicationActor(
            _storage, _sfStorage, _replicationService, SiteRole,
            NullLogger<SiteReplicationActor>.Instance, fetcher)));

    private IActorRef CreateReplicationActor(
        IDeploymentConfigFetcher fetcher, SiteRuntimeOptions options, TimeSpan retryDelay) =>
        ActorOf(Props.Create(() => new SiteReplicationActor(
            _storage, _sfStorage, _replicationService, SiteRole,
            NullLogger<SiteReplicationActor>.Instance, fetcher, null, options, retryDelay)));

    [Fact]
    public async Task ReplicatedFetch_RetriesUpToConfigFetchRetryCount()
    {
        // The first two fetches fail transiently; the third succeeds. With
        // ConfigFetchRetryCount = 3 the standby must retry to the third attempt and
        // then guarded-write the fetched config (a short retry delay keeps the test fast).
        var attempts = 0;
        var fetcher = new FakeConfigFetcher(_ =>
            Interlocked.Increment(ref attempts) < 3
                ? Task.FromException<string>(new InvalidOperationException("central hiccup"))
                : Task.FromResult("{\"instanceUniqueName\":\"RetryPump\"}"));
        var actor = CreateReplicationActor(
            fetcher, new SiteRuntimeOptions { ConfigFetchRetryCount = 3 },
            TimeSpan.FromMilliseconds(50));

        actor.Tell(new ApplyConfigDeploy(
            "RetryPump", "dep-r1", "sha256:r1", true,
            "http://central:9000", "tok-r1"));

        await AwaitAssertAsync(async () =>
        {
            Assert.Equal(3, Volatile.Read(ref attempts));
            var configs = await _storage.GetAllDeployedConfigsAsync();
            Assert.Single(configs, c => c.InstanceUniqueName == "RetryPump");
        }, TimeSpan.FromSeconds(10));
    }

    [Fact]
    public async Task ApplyConfigDeploy_StandbyFetchesConfigAndGuardedWrites()
    {
        // The standby receives an id-only ApplyConfigDeploy; it fetches the config from
        // central using the message's coords, then guarded-writes the fetched config.
        const string configJson = "{\"instanceUniqueName\":\"Pump1\"}";
        var fetcher = new FakeConfigFetcher(_ => Task.FromResult(configJson));
        var actor = CreateReplicationActor(fetcher);

        actor.Tell(new ApplyConfigDeploy(
            "Pump1", "dep-100", "sha256:abc", true,
            "http://central:9000", "tok-xyz"));

        // The continuation runs off-thread; await the guarded write landing.
        await AwaitAssertAsync(async () =>
        {
            var configs = await _storage.GetAllDeployedConfigsAsync();
            var row = Assert.Single(configs, c => c.InstanceUniqueName == "Pump1");
            Assert.Equal(configJson, row.ConfigJson);
            Assert.Equal("dep-100", row.DeploymentId);
            Assert.Equal("sha256:abc", row.RevisionHash);
            Assert.True(row.IsEnabled);
        }, TimeSpan.FromSeconds(5));

        // The fetcher was called with the message's coords.
        var call = Assert.Single(fetcher.Calls);
        Assert.Equal("http://central:9000", call.BaseUrl);
        Assert.Equal("dep-100", call.DeploymentId);
        Assert.Equal("tok-xyz", call.Token);
    }

    [Fact]
    public async Task ApplyConfigDeploy_Superseded404_SkipsWriteAndActorSurvives()
    {
        // A 404 (superseded/expired) surfaces as DeploymentConfigFetchException{IsSuperseded}.
        // The standby must skip the write, observe the exception (no crash), and stay alive.
        var fetcher = new FakeConfigFetcher(_ =>
            Task.FromException<string>(
                new DeploymentConfigFetchException("expired", isSuperseded: true)));
        var actor = CreateReplicationActor(fetcher);

        actor.Tell(new ApplyConfigDeploy(
            "GonePump", "dep-stale", "sha256:gone", true,
            "http://central:9000", "tok-stale"));

        // The fetch was attempted...
        await AwaitAssertAsync(() =>
        {
            Assert.Single(fetcher.Calls);
            return Task.CompletedTask;
        }, TimeSpan.FromSeconds(5));

        // ...the actor did not crash (no Terminated to its watcher within the window)...
        Watch(actor);
        ExpectNoMsg(TimeSpan.FromMilliseconds(500));

        // ...and nothing was written for the superseded instance.
        var configs = await _storage.GetAllDeployedConfigsAsync();
        Assert.DoesNotContain(configs, c => c.InstanceUniqueName == "GonePump");
    }

    [Fact]
    public async Task ApplyConfigDeploy_EmptyFetchCoords_SkipsFetchAndWrite()
    {
        // The direct DeployInstanceCommand cross-cluster wire path was retired in Task 14.
        // This tests the defensive guard: if empty coords arrive, the actor must skip quietly
        // — no FetchAsync("") call, no write — rather than erroring.
        var fetcher = new FakeConfigFetcher(_ => Task.FromResult("never"));
        var actor = CreateReplicationActor(fetcher);

        actor.Tell(new ApplyConfigDeploy(
            "NoCoordsPump", "dep-direct", "sha256:nc", true,
            CentralFetchBaseUrl: "", FetchToken: ""));

        // Give any (erroneous) async continuation time to run, then prove neither happened.
        Watch(actor);
        ExpectNoMsg(TimeSpan.FromMilliseconds(500));
        Assert.Empty(fetcher.Calls);
        var configs = await _storage.GetAllDeployedConfigsAsync();
        Assert.DoesNotContain(configs, c => c.InstanceUniqueName == "NoCoordsPump");
    }

    [Fact]
    public void ReplicateConfigDeploy_MapsToIdOnlyApplyConfigDeploy_ForPeer()
    {
        // The outbound mapping must forward an id-only ApplyConfigDeploy carrying the fetch
        // coords (and NO inline config) to the peer.
        var probe = CreateTestProbe();
        var fetcher = new FakeConfigFetcher(_ => Task.FromResult("unused"));
        var actor = ActorOf(Props.Create(() => new ProbeForwardingReplicationActor(
            _storage, _sfStorage, _replicationService, SiteRole,
            NullLogger<SiteReplicationActor>.Instance, fetcher, probe.Ref)));

        actor.Tell(new ReplicateConfigDeploy(
            "Pump2", "dep-200", "sha256:def", false,
            "http://central:9000", "tok-abc"));

        var applied = probe.ExpectMsg<ApplyConfigDeploy>(TimeSpan.FromSeconds(3));
        Assert.Equal("Pump2", applied.InstanceName);
        Assert.Equal("dep-200", applied.DeploymentId);
        Assert.Equal("sha256:def", applied.RevisionHash);
        Assert.False(applied.IsEnabled);
        Assert.Equal("http://central:9000", applied.CentralFetchBaseUrl);
        Assert.Equal("tok-abc", applied.FetchToken);
    }

    // ── Task 21: peer-join S&F buffer resync (anti-entropy) ──

    [Fact]
    public void StandbyTrackingPeer_SendsResyncRequest()
    {
        var probe = CreateTestProbe();
        var actor = ActorOf(Props.Create(() => new ResyncTestActor(
            _storage, _sfStorage, _replicationService, SiteRole,
            NullLogger<SiteReplicationActor>.Instance, probe.Ref, () => false)));

        actor.Tell(new TriggerPeerTracked()); // stands in for TryTrackPeer's MemberUp path

        probe.ExpectMsg<RequestSfBufferResync>(TimeSpan.FromSeconds(3));
    }

    [Fact]
    public void ActiveTrackingPeer_DoesNotRequestResync()
    {
        var probe = CreateTestProbe();
        var actor = ActorOf(Props.Create(() => new ResyncTestActor(
            _storage, _sfStorage, _replicationService, SiteRole,
            NullLogger<SiteReplicationActor>.Instance, probe.Ref, () => true)));

        actor.Tell(new TriggerPeerTracked());

        probe.ExpectNoMsg(TimeSpan.FromMilliseconds(300)); // active node never requests a resync
    }

    [Fact]
    public async Task ActiveNode_AnswersResyncRequest_WithChunkedSnapshot()
    {
        // Post-R2-T5 the active node answers with byte-budgeted SfBufferSnapshotChunk(s)
        // (a single small row rides one chunk) rather than the monolithic SfBufferSnapshot.
        await _sfStorage.EnqueueAsync(NewSfMessage("m1"));
        var probe = CreateTestProbe();
        var actor = ActorOf(Props.Create(() => new ResyncTestActor(
            _storage, _sfStorage, _replicationService, SiteRole,
            NullLogger<SiteReplicationActor>.Instance, probe.Ref, () => true)));

        actor.Tell(new RequestSfBufferResync(), TestActor);

        var chunk = ExpectMsg<SfBufferSnapshotChunk>(TimeSpan.FromSeconds(3));
        Assert.Equal(1, chunk.TotalChunks);
        Assert.Equal(1, chunk.Sequence);
        Assert.Single(chunk.Messages);
        Assert.False(chunk.Truncated);
    }

    [Fact]
    public async Task StandbyNode_AppliesSnapshot_ReplacingItsBuffer()
    {
        await _sfStorage.EnqueueAsync(NewSfMessage("stale"));
        var probe = CreateTestProbe();
        var actor = ActorOf(Props.Create(() => new ResyncTestActor(
            _storage, _sfStorage, _replicationService, SiteRole,
            NullLogger<SiteReplicationActor>.Instance, probe.Ref, () => false)));

        actor.Tell(new SfBufferSnapshot(new List<StoreAndForwardMessage> { NewSfMessage("fresh") }, false));

        await AwaitAssertAsync(async () =>
        {
            Assert.Null(await _sfStorage.GetMessageByIdAsync("stale"));
            Assert.NotNull(await _sfStorage.GetMessageByIdAsync("fresh"));
        }, TimeSpan.FromSeconds(5));
    }

    // ── R2 T5: chunked resync answer ──

    [Fact]
    public void ChunkForRemoting_SplitsByByteBudget_PreservingOrderAndSequence()
    {
        var rows = Enumerable.Range(0, 10)
            .Select(i => NewMessage($"m{i}", payloadJson: new string('x', 20_000)))
            .ToList();

        var chunks = SiteReplicationActor.ChunkForRemoting(rows, maxChunkBytes: 64_000, maxChunkRows: 200);

        Assert.True(chunks.Count > 1); // 10 × 20 KB cannot ride one 64 KB chunk
        Assert.Equal(rows.Select(r => r.Id), chunks.SelectMany(c => c).Select(r => r.Id)); // order preserved
        Assert.All(chunks, c => Assert.True(
            c.Sum(r => r.PayloadJson.Length) <= 64_000 || c.Count == 1)); // budget honored (oversized row isolated)
    }

    [Fact]
    public void ChunkForRemoting_RowCapHonored_AndSingleOversizedRowIsolated()
    {
        var many = Enumerable.Range(0, 500).Select(i => NewMessage($"s{i}", payloadJson: "{}")).ToList();
        Assert.All(SiteReplicationActor.ChunkForRemoting(many, 64_000, 200), c => Assert.True(c.Count <= 200));

        var oversized = new List<StoreAndForwardMessage>
            { NewMessage("big", payloadJson: new string('y', 100_000)), NewMessage("small", payloadJson: "{}") };
        var chunks = SiteReplicationActor.ChunkForRemoting(oversized, 64_000, 200);
        Assert.Equal(2, chunks.Count); // the oversized row rides alone
    }

    [Fact]
    public async Task ActiveNode_AnswersResyncRequest_WithSequencedChunks_SharingOneResyncId()
    {
        for (var i = 0; i < 3; i++)
            await _sfStorage.EnqueueAsync(NewMessage($"c{i}", payloadJson: new string('z', 30_000)));
        var actor = CreateResyncActor(isActive: () => true);

        actor.Tell(new RequestSfBufferResync(), TestActor);

        var first = ExpectMsg<SfBufferSnapshotChunk>(TimeSpan.FromSeconds(5));
        var rest = Enumerable.Range(1, first.TotalChunks - 1)
            .Select(_ => ExpectMsg<SfBufferSnapshotChunk>(TimeSpan.FromSeconds(5)))
            .Prepend(first)
            .ToList();

        Assert.True(first.TotalChunks > 1);
        Assert.All(rest, c => Assert.Equal(first.ResyncId, c.ResyncId));
        Assert.Equal(Enumerable.Range(1, first.TotalChunks), rest.Select(c => c.Sequence));
        Assert.Equal(3, rest.Sum(c => c.Messages.Count));
    }

    // ── R2 T6: standby chunk assembly + atomic apply + ack ──

    [Fact]
    public async Task StandbyNode_AssemblesChunks_AppliesOnce_AndAcks()
    {
        await _sfStorage.EnqueueAsync(NewMessage("stale"));
        var actor = CreateResyncActor(isActive: () => false);
        var resyncId = "r1";

        actor.Tell(new SfBufferSnapshotChunk(resyncId, 1, 2,
            new List<StoreAndForwardMessage> { NewMessage("f1") }, false), TestActor);
        actor.Tell(new SfBufferSnapshotChunk(resyncId, 2, 2,
            new List<StoreAndForwardMessage> { NewMessage("f2") }, false), TestActor);

        var ack = ExpectMsg<SfBufferResyncAck>(TimeSpan.FromSeconds(5));
        Assert.Equal(resyncId, ack.ResyncId);
        Assert.Equal(2, ack.RowCount);
        await AwaitAssertAsync(async () =>
        {
            Assert.Null(await _sfStorage.GetMessageByIdAsync("stale"));   // replaced wholesale
            Assert.NotNull(await _sfStorage.GetMessageByIdAsync("f1"));
            Assert.NotNull(await _sfStorage.GetMessageByIdAsync("f2"));
        });
    }

    [Fact]
    public async Task StandbyNode_NewResyncId_DiscardsStalePartialAssembly()
    {
        var actor = CreateResyncActor(isActive: () => false);
        actor.Tell(new SfBufferSnapshotChunk("old", 1, 2,
            new List<StoreAndForwardMessage> { NewMessage("orphan") }, false), TestActor);
        actor.Tell(new SfBufferSnapshotChunk("new", 1, 1,
            new List<StoreAndForwardMessage> { NewMessage("fresh") }, false), TestActor);

        ExpectMsg<SfBufferResyncAck>(TimeSpan.FromSeconds(5)); // "new" completed
        await AwaitAssertAsync(async () =>
        {
            Assert.NotNull(await _sfStorage.GetMessageByIdAsync("fresh"));
            Assert.Null(await _sfStorage.GetMessageByIdAsync("orphan")); // stale partial never applied
        });
    }

    [Fact]
    public void ActiveNode_IgnoresChunks_NeverAcks()
    {
        var actor = CreateResyncActor(isActive: () => true);
        actor.Tell(new SfBufferSnapshotChunk("r", 1, 1,
            new List<StoreAndForwardMessage> { NewMessage("x") }, false), TestActor);
        ExpectNoMsg(TimeSpan.FromMilliseconds(300));
    }

    // ── R2 T7: active-side resync ack confirmation + telemetry ──
    //
    // NOTE (deviation from plan): the actor logs via Microsoft ILogger (NullLogger in
    // tests), NOT Akka's EventStream, so the plan's EventFilter.Warning assertions can
    // never observe these warnings. We observe the two OTel counters via a MeterListener
    // instead — the equivalent, and stronger, observable signal.

    [Fact]
    public async Task ActiveNode_ReceivingAck_CountsResyncCompleted()
    {
        long completed = 0;
        using var listener = ListenCounter("scadabridge.store_and_forward.resync.completed",
            m => Interlocked.Add(ref completed, m));

        await _sfStorage.EnqueueAsync(NewMessage("m1"));
        var actor = CreateResyncActor(isActive: () => true, ackTimeout: TimeSpan.FromSeconds(30));
        actor.Tell(new RequestSfBufferResync(), TestActor);
        var chunk = ExpectMsg<SfBufferSnapshotChunk>(TimeSpan.FromSeconds(5));

        actor.Tell(new SfBufferResyncAck(chunk.ResyncId, 1), TestActor);

        await AwaitAssertAsync(() =>
        {
            Assert.True(Interlocked.Read(ref completed) >= 1); // ack recorded the completion
            return Task.CompletedTask;
        }, TimeSpan.FromSeconds(5));
    }

    [Fact]
    public async Task ActiveNode_MissingAck_WarnsAfterAckTimeout()
    {
        long ackMissing = 0;
        using var listener = ListenCounter("scadabridge.store_and_forward.resync.ack_missing",
            m => Interlocked.Add(ref ackMissing, m));

        await _sfStorage.EnqueueAsync(NewMessage("m1"));
        var actor = CreateResyncActor(isActive: () => true, ackTimeout: TimeSpan.FromMilliseconds(200));
        actor.Tell(new RequestSfBufferResync(), TestActor);
        ExpectMsg<SfBufferSnapshotChunk>(TimeSpan.FromSeconds(5));
        // No ack is sent → the ack window expires and the resync is counted unacknowledged.

        await AwaitAssertAsync(() =>
        {
            Assert.True(Interlocked.Read(ref ackMissing) >= 1);
            return Task.CompletedTask;
        }, TimeSpan.FromSeconds(5));
    }

    /// <summary>Attaches a <see cref="MeterListener"/> to a single ScadaBridge counter by name,
    /// forwarding each recorded increment to <paramref name="onMeasurement"/>.</summary>
    private static MeterListener ListenCounter(string instrumentName, Action<long> onMeasurement)
    {
        var listener = new MeterListener();
        listener.InstrumentPublished = (inst, l) =>
        {
            if (inst.Meter.Name == ScadaBridgeTelemetry.MeterName && inst.Name == instrumentName)
                l.EnableMeasurementEvents(inst);
        };
        listener.SetMeasurementEventCallback<long>((_, m, _, _) => onMeasurement(m));
        listener.Start();
        return listener;
    }

    private static StoreAndForwardMessage NewSfMessage(string id) => new()
    {
        Id = id,
        Category = StoreAndForwardCategory.ExternalSystem,
        Target = "t",
        PayloadJson = "{}",
        RetryCount = 0,
        MaxRetries = 50,
        RetryIntervalMs = 30000,
        CreatedAt = DateTimeOffset.UtcNow,
        Status = StoreAndForwardMessageStatus.Pending,
    };

    /// <summary>
    /// Builds a resync-test message with a settable payload (additive to
    /// <see cref="NewSfMessage"/> — the chunker sizes on <c>PayloadJson</c> length).
    /// </summary>
    private static StoreAndForwardMessage NewMessage(string id, string payloadJson = "{}") => new()
    {
        Id = id,
        Category = StoreAndForwardCategory.ExternalSystem,
        Target = "t",
        PayloadJson = payloadJson,
        RetryCount = 0,
        MaxRetries = 50,
        RetryIntervalMs = 30000,
        CreatedAt = DateTimeOffset.UtcNow,
        Status = StoreAndForwardMessageStatus.Pending,
    };

    /// <summary>Constructs a <see cref="ResyncTestActor"/> with the given active-node check
    /// (the resync chunk/ack tests Tell to and expect from <see cref="TestKit.TestActor"/>).
    /// <paramref name="ackTimeout"/> is the active-side ack window seam (T7).</summary>
    private IActorRef CreateResyncActor(Func<bool> isActive, TimeSpan? ackTimeout = null) =>
        ActorOf(Props.Create(() => new ResyncTestActor(
            _storage, _sfStorage, _replicationService, SiteRole,
            NullLogger<SiteReplicationActor>.Instance, CreateTestProbe().Ref, isActive, ackTimeout)));

    /// <summary>Test message: drives <see cref="SiteReplicationActor.OnPeerTracked"/> directly,
    /// standing in for the MemberUp→TryTrackPeer path (a single-node TestKit cannot form a real peer).</summary>
    private sealed record TriggerPeerTracked;

    /// <summary>
    /// Test subclass for the resync tests: captures peer sends to a probe, injects the
    /// active-node check, and exposes <see cref="OnPeerTracked"/> via a test message.
    /// </summary>
    private sealed class ResyncTestActor : SiteReplicationActor
    {
        private readonly IActorRef _peerProbe;

        public ResyncTestActor(
            SiteStorageService storage, StoreAndForwardStorage sfStorage,
            ReplicationService replicationService, string siteRole,
            ILogger<SiteReplicationActor> logger, IActorRef peerProbe, Func<bool> isActive,
            TimeSpan? ackTimeout = null)
            : base(storage, sfStorage, replicationService, siteRole, logger,
                   configFetcher: null, isActiveOverride: isActive, resyncAckTimeout: ackTimeout)
        {
            _peerProbe = peerProbe;
            Receive<TriggerPeerTracked>(_ => OnPeerTracked());
        }

        protected override void SendToPeer(object message) => _peerProbe.Tell(message, Self);
    }

    /// <summary>
    /// Test subclass exposing the peer send: <see cref="SiteReplicationActor.SendToPeer"/> is
    /// overridden to forward to a probe so the outbound mapping can be asserted without a real
    /// two-node cluster (a single-node TestKit has no peer address, so the real send is dropped).
    /// </summary>
    private sealed class ProbeForwardingReplicationActor : SiteReplicationActor
    {
        private readonly IActorRef _peerProbe;

        public ProbeForwardingReplicationActor(
            SiteStorageService storage, StoreAndForwardStorage sfStorage,
            ReplicationService replicationService, string siteRole,
            ILogger<SiteReplicationActor> logger, IDeploymentConfigFetcher configFetcher,
            IActorRef peerProbe)
            : base(storage, sfStorage, replicationService, siteRole, logger, configFetcher)
            => _peerProbe = peerProbe;

        protected override void SendToPeer(object message) => _peerProbe.Tell(message, Self);
    }

    /// <summary>
    /// In-test fake <see cref="IDeploymentConfigFetcher"/>: runs a per-deploymentId behavior
    /// (return config JSON or throw, as a Task — mirroring the real async HTTP fetcher) and
    /// records every call's coords thread-safely (the continuation runs on a pool thread).
    /// </summary>
    private sealed class FakeConfigFetcher : IDeploymentConfigFetcher
    {
        private readonly Func<string, Task<string>> _behavior;
        public ConcurrentQueue<(string BaseUrl, string DeploymentId, string Token)> Calls { get; } = new();

        public FakeConfigFetcher(Func<string, Task<string>> behavior) => _behavior = behavior;

        public async Task<string> FetchAsync(
            string centralFetchBaseUrl, string deploymentId, string token, CancellationToken ct)
        {
            Calls.Enqueue((centralFetchBaseUrl, deploymentId, token));
            await Task.Yield();
            return await _behavior(deploymentId);
        }
    }
}