pgsql-jellyfin/Jellyfin.Server.Implementations/Clustering/FileSystemChangeProcessor.cs

// <copyright file="FileSystemChangeProcessor.cs" company="PlaceholderCompany">
// Copyright (c) PlaceholderCompany. All rights reserved.
// </copyright>

namespace Jellyfin.Server.Implementations.Clustering
{
    using System;
    using System.Linq;
    using System.Threading;
    using System.Threading.Tasks;
    using Jellyfin.Database.Implementations;
    using Jellyfin.Database.Implementations.Entities;
    using MediaBrowser.Controller.Library;
    using Microsoft.EntityFrameworkCore;
    using Microsoft.Extensions.Hosting;
    using Microsoft.Extensions.Logging;

    /// <summary>
    /// Processes file system changes from the database.
    /// Only the primary instance processes changes; all instances can record them.
    /// </summary>
    public sealed class FileSystemChangeProcessor : IFileSystemChangeProcessor, IHostedService, IAsyncDisposable
    {
        private readonly IDbContextFactory<JellyfinDbContext> _dbContextFactory;
        private readonly IInstanceRegistry _instanceRegistry;
        private readonly IPrimaryElectionService _primaryElectionService;
        private readonly ILibraryManager? _libraryManager;
        private readonly ILogger<FileSystemChangeProcessor> _logger;
        private CancellationTokenSource? _processingCts;
        private Task? _processingTask;
        private bool _disposed;

        /// <summary>
        /// Initializes a new instance of the <see cref="FileSystemChangeProcessor"/> class.
        /// </summary>
        /// <param name="dbContextFactory">The database context factory.</param>
        /// <param name="instanceRegistry">The instance registry.</param>
        /// <param name="primaryElectionService">The primary election service.</param>
        /// <param name="libraryManager">The library manager (optional, may not be available at startup).</param>
        /// <param name="logger">The logger.</param>
        public FileSystemChangeProcessor(
            IDbContextFactory<JellyfinDbContext> dbContextFactory,
            IInstanceRegistry instanceRegistry,
            IPrimaryElectionService primaryElectionService,
            ILibraryManager? libraryManager,
            ILogger<FileSystemChangeProcessor> logger)
        {
            _dbContextFactory = dbContextFactory;
            _instanceRegistry = instanceRegistry;
            _primaryElectionService = primaryElectionService;
            _libraryManager = libraryManager;
            _logger = logger;
        }

        /// <inheritdoc/>
        Task IHostedService.StartAsync(CancellationToken cancellationToken)
        {
            return StartAsync(cancellationToken);
        }

        /// <inheritdoc/>
        Task IHostedService.StopAsync(CancellationToken cancellationToken)
        {
            return StopAsync(cancellationToken);
        }

        /// <inheritdoc/>
        public async Task StartAsync(CancellationToken cancellationToken = default)
        {
            _logger.LogInformation("Starting file system change processor");

            // Subscribe to primary changes
            _primaryElectionService.PrimaryInstanceChanged += OnPrimaryInstanceChanged;

            // If we're currently primary, start processing
            if (_primaryElectionService.IsPrimary)
            {
                StartProcessingLoop();
            }

            _logger.LogInformation("File system change processor started");
            await Task.CompletedTask.ConfigureAwait(false);
        }

        /// <inheritdoc/>
        public async Task StopAsync(CancellationToken cancellationToken = default)
        {
            _logger.LogInformation("Stopping file system change processor");

            _primaryElectionService.PrimaryInstanceChanged -= OnPrimaryInstanceChanged;

            StopProcessingLoop();

            _logger.LogInformation("File system change processor stopped");
            await Task.CompletedTask.ConfigureAwait(false);
        }

        /// <inheritdoc/>
        public async Task RecordChangeAsync(string path, string changeType, string? oldPath = null, CancellationToken cancellationToken = default)
        {
            try
            {
                await using var context = await _dbContextFactory.CreateDbContextAsync(cancellationToken).ConfigureAwait(false);

                var change = new FileSystemChange
                {
                    Path = path,
                    ChangeType = changeType,
                    OldPath = oldPath,
                    DetectedAt = DateTime.UtcNow,
                    DetectedBy = _instanceRegistry.CurrentInstanceId
                };

                context.FileSystemChanges.Add(change);
                await context.SaveChangesAsync(cancellationToken).ConfigureAwait(false);

                _logger.LogDebug(
                    "Recorded file system change: {ChangeType} - {Path}",
                    changeType,
                    path);
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "Failed to record file system change for {Path}", path);
            }
        }

        /// <inheritdoc/>
        public async ValueTask DisposeAsync()
        {
            if (_disposed)
            {
                return;
            }

            _disposed = true;

            await StopAsync(CancellationToken.None).ConfigureAwait(false);

            _processingCts?.Dispose();
        }

        private void OnPrimaryInstanceChanged(object? sender, PrimaryInstanceChangedEventArgs e)
        {
            if (e.IsCurrentInstance)
            {
                _logger.LogInformation("Became primary instance, starting file system change processing");
                StartProcessingLoop();
            }
            else
            {
                _logger.LogInformation("No longer primary instance, stopping file system change processing");
                StopProcessingLoop();
            }
        }

        private void StartProcessingLoop()
        {
            if (_processingTask is not null)
            {
                return; // Already running
            }

            _processingCts = new CancellationTokenSource();
            _processingTask = ProcessChangesLoopAsync(_processingCts.Token);
        }

        private void StopProcessingLoop()
        {
            if (_processingTask is null)
            {
                return; // Not running
            }

            _processingCts?.Cancel();
            _processingTask = null;
            _processingCts?.Dispose();
            _processingCts = null;
        }

        private async Task ProcessChangesLoopAsync(CancellationToken cancellationToken)
        {
            _logger.LogInformation("File system change processing loop started");

            while (!cancellationToken.IsCancellationRequested)
            {
                try
                {
                    // Process a batch of changes
                    await ProcessPendingChangesAsync(cancellationToken).ConfigureAwait(false);

                    // Wait before next batch (5 seconds)
                    await Task.Delay(TimeSpan.FromSeconds(5), cancellationToken).ConfigureAwait(false);
                }
                catch (OperationCanceledException)
                {
                    break;
                }
                catch (Exception ex)
                {
                    _logger.LogError(ex, "Error in file system change processing loop");

                    // Wait before retrying on error
                    try
                    {
                        await Task.Delay(TimeSpan.FromSeconds(10), cancellationToken).ConfigureAwait(false);
                    }
                    catch (OperationCanceledException)
                    {
                        break;
                    }
                }
            }

            _logger.LogInformation("File system change processing loop stopped");
        }

        private async Task ProcessPendingChangesAsync(CancellationToken cancellationToken)
        {
            await using var context = await _dbContextFactory.CreateDbContextAsync(cancellationToken).ConfigureAwait(false);

            // Get unprocessed changes (oldest first, limit 100 per batch)
            var changes = await context.FileSystemChanges
                .Where(c => c.ProcessedAt == null)
                .OrderBy(c => c.DetectedAt)
                .Take(100)
                .ToListAsync(cancellationToken)
                .ConfigureAwait(false);

            if (changes.Count == 0)
            {
                return; // Nothing to process
            }

            _logger.LogDebug("Processing {Count} file system changes", changes.Count);

            foreach (var change in changes)
            {
                if (cancellationToken.IsCancellationRequested)
                {
                    break;
                }

                await ProcessSingleChangeAsync(change, context, cancellationToken).ConfigureAwait(false);
            }

            // Save all processed changes
            try
            {
                await context.SaveChangesAsync(cancellationToken).ConfigureAwait(false);
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "Failed to save processed file system changes");
            }
        }

        private async Task ProcessSingleChangeAsync(FileSystemChange change, JellyfinDbContext context, CancellationToken cancellationToken)
        {
            try
            {
                if (_libraryManager is null)
                {
                    // LibraryManager not available yet (still starting up)
                    return;
                }

                // Process the change based on type
                switch (change.ChangeType)
                {
                    case "Created":
                    case "Modified":
                        // LibraryManager will handle these through its file watcher
                        // We just mark as processed since detection happened
                        _logger.LogDebug("File change detected: {Type} - {Path}", change.ChangeType, change.Path);
                        break;

                    case "Deleted":
                        _logger.LogDebug("File deleted: {Path}", change.Path);
                        break;

                    case "Renamed":
                        _logger.LogDebug("File renamed from {OldPath} to {NewPath}", change.OldPath, change.Path);
                        break;
                }

                // Mark as processed
                change.ProcessedAt = DateTime.UtcNow;
                change.ProcessedBy = _instanceRegistry.CurrentInstanceId;

                // Note: Actual file refresh is handled by LibraryManager's existing file watcher
                // This processor just coordinates change detection across instances
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "Error processing file system change for {Path}", change.Path);
                change.Error = ex.Message;
                change.ProcessedAt = DateTime.UtcNow;
                change.ProcessedBy = _instanceRegistry.CurrentInstanceId;
            }

            await Task.CompletedTask.ConfigureAwait(false);
        }
    }
}