lmxopcua/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/TwoNodeClusterHarness.cs

using System.Net.Sockets;
using Akka.Actor;
using Akka.Cluster;
using Akka.Hosting;
using Microsoft.AspNetCore.Builder;
using Microsoft.AspNetCore.Hosting;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using ZB.MOM.WW.OtOpcUa.AdminUI;
using ZB.MOM.WW.OtOpcUa.AdminUI.Api;
using ZB.MOM.WW.OtOpcUa.AdminUI.Clients;
using ZB.MOM.WW.OtOpcUa.AdminUI.Hubs;
using ZB.MOM.WW.OtOpcUa.Cluster;
using ZB.MOM.WW.OtOpcUa.Configuration;
using ZB.MOM.WW.OtOpcUa.ControlPlane;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Host.Health;
using ZB.MOM.WW.OtOpcUa.Runtime;
using ZB.MOM.WW.OtOpcUa.Security;
using ZB.MOM.WW.OtOpcUa.Security.Endpoints;
using ZB.MOM.WW.OtOpcUa.Security.Ldap;

namespace ZB.MOM.WW.OtOpcUa.Host.IntegrationTests;

/// <summary>
/// Spins up two in-process <c>OtOpcUa.Host</c>-equivalent <see cref="WebApplication"/> instances
/// that share an <see cref="OtOpcUaConfigDbContext"/> and form a 2-member Akka cluster. Both
/// nodes carry the <c>admin</c> + <c>driver</c> roles, matching design §8's failover-test 2-node
/// profile.
///
/// <para>Default mode uses EF InMemoryDatabase + <see cref="StubLdapAuthService"/>. Optional
/// real-infra modes (env-var driven, see <see cref="HarnessMode.FromEnvironment"/>):</para>
/// <list type="bullet">
///   <item><c>OTOPCUA_HARNESS_USE_SQL=1</c> → swap the in-memory DB for SQL Server on
///         <c>localhost:14331</c> (see <c>docker-compose.yml</c>). Each harness gets a unique
///         database name (<c>OtOpcUa_Harness_{guid}</c>) created via <c>EnsureCreated</c>
///         and dropped via <c>EnsureDeleted</c> on dispose.</item>
///   <item><c>OTOPCUA_HARNESS_USE_LDAP=1</c> → drop the stub and point <c>LdapAuthService</c>
///         at the GLAuth container on <c>localhost:3894</c> (see Docker/glauth/config.toml).</item>
/// </list>
///
/// <para>Why not <c>WebApplicationFactory&lt;Program&gt;</c>? Program.cs reads <c>OTOPCUA_ROLES</c>
/// from process env (shared across in-process WAF instances) and writes both Serilog file sinks
/// + Akka cluster TCP listener to the host process — neither survives two parallel WAFs cleanly.
/// This harness instead replays the Program.cs DI graph from a clean
/// <see cref="WebApplicationBuilder"/> per node with per-node config overrides.</para>
/// </summary>
public sealed class TwoNodeClusterHarness : IAsyncDisposable
{
    public const string TestRoles = "admin,driver";
    /// <summary>The deploy-API key both harness nodes are configured with (Security:DeployApiKey).</summary>
    public const string HarnessDeployApiKey = "test-deploy-key";
    /// <summary>Gets the shared database name for both cluster nodes.</summary>
    public string SharedDbName { get; } = $"two-node-cluster-{Guid.NewGuid():N}";

    /// <summary>Resolves node A's bound HTTP base address (Kestrel binds an ephemeral port), for
    /// driving the REST surface over real HTTP in tests.</summary>
    public string NodeABaseAddress => ResolveBaseAddress(NodeA);

    private static string ResolveBaseAddress(WebApplication app)
    {
        var feature = app.Services.GetRequiredService<Microsoft.AspNetCore.Hosting.Server.IServer>()
            .Features.Get<Microsoft.AspNetCore.Hosting.Server.Features.IServerAddressesFeature>();
        return feature?.Addresses.FirstOrDefault()
            ?? throw new InvalidOperationException("Node A has no bound HTTP address.");
    }

    /// <summary>Gets the harness mode configuration from environment variables.</summary>
    public HarnessMode Mode { get; } = HarnessMode.FromEnvironment();
    private string? _sqlDbName;
    private string? _sqlConnString;

    /// <summary>Optional opt-in <see cref="IDriverFactory"/> both nodes register so deployed drivers can
    /// reach <c>Connected</c>/Healthy in-process. Null (default) leaves the production
    /// <see cref="NullDriverFactory"/> from <c>AddOtOpcUaRuntime()</c> in place — unchanged behaviour for
    /// existing tests.</summary>
    private IDriverFactory? _driverFactory;

    /// <summary>Gets the first web application node.</summary>
    public WebApplication NodeA { get; private set; } = null!;
    /// <summary>Gets the second web application node.</summary>
    public WebApplication NodeB { get; private set; } = null!;

    /// <summary>Gets the Akka port allocated for node A.</summary>
    public int NodeAAkkaPort { get; private set; }
    /// <summary>Gets the Akka port allocated for node B.</summary>
    public int NodeBAkkaPort { get; private set; }

    // Both nodes bind to 127.0.0.1 — ClusterRoleInfo + ConfigPublishCoordinator encode
    // host:port into NodeId so the cluster membership stays distinct on different ports.
    public const string LoopbackHost = "127.0.0.1";

    /// <summary>Gets the Akka ActorSystem for node A.</summary>
    public ActorSystem NodeASystem => NodeA.Services.GetRequiredService<ActorSystem>();
    /// <summary>Gets the Akka ActorSystem for node B.</summary>
    public ActorSystem NodeBSystem => NodeB.Services.GetRequiredService<ActorSystem>();

    /// <summary>
    /// The <c>host:port</c> identity each node's <c>DriverHostActor</c> / <c>NodeDeploymentState</c>
    /// row uses, derived the same way <c>ClusterRoleInfo</c> derives <c>LocalNode</c> from
    /// <c>Cluster:PublicHostname</c> + <c>Cluster:Port</c>. Seed a <c>ClusterNode.NodeId</c> with this
    /// value to bind a node to a logical ClusterId for multi-cluster scoping tests.
    /// </summary>
    public string NodeANodeId => $"{LoopbackHost}:{NodeAAkkaPort}";

    /// <inheritdoc cref="NodeANodeId"/>
    public string NodeBNodeId => $"{LoopbackHost}:{NodeBAkkaPort}";

    /// <summary>Opens a new <see cref="OtOpcUaConfigDbContext"/> over the shared ConfigDb (the same
    /// store both nodes read) so a test can seed clusters/nodes/drivers before triggering a deploy.</summary>
    /// <returns>A new DbContext the caller is responsible for disposing.</returns>
    public Task<OtOpcUaConfigDbContext> CreateConfigDbContextAsync()
        => NodeA.Services.GetRequiredService<IDbContextFactory<OtOpcUaConfigDbContext>>().CreateDbContextAsync();

    /// <summary>Boots both nodes and waits up to <paramref name="formationTimeout"/> for cluster convergence.</summary>
    /// <param name="formationTimeout">Maximum time to wait for cluster formation; defaults to 20 seconds if not provided.</param>
    /// <param name="driverFactory">Optional opt-in <see cref="IDriverFactory"/> both nodes register so deployed
    /// drivers reach <c>Connected</c>/Healthy in-process. Null (default) leaves the production
    /// <see cref="NullDriverFactory"/> default in place — unchanged behaviour for existing tests.</param>
    public static async Task<TwoNodeClusterHarness> StartAsync(
        TimeSpan? formationTimeout = null,
        IDriverFactory? driverFactory = null)
    {
        var harness = new TwoNodeClusterHarness();
        harness._driverFactory = driverFactory;
        harness.NodeAAkkaPort = AllocateFreePort();
        harness.NodeBAkkaPort = AllocateFreePort();

        if (harness.Mode.UseSqlServer)
        {
            harness._sqlDbName = $"OtOpcUa_Harness_{Guid.NewGuid():N}";
            harness._sqlConnString = $"Server=localhost,14331;Database={harness._sqlDbName};User Id=sa;Password=OtOpcUa!Harness123;TrustServerCertificate=True;";
            await EnsureSqlSchemaCreatedAsync(harness._sqlConnString);
        }

        // Node A boots first as the seed.
        harness.NodeA = await BuildNodeAsync(harness, NodeRole.Seed);
        harness.NodeB = await BuildNodeAsync(harness, NodeRole.Joiner);

        await WaitForClusterFormationAsync(
            harness.NodeASystem,
            harness.NodeBSystem,
            formationTimeout ?? TimeSpan.FromSeconds(20));

        return harness;
    }

    /// <summary>
    /// Gracefully shuts down node B via <see cref="WebApplication.DisposeAsync"/>, which runs
    /// CoordinatedShutdown → Cluster.Leave. Node A sees the member transition to Removed within
    /// a couple of seconds. Use this for failover scenarios; call <see cref="RestartNodeBAsync"/>
    /// to bring it back on the same Akka port.
    /// </summary>
    public async Task StopNodeBAsync()
    {
        if (NodeB is null) return;
        await NodeB.DisposeAsync();
        NodeB = null!;
    }

    /// <summary>
    /// Rebuilds node B on the same Akka port + same ConfigDb and waits for the cluster
    /// to re-converge to 2 Up members. Use after <see cref="StopNodeBAsync"/> to test rejoin.
    /// </summary>
    /// <param name="formationTimeout">The maximum time to wait for cluster formation; defaults to 20 seconds.</param>
    public async Task RestartNodeBAsync(TimeSpan? formationTimeout = null)
    {
        NodeB = await BuildNodeAsync(this, NodeRole.Joiner);

        await WaitForClusterFormationAsync(
            NodeASystem,
            NodeBSystem,
            formationTimeout ?? TimeSpan.FromSeconds(20));
    }

    /// <summary>
    /// Abruptly crashes node A (the seed / oldest member) by shutting down its remoting transport — the
    /// canonical in-process crash simulation. All associations drop and heartbeats/gossip stop instantly
    /// with NO graceful <c>Cluster.Leave</c> (the transport is dead, so A cannot gossip anything). Node B's
    /// failure detector therefore marks A <b>Unreachable</b> (not Removed), which is the only state the
    /// split-brain resolver acts on: with an active resolver (keep-oldest + down-if-alone, active here via
    /// both the akka.conf block and Akka.Cluster.Hosting's default downing provider) B downs the alone
    /// oldest node and takes over; with downing disabled (explicit NoDowning) B keeps A Unreachable forever
    /// and never fails over. This is the crash the graceful <see cref="StopNodeBAsync"/> path (a real
    /// <c>Cluster.Leave</c>, which removes a member with no downing decision) can NOT simulate.
    /// The ActorSystem itself stays alive with a dead transport until harness teardown disposes it.
    /// </summary>
    public async Task HardKillNodeAAsync()
    {
        if (NodeA is null) return;
        var provider = (Akka.Remote.IRemoteActorRefProvider)((ExtendedActorSystem)NodeASystem).Provider;
        await provider.Transport.Shutdown();
    }

    /// <summary>
    /// Waits until node B's own cluster view has the crashed node A removed (B is the sole Up member) AND
    /// B has become the <c>driver</c>-role leader — i.e. the split-brain resolver downed the alone oldest
    /// node and B took over. Queries <see cref="NodeBSystem"/> (node A is dead after a hard kill, so its
    /// view can't be read). Returns the elapsed time to takeover; throws <see cref="TimeoutException"/> if
    /// B never becomes the sole driver-leader within <paramref name="timeout"/> (the failure signature of
    /// an INACTIVE resolver — A stays Unreachable and B never fails over).
    /// </summary>
    /// <param name="timeout">Upper bound; must exceed acceptable-heartbeat-pause (10s) + stable-after (15s).</param>
    public async Task<TimeSpan> WaitForNodeBSoleDriverLeaderAsync(TimeSpan timeout)
    {
        var start = DateTime.UtcNow;
        var deadline = start + timeout;
        // Require the outcome to hold across several consecutive polls so a transient during A's
        // Up -> Down convergence (the members list can briefly flap while gossip settles) can't be
        // mistaken for a settled takeover.
        const int requiredStableStreak = 4;
        var streak = 0;
        while (DateTime.UtcNow < deadline)
        {
            var clusterB = Akka.Cluster.Cluster.Get(NodeBSystem);
            var state = clusterB.State;
            var upCount = state.Members.Count(m => m.Status == MemberStatus.Up);
            var driverLeader = state.RoleLeader("driver");
            // SBR proof: the crashed oldest node has been DOWNED (no longer Up, so B is the sole Up
            // member) and B has become the driver-role leader. With the resolver INACTIVE (NoDowning) A
            // would instead stay Up-but-Unreachable — B would still see 2 Up members and never gain a
            // stable role leader. (A lingers as a Down member in the Unreachable set until reaped; that
            // reaping is not part of the failover contract, so it is NOT required here.)
            if (upCount == 1 && driverLeader == clusterB.SelfAddress)
            {
                if (++streak >= requiredStableStreak)
                    return DateTime.UtcNow - start;
            }
            else
            {
                streak = 0;
            }
            await Task.Delay(250);
        }
        var finalState = Akka.Cluster.Cluster.Get(NodeBSystem).State;
        throw new TimeoutException(
            $"Node B did not become the sole driver-role leader within {timeout} after node A was hard-killed. " +
            $"Up={finalState.Members.Count(m => m.Status == MemberStatus.Up)}, " +
            $"Unreachable={finalState.Unreachable.Count}, " +
            $"driverLeader={finalState.RoleLeader("driver")}, selfB={Akka.Cluster.Cluster.Get(NodeBSystem).SelfAddress}. " +
            "A stuck-Unreachable A with no takeover is the signature of an INACTIVE split-brain resolver.");
    }

    /// <summary>
    /// Waits for node A's cluster view to reach <paramref name="expectedUpMembers"/> members in
    /// <see cref="MemberStatus.Up"/>. Used for asserting shrink-after-stop or grow-after-restart.
    /// </summary>
    /// <param name="expectedUpMembers">The expected number of Up members in the cluster.</param>
    /// <param name="timeout">The maximum time to wait for the expected cluster size.</param>
    public async Task WaitForClusterSizeAsync(int expectedUpMembers, TimeSpan timeout)
    {
        var deadline = DateTime.UtcNow + timeout;
        while (DateTime.UtcNow < deadline)
        {
            var count = Akka.Cluster.Cluster.Get(NodeASystem).State.Members
                .Count(m => m.Status == MemberStatus.Up);
            if (count == expectedUpMembers) return;
            await Task.Delay(200);
        }
        var actual = Akka.Cluster.Cluster.Get(NodeASystem).State.Members
            .Count(m => m.Status == MemberStatus.Up);
        throw new TimeoutException(
            $"Cluster did not converge to {expectedUpMembers} Up members within {timeout}. Actual={actual}");
    }

    private enum NodeRole { Seed, Joiner }

    private static async Task<WebApplication> BuildNodeAsync(TwoNodeClusterHarness harness, NodeRole role)
    {
        var akkaPort = role == NodeRole.Seed ? harness.NodeAAkkaPort : harness.NodeBAkkaPort;
        var builder = WebApplication.CreateBuilder(new WebApplicationOptions { Args = [] });

        builder.WebHost.UseKestrel(o => o.Listen(System.Net.IPAddress.Parse(LoopbackHost), 0));

        var configOverrides = new Dictionary<string, string?>
        {
            ["ConnectionStrings:ConfigDb"] = harness._sqlConnString
                ?? "Server=test;Database=test;Trusted_Connection=True;TrustServerCertificate=True;",
            ["Cluster:Hostname"] = LoopbackHost,
            ["Cluster:Port"] = akkaPort.ToString(),
            ["Cluster:PublicHostname"] = LoopbackHost,
            ["Cluster:SeedNodes:0"] = $"akka.tcp://otopcua@{LoopbackHost}:{harness.NodeAAkkaPort}",
            ["Cluster:Roles:0"] = "admin",
            ["Cluster:Roles:1"] = "driver",
            ["Security:Jwt:SigningKey"] = "two-node-harness-test-signing-key-with-enough-bytes-for-hs256",
            ["Security:Jwt:Issuer"] = "otopcua-test",
            ["Security:DeployApiKey"] = HarnessDeployApiKey,
            ["Security:Jwt:Audience"] = "otopcua-test",
        };

        if (harness.Mode.UseRealLdap)
        {
            // Bound section is Security:Ldap (see LdapOptions.SectionName); Transport replaces the
            // old UseTls bool and AllowInsecure replaces AllowInsecureLdap (Task 1.4).
            configOverrides["Security:Ldap:Enabled"] = "true";
            configOverrides["Security:Ldap:Server"] = "localhost";
            configOverrides["Security:Ldap:Port"] = "3894";
            configOverrides["Security:Ldap:Transport"] = "None";
            configOverrides["Security:Ldap:AllowInsecure"] = "true";
            configOverrides["Security:Ldap:SearchBase"] = "dc=zb,dc=local";
            // GLAuth's search-capable bind account (see Docker/glauth/config.toml) — matches the
            // shared GLAuth the data-plane binds against. Replaced the retired OpenLDAP cn=admin.
            configOverrides["Security:Ldap:ServiceAccountDn"] = "cn=serviceaccount,dc=zb,dc=local";
            configOverrides["Security:Ldap:ServiceAccountPassword"] = "serviceaccount123";
        }

        builder.Configuration.AddInMemoryCollection(configOverrides);

        // Provider swap: same DbContext type wired to either InMemory or SqlServer at startup.
        if (harness.Mode.UseSqlServer)
        {
            builder.Services.AddDbContextFactory<OtOpcUaConfigDbContext>(opt => opt.UseSqlServer(harness._sqlConnString));
            builder.Services.AddDbContext<OtOpcUaConfigDbContext>(opt => opt.UseSqlServer(harness._sqlConnString));
        }
        else
        {
            builder.Services.AddDbContextFactory<OtOpcUaConfigDbContext>(opt => opt.UseInMemoryDatabase(harness.SharedDbName));
            builder.Services.AddDbContext<OtOpcUaConfigDbContext>(opt => opt.UseInMemoryDatabase(harness.SharedDbName));
        }

        // Production fidelity: Program.cs calls AddOtOpcUaRuntime() (DI registration) AND
        // WithOtOpcUaRuntimeActors() (Akka spawn). The harness historically called only the latter,
        // so IDriverHealthPublisher fell back to NullDriverHealthPublisher (no health on the
        // driver-health DPS topic) and IDriverFactory to NullDriverFactory (deployed drivers never
        // reached Connected). Register the opt-in fake factory FIRST so its registration wins over the
        // TryAddSingleton<IDriverFactory>(NullDriverFactory) seeded inside AddOtOpcUaRuntime; when no
        // factory was passed, the Null default stays — unchanged behaviour for existing tests. Must run
        // BEFORE AddAkka (see the AddOtOpcUaRuntime XML doc).
        if (harness._driverFactory is not null)
            builder.Services.AddSingleton(harness._driverFactory);
        builder.Services.AddOtOpcUaRuntime();

        builder.Services.AddOtOpcUaCluster(builder.Configuration);

        builder.Services.AddAkka("otopcua", (ab, sp) =>
        {
            ab.WithOtOpcUaClusterBootstrap(sp);
            ab.WithOtOpcUaControlPlaneSingletons();
            ab.WithOtOpcUaRuntimeActors();
        });

        builder.Services.AddOtOpcUaAuth(builder.Configuration);
        if (!harness.Mode.UseRealLdap)
            builder.Services.AddSingleton<ILdapAuthService, StubLdapAuthService>();
        builder.Services.AddAdminUI();
        builder.Services.AddSignalR();
        builder.Services.AddOtOpcUaAdminClients();
        builder.Services.AddOtOpcUaHealth();

        var app = builder.Build();
        app.UseAuthentication();
        app.UseAuthorization();
        app.MapOtOpcUaAuth();
        app.MapOtOpcUaHubs();
        app.MapOtOpcUaHealth();
        app.MapOtOpcUaDeployApi(app.Configuration);

        await app.StartAsync();
        return app;
    }

    private static async Task EnsureSqlSchemaCreatedAsync(string connString)
    {
        var opts = new DbContextOptionsBuilder<OtOpcUaConfigDbContext>()
            .UseSqlServer(connString)
            .Options;
        await using var db = new OtOpcUaConfigDbContext(opts);
        // EnsureCreated bypasses migrations but builds the model in one shot — fine for tests.
        // Production deployments use Migrate-To-V2.ps1 to apply EF migrations.
        await db.Database.EnsureCreatedAsync();
    }

    private static async Task DropSqlDatabaseAsync(string connString)
    {
        var opts = new DbContextOptionsBuilder<OtOpcUaConfigDbContext>()
            .UseSqlServer(connString)
            .Options;
        await using var db = new OtOpcUaConfigDbContext(opts);
        await db.Database.EnsureDeletedAsync();
    }

    private static async Task WaitForClusterFormationAsync(ActorSystem a, ActorSystem b, TimeSpan timeout)
    {
        var deadline = DateTime.UtcNow + timeout;
        while (DateTime.UtcNow < deadline)
        {
            var aMembers = Akka.Cluster.Cluster.Get(a).State.Members
                .Where(m => m.Status == MemberStatus.Up).ToArray();
            var bMembers = Akka.Cluster.Cluster.Get(b).State.Members
                .Where(m => m.Status == MemberStatus.Up).ToArray();
            if (aMembers.Length >= 2 && bMembers.Length >= 2) return;
            await Task.Delay(200);
        }
        throw new TimeoutException(
            $"Cluster did not form within {timeout}. " +
            $"A up={Akka.Cluster.Cluster.Get(a).State.Members.Count(m => m.Status == MemberStatus.Up)}, " +
            $"B up={Akka.Cluster.Cluster.Get(b).State.Members.Count(m => m.Status == MemberStatus.Up)}");
    }

    private static int AllocateFreePort()
    {
        using var listener = new TcpListener(System.Net.IPAddress.Parse(LoopbackHost), 0);
        listener.Start();
        var port = ((System.Net.IPEndPoint)listener.LocalEndpoint).Port;
        listener.Stop();
        return port;
    }

    /// <summary>Asynchronously disposes both nodes and cleans up the SQL database if used.</summary>
    public async ValueTask DisposeAsync()
    {
        if (NodeB is not null) await NodeB.DisposeAsync();
        if (NodeA is not null) await NodeA.DisposeAsync();
        if (_sqlConnString is not null)
        {
            try { await DropSqlDatabaseAsync(_sqlConnString); }
            catch { /* best-effort cleanup */ }
        }
    }

    /// <summary>Captures the env-var driven harness mode at construction time.</summary>
    public sealed record HarnessMode(bool UseSqlServer, bool UseRealLdap)
    {
        /// <summary>Creates a HarnessMode from environment variable settings.</summary>
        /// <returns>A HarnessMode configured from OTOPCUA_HARNESS_USE_SQL and OTOPCUA_HARNESS_USE_LDAP environment variables.</returns>
        public static HarnessMode FromEnvironment() => new(
            UseSqlServer: Environment.GetEnvironmentVariable("OTOPCUA_HARNESS_USE_SQL") == "1",
            UseRealLdap: Environment.GetEnvironmentVariable("OTOPCUA_HARNESS_USE_LDAP") == "1");
    }

    private sealed class StubLdapAuthService : ILdapAuthService
    {
        /// <summary>Asynchronously authenticates a user with the stub LDAP service.</summary>
        /// <param name="username">The username to authenticate.</param>
        /// <param name="password">The password to authenticate against.</param>
        /// <param name="ct">The cancellation token.</param>
        /// <returns>A task that returns the LDAP authentication result.</returns>
        public Task<LdapAuthResult> AuthenticateAsync(string username, string password, CancellationToken ct = default)
            => Task.FromResult(new LdapAuthResult(
                Success: password == "valid-password",
                DisplayName: username,
                Username: username,
                Groups: ["Administrator"],
                Roles: ["Administrator"],
                Error: null));
    }
}