natsdotnet/tests/NATS.Server.Gateways.Tests/Gateways/GatewayInterestModeTests.cs

using System.Net;
using System.Net.Sockets;
using System.Text;
using Microsoft.Extensions.Logging.Abstractions;
using NATS.Client.Core;
using NATS.Server.Auth;
using NATS.Server.Configuration;
using NATS.Server.Gateways;
using NATS.Server.Subscriptions;

namespace NATS.Server.Gateways.Tests.Gateways;

/// <summary>
/// Gateway interest-only mode, account interest, subject interest propagation,
/// and subscription lifecycle tests.
/// Ported from golang/nats-server/server/gateway_test.go.
/// </summary>
public class GatewayInterestModeTests
{
    // ── Remote Interest Tracking via SubList ─────────────────────────────

    // Go: TestGatewayAccountInterest server/gateway_test.go:1794
    [Fact]
    public void Remote_interest_tracked_for_literal_subject()
    {
        using var subList = new SubList();
        subList.ApplyRemoteSub(new RemoteSubscription("orders.created", null, "gw1", "$G"));

        subList.HasRemoteInterest("$G", "orders.created").ShouldBeTrue();
        subList.HasRemoteInterest("$G", "orders.updated").ShouldBeFalse();
    }

    // Go: TestGatewayAccountInterest server/gateway_test.go:1794
    [Fact]
    public void Remote_interest_tracked_for_wildcard_subject()
    {
        using var subList = new SubList();
        subList.ApplyRemoteSub(new RemoteSubscription("orders.*", null, "gw1", "$G"));

        subList.HasRemoteInterest("$G", "orders.created").ShouldBeTrue();
        subList.HasRemoteInterest("$G", "orders.updated").ShouldBeTrue();
        subList.HasRemoteInterest("$G", "orders.deep.nested").ShouldBeFalse();
    }

    // Go: TestGatewayAccountInterest server/gateway_test.go:1794
    [Fact]
    public void Remote_interest_tracked_for_fwc_subject()
    {
        using var subList = new SubList();
        subList.ApplyRemoteSub(new RemoteSubscription("events.>", null, "gw1", "$G"));

        subList.HasRemoteInterest("$G", "events.one").ShouldBeTrue();
        subList.HasRemoteInterest("$G", "events.one.two.three").ShouldBeTrue();
        subList.HasRemoteInterest("$G", "other").ShouldBeFalse();
    }

    // Go: TestGatewayAccountInterest server/gateway_test.go:1794
    [Fact]
    public void Remote_interest_scoped_to_account()
    {
        using var subList = new SubList();
        subList.ApplyRemoteSub(new RemoteSubscription("orders.>", null, "gw1", "ACCT_A"));

        subList.HasRemoteInterest("ACCT_A", "orders.created").ShouldBeTrue();
        subList.HasRemoteInterest("ACCT_B", "orders.created").ShouldBeFalse();
        subList.HasRemoteInterest("$G", "orders.created").ShouldBeFalse();
    }

    // Go: TestGatewayAccountUnsub server/gateway_test.go:1912
    [Fact]
    public void Remote_interest_removed_on_aminus()
    {
        using var subList = new SubList();
        subList.ApplyRemoteSub(new RemoteSubscription("orders.>", null, "gw1", "$G"));
        subList.HasRemoteInterest("$G", "orders.created").ShouldBeTrue();

        subList.ApplyRemoteSub(RemoteSubscription.Removal("orders.>", null, "gw1", "$G"));
        subList.HasRemoteInterest("$G", "orders.created").ShouldBeFalse();
    }

    // Go: TestGatewayAccountInterest server/gateway_test.go:1794
    [Fact]
    public void Multiple_remote_interests_from_different_routes()
    {
        using var subList = new SubList();
        subList.ApplyRemoteSub(new RemoteSubscription("orders.*", null, "gw1", "$G"));
        subList.ApplyRemoteSub(new RemoteSubscription("orders.*", null, "gw2", "$G"));

        subList.HasRemoteInterest("$G", "orders.created").ShouldBeTrue();
        subList.MatchRemote("$G", "orders.created").Count.ShouldBe(2);
    }

    // Go: TestGatewayAccountUnsub server/gateway_test.go:1912
    [Fact]
    public void Removing_one_route_interest_keeps_other()
    {
        using var subList = new SubList();
        subList.ApplyRemoteSub(new RemoteSubscription("orders.*", null, "gw1", "$G"));
        subList.ApplyRemoteSub(new RemoteSubscription("orders.*", null, "gw2", "$G"));

        subList.ApplyRemoteSub(RemoteSubscription.Removal("orders.*", null, "gw1", "$G"));
        subList.HasRemoteInterest("$G", "orders.created").ShouldBeTrue();
        subList.MatchRemote("$G", "orders.created").Count.ShouldBe(1);
    }

    // ── Interest Change Events ──────────────────────────────────────────

    // Go: TestGatewaySwitchToInterestOnlyModeImmediately server/gateway_test.go:6934
    [Fact]
    public void Interest_change_event_fired_on_remote_add()
    {
        using var subList = new SubList();
        var changes = new List<InterestChange>();
        subList.InterestChanged += change => changes.Add(change);

        subList.ApplyRemoteSub(new RemoteSubscription("test.>", null, "gw1", "$G"));

        changes.Count.ShouldBe(1);
        changes[0].Kind.ShouldBe(InterestChangeKind.RemoteAdded);
        changes[0].Subject.ShouldBe("test.>");
        changes[0].Account.ShouldBe("$G");
    }

    // Go: TestGatewayAccountUnsub server/gateway_test.go:1912
    [Fact]
    public void Interest_change_event_fired_on_remote_remove()
    {
        using var subList = new SubList();
        var changes = new List<InterestChange>();
        subList.InterestChanged += change => changes.Add(change);

        subList.ApplyRemoteSub(new RemoteSubscription("test.>", null, "gw1", "$G"));
        subList.ApplyRemoteSub(RemoteSubscription.Removal("test.>", null, "gw1", "$G"));

        changes.Count.ShouldBe(2);
        changes[1].Kind.ShouldBe(InterestChangeKind.RemoteRemoved);
    }

    // Go: TestGatewaySwitchToInterestOnlyModeImmediately server/gateway_test.go:6934
    [Fact]
    public void Duplicate_remote_add_does_not_fire_extra_event()
    {
        using var subList = new SubList();
        var addCount = 0;
        subList.InterestChanged += change =>
        {
            if (change.Kind == InterestChangeKind.RemoteAdded)
                addCount++;
        };

        subList.ApplyRemoteSub(new RemoteSubscription("test.>", null, "gw1", "$G"));
        subList.ApplyRemoteSub(new RemoteSubscription("test.>", null, "gw1", "$G"));

        addCount.ShouldBe(1);
    }

    // Go: TestGatewayAccountUnsub server/gateway_test.go:1912
    [Fact]
    public void Remove_nonexistent_subscription_does_not_fire_event()
    {
        using var subList = new SubList();
        var removeCount = 0;
        subList.InterestChanged += change =>
        {
            if (change.Kind == InterestChangeKind.RemoteRemoved)
                removeCount++;
        };

        subList.ApplyRemoteSub(RemoteSubscription.Removal("nonexistent", null, "gw1", "$G"));

        removeCount.ShouldBe(0);
    }

    // ── Queue Weight in MatchRemote ─────────────────────────────────────

    // Go: TestGatewayTotalQSubs server/gateway_test.go:2484
    [Fact]
    public void Match_remote_expands_queue_weight()
    {
        using var subList = new SubList();
        subList.ApplyRemoteSub(new RemoteSubscription("foo", "bar", "gw1", "$G", QueueWeight: 3));

        var matches = subList.MatchRemote("$G", "foo");
        matches.Count.ShouldBe(3);
    }

    // Go: TestGatewayTotalQSubs server/gateway_test.go:2484
    [Fact]
    public void Match_remote_default_weight_is_one()
    {
        using var subList = new SubList();
        subList.ApplyRemoteSub(new RemoteSubscription("foo", "bar", "gw1", "$G"));

        var matches = subList.MatchRemote("$G", "foo");
        matches.Count.ShouldBe(1);
    }

    // ── End-to-End Interest Propagation via Gateway ─────────────────────

    // Go: TestGatewayDontSendSubInterest server/gateway_test.go:1755
    [Fact]
    public async Task Local_subscription_propagated_to_remote_via_gateway()
    {
        await using var fixture = await InterestModeFixture.StartAsync();

        await using var localConn = new NatsConnection(new NatsOpts
        {
            Url = $"nats://127.0.0.1:{fixture.Local.Port}",
        });
        await localConn.ConnectAsync();

        await using var sub = await localConn.SubscribeCoreAsync<string>("prop.test");
        await localConn.PingAsync();

        // The remote server should see the interest
        using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(5));
        while (!timeout.IsCancellationRequested && !fixture.Remote.HasRemoteInterest("prop.test"))
            await Task.Delay(50, timeout.Token).ContinueWith(_ => { }, TaskScheduler.Default);

        fixture.Remote.HasRemoteInterest("prop.test").ShouldBeTrue();
    }

    // Go: TestGatewayAccountUnsub server/gateway_test.go:1912
    [Fact]
    public async Task Unsubscribe_propagated_to_remote_via_gateway()
    {
        await using var fixture = await InterestModeFixture.StartAsync();

        await using var localConn = new NatsConnection(new NatsOpts
        {
            Url = $"nats://127.0.0.1:{fixture.Local.Port}",
        });
        await localConn.ConnectAsync();

        var sub = await localConn.SubscribeCoreAsync<string>("unsub.test");
        await localConn.PingAsync();

        using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(5));
        while (!timeout.IsCancellationRequested && !fixture.Remote.HasRemoteInterest("unsub.test"))
            await Task.Delay(50, timeout.Token).ContinueWith(_ => { }, TaskScheduler.Default);

        fixture.Remote.HasRemoteInterest("unsub.test").ShouldBeTrue();

        // Unsubscribe
        await sub.DisposeAsync();
        await localConn.PingAsync();

        // Wait for interest to be removed
        using var unsubTimeout = new CancellationTokenSource(TimeSpan.FromSeconds(5));
        while (!unsubTimeout.IsCancellationRequested && fixture.Remote.HasRemoteInterest("unsub.test"))
            await Task.Delay(50, unsubTimeout.Token).ContinueWith(_ => { }, TaskScheduler.Default);

        fixture.Remote.HasRemoteInterest("unsub.test").ShouldBeFalse();
    }

    // Go: TestGatewaySubjectInterest server/gateway_test.go:1972
    [Fact]
    public async Task Remote_wildcard_subscription_establishes_interest()
    {
        await using var fixture = await InterestModeFixture.StartAsync();

        await using var remoteConn = new NatsConnection(new NatsOpts
        {
            Url = $"nats://127.0.0.1:{fixture.Remote.Port}",
        });
        await remoteConn.ConnectAsync();

        await using var sub = await remoteConn.SubscribeCoreAsync<string>("interest.>");
        await remoteConn.PingAsync();

        using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(5));
        while (!timeout.IsCancellationRequested && !fixture.Local.HasRemoteInterest("interest.test"))
            await Task.Delay(50, timeout.Token).ContinueWith(_ => { }, TaskScheduler.Default);

        fixture.Local.HasRemoteInterest("interest.test").ShouldBeTrue();
        fixture.Local.HasRemoteInterest("interest.deep.nested").ShouldBeTrue();
    }

    // Go: TestGatewayDontSendSubInterest server/gateway_test.go:1755
    [Fact]
    public async Task Multiple_subscribers_same_subject_produces_single_interest()
    {
        await using var fixture = await InterestModeFixture.StartAsync();

        await using var conn1 = new NatsConnection(new NatsOpts
        {
            Url = $"nats://127.0.0.1:{fixture.Remote.Port}",
        });
        await conn1.ConnectAsync();

        await using var conn2 = new NatsConnection(new NatsOpts
        {
            Url = $"nats://127.0.0.1:{fixture.Remote.Port}",
        });
        await conn2.ConnectAsync();

        await using var sub1 = await conn1.SubscribeCoreAsync<string>("multi.interest");
        await using var sub2 = await conn2.SubscribeCoreAsync<string>("multi.interest");
        await conn1.PingAsync();
        await conn2.PingAsync();

        using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(5));
        while (!timeout.IsCancellationRequested && !fixture.Local.HasRemoteInterest("multi.interest"))
            await Task.Delay(50, timeout.Token).ContinueWith(_ => { }, TaskScheduler.Default);

        fixture.Local.HasRemoteInterest("multi.interest").ShouldBeTrue();
    }

    // Go: TestGatewayAccountInterest server/gateway_test.go:1794
    [Fact]
    public async Task Account_scoped_interest_propagated_via_gateway()
    {
        var users = new User[]
        {
            new() { Username = "acct_user", Password = "pass", Account = "MYACCT" },
        };

        await using var fixture = await InterestModeFixture.StartWithUsersAsync(users);

        await using var conn = new NatsConnection(new NatsOpts
        {
            Url = $"nats://acct_user:pass@127.0.0.1:{fixture.Remote.Port}",
        });
        await conn.ConnectAsync();

        await using var sub = await conn.SubscribeCoreAsync<string>("acct.interest");
        await conn.PingAsync();

        using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(5));
        while (!timeout.IsCancellationRequested && !fixture.Local.HasRemoteInterest("MYACCT", "acct.interest"))
            await Task.Delay(50, timeout.Token).ContinueWith(_ => { }, TaskScheduler.Default);

        fixture.Local.HasRemoteInterest("MYACCT", "acct.interest").ShouldBeTrue();
    }

    // ── RemoteSubscription Record Tests ─────────────────────────────────

    // Go: TestGatewayAccountInterest server/gateway_test.go:1794
    [Fact]
    public void RemoteSubscription_record_equality()
    {
        var a = new RemoteSubscription("foo", null, "gw1", "$G");
        var b = new RemoteSubscription("foo", null, "gw1", "$G");
        a.ShouldBe(b);
    }

    // Go: TestGatewayAccountInterest server/gateway_test.go:1794
    [Fact]
    public void RemoteSubscription_removal_factory()
    {
        var removal = RemoteSubscription.Removal("foo", "bar", "gw1", "$G");
        removal.IsRemoval.ShouldBeTrue();
        removal.Subject.ShouldBe("foo");
        removal.Queue.ShouldBe("bar");
        removal.RouteId.ShouldBe("gw1");
    }

    // Go: TestGatewayAccountInterest server/gateway_test.go:1794
    [Fact]
    public void RemoteSubscription_default_account_is_global()
    {
        var sub = new RemoteSubscription("foo", null, "gw1");
        sub.Account.ShouldBe("$G");
    }

    // Go: TestGatewayTotalQSubs server/gateway_test.go:2484
    [Fact]
    public void RemoteSubscription_default_queue_weight_is_one()
    {
        var sub = new RemoteSubscription("foo", "bar", "gw1");
        sub.QueueWeight.ShouldBe(1);
    }

    // Go: TestGatewayAccountUnsub server/gateway_test.go:1912
    [Fact]
    public void RemoteSubscription_default_is_not_removal()
    {
        var sub = new RemoteSubscription("foo", null, "gw1");
        sub.IsRemoval.ShouldBeFalse();
    }

    // ── Subscription Propagation by GatewayManager ──────────────────────

    // Go: TestGatewayDontSendSubInterest server/gateway_test.go:1755
    [Fact]
    public async Task Gateway_manager_propagate_subscription_sends_aplus()
    {
        using var listener = new TcpListener(IPAddress.Loopback, 0);
        listener.Start();
        var port = ((IPEndPoint)listener.LocalEndpoint).Port;

        var options = new GatewayOptions
        {
            Name = "LOCAL",
            Host = "127.0.0.1",
            Port = 0,
            Remotes = [$"127.0.0.1:{port}"],
        };
        var manager = new GatewayManager(
            options,
            new ServerStats(),
            "SERVER1",
            _ => { },
            _ => { },
            NullLogger<GatewayManager>.Instance);

        await manager.StartAsync(CancellationToken.None);

        // Accept the connection from gateway manager
        using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
        using var gwSocket = await listener.AcceptSocketAsync(cts.Token);

        // Exchange handshakes
        var line = await ReadLineAsync(gwSocket, cts.Token);
        line.ShouldStartWith("GATEWAY ");
        await WriteLineAsync(gwSocket, "GATEWAY REMOTE1", cts.Token);

        // Wait for connection to be registered
        await Task.Delay(200);

        // Propagate a subscription
        manager.PropagateLocalSubscription("$G", "orders.>", null);

        // Read the A+ message
        await Task.Delay(100);
        var aplusLine = await ReadLineAsync(gwSocket, cts.Token);
        aplusLine.ShouldBe("A+ $G orders.>");

        await manager.DisposeAsync();
    }

    // Go: TestGatewayAccountUnsub server/gateway_test.go:1912
    [Fact]
    public async Task Gateway_manager_propagate_unsubscription_sends_aminus()
    {
        using var listener = new TcpListener(IPAddress.Loopback, 0);
        listener.Start();
        var port = ((IPEndPoint)listener.LocalEndpoint).Port;

        var options = new GatewayOptions
        {
            Name = "LOCAL",
            Host = "127.0.0.1",
            Port = 0,
            Remotes = [$"127.0.0.1:{port}"],
        };
        var manager = new GatewayManager(
            options,
            new ServerStats(),
            "SERVER1",
            _ => { },
            _ => { },
            NullLogger<GatewayManager>.Instance);

        await manager.StartAsync(CancellationToken.None);

        using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
        using var gwSocket = await listener.AcceptSocketAsync(cts.Token);

        var line = await ReadLineAsync(gwSocket, cts.Token);
        line.ShouldStartWith("GATEWAY ");
        await WriteLineAsync(gwSocket, "GATEWAY REMOTE1", cts.Token);

        await Task.Delay(200);

        manager.PropagateLocalUnsubscription("$G", "orders.>", null);

        await Task.Delay(100);
        var aminusLine = await ReadLineAsync(gwSocket, cts.Token);
        aminusLine.ShouldBe("A- $G orders.>");

        await manager.DisposeAsync();
    }

    // ── Helpers ─────────────────────────────────────────────────────────

    private static async Task<string> ReadLineAsync(Socket socket, CancellationToken ct)
    {
        var bytes = new List<byte>(64);
        var single = new byte[1];
        while (true)
        {
            var read = await socket.ReceiveAsync(single, SocketFlags.None, ct);
            if (read == 0)
                break;
            if (single[0] == (byte)'\n')
                break;
            if (single[0] != (byte)'\r')
                bytes.Add(single[0]);
        }

        return Encoding.ASCII.GetString([.. bytes]);
    }

    private static Task WriteLineAsync(Socket socket, string line, CancellationToken ct)
        => socket.SendAsync(Encoding.ASCII.GetBytes($"{line}\r\n"), SocketFlags.None, ct).AsTask();
}

/// <summary>
/// Shared fixture for interest mode tests.
/// </summary>
internal sealed class InterestModeFixture : IAsyncDisposable
{
    private readonly CancellationTokenSource _localCts;
    private readonly CancellationTokenSource _remoteCts;

    private InterestModeFixture(NatsServer local, NatsServer remote, CancellationTokenSource localCts, CancellationTokenSource remoteCts)
    {
        Local = local;
        Remote = remote;
        _localCts = localCts;
        _remoteCts = remoteCts;
    }

    public NatsServer Local { get; }
    public NatsServer Remote { get; }

    public static Task<InterestModeFixture> StartAsync()
        => StartWithUsersAsync(null);

    public static async Task<InterestModeFixture> StartWithUsersAsync(IReadOnlyList<User>? users)
    {
        var localOptions = new NatsOptions
        {
            Host = "127.0.0.1",
            Port = 0,
            Users = users,
            Gateway = new GatewayOptions
            {
                Name = "LOCAL",
                Host = "127.0.0.1",
                Port = 0,
            },
        };

        var local = new NatsServer(localOptions, NullLoggerFactory.Instance);
        var localCts = new CancellationTokenSource();
        _ = local.StartAsync(localCts.Token);
        await local.WaitForReadyAsync();

        var remoteOptions = new NatsOptions
        {
            Host = "127.0.0.1",
            Port = 0,
            Users = users,
            Gateway = new GatewayOptions
            {
                Name = "REMOTE",
                Host = "127.0.0.1",
                Port = 0,
                Remotes = [local.GatewayListen!],
            },
        };

        var remote = new NatsServer(remoteOptions, NullLoggerFactory.Instance);
        var remoteCts = new CancellationTokenSource();
        _ = remote.StartAsync(remoteCts.Token);
        await remote.WaitForReadyAsync();

        using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(5));
        while (!timeout.IsCancellationRequested && (local.Stats.Gateways == 0 || remote.Stats.Gateways == 0))
            await Task.Delay(50, timeout.Token).ContinueWith(_ => { }, TaskScheduler.Default);

        return new InterestModeFixture(local, remote, localCts, remoteCts);
    }

    public async ValueTask DisposeAsync()
    {
        await _localCts.CancelAsync();
        await _remoteCts.CancelAsync();
        Local.Dispose();
        Remote.Dispose();
        _localCts.Dispose();
        _remoteCts.Dispose();
    }
}