// // Copyright (c) PlaceholderCompany. All rights reserved. // namespace Jellyfin.Server.Implementations.Clustering { using System; using System.Text.Json; using System.Threading; using System.Threading.Tasks; using MediaBrowser.Controller.Library; using Microsoft.Extensions.Logging; /// /// Coordinates cache invalidation across multiple Jellyfin instances using PostgreSQL LISTEN/NOTIFY. /// public sealed class CacheCoordinator : ICacheCoordinator, IAsyncDisposable { private const string CacheInvalidationChannel = "jellyfin_cache_invalidation"; private readonly IPostgresNotificationListener _notificationListener; private readonly IInstanceRegistry _instanceRegistry; private readonly ILogger _logger; private readonly ILibraryManager? _libraryManager; private bool _disposed; private bool _started; /// /// Initializes a new instance of the class. /// /// The PostgreSQL notification listener. /// The instance registry. /// The logger. /// The library manager (optional). public CacheCoordinator( IPostgresNotificationListener notificationListener, IInstanceRegistry instanceRegistry, ILogger logger, ILibraryManager? libraryManager = null) { _notificationListener = notificationListener; _instanceRegistry = instanceRegistry; _logger = logger; _libraryManager = libraryManager; } /// public async Task StartAsync(CancellationToken cancellationToken = default) { if (_started) { _logger.LogWarning("Cache coordinator already started"); return; } try { _notificationListener.NotificationReceived += OnNotificationReceived; await _notificationListener.StartListeningAsync(CacheInvalidationChannel, cancellationToken).ConfigureAwait(false); _started = true; _logger.LogInformation("Cache coordinator started successfully"); } catch (Exception ex) { _logger.LogError(ex, "Failed to start cache coordinator"); throw; } } /// public async Task StopAsync(CancellationToken cancellationToken = default) { if (!_started) { return; } try { _notificationListener.NotificationReceived -= OnNotificationReceived; await _notificationListener.StopListeningAsync(cancellationToken).ConfigureAwait(false); _started = false; _logger.LogInformation("Cache coordinator stopped"); } catch (Exception ex) { _logger.LogError(ex, "Error stopping cache coordinator"); } } /// public async Task InvalidateItemAsync(Guid itemId, CancellationToken cancellationToken = default) { var message = new CacheInvalidationMessage { Type = CacheInvalidationType.Item, EntityId = itemId, Timestamp = DateTime.UtcNow, SourceInstanceId = _instanceRegistry.CurrentInstanceId }; await SendInvalidationMessageAsync(message, cancellationToken).ConfigureAwait(false); } /// public async Task InvalidateUserDataAsync(Guid userId, CancellationToken cancellationToken = default) { var message = new CacheInvalidationMessage { Type = CacheInvalidationType.UserData, EntityId = userId, Timestamp = DateTime.UtcNow, SourceInstanceId = _instanceRegistry.CurrentInstanceId }; await SendInvalidationMessageAsync(message, cancellationToken).ConfigureAwait(false); } /// public async Task InvalidateImageAsync(Guid itemId, CancellationToken cancellationToken = default) { var message = new CacheInvalidationMessage { Type = CacheInvalidationType.Image, EntityId = itemId, Timestamp = DateTime.UtcNow, SourceInstanceId = _instanceRegistry.CurrentInstanceId }; await SendInvalidationMessageAsync(message, cancellationToken).ConfigureAwait(false); } /// public async Task InvalidateAllAsync(CacheInvalidationType cacheType, CancellationToken cancellationToken = default) { var message = new CacheInvalidationMessage { Type = cacheType, EntityId = null, Timestamp = DateTime.UtcNow, SourceInstanceId = _instanceRegistry.CurrentInstanceId }; await SendInvalidationMessageAsync(message, cancellationToken).ConfigureAwait(false); } /// public async Task InvalidateLibraryAsync(Guid libraryId, CancellationToken cancellationToken = default) { var message = new CacheInvalidationMessage { Type = CacheInvalidationType.Library, EntityId = libraryId, Timestamp = DateTime.UtcNow, SourceInstanceId = _instanceRegistry.CurrentInstanceId }; await SendInvalidationMessageAsync(message, cancellationToken).ConfigureAwait(false); } /// public async Task InvalidatePersonAsync(Guid personId, CancellationToken cancellationToken = default) { var message = new CacheInvalidationMessage { Type = CacheInvalidationType.Person, EntityId = personId, Timestamp = DateTime.UtcNow, SourceInstanceId = _instanceRegistry.CurrentInstanceId }; await SendInvalidationMessageAsync(message, cancellationToken).ConfigureAwait(false); } /// public async Task InvalidateMetadataAsync(Guid itemId, CancellationToken cancellationToken = default) { var message = new CacheInvalidationMessage { Type = CacheInvalidationType.Metadata, EntityId = itemId, Timestamp = DateTime.UtcNow, SourceInstanceId = _instanceRegistry.CurrentInstanceId }; await SendInvalidationMessageAsync(message, cancellationToken).ConfigureAwait(false); } /// public async ValueTask DisposeAsync() { if (_disposed) { return; } _disposed = true; await StopAsync(CancellationToken.None).ConfigureAwait(false); } private async Task SendInvalidationMessageAsync(CacheInvalidationMessage message, CancellationToken cancellationToken) { try { var json = JsonSerializer.Serialize(message); await _notificationListener.NotifyAsync(CacheInvalidationChannel, json, cancellationToken).ConfigureAwait(false); _logger.LogDebug( "Sent cache invalidation: Type={Type}, EntityId={EntityId}", message.Type, message.EntityId); } catch (Exception ex) { _logger.LogError( ex, "Failed to send cache invalidation message: Type={Type}, EntityId={EntityId}", message.Type, message.EntityId); } } private void OnNotificationReceived(object? sender, PostgresNotificationEventArgs e) { try { var message = JsonSerializer.Deserialize(e.Payload); if (message is null) { _logger.LogWarning("Received invalid cache invalidation message"); return; } // Don't process messages from this instance if (message.SourceInstanceId.Equals(_instanceRegistry.CurrentInstanceId)) { return; } _logger.LogDebug( "Processing cache invalidation from instance {SourceInstanceId}: Type={Type}, EntityId={EntityId}", message.SourceInstanceId, message.Type, message.EntityId); ProcessInvalidationMessage(message); } catch (Exception ex) { _logger.LogError(ex, "Error processing cache invalidation notification"); } } private void ProcessInvalidationMessage(CacheInvalidationMessage message) { // TODO: Actual cache invalidation logic would go here // This is a placeholder that demonstrates the pattern // The actual implementation would call into Jellyfin's cache managers switch (message.Type) { case CacheInvalidationType.Item: if (message.EntityId.HasValue && _libraryManager is not null) { // Example: _libraryManager.InvalidateItemCache(message.EntityId.Value); _logger.LogDebug("Invalidating item cache for {ItemId}", message.EntityId.Value); } break; case CacheInvalidationType.UserData: if (message.EntityId.HasValue) { _logger.LogDebug("Invalidating user data cache for {UserId}", message.EntityId.Value); } break; case CacheInvalidationType.Image: if (message.EntityId.HasValue) { _logger.LogDebug("Invalidating image cache for {ItemId}", message.EntityId.Value); } break; case CacheInvalidationType.Library: if (message.EntityId.HasValue) { _logger.LogDebug("Invalidating library cache for {LibraryId}", message.EntityId.Value); } break; case CacheInvalidationType.All: _logger.LogDebug("Invalidating all caches of type {CacheType}", message.Type); break; default: _logger.LogWarning("Unknown cache invalidation type: {Type}", message.Type); break; } } } }