I'm using FusionCache!

[lootag]

Scenario

I need to bill users of an API for each second they're connected to a specific socket, and I'm using FusionCache to do the realtime billing of credits. Works like a charm.

Liked

Pretty much everything. The library is so versatile it can literally go from being just a simple cache, all the way to a full fledged in-memory-data backed by redis.

Unliked/Missing

I don't think anything is missing in particular, and in fact I would not add any more features than it currently supports in order to prevent scope creep.

Closing Thoughts

Good job @jodydonetti, keep em coming

[pietrodicaprio]

😮could you share a bit of your implementation?
How are you using FusionCache to track the connection?

Nice out of the box thinking by the way!

[lootag]

public class SpeechRecognitionService : IDisposable
{
    private readonly WebSocket _webSocket;
    private ILiveTranscriptionClient _liveTranscriptionClient;
    private IFusionCache _cache;
    private readonly uint _freeTrialCredits = 0;
    private string _username;
    private Language _language;
    private string[] _languageCodes = ["en-US", "en-GB", "en-AU", "en-NZ", "en-IN", "es", "es-419"];

    private SpeechRecognitionService(
        WebSocket webSocket,
        IFusionCache cache,
        string username,
        Language language
        )
    {
        _webSocket = webSocket;
        _language = language;
        var credentials = new Credentials("<DEEPGRAM_KEY>");
        var deepgramClient = new DeepgramClient(credentials);
        _liveTranscriptionClient = deepgramClient.CreateLiveTranscriptionClient();
        _liveTranscriptionClient.ConnectionOpened += OnConnectionOpen;
        _liveTranscriptionClient.TranscriptReceived += OnReceiveTranscript;
        _liveTranscriptionClient.ConnectionClosed += OnConnectionClosed;
        _liveTranscriptionClient.ConnectionError += OnConnectionError;
        _cache = cache;
        _username = username;
    }

    public static async Task<SpeechRecognitionService> Create(
        HttpContext context,
        IFusionCache cache,
        string username,
        Language language
        )
    {
        var webSocket = await context.WebSockets.AcceptWebSocketAsync();
        return new SpeechRecognitionService(webSocket, cache, username, language);
    }
    public void Dispose()
    {
        _webSocket.Dispose();
        _liveTranscriptionClient.Dispose();
    }

    public async Task Listen(LiveTranscriptionOptions deepgramOptions)
    {
        var closeStatus = WebSocketCloseStatus.NormalClosure;

        var audioBuffer = new byte[4096];
        var ct = new CancellationTokenSource();
        var timeIsUpSignal = ct.Token;
        var timerCancellationTokenSource = new CancellationTokenSource();
        await _liveTranscriptionClient.StartConnectionAsync(deepgramOptions);
        var timer = Task.Run(async () => await Timer(ct, timerCancellationTokenSource.Token), timeIsUpSignal);
        while (_webSocket.State == WebSocketState.Open)
        {
            try
            {
                var receiveCancellation = new CancellationTokenSource();
                receiveCancellation.CancelAfter(5000);
                var result = await _webSocket.ReceiveAsync(new ArraySegment<byte>(audioBuffer), receiveCancellation.Token);
                if (result?.MessageType == WebSocketMessageType.Binary && !timeIsUpSignal.IsCancellationRequested)
                {
                    var audioData = new ArraySegment<byte>(audioBuffer, 0, result.Count);
                    _liveTranscriptionClient.SendData([.. audioData]);
                }
                else
                {
                    timerCancellationTokenSource.Cancel();
                    await _liveTranscriptionClient.FinishAsync();
                    closeStatus = WebSocketCloseStatus.ProtocolError;
                    break;
                }
            }
            catch (Exception ex)
            {
                var message = ex.Message;
                Console.WriteLine(message);
                closeStatus = WebSocketCloseStatus.InvalidMessageType;
                timerCancellationTokenSource.Cancel();
                await _liveTranscriptionClient.FinishAsync();
            }
        }
        await _liveTranscriptionClient.FinishAsync();
        timerCancellationTokenSource.Cancel();
        await _webSocket.CloseAsync(closeStatus, "Closing connection", CancellationToken.None);
    }

    async Task Timer(CancellationTokenSource ct, CancellationToken token = default)
    {
        var user = await _cache.GetOrSetAsync(_username, async _ => new User(
            _freeTrialCredits
            ), TimeSpan.MaxValue);
        while (user.AvailableCredits > 0 && !token.IsCancellationRequested && _webSocket.State == WebSocketState.Open)
        {
            await Task.Delay(990);
            user = _cache.GetOrSet(_username, _ => new User(_freeTrialCredits), TimeSpan.MaxValue);
            await _cache.SetAsync(_username, user.UseAudioSeconds(1), (options) =>
            {
                options.Duration = TimeSpan.MaxValue;
                options.AllowBackgroundDistributedCacheOperations = true;
            });
        }
        ct.Cancel();
        await _liveTranscriptionClient.FinishAsync();
    }

    void OnConnectionOpen(object? sender, EventArgs eventArgs)
    {
        Console.WriteLine("Connection opened");
    }

    async void OnConnectionClosed(object? sender, System.EventArgs eventArgs)
    {
        if (_webSocket.State == WebSocketState.Open)
        {
            await _webSocket.CloseAsync(WebSocketCloseStatus.InternalServerError, "Service unexpectedly closed the socket.", CancellationToken.None);
        }
        Console.WriteLine($"Connection closed {eventArgs}");
        var ea = eventArgs as ConnectionClosedEventArgs;
    }

    async void OnConnectionError(object? sender, System.EventArgs eventArgs)
    {

        if (_webSocket.State == WebSocketState.Open)
        {
            await _webSocket.CloseAsync(WebSocketCloseStatus.InternalServerError, "Service unexpectedly closed the socket.", CancellationToken.None);
        }
        Console.WriteLine($"Connection error {eventArgs}");
    }

    async void OnReceiveTranscript(object? sender, TranscriptReceivedEventArgs eventArgs)
    {
        var user = await _cache.GetOrSetAsync(_username, async _ => new User(_freeTrialCredits), TimeSpan.MaxValue);
        var transcription = eventArgs.Transcript.Channel?.Alternatives.FirstOrDefault()?.Transcript ?? string.Empty;
        if (eventArgs.Transcript.IsFinal && !string.IsNullOrEmpty(transcription) && _webSocket.State == WebSocketState.Open)
        {
            await _webSocket.SendAsync(Encoding.UTF8.GetBytes(transcription), WebSocketMessageType.Text, true, CancellationToken.None);
        }
    }
}

[lootag]

@pietrodicaprio there you go :-) You actually made me think that the entire server could be open sourced, I'll probably do that actually at some point, I just need to remove some minor proprietary configurations xD

[pietrodicaprio]

Oh, so FusionCache is used to avoid to being affected by the performance of the chosen data layer and other systems take care of the written data in order to bill (Redis pub/sub ?), do I get it right?

[lootag]

Yeah, so the application uses credits. Users buy an amount of credits, and certain actions (including voice transcription, which happens over WebSocket) consume these credits. Since I need the consumption of credits to be both fairly consistent and very performant, I use FusionCache + Redis to keep track of how many credits a user has left. When they connect to a socket, I start a timer that consumes their credits as time goes by.

If the question is whether I do actual billing through a proper payment processor in the tight loop above... well, the answer is no xD that would be too slow, and I'm not even sure that any payment processor would be okay with me issuing one request per second per user.

[pietrodicaprio]

I noticed the credits, I didn't mean to do payment processing every second ehehe
Thanks for the details, awesome job!

[jodydonetti]

Hi @lootag , thanks for sharing!

I need to bill users of an API for each second they're connected to a specific socket, and I'm using FusionCache to do the realtime billing of credits. Works like a charm.

Interesting use case, and happy it's working great!

Liked

Pretty much everything. The library is so versatile it can literally go from being just a simple cache, all the way to a full fledged in-memory-data backed by redis.

Unliked/Missing

I don't think anything is missing in particular, and in fact I would not add any more features than it currently supports in order to prevent scope creep.

Eheh, I see what you mean, and I generally agree: I have some other things in my backlog, but I try to be very careful when I have to decide to add something, to avoid feature creep.