jellyfin-ha-src/Emby.Server.Implementations/Session/RedisSessionDirectory.cs

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text.Json;
using System.Threading;
using System.Threading.Tasks;
using Jellyfin.Extensions.Json;
using MediaBrowser.Controller.Session;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using StackExchange.Redis;

namespace Emby.Server.Implementations.Session;

/// <summary>
/// A Redis-backed <see cref="ISessionDirectory"/>. Each entry is a key with an expiry, so the sessions of
/// an instance that stops refreshing them disappear on their own.
/// </summary>
public sealed class RedisSessionDirectory : ISessionDirectory
{
    private const string KeyPrefix = "jellyfin:session:";

    private static readonly JsonSerializerOptions _jsonOptions = JsonDefaults.Options;

    private readonly IConnectionMultiplexer _redis;
    private readonly IDatabase _db;
    private readonly SessionDirectoryOptions _options;
    private readonly ILogger<RedisSessionDirectory> _logger;

    /// <summary>
    /// Initializes a new instance of the <see cref="RedisSessionDirectory"/> class.
    /// </summary>
    /// <param name="redis">The Redis connection multiplexer.</param>
    /// <param name="options">The session directory configuration options.</param>
    /// <param name="logger">The logger.</param>
    public RedisSessionDirectory(
        IConnectionMultiplexer redis,
        IOptions<SessionDirectoryOptions> options,
        ILogger<RedisSessionDirectory> logger)
    {
        ArgumentNullException.ThrowIfNull(redis);
        ArgumentNullException.ThrowIfNull(options);

        _redis = redis;
        _db = redis.GetDatabase();
        _options = options.Value;
        _logger = logger;
    }

    private TimeSpan EntryTtl => TimeSpan.FromSeconds(Math.Max(1, _options.EntryTtlSeconds));

    /// <inheritdoc />
    public async Task PublishAsync(SessionDirectoryEntry entry, CancellationToken cancellationToken = default)
    {
        ArgumentNullException.ThrowIfNull(entry);

        var sessionId = entry.Session?.Id;
        if (string.IsNullOrEmpty(sessionId))
        {
            return;
        }

        try
        {
            await _db.StringSetAsync(KeyPrefix + sessionId, JsonSerializer.Serialize(entry, _jsonOptions), EntryTtl).ConfigureAwait(false);
        }
        catch (Exception ex)
        {
            _logger.LogWarning(ex, "Failed to publish session {SessionId}; it stays invisible to the other instances.", sessionId);
        }
    }

    /// <inheritdoc />
    public async Task RemoveAsync(string sessionId, CancellationToken cancellationToken = default)
    {
        try
        {
            await _db.KeyDeleteAsync(KeyPrefix + sessionId).ConfigureAwait(false);
        }
        catch (Exception ex)
        {
            _logger.LogWarning(ex, "Failed to remove session {SessionId}; it expires on its own.", sessionId);
        }
    }

    /// <inheritdoc />
    public async Task<IReadOnlyList<SessionDirectoryEntry>> GetAllAsync(CancellationToken cancellationToken = default)
    {
        var entries = new List<SessionDirectoryEntry>();

        try
        {
            foreach (var server in _redis.GetServers())
            {
                if (!server.IsConnected)
                {
                    continue;
                }

                var keys = new List<RedisKey>();
                await foreach (var key in server.KeysAsync(database: _db.Database, pattern: KeyPrefix + "*", pageSize: 1000).WithCancellation(cancellationToken).ConfigureAwait(false))
                {
                    keys.Add(key);
                }

                var values = await Task.WhenAll(keys.Select(key => _db.StringGetAsync(key))).ConfigureAwait(false);

                foreach (var raw in values)
                {
                    if (!raw.HasValue)
                    {
                        continue;
                    }

                    try
                    {
                        var entry = JsonSerializer.Deserialize<SessionDirectoryEntry>(raw.ToString(), _jsonOptions);
                        if (entry?.Session is not null)
                        {
                            entries.Add(entry);
                        }
                    }
                    catch (JsonException ex)
                    {
                        _logger.LogWarning(ex, "Failed to deserialize a session directory entry.");
                    }
                }
            }
        }
        catch (Exception ex)
        {
            // Degrade to the sessions this instance holds rather than failing the request outright.
            _logger.LogWarning(ex, "Failed to read the session directory; only local sessions are reported.");
            return Array.Empty<SessionDirectoryEntry>();
        }

        return entries;
    }
}