//
// Copyright (c) PlaceholderCompany. All rights reserved.
//
namespace Jellyfin.Server.Implementations.Clustering
{
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Jellyfin.Database.Implementations;
using Jellyfin.Database.Implementations.Entities;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Logging;
///
/// Manages Jellyfin instance registration and lifecycle.
///
public sealed class InstanceRegistry : IInstanceRegistry, IAsyncDisposable
{
private readonly IDbContextFactory _dbContextFactory;
private readonly ILogger _logger;
private readonly Timer _heartbeatTimer;
private bool _disposed;
///
/// Initializes a new instance of the class.
///
/// The database context factory.
/// The logger.
public InstanceRegistry(
IDbContextFactory dbContextFactory,
ILogger logger)
{
_dbContextFactory = dbContextFactory;
_logger = logger;
CurrentInstanceId = Guid.NewGuid();
// Send heartbeat every 30 seconds
_heartbeatTimer = new Timer(
HeartbeatCallback,
null,
TimeSpan.FromSeconds(30),
TimeSpan.FromSeconds(30));
}
///
/// Gets the unique identifier for the current instance.
///
public Guid CurrentInstanceId { get; }
///
public async Task RegisterInstanceAsync(
string hostname,
int processId,
int httpPort,
int? httpsPort,
string version,
string? capabilities,
CancellationToken cancellationToken = default)
{
try
{
await using var dbContext = await _dbContextFactory.CreateDbContextAsync(cancellationToken).ConfigureAwait(false);
var instance = new Instance(
CurrentInstanceId,
hostname,
processId,
httpPort,
version)
{
HttpsPort = httpsPort
};
if (!string.IsNullOrEmpty(capabilities))
{
instance.SetCapabilities(capabilities);
}
dbContext.Instances.Add(instance);
await dbContext.SaveChangesAsync(cancellationToken).ConfigureAwait(false);
_logger.LogInformation(
"Registered instance {InstanceId} ({Hostname}:{Port})",
CurrentInstanceId,
hostname,
httpPort);
}
catch (Exception ex)
{
_logger.LogError(ex, "Failed to register instance {InstanceId}", CurrentInstanceId);
throw;
}
}
///
public async Task UpdateHeartbeatAsync(CancellationToken cancellationToken = default)
{
try
{
await using var dbContext = await _dbContextFactory.CreateDbContextAsync(cancellationToken).ConfigureAwait(false);
var instance = await dbContext.Instances
.FirstOrDefaultAsync(i => i.InstanceId.Equals(CurrentInstanceId), cancellationToken)
.ConfigureAwait(false);
if (instance is not null)
{
instance.UpdateHeartbeat();
await dbContext.SaveChangesAsync(cancellationToken).ConfigureAwait(false);
}
}
catch (Exception ex)
{
_logger.LogWarning(ex, "Failed to update heartbeat for instance {InstanceId}", CurrentInstanceId);
}
}
///
public async Task ShutdownInstanceAsync(CancellationToken cancellationToken = default)
{
try
{
await using var dbContext = await _dbContextFactory.CreateDbContextAsync(cancellationToken).ConfigureAwait(false);
var instance = await dbContext.Instances
.FirstOrDefaultAsync(i => i.InstanceId.Equals(CurrentInstanceId), cancellationToken)
.ConfigureAwait(false);
if (instance is not null)
{
instance.UpdateStatus(InstanceStatus.Shutdown);
await dbContext.SaveChangesAsync(cancellationToken).ConfigureAwait(false);
_logger.LogInformation("Instance {InstanceId} marked as shutdown", CurrentInstanceId);
}
}
catch (Exception ex)
{
_logger.LogError(ex, "Failed to shutdown instance {InstanceId}", CurrentInstanceId);
}
}
///
public async Task> GetActiveInstancesAsync(CancellationToken cancellationToken = default)
{
await using var dbContext = await _dbContextFactory.CreateDbContextAsync(cancellationToken).ConfigureAwait(false);
return await dbContext.Instances
.Where(i => i.Status == InstanceStatus.Active)
.Where(i => !i.IsStale(TimeSpan.FromMinutes(2)))
.ToListAsync(cancellationToken)
.ConfigureAwait(false);
}
///
public async Task GetInstanceAsync(Guid instanceId, CancellationToken cancellationToken = default)
{
await using var dbContext = await _dbContextFactory.CreateDbContextAsync(cancellationToken).ConfigureAwait(false);
return await dbContext.Instances
.FirstOrDefaultAsync(i => i.InstanceId.Equals(instanceId), cancellationToken)
.ConfigureAwait(false);
}
///
public async Task CleanupStaleInstancesAsync(TimeSpan staleThreshold, CancellationToken cancellationToken = default)
{
try
{
await using var dbContext = await _dbContextFactory.CreateDbContextAsync(cancellationToken).ConfigureAwait(false);
var cutoffTime = DateTime.UtcNow - staleThreshold;
var staleInstances = await dbContext.Instances
.Where(i => i.LastHeartbeat < cutoffTime)
.Where(i => i.Status != InstanceStatus.Shutdown)
.ToListAsync(cancellationToken)
.ConfigureAwait(false);
if (staleInstances.Count > 0)
{
foreach (var instance in staleInstances)
{
instance.UpdateStatus(InstanceStatus.Failed);
}
await dbContext.SaveChangesAsync(cancellationToken).ConfigureAwait(false);
_logger.LogInformation(
"Marked {Count} stale instances as failed",
staleInstances.Count);
}
return staleInstances.Count;
}
catch (Exception ex)
{
_logger.LogError(ex, "Failed to cleanup stale instances");
return 0;
}
}
///
public async Task IsPrimaryInstanceAsync(CancellationToken cancellationToken = default)
{
await using var dbContext = await _dbContextFactory.CreateDbContextAsync(cancellationToken).ConfigureAwait(false);
var instance = await dbContext.Instances
.FirstOrDefaultAsync(i => i.InstanceId.Equals(CurrentInstanceId), cancellationToken)
.ConfigureAwait(false);
return instance?.IsPrimary ?? false;
}
///
public async ValueTask DisposeAsync()
{
if (_disposed)
{
return;
}
_disposed = true;
await _heartbeatTimer.DisposeAsync().ConfigureAwait(false);
await ShutdownInstanceAsync(CancellationToken.None).ConfigureAwait(false);
}
private void HeartbeatCallback(object? state)
{
if (_disposed)
{
return;
}
Task.Run(async () =>
{
try
{
await UpdateHeartbeatAsync(CancellationToken.None).ConfigureAwait(false);
}
catch (Exception ex)
{
_logger.LogError(ex, "Heartbeat timer error for instance {InstanceId}", CurrentInstanceId);
}
});
}
}
}