proxd-man/Proxd-man/Subscribe/SubscribeManager.cs

using System.Buffers.Text;
using System.Collections.Immutable;
using System.Text;
using System.Text.Json;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using Proxd_man.Config;
using Proxd_man.Config.Outbound;
using Proxd_man.ShareLink;

namespace Proxd_man.Subscribe;

public class SubscribeManager
{
    
    private readonly Dictionary<SubscribeSource, List<BaseOutboundSetting>> _subscribeOutbounds = new();
    private CancellationTokenSource? _cts;
    private readonly ILogger<SubscribeManager> _logger;
    private readonly IOptionsMonitor<ProxdConfig> _config;

    public SubscribeManager(ILogger<SubscribeManager> logger, IOptionsMonitor<ProxdConfig> config)
    {
        _logger = logger;
        _config = config;

        _config.OnChange((_, _) => Task.Run(async () => await CheckAndUpdateSubscribes()));
    }

    public delegate void SubscribeUpdatedEventHandler(IDictionary<SubscribeSource, List<BaseOutboundSetting>> subscribeOutbounds);
    public event SubscribeUpdatedEventHandler? SubscribeUpdated;

    private ProxdConfig Config => _config.Get(null);
    
    public async Task CheckAndUpdateSubscribes(CancellationToken cancellationToken = default)
    {
        var httpClient = new HttpClient();

        _logger.LogInformation("开始更新订阅...");
        
        var newSubscribeOutbounds = new Dictionary<SubscribeSource, List<BaseOutboundSetting>>();
        for (var i = 0; i < Config.Subscribe.Subscribes.Count; i++)
        {
            var subscribe = Config.Subscribe.Subscribes[i];
            var outbounds = new List<BaseOutboundSetting>();

            _logger.LogInformation("正在更新订阅 {subscribeName} ({current}/{total})...", subscribe.Name, i + 1, Config.Subscribe.Subscribes.Count);
            try
            {
                var response = await httpClient.GetAsync(subscribe.Url, cancellationToken);
                response.EnsureSuccessStatusCode();
                var content = await response.Content.ReadAsStringAsync(cancellationToken);

                if (Base64.IsValid(content))
                {
                    content = Encoding.UTF8.GetString(Convert.FromBase64String(content));
                }
                
                var shareLinks = content.Split('\n').Select(l => l.Trim()).Where(l => !string.IsNullOrEmpty(l));
                foreach (var shareLink in shareLinks)
                {
                    try
                    {
                        var outbound = ShareLinkParser.Parse(shareLink);
                        outbounds.Add(outbound);
                    }
                    catch (InvalidShareLinkException e)
                    {
                        _logger.LogWarning(e, "解析分享链接失败: {shareLink}\n", shareLink);
                    }
                }
            }
            catch (Exception e)
            {
                _logger.LogWarning(e, "更新订阅失败: {subscribeName}", subscribe.Name);
            }

            if (outbounds.Count > 0)
            {
                newSubscribeOutbounds[subscribe] = outbounds;
            }
            _logger.LogInformation("订阅 {subscribeName} 更新完成. (共 {outboundCount} 个节点)", subscribe.Name, outbounds.Count);
        }
        
        _subscribeOutbounds.Clear();
        foreach (var (source, outbounds) in newSubscribeOutbounds)
        {
            _subscribeOutbounds[source] = outbounds;
        }
        
        _logger.LogInformation("订阅更新完成.");
        SubscribeUpdated?.Invoke(_subscribeOutbounds.ToImmutableDictionary());
    }
    
    public IDictionary<SubscribeSource, List<BaseOutboundSetting>> GetSubscribeOutbounds()
    {
        return _subscribeOutbounds.ToImmutableDictionary();
    }
    
    public void StartUpdateSubscribes()
    {
        if (_cts != null)
        {
            StopUpdateSubscribes();
        }
        
        var cts = new CancellationTokenSource();
        
        Task.Run(async () =>
        {
            while (!cts.Token.IsCancellationRequested)
            {
                if (_subscribeOutbounds.Count > 0)
                {
                    _logger.LogDebug("订阅更新完成, 将在 12 小时后再次更新.");
                    await Task.Delay(TimeSpan.FromHours(12), cts.Token);
                }
                else
                {
                    _logger.LogDebug("没有可用的订阅, 将在 5 分钟后再次更新.");
                    await Task.Delay(TimeSpan.FromMinutes(5), cts.Token);
                }
                
                _logger.LogInformation("开始更新订阅...");
                await CheckAndUpdateSubscribes(cts.Token);
            }
        }, cts.Token);
        _cts = cts;
        _logger.LogInformation("订阅更新任务已启动.");
    }
    
    public void StopUpdateSubscribes()
    {
        _cts?.Cancel();
        _cts = null;
    }
    
}