lmxopcua/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/CannedAgentClient.cs

using System.Collections.Concurrent;
using System.Runtime.CompilerServices;
using System.Threading.Channels;

namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests;

/// <summary>
///     The shared <see cref="IMTConnectAgentClient"/> test double: an Agent that serves the canned
///     <c>Fixtures/probe.xml</c> + <c>Fixtures/current.xml</c> documents, counts every call, records
///     its own disposal, and lets a test drive the <c>/sample</c> chunk sequence by hand.
/// </summary>
/// <remarks>
///     <para>
///         <b>Deterministic by construction — no timers, no sleeps, no polling.</b> The
///         <c>/sample</c> leg reads from an unbounded <see cref="Channel{T}"/> that only a test
///         fills, and <see cref="PumpAsync"/> does not return until the driver has actually consumed
///         the chunk it wrote (each chunk carries its own completion source, signalled after the
///         enumerator's <c>yield return</c> resumes). A subscription test can therefore say "one
///         chunk has now been fully processed" as a fact rather than as a timing hope.
///     </para>
///     <para>
///         <b>It honours the seam's stream-end contract</b> (see
///         <see cref="IMTConnectAgentClient.SampleAsync"/>): cancelling the token is the only way the
///         enumeration ends without throwing. Closing the scripted stream via
///         <see cref="EndStream"/> raises <see cref="MTConnectStreamEndedException"/>, exactly as the
///         production client does when an Agent drops the connection — so a pump written against the
///         fake cannot silently pass while mishandling the real one.
///     </para>
///     <para>
///         <b>Disposal is observable</b> (<see cref="DisposeCount"/>) because "did the driver
///         actually release the client?" is a behaviour, not an implementation detail: a
///         <c>ReinitializeAsync</c> that re-points the Agent but leaks the old client keeps a live
///         connection pool per re-deploy, and nothing else in the test surface would notice.
///     </para>
/// </remarks>
internal sealed class CannedAgentClient : IMTConnectAgentClient, IDisposable
{
    private readonly CancellationTokenSource _disposeCts = new();

    private readonly Lock _sampleWaitersLock = new();
    private readonly List<(int Threshold, TaskCompletionSource Waiter)> _sampleWaiters = [];

    /// <summary>Chunks a test has scripted but not yet pumped — see <see cref="ScriptChunks"/>.</summary>
    private readonly ConcurrentQueue<MTConnectStreamsResult> _script = new();

    /// <summary>
    ///     The <b>current</b> <c>/sample</c> stream generation. Replaced (not merely completed) by
    ///     <see cref="EndStream"/> so that a pump which reconnects after a dropped stream gets a
    ///     live channel to read from instead of one that is permanently closed — without that, a
    ///     reconnect test could only ever observe the reconnect failing.
    /// </summary>
    private Channel<ScriptedChunk> _chunks = Channel.CreateUnbounded<ScriptedChunk>();

    private int _probeCallCount;
    private int _currentCallCount;
    private int _sampleCallCount;
    private int _disposeCount;

    private CannedAgentClient(MTConnectProbeModel probe, MTConnectStreamsResult current)
    {
        Probe = probe;
        Current = current;
    }

    /// <summary>
    ///     Builds a client serving the repo's canned fixtures — <c>Fixtures/probe.xml</c> for
    ///     <c>/probe</c> and <c>Fixtures/current.xml</c> for <c>/current</c>.
    /// </summary>
    /// <param name="probeFixture">Probe-document fixture path, relative to the test binaries.</param>
    /// <param name="currentFixture">Streams-document fixture path, relative to the test binaries.</param>
    public static CannedAgentClient FromFixtures(
        string probeFixture = "Fixtures/probe.xml",
        string currentFixture = "Fixtures/current.xml") =>
        new(
            MTConnectProbeParser.Parse(File.ReadAllText(probeFixture)),
            MTConnectStreamsParser.Parse(File.ReadAllText(currentFixture)));

    /// <summary>Parses a streams fixture into a chunk a test can script onto the sample stream.</summary>
    /// <param name="fixture">Streams-document fixture path, relative to the test binaries.</param>
    public static MTConnectStreamsResult Chunk(string fixture) =>
        MTConnectStreamsParser.Parse(File.ReadAllText(fixture));

    /// <summary>
    ///     An Agent scripted for the ring-buffer-overflow story: the driver primes from
    ///     <c>Fixtures/current.xml</c> (<c>nextSequence</c> 108), then the first scripted chunk is
    ///     <c>Fixtures/sample-gap.xml</c> — whose <c>firstSequence</c> (5000) is strictly newer than
    ///     the cursor, i.e. the buffer rolled past the driver — and the second is an ordinary chunk
    ///     that continues contiguously from the re-baseline.
    /// </summary>
    /// <remarks>
    ///     The <b>second</b> <c>/current</c> answer is what makes this a story rather than a single
    ///     event: it advertises the buffer the gap chunk revealed (<c>firstSequence</c> 5000 /
    ///     <c>nextSequence</c> 5005), so a driver that re-baselines and resumes from that answer sees
    ///     the follow-up chunk as contiguous. A driver that re-baselined but resumed from the stale
    ///     cursor would report a gap on every subsequent chunk instead.
    /// </remarks>
    public static CannedAgentClient WithGapThenResume()
    {
        var client = FromFixtures();
        var gap = Chunk("Fixtures/sample-gap.xml");

        // 1st /current = the InitializeAsync prime (the fixture). 2nd = the post-gap re-baseline.
        client.CurrentAnswers.Enqueue(client.Current);
        client.CurrentAnswers.Enqueue(
            new MTConnectStreamsResult(gap.InstanceId, gap.NextSequence, gap.FirstSequence, gap.Observations));

        client.ScriptChunks(
            gap,
            new MTConnectStreamsResult(
                gap.InstanceId,
                gap.NextSequence + 5,
                gap.NextSequence,
                [
                    new MTConnectObservation(
                        "dev1_pos", "201.5000", new DateTime(2026, 7, 24, 12, 6, 0, DateTimeKind.Utc)),
                ]));

        return client;
    }

    /// <summary>The document <c>/probe</c> answers with. Settable so a test can re-shape the model.</summary>
    public MTConnectProbeModel Probe { get; set; }

    /// <summary>
    ///     The document <c>/current</c> answers with. Settable so a re-baseline (or an agent-restart
    ///     <c>instanceId</c> change) can be scripted between calls.
    /// </summary>
    public MTConnectStreamsResult Current { get; set; }

    /// <summary>When set, <c>/probe</c> throws this instead of answering.</summary>
    public Exception? ProbeFailure { get; set; }

    /// <summary>When set, <c>/current</c> throws this instead of answering.</summary>
    public Exception? CurrentFailure { get; set; }

    /// <summary>
    ///     Scripted <c>/current</c> answers, consumed in order: each call dequeues the next one and
    ///     <b>promotes it to <see cref="Current"/></b>, so once the script runs out the Agent keeps
    ///     answering with the last thing it said rather than travelling back in time. This is how a
    ///     test scripts "the prime, then the post-gap re-baseline" without racing the pump for a
    ///     property setter.
    /// </summary>
    public ConcurrentQueue<MTConnectStreamsResult> CurrentAnswers { get; } = new();

    /// <summary>
    ///     Scripted one-shot <c>/sample</c> failures: each enumeration dequeues one and throws it.
    ///     Models a transient stream fault the Agent recovers from — e.g. the
    ///     <see cref="InvalidDataException"/> a real cppagent's <c>OUT_OF_RANGE</c> error document
    ///     (served under HTTP 200) surfaces as.
    /// </summary>
    public ConcurrentQueue<Exception> SampleFailures { get; } = new();

    /// <summary>
    ///     A <b>sticky</b> <c>/sample</c> failure: every enumeration throws it, forever. Models a
    ///     configuration-level fault such as
    ///     <see cref="MTConnectStreamNotSupportedException"/>, where reconnecting reproduces the
    ///     identical answer — the shape a pump must NOT retry-loop against.
    /// </summary>
    public Exception? SampleFailure { get; set; }

    /// <summary>
    ///     When set, the <b>next</b> <c>/probe</c> parks here until the test completes it, then the
    ///     gate clears itself so later calls answer immediately.
    /// </summary>
    /// <remarks>
    ///     This is how a test holds a request "in flight" across a concurrent lifecycle change —
    ///     a shutdown or a re-initialize landing mid-fetch — with no timers and no races of its own.
    ///     The answer is captured when the request lands, not when it completes, so a test can
    ///     change <see cref="Probe"/> meanwhile and still tell the two documents apart.
    /// </remarks>
    public TaskCompletionSource? ProbeGate { get; set; }

    /// <summary>
    ///     When set, <c>/current</c> does not answer until this source completes — an Agent that
    ///     accepted the request and then went quiet. The wait is cancellation-observing, so the
    ///     caller's own deadline is what ends it; the test never sleeps and never races a timer it
    ///     did not set. Leave <c>null</c> for the ordinary immediate answer.
    /// </summary>
    public TaskCompletionSource? CurrentGate { get; set; }

    /// <summary>
    ///     Signalled the moment a <c>/current</c> request <b>lands</b> — before
    ///     <see cref="CurrentGate"/> parks it. This is the deterministic "the caller is now inside
    ///     the request" barrier a test needs before landing a concurrent lifecycle change on it;
    ///     without it the only alternative is polling <see cref="CurrentCallCount"/> on a timer.
    /// </summary>
    public TaskCompletionSource? CurrentEntered { get; set; }

    /// <summary>Number of <c>/probe</c> requests issued against this client.</summary>
    public int ProbeCallCount => Volatile.Read(ref _probeCallCount);

    /// <summary>Number of <c>/current</c> requests issued against this client.</summary>
    public int CurrentCallCount => Volatile.Read(ref _currentCallCount);

    /// <summary>Number of <c>/sample</c> enumerations started against this client.</summary>
    public int SampleCallCount => Volatile.Read(ref _sampleCallCount);

    /// <summary>Number of <see cref="Dispose"/> calls (not clamped — a double-dispose is visible).</summary>
    public int DisposeCount => Volatile.Read(ref _disposeCount);

    /// <summary>Whether the client has been disposed at least once.</summary>
    public bool IsDisposed => DisposeCount > 0;

    /// <summary>The <c>from</c> sequence the most recent <c>/sample</c> enumeration was opened at.</summary>
    public long? LastSampleFrom { get; private set; }

    /// <summary>
    ///     A task completing once <see cref="SampleCallCount"/> has reached
    ///     <paramref name="count"/> — the deterministic "the pump has now opened its Nth stream"
    ///     barrier, for the reconnect paths where no chunk is ever delivered and
    ///     <see cref="PumpAsync"/> therefore cannot be the barrier.
    /// </summary>
    /// <param name="count">The enumeration count to wait for.</param>
    public Task WaitForSampleCallsAsync(int count)
    {
        lock (_sampleWaitersLock)
        {
            if (Volatile.Read(ref _sampleCallCount) >= count)
            {
                return Task.CompletedTask;
            }

            var waiter = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
            _sampleWaiters.Add((count, waiter));

            return waiter.Task;
        }
    }

    /// <summary>Releases any <see cref="WaitForSampleCallsAsync"/> waiter the new call count satisfies.</summary>
    private void ReleaseSampleWaiters(int reached)
    {
        lock (_sampleWaitersLock)
        {
            for (var i = _sampleWaiters.Count - 1; i >= 0; i--)
            {
                if (_sampleWaiters[i].Threshold > reached)
                {
                    continue;
                }

                _sampleWaiters[i].Waiter.TrySetResult();
                _sampleWaiters.RemoveAt(i);
            }
        }
    }

    /// <inheritdoc/>
    public async Task<MTConnectProbeModel> ProbeAsync(CancellationToken ct)
    {
        ObjectDisposedException.ThrowIf(IsDisposed, this);
        ct.ThrowIfCancellationRequested();
        Interlocked.Increment(ref _probeCallCount);

        if (ProbeFailure is not null)
        {
            throw ProbeFailure;
        }

        // Captured now: the Agent answers with the document it held when the request landed.
        var answer = Probe;

        if (ProbeGate is { } gate)
        {
            ProbeGate = null;
            await gate.Task.WaitAsync(ct).ConfigureAwait(false);

            // A real client whose handler was disposed mid-request fails the request; so does this.
            ObjectDisposedException.ThrowIf(IsDisposed, this);
        }

        return answer;
    }

    /// <inheritdoc/>
    public async Task<MTConnectStreamsResult> CurrentAsync(CancellationToken ct)
    {
        ObjectDisposedException.ThrowIf(IsDisposed, this);
        ct.ThrowIfCancellationRequested();
        Interlocked.Increment(ref _currentCallCount);
        CurrentEntered?.TrySetResult();

        // The request was accepted; the answer is withheld until the test releases the gate or the
        // caller's deadline cancels the token.
        var gate = CurrentGate;
        if (gate is not null)
        {
            await gate.Task.WaitAsync(ct).ConfigureAwait(false);
        }

        // A scripted answer supersedes — and then becomes — the standing one.
        if (CurrentAnswers.TryDequeue(out var scripted))
        {
            Current = scripted;
        }

        return CurrentFailure is null ? Current : throw CurrentFailure;
    }

    /// <inheritdoc/>
    public async IAsyncEnumerable<MTConnectStreamsResult> SampleAsync(
        long from, [EnumeratorCancellation] CancellationToken ct)
    {
        ObjectDisposedException.ThrowIf(IsDisposed, this);
        ReleaseSampleWaiters(Interlocked.Increment(ref _sampleCallCount));
        LastSampleFrom = from;

        // Thrown before the first chunk, exactly like a real client that fails its /sample request:
        // one-shot scripts first, then the sticky "this endpoint will never stream" failure.
        if (SampleFailures.TryDequeue(out var oneShot))
        {
            throw oneShot;
        }

        if (SampleFailure is { } sticky)
        {
            throw sticky;
        }

        // Bound to THIS generation: EndStream swaps in a fresh channel, so a consumer that
        // reconnects reads the new one rather than the closed one it just lost.
        var chunks = Volatile.Read(ref _chunks);
        using var lifetime = CancellationTokenSource.CreateLinkedTokenSource(ct, _disposeCts.Token);
        var delivered = 0L;

        while (true)
        {
            ScriptedChunk scripted;
            try
            {
                scripted = await chunks.Reader.ReadAsync(lifetime.Token).ConfigureAwait(false);
            }
            catch (ChannelClosedException)
            {
                // Mirrors the production client: a stream that stops for any reason other than the
                // caller cancelling is an exception, never a quiet end of enumeration.
                throw new MTConnectStreamEndedException(
                    MTConnectStreamEndReason.ConnectionClosed, "canned://agent", delivered);
            }

            delivered++;

            try
            {
                yield return scripted.Result;
            }
            finally
            {
                // Signalled when the consumer is DONE with this chunk — whether it came back for the
                // next one or abandoned the enumeration. The finally is load-bearing: a chunk that
                // makes the pump break out (a sequence gap, an agent restart) is never resumed past,
                // so signalling only after the resume would hang every re-baseline test.
                scripted.Consumed.TrySetResult();
            }
        }
    }

    /// <summary>
    ///     Queues a chunk onto the scripted <c>/sample</c> stream and waits until the consumer has
    ///     processed it. This is the deterministic replacement for "wait a bit and hope the pump
    ///     ran".
    /// </summary>
    /// <param name="chunk">The chunk the Agent should send next.</param>
    /// <returns>A task completing once the enumerating consumer has moved past <paramref name="chunk"/>.</returns>
    public Task PumpAsync(MTConnectStreamsResult chunk)
    {
        ArgumentNullException.ThrowIfNull(chunk);

        var scripted = new ScriptedChunk(
            chunk, new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously));

        if (!Volatile.Read(ref _chunks).Writer.TryWrite(scripted))
        {
            throw new InvalidOperationException("The scripted /sample stream is already closed.");
        }

        return scripted.Consumed.Task;
    }

    /// <summary>
    ///     Queues chunks for <see cref="PumpOnce"/> to deliver one at a time. Nothing is sent until
    ///     a test asks for it — the script is a plan, not a schedule.
    /// </summary>
    /// <param name="chunks">The chunks the Agent should send, in order.</param>
    public void ScriptChunks(params MTConnectStreamsResult[] chunks)
    {
        ArgumentNullException.ThrowIfNull(chunks);

        foreach (var chunk in chunks)
        {
            _script.Enqueue(chunk);
        }
    }

    /// <summary>
    ///     Pumps the next scripted chunk (see <see cref="ScriptChunks"/>) and waits until the
    ///     consumer has finished with it.
    /// </summary>
    /// <returns>A task completing once the consumer has processed — or abandoned the stream on — that chunk.</returns>
    /// <exception cref="InvalidOperationException">The script is exhausted.</exception>
    public Task PumpOnce() =>
        _script.TryDequeue(out var next)
            ? PumpAsync(next)
            : throw new InvalidOperationException(
                "No scripted /sample chunk left to pump; call ScriptChunks first.");

    /// <summary>
    ///     Closes the current scripted <c>/sample</c> stream, which surfaces to its consumer as an
    ///     <see cref="MTConnectStreamEndedException"/> — the transient, reconnectable end — and opens
    ///     a fresh one so a reconnecting consumer has somewhere to land.
    /// </summary>
    public void EndStream()
    {
        var closed = Interlocked.Exchange(ref _chunks, Channel.CreateUnbounded<ScriptedChunk>());
        closed.Writer.TryComplete();
    }

    /// <inheritdoc/>
    /// <remarks>
    ///     <para>
    ///         <b>The first-disposer test uses the value <see cref="Interlocked.Increment(ref int)"/>
    ///         returned</b>, not a re-read of the counter. Re-reading is a race: two concurrent
    ///         disposes can both increment and then both read 2, so BOTH take the early return and
    ///         the stream is never closed — a subscription test would then hang waiting for a
    ///         teardown that silently did nothing.
    ///     </para>
    ///     <para>
    ///         <b><see cref="_disposeCts"/> is cancelled but deliberately NOT disposed.</b> A
    ///         <c>/sample</c> enumeration that is starting concurrently reaches
    ///         <c>CreateLinkedTokenSource(ct, _disposeCts.Token)</c>, and reading <c>.Token</c> on a
    ///         disposed source throws <see cref="ObjectDisposedException"/> — a flake that would
    ///         surface as a random unrelated failure in whichever test happened to lose the race.
    ///         Leaking one cancelled source per fake is free; a cancelled source needs no disposal
    ///         to release anything a test cares about.
    ///     </para>
    /// </remarks>
    public void Dispose()
    {
        if (Interlocked.Increment(ref _disposeCount) > 1)
        {
            return;
        }

        _disposeCts.Cancel();
        Volatile.Read(ref _chunks).Writer.TryComplete();
    }

    private sealed record ScriptedChunk(MTConnectStreamsResult Result, TaskCompletionSource Consumed);
}