junction/src/Junction.Core/Monitoring/MachineMonitor.cs

using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using System.Collections.Concurrent;
using Junction.Core.Plugins;
using Junction.Core.Polling;
using Junction.Domain;
using Junction.Domain.Models;
using Junction.Domain.Persistence;
using Junction.Domain.Protocols;
using Microsoft.Extensions.Logging;

namespace Junction.Core.Monitoring
{
    /// <summary>
    /// Default <see cref="IMachineMonitor"/>. See the interface for the orchestration contract
    /// and partial-start policy.
    /// <para>
    /// PollingEngine-per-machine: each machine gets its own <see cref="IPollingEngine"/> built
    /// from an injected <see cref="Func{IPollingEngine}"/> factory. The factory keeps the
    /// monitor decoupled from engine construction (and from <see cref="ILogger{PollingEngine}"/>
    /// wiring), which makes the monitor trivially unit-testable with a stub engine. DI registers
    /// the default factory as <c>() =&gt; new PollingEngine(loggerFactory.CreateLogger&lt;PollingEngine&gt;())</c>.
    /// </para>
    /// <para>
    /// Thread-safety: snapshots arrive concurrently from multiple poll loops. The latest-snapshot
    /// cache is a <see cref="ConcurrentDictionary{Guid, MachineSnapshot}"/> (lock-free reads for
    /// the dashboard), and the <see cref="SnapshotUpdated"/> event is raised through a captured
    /// local delegate. Start/Stop mutate the running-loop set under a private lock.
    /// </para>
    /// </summary>
    public sealed class MachineMonitor : IMachineMonitor
    {
        private const string Source = "MachineMonitor";

        private readonly IMachineRepository _repository;
        private readonly IPluginLoader _pluginLoader;
        private readonly Func<IPollingEngine> _engineFactory;
        private readonly ILogger<MachineMonitor> _logger;

        private readonly ConcurrentDictionary<Guid, MachineSnapshot> _snapshots =
            new ConcurrentDictionary<Guid, MachineSnapshot>();

        private readonly object _lifecycleLock = new object();
        private readonly List<RunningLoop> _running = new List<RunningLoop>();

        // Built once by StartAsync and kept so dynamic add/update can resolve drivers at runtime.
        // Guarded by _lifecycleLock. Null until StartAsync completes = "monitor not started".
        private Dictionary<string, IProtocolDriverFactory>? _factories;

        public MachineMonitor(
            IMachineRepository repository,
            IPluginLoader pluginLoader,
            Func<IPollingEngine> engineFactory,
            ILogger<MachineMonitor> logger)
        {
            _repository = repository ?? throw new ArgumentNullException(nameof(repository));
            _pluginLoader = pluginLoader ?? throw new ArgumentNullException(nameof(pluginLoader));
            _engineFactory = engineFactory ?? throw new ArgumentNullException(nameof(engineFactory));
            _logger = logger ?? throw new ArgumentNullException(nameof(logger));
        }

        /// <inheritdoc />
        public event EventHandler<MachineSnapshot>? SnapshotUpdated;

        /// <inheritdoc />
        public IReadOnlyDictionary<Guid, MachineSnapshot> LatestSnapshots => _snapshots;

        /// <inheritdoc />
        public IReadOnlyCollection<string> AvailableProtocols
        {
            get
            {
                lock (_lifecycleLock)
                {
                    if (_factories is null)
                    {
                        return Array.Empty<string>();
                    }

                    // Independent snapshot of the keys so callers can't observe later mutations.
                    return new List<string>(_factories.Keys);
                }
            }
        }

        /// <inheritdoc />
        public async Task<Result> StartAsync(string pluginsDirectory, CancellationToken cancellationToken)
        {
            if (cancellationToken.IsCancellationRequested)
            {
                return Result.Cancelled();
            }

            // 1. Load plugins. A failure here (missing/invalid directory) is a hard infra failure.
            Result<IReadOnlyList<LoadedPlugin>> loadResult = _pluginLoader.LoadFrom(pluginsDirectory);
            if (!loadResult.IsSuccess)
            {
                _logger.LogError("Plugin load failed; monitor cannot start: {Errors}", DescribeErrors(loadResult.Errors));
                return Result.Fail(loadResult.Errors);
            }

            var factories = BuildFactoryMap(loadResult.Value);

            // 2. Load configured machines. A repository failure is a hard infra failure.
            Result<IReadOnlyList<Machine>> machinesResult;
            try
            {
                machinesResult = await _repository.GetAllAsync(cancellationToken).ConfigureAwait(false);
            }
            catch (OperationCanceledException)
            {
                return Result.Cancelled();
            }

            if (machinesResult.WasCancelled)
            {
                return Result.Cancelled();
            }

            if (!machinesResult.IsSuccess)
            {
                _logger.LogError("Machine load failed; monitor cannot start: {Errors}", DescribeErrors(machinesResult.Errors));
                return Result.Fail(machinesResult.Errors);
            }

            // 3. Start one poll loop per machine that resolves a driver. Skips are non-fatal.
            int started = 0;
            int skipped = 0;

            lock (_lifecycleLock)
            {
                // Retain the factory map so AddOrUpdateMachineAsync can build drivers later.
                _factories = factories;

                foreach (Machine machine in machinesResult.Value)
                {
                    if (!factories.TryGetValue(machine.ProtocolId, out IProtocolDriverFactory factory))
                    {
                        skipped++;
                        _logger.LogWarning(
                            "No plugin loaded for protocol '{ProtocolId}'; skipping machine {MachineId} ({MachineName}).",
                            machine.ProtocolId, machine.Id, machine.Name);
                        continue;
                    }

                    Result<IProtocolDriver> driverResult = factory.Create(machine);
                    if (!driverResult.IsSuccess)
                    {
                        skipped++;
                        _logger.LogWarning(
                            "Driver creation failed for machine {MachineId} ({MachineName}) via protocol '{ProtocolId}'; skipping: {Errors}",
                            machine.Id, machine.Name, machine.ProtocolId, DescribeErrors(driverResult.Errors));
                        continue;
                    }

                    IProtocolDriver driver = driverResult.Value;
                    var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
                    IPollingEngine engine = _engineFactory();

                    CancellationToken loopToken = cts.Token;
                    Task loop = engine.RunAsync(
                        machine,
                        driver,
                        snapshot => OnSnapshot(snapshot, loopToken),
                        loopToken);

                    _running.Add(new RunningLoop(machine.Id, cts, loop));
                    started++;
                }
            }

            _logger.LogInformation(
                "MachineMonitor started: {Started} machine(s) polling, {Skipped} skipped.",
                started, skipped);

            // Partial start is a success.
            return Result.Ok();
        }

        /// <inheritdoc />
        public async Task StopAsync()
        {
            List<RunningLoop> loops;
            lock (_lifecycleLock)
            {
                loops = new List<RunningLoop>(_running);
                _running.Clear();
            }

            if (loops.Count == 0)
            {
                return;
            }

            foreach (RunningLoop loop in loops)
            {
                try
                {
                    loop.Cts.Cancel();
                }
                catch (ObjectDisposedException)
                {
                    // Already disposed; ignore.
                }
            }

            var tasks = new Task[loops.Count];
            for (int i = 0; i < loops.Count; i++)
            {
                tasks[i] = loops[i].Loop;
            }

            try
            {
                // Poll loops never fault (they swallow driver/cancellation errors), but guard anyway.
                await Task.WhenAll(tasks).ConfigureAwait(false);
            }
            catch (Exception ex)
            {
                _logger.LogWarning(ex, "A poll loop faulted during shutdown; continuing cleanup.");
            }

            foreach (RunningLoop loop in loops)
            {
                loop.Cts.Dispose();
            }

            _logger.LogInformation("MachineMonitor stopped: {Count} poll loop(s) shut down.", loops.Count);
        }

        /// <inheritdoc />
        public async Task<Result> AddOrUpdateMachineAsync(Machine machine, CancellationToken cancellationToken)
        {
            if (machine is null)
            {
                return Result.Fail(OperationError.Of(Source, "machine is null."));
            }

            if (cancellationToken.IsCancellationRequested)
            {
                return Result.Cancelled();
            }

            // Phase 1: detach any existing loop for this id (under lock), enforcing "not started".
            RunningLoop? existing;
            lock (_lifecycleLock)
            {
                if (_factories is null)
                {
                    return Result.Fail(OperationError.Of(Source, "Monitor not started; call StartAsync first."));
                }

                existing = FindAndRemoveLocked(machine.Id);
            }

            // Phase 2: await the old loop's clean shutdown OUTSIDE the lock (no await under lock).
            if (existing != null)
            {
                await StopLoopAsync(existing).ConfigureAwait(false);
                _snapshots.TryRemove(machine.Id, out _);
            }

            // Phase 3: start a fresh loop (under lock).
            lock (_lifecycleLock)
            {
                if (_factories is null)
                {
                    return Result.Fail(OperationError.Of(Source, "Monitor not started; call StartAsync first."));
                }

                var cts = new CancellationTokenSource();
                Result started = TryStartLoopLocked(machine, cts);
                if (!started.IsSuccess)
                {
                    cts.Dispose();
                    _logger.LogWarning(
                        "AddOrUpdate failed to start machine {MachineId} ({MachineName}): {Errors}",
                        machine.Id, machine.Name, DescribeErrors(started.Errors));
                    return started;
                }

                _logger.LogInformation(
                    "Machine {MachineId} ({MachineName}) started/restarted via AddOrUpdate.",
                    machine.Id, machine.Name);
                return Result.Ok();
            }
        }

        /// <inheritdoc />
        public async Task<Result> RemoveMachineAsync(Guid machineId, CancellationToken cancellationToken)
        {
            RunningLoop? existing;
            lock (_lifecycleLock)
            {
                existing = FindAndRemoveLocked(machineId);
            }

            if (existing != null)
            {
                await StopLoopAsync(existing).ConfigureAwait(false);
            }

            _snapshots.TryRemove(machineId, out _);
            _logger.LogInformation("Machine {MachineId} removed from monitor (was running: {Running}).", machineId, existing != null);

            // Idempotent: Ok even when nothing was running for the id.
            return Result.Ok();
        }

        /// <summary>
        /// Removes and returns the tracked loop for <paramref name="machineId"/>, or null.
        /// Caller must hold <see cref="_lifecycleLock"/>.
        /// </summary>
        private RunningLoop? FindAndRemoveLocked(Guid machineId)
        {
            for (int i = 0; i < _running.Count; i++)
            {
                if (_running[i].MachineId == machineId)
                {
                    RunningLoop loop = _running[i];
                    _running.RemoveAt(i);
                    return loop;
                }
            }

            return null;
        }

        /// <summary>Cancels, awaits and disposes a single loop. Never throws.</summary>
        private async Task StopLoopAsync(RunningLoop loop)
        {
            try
            {
                loop.Cts.Cancel();
            }
            catch (ObjectDisposedException)
            {
                // Already disposed; ignore.
            }

            try
            {
                await loop.Loop.ConfigureAwait(false);
            }
            catch (Exception ex)
            {
                _logger.LogWarning(ex, "Poll loop faulted while stopping machine {MachineId}; continuing.", loop.MachineId);
            }

            loop.Cts.Dispose();
        }

        /// <summary>
        /// Resolves a driver for <paramref name="machine"/> via the retained factory map and starts
        /// a poll loop tracked in <see cref="_running"/>. Caller must hold <see cref="_lifecycleLock"/>
        /// and have verified <see cref="_factories"/> is non-null. Returns Fail (unknown protocol /
        /// driver-create failure) without adding a loop.
        /// </summary>
        private Result TryStartLoopLocked(Machine machine, CancellationTokenSource cts)
        {
            Dictionary<string, IProtocolDriverFactory> factories = _factories!;

            if (!factories.TryGetValue(machine.ProtocolId, out IProtocolDriverFactory factory))
            {
                return Result.Fail(OperationError.Of(
                    Source,
                    $"No plugin loaded for protocol '{machine.ProtocolId}'."));
            }

            Result<IProtocolDriver> driverResult = factory.Create(machine);
            if (!driverResult.IsSuccess)
            {
                return Result.Fail(driverResult.Errors);
            }

            IProtocolDriver driver = driverResult.Value;
            IPollingEngine engine = _engineFactory();
            CancellationToken loopToken = cts.Token;
            Task loop = engine.RunAsync(
                machine,
                driver,
                snapshot => OnSnapshot(snapshot, loopToken),
                loopToken);

            _running.Add(new RunningLoop(machine.Id, cts, loop));
            return Result.Ok();
        }

        /// <summary>
        /// Snapshot callback invoked from a poll loop: updates the latest cache, raises the
        /// event, then fires the persistence write off (fire-and-forget with error logging) so
        /// a slow/failed save never stalls or breaks the poll loop.
        /// </summary>
        private void OnSnapshot(MachineSnapshot snapshot, CancellationToken cancellationToken)
        {
            _snapshots[snapshot.MachineId] = snapshot;

            EventHandler<MachineSnapshot>? handler = SnapshotUpdated;
            if (handler != null)
            {
                try
                {
                    handler(this, snapshot);
                }
                catch (Exception ex)
                {
                    _logger.LogWarning(ex, "SnapshotUpdated subscriber threw for machine {MachineId}; continuing.", snapshot.MachineId);
                }
            }

            _ = PersistAsync(snapshot, cancellationToken);
        }

        private async Task PersistAsync(MachineSnapshot snapshot, CancellationToken cancellationToken)
        {
            try
            {
                Result result = await _repository.SaveSnapshotAsync(snapshot, cancellationToken).ConfigureAwait(false);
                if (!result.IsSuccess && !result.WasCancelled)
                {
                    _logger.LogWarning(
                        "Failed to persist snapshot for machine {MachineId}: {Errors}",
                        snapshot.MachineId, DescribeErrors(result.Errors));
                }
            }
            catch (OperationCanceledException)
            {
                // Shutdown in progress; nothing to do.
            }
            catch (Exception ex)
            {
                _logger.LogWarning(ex, "Persisting snapshot threw for machine {MachineId}; continuing.", snapshot.MachineId);
            }
        }

        /// <summary>
        /// Maps ProtocolId (case-insensitive) to factory. On duplicate protocol ids the first
        /// loaded plugin wins; the collision is logged.
        /// </summary>
        private Dictionary<string, IProtocolDriverFactory> BuildFactoryMap(IReadOnlyList<LoadedPlugin> plugins)
        {
            var map = new Dictionary<string, IProtocolDriverFactory>(StringComparer.OrdinalIgnoreCase);
            foreach (LoadedPlugin plugin in plugins)
            {
                string protocolId = plugin.Factory.ProtocolId ?? "";
                if (map.ContainsKey(protocolId))
                {
                    _logger.LogWarning("Duplicate plugin for protocol '{ProtocolId}'; keeping the first loaded.", protocolId);
                    continue;
                }

                map[protocolId] = plugin.Factory;
            }

            return map;
        }

        private static string DescribeErrors(IReadOnlyList<OperationError> errors)
        {
            if (errors is null || errors.Count == 0)
            {
                return "(no detail)";
            }

            if (errors.Count == 1)
            {
                return errors[0].ToString();
            }

            var parts = new string[errors.Count];
            for (int i = 0; i < errors.Count; i++)
            {
                parts[i] = errors[i].ToString();
            }

            return string.Join("; ", parts);
        }

        /// <summary>A running per-machine poll loop with its cancellation source.</summary>
        private sealed class RunningLoop
        {
            public Guid MachineId { get; }
            public CancellationTokenSource Cts { get; }
            public Task Loop { get; }

            public RunningLoop(Guid machineId, CancellationTokenSource cts, Task loop)
            {
                MachineId = machineId;
                Cts = cts;
                Loop = loop;
            }
        }
    }
}