lmxopcua/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/ServiceCollectionExtensions.cs

using Akka.Cluster.Hosting;
using Akka.Cluster.Hosting.SBR;
using Akka.Event;
using Akka.Hosting;
using Akka.Logger.Serilog;
using Akka.Remote.Hosting;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Options;
using ZB.MOM.WW.Configuration;
using ZB.MOM.WW.OtOpcUa.Commons.Interfaces;

namespace ZB.MOM.WW.OtOpcUa.Cluster;

public static class ServiceCollectionExtensions
{
    /// <summary>
    /// Binds <see cref="AkkaClusterOptions"/> and registers <see cref="IClusterRoleInfo"/>. The
    /// actual ActorSystem + cluster bootstrap is layered on inside the host's <c>AddAkka(...)</c>
    /// configurator via <see cref="WithOtOpcUaClusterBootstrap"/> — keeping the entire Akka graph
    /// under Akka.Hosting's management so cluster singletons land on the same ActorSystem.
    /// </summary>
    /// <remarks>
    /// The binding is validated at startup (<c>ValidateOnStart</c>) by
    /// <see cref="AkkaClusterOptionsValidator"/>, so a seed list that cannot bootstrap this node —
    /// self listed behind its partner — fails the host loudly instead of leaving it running but
    /// permanently outside the cluster.
    /// </remarks>
    /// <param name="services">The service collection to configure.</param>
    /// <param name="configuration">The application configuration containing cluster options.</param>
    /// <returns>The same service collection, for chaining.</returns>
    public static IServiceCollection AddOtOpcUaCluster(this IServiceCollection services, IConfiguration configuration)
    {
        services.AddValidatedOptions<AkkaClusterOptions, AkkaClusterOptionsValidator>(
            configuration, AkkaClusterOptions.SectionName);

        // Per-cluster mesh Phase 2: which transport carries central→node commands. Validated at
        // startup for the same reason the cluster options are — a contact point that cannot resolve
        // produces silence, not an error, and silence is indistinguishable from "nothing to do".
        services.AddValidatedOptions<MeshTransportOptions, MeshTransportOptionsValidator>(
            configuration, MeshTransportOptions.SectionName);

        // Per-cluster mesh Phase 3: where a driver reads its config (Direct SQL, or fetch-and-cache
        // over gRPC). Validated at startup for the same reason — a FetchAndCache node with no endpoint
        // or no key produces silence, not an error. ConfigServe (the central serve surface) needs no
        // cross-field validation: a 0 port is a legitimate "disabled", so a plain Configure suffices.
        services.AddValidatedOptions<ConfigSourceOptions, ConfigSourceOptionsValidator>(
            configuration, ConfigSourceOptions.SectionName);
        services.Configure<ConfigServeOptions>(configuration.GetSection(ConfigServeOptions.SectionName));

        // Per-cluster mesh Phase 5: which transport carries a node's live-telemetry stream (node
        // serve side + central dial side). Validated at startup for the same reason ConfigSource is —
        // a Grpc-mode node with no listen port or no key produces silence, not an error.
        services.AddValidatedOptions<TelemetryOptions, TelemetryOptionsValidator>(
            configuration, TelemetryOptions.SectionName);
        services.AddValidatedOptions<TelemetryDialOptions, TelemetryDialOptionsValidator>(
            configuration, TelemetryDialOptions.SectionName);

        services.AddSingleton<IClusterRoleInfo, ClusterRoleInfo>();

        return services;
    }

    /// <summary>
    /// Configures the Akka.Hosting builder with the embedded OtOpcUa HOCON (split-brain resolver,
    /// pinned dispatcher, failure detector tuning) + remote endpoint + cluster bootstrap derived
    /// from <see cref="AkkaClusterOptions"/>.
    ///
    /// Wire from Program.cs:
    /// <code>
    /// services.AddAkka("otopcua", (ab, sp) =>
    /// {
    ///     ab.WithOtOpcUaClusterBootstrap(sp);
    ///     if (hasAdmin) ab.WithOtOpcUaControlPlaneSingletons();
    ///     if (hasDriver) ab.WithOtOpcUaRuntimeActors();
    /// });
    /// </code>
    /// </summary>
    /// <param name="builder">The Akka configuration builder to configure.</param>
    /// <param name="serviceProvider">The service provider for resolving cluster options.</param>
    /// <returns>The same builder, for chaining.</returns>
    public static AkkaConfigurationBuilder WithOtOpcUaClusterBootstrap(
        this AkkaConfigurationBuilder builder,
        IServiceProvider serviceProvider)
    {
        var options = serviceProvider.GetRequiredService<IOptions<AkkaClusterOptions>>().Value;

        builder.AddHocon(HoconLoader.LoadBaseConfig(), HoconAddMode.Append);

        // Route Akka's internal ILoggingAdapter (DriverHostActor, DriverInstanceActor, cluster
        // events, …) into Serilog so those logs reach the same sinks as the MEL/Serilog application
        // logs. Akka.Hosting owns logger setup, so HOCON `akka.loggers` alone is not honored — the
        // logger must be registered through ConfigureLoggers. Without this the actor graph logs only
        // to the default StandardOutLogger (discarded under the Windows service host), which is why
        // the driver-role actors were invisible during the 2026-06 data-plane investigation.
        builder.ConfigureLoggers(setup =>
        {
            setup.LogLevel = LogLevel.DebugLevel;
            setup.ClearLoggers();
            setup.AddLogger<SerilogLogger>();
        });

        builder.WithRemoting(new RemoteOptions
        {
            HostName = options.Hostname,
            Port = options.Port,
            PublicHostName = options.PublicHostname,
        });

        builder.WithClustering(BuildClusterOptions(options));

        // Must come AFTER WithClustering, which always emits a downing-provider-class of its own:
        // Akka.Cluster.Hosting applies SplitBrainResolverOption.Default whenever the typed
        // ClusterOptions.SplitBrainResolver is null, so there is always a competing value to beat.
        // Prepend is the highest-precedence mode, which makes this block win irrespective of how
        // many fragments are added or in what order.
        //
        // Do not take this comment's word for it. Which fragment actually wins is not obvious from
        // the mode names — measured, Append also wins here, purely because it happens to be added
        // last. SplitBrainResolverActivationTests reads the provider back off a *running*
        // ActorSystem for exactly that reason: the effective value is the only one worth asserting.
        builder.AddHocon(BuildDowningHocon(options), HoconAddMode.Prepend);

        // NOTE (2026-07-22): no bootstrap watchdog is armed here any more. The cold-start-alone gap
        // it existed for — Akka lets only seed-nodes[0] form a NEW cluster — is now closed by
        // self-first seed ordering (AkkaClusterOptions.SeedNodes), enforced at boot by
        // AkkaClusterOptionsValidator. The retired ClusterBootstrapFallback timer sat OUTSIDE Akka's
        // join handshake, so it could not distinguish "no seed answered" from "a seed answered and
        // the join is in flight", and Cluster.Join(SelfAddress) is not ignored mid-handshake — it
        // wins. See docs/Redundancy.md → "Bootstrap: self-first seed ordering".

        return builder;
    }

    /// <summary>
    /// How long a member must stay unreachable before it is downed. Shared by both strategies
    /// (<c>auto-down-unreachable-after</c> and the SBR resolver's <c>stable-after</c>) so the two
    /// have the same failover latency. Must stay above
    /// <c>akka.cluster.failure-detector.acceptable-heartbeat-pause</c> or a merely-slow node gets
    /// downed; both constraints are pinned by tests.
    /// </summary>
    public static readonly TimeSpan DowningStableAfter = TimeSpan.FromSeconds(15);

    /// <summary>
    /// Builds the HOCON that selects the downing provider, per
    /// <see cref="AkkaClusterOptions.SplitBrainResolverStrategy"/>.
    /// </summary>
    /// <param name="options">The bound cluster options carrying the strategy.</param>
    /// <returns>
    /// For <c>auto-down</c>, a block installing Akka's <c>AutoDowning</c> provider with a downing
    /// window of <see cref="DowningStableAfter"/>. For <c>keep-oldest</c>, an empty string — the
    /// typed <see cref="KeepOldestOption"/> from <see cref="BuildClusterOptions"/> already installs
    /// the SBR provider, and the resolver's own settings live in <c>Resources/akka.conf</c>.
    /// </returns>
    /// <exception cref="ArgumentOutOfRangeException">The strategy is not a recognised value.</exception>
    public static string BuildDowningHocon(AkkaClusterOptions options)
    {
        ArgumentNullException.ThrowIfNull(options);

        if (IsAutoDown(options))
        {
            // auto-down-unreachable-after is load-bearing: it defaults to `off`, under which
            // AutoDowning is installed but never downs anything — failover silently disabled.
            return $$"""
                akka.cluster {
                    downing-provider-class = "Akka.Cluster.AutoDowning, Akka.Cluster"
                    auto-down-unreachable-after = {{(int)DowningStableAfter.TotalMilliseconds}}ms
                }
                """;
        }

        if (IsKeepOldest(options))
        {
            return string.Empty;
        }

        throw new ArgumentOutOfRangeException(
            nameof(options),
            options.SplitBrainResolverStrategy,
            $"Unknown {AkkaClusterOptions.SectionName}:{nameof(AkkaClusterOptions.SplitBrainResolverStrategy)}. "
            + "Expected 'auto-down' (default; survives a crash of either node) or 'keep-oldest' "
            + "(partition-safe, but a two-node pair cannot survive a crash of the oldest node).");
    }

    private static bool IsAutoDown(AkkaClusterOptions options) =>
        string.Equals(options.SplitBrainResolverStrategy, "auto-down", StringComparison.OrdinalIgnoreCase);

    private static bool IsKeepOldest(AkkaClusterOptions options) =>
        string.Equals(options.SplitBrainResolverStrategy, "keep-oldest", StringComparison.OrdinalIgnoreCase);

    /// <summary>
    /// Builds the <see cref="ClusterOptions"/> for the fused-host cluster.
    /// </summary>
    /// <remarks>
    ///     <para>
    ///         <b>Activation note (arch-review 03/S1, corrected twice).</b> Akka.Cluster.Hosting's
    ///         <c>WithClustering</c> always installs a downing provider: when
    ///         <see cref="ClusterOptions.SplitBrainResolver"/> is <c>null</c> it applies
    ///         <c>SplitBrainResolverOption.Default</c>, which registers
    ///         <c>Akka.Cluster.SBR.SplitBrainResolverProvider</c> and reads the
    ///         <c>split-brain-resolver</c> block in <c>Resources/akka.conf</c>. The cluster is
    ///         therefore never <c>NoDowning</c>. What this method chooses is only <i>which</i>
    ///         provider — and under the default <c>auto-down</c> strategy the choice is not made
    ///         here at all but by the Prepended HOCON in
    ///         <see cref="WithOtOpcUaClusterBootstrap"/>, which outranks whatever
    ///         <c>WithClustering</c> emits.
    ///     </para>
    ///     <para>
    ///         <b>Why keep-oldest is no longer the default (2026-07-21).</b> The previous
    ///         documentation here asserted that <see cref="KeepOldestOption"/> with
    ///         <c>DownIfAlone=true</c> was "the correct strategy for a 2-node warm-redundancy pair"
    ///         because <c>down-if-alone</c> would down a node that lost its peer. That is the
    ///         opposite of what Akka.NET does. In <c>KeepOldest.OldestDecision</c> the
    ///         <c>down-if-alone</c> branch requires the surviving side to hold &gt;= 2 members, so
    ///         in a 1-vs-1 split the survivor falls through to <c>DownReachable</c> and downs
    ///         <i>itself</i>; with <c>run-coordinated-shutdown-when-down = on</c> it then exits. A
    ///         two-node pair running keep-oldest converts a crash of the oldest node into a total
    ///         outage — precisely the failure redundancy exists to absorb. Live-proven on the sister
    ///         project's rig; see <c>docs/Redundancy.md</c> and
    ///         <see cref="AkkaClusterOptions.SplitBrainResolverStrategy"/> for the trade.
    ///     </para>
    /// </remarks>
    /// <param name="options">The bound cluster options carrying seed nodes, roles and the strategy.</param>
    /// <returns>
    /// The cluster options with seed nodes and roles. <see cref="ClusterOptions.SplitBrainResolver"/>
    /// is set only under <c>keep-oldest</c>, where the SBR provider is the one wanted; under
    /// <c>auto-down</c> it is left null because the Prepended HOCON replaces the provider anyway,
    /// and naming a resolver that is not in force would be misleading.
    /// </returns>
    public static ClusterOptions BuildClusterOptions(AkkaClusterOptions options)
    {
        ArgumentNullException.ThrowIfNull(options);

        return new ClusterOptions
        {
            SeedNodes = options.SeedNodes,
            Roles = options.Roles,
            SplitBrainResolver = IsKeepOldest(options) ? new KeepOldestOption { DownIfAlone = true } : null,
        };
    }
}