ScadaBridge/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/SiteRuntimeFixture.cs

using System.Diagnostics;
using System.Text.Json;
using Akka.Actor;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging.Abstractions;
using ZB.MOM.WW.LocalDb;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.DataConnection;
using ZB.MOM.WW.ScadaBridge.Communication.Grpc;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Flattening;
using ZB.MOM.WW.ScadaBridge.DataConnectionLayer;
using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Actors;
using ZB.MOM.WW.ScadaBridge.HealthMonitoring;
using ZB.MOM.WW.ScadaBridge.LoadHarness.Metrics;
using ZB.MOM.WW.ScadaBridge.LoadHarness.Probes;
using ZB.MOM.WW.ScadaBridge.SiteRuntime;
using ZB.MOM.WW.ScadaBridge.SiteRuntime.Actors;
using ZB.MOM.WW.ScadaBridge.SiteRuntime.Persistence;
using ZB.MOM.WW.ScadaBridge.SiteRuntime.Scripts;
using ZB.MOM.WW.ScadaBridge.SiteRuntime.Streaming;
using ZB.MOM.WW.ScadaBridge.StoreAndForward;

namespace ZB.MOM.WW.ScadaBridge.LoadHarness;

/// <summary>
/// One simulated site: its own <see cref="ActorSystem"/>, its own LocalDb SQLite
/// file, a real Data Connection Layer over <see cref="SimulatedDataConnection"/>
/// adapters, real Instance Actors, a real <see cref="SiteStreamManager"/>, real
/// store-and-forward, and a real <see cref="SiteHealthCollector"/>.
///
/// <para>
/// The sites are separate, non-clustered ActorSystems rather than 10 real two-node
/// Akka clusters. Cluster membership, singleton placement and failover timing are
/// already measured on a real two-node rig by
/// <c>PerformanceTests/Failover/FailoverTimingTests.cs</c>; what matters here is
/// the load-bearing hierarchy under each singleton, which is what this builds.
/// </para>
/// </summary>
public sealed class SiteRuntimeFixture : IAsyncDisposable
{
    private readonly HarnessConfig _config;
    private readonly List<SimulatedDataConnection> _connections;
    private readonly ServiceProvider _localDbProvider;
    private readonly ILocalDb _localDb;
    private readonly List<IActorRef> _instanceActors = new();
    private readonly List<StreamSubscriberProbe> _probes = new();
    private readonly List<WireStreamSubscriber> _wireSubscribers = new();
    private readonly string _dataDirectory;
    private SiteWireStreamHost? _wireHost;
    private SiteStreamGrpcClient? _wireClient;
    private CancellationTokenSource? _wireCts;

    /// <summary>Site identifier, for example a zero-padded numbered name for the first site.</summary>
    public string SiteId { get; }

    /// <summary>This site's actor system.</summary>
    public ActorSystem System { get; }

    /// <summary>The real site-wide broadcast stream.</summary>
    public SiteStreamManager StreamManager { get; }

    /// <summary>The real site health collector feeding the 30 s report.</summary>
    public SiteHealthCollector HealthCollector { get; }

    /// <summary>The real store-and-forward engine for this site.</summary>
    public StoreAndForwardService StoreAndForward { get; }

    /// <summary>The real DCL manager actor.</summary>
    public IActorRef DataConnectionManager { get; }

    /// <summary>The simulated adapters, one per data connection, in creation order.</summary>
    public IReadOnlyList<SimulatedDataConnection> Connections
    {
        get { lock (_connections) return _connections.ToList(); }
    }

    /// <summary>The live stream subscribers attached to this site.</summary>
    public IReadOnlyList<StreamSubscriberProbe> Probes => _probes;

    /// <summary>The wire-path (real gRPC) subscribers attached to this site, if any.</summary>
    public IReadOnlyList<WireStreamSubscriber> WireSubscribers => _wireSubscribers;

    /// <summary>The site's real gRPC listener, when wire probes are attached.</summary>
    public SiteWireStreamHost? WireHost => _wireHost;

    /// <summary>Instance Actors created on this site.</summary>
    public IReadOnlyList<IActorRef> InstanceActors => _instanceActors;

    /// <summary>Tag paths per connection index, in the order they were assigned.</summary>
    public IReadOnlyList<List<string>> TagPathsByConnection { get; }

    /// <summary>Wall-clock time the instance ramp took, measured by <see cref="StartInstancesAsync"/>.</summary>
    public TimeSpan InstanceRampDuration { get; private set; }

    private SiteRuntimeFixture(
        string siteId,
        HarnessConfig config,
        string dataDirectory,
        ServiceProvider localDbProvider,
        ILocalDb localDb,
        ActorSystem system,
        SiteStreamManager streamManager,
        SiteHealthCollector healthCollector,
        StoreAndForwardService storeAndForward,
        IActorRef dataConnectionManager,
        SiteStorageService storage,
        ScriptCompilationService compilationService,
        SharedScriptLibrary sharedScriptLibrary,
        SiteRuntimeOptions siteOptions,
        List<List<string>> tagPathsByConnection,
        List<SimulatedDataConnection> connections)
    {
        _connections = connections;
        SiteId = siteId;
        _config = config;
        _dataDirectory = dataDirectory;
        _localDbProvider = localDbProvider;
        _localDb = localDb;
        System = system;
        StreamManager = streamManager;
        HealthCollector = healthCollector;
        StoreAndForward = storeAndForward;
        DataConnectionManager = dataConnectionManager;
        Storage = storage;
        CompilationService = compilationService;
        SharedScriptLibrary = sharedScriptLibrary;
        SiteOptions = siteOptions;
        TagPathsByConnection = tagPathsByConnection;
    }

    private SiteStorageService Storage { get; }
    private ScriptCompilationService CompilationService { get; }
    private SharedScriptLibrary SharedScriptLibrary { get; }
    private SiteRuntimeOptions SiteOptions { get; }

    /// <summary>
    /// Builds a site: LocalDb file, actor system, DCL with its simulated connections,
    /// stream manager, health collector and store-and-forward engine. Instance Actors
    /// are created separately by <see cref="StartInstancesAsync"/> so the deployment
    /// ramp can be timed on its own.
    /// </summary>
    /// <param name="siteIndex">Zero-based site index.</param>
    /// <param name="config">Harness configuration.</param>
    /// <param name="rootDataDirectory">Directory under which this site's SQLite files live.</param>
    /// <param name="connectionsPerSite">Number of data connections to spread the site's tags across.</param>
    /// <returns>The started fixture.</returns>
    public static async Task<SiteRuntimeFixture> CreateAsync(
        int siteIndex, HarnessConfig config, string rootDataDirectory, int connectionsPerSite)
    {
        var siteId = $"site-{siteIndex + 1:D2}";
        var dataDirectory = Path.Combine(rootDataDirectory, siteId);
        Directory.CreateDirectory(dataDirectory);

        var configuration = new ConfigurationBuilder()
            .AddInMemoryCollection(new Dictionary<string, string?>
            {
                ["LocalDb:Path"] = Path.Combine(dataDirectory, "site-localdb.db"),
            })
            .Build();

        var localDbProvider = new ServiceCollection()
            .AddZbLocalDb(configuration)
            .BuildServiceProvider();
        var localDb = localDbProvider.GetRequiredService<ILocalDb>();

        var storage = new SiteStorageService(localDb, NullLogger<SiteStorageService>.Instance);
        await storage.InitializeAsync();

        var compilationService = new ScriptCompilationService(NullLogger<ScriptCompilationService>.Instance);
        var sharedScriptLibrary = new SharedScriptLibrary(compilationService, NullLogger<SharedScriptLibrary>.Instance);

        // Production defaults throughout — the point of the run is to measure the
        // shipped configuration, so nothing here is tuned for the harness.
        var siteOptions = new SiteRuntimeOptions();

        // WARNING-level logging. At 37,500 updates/s per site, Akka's INFO output would
        // itself become a measured load; and the InstanceActorInitialized dead letters are
        // a harness artifact (see StartInstancesAsync) rather than a real condition.
        var system = ActorSystem.Create($"loadharness-{siteId}", Akka.Configuration.ConfigurationFactory.ParseString(
            "akka.loglevel = WARNING\nakka.stdout-loglevel = WARNING\nakka.log-dead-letters = 0\nakka.log-dead-letters-during-shutdown = off"));

        var streamManager = new SiteStreamManager(siteOptions, NullLogger<SiteStreamManager>.Instance);
        streamManager.Initialize(system);

        var healthCollector = new SiteHealthCollector();
        healthCollector.SetActiveNode(true);
        healthCollector.SetNodeHostname($"{siteId}-node-a");

        var sfStorage = new StoreAndForwardStorage(localDb, NullLogger<StoreAndForwardStorage>.Instance);
        var storeAndForward = new StoreAndForwardService(
            sfStorage,
            new StoreAndForwardOptions(),
            NullLogger<StoreAndForwardService>.Instance,
            siteId: siteId);
        await storeAndForward.StartAsync();

        // Real DCL, with the simulated adapter registered on the real factory via its
        // documented RegisterAdapter extension point.
        var loggerFactory = NullLoggerFactory.Instance;
        var factory = new DataConnectionFactory(loggerFactory);
        var connections = new List<SimulatedDataConnection>();
        factory.RegisterAdapter(SimulatedDataConnection.ProtocolName, _ =>
        {
            var connection = new SimulatedDataConnection();
            lock (connections) connections.Add(connection);
            return connection;
        });

        var dclManager = system.ActorOf(
            Props.Create(() => new DataConnectionManagerActor(
                factory, new DataConnectionOptions(), healthCollector, null, null)),
            "data-connection-manager");

        for (var c = 0; c < connectionsPerSite; c++)
        {
            dclManager.Tell(new CreateConnectionCommand(
                ConnectionName: ConnectionName(c),
                ProtocolType: SimulatedDataConnection.ProtocolName,
                PrimaryConnectionDetails: new Dictionary<string, string>
                {
                    ["endpoint"] = $"sim://{siteId}/{c}",
                    [SimulatedDataConnection.ConnectionNameKey] = ConnectionName(c),
                }));
        }

        var tagPathsByConnection = new List<List<string>>();
        for (var c = 0; c < connectionsPerSite; c++)
            tagPathsByConnection.Add(new List<string>());

        var fixture = new SiteRuntimeFixture(
            siteId, config, dataDirectory, localDbProvider, localDb, system, streamManager,
            healthCollector, storeAndForward, dclManager, storage, compilationService,
            sharedScriptLibrary, siteOptions, tagPathsByConnection, connections);

        return fixture;
    }

    /// <summary>Deterministic connection name for a connection index.</summary>
    /// <param name="connectionIndex">Zero-based connection index.</param>
    /// <returns>The connection name used in configs and DCL commands.</returns>
    public static string ConnectionName(int connectionIndex) => $"sim-conn-{connectionIndex:D2}";

    /// <summary>
    /// Creates this site's Instance Actors in production-shaped staggered batches
    /// (<see cref="SiteRuntimeOptions.StartupBatchSize"/> /
    /// <see cref="SiteRuntimeOptions.StartupBatchDelayMs"/>) and records how long the
    /// ramp took. This is the harness's stand-in for "deployment of 500 instances to
    /// a site" — it exercises the same per-instance construction, config
    /// deserialization, override load and DCL subscribe that a real deploy triggers.
    /// </summary>
    /// <param name="cancellationToken">Cancels the ramp.</param>
    /// <returns>A task that completes when every instance actor exists.</returns>
    public async Task StartInstancesAsync(CancellationToken cancellationToken)
    {
        var started = Stopwatch.StartNew();
        var connectionCount = TagPathsByConnection.Count;

        for (var i = 0; i < _config.InstancesPerSite; i++)
        {
            var instanceName = InstanceName(i);
            var connectionIndex = i % connectionCount;
            var connectionName = ConnectionName(connectionIndex);

            var attributes = new List<ResolvedAttribute>(_config.TagsPerInstance);
            for (var t = 0; t < _config.TagsPerInstance; t++)
            {
                var tagPath = $"{instanceName}.Tag{t:D3}";
                TagPathsByConnection[connectionIndex].Add(tagPath);
                attributes.Add(new ResolvedAttribute
                {
                    CanonicalName = $"Tag{t:D3}",
                    DataType = "Double",
                    DataSourceReference = tagPath,
                    BoundDataConnectionId = connectionIndex + 1,
                    BoundDataConnectionName = connectionName,
                    BoundDataConnectionProtocol = SimulatedDataConnection.ProtocolName,
                });
            }

            var configuration = new FlattenedConfiguration
            {
                InstanceUniqueName = instanceName,
                TemplateId = 1,
                SiteId = 1,
                Attributes = attributes,
                Connections = new Dictionary<string, ConnectionConfig>
                {
                    [connectionName] = new()
                    {
                        Protocol = SimulatedDataConnection.ProtocolName,
                        ConfigurationJson = "{}",
                    },
                },
            };

            var configJson = JsonSerializer.Serialize(configuration);
            var actor = System.ActorOf(
                Props.Create(() => new InstanceActor(
                    instanceName, configJson, Storage, CompilationService, SharedScriptLibrary,
                    StreamManager, SiteOptions, NullLogger<InstanceActor>.Instance,
                    DataConnectionManager, HealthCollector, null, null)),
                instanceName);
            _instanceActors.Add(actor);

            // Production staggered-startup pacing (SiteRuntimeOptions defaults):
            // batches of StartupBatchSize separated by StartupBatchDelayMs, which is
            // exactly what DeploymentManagerActor does on a real site start.
            if ((i + 1) % SiteOptions.StartupBatchSize == 0)
                await Task.Delay(SiteOptions.StartupBatchDelayMs, cancellationToken);
        }

        started.Stop();
        InstanceRampDuration = started.Elapsed;

        HealthCollector.SetInstanceCounts(
            _config.InstancesPerSite, _config.InstancesPerSite, 0);
        for (var c = 0; c < connectionCount; c++)
        {
            HealthCollector.UpdateTagResolution(
                ConnectionName(c), TagPathsByConnection[c].Count, TagPathsByConnection[c].Count);
        }
    }

    /// <summary>Deterministic instance unique name for an instance index.</summary>
    /// <param name="instanceIndex">Zero-based instance index.</param>
    /// <returns>The instance unique name.</returns>
    public string InstanceName(int instanceIndex) => $"{SiteId}-inst-{instanceIndex:D4}";

    /// <summary>
    /// Attaches <see cref="HarnessConfig.StreamProbesPerSite"/> live stream subscribers,
    /// each built from the production pieces the gRPC server uses: a real
    /// <see cref="StreamRelayActor"/> writing into a bounded <c>DropOldest</c> channel
    /// of the production capacity, subscribed through the real
    /// <see cref="SiteStreamManager.Subscribe"/>. Only the socket writer is replaced —
    /// by a reader task under the harness's control, which is what makes the
    /// slow-subscriber scenario possible at all.
    ///
    /// <para>
    /// Attaching at least one subscriber is also load-bearing:
    /// <c>PublishAttributeValueChanged</c> short-circuits at zero subscribers, so an
    /// unsubscribed site would publish nothing and measure nothing.
    /// </para>
    /// </summary>
    /// <param name="latency">Histogram that receives end-to-end tag update latencies.</param>
    public void AttachStreamProbes(LatencyHistogram latency)
    {
        var stride = Math.Max(1, _config.InstancesPerSite / Math.Max(1, _config.StreamProbesPerSite));
        for (var p = 0; p < _config.StreamProbesPerSite; p++)
        {
            var instanceIndex = Math.Min(p * stride, _config.InstancesPerSite - 1);
            var instanceName = InstanceName(instanceIndex);
            var probe = StreamSubscriberProbe.Attach(
                System, StreamManager, instanceName, $"{SiteId}-probe-{p:D2}", latency);
            _probes.Add(probe);
        }
    }

    /// <summary>
    /// Starts the site's real Kestrel/gRPC listener and attaches
    /// <paramref name="count"/> subscribers through the production
    /// <see cref="SiteStreamGrpcClient"/> — one channel per site with the streams
    /// multiplexed on it, exactly the production shape (central keys one client per
    /// <c>(siteId, endpoint)</c>). Instance selection uses the same stride as
    /// <see cref="AttachStreamProbes"/>, so the wire probes watch the SAME instances
    /// as the in-process probes and the two latency distributions are directly
    /// comparable within one run.
    /// </summary>
    /// <param name="count">Wire subscribers to attach.</param>
    /// <param name="latency">Histogram that receives end-to-end (over-the-wire) latencies.</param>
    /// <param name="log">Progress sink for fault reporting.</param>
    /// <returns>
    /// A task that completes when the listener is up and every stream has been
    /// initiated. Stream ESTABLISHMENT finishes asynchronously — comfortably inside
    /// the subscribe-settle + settle windows that separate attach from measurement.
    /// </returns>
    public async Task AttachWireStreamProbesAsync(int count, LatencyHistogram latency, Action<string> log)
    {
        _wireHost = await SiteWireStreamHost.StartAsync(this);
        _wireClient = new SiteStreamGrpcClient(_wireHost.Endpoint, NullLogger.Instance);
        _wireCts = new CancellationTokenSource();

        var stride = Math.Max(1, _config.InstancesPerSite / Math.Max(1, count));
        for (var p = 0; p < count; p++)
        {
            var instanceIndex = Math.Min(p * stride, _config.InstancesPerSite - 1);
            var subscriber = WireStreamSubscriber.Attach(
                _wireClient, InstanceName(instanceIndex), $"wire-{SiteId}-p{p:D2}",
                latency, log, _wireCts.Token);
            _wireSubscribers.Add(subscriber);
        }
    }

    /// <summary>Total tag paths registered across this site's connections.</summary>
    public int TotalTagPaths => TagPathsByConnection.Sum(list => list.Count);

    /// <summary>Stops the wire path, the actor system and the data directory in production teardown order.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    public async ValueTask DisposeAsync()
    {
        // Wire path first: cancel the client streams, wait for their loops to exit,
        // then drop the channel and the listener — the same order production teardown
        // takes (client goes away, server cancels, actor system last).
        if (_wireCts is not null)
        {
            await _wireCts.CancelAsync();
            foreach (var subscriber in _wireSubscribers)
                await subscriber.WaitForShutdownAsync();
            _wireCts.Dispose();
        }

        if (_wireClient is not null)
            await _wireClient.DisposeAsync();
        if (_wireHost is not null)
            await _wireHost.DisposeAsync();

        foreach (var probe in _probes)
            await probe.DisposeAsync();

        await StoreAndForward.StopAsync();
        await System.Terminate();
        await _localDbProvider.DisposeAsync();

        try
        {
            if (Directory.Exists(_dataDirectory))
                Directory.Delete(_dataDirectory, recursive: true);
        }
        catch (IOException)
        {
            // Best-effort cleanup of a temp directory; a lingering WAL handle is not
            // a harness failure and must not mask the measured result.
        }
    }
}