using Akka.Actor;
using Akka.TestKit.Xunit2;
using Microsoft.CodeAnalysis.CSharp.Scripting;
using Microsoft.CodeAnalysis.Scripting;
using Microsoft.Extensions.Logging.Abstractions;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.ScriptExecution;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Flattening;
using ZB.MOM.WW.ScadaBridge.HealthMonitoring;
using ZB.MOM.WW.ScadaBridge.SiteRuntime.Actors;
using ZB.MOM.WW.ScadaBridge.SiteRuntime.Scripts;
using ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests.TestSupport;

namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests.Actors;

/// <summary>
/// The per-script in-flight cap (<see cref="SiteRuntimeOptions.MaxConcurrentRunsPerScript"/>)
/// and the recursion limit (<see cref="SiteRuntimeOptions.MaxScriptCallDepth"/>) govern two
/// different things, and the cap must not usurp the recursion limit's job.
///
/// <para>A run awaiting a nested <c>CallScript</c> still holds its in-flight slot — the slot is
/// released only by <c>ScriptExecutionCompleted</c>, sent after the body returns — and a script
/// that calls ITSELF routes the nested request back to the SAME <see cref="ScriptActor"/>. So
/// counting nested launches against the cap made self-recursion fail at depth 4 (cap) instead
/// of depth 10 (<c>MaxScriptCallDepth</c>), reported as a misleading "shed" with a spurious shed
/// counter, and left the documented recursion-limit path — the site event emitted by
/// <c>ScriptRuntimeContext.CallScript</c> — unreachable.</para>
///
/// <para>The cap now applies to <c>callDepth == 0</c> launches only: trigger-driven runs and
/// depth-0 Ask calls, i.e. genuinely NEW work. Both halves are pinned below.</para>
/// </summary>
public class ScriptRecursionVsRunCapTests : TestKit, IDisposable
{
    private readonly SharedScriptLibrary _sharedLibrary;
    private readonly ScriptExecutionScheduler _scheduler = new(8);

    /// <summary>Initializes the shared script library and resets the per-test hooks.</summary>
    public ScriptRecursionVsRunCapTests()
    {
        var compilationService = new ScriptCompilationService(
            NullLogger<ScriptCompilationService>.Instance);
        _sharedLibrary = new SharedScriptLibrary(
            compilationService, NullLogger<SharedScriptLibrary>.Instance);
        RecursionHooks.Reset();
    }

    void IDisposable.Dispose()
    {
        RecursionHooks.Gate.Release(64);
        Shutdown();
        _scheduler.Dispose();
    }

    private static Script<object?> CompileRaw(string code)
    {
        var options = ScriptOptions.Default
            .WithReferences(typeof(object).Assembly, typeof(Enumerable).Assembly,
                typeof(RecursionHooks).Assembly)
            .WithImports("System", "System.Collections.Generic", "System.Linq", "System.Threading.Tasks");
        var script = CSharpScript.Create<object?>(code, options, typeof(ScriptGlobals));
        script.Compile();
        return script;
    }

    private static SiteRuntimeOptions Options(int maxCallDepth) => new()
    {
        MaxConcurrentRunsPerScript = 4,
        MaxScriptCallDepth = maxCallDepth,
        // Long enough that nothing times out inside the test window — the depth limit, not a
        // deadline and not the cap, must be what stops the recursion.
        ScriptExecutionTimeoutSeconds = 120,
        StuckScriptGraceMs = 120_000
    };

    /// <summary>
    /// End-to-end self-recursion through the real actor path: the script calls itself, the
    /// nested <c>ScriptCallRequest</c> is routed back to the same actor by a stand-in instance
    /// actor, and the chain must run all the way to <c>MaxScriptCallDepth</c> and then be
    /// stopped by the recursion limit — with its site event — never by the concurrency cap.
    /// </summary>
    [Fact]
    public void SelfRecursion_ReachesMaxScriptCallDepth_AndStopsAtTheRecursionLimit_NotAShed()
    {
        const int maxCallDepth = 6;   // > MaxConcurrentRunsPerScript (4): the whole point
        var siteLog = new FakeSiteEventLogger();
        var health = new SiteHealthCollector();
        var options = Options(maxCallDepth);

        // Stand-in Instance Actor: routes ScriptCallRequest straight back to the one script
        // actor, which is exactly what an InstanceActor does for a same-instance CallScript —
        // and makes the nested request land on the SAME actor whose cap is under test.
        var router = ActorOf(Props.Create(() => new SelfCallRouter()), "router-" + Guid.NewGuid().ToString("N"));

        var actor = ActorOf(
            Props.Create(() => new ScriptActor(
                "Self", "Inst1", router,
                CompileRaw(
                    "ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests.Actors.RecursionHooks.Enter();" +
                    "await Instance.CallScript(\"Self\");" +
                    "return null;"),
                new ResolvedScript { CanonicalName = "Self", TriggerType = "Call" },
                _sharedLibrary, options, NullLogger<ScriptActor>.Instance,
                null, null, health, new SingleServiceProvider(siteLog), _scheduler, null)),
            "self-" + Guid.NewGuid().ToString("N"));

        router.Tell(new SelfCallRouter.SetTarget(actor));

        var caller = CreateTestProbe();
        actor.Tell(new ScriptCallRequest("Self", null, 0, "corr-root"), caller.Ref);

        // The whole chain unwinds back to the root caller: the deepest call is refused by the
        // recursion limit, that failure propagates up through each awaiting CallScript.
        var result = caller.ExpectMsg<ScriptCallResult>(TimeSpan.FromSeconds(30));
        Assert.False(result.Success);
        Assert.Equal("corr-root", result.CorrelationId);
        // NOT a shed — the reply carries the depth diagnosis, which is the actionable one.
        Assert.DoesNotContain("shed", result.ErrorMessage, StringComparison.OrdinalIgnoreCase);

        // Root run + one run per nesting level: recursion went the full documented distance
        // instead of stopping at the cap (which would have given 4).
        Assert.Equal(maxCallDepth + 1, RecursionHooks.Runs);

        AwaitAssert(() =>
        {
            var scriptEvents = siteLog.OfType("script");

            // The recursion-limit path is REACHABLE: its Error site event is emitted by
            // ScriptRuntimeContext, sourced at the INSTANCE (the per-run failure events the
            // unwind also produces are sourced at the script actor and merely quote it), and
            // it fires exactly once — at the bottom of the chain.
            var limitEvent = Assert.Single(
                scriptEvents, r => r.Source == "InstanceScript:Inst1");
            Assert.Equal("Error", limitEvent.Severity);
            Assert.StartsWith("Script call depth exceeded", limitEvent.Message);
            Assert.Contains($"maximum of {maxCallDepth}", limitEvent.Message);
            Assert.Contains($"rejected at depth {maxCallDepth + 1}", limitEvent.Message);

            // …and nothing was shed on the way there.
            Assert.DoesNotContain(scriptEvents, r => r.Message.Contains("shed", StringComparison.OrdinalIgnoreCase));
        }, TimeSpan.FromSeconds(10));

        Assert.Equal(0, health.CollectReport("site-1").ScriptRunShedCount);
    }

    /// <summary>
    /// The exemption is scoped to nesting: with the cap already full of depth-0 runs, another
    /// depth-0 request is still shed, while a nested (<c>callDepth &gt; 0</c>) request is
    /// launched. This is the discriminating pin — a blanket "skip the cap" would fail the first
    /// half, and the pre-fix behaviour fails the second.
    /// </summary>
    [Fact]
    public void WithTheCapFull_DepthZeroIsStillShed_ButANestedCallIsLaunched()
    {
        var siteLog = new FakeSiteEventLogger();
        var health = new SiteHealthCollector();
        var instance = CreateTestProbe().Ref;
        var options = Options(maxCallDepth: 10);

        var actor = ActorOfAsTestActorRef<ScriptActor>(
            Props.Create(() => new ScriptActor(
                "Hot", "Inst1", instance,
                CompileRaw("ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests.Actors.RecursionHooks.Gate.Wait(); return null;"),
                new ResolvedScript { CanonicalName = "Hot", TriggerType = "Call" },
                _sharedLibrary, options, NullLogger<ScriptActor>.Instance,
                null, null, health, new SingleServiceProvider(siteLog), _scheduler, null)),
            "cap-" + Guid.NewGuid().ToString("N"));

        // Fill the cap with four depth-0 runs, all blocked in their bodies.
        for (var i = 0; i < 4; i++)
            actor.Tell(new ScriptCallRequest("Hot", null, 0, $"corr-{i}"), ActorRefs.NoSender);

        AwaitAssert(() =>
        {
            Assert.Equal(4, actor.UnderlyingActor.RunsInFlight);
            Assert.Equal(4, _scheduler.BusyThreadCount);
        }, TimeSpan.FromSeconds(15));

        // Depth 0 — new work — is still refused.
        var newWork = CreateTestProbe();
        actor.Tell(new ScriptCallRequest("Hot", null, 0, "corr-depth0"), newWork.Ref);
        var shed = newWork.ExpectMsg<ScriptCallResult>(TimeSpan.FromSeconds(10));
        Assert.False(shed.Success);
        Assert.Contains("shed", shed.ErrorMessage, StringComparison.OrdinalIgnoreCase);
        Assert.Equal(4, actor.UnderlyingActor.RunsInFlight);

        // Depth 1 — a nested call, bounded by MaxScriptCallDepth instead — is launched.
        var nested = CreateTestProbe();
        actor.Tell(new ScriptCallRequest("Hot", null, 1, "corr-depth1"), nested.Ref);

        AwaitAssert(() =>
        {
            Assert.Equal(5, actor.UnderlyingActor.RunsInFlight);
            Assert.Equal(5, _scheduler.BusyThreadCount);
        }, TimeSpan.FromSeconds(15));

        // It is running, not answered: no shed reply reached the nested caller.
        nested.ExpectNoMsg(TimeSpan.FromMilliseconds(500));

        // Exactly the one depth-0 refusal was counted.
        Assert.Equal(1, health.CollectReport("site-1").ScriptRunShedCount);
    }

    /// <summary>
    /// Stand-in Instance Actor that forwards every message to one script actor, so a script's
    /// nested <c>CallScript</c> Ask is routed back to itself (self-recursion) with the original
    /// Ask sender preserved.
    /// </summary>
    private sealed class SelfCallRouter : ReceiveActor
    {
        /// <summary>Sets the actor every subsequent message is forwarded to.</summary>
        /// <param name="Target">The script actor to forward to.</param>
        public sealed record SetTarget(IActorRef Target);

        private IActorRef? _target;

        /// <summary>Initializes the router with no target until <see cref="SetTarget"/> arrives.</summary>
        public SelfCallRouter()
        {
            Receive<SetTarget>(m => _target = m.Target);
            ReceiveAny(msg => _target?.Forward(msg));
        }
    }
}

/// <summary>Test hooks for the recursion-versus-cap tests.</summary>
public static class RecursionHooks
{
    private static int _runs;

    /// <summary>Gate the blocking cap-filling script waits on; reset per test.</summary>
    public static SemaphoreSlim Gate = new(0);

    /// <summary>Number of recursive script bodies that started, across the whole chain.</summary>
    public static int Runs => Volatile.Read(ref _runs);

    /// <summary>Called at the top of each recursive script body.</summary>
    public static void Enter() => Interlocked.Increment(ref _runs);

    /// <summary>Resets the hooks between tests.</summary>
    public static void Reset()
    {
        Interlocked.Exchange(ref _runs, 0);
        Gate = new SemaphoreSlim(0);
    }
}