ScadaBridge/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/ReconcileServiceTests.cs

using Microsoft.Extensions.Logging.Abstractions;
using Microsoft.Extensions.Options;
using NSubstitute;
using ZB.MOM.WW.ScadaBridge.Commons.Entities.Deployment;
using ZB.MOM.WW.ScadaBridge.Commons.Entities.Sites;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Deployment;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Deployment;

namespace ZB.MOM.WW.ScadaBridge.Communication.Tests;

/// <summary>
/// Unit tests for the central-side startup-reconciliation handler
/// (<see cref="ReconcileService"/>). A site node reports its local inventory; the
/// service diffs it against central's expected deployed set, stages fresh fetch
/// tokens for the gap (missing/stale), and reports orphans.
/// </summary>
public class ReconcileServiceTests
{
    private const string BaseUrl = "https://central.example:9000";
    private const string SiteIdentifier = "site1";
    private const int SiteId = 7;

    private readonly IDeploymentManagerRepository _deploymentRepo =
        Substitute.For<IDeploymentManagerRepository>();
    private readonly ISiteRepository _siteRepo = Substitute.For<ISiteRepository>();

    // Fake repository state behind the two batched reconcile calls.
    private readonly Dictionary<int, DeployedConfigSnapshot> _snapshots = new();
    private readonly Dictionary<int, PendingDeployment> _existingLive = new();
    private readonly HashSet<int> _racedAway = new();

    /// <summary>Wires the batched snapshot read and batched stage to the fake state above.</summary>
    public ReconcileServiceTests()
    {
        _deploymentRepo.GetDeployedSnapshotsByInstanceIdsAsync(
                Arg.Any<IReadOnlyCollection<int>>(), Arg.Any<CancellationToken>())
            .Returns(ci => (IReadOnlyDictionary<int, DeployedConfigSnapshot>)ci.ArgAt<IReadOnlyCollection<int>>(0)
                .Where(_snapshots.ContainsKey)
                .ToDictionary(id => id, id => _snapshots[id]));

        _deploymentRepo.StagePendingBatchIfAbsentAsync(
                Arg.Any<IReadOnlyList<PendingDeployment>>(), Arg.Any<DateTimeOffset>(), Arg.Any<CancellationToken>())
            .Returns(ci =>
            {
                var live = new Dictionary<int, PendingDeployment>();
                foreach (var candidate in ci.ArgAt<IReadOnlyList<PendingDeployment>>(0))
                {
                    if (_existingLive.TryGetValue(candidate.InstanceId, out var existing))
                        live[candidate.InstanceId] = existing;
                    else if (!_racedAway.Contains(candidate.InstanceId))
                        live[candidate.InstanceId] = candidate;
                }
                return (IReadOnlyDictionary<int, PendingDeployment>)live;
            });
    }

    private ReconcileService CreateService(TimeSpan? ttl = null, TimeSpan? queryTimeout = null)
    {
        var options = Options.Create(new CommunicationOptions
        {
            CentralFetchBaseUrl = BaseUrl,
            PendingDeploymentTtl = ttl ?? TimeSpan.FromMinutes(5),
            QueryTimeout = queryTimeout ?? TimeSpan.FromSeconds(30),
        });
        return new ReconcileService(
            _deploymentRepo, _siteRepo, options, NullLogger<ReconcileService>.Instance);
    }

    private void SiteResolves() =>
        _siteRepo.GetSiteByIdentifierAsync(SiteIdentifier, Arg.Any<CancellationToken>())
            .Returns(new Site("Site One", SiteIdentifier) { Id = SiteId });

    private static ExpectedDeployment Expected(int id, string name, string rev, string dep, bool enabled = true) =>
        new(id, name, rev, dep, enabled);

    private void ExpectedSet(params ExpectedDeployment[] expected) =>
        _deploymentRepo.GetExpectedDeploymentsForSiteAsync(SiteId, Arg.Any<CancellationToken>())
            .Returns(expected.ToList());

    private void SnapshotFor(ExpectedDeployment exp) =>
        _snapshots[exp.InstanceId] = new DeployedConfigSnapshot(
            exp.DeploymentId, exp.RevisionHash, $"{{\"cfg\":\"{exp.InstanceUniqueName}\"}}")
        {
            InstanceId = exp.InstanceId
        };

    private static ReconcileSiteRequest Request(params (string Name, string Rev)[] local) =>
        new(SiteIdentifier, "node-a",
            local.ToDictionary(x => x.Name, x => x.Rev));

    /// <summary>Verifies that reconciliation reports missing and stale instances as gaps with fresh tokens and snapshot deployment ids, and omits current ones.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task Reconcile_GapIsMissingAndStale_CurrentOmitted_WithFreshTokensAndSnapshotDeploymentIds()
    {
        SiteResolves();
        var a = Expected(1, "inst-A", "rev1", "dep-A");
        var b = Expected(2, "inst-B", "rev2", "dep-B");
        var c = Expected(3, "inst-C", "rev3", "dep-C", enabled: false);
        ExpectedSet(a, b, c);
        SnapshotFor(b);
        SnapshotFor(c);

        // Node has A current, B stale (revOLD), C missing entirely.
        var response = await CreateService().ReconcileAsync(
            Request(("inst-A", "rev1"), ("inst-B", "revOLD")));

        Assert.Equal(BaseUrl, response.CentralFetchBaseUrl);
        Assert.Equal(2, response.Gap.Count);
        Assert.DoesNotContain(response.Gap, g => g.InstanceUniqueName == "inst-A");

        var gapB = Assert.Single(response.Gap, g => g.InstanceUniqueName == "inst-B");
        Assert.Equal("dep-B", gapB.DeploymentId);
        Assert.Equal("rev2", gapB.RevisionHash);
        Assert.True(gapB.IsEnabled);
        Assert.False(string.IsNullOrWhiteSpace(gapB.FetchToken));

        var gapC = Assert.Single(response.Gap, g => g.InstanceUniqueName == "inst-C");
        Assert.Equal("dep-C", gapC.DeploymentId);
        Assert.Equal("rev3", gapC.RevisionHash);
        Assert.False(gapC.IsEnabled);
        Assert.False(string.IsNullOrWhiteSpace(gapC.FetchToken));

        // Fresh, distinct tokens per gap item.
        Assert.NotEqual(gapB.FetchToken, gapC.FetchToken);
        Assert.Empty(response.OrphanNames);
    }

    /// <summary>Verifies that a local instance name that is not expected is reported as an orphan.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task Reconcile_LocalNameNotInExpected_IsReportedAsOrphan()
    {
        SiteResolves();
        var a = Expected(1, "inst-A", "rev1", "dep-A");
        ExpectedSet(a);

        // inst-A is current; inst-Z is not deployed centrally → orphan.
        var response = await CreateService().ReconcileAsync(
            Request(("inst-A", "rev1"), ("inst-Z", "revX")));

        Assert.Empty(response.Gap);
        var orphan = Assert.Single(response.OrphanNames);
        Assert.Equal("inst-Z", orphan);
    }

    /// <summary>Verifies that when a live pending row already exists for a gap instance, the gap carries the existing token.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task Reconcile_LiveRowAlreadyExists_ReturnsExistingTokenInGap()
    {
        // Live-found bug: two site nodes concurrently missing the SAME instance both reconcile at
        // startup. The first stages a pending row; the second's StagePendingIfAbsent returns false.
        // The second node must STILL heal — the handler reads the existing pending row and returns
        // ITS deploymentId/revHash/token (the fetch token is multi-use within its TTL).
        SiteResolves();
        var b = Expected(2, "inst-B", "rev2", "dep-B");
        var c = Expected(3, "inst-C", "rev3", "dep-C");
        ExpectedSet(b, c);
        SnapshotFor(b);
        SnapshotFor(c);

        // B stages fresh (true). C already has a pending row (a concurrent reconcile from the
        // other node, or an in-flight deploy, staged it first) → stage returns false.
        // The existing pending row for inst-C carries the FIRST node's deploymentId + token.
        var nowUtc = DateTimeOffset.UtcNow;
        _existingLive[3] = new PendingDeployment(
            "dep-C-existing", 3, "rev3-existing", string.Empty,
            "tok-C-existing", nowUtc, nowUtc.AddMinutes(5));

        var response = await CreateService().ReconcileAsync(Request(/* empty local inventory */));

        Assert.Equal(2, response.Gap.Count);

        var gapB = Assert.Single(response.Gap, g => g.InstanceUniqueName == "inst-B");
        Assert.Equal("dep-B", gapB.DeploymentId);

        // inst-C IS included, carrying the EXISTING row's deploymentId/revHash/token — NOT the
        // snapshot's (snapshot.DeploymentId would be "dep-C").
        var gapC = Assert.Single(response.Gap, g => g.InstanceUniqueName == "inst-C");
        Assert.Equal("dep-C-existing", gapC.DeploymentId);
        Assert.Equal("rev3-existing", gapC.RevisionHash);
        Assert.Equal("tok-C-existing", gapC.FetchToken);
        Assert.True(gapC.IsEnabled);
    }

    /// <summary>Verifies that when no pending row can be staged or found, that gap item is omitted.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task Reconcile_NoLiveRowStagedOrFound_OmitsThatGapItem()
    {
        // Fallback path: stage returns false, but the pending row raced away (was purged between
        // the stage attempt and the read). Omit the item — the node retries on the next reconcile.
        SiteResolves();
        var b = Expected(2, "inst-B", "rev2", "dep-B");
        var c = Expected(3, "inst-C", "rev3", "dep-C");
        ExpectedSet(b, c);
        SnapshotFor(b);
        SnapshotFor(c);

        // No row could be staged or found for inst-C (raced away).
        _racedAway.Add(3);

        var response = await CreateService().ReconcileAsync(Request(/* empty local inventory */));

        var gap = Assert.Single(response.Gap);
        Assert.Equal("inst-B", gap.InstanceUniqueName);
        Assert.DoesNotContain(response.Gap, g => g.InstanceUniqueName == "inst-C");
    }

    /// <summary>Verifies that a gap item whose snapshot is missing is skipped.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task Reconcile_SnapshotMissing_SkipsGapItem()
    {
        SiteResolves();
        var b = Expected(2, "inst-B", "rev2", "dep-B");
        ExpectedSet(b);
        // No snapshot configured for inst-B → repo returns null (deleted race).

        var response = await CreateService().ReconcileAsync(Request());

        Assert.Empty(response.Gap);
        Assert.Empty(response.OrphanNames);
    }

    /// <summary>Verifies that reconciling an unknown site returns an empty response without throwing.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task Reconcile_UnknownSite_ReturnsEmptyResponse_NoThrow()
    {
        _siteRepo.GetSiteByIdentifierAsync(SiteIdentifier, Arg.Any<CancellationToken>())
            .Returns((Site?)null);

        var response = await CreateService().ReconcileAsync(
            Request(("inst-A", "rev1")));

        Assert.Empty(response.Gap);
        Assert.Empty(response.OrphanNames);
        Assert.Equal(BaseUrl, response.CentralFetchBaseUrl);
        await _deploymentRepo.DidNotReceive()
            .GetExpectedDeploymentsForSiteAsync(Arg.Any<int>(), Arg.Any<CancellationToken>());
    }

    // ── review 2026-10-21 03/P-1: batched reconcile + bounded work ──

    /// <summary>
    /// The reconcile pass is a fixed number of round trips: snapshot JSON is read ONCE, only
    /// for the gap, and the pending rows are staged in ONE batch — never per gap instance.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task Reconcile_ReadsGapSnapshotsOnce_AndStagesInOneBatch()
    {
        SiteResolves();
        var a = Expected(1, "inst-A", "rev1", "dep-A");
        var b = Expected(2, "inst-B", "rev2", "dep-B");
        var c = Expected(3, "inst-C", "rev3", "dep-C");
        ExpectedSet(a, b, c);
        SnapshotFor(a);
        SnapshotFor(b);
        SnapshotFor(c);

        var response = await CreateService().ReconcileAsync(Request(("inst-A", "rev1")));

        Assert.Equal(2, response.Gap.Count);
        await _deploymentRepo.Received(1).GetDeployedSnapshotsByInstanceIdsAsync(
            Arg.Is<IReadOnlyCollection<int>>(ids => ids.Order().SequenceEqual(new[] { 2, 3 })),
            Arg.Any<CancellationToken>());
        await _deploymentRepo.Received(1).StagePendingBatchIfAbsentAsync(
            Arg.Is<IReadOnlyList<PendingDeployment>>(rows => rows.Count == 2),
            Arg.Any<DateTimeOffset>(), Arg.Any<CancellationToken>());
        await _deploymentRepo.DidNotReceive().GetDeployedSnapshotByInstanceIdAsync(
            Arg.Any<int>(), Arg.Any<CancellationToken>());
        await _deploymentRepo.DidNotReceive().StagePendingIfAbsentAsync(
            Arg.Any<int>(), Arg.Any<string>(), Arg.Any<string>(), Arg.Any<string>(),
            Arg.Any<string>(), Arg.Any<DateTimeOffset>(), Arg.Any<DateTimeOffset>(),
            Arg.Any<CancellationToken>());
    }

    /// <summary>A node with nothing missing costs no snapshot read and no staging at all.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task Reconcile_NoGap_DoesNoSnapshotReadOrStaging()
    {
        SiteResolves();
        ExpectedSet(Expected(1, "inst-A", "rev1", "dep-A"));

        var response = await CreateService().ReconcileAsync(Request(("inst-A", "rev1")));

        Assert.Empty(response.Gap);
        await _deploymentRepo.DidNotReceive().GetDeployedSnapshotsByInstanceIdsAsync(
            Arg.Any<IReadOnlyCollection<int>>(), Arg.Any<CancellationToken>());
        await _deploymentRepo.DidNotReceive().StagePendingBatchIfAbsentAsync(
            Arg.Any<IReadOnlyList<PendingDeployment>>(), Arg.Any<DateTimeOffset>(), Arg.Any<CancellationToken>());
    }

    /// <summary>
    /// Central stops working once the site's own deadline (QueryTimeout) has passed instead of
    /// staging tokens for a caller that has already given up.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task Reconcile_WorkIsBoundedByQueryTimeout()
    {
        SiteResolves();
        _deploymentRepo.GetExpectedDeploymentsForSiteAsync(SiteId, Arg.Any<CancellationToken>())
            .Returns(async ci =>
            {
                await Task.Delay(TimeSpan.FromSeconds(30), ci.ArgAt<CancellationToken>(1));
                return (IReadOnlyList<ExpectedDeployment>)new List<ExpectedDeployment>();
            });

        var service = CreateService(queryTimeout: TimeSpan.FromMilliseconds(100));

        await Assert.ThrowsAnyAsync<OperationCanceledException>(
            () => service.ReconcileAsync(Request()));
    }
}