// // Copyright (c) PlaceholderCompany. All rights reserved. // namespace Jellyfin.Server.Implementations.Clustering { using System; using System.Threading; using System.Threading.Tasks; using Jellyfin.Database.Implementations; using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Logging; /// /// Service that manages primary instance election and coordination. /// public sealed class PrimaryElectionService : IPrimaryElectionService, IHostedService, IAsyncDisposable { private readonly IDbContextFactory _dbContextFactory; private readonly IInstanceRegistry _instanceRegistry; private readonly ILogger _logger; private CancellationTokenSource? _monitoringCts; private Task? _monitoringTask; private bool _disposed; /// /// Initializes a new instance of the class. /// /// The database context factory. /// The instance registry. /// The logger. public PrimaryElectionService( IDbContextFactory dbContextFactory, IInstanceRegistry instanceRegistry, ILogger logger) { _dbContextFactory = dbContextFactory; _instanceRegistry = instanceRegistry; _logger = logger; } /// public event EventHandler? PrimaryInstanceChanged; /// public bool IsPrimary { get; private set; } /// public Guid? PrimaryInstanceId { get; private set; } /// Task IHostedService.StartAsync(CancellationToken cancellationToken) { return StartAsync(cancellationToken); } /// Task IHostedService.StopAsync(CancellationToken cancellationToken) { return StopAsync(cancellationToken); } /// public async Task StartAsync(CancellationToken cancellationToken = default) { _logger.LogInformation("Starting primary election service"); // Run initial election await ElectPrimaryAsync(cancellationToken).ConfigureAwait(false); // Start background monitoring _monitoringCts = new CancellationTokenSource(); _monitoringTask = MonitorPrimaryStatusAsync(_monitoringCts.Token); _logger.LogInformation( "Primary election service started. Current instance is primary: {IsPrimary}", IsPrimary); } /// public async Task StopAsync(CancellationToken cancellationToken = default) { _logger.LogInformation("Stopping primary election service"); if (_monitoringCts is not null) { _monitoringCts.Cancel(); if (_monitoringTask is not null) { try { await _monitoringTask.ConfigureAwait(false); } catch (OperationCanceledException) { // Expected } } } // If we were primary, relinquish it if (IsPrimary) { await RelinquishPrimaryAsync(cancellationToken).ConfigureAwait(false); } _logger.LogInformation("Primary election service stopped"); } /// public async Task ElectPrimaryAsync(CancellationToken cancellationToken = default) { try { await using var context = await _dbContextFactory.CreateDbContextAsync(cancellationToken).ConfigureAwait(false); // Call the database function to elect primary var result = await context.Database .SqlQuery($"SELECT library.elect_primary_instance()") .FirstOrDefaultAsync(cancellationToken) .ConfigureAwait(false); var previousPrimaryId = PrimaryInstanceId; PrimaryInstanceId = result; IsPrimary = result.HasValue && result.Value.Equals(_instanceRegistry.CurrentInstanceId); // Raise event if primary changed if (!previousPrimaryId.Equals(PrimaryInstanceId)) { _logger.LogInformation( "Primary instance changed from {PreviousPrimaryId} to {NewPrimaryId}. Current instance is primary: {IsPrimary}", previousPrimaryId, PrimaryInstanceId, IsPrimary); PrimaryInstanceChanged?.Invoke( this, new PrimaryInstanceChangedEventArgs(previousPrimaryId, PrimaryInstanceId, IsPrimary)); } return PrimaryInstanceId; } catch (Exception ex) { _logger.LogError(ex, "Failed to elect primary instance"); throw; } } /// public bool ShouldExecuteScheduledTasks() { // Only the primary instance should execute scheduled tasks return IsPrimary; } /// public async ValueTask DisposeAsync() { if (_disposed) { return; } _disposed = true; await StopAsync(CancellationToken.None).ConfigureAwait(false); _monitoringCts?.Dispose(); } private async Task MonitorPrimaryStatusAsync(CancellationToken cancellationToken) { _logger.LogDebug("Primary status monitoring started"); while (!cancellationToken.IsCancellationRequested) { try { // Check primary status every 30 seconds await Task.Delay(TimeSpan.FromSeconds(30), cancellationToken).ConfigureAwait(false); await CheckAndUpdatePrimaryStatusAsync(cancellationToken).ConfigureAwait(false); } catch (OperationCanceledException) { break; } catch (Exception ex) { _logger.LogError(ex, "Error in primary status monitoring loop"); // Wait before retrying try { await Task.Delay(TimeSpan.FromSeconds(5), cancellationToken).ConfigureAwait(false); } catch (OperationCanceledException) { break; } } } _logger.LogDebug("Primary status monitoring stopped"); } private async Task CheckAndUpdatePrimaryStatusAsync(CancellationToken cancellationToken) { try { await using var context = await _dbContextFactory.CreateDbContextAsync(cancellationToken).ConfigureAwait(false); // Check if current primary is still active var currentPrimary = await context.Database .SqlQuery($"SELECT library.get_primary_instance()") .FirstOrDefaultAsync(cancellationToken) .ConfigureAwait(false); // If no primary exists or primary changed, run election if (!currentPrimary.Equals(PrimaryInstanceId)) { _logger.LogWarning( "Primary instance status changed. Current: {CurrentPrimary}, Expected: {ExpectedPrimary}. Running election.", currentPrimary, PrimaryInstanceId); await ElectPrimaryAsync(cancellationToken).ConfigureAwait(false); } } catch (Exception ex) { _logger.LogError(ex, "Failed to check primary status"); } } private async Task RelinquishPrimaryAsync(CancellationToken cancellationToken) { try { await using var context = await _dbContextFactory.CreateDbContextAsync(cancellationToken).ConfigureAwait(false); // Clear IsPrimary flag for this instance await context.Database .ExecuteSqlAsync( $@"UPDATE library.""Instances"" SET ""IsPrimary"" = FALSE WHERE ""InstanceId"" = {_instanceRegistry.CurrentInstanceId}", cancellationToken) .ConfigureAwait(false); IsPrimary = false; _logger.LogInformation("Relinquished primary status"); } catch (Exception ex) { _logger.LogError(ex, "Failed to relinquish primary status"); } } } }