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

using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Jellyfin.Data.Enums;
using Jellyfin.Plugin.RecentlyAddedGuard.Core;
using Jellyfin.Plugin.RecentlyAddedGuard.Ledger;
using MediaBrowser.Controller.Entities;
using MediaBrowser.Controller.Entities.TV;
using MediaBrowser.Controller.Library;
using Microsoft.Extensions.Logging;

namespace Jellyfin.Plugin.RecentlyAddedGuard.Services;

/// <summary>
/// Seeds and synchronizes the ledger with the library. Only registers items; never changes them.
/// </summary>
public sealed class LedgerSynchronizer : IDisposable
{
    private const int BatchSize = 500;

    private static readonly BaseItemKind[] _trackedKinds = [BaseItemKind.Movie, BaseItemKind.Episode];

    private readonly SemaphoreSlim _gate = new(1, 1);
    private readonly ILibraryManager _libraryManager;
    private readonly IdentityResolver _resolver;
    private readonly Reconciler _reconciler;
    private readonly LedgerStore _store;
    private readonly IExternalDateSource[] _externalSources;
    private readonly ILogger<LedgerSynchronizer> _logger;

    /// <summary>
    /// Initializes a new instance of the <see cref="LedgerSynchronizer"/> class.
    /// </summary>
    /// <param name="libraryManager">The library manager.</param>
    /// <param name="resolver">The identity resolver.</param>
    /// <param name="reconciler">The reconciler.</param>
    /// <param name="store">The ledger.</param>
    /// <param name="externalSources">External date sources (none in v1).</param>
    /// <param name="logger">The logger.</param>
    public LedgerSynchronizer(
        ILibraryManager libraryManager,
        IdentityResolver resolver,
        Reconciler reconciler,
        LedgerStore store,
        IEnumerable<IExternalDateSource> externalSources,
        ILogger<LedgerSynchronizer> logger)
    {
        _libraryManager = libraryManager;
        _resolver = resolver;
        _reconciler = reconciler;
        _store = store;
        _externalSources = externalSources.ToArray();
        _logger = logger;
    }

    /// <summary>
    /// Registers all movies and episodes, detaches items that no longer exist and prunes old entries.
    /// The first run on an empty ledger is the seed. Concurrent calls wait for each other.
    /// </summary>
    /// <param name="progress">Reports progress from 0 to 100.</param>
    /// <param name="cancellationToken">The cancellation token.</param>
    /// <returns>A task that completes when the ledger is synchronized and saved.</returns>
    public async Task RunAsync(IProgress<double> progress, CancellationToken cancellationToken)
    {
        ArgumentNullException.ThrowIfNull(progress);

        await _gate.WaitAsync(cancellationToken).ConfigureAwait(false);
        try
        {
            await RunCoreAsync(progress, cancellationToken).ConfigureAwait(false);
        }
        finally
        {
            _gate.Release();
        }
    }

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

    private async Task RunCoreAsync(IProgress<double> progress, CancellationToken cancellationToken)
    {
        var isSeed = !_store.SeedCompleted;
        var stopwatch = Stopwatch.StartNew();
        progress.Report(0);

        // Ids first, then the items in batches: keeps memory flat for libraries with tens of thousands of episodes.
        var itemIds = _libraryManager.GetItemIds(new InternalItemsQuery
        {
            IncludeItemTypes = _trackedKinds,
            Recursive = true,
            IsVirtualItem = false
        });

        var seen = new HashSet<Guid>(itemIds.Count);
        var seriesCache = new Dictionary<Guid, Series?>();
        int registered = 0, lowered = 0, ignored = 0, external = 0;

        for (var offset = 0; offset < itemIds.Count; offset += BatchSize)
        {
            cancellationToken.ThrowIfCancellationRequested();

            var batch = itemIds.Skip(offset).Take(BatchSize).ToArray();
            var items = _libraryManager.GetItemList(new InternalItemsQuery
            {
                ItemIds = batch,
                IncludeItemTypes = _trackedKinds,
                Recursive = true,
                IsVirtualItem = false
            });

            foreach (var item in items)
            {
                seen.Add(item.Id);
                var identity = _resolver.Resolve(item, seriesId => LookupSeries(seriesId, seriesCache));
                if (identity is null || identity.Keys.Count == 0)
                {
                    ignored++;
                    continue;
                }

                switch (_reconciler.Register(identity).Action)
                {
                    case ReconcileAction.Registered:
                        registered++;
                        break;
                    case ReconcileAction.FirstAddedLowered:
                        lowered++;
                        break;
                    default:
                        break;
                }

                if (_externalSources.Length > 0 && await ApplyExternalSourcesAsync(identity, cancellationToken).ConfigureAwait(false))
                {
                    external++;
                }
            }

            progress.Report(90.0 * Math.Min(offset + BatchSize, itemIds.Count) / itemIds.Count);
        }

        // Live items the query did not return may be alternate versions or added meanwhile; only drop the ones that are gone.
        var stale = _reconciler.GetUnseenLiveItemIds(seen)
            .Where(itemId => _libraryManager.GetItemById(itemId) is null)
            .ToList();
        var detached = _reconciler.RemoveStaleItems(stale);
        var pruned = _reconciler.Prune();

        _store.MarkSynced(DateTime.UtcNow);
        _store.Flush();
        progress.Report(100);

        var statistics = _store.GetStatistics();
        _logger.LogInformation(
            "{Kind} abgeschlossen in {Elapsed:g}: {Items} Items geprüft, {Registered} neu registriert, {Lowered} Datum vorverlegt, "
            + "{External} aus externen Quellen, {Ignored} ohne Kennung übersprungen, {Detached} entfernte Items ausgetragen, {Pruned} Einträge bereinigt. "
            + "Ledger: {Entries} Einträge ({Live} mit Items, {Removed} in der Frist)",
            isSeed ? "Ledger-Seed" : "Ledger-Synchronisation",
            stopwatch.Elapsed,
            itemIds.Count,
            registered,
            lowered,
            external,
            ignored,
            detached,
            pruned,
            statistics.Entries,
            statistics.LiveEntries,
            statistics.RemovedEntries);
    }

    private Series? LookupSeries(Guid seriesId, Dictionary<Guid, Series?> cache)
    {
        if (!cache.TryGetValue(seriesId, out var series))
        {
            series = _libraryManager.GetItemById(seriesId) as Series;
            cache[seriesId] = series;
        }

        return series;
    }

#pragma warning disable CA1031 // A failing external source must not abort the synchronization.
    private async Task<bool> ApplyExternalSourcesAsync(MediaIdentity identity, CancellationToken cancellationToken)
    {
        var lowered = false;
        foreach (var source in _externalSources)
        {
            try
            {
                if (await source.GetFirstAddedAsync(identity, cancellationToken).ConfigureAwait(false) is { } firstAdded)
                {
                    lowered |= _reconciler.ApplyExternalFirstAdded(identity.Keys, firstAdded);
                }
            }
            catch (Exception ex) when (ex is not OperationCanceledException)
            {
                _logger.LogWarning(ex, "Externe Datumsquelle {Source} ist für {Name} fehlgeschlagen", source.Name, identity.DisplayName);
            }
        }

        return lowered;
    }
#pragma warning restore CA1031
}