Finlytic/FinlyticCore/Services/TradeRepublic/TradeRepublicClient.cs

using System;
using System.Collections.Concurrent;
using System.Text.Json;
using System.Threading;
using System.Threading.Tasks;
using FinlyticCore.Models.TradeRepublic;
using FinlyticCore.Util;
using Microsoft.Extensions.Logging;

namespace FinlyticCore.Services.TradeRepublic;

/// <summary>
/// A managed, thread-safe WebSocket client designed to communicate with the Trade Republic API.
/// Supports both single RPC requests and real-time live ticker subscriptions (e.g. {isin}.TIB).
/// </summary>
public class TradeRepublicClient : ManagedWebSocket
{
    private readonly ILogger<TradeRepublicClient> _logger;
    private int _currentSub;
    private readonly ConcurrentDictionary<int, TaskCompletionSource<ReceivedMessage>> _pendingRequests = new();
    private readonly ConcurrentDictionary<int, Action<string>> _tickerSubscriptions = new();

    public event Action<ReceivedMessage>? UnhandledMessageReceived;
    public event Action<string>? SystemMessageReceived;

    /// <summary>
    /// Initializes a new instance of the <see cref="TradeRepublicClient"/> class.
    /// </summary>
    /// <param name="logger">The logger instance.</param>
    public TradeRepublicClient(ILogger<TradeRepublicClient> logger)
    {
        _logger = logger;
    }

    /// <summary>
    /// Connects to the Trade Republic WebSocket API.
    /// </summary>
    /// <param name="cancellationToken">The cancellation token.</param>
    /// <returns>A task that represents the asynchronous operation. The task result contains a boolean indicating whether the connection was successful.</returns>
    public async Task<bool> InitAsync(CancellationToken cancellationToken = default)
    {
        if (IsConnected) return true;

        await ConnectAsync("wss://api.traderepublic.com/", TimeSpan.FromSeconds(10));

        var tcs = new TaskCompletionSource<ReceivedMessage>(TaskCreationOptions.RunContinuationsAsynchronously);
        _pendingRequests.TryAdd(-1, tcs);

        try
        {
            var json = JsonSerializer.Serialize(new TradeRepublicConnectRequest(), typeof(TradeRepublicConnectRequest), FinlyticJsonSerializerContext.Default);
            await SendAsync($"connect 34 {json}");

            var res = await tcs.Task.WaitAsync(TimeSpan.FromSeconds(5), cancellationToken);
            if (string.IsNullOrWhiteSpace(res.Type)) return false;

            var isConnected = res.Type == "connected";
            if (isConnected)
            {
                _logger.LogInformation("[{Channel}] WebSocket connection to Trade Republic established.", "TradeRepublicChannel");
            }

            return isConnected;
        }
        catch (Exception ex)
        {
            _pendingRequests.TryRemove(-1, out _);
            _logger.LogWarning(ex, "[{Channel}] Failed or timed out establishing Trade Republic WebSocket connection.", "TradeRepublicChannel");
            return false;
        }
    }

    /// <summary>
    /// Sends a JSON request to the Trade Republic WebSocket API and waits for the response.
    /// </summary>
    /// <typeparam name="TResponse">The expected response type.</typeparam>
    /// <typeparam name="TRequest">The request type.</typeparam>
    /// <param name="request">The request to send.</param>
    /// <param name="cancellationToken">The cancellation token.</param>
    /// <returns>A task that represents the asynchronous operation. The task result contains the deserialized response, or null if the request failed or timed out.</returns>
    public async Task<TResponse?> SendRequestAsync<TResponse, TRequest>(TRequest request, CancellationToken cancellationToken = default)
        where TResponse : class where TRequest : class
    {
        var tempSub = Interlocked.Increment(ref _currentSub);
        var msg = $"sub {tempSub} {JsonSerializer.Serialize(request, typeof(TRequest), FinlyticJsonSerializerContext.Default)}";

        _logger.LogDebug("[{Channel}] TR WS Sent (Request): {Message}", "TradeRepublicChannel", msg);
        var tcs = new TaskCompletionSource<ReceivedMessage>(TaskCreationOptions.RunContinuationsAsynchronously);

        _pendingRequests.TryAdd(tempSub, tcs);
        await SendAsync(msg);

        try
        {
            var res = await tcs.Task.WaitAsync(TimeSpan.FromSeconds(8), cancellationToken);
            if (string.IsNullOrWhiteSpace(res.Data) || !res.Type.Contains('A')) return null;

            return (TResponse?)JsonSerializer.Deserialize(res.Data, typeof(TResponse), FinlyticJsonSerializerContext.Default);
        }
        catch (Exception ex)
        {
            _logger.LogWarning(ex, "[{Channel}] Error waiting for Trade Republic response ID {SubId}", "TradeRepublicChannel", tempSub);
            return null;
        }
        finally
        {
            _pendingRequests.TryRemove(tempSub, out _);
            try { await SendAsync($"unsub {tempSub}"); } catch { }
        }
    }

    /// <summary>
    /// Subscribes to the real-time ticker stream for a specific ISIN (e.g., US5398301094.TIB).
    /// </summary>
    public async Task<int?> SubscribeTickerAsync(string isin, Action<TradeRepublicTickerResponse> onTick, CancellationToken cancellationToken = default)
    {
        if (string.IsNullOrWhiteSpace(isin)) return null;

        var cleanIsin = isin.Trim().ToUpperInvariant();
        var tickerId = cleanIsin.EndsWith(".TIB") ? cleanIsin : $"{cleanIsin}.TIB";

        var tempSub = Interlocked.Increment(ref _currentSub);
        var req = new TradeRepublicTickerRequest(tickerId);
        var msg = $"sub {tempSub} {JsonSerializer.Serialize(req, typeof(TradeRepublicTickerRequest), FinlyticJsonSerializerContext.Default)}";

        _tickerSubscriptions[tempSub] = jsonPayload =>
        {
            // Skip empty or non-JSON payloads (e.g. TR protocol ack messages)
            if (string.IsNullOrWhiteSpace(jsonPayload) || (!jsonPayload.TrimStart().StartsWith('{') && !jsonPayload.TrimStart().StartsWith('[')))
                return;

            try
            {
                var tickerRes = (TradeRepublicTickerResponse?)JsonSerializer.Deserialize(jsonPayload, typeof(TradeRepublicTickerResponse), FinlyticJsonSerializerContext.Default);
                if (tickerRes != null)
                {
                    onTick(tickerRes);
                }
            }
            catch (Exception ex)
            {
                _logger.LogWarning(ex, "[{Channel}] Failed to parse real-time ticker payload for {TickerId}", "TradeRepublicChannel", tickerId);
            }
        };

        _logger.LogInformation("[{Channel}] Subscribing to Trade Republic real-time ticker {TickerId} (Sub ID: {SubId})", "TradeRepublicChannel", tickerId, tempSub);
        _logger.LogDebug("[{Channel}] TR WS Sent: {Message}", "TradeRepublicChannel", msg);
        await SendAsync(msg);
        return tempSub;
    }

    /// <summary>
    /// Unsubscribes from a real-time ticker stream.
    /// </summary>
    /// <param name="subId">The subscription ID to unsubscribe.</param>
    /// <returns>A task representing the async operation.</returns>
    public async Task UnsubscribeTickerAsync(int subId)
    {
        _tickerSubscriptions.TryRemove(subId, out _);
        try
        {
            await SendAsync($"unsub {subId}");
        }
        catch { }
    }

    /// <inheritdoc />
    protected override void OnMessageReceived(string message)
    {
        if (string.IsNullOrWhiteSpace(message)) return;

        _logger.LogDebug("[{Channel}] TR WS Recv: {Message}", "TradeRepublicChannel", message);

        // Trade Republic message formats:
        // "34 connected" -> subId = 34, type = "connected", payload = "connected"
        // "22A {...}" or "22A{...}" -> subId = 22, type = "A", payload = "{...}"
        var digitLen = 0;
        while (digitLen < message.Length && char.IsDigit(message[digitLen]))
        {
            digitLen++;
        }

        if (digitLen == 0)
        {
            SystemMessageReceived?.Invoke(message);
            return;
        }

        if (!int.TryParse(message.Substring(0, digitLen), out var subId))
        {
            SystemMessageReceived?.Invoke(message);
            return;
        }

        var remainder = message.Substring(digitLen).TrimStart();
        string type;
        string payload;

        if (remainder.StartsWith("connected"))
        {
            type = "connected";
            payload = remainder;
        }
        else if (remainder.Length > 0)
        {
            // Type is usually a single character like 'A' or 'E'
            // The JSON payload (or ack) starts immediately after or after a space
            type = remainder[0].ToString();
            payload = remainder.Substring(1).TrimStart();
        }
        else
        {
            type = "ack";
            payload = string.Empty;
        }

        var received = new ReceivedMessage(subId, type, payload);

        if (_pendingRequests.TryGetValue(subId, out var tcs))
        {
            tcs.TrySetResult(received);
        }

        if (_tickerSubscriptions.TryGetValue(subId, out var handler))
        {
            handler(payload);
        }

        UnhandledMessageReceived?.Invoke(received);
    }
}

/// <summary>
/// Represents a received message from the Trade Republic WebSocket.
/// </summary>
/// <param name="SubId">The subscription ID.</param>
/// <param name="Type">The message type.</param>
/// <param name="Data">The payload data.</param>
public record ReceivedMessage(int SubId, string Type, string Data);