LightlessServer/LightlessSyncServer/LightlessSyncStaticFilesServer/Services/ShardFileInventoryReporter.cs

using LightlessSync.API.Routes;
using LightlessSyncShared.Models;
using LightlessSyncShared.Services;
using LightlessSyncShared.Utils;
using LightlessSyncShared.Utils.Configuration;
using System.Net.Http.Json;
using System.Linq;

namespace LightlessSyncStaticFilesServer.Services;

public interface IShardFileInventoryReporter
{
    void ReportAdded(string hash);
    void ReportRemoved(string hash);
}

public sealed class NullShardFileInventoryReporter : IShardFileInventoryReporter
{
    public void ReportAdded(string hash) { }
    public void ReportRemoved(string hash) { }
}

public sealed class ShardFileInventoryReporter : IHostedService, IShardFileInventoryReporter
{
    private static readonly TimeSpan ResyncInterval = TimeSpan.FromMinutes(30);
    private static readonly TimeSpan RetryDelay = TimeSpan.FromSeconds(10);
    private static readonly TimeSpan BatchDelay = TimeSpan.FromSeconds(2);

    private readonly IConfigurationService<StaticFilesServerConfiguration> _configurationService;
    private readonly ILogger<ShardFileInventoryReporter> _logger;
    private readonly HttpClient _httpClient = new();
    private readonly string _cacheDir;
    private readonly object _pendingLock = new();
    private readonly SemaphoreSlim _signal = new(0);

    private HashSet<string> _pendingAdds = new(StringComparer.OrdinalIgnoreCase);
    private HashSet<string> _pendingRemoves = new(StringComparer.OrdinalIgnoreCase);
    private CancellationTokenSource? _cts;
    private Task? _processTask;
    private Task? _resyncTask;
    private long _sequence;
    private bool _resyncRequested;

    public ShardFileInventoryReporter(
        IConfigurationService<StaticFilesServerConfiguration> configurationService,
        ILogger<ShardFileInventoryReporter> logger,
        ServerTokenGenerator serverTokenGenerator)
    {
        _configurationService = configurationService;
        _logger = logger;
        _cacheDir = configurationService.GetValue<string>(nameof(StaticFilesServerConfiguration.CacheDirectory));
        _httpClient.DefaultRequestHeaders.Authorization =
            new System.Net.Http.Headers.AuthenticationHeaderValue("Bearer", serverTokenGenerator.Token);
    }

    public Task StartAsync(CancellationToken cancellationToken)
    {
        _cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
        _resyncRequested = true;
        _signal.Release();

        _processTask = Task.Run(() => ProcessUpdatesAsync(_cts.Token), _cts.Token);
        _resyncTask = Task.Run(() => ResyncLoopAsync(_cts.Token), _cts.Token);
        return Task.CompletedTask;
    }

    public async Task StopAsync(CancellationToken cancellationToken)
    {
        if (_cts == null)
            return;

        _cts.Cancel();

        try
        {
            if (_processTask != null) await _processTask.ConfigureAwait(false);
        }
        catch
        {
            // ignore
        }

        try
        {
            if (_resyncTask != null) await _resyncTask.ConfigureAwait(false);
        }
        catch
        {
            // ignore
        }

        _httpClient.Dispose();
    }

    public void ReportAdded(string hash)
    {
        if (!IsValidHash(hash))
            return;

        lock (_pendingLock)
        {
            _pendingAdds.Add(hash);
            _pendingRemoves.Remove(hash);
        }

        _signal.Release();
    }

    public void ReportRemoved(string hash)
    {
        if (!IsValidHash(hash))
            return;

        lock (_pendingLock)
        {
            _pendingRemoves.Add(hash);
            _pendingAdds.Remove(hash);
        }

        _signal.Release();
    }

    private async Task ResyncLoopAsync(CancellationToken ct)
    {
        while (!ct.IsCancellationRequested)
        {
            try
            {
                await Task.Delay(ResyncInterval, ct).ConfigureAwait(false);
            }
            catch (OperationCanceledException)
            {
                break;
            }

            _resyncRequested = true;
            _signal.Release();
        }
    }

    private async Task ProcessUpdatesAsync(CancellationToken ct)
    {
        while (!ct.IsCancellationRequested)
        {
            try
            {
                await _signal.WaitAsync(BatchDelay, ct).ConfigureAwait(false);
            }
            catch (OperationCanceledException)
            {
                break;
            }

            while (!ct.IsCancellationRequested)
            {
                ShardFileInventoryUpdateDto? update = null;

                if (_resyncRequested)
                {
                    var snapshot = BuildSnapshot();
                    update = new ShardFileInventoryUpdateDto
                    {
                        Sequence = Interlocked.Increment(ref _sequence),
                        IsFullSnapshot = true,
                        Added = snapshot
                    };
                }
                else
                {
                    HashSet<string> adds;
                    HashSet<string> removes;

                    lock (_pendingLock)
                    {
                        if (_pendingAdds.Count == 0 && _pendingRemoves.Count == 0)
                            break;

                        adds = _pendingAdds;
                        removes = _pendingRemoves;
                        _pendingAdds = new HashSet<string>(StringComparer.OrdinalIgnoreCase);
                        _pendingRemoves = new HashSet<string>(StringComparer.OrdinalIgnoreCase);
                    }

                    update = new ShardFileInventoryUpdateDto
                    {
                        Sequence = Interlocked.Increment(ref _sequence),
                        Added = adds.ToList(),
                        Removed = removes.ToList()
                    };
                }

                if (update == null)
                    break;

                await SendUpdateWithRetryAsync(update, ct).ConfigureAwait(false);

                if (update.IsFullSnapshot)
                {
                    lock (_pendingLock)
                    {
                        _pendingAdds.Clear();
                        _pendingRemoves.Clear();
                        _resyncRequested = false;
                    }
                }
            }
        }
    }

    private async Task SendUpdateWithRetryAsync(ShardFileInventoryUpdateDto update, CancellationToken ct)
    {
        while (!ct.IsCancellationRequested)
        {
            try
            {
                await SendUpdateAsync(update, ct).ConfigureAwait(false);
                return;
            }
            catch (OperationCanceledException) when (ct.IsCancellationRequested)
            {
                throw;
            }
            catch (Exception ex)
            {
                _logger.LogWarning(ex, "Failed to send shard file inventory update (seq {seq})", update.Sequence);
                try
                {
                    await Task.Delay(RetryDelay, ct).ConfigureAwait(false);
                }
                catch (OperationCanceledException)
                {
                    throw;
                }
            }
        }
    }

    private async Task SendUpdateAsync(ShardFileInventoryUpdateDto update, CancellationToken ct)
    {
        var mainServer = _configurationService.GetValue<Uri>(nameof(StaticFilesServerConfiguration.MainFileServerAddress));
        if (mainServer == null)
            throw new InvalidOperationException("Main server address is not configured.");

        using var response = await _httpClient.PostAsJsonAsync(
            LightlessFiles.MainShardFilesFullPath(mainServer),
            update,
            ct).ConfigureAwait(false);

        response.EnsureSuccessStatusCode();

        var ack = await response.Content.ReadFromJsonAsync<ShardFileInventoryUpdateAckDto>(cancellationToken: ct)
            .ConfigureAwait(false);

        if (ack == null || ack.AppliedSequence < update.Sequence)
            throw new InvalidOperationException($"Main server did not apply update {update.Sequence}.");
    }

    private List<string> BuildSnapshot()
    {
        var hashes = new List<string>();

        if (string.IsNullOrWhiteSpace(_cacheDir) || !Directory.Exists(_cacheDir))
            return hashes;

        foreach (var file in Directory.EnumerateFiles(_cacheDir, "*", SearchOption.AllDirectories))
        {
            var name = Path.GetFileName(file);
            if (name.EndsWith(".dl", StringComparison.OrdinalIgnoreCase))
                continue;

            if (IsValidHash(name))
                hashes.Add(name);
        }

        return hashes;
    }

    private static bool IsValidHash(string hash)
    {
        return hash.Length == 40 && hash.All(char.IsAsciiLetterOrDigit);
    }
}