ScadaBridge/src/ZB.MOM.WW.ScadaBridge.Host/Health/AkkaManualFailoverService.cs

using System.Text.Json;
using Akka.Actor;
using Akka.Cluster;
using Microsoft.Extensions.Logging;
using ZB.MOM.WW.ScadaBridge.CentralUI.Services;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
using ZB.MOM.WW.ScadaBridge.Communication;
using ZB.MOM.WW.ScadaBridge.Communication.ClusterState;
using ZB.MOM.WW.ScadaBridge.Host.Actors;

namespace ZB.MOM.WW.ScadaBridge.Host.Health;

/// <summary>
/// Production <see cref="IManualFailoverService"/> backed by the running Akka.NET cluster.
/// Registered only in the Central-role branch of <c>Program.cs</c>.
///
/// <para><b>Leave, never Down.</b> The target is asked to leave gracefully, so
/// <c>ClusterSingletonManager</c> hands its singletons to the survivor before the member is
/// removed. A <c>Down</c> would skip that hand-off and leave the pair to the downing strategy
/// — the wrong tool for a deliberate, planned role swap.</para>
///
/// <para><b>The target is the oldest Up member, not the leader.</b> That mirrors
/// <c>ActiveNodeEvaluator</c>'s rule, which is where the singletons actually live; Akka's
/// cluster leadership is address-ordered and diverges from singleton placement after a
/// restart (review 01 [High]).</para>
///
/// <para><b>Audit before acting.</b> The row is written before the Leave is issued. The node
/// serving this call is usually NOT the one leaving, but it can be (an admin routed to the
/// active node fails that node over), and an audit written afterwards could be lost to the
/// very shutdown it describes.</para>
/// </summary>
public sealed class AkkaManualFailoverService : IManualFailoverService
{
    private readonly AkkaHostedService _akka;
    private readonly ICentralAuditWriter _audit;
    private readonly CommunicationService _communication;
    private readonly ILogger<AkkaManualFailoverService> _logger;

    /// <summary>Initializes a new <see cref="AkkaManualFailoverService"/>.</summary>
    /// <param name="akka">The Akka hosted service exposing the cluster's actor system.</param>
    /// <param name="audit">Central direct-write audit writer.</param>
    /// <param name="communication">Central→site command/control transport, used for site failover.</param>
    /// <param name="logger">Logger.</param>
    public AkkaManualFailoverService(
        AkkaHostedService akka,
        ICentralAuditWriter audit,
        CommunicationService communication,
        ILogger<AkkaManualFailoverService> logger)
    {
        _akka = akka;
        _audit = audit;
        _communication = communication;
        _logger = logger;
    }

    /// <inheritdoc />
    public async Task<string?> FailOverCentralAsync(string actor)
    {
        var system = _akka.GetOrCreateActorSystem();

        // Resolve first so the audit row can name the target, and so the peer guard rejects
        // before anything observable happens.
        var target = FailOverCore(system, role: CentralRole, dryRun: true);
        if (target is null)
        {
            _logger.LogWarning(
                "Manual failover requested by {Actor} but refused: fewer than 2 Up '{Role}' members, "
                + "so there is no standby to take over.", actor, CentralRole);
            return null;
        }

        await WriteAuditAsync(actor, target);

        _logger.LogWarning(
            "Manual failover triggered by {Actor}: {Target} is leaving the cluster gracefully; "
            + "its singletons hand over to the standby, it restarts via its supervisor and rejoins "
            + "as the youngest member.", actor, target);

        FailOverCore(system, role: CentralRole);
        return target.ToString();
    }

    /// <inheritdoc />
    public async Task<SiteFailoverOutcome> FailOverSiteAsync(string siteId, string actor)
    {
        // Audit BEFORE relaying, for the same reason as central: the row must exist even if
        // the outcome is never observed. Unlike central, the acting node is never the one
        // leaving — but a relay can still time out, and an un-acked request that DID take
        // effect at the site would otherwise be unattributed.
        await WriteAuditAsync(actor, target: siteId, sourceSiteId: siteId);

        _logger.LogWarning(
            "Manual site failover triggered by {Actor} for site {SiteId}; relaying to the site cluster.",
            actor, siteId);

        try
        {
            var ack = await _communication.TriggerSiteFailoverAsync(siteId, Guid.NewGuid().ToString());
            return new SiteFailoverOutcome(ack.Accepted, ack.TargetAddress, ack.ErrorMessage);
        }
        catch (Exception ex)
        {
            // Central never buffers for an unreachable site — the Ask simply times out. Report
            // that distinctly from a refusal, which is a definitive answer FROM the site.
            _logger.LogWarning(ex,
                "Manual site failover for {SiteId} did not get an ack from the site.", siteId);
            return new SiteFailoverOutcome(
                Accepted: false,
                TargetAddress: null,
                ErrorMessage: $"Site did not respond: {ex.Message}");
        }
    }

    /// <summary>The Akka role scoping central-cluster membership.</summary>
    private const string CentralRole = "Central";

    /// <summary>
    /// Oldest Up member with the role leaves. Delegates to
    /// <see cref="ClusterFailoverCoordinator.FailOverOldest"/>, which lives in Communication so
    /// the site-pair failover path (inside <c>SiteCommunicationActor</c>, which cannot reference
    /// Host) shares one implementation of the rule.
    /// </summary>
    /// <param name="system">The actor system whose cluster is acted on.</param>
    /// <param name="role">Role scope for membership.</param>
    /// <param name="dryRun">When true, resolve and return the target without issuing the Leave.</param>
    /// <returns>The address that leaves (or would leave), or null when there is no peer.</returns>
    public static Address? FailOverCore(ActorSystem system, string role, bool dryRun = false)
        => ClusterFailoverCoordinator.FailOverOldest(system, role, dryRun);

    /// <summary>
    /// Best-effort audit row. Audit failure must never block the failover the admin asked for
    /// — the same rule the rest of the Audit Log follows (audit is best-effort; the
    /// user-facing action's own success path is authoritative).
    /// </summary>
    private Task WriteAuditAsync(string actor, Address target)
        => WriteAuditAsync(actor, target.ToString(), sourceSiteId: null);

    /// <inheritdoc cref="WriteAuditAsync(string, Address)"/>
    private async Task WriteAuditAsync(string actor, string target, string? sourceSiteId)
    {
        try
        {
            var evt = ScadaBridgeAuditEventFactory.Create(
                channel: AuditChannel.Cluster,
                kind: AuditKind.ManualFailover,
                status: AuditStatus.Submitted,
                actor: actor,
                target: target,
                sourceSiteId: sourceSiteId,
                extra: JsonSerializer.Serialize(new
                {
                    target,
                    // Central rows name the Central role; site rows name the site, so a query
                    // can tell which pair an operator moved.
                    scope = sourceSiteId is null ? CentralRole : $"site-{sourceSiteId}"
                }));

            await _audit.WriteAsync(evt);
        }
        catch (Exception ex)
        {
            _logger.LogWarning(
                ex,
                "Best-effort manual-failover audit emission failed (actor={Actor}, target={Target}); "
                + "the failover itself proceeds.", actor, target);
        }
    }
}