ScadaBridge/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/AkkaSiteTransport.cs

using System.Collections.Immutable;
using Akka.Actor;
using Akka.Cluster.Tools.Client;
using Akka.Event;

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

/// <summary>
/// The default (and, until the Phase 1B cutover, the shipping) central→site transport: routes each
/// <see cref="SiteEnvelope"/> through a per-site Akka <see cref="ClusterClient"/>, exactly as
/// <c>CentralCommunicationActor</c> did inline before the seam was extracted.
/// </summary>
/// <remarks>
/// <para>
/// Behaviour is identical to the pre-seam code: the per-site client map is (re)built from the DB
/// refresh cache; a send to a site with no client is warned and dropped (the caller's Ask times
/// out — central never buffers); a send preserves the reply-to sender so the site's reply routes
/// straight back to the waiting Ask (or the debug-bridge actor).
/// </para>
/// <para>
/// Both members run on the owning actor's thread, so the client map needs no synchronisation and
/// the stored <see cref="IActorContext"/> (used for <c>Context.Stop</c> and <c>Context.System</c>)
/// is only ever touched there.
/// </para>
/// </remarks>
public sealed class AkkaSiteTransport : ISiteCommandTransport
{
    private readonly ISiteClientFactory _siteClientFactory;
    private readonly IActorContext _context;
    private readonly ILoggingAdapter _log;

    /// <summary>
    /// Per-site ClusterClient instances and their contact addresses.
    /// Maps SiteIdentifier → (ClusterClient actor, set of contact address strings).
    /// Refreshed by <see cref="ReconcileSites"/>.
    /// </summary>
    private readonly Dictionary<string, (IActorRef Client, ImmutableHashSet<string> ContactAddresses)> _siteClients = new();

    /// <summary>Creates the Akka transport bound to the owning actor's context.</summary>
    /// <param name="siteClientFactory">Factory that creates a ClusterClient per site.</param>
    /// <param name="context">The owning actor's context (for <c>Stop</c>/<c>System</c>); calls stay on the actor thread.</param>
    /// <param name="log">The owning actor's logger, so warnings keep the actor's log source.</param>
    public AkkaSiteTransport(ISiteClientFactory siteClientFactory, IActorContext context, ILoggingAdapter log)
    {
        _siteClientFactory = siteClientFactory ?? throw new ArgumentNullException(nameof(siteClientFactory));
        _context = context ?? throw new ArgumentNullException(nameof(context));
        _log = log ?? throw new ArgumentNullException(nameof(log));
    }

    /// <inheritdoc />
    public void Send(SiteEnvelope envelope, IActorRef replyTo)
    {
        ArgumentNullException.ThrowIfNull(envelope);

        if (!_siteClients.TryGetValue(envelope.SiteId, out var entry))
        {
            _log.Warning("No ClusterClient for site {0}, cannot route message {1}",
                envelope.SiteId, envelope.Message.GetType().Name);

            // The Ask will timeout on the caller side — no central buffering
            return;
        }

        // Route via ClusterClient — replyTo is preserved for Ask response routing
        entry.Client.Tell(
            new ClusterClient.Send("/user/site-communication", envelope.Message),
            replyTo);
    }

    /// <inheritdoc />
    public void ReconcileSites(SiteAddressCacheLoaded cache)
    {
        ArgumentNullException.ThrowIfNull(cache);

        var newSiteIds = cache.SiteContacts.Keys.ToHashSet();
        var existingSiteIds = _siteClients.Keys.ToHashSet();

        // Stop ClusterClients for removed sites
        foreach (var removed in existingSiteIds.Except(newSiteIds))
        {
            _log.Info("Stopping ClusterClient for removed site {0}", removed);
            _context.Stop(_siteClients[removed].Client);
            _siteClients.Remove(removed);
        }

        // Add or update
        foreach (var (siteId, addresses) in cache.SiteContacts)
        {
            // Parse all addresses up front inside a try/catch so a
            // single malformed site row cannot abort the whole refresh loop and leave
            // the cache half-updated. A bad site is logged and skipped; others proceed.
            ImmutableHashSet<ActorPath> contactPaths;
            try
            {
                contactPaths = addresses
                    .Select(a => ActorPath.Parse($"{a}/system/receptionist"))
                    .ToImmutableHashSet();
            }
            catch (Exception ex)
            {
                _log.Warning(ex,
                    "Malformed contact address for site {0}; skipping this site in the refresh "
                    + "(other sites are unaffected)", siteId);
                continue;
            }

            var contactStrings = addresses.ToImmutableHashSet();

            // Skip if unchanged
            if (_siteClients.TryGetValue(siteId, out var existing) && existing.ContactAddresses.SetEquals(contactStrings))
                continue;

            // Stop old client if addresses changed
            if (_siteClients.ContainsKey(siteId))
            {
                _log.Info("Updating ClusterClient for site {0} (addresses changed)", siteId);
                _context.Stop(_siteClients[siteId].Client);
                // Remove now: if the replacement create below fails, a stale entry
                // would route envelopes to a stopping actor.
                _siteClients.Remove(siteId);
            }

            IActorRef client;
            try
            {
                client = _siteClientFactory.Create(_context.System, siteId, contactPaths);
            }
            catch (Exception ex)
            {
                _log.Error(ex,
                    "Failed to create ClusterClient for site {0}; site is unroutable until the next refresh",
                    siteId);
                continue;
            }
            _siteClients[siteId] = (client, contactStrings);
            _log.Info("Created ClusterClient for site {0} with {1} contact(s)", siteId, addresses.Count);
        }

        _log.Info("Site ClusterClient cache refreshed with {0} site(s)", _siteClients.Count);
    }
}