lux/EgwCoreLib.Lux.Data/Services/General/RedisService.cs

namespace EgwCoreLib.Lux.Data.Services.General
{
    /// <summary>
    /// Implementazione interfaccia REDIS:
    /// - gestione PubSub
    /// - gestione caching
    /// </summary>
    public class RedisService : IRedisService
    {
        #region Public Constructors

        public RedisService(IConnectionMultiplexer connection)
        {
            _connection = connection;
            _db = connection.GetDatabase();
            _subscriber = connection.GetSubscriber();

            // json serializer... FIX errore loop circolare https://www.ryadel.com/en/jsonserializationexception-self-referencing-loop-detected-error-fix-entity-framework-asp-net-core/
            JSSettings = new JsonSerializerSettings()
            {
                ReferenceLoopHandling = ReferenceLoopHandling.Ignore
            };
            Log.Info("Startup completed");
        }

        #endregion Public Constructors

        #region Public Methods

        /// <summary>
        /// <inheritdoc />
        /// </summary>
        public bool Delete(string key)
        {
            Log.Trace($"Set request for {key}");
            return _db.KeyDelete(key);
        }

        /// <summary>
        /// <inheritdoc />
        /// </summary>
        public async Task<bool> DeleteAsync(string key)
        {
            Log.Trace($"SetAsync request for {key}");
            return await _db.KeyDeleteAsync(key);
        }

        /// <summary>
        /// <inheritdoc />
        /// </summary>
        public async Task<bool> FlushPatternAsync(RedisValue pattern)
        {
            try
            {
                var connServ = _connection.GetEndPoints()
                    .Select(endpoint => _connection.GetServer(endpoint))
                    .Where(x => x.IsConnected && !x.IsReplica)
                    .FirstOrDefault();

                if (connServ == null)
                {
                    Log.Error("Master REDIS server non trovato");
                    return false;
                }

                if (pattern.Equals(RedisValue.Null) || pattern.ToString() == "*")
                {
                    connServ.FlushDatabase(database: _db.Database);
                    return true;
                }

                var keys = connServ.Keys(database: _db.Database, pattern: pattern, pageSize: 1000);
                var deleteTasks = new List<Task>();

                foreach (var key in keys)
                {
                    deleteTasks.Add(_db.KeyDeleteAsync(key));
                    if (deleteTasks.Count >= 1000)
                    {
                        await Task.WhenAll(deleteTasks);
                        deleteTasks.Clear();
                    }
                }

                if (deleteTasks.Count > 0)
                    await Task.WhenAll(deleteTasks);

                return true;
            }
            catch (Exception ex)
            {
                Log.Error($"Eccezione durante FlushPatternAsync: {ex}");
                return false;
            }
        }

        /// <summary>
        /// <inheritdoc />
        /// </summary>
        public string? Get(string key)
        {
            Log.Trace($"Get request for {key}");
            var value = _db.StringGet(key);
            return value.HasValue ? value.ToString() : null;
        }

        /// <summary>
        /// <inheritdoc />
        /// </summary>
        public async Task<string?> GetAsync(string key)
        {
            Log.Trace($"GetAsync request for {key}");
            var value = await _db.StringGetAsync(key);
            return value.HasValue ? value.ToString() : null;
        }

        /// <summary>
        /// <inheritdoc />
        /// </summary>
        public async Task<List<HashEntry>> HashGetAllAsync(string hashKey)
        {
            Log.Trace($"HashGetAllAsync request for {hashKey}");
            var results = await _db.HashGetAllAsync(hashKey);
            return results.ToList();
        }

        /// <summary>
        /// <inheritdoc />
        /// </summary>
        public async Task<RedisValue?> HashGetAsync(string hashKey, string field)
        {
            Log.Trace($"HashGetAsync request for {hashKey}.{field}");
            return await _db.HashGetAsync(hashKey, field);
        }

        /// <summary>
        /// <inheritdoc />
        /// </summary>
        public long Publish(string channel, string message)
        {
            Log.Trace($"Publish: channel {channel}");
            RedisChannel rChannel = new RedisChannel(channel, RedisChannel.PatternMode.Literal);
            long numCli = _subscriber.Publish(rChannel, message);
            return numCli;
        }

        /// <summary>
        /// <inheritdoc />
        /// </summary>
        public async Task<long> PublishAsync(string channel, string message)
        {
            Log.Trace($"PublishAsync: channel {channel}");
            RedisChannel rChannel = new RedisChannel(channel, RedisChannel.PatternMode.Literal);
            long numCli = await _subscriber.PublishAsync(rChannel, message);
            return numCli;
        }

        /// <summary>
        /// <inheritdoc />
        /// </summary>
        public long QueueCount(RedisKey queueName)
        {
            return _db.ListLength(queueName);
        }

        /// <summary>
        /// <inheritdoc />
        /// </summary>
        public async Task<long> QueueCountAsync(RedisKey queueName)
        {
            return await _db.ListLengthAsync(queueName);
        }

        /// <summary>
        /// <inheritdoc />
        /// </summary>
        public List<RedisValue> QueueListAll(RedisKey queueName)
        {
            // lettura + reset in blocco
            var listData = _db.ListRange(queueName, 0, -1).ToList();
            return listData;
        }

        /// <summary>
        /// <inheritdoc />
        /// </summary>
        public async Task<List<RedisValue>> QueueListAllAsync(RedisKey queueName)
        {
            // lettura + reset in blocco
            var listData = await _db.ListRangeAsync(queueName, 0, -1);
            return listData.ToList();
        }

        /// <summary>
        /// <inheritdoc />
        /// </summary>
        public RedisValue QueuePop(RedisKey queueName)
        {
            return _db.ListLeftPop(queueName);
        }

        /// <summary>
        /// <inheritdoc />
        /// </summary>
        public List<RedisValue> QueuePopAll(RedisKey queueName)
        {
            // lettura + reset in blocco
            var listData = _db.ListRange(queueName, 0, -1).ToList();
            if (listData.Count > 0)
            {
                _db.KeyDelete(queueName); // remove the entire list
            }
            return listData;
        }

        /// <summary>
        /// <inheritdoc />
        /// </summary>
        public async Task<List<RedisValue>> QueuePopAllAsync(RedisKey queueName)
        {
            // lettura + reset in blocco
            var rawData = await _db.ListRangeAsync(queueName, 0, -1);
            var listData = rawData.ToList();
            if (listData.Count > 0)
            {
                _db.KeyDelete(queueName); // remove the entire list
            }
            return listData;
        }

        /// <summary>
        /// <inheritdoc />
        /// </summary>
        public async Task<RedisValue> QueuePopAsync(RedisKey queueName)
        {
            return await _db.ListLeftPopAsync(queueName);
        }

        /// <summary>
        /// <inheritdoc />
        /// </summary>
        public List<RedisValue> QueuePopList(RedisKey queueName, int maxElem)
        {
            // nuovo metodo con rimozione
            var results = new List<RedisValue>(maxElem);
            for (int i = 0; i < maxElem; i++)
            {
                var item = _db.ListLeftPop(queueName);
                if (item.IsNull) break; // queue empty
                results.Add(item);
            }
            return results;
        }

        /// <summary>
        /// <inheritdoc />
        /// </summary>
        public async Task<List<RedisValue>> QueuePopListAsync(RedisKey queueName, int maxElem)
        {
            // nuovo metodo con rimozione
            var results = new List<RedisValue>(maxElem);
            for (int i = 0; i < maxElem; i++)
            {
                var item = await _db.ListLeftPopAsync(queueName);
                if (item.IsNull) break; // queue empty
                results.Add(item);
            }
            return results;
        }

        /// <summary>
        /// <inheritdoc />
        /// </summary>
        public long QueuePush(RedisKey queueName, RedisValue value)
        {
            long qLen = _db.ListRightPush(queueName, value);
            return qLen;
        }

        /// <summary>
        /// <inheritdoc />
        /// </summary>
        public async Task<long> QueuePushAsync(RedisKey queueName, RedisValue value)
        {
            long qLen = await _db.ListRightPushAsync(queueName, value);
            return qLen;
        }

        /// <summary>
        /// <inheritdoc />
        /// </summary>
        public long QueueRemove(RedisKey queueName, RedisValue value)
        {
            // count = 0 → rimuove TUTTE le occorrenze del valore
            return _db.ListRemove(queueName, value, 0);
        }

        /// <summary>
        /// <inheritdoc />
        /// </summary>
        public async Task<long> QueueRemoveAsync(RedisKey queueName, RedisValue value)
        {
            // count = 0 → rimuove TUTTE le occorrenze del valore
            return await _db.ListRemoveAsync(queueName, value, 0);
        }

        /// <summary>
        /// <inheritdoc />
        /// </summary>
        public bool QueueReset(RedisKey queueName)
        {
            bool answ = _db.KeyDelete(queueName);
            return answ;
        }

        /// <summary>
        /// <inheritdoc />
        /// </summary>
        public async Task<bool> QueueResetAsync(RedisKey queueName)
        {
            bool answ = await _db.KeyDeleteAsync(queueName);
            return answ;
        }

        /// <summary>
        /// <inheritdoc />
        /// </summary>
        public bool Set(string key, string value, TimeSpan? tsExpiry = null)
        {
            Log.Trace($"Set request for {key}");
            return _db.StringSet(key, value, tsExpiry, When.Always);
        }

        /// <summary>
        /// <inheritdoc />
        /// </summary>
        public async Task<bool> SetAsync(string key, string value, TimeSpan? tsExpiry = null)
        {
            Log.Trace($"SetAsync request for {key}");
            return await _db.StringSetAsync(key, value, tsExpiry, When.Always);
        }

        /// <summary>
        /// <inheritdoc />
        /// </summary>
        public void Subscribe(string channel, Action<RedisChannel, RedisValue> handler)
        {
            Log.Trace($"Subscribed to channel {channel}");
            RedisChannel rChannel = new RedisChannel(channel, RedisChannel.PatternMode.Literal);
            _subscriber.Subscribe(rChannel, handler);
        }

        #endregion Public Methods

        #region Private Fields

        private static Logger Log = LogManager.GetCurrentClassLogger();

        private readonly IConnectionMultiplexer _connection;

        private readonly IDatabase _db;

        private readonly ISubscriber _subscriber;

        /// <summary>
        /// conf speciale serializzatore  JSON
        /// </summary>
        private JsonSerializerSettings? JSSettings;

        #endregion Private Fields
    }
}