suitelinkclient/tests/SuiteLink.Client.Tests/SuiteLinkClientSubscriptionRegistryTests.cs

using SuiteLink.Client.Protocol;
using SuiteLink.Client.Transport;

namespace SuiteLink.Client.Tests;

public sealed class SuiteLinkClientSubscriptionRegistryTests
{
    [Fact]
    public async Task SubscribeAsync_StoresDurableSubscriptionIntent()
    {
        var transport = new FakeTransport();
        transport.EnqueueReceive(BuildHandshakeAckFrame());
        transport.EnqueueReceive(BuildAdviseAckFrame(1));

        var client = new SuiteLinkClient(transport);
        await client.ConnectAsync(CreateOptions());

        _ = await client.SubscribeAsync("Pump001.Run", _ => { });

        Assert.True(client.DebugHasDurableSubscription("Pump001.Run"));
    }

    [Fact]
    public async Task SubscribeAsync_DuplicateItem_Throws_AndKeepsOriginalCallbackRegistration()
    {
        var transport = new FakeTransport();
        transport.EnqueueReceive(BuildHandshakeAckFrame());
        transport.EnqueueReceive(BuildAdviseAckFrame(1));
        transport.EnqueueReceive(BuildBooleanUpdateFrame(1, true));

        var client = new SuiteLinkClient(transport);
        await client.ConnectAsync(CreateOptions());

        var firstCallbackCount = 0;
        var secondCallbackCount = 0;
        _ = await client.SubscribeAsync("Pump001.Run", _ => firstCallbackCount++);

        var duplicateException = await Assert.ThrowsAsync<InvalidOperationException>(
            () => client.SubscribeAsync("Pump001.Run", _ => secondCallbackCount++));
        Assert.Contains("already subscribed", duplicateException.Message, StringComparison.OrdinalIgnoreCase);

        await client.ProcessIncomingAsync();

        Assert.True(client.DebugHasDurableSubscription("Pump001.Run"));
        Assert.Equal(1, firstCallbackCount);
        Assert.Equal(0, secondCallbackCount);
    }

    [Fact]
    public async Task SubscriptionHandleDisposeAsync_RemovesDurableSubscriptionIntent()
    {
        var transport = new FakeTransport();
        transport.EnqueueReceive(BuildHandshakeAckFrame());
        transport.EnqueueReceive(BuildAdviseAckFrame(1));

        var client = new SuiteLinkClient(transport);
        await client.ConnectAsync(CreateOptions());

        var handle = await client.SubscribeAsync("Pump001.Run", _ => { });
        Assert.True(client.DebugHasDurableSubscription("Pump001.Run"));

        await handle.DisposeAsync();

        Assert.False(client.DebugHasDurableSubscription("Pump001.Run"));
    }

    [Fact]
    public async Task SubscriptionHandleDisposeAsync_RemovesDurableIntent_WhenUnadviseSendFails()
    {
        var transport = new FakeTransport();
        transport.EnqueueReceive(BuildHandshakeAckFrame());
        transport.EnqueueReceive(BuildAdviseAckFrame(1));
        transport.SendFailureFactory = frameBytes =>
        {
            var span = frameBytes.Span;
            var isUnadviseFrame = span.Length >= 4 &&
                span[2] == 0x04 &&
                span[3] == 0x80;
            return isUnadviseFrame ? new IOException("Synthetic unadvise send failure.") : null;
        };

        var client = new SuiteLinkClient(transport);
        await client.ConnectAsync(CreateOptions());

        var handle = await client.SubscribeAsync("Pump001.Run", _ => { });
        Assert.True(client.DebugHasDurableSubscription("Pump001.Run"));

        await Assert.ThrowsAsync<IOException>(() => handle.DisposeAsync().AsTask());
        Assert.False(client.DebugHasDurableSubscription("Pump001.Run"));
    }

    private static SuiteLinkConnectionOptions CreateOptions()
    {
        return new SuiteLinkConnectionOptions(
            host: "127.0.0.1",
            application: "App",
            topic: "Topic",
            clientName: "Client",
            clientNode: "Node",
            userName: "User",
            serverNode: "Server",
            timezone: "UTC",
            port: 5413);
    }

    private static byte[] BuildHandshakeAckFrame()
    {
        return [0x06, 0x00, 0x01, 0x00, 0xA1, 0xB2, 0xC3, 0xA5];
    }

    private static byte[] BuildAdviseAckFrame(uint tagId)
    {
        Span<byte> payload = stackalloc byte[5];
        SuiteLinkEncoding.WriteUInt32LittleEndian(payload[..4], tagId);
        payload[4] = 0x00;
        return SuiteLinkFrameWriter.WriteFrame(SuiteLinkSubscriptionCodec.AdviseAckMessageType, payload);
    }

    private static byte[] BuildBooleanUpdateFrame(uint tagId, bool value)
    {
        Span<byte> payload = stackalloc byte[10];
        SuiteLinkEncoding.WriteUInt32LittleEndian(payload[..4], tagId);
        SuiteLinkEncoding.WriteUInt16LittleEndian(payload.Slice(4, 2), 1);
        SuiteLinkEncoding.WriteUInt16LittleEndian(payload.Slice(6, 2), 0x00C0);
        payload[8] = (byte)SuiteLinkWireValueType.Binary;
        payload[9] = value ? (byte)1 : (byte)0;
        return SuiteLinkFrameWriter.WriteFrame(SuiteLinkUpdateCodec.UpdateMessageType, payload);
    }

    private sealed class FakeTransport : ISuiteLinkTransport
    {
        private readonly Queue<byte[]> _receiveChunks = [];
        public Func<ReadOnlyMemory<byte>, Exception?>? SendFailureFactory { get; set; }

        public bool IsConnected { get; private set; }

        public void EnqueueReceive(byte[] bytes)
        {
            _receiveChunks.Enqueue(bytes);
        }

        public ValueTask ConnectAsync(string host, int port, CancellationToken cancellationToken = default)
        {
            IsConnected = true;
            return ValueTask.CompletedTask;
        }

        public ValueTask SendAsync(ReadOnlyMemory<byte> buffer, CancellationToken cancellationToken = default)
        {
            var sendFailure = SendFailureFactory?.Invoke(buffer);
            if (sendFailure is not null)
            {
                throw sendFailure;
            }

            return ValueTask.CompletedTask;
        }

        public ValueTask<int> ReceiveAsync(Memory<byte> buffer, CancellationToken cancellationToken = default)
        {
            if (_receiveChunks.Count == 0)
            {
                return ValueTask.FromResult(0);
            }

            var bytes = _receiveChunks.Dequeue();
            bytes.CopyTo(buffer);
            return ValueTask.FromResult(bytes.Length);
        }

        public ValueTask DisposeAsync()
        {
            IsConnected = false;
            return ValueTask.CompletedTask;
        }
    }
}