lmxopcua/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/PlainMqttLiveTests.cs

using System.Security.Cryptography;
using System.Security.Cryptography.X509Certificates;
using System.Text;
using MQTTnet;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;

namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests;

/// <summary>
///     Live validation of the plain-MQTT (P1) driver against the real Mosquitto fixture in
///     <c>Docker/docker-compose.yml</c> — auth + TLS on <c>:8883</c>, authenticated plaintext on
///     <c>:1883</c>, and a publisher emitting retained JSON. Every component exercised here is the
///     production one: <see cref="MqttConnection"/> (broker session, TLS, CA pin),
///     <see cref="MqttSubscriptionManager"/> (SUBSCRIBE, routing, extraction) and
///     <see cref="MqttDriver"/> (the <c>ISubscribable</c>/<c>IReadable</c> surface).
///     <para>
///         <b>Env-gated + skip-clean.</b> Without <c>MQTT_FIXTURE_ENDPOINT</c> (and the credentials)
///         every test calls <c>Assert.Skip</c>, so <c>dotnet test</c> is green on a laptop with no
///         Docker. See <see cref="MqttFixture"/> for the full env-var list.
///     </para>
///     <para>
///         <b>Why these four things.</b> The TLS/auth legs are the first time the Task-3 handshake
///         code meets a real broker — until now the CA-pin path had only ever seen synthetic certs,
///         and its failure mode (a pin that accepts everything) is invisible without a negative
///         control, which is why <see cref="Tls_connect_pinned_to_a_foreign_ca_is_rejected"/> exists
///         alongside the positive. The subscribe leg asserts the published <b>RawPath</b>, the single
///         most defect-prone identity in this plan. And the unauthored-topic leg is written so it
///         cannot pass vacuously — it proves an authored tag was flowing throughout the window in
///         which the unauthored topic produced nothing.
///     </para>
/// </summary>
[Collection(MqttFixtureCollection.Name)]
[Trait("Category", "LiveIntegration")]
public sealed class PlainMqttLiveTests(MqttFixture fixture)
{
    /// <summary>OPC UA <c>BadWaitingForInitialData</c> — "no value has ever been observed".</summary>
    private const uint BadWaitingForInitialData = 0x80320000u;

    /// <summary>OPC UA <c>BadNodeIdUnknown</c> — "that reference is not an authored MQTT tag".</summary>
    private const uint BadNodeIdUnknown = 0x80340000u;

    /// <summary>OPC UA <c>Good</c>.</summary>
    private const uint Good = 0x00000000u;

    /// <summary>
    ///     Bound on every wait for broker traffic. The fixture publisher emits every 2s by default,
    ///     so 30s is ~15 publish cycles — generous enough to absorb a slow container start, short
    ///     enough that a genuinely dark subscription fails rather than hangs (the plan's per-op
    ///     bounded-deadline rule).
    /// </summary>
    private static readonly TimeSpan MessageTimeout = TimeSpan.FromSeconds(30);

    private readonly MqttFixture _fx = fixture;

    // -----------------------------------------------------------------------
    // TLS + auth
    // -----------------------------------------------------------------------

    /// <summary>
    ///     Connects over the TLS listener with real credentials and asserts the session establishes.
    ///     Pins to the fixture CA when <c>MQTT_FIXTURE_CA_CERT</c> is supplied — so on a fully
    ///     provisioned rig this is simultaneously the CA-pin happy path — and falls back to
    ///     <c>AllowUntrustedServerCertificate</c> otherwise so the auth leg still runs.
    /// </summary>
    [Fact]
    public async Task Tls_auth_connect_succeeds()
    {
        SkipIfUnconfigured();

        var ct = TestContext.Current.CancellationToken;
        await using var connection = new MqttConnection(TlsOptions(), "mqtt-live-tls");

        await connection.ConnectAsync(ct);

        connection.IsConnected.ShouldBeTrue("a TLS+auth connect against the fixture broker must establish a session");
        connection.State.ShouldBe(MqttConnectionState.Connected);

        TestContext.Current.SendDiagnosticMessage(
            $"TLS+auth connect to {_fx.TlsHost}:{_fx.TlsPort} succeeded "
            + $"(ca pin: {_fx.CaCertificatePath ?? "<none — AllowUntrustedServerCertificate>"}).");
    }

    /// <summary>
    ///     The auth negative, and the assertion that gives every other test in this file its meaning:
    ///     a wrong password must never yield a usable session. The broker runs
    ///     <c>allow_anonymous false</c>, so if bad credentials produced a session the fixture would be
    ///     misconfigured and "connected" would prove nothing about auth.
    ///     <para>
    ///         <b>LIVE-GATE FINDING (2026-07-24), now FIXED — this test was tightened accordingly.</b>
    ///         Against real Mosquitto, <see cref="MqttConnection.ConnectAsync"/> originally
    ///         <b>returned normally</b> for a CONNECT the broker rejected as not-authorized: MQTTnet 5
    ///         does not throw on an unsuccessful CONNACK, and <c>ConnectCoreAsync</c> then
    ///         unconditionally published <see cref="MqttConnectionState.Connected"/> and started the
    ///         reconnect supervisor. Measured at +300 ms with a wrong password:
    ///         <c>thrown=&lt;none&gt;</c>, <c>IsConnected=False</c>, <c>State=Reconnecting</c>. Broker
    ///         side: <c>Client … disconnected, not authorised.</c> The operator-visible consequence was
    ///         that <c>InitializeAsync</c> reported <c>DriverState.Healthy</c> / <c>HostState.Running</c>
    ///         off a connect that never authenticated, so a deployment carrying the wrong broker
    ///         password sealed green.
    ///     </para>
    ///     <para>
    ///         <c>ConnectCoreAsync</c> now inspects the CONNACK and raises
    ///         <see cref="MqttConnectRejectedException"/>, so this asserts the strong form: the
    ///         rejection is <b>surfaced</b>, classified as unrecoverable, and lands the connection in
    ///         <see cref="MqttConnectionState.Faulted"/>. The original weak invariant — polled over 5 s
    ///         so the optimistic <c>Connected</c> window cannot pass it for the wrong reason — is kept
    ///         underneath it, because "no session is ever established" is the property that actually
    ///         matters to a consumer and it must hold regardless of how the failure is reported.
    ///     </para>
    /// </summary>
    [Fact]
    public async Task Tls_connect_with_wrong_password_is_rejected_and_surfaced()
    {
        SkipIfUnconfigured();

        var ct = TestContext.Current.CancellationToken;
        var options = TlsOptions() with { Password = _fx.Password + "-definitely-wrong" };
        await using var connection = new MqttConnection(options, "mqtt-live-badpass");

        var rejection = await Should.ThrowAsync<MqttConnectRejectedException>(
            async () => await connection.ConnectAsync(ct));

        // Mosquitto answers bad credentials with NotAuthorized (MQTT 5) / 0x05 (3.1.1); either maps to
        // the same unrecoverable classification, which is what stops the reconnect supervisor.
        rejection.IsUnrecoverable.ShouldBeTrue(
            $"a broker that refused the credentials returned {rejection.ResultCode}, classified as retryable");
        rejection.Message.ShouldNotContain(options.Password ?? "<unset>", Case.Sensitive);
        connection.State.ShouldBe(MqttConnectionState.Faulted);

        // Poll rather than sample once: a regression that went back to publishing an optimistic
        // Connected would still read "not connected" in the first millisecond. Every sample must show
        // no session — a correct password would make IsConnected true within the first sample.
        var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(5);
        while (DateTime.UtcNow < deadline)
        {
            connection.IsConnected.ShouldBeFalse(
                "a CONNECT the broker rejected as not-authorized must never yield a usable session");
            connection.State.ShouldNotBe(
                MqttConnectionState.Connected,
                "a rejected CONNECT must never advertise a healthy session");
            await Task.Delay(100, ct);
        }

        TestContext.Current.SendDiagnosticMessage(
            $"wrong-password connect: {rejection.ResultCode} (unrecoverable={rejection.IsUnrecoverable}), "
            + $"IsConnected={connection.IsConnected}, State={connection.State}.");
    }

    /// <summary>
    ///     CA-pin happy path: <c>CaCertificatePath</c> set to the fixture's own CA, with the
    ///     untrusted-certificate escape hatch explicitly OFF, so the handshake can only succeed by
    ///     chaining the broker's leaf up to the pinned root through
    ///     <c>MqttConnection.ValidateAgainstPinnedCa</c>.
    /// </summary>
    [Fact]
    public async Task Tls_connect_pinned_to_fixture_ca_succeeds()
    {
        SkipIfUnconfigured();
        if (_fx.CaCertificatePath is null)
        {
            Assert.Skip(
                $"Skipped: set {MqttFixture.CaCertEnvVar} to the fixture CA (Docker/secrets/ca.crt) to run the CA-pin leg.");
        }

        File.Exists(_fx.CaCertificatePath).ShouldBeTrue(
            $"{MqttFixture.CaCertEnvVar}='{_fx.CaCertificatePath}' must point at the fixture CA PEM");

        var ct = TestContext.Current.CancellationToken;
        var options = TlsOptions() with
        {
            CaCertificatePath = _fx.CaCertificatePath,
            AllowUntrustedServerCertificate = false,
        };
        await using var connection = new MqttConnection(options, "mqtt-live-capin");

        await connection.ConnectAsync(ct);

        connection.IsConnected.ShouldBeTrue("the broker leaf must chain up to the pinned fixture CA");
    }

    /// <summary>
    ///     The pin's falsifiability control, and the reason the positive above means anything: pin to
    ///     a CA the broker's certificate was <b>not</b> issued by and assert the handshake fails. A
    ///     validator that returned <c>true</c> unconditionally — the silent failure mode for this
    ///     kind of code — would pass the positive test and fail here.
    ///     <para>
    ///         The foreign CA is generated in-process rather than taken from the fixture, so this leg
    ///         needs no extra rig provisioning and cannot accidentally be handed the real CA.
    ///     </para>
    /// </summary>
    [Fact]
    public async Task Tls_connect_pinned_to_a_foreign_ca_is_rejected()
    {
        SkipIfUnconfigured();

        var ct = TestContext.Current.CancellationToken;
        var foreignCaPath = Path.Combine(Path.GetTempPath(), $"otopcua-mqtt-foreign-ca-{Guid.NewGuid():N}.pem");
        await File.WriteAllTextAsync(foreignCaPath, CreateForeignCaPem(), ct);

        try
        {
            var options = TlsOptions() with
            {
                CaCertificatePath = foreignCaPath,
                AllowUntrustedServerCertificate = false,
                // The handshake fails fast; no need to sit on the default 15s budget.
                ConnectTimeoutSeconds = 10,
            };
            await using var connection = new MqttConnection(options, "mqtt-live-foreignca");

            await Should.ThrowAsync<Exception>(async () => await connection.ConnectAsync(ct));

            connection.IsConnected.ShouldBeFalse(
                "a broker certificate that does not chain to the pinned CA must be refused — fail closed");
        }
        finally
        {
            File.Delete(foreignCaPath);
        }
    }

    // -----------------------------------------------------------------------
    // Plain subscribe / read
    // -----------------------------------------------------------------------

    /// <summary>
    ///     The identity assertion. Subscribes an authored tag whose RawPath is deliberately unrelated
    ///     to its MQTT topic, and asserts the driver publishes the change under the <b>RawPath</b> —
    ///     the v3 driver wire reference <c>DriverHostActor</c> fans out to the raw + UNS NodeIds. A
    ///     driver that keyed notifications by topic (or by the TagConfig blob, the retired pre-v3
    ///     shape) would deliver a value that no node in the address space is listening for, and every
    ///     "did a value arrive" assertion would still pass.
    /// </summary>
    [Fact]
    public async Task Plain_subscribe_delivers_value_under_rawpath()
    {
        SkipIfUnconfigured();

        var ct = TestContext.Current.CancellationToken;
        const string rawPath = "Fixture/Mqtt/Line1/Temperature";
        var tag = JsonTag(rawPath, _fx.Topic("line1/temperature"), "$.value", DriverDataType.Float64);

        await using var driver = new MqttDriver(PlainOptions(tag), "mqtt-live-subscribe");
        var recorder = new DataChangeRecorder(driver);

        await driver.InitializeAsync(string.Empty, ct);
        await driver.SubscribeAsync([rawPath], TimeSpan.FromSeconds(1), ct);

        var change = await recorder.WaitForAsync(rawPath, MessageTimeout, ct);

        change.FullReference.ShouldBe(
            rawPath,
            "the driver must publish under the tag's RawPath, never its topic or its TagConfig blob");
        change.Snapshot.StatusCode.ShouldBe(Good);
        change.Snapshot.Value.ShouldBeOfType<double>().ShouldBeInRange(-100d, 1000d);

        // ...and the read path serves the same last value under the same key.
        var read = await driver.ReadAsync([rawPath], ct);
        read.Count.ShouldBe(1);
        read[0].StatusCode.ShouldBe(Good);
        read[0].Value.ShouldNotBeNull();
    }

    /// <summary>
    ///     Retained-message seeding. The fixture publisher emits <c>retained/seed</c> exactly once,
    ///     at container start, and never republishes it — so a subscriber connecting later can only
    ///     obtain a value from the broker's retained copy. Asserting the value on a topic that is
    ///     <i>also</i> republished every couple of seconds would prove nothing: a live publish
    ///     arriving quickly looks identical to a retained seed.
    /// </summary>
    [Fact]
    public async Task Retained_message_seeds_value_on_subscribe()
    {
        SkipIfUnconfigured();

        var ct = TestContext.Current.CancellationToken;
        const string rawPath = "Fixture/Mqtt/Retained/Seed";
        var tag = JsonTag(rawPath, _fx.Topic("retained/seed"), "$.value", DriverDataType.Float64);

        await using var driver = new MqttDriver(PlainOptions(tag), "mqtt-live-retained");
        var recorder = new DataChangeRecorder(driver);

        await driver.InitializeAsync(string.Empty, ct);
        await driver.SubscribeAsync([rawPath], TimeSpan.FromSeconds(1), ct);

        var change = await recorder.WaitForAsync(rawPath, MessageTimeout, ct);

        change.FullReference.ShouldBe(rawPath);
        change.Snapshot.StatusCode.ShouldBe(Good);
        change.Snapshot.Value.ShouldBeOfType<double>().ShouldBe(
            42.5d,
            "the retained seed carries the publisher's one-shot value; anything else means the "
            + "subscription was seeded from somewhere other than the broker's retained store");
    }

    /// <summary>
    ///     A chatty broker must not auto-provision. Publishes on a topic no tag was authored for and
    ///     asserts the driver never raises a notification for it.
    ///     <para>
    ///         <b>Deliberately falsifiable.</b> A version of this test that authored nothing would
    ///         pass vacuously — silence proves nothing if the pipeline is dead. So a real tag is
    ///         authored and subscribed first, and the test asserts that authored tag kept delivering
    ///         <i>throughout</i> the window in which the unauthored topic was being published. The
    ///         only way both assertions hold is if the driver was live and chose to ignore the
    ///         unauthored traffic.
    ///     </para>
    /// </summary>
    [Fact]
    public async Task Unauthored_topic_is_silent_while_an_authored_tag_keeps_flowing()
    {
        SkipIfUnconfigured();

        var ct = TestContext.Current.CancellationToken;
        const string authoredRawPath = "Fixture/Mqtt/Line1/Counter";
        const string unauthoredRawPath = "Fixture/Mqtt/Line1/NotAuthored";

        // Unique per run so a previous run's retained message can never seed it.
        var unauthoredTopic = _fx.Topic($"unauthored/{Guid.NewGuid():N}");

        var tag = JsonTag(authoredRawPath, _fx.Topic("line1/counter"), "$.value", DriverDataType.Int64);

        await using var driver = new MqttDriver(PlainOptions(tag), "mqtt-live-unauthored");
        var recorder = new DataChangeRecorder(driver);

        await driver.InitializeAsync(string.Empty, ct);

        // Subscribing an unauthored reference alongside the authored one: the driver must say
        // BadNodeIdUnknown rather than leaving it forever "waiting for initial data" — and must
        // never later turn it Good off broker traffic.
        await driver.SubscribeAsync([authoredRawPath, unauthoredRawPath], TimeSpan.FromSeconds(1), ct);

        // Proof the pipeline is live BEFORE the unauthored traffic starts.
        await recorder.WaitForAsync(authoredRawPath, MessageTimeout, ct);
        var beforeCount = recorder.CountFor(authoredRawPath);

        // Emit unauthored traffic with our own client — the fixture publisher's own noise topic is
        // not enough, because a test must be able to point at the exact messages it caused.
        await PublishUnauthoredAsync(unauthoredTopic, messages: 5, ct);

        // Let the authored tag tick at least once more while the unauthored messages sit delivered.
        await recorder.WaitForCountAsync(authoredRawPath, beforeCount + 1, MessageTimeout, ct);

        var seen = recorder.Snapshot();
        seen.ShouldAllBe(
            e => e.FullReference == authoredRawPath,
            "the driver must raise notifications ONLY for authored tags; a broker's other topics are none of its business. "
            + $"Saw: {string.Join(", ", seen.Select(e => e.FullReference).Distinct())}");

        var read = await driver.ReadAsync([authoredRawPath, unauthoredRawPath], ct);
        read[0].StatusCode.ShouldBe(Good, "the authored tag was flowing throughout — this is the falsifiability control");
        read[1].StatusCode.ShouldBeOneOf(
            BadNodeIdUnknown,
            BadWaitingForInitialData);
        read[1].Value.ShouldBeNull("an unauthored reference must never acquire a value from broker traffic");
    }

    // -----------------------------------------------------------------------
    // helpers
    // -----------------------------------------------------------------------

    private void SkipIfUnconfigured()
    {
        if (_fx.NotConfigured)
        {
            Assert.Skip(_fx.SkipReason!);
        }
    }

    /// <summary>Driver options for the TLS listener, pinning the fixture CA when one was supplied.</summary>
    private MqttDriverOptions TlsOptions(params RawTagEntry[] rawTags) => new()
    {
        Host = _fx.TlsHost,
        Port = _fx.TlsPort,
        UseTls = true,
        // With no CA to pin, the fixture's self-issued chain cannot validate against the OS trust
        // store, so the documented dev escape hatch is the only way the auth leg can run at all.
        CaCertificatePath = _fx.CaCertificatePath,
        AllowUntrustedServerCertificate = _fx.CaCertificatePath is null,
        Username = _fx.Username,
        Password = _fx.Password,
        ClientId = $"otopcua-live-{Guid.NewGuid():N}",
        ConnectTimeoutSeconds = 15,
        Mode = MqttMode.Plain,
        Plain = new MqttPlainOptions { DefaultQos = 1 },
        RawTags = rawTags,
    };

    /// <summary>Driver options for the plaintext listener — still authenticated.</summary>
    private MqttDriverOptions PlainOptions(params RawTagEntry[] rawTags) => new()
    {
        Host = _fx.PlainHost,
        Port = _fx.PlainPort,
        UseTls = false,
        Username = _fx.Username,
        Password = _fx.Password,
        ClientId = $"otopcua-live-{Guid.NewGuid():N}",
        ConnectTimeoutSeconds = 15,
        Mode = MqttMode.Plain,
        Plain = new MqttPlainOptions { DefaultQos = 1 },
        RawTags = rawTags,
    };

    private static RawTagEntry JsonTag(string rawPath, string topic, string jsonPath, DriverDataType dataType) =>
        new(
            rawPath,
            $$"""
              {"topic":"{{topic}}","payloadFormat":"Json","jsonPath":"{{jsonPath}}","dataType":"{{dataType}}","qos":1,"retainSeed":true}
              """,
            WriteIdempotent: false);

    /// <summary>
    ///     Publishes on a topic no tag is authored for, with a raw MQTTnet client rather than through
    ///     the driver (the driver has no write path in P1, and the point is traffic the driver did
    ///     not originate).
    /// </summary>
    private async Task PublishUnauthoredAsync(string topic, int messages, CancellationToken ct)
    {
        using var client = new MqttClientFactory().CreateMqttClient();
        var options = new MqttClientOptionsBuilder()
            .WithTcpServer(_fx.PlainHost, _fx.PlainPort)
            .WithCredentials(_fx.Username, _fx.Password)
            .WithClientId($"otopcua-live-noise-{Guid.NewGuid():N}")
            .WithTimeout(TimeSpan.FromSeconds(10))
            .Build();

        await client.ConnectAsync(options, ct);
        try
        {
            for (var i = 0; i < messages; i++)
            {
                await client.PublishStringAsync(
                    topic,
                    $$"""{"value":{{i}},"source":"unauthored-live-test"}""",
                    MQTTnet.Protocol.MqttQualityOfServiceLevel.AtLeastOnce,
                    retain: false,
                    ct);
            }
        }
        finally
        {
            await client.DisconnectAsync(cancellationToken: ct);
        }
    }

    /// <summary>
    ///     A self-signed CA the fixture broker's certificate was definitively NOT issued by, as PEM.
    ///     Generated per call so nothing about it can be shared with — or mistaken for — the real
    ///     fixture CA.
    /// </summary>
    private static string CreateForeignCaPem()
    {
        using var key = RSA.Create(2048);
        var request = new CertificateRequest(
            "CN=OtOpcUa MQTT foreign CA (negative control)",
            key,
            HashAlgorithmName.SHA256,
            RSASignaturePadding.Pkcs1);

        request.CertificateExtensions.Add(new X509BasicConstraintsExtension(true, false, 0, true));
        request.CertificateExtensions.Add(
            new X509KeyUsageExtension(X509KeyUsageFlags.KeyCertSign | X509KeyUsageFlags.CrlSign, true));

        using var certificate = request.CreateSelfSigned(
            DateTimeOffset.UtcNow.AddDays(-1),
            DateTimeOffset.UtcNow.AddDays(1));

        var builder = new StringBuilder();
        builder.AppendLine("-----BEGIN CERTIFICATE-----");
        builder.AppendLine(Convert.ToBase64String(certificate.RawData, Base64FormattingOptions.InsertLineBreaks));
        builder.AppendLine("-----END CERTIFICATE-----");
        return builder.ToString();
    }

    /// <summary>
    ///     Records every <c>OnDataChange</c> the driver raises and lets a test await a specific
    ///     RawPath. Attached in the constructor — <b>before</b> the MQTT subscribe — because the
    ///     retained seed can arrive inside the SUBSCRIBE round-trip, and a handler attached
    ///     afterwards would miss exactly the message the retained-seed test is about.
    /// </summary>
    private sealed class DataChangeRecorder
    {
        private readonly List<DataChangeEventArgs> _events = [];
        private readonly Lock _gate = new();
        private readonly List<Waiter> _waiters = [];

        public DataChangeRecorder(MqttDriver driver) => driver.OnDataChange += OnDataChange;

        /// <summary>A point-in-time copy of everything recorded so far.</summary>
        /// <returns>The recorded events.</returns>
        public IReadOnlyList<DataChangeEventArgs> Snapshot()
        {
            lock (_gate)
            {
                return [.. _events];
            }
        }

        /// <summary>How many notifications have been recorded for one RawPath.</summary>
        /// <param name="rawPath">The RawPath to count.</param>
        /// <returns>The count.</returns>
        public int CountFor(string rawPath)
        {
            lock (_gate)
            {
                return _events.Count(e => e.FullReference == rawPath);
            }
        }

        /// <summary>Awaits the first notification for <paramref name="rawPath"/>.</summary>
        /// <param name="rawPath">The RawPath to await.</param>
        /// <param name="timeout">Bound on the wait.</param>
        /// <param name="cancellationToken">Caller cancellation.</param>
        /// <returns>The first matching notification.</returns>
        public Task<DataChangeEventArgs> WaitForAsync(
            string rawPath,
            TimeSpan timeout,
            CancellationToken cancellationToken)
            => WaitForCountAsync(rawPath, count: 1, timeout, cancellationToken);

        /// <summary>
        ///     Awaits the <paramref name="count"/>-th notification for <paramref name="rawPath"/>.
        ///     Pre-scans what has already been recorded, so a caller that registers the wait after
        ///     the message arrived is served immediately rather than timing out.
        /// </summary>
        /// <param name="rawPath">The RawPath to await.</param>
        /// <param name="count">How many notifications must have been seen.</param>
        /// <param name="timeout">Bound on the wait.</param>
        /// <param name="cancellationToken">Caller cancellation.</param>
        /// <returns>The <paramref name="count"/>-th matching notification.</returns>
        public async Task<DataChangeEventArgs> WaitForCountAsync(
            string rawPath,
            int count,
            TimeSpan timeout,
            CancellationToken cancellationToken)
        {
            Waiter waiter;
            lock (_gate)
            {
                var matches = _events.Where(e => e.FullReference == rawPath).ToList();
                if (matches.Count >= count)
                {
                    return matches[count - 1];
                }

                waiter = new Waiter(rawPath, count);
                _waiters.Add(waiter);
            }

            try
            {
                return await waiter.Completion.Task.WaitAsync(timeout, cancellationToken);
            }
            catch (TimeoutException)
            {
                throw new TimeoutException(
                    $"No notification #{count} for RawPath '{rawPath}' within {timeout.TotalSeconds:0}s. "
                    + $"Recorded so far: {DescribeRecorded()}. Is the fixture publisher container running "
                    + "(`docker logs otopcua-mqtt-publisher`), and does MQTT_FIXTURE_TOPIC_PREFIX match it?");
            }
            finally
            {
                lock (_gate)
                {
                    _waiters.Remove(waiter);
                }
            }
        }

        private string DescribeRecorded()
        {
            var snapshot = Snapshot();
            return snapshot.Count == 0
                ? "(nothing)"
                : string.Join(", ", snapshot.GroupBy(e => e.FullReference).Select(g => $"{g.Key}×{g.Count()}"));
        }

        private void OnDataChange(object? sender, DataChangeEventArgs args)
        {
            List<Waiter>? satisfied = null;
            lock (_gate)
            {
                _events.Add(args);
                foreach (var waiter in _waiters)
                {
                    if (waiter.RawPath != args.FullReference) continue;
                    if (_events.Count(e => e.FullReference == waiter.RawPath) < waiter.Count) continue;

                    (satisfied ??= []).Add(waiter);
                }
            }

            // Completed OUTSIDE the lock, and with RunContinuationsAsynchronously below, so a test
            // continuation never runs on MQTTnet's dispatcher thread while holding this lock —
            // that would stall delivery for every other subscription.
            if (satisfied is null) return;
            foreach (var waiter in satisfied)
            {
                waiter.Completion.TrySetResult(args);
            }
        }

        private sealed record Waiter(string RawPath, int Count)
        {
            public TaskCompletionSource<DataChangeEventArgs> Completion { get; } =
                new(TaskCreationOptions.RunContinuationsAsynchronously);
        }
    }
}