jellyfin-plugin-recently-added-guard/Jellyfin.Plugin.RecentlyAddedGuard/Services/GuardHostedService.cs

using System;
using System.Collections.Generic;
using System.IO;
using System.Threading;
using System.Threading.Channels;
using System.Threading.Tasks;
using Jellyfin.Plugin.RecentlyAddedGuard.Core;
using Jellyfin.Plugin.RecentlyAddedGuard.Ledger;
using MediaBrowser.Controller.Entities;
using MediaBrowser.Controller.Library;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;

namespace Jellyfin.Plugin.RecentlyAddedGuard.Services;

/// <summary>
/// Listens to library events and processes them on a background worker; runs the seed on first start.
/// </summary>
public sealed class GuardHostedService : IHostedService, IDisposable
{
    private static readonly TimeSpan _debounceDelay = TimeSpan.FromSeconds(2);

    private readonly ILibraryManager _libraryManager;
    private readonly IdentityResolver _resolver;
    private readonly GuardProcessor _processor;
    private readonly LibraryItemDateWriter _writer;
    private readonly LedgerStore _store;
    private readonly LedgerSynchronizer _synchronizer;
    private readonly ILogger<GuardHostedService> _logger;
    private readonly Channel<GuardEvent> _events = Channel.CreateUnbounded<GuardEvent>(
        new UnboundedChannelOptions { SingleReader = true, SingleWriter = false });

    private readonly CancellationTokenSource _stopping = new();
    private Task _seed = Task.CompletedTask;
    private Task _worker = Task.CompletedTask;

    /// <summary>
    /// Initializes a new instance of the <see cref="GuardHostedService"/> class.
    /// </summary>
    /// <param name="libraryManager">The library manager.</param>
    /// <param name="resolver">The identity resolver.</param>
    /// <param name="processor">The processor.</param>
    /// <param name="writer">The date writer.</param>
    /// <param name="store">The ledger.</param>
    /// <param name="synchronizer">The synchronizer.</param>
    /// <param name="logger">The logger.</param>
    public GuardHostedService(
        ILibraryManager libraryManager,
        IdentityResolver resolver,
        GuardProcessor processor,
        LibraryItemDateWriter writer,
        LedgerStore store,
        LedgerSynchronizer synchronizer,
        ILogger<GuardHostedService> logger)
    {
        _libraryManager = libraryManager;
        _resolver = resolver;
        _processor = processor;
        _writer = writer;
        _store = store;
        _synchronizer = synchronizer;
        _logger = logger;
    }

    /// <inheritdoc />
    public Task StartAsync(CancellationToken cancellationToken)
    {
        _store.Load();

        _libraryManager.ItemAdded += OnItemAddedOrUpdated;
        _libraryManager.ItemUpdated += OnItemAddedOrUpdated;
        _libraryManager.ItemRemoved += OnItemRemoved;

        if (!_store.SeedCompleted)
        {
            _seed = Task.Run(() => RunSeedAsync(_stopping.Token), CancellationToken.None);
        }

        _worker = Task.Run(() => RunWorkerAsync(_stopping.Token), CancellationToken.None);
        return Task.CompletedTask;
    }

    /// <inheritdoc />
    public async Task StopAsync(CancellationToken cancellationToken)
    {
        _libraryManager.ItemAdded -= OnItemAddedOrUpdated;
        _libraryManager.ItemUpdated -= OnItemAddedOrUpdated;
        _libraryManager.ItemRemoved -= OnItemRemoved;
        _events.Writer.TryComplete();

        // An unfinished seed starts over on the next start; pending events are still processed.
        if (!_seed.IsCompleted)
        {
            await _stopping.CancelAsync().ConfigureAwait(false);
        }

        try
        {
            await Task.WhenAll(_seed, _worker).WaitAsync(cancellationToken).ConfigureAwait(false);
        }
        catch (OperationCanceledException)
        {
            await _stopping.CancelAsync().ConfigureAwait(false);
        }

        try
        {
            _store.Flush();
        }
        catch (Exception ex) when (ex is IOException or UnauthorizedAccessException)
        {
            _logger.LogError(ex, "Ledger konnte beim Beenden nicht gespeichert werden");
        }
    }

    /// <inheritdoc />
    public void Dispose() => _stopping.Dispose();

    private void OnItemAddedOrUpdated(object? sender, ItemChangeEventArgs e)
    {
        var item = e.Item;
        if (item is null || !IdentityResolver.IsTracked(item) || _writer.IsWriting(item.Id))
        {
            return;
        }

        _events.Writer.TryWrite(new GuardEvent(item.Id, null));
    }

    private void OnItemRemoved(object? sender, ItemChangeEventArgs e)
    {
        var item = e.Item;
        if (item is null || !IdentityResolver.IsTracked(item))
        {
            return;
        }

        // Keep the object: its provider ids are needed later, the database row is already gone.
        _events.Writer.TryWrite(new GuardEvent(item.Id, item));
    }

#pragma warning disable CA1031 // Background loops log failures and keep running.
    private async Task RunSeedAsync(CancellationToken cancellationToken)
    {
        try
        {
            _logger.LogInformation("Ledger wurde noch nicht befüllt, Seed läuft im Hintergrund");
            await _synchronizer.RunAsync(new Progress<double>(), cancellationToken).ConfigureAwait(false);
        }
        catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
        {
            _logger.LogInformation("Ledger-Seed abgebrochen, er läuft beim nächsten Start erneut");
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Ledger-Seed fehlgeschlagen; der Scheduled Task \"Ledger synchronisieren\" holt ihn nach");
        }
    }

    private async Task RunWorkerAsync(CancellationToken cancellationToken)
    {
        try
        {
            // Events raised during the seed wait in the channel, so they are matched against a complete ledger.
            await _seed.ConfigureAwait(false);

            var reader = _events.Reader;
            var pending = new OrderedDictionary<Guid, GuardEvent>();
            while (await reader.WaitToReadAsync(cancellationToken).ConfigureAwait(false))
            {
                // A scan raises several events per item within moments; collect the burst, keep the latest per item.
                await Task.Delay(_debounceDelay, cancellationToken).ConfigureAwait(false);
                while (reader.TryRead(out var guardEvent))
                {
                    pending[guardEvent.ItemId] = guardEvent;
                }

                foreach (var guardEvent in pending.Values)
                {
                    await ProcessAsync(guardEvent, cancellationToken).ConfigureAwait(false);
                }

                pending.Clear();
            }
        }
        catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
        {
            // Shutting down.
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Event-Verarbeitung von Recently Added Guard wurde unerwartet beendet");
        }
    }

    private async Task ProcessAsync(GuardEvent guardEvent, CancellationToken cancellationToken)
    {
        try
        {
            if (guardEvent.RemovedItem is { } removed)
            {
                _processor.ProcessRemoved(guardEvent.ItemId, _resolver.ResolveKeysForRemoval(removed));
                return;
            }

            var item = _libraryManager.GetItemById(guardEvent.ItemId);
            var identity = item is null ? null : _resolver.Resolve(item);
            if (identity is not null)
            {
                await _processor.ProcessSeenAsync(identity, cancellationToken).ConfigureAwait(false);
            }
        }
        catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
        {
            throw;
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Item {ItemId} konnte nicht verarbeitet werden", guardEvent.ItemId);
        }
    }
#pragma warning restore CA1031

    /// <summary>
    /// A queued library event. <see cref="RemovedItem"/> is set for removals.
    /// </summary>
    private sealed record GuardEvent(Guid ItemId, BaseItem? RemovedItem);
}