track/src/SJZY.Track.Application/Services/ForewarningServices/ForewarningCacheService.cs

using System.Collections.Concurrent;
using SJZY.Expand.ABP.Core;
using SJZY.Expand.ABP.Core.PublicMethords;
using SJZY.Track.Application.Contract.Dtos.Forewarning;
using SJZY.Track.Application.Contract.Events;
using SJZY.Track.Application.Contract.Services;
using SJZY.Track.Domain.Aggregates.KPIAreaAggregate;
using SJZY.Track.Domain.Dtos.Forewarning;
using SJZY.Track.Domain.Queries;
using Volo.Abp.DependencyInjection;

namespace SJZY.Track.Application.Services.ForewarningServices;

public class ForewarningCacheService : IForewarningCacheService, ISingletonDependency
{
    private const string GlobalWaningLockKey = "GlobalWaningLockKey_GlobalLock";
    private readonly ConcurrentDictionary<int, List<AreaCacheEntry>> _areas = new();
    readonly IKpiAreaRepository _kpiAreaRepository;

    private readonly ConcurrentDictionary<string, List<ForewarningRuleCacheEntry>> _rules = new();
    private readonly IWarningQueries _warningQueries;
    object _lock = new object();

    public ForewarningCacheService(IKpiAreaRepository kpiAreaRepository, IWarningQueries warningQueries)
    {
        _kpiAreaRepository = kpiAreaRepository;
        _warningQueries = warningQueries;
    }

    public async Task<List<ForewarningRuleCacheEntry>?> GetRulesAsync(int subSystemId, int customerId)
    {
        var key = WarningKey(subSystemId, customerId);
        if (_rules.TryGetValue(key, out var rules))
        {
            return rules;
        }

        return await GetRules(subSystemId, customerId);
    }

    public Task Clear(ClearRulCacheEvent @event)
    {
        foreach (var item in @event.Items)
        {
            var redisKey = WarningKey(item.SystemId, item.CustomerId);
            _rules.Remove(redisKey, out _);
        }

        return Task.CompletedTask;
    }

    public async Task ClearRedisBySystemIdAndCustomer(int subSystemId, int customerId)
    {
        await RedisHelper.Client.DelAsync(WarningKey(subSystemId, customerId));
        var eventModel = new ClearRulCacheEvent
        {
            Items = new List<RuleAreaBaseInfo>()
            {
                new RuleAreaBaseInfo(subSystemId, customerId)
            }
        };

        await PublishEvent(eventModel);
    }

    public async Task ClearRedisByAreaId(int areaId)
    {
        var baseInfos = await _warningQueries.GetWarningRuleByAreaId(areaId);
        if (!baseInfos.Any()) return;
        baseInfos = baseInfos.DistinctBy(val => WarningKey(val.SystemId, val.CustomerId)).ToList();

        var willDelKeys = baseInfos.Select(val => WarningKey(val.SystemId, val.CustomerId)).ToArray();
        await RedisHelper.Client.DelAsync(willDelKeys);
        await RedisHelper.Client.DelAsync(AreaKey(areaId));
        var eventModel = new ClearRulCacheEvent
        {
            Items = baseInfos.Select(x => new RuleAreaBaseInfo
            {
                SystemId = x.SystemId,
                CustomerId = x.CustomerId
            }).ToList()
        };

        await PublishEvent(eventModel);
    }

    string WarningKey(int subSystemId, int customerId) =>
        $"{GlobalConst.TrackWarningKeyPrefix}:{subSystemId}:{customerId}";

    string AreaKey(int areaId) => $"{GlobalConst.TrackAreaKeyPrefix}:{areaId}";

    private List<ForewarningRuleCacheEntry> WriteRules(List<WarningRuleCacheDto> rules)
    {
        var result = new List<ForewarningRuleCacheEntry>();
        foreach (var item in rules)
        {
            List<AreaCacheEntry>? areaList = null;
            if (item.AreaId.HasValue)
            {
                areaList = GetAreaInfoByAreaId(item.AreaId!.Value);
            }

            var entry = new ForewarningRuleCacheEntry
            {
                Id = item.Id,
                WarningId = item.WarningId,
                TotalTimeSpan = item.TotalTimeSpan,
                WarnTimeSpan = item.WarnTimeSpan,
                WarnCount = item.WarnCount,
                StartTrackStatusId = item.StartTrackStatusId,
                EndTrackStatusId = item.EndTrackStatusId,
                CargoType = item.CargoType,
                Months = string.IsNullOrEmpty(item.Months)
                    ? new List<int>()
                    : item.Months.Split('/').Select(int.Parse).ToList(),
                IsTransit = item.IsTransit,
                Areas = areaList,
                ProductCode = item.ProductCode,
                LeavePort = item.LeavePort
            };

            result.Add(entry);
        }

        return result;
    }

    private async Task<List<ForewarningRuleCacheEntry>?> GetRules(int systemId, int customerId)
    {
        var key = WarningKey(systemId, customerId);
        var result = RedisHelper.Client.Get<List<WarningRuleCacheDto>>(key);
        if (result != null)
        {
            // 非线程安全，需要加锁
            lock (_lock)
            {
                if (_rules.TryGetValue(key, out var ruleResult))
                {
                    return ruleResult;
                }

                ruleResult = WriteRules(result);
                _rules.TryAdd(key, ruleResult);
                return ruleResult;
            }
        }

        // 非线程安全，需要加锁
        using (RedisHelper.Client.Lock(GlobalWaningLockKey, 20))
        {
            if (_rules.TryGetValue(key, out var ruleResult)) return ruleResult;

            var warnings = await _warningQueries.GetWarningRule(systemId, customerId);
            if (warnings == null || warnings.Count == 0)
            {
                await RedisHelper.Client.SetAsync(key, new List<WarningRuleCacheDto>(), TimeSpan.FromSeconds(20));
                return new List<ForewarningRuleCacheEntry>();
            }

            await RedisHelper.Client.SetAsync(key, warnings);
            ruleResult = WriteRules(warnings);
            _rules.TryAdd(key, ruleResult);
            return ruleResult;
        }
    }

    private List<AreaCacheEntry>? GetAreaInfoByAreaId(int areaId)
    {
        if (_areas.TryGetValue(areaId, out var areaInfos)) return areaInfos;

        var areas = GetAreas(areaId);
        if (areas.Count == 0) return null;

        areaInfos = areas.Select(KpiAreaInfoToEntry).ToList();
        _areas.TryAdd(areaId, areaInfos);
        return areaInfos;
    }

    AreaCacheEntry KpiAreaInfoToEntry(KpiAreaInfo model)
    {
        return new AreaCacheEntry
        {
            Country = model.Country,
            Province = model.Province,
            City = model.City,
            LeftPostcode = model.LeftPostcode,
            RightPostcode = model.RightPostcode
        };
    }

    /// <summary>
    /// 获取区域
    /// </summary>
    /// <returns></returns>
    private List<KpiAreaInfo> GetAreas(int areaId)
    {
        var key = AreaKey(areaId);
        var result = RedisHelper.Client.Get<List<KpiAreaInfo>>(key);
        if (result == null)
        {
            result = RedisHelper.Client.Get<List<KpiAreaInfo>>(key);
            if (result != null) return new List<KpiAreaInfo>();

            var kpiArea = _kpiAreaRepository.GetAreaInfosByAreaId(areaId);
            if (kpiArea == null)
            {
                RedisHelper.Client.Set(key, new List<KpiAreaInfo>(), TimeSpan.FromSeconds(20));
                return new List<KpiAreaInfo>();
            }

            RedisHelper.Client.Set(key, kpiArea);
            return kpiArea;
        }

        return result;
    }

    private async Task PublishEvent(ClearRulCacheEvent @event)
    {
        await RedisHelper.Client.PublishAsync(GlobalConst.PublishNotifyChannelKey, @event.ToJson());
    }
}