using Cleanuparr.Domain.Entities.Arr.Queue; using Cleanuparr.Domain.Enums; using Cleanuparr.Infrastructure.Events.Interfaces; using Cleanuparr.Infrastructure.Features.Context; using Cleanuparr.Infrastructure.Features.Notifications; using Cleanuparr.Infrastructure.Realtime; using Cleanuparr.Infrastructure.Interceptors; using Cleanuparr.Persistence; using Cleanuparr.Persistence.Models.Configuration.Arr; using Cleanuparr.Persistence.Models.Events; using Cleanuparr.Persistence.Models.State; using Cleanuparr.Persistence.Providers; using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.Logging; namespace Cleanuparr.Infrastructure.Events; /// /// Service for publishing events to the database and to connected clients /// public class EventPublisher : IEventPublisher { private readonly EventsContext _context; private readonly IEventNotifier _eventNotifier; private readonly ILogger _logger; private readonly INotificationPublisher _notificationPublisher; private readonly IDryRunInterceptor _dryRunInterceptor; private readonly IDatabaseProvider _databaseProvider; private readonly TimeProvider _timeProvider; public EventPublisher( EventsContext context, IEventNotifier eventNotifier, ILogger logger, INotificationPublisher notificationPublisher, IDryRunInterceptor dryRunInterceptor, IDatabaseProvider databaseProvider, TimeProvider timeProvider) { _context = context; _eventNotifier = eventNotifier; _logger = logger; _notificationPublisher = notificationPublisher; _dryRunInterceptor = dryRunInterceptor; _databaseProvider = databaseProvider; _timeProvider = timeProvider; } /// /// Generic method for publishing events to the database and to connected clients. /// Common context fields are populated here; sets event-type-specific typed fields. /// public async Task PublishAsync(EventType eventType, string message, EventSeverity severity, Action? configure = null, Guid? trackingId = null, Guid? strikeId = null, bool? isDryRun = null) { AppEvent eventEntity = new() { EventType = eventType, Message = message, Severity = severity, TrackingId = trackingId, StrikeId = strikeId, JobRunId = ContextProvider.TryGetJobRunId(), ArrInstanceId = ContextProvider.Get(ContextProvider.Keys.ArrInstanceId) as Guid?, DownloadClientId = ContextProvider.Get(ContextProvider.Keys.DownloadClientId) as Guid?, InstanceType = ContextProvider.Get(nameof(InstanceType)) is InstanceType it ? it : null, InstanceUrl = (ContextProvider.Get(ContextProvider.Keys.ArrInstanceUrl) as Uri)?.ToString(), DownloadClientType = ContextProvider.Get(ContextProvider.Keys.DownloadClientType) is DownloadClientTypeName dct ? dct : null, DownloadClientName = ContextProvider.Get(ContextProvider.Keys.DownloadClientName) as string, }; configure?.Invoke(eventEntity); eventEntity.IsDryRun = isDryRun ?? await _dryRunInterceptor.IsDryRunEnabled(); _context.Events.Add(eventEntity); await _context.SaveChangesAsync(); await NotifyClientsAsync(eventEntity); _logger.LogTrace("Published event: {eventType}", eventType); } /// /// Publishes a manual event, gated to avoid duplicates. Common context fields are populated here; /// sets event-type-specific typed fields. When an item hash is set, /// the event is suppressed if an unresolved event of the same type/hash already exists, or if one /// was resolved within the post-resolve cooldown window. /// public async Task PublishManualAsync(ManualEventType type, string message, EventSeverity severity, Action? configure = null, bool? isDryRun = null) { ManualEvent eventEntity = new() { Type = type, Message = message, Severity = severity, JobRunId = ContextProvider.TryGetJobRunId(), InstanceType = ContextProvider.Get(nameof(InstanceType)) is InstanceType it ? it : null, InstanceUrl = (ContextProvider.Get(ContextProvider.Keys.ArrInstanceUrl) as Uri)?.ToString(), DownloadClientType = ContextProvider.Get(ContextProvider.Keys.DownloadClientType) is DownloadClientTypeName dct ? dct : null, DownloadClientName = ContextProvider.Get(ContextProvider.Keys.DownloadClientName) as string, }; configure?.Invoke(eventEntity); string? normalizedHash = eventEntity.ItemHash?.ToLowerInvariant(); eventEntity.ItemHash = normalizedHash; if (normalizedHash is not null) { // 1h cooldown is hardcoded by request; make it a config value only if it needs tuning. DateTimeOffset cutoff = _timeProvider.GetUtcNow().AddHours(-1); // Suppress if an unresolved event already exists (dedup) OR one was resolved < 1h ago (post-resolve cooldown). bool suppress = await _context.ManualEvents.AnyAsync(e => e.Type == type && e.ItemHash == normalizedHash && (!e.IsResolved || (e.ResolvedAt != null && e.ResolvedAt >= cutoff))); if (suppress) { _logger.LogDebug("Skipping manual event {type} for {hash} (unresolved or within cooldown)", type, normalizedHash); return; } } eventEntity.IsDryRun = isDryRun ?? await _dryRunInterceptor.IsDryRunEnabled(); try { _context.ManualEvents.Add(eventEntity); await _context.SaveChangesAsync(); } catch (DbUpdateException ex) when (normalizedHash is not null && _databaseProvider.IsUniqueConstraintViolation(ex)) { _logger.LogDebug("Manual event {type} for {hash} rejected by unique index", type, normalizedHash); _context.Entry(eventEntity).State = EntityState.Detached; return; } await NotifyClientsAsync(eventEntity); _logger.LogTrace("Published manual event: {message}", message); } /// /// Publishes a strike event with context data and notifications /// public async Task PublishStrike(StrikeType strikeType, int strikeCount, string hash, string itemName, Guid strikeId) { // Determine the appropriate EventType based on StrikeType EventType eventType = strikeType switch { StrikeType.Stalled => EventType.StalledStrike, StrikeType.DownloadingMetadata => EventType.DownloadingMetadataStrike, StrikeType.FailedImport => EventType.FailedImportStrike, StrikeType.SlowSpeed => EventType.SlowSpeedStrike, StrikeType.SlowTime => EventType.SlowTimeStrike, StrikeType.DeadTorrent => EventType.DeadTorrentStrike, StrikeType.Errored => EventType.ErroredStrike, _ => throw new ArgumentOutOfRangeException(nameof(strikeType), strikeType, null) }; List failedImportReasons = []; if (strikeType is StrikeType.FailedImport) { QueueRecord record = ContextProvider.Get(nameof(QueueRecord)); failedImportReasons = record.StatusMessages? .Select(m => m.Messages is { Count: > 0 } ? $"{m.Title}: {string.Join("; ", m.Messages)}" : m.Title) .Where(s => !string.IsNullOrWhiteSpace(s)) .ToList() ?? []; } bool isDryRun = await _dryRunInterceptor.IsDryRunEnabled(); // Publish the event await PublishAsync( eventType, $"Item '{itemName}' has been struck {strikeCount} times for reason '{strikeType}'", EventSeverity.Important, configure: e => { e.ItemTitle = itemName; e.ItemHash = hash; e.StrikeCount = strikeCount; e.FailedImportReasons = failedImportReasons; }, strikeId: strikeId, isDryRun: isDryRun); // Broadcast strike for real-time dashboard updates await BroadcastStrikeAsync(strikeId, strikeType, hash, itemName, isDryRun); // Send notification (uses ContextProvider internally) await _notificationPublisher.NotifyStrike(strikeType, strikeCount); } /// /// Publishes a strike reset event: emitted when a download recovers and its strikes of a given type are cleared. /// public async Task PublishStrikeReset(StrikeType strikeType, int strikeCount, string hash, string itemName) { await PublishAsync( EventType.StrikeReset, $"'{itemName}' recovered — {strikeCount} '{strikeType}' strike(s) reset", EventSeverity.Information, configure: e => { e.ItemTitle = itemName; e.ItemHash = hash; e.StrikeCount = strikeCount; }); } /// /// Publishes a queue item deleted event with context data and notifications /// public async Task PublishQueueItemDeleted(bool removeFromClient, DeleteReason deleteReason) { // Get context data for the event string itemName = ContextProvider.Get(ContextProvider.Keys.ItemName); string hash = ContextProvider.Get(ContextProvider.Keys.Hash); // Publish the event await PublishAsync( EventType.QueueItemDeleted, $"Deleting item from queue with reason: {deleteReason}", EventSeverity.Important, configure: e => { e.ItemTitle = itemName; e.ItemHash = hash; e.DeleteReason = deleteReason; e.RemoveFromClient = removeFromClient; }); // Send notification (uses ContextProvider internally) await _notificationPublisher.NotifyQueueItemDeleted(removeFromClient, deleteReason); } /// /// Publishes a force imported event with notifications /// public async Task PublishForceImported(string itemTitle, string hash) { await PublishAsync( EventType.ForceImported, "Imported a download the arr had blocked", EventSeverity.Important, configure: e => { e.ItemTitle = itemTitle; e.ItemHash = hash; }); await _notificationPublisher.NotifyForceImported(); } /// /// Publishes a download cleaned event with context data and notifications /// public async Task PublishDownloadCleaned(double ratio, TimeSpan seedingTime, string categoryName, CleanReason reason) { // Get context data for the event string itemName = ContextProvider.Get(ContextProvider.Keys.ItemName); string hash = ContextProvider.Get(ContextProvider.Keys.Hash); // Publish the event await PublishAsync( EventType.DownloadCleaned, $"Cleaned item from download client with reason: {reason}", EventSeverity.Important, configure: e => { e.ItemTitle = itemName; e.ItemHash = hash; e.CleanedCategory = categoryName; e.SeedRatio = ratio; e.SeedingTimeHours = seedingTime.TotalHours; e.CleanReason = reason; }); // Send notification (uses ContextProvider internally) await _notificationPublisher.NotifyDownloadCleaned(ratio, seedingTime, categoryName, reason); } /// /// Publishes a download stopped event with context data and notifications /// public async Task PublishDownloadStopped(double ratio, TimeSpan seedingTime, string categoryName, CleanReason reason) { string itemName = ContextProvider.Get(ContextProvider.Keys.ItemName); string hash = ContextProvider.Get(ContextProvider.Keys.Hash); await PublishAsync( EventType.DownloadStopped, $"Stopped item in download client with reason: {reason}", EventSeverity.Important, configure: e => { e.ItemTitle = itemName; e.ItemHash = hash; e.CleanedCategory = categoryName; e.SeedRatio = ratio; e.SeedingTimeHours = seedingTime.TotalHours; e.CleanReason = reason; }); await _notificationPublisher.NotifyDownloadStopped(ratio, seedingTime, categoryName, reason); } /// /// Publishes a category changed event with context data and notifications /// public async Task PublishCategoryChanged(string oldCategory, string newCategory, bool isTag = false) { // Get context data for the event string itemName = ContextProvider.Get(ContextProvider.Keys.ItemName); string hash = ContextProvider.Get(ContextProvider.Keys.Hash); // Publish the event await PublishAsync( EventType.CategoryChanged, isTag ? $"Tag '{newCategory}' added to download" : $"Category changed from '{oldCategory}' to '{newCategory}'", EventSeverity.Information, configure: e => { e.ItemTitle = itemName; e.ItemHash = hash; e.OldCategory = oldCategory; e.NewCategory = newCategory; e.IsCategoryTag = isTag; }); // Send notification (uses ContextProvider internally) await _notificationPublisher.NotifyCategoryChanged(oldCategory, newCategory, isTag); } /// /// Publishes an event alerting that an item keeps coming back /// public async Task PublishRecurringItem(string hash, string itemName, int strikeCount) { await PublishManualAsync( ManualEventType.RecurringDownload, "Download keeps coming back after deletion\nTo prevent further issues, please consult the prerequisites: https://cleanuparr.github.io/Cleanuparr/docs/installation/", EventSeverity.Important, configure: e => { e.ItemTitle = itemName; e.ItemHash = hash; e.StrikeCount = strikeCount; } ); } /// /// Publishes a search triggered event with context data and notifications. /// Returns the event ID so the SeekerCommandMonitor can update it on completion. /// public async Task PublishSearchTriggered(string itemTitle, SeekerSearchType searchType, SeekerSearchReason searchReason, Guid? cycleId = null, bool? isDryRun = null, SeekerContentKind? contentKind = null) { bool dryRun = isDryRun ?? await _dryRunInterceptor.IsDryRunEnabled(); AppEvent eventEntity = new() { EventType = EventType.SearchTriggered, Message = $"Search triggered for {itemTitle}", Severity = EventSeverity.Information, SearchStatus = dryRun ? null : SearchCommandStatus.Pending, JobRunId = ContextProvider.TryGetJobRunId(), ArrInstanceId = ContextProvider.Get(ContextProvider.Keys.ArrInstanceId) as Guid?, DownloadClientId = ContextProvider.Get(ContextProvider.Keys.DownloadClientId) as Guid?, InstanceType = ContextProvider.Get(nameof(InstanceType)) is InstanceType it ? it : null, InstanceUrl = (ContextProvider.Get(ContextProvider.Keys.ArrInstanceUrl) as Uri)?.ToString(), DownloadClientType = ContextProvider.Get(ContextProvider.Keys.DownloadClientType) is DownloadClientTypeName dct ? dct : null, DownloadClientName = ContextProvider.Get(ContextProvider.Keys.DownloadClientName) as string, CycleId = cycleId, ItemTitle = itemTitle, SearchType = searchType, SearchReason = searchReason, ContentKind = contentKind, }; eventEntity.IsDryRun = dryRun; _context.Events.Add(eventEntity); await _context.SaveChangesAsync(); await NotifyClientsAsync(eventEntity); await _notificationPublisher.NotifySearchTriggered(itemTitle, searchType, searchReason); return eventEntity.Id; } /// /// Updates an existing search event with completion status and optional grabbed item titles /// public async Task PublishSearchCompleted(Guid eventId, SearchCommandStatus status, InstanceType instanceType, string instanceUrl, List? grabbedItems = null) { var existingEvent = await _context.Events .FirstOrDefaultAsync(e => e.Id == eventId); if (existingEvent is null) { _logger.LogWarning("Could not find search event {EventId} to update completion status", eventId); return; } existingEvent.SearchStatus = status; existingEvent.CompletedAt = _timeProvider.GetUtcNow(); if (grabbedItems is { Count: > 0 }) { existingEvent.GrabbedItems = grabbedItems; } await _context.SaveChangesAsync(); await NotifyClientsAsync(existingEvent); if (status is SearchCommandStatus.Completed && grabbedItems is { Count: > 0 }) { await _notificationPublisher.NotifySearchItemGrabbed(existingEvent.ItemTitle ?? string.Empty, grabbedItems, instanceType, instanceUrl); } } public async Task PublishSeekerCycleCompleted(SeekerCycle cycle, string instanceName) { bool alreadyPublished = await _context.Events .AnyAsync(e => e.EventType == EventType.SeekerCycleCompleted && e.CycleId == cycle.Id); if (alreadyPublished) { return; } int upgradesGrabbed = await _context.Events .Where(e => e.CycleId == cycle.Id && e.EventType == EventType.SearchTriggered) .Where(e => e.SearchReason == SeekerSearchReason.QualityCutoffNotMet || e.SearchReason == SeekerSearchReason.CustomFormatScoreBelowCutoff) .Where(e => e.GrabbedItems.Count > 0) .CountAsync(); TimeSpan duration = (cycle.CompletedAt ?? _timeProvider.GetUtcNow()) - cycle.StartedAt; await PublishAsync( EventType.SeekerCycleCompleted, $"Cycle completed on {instanceName} after {cycle.SearchedUnits} searches, {cycle.GrabbedUnits} grabbed", EventSeverity.Information, configure: e => e.CycleId = cycle.Id, isDryRun: cycle.IsDryRun); await _notificationPublisher.NotifySeekerCycleCompleted( instanceName, duration, cycle.SearchedUnits, cycle.GrabbedUnits, upgradesGrabbed); } /// /// Updates an existing search event with the started status /// public async Task PublishSearchStarted(Guid eventId) { AppEvent? existingEvent = await _context.Events .FirstOrDefaultAsync(e => e.Id == eventId); if (existingEvent is null) { _logger.LogWarning("Could not find search event {EventId} to update its status", eventId); return; } if (existingEvent.SearchStatus is not SearchCommandStatus.Pending) { return; } existingEvent.SearchStatus = SearchCommandStatus.Started; await _context.SaveChangesAsync(); await NotifyClientsAsync(existingEvent); } /// /// Fails every search event of an arr instance that never reached a terminal status /// public Task FailStrandedSearchEvents(Guid arrInstanceId) => FailSearchEventsAsync(InProgressSearchEvents().Where(e => e.ArrInstanceId == arrInstanceId)); /// /// Fails every search event that never got a command tracker and is older than the given cutoff /// public Task FailAbandonedSearchEvents(DateTimeOffset cutoff) => FailSearchEventsAsync(InProgressSearchEvents() .Where(e => e.Timestamp < cutoff && !_context.SeekerCommandTrackers.Any(t => t.EventId == e.Id))); private IQueryable InProgressSearchEvents() => _context.Events .Where(e => e.EventType == EventType.SearchTriggered && (e.SearchStatus == SearchCommandStatus.Pending || e.SearchStatus == SearchCommandStatus.Started)); private async Task FailSearchEventsAsync(IQueryable query) { List strandedEvents = await query.ToListAsync(); if (strandedEvents.Count == 0) { return 0; } foreach (AppEvent strandedEvent in strandedEvents) { strandedEvent.SearchStatus = SearchCommandStatus.Failed; strandedEvent.CompletedAt = _timeProvider.GetUtcNow(); } await _context.SaveChangesAsync(); foreach (AppEvent strandedEvent in strandedEvents) { await NotifyClientsAsync(strandedEvent); } return strandedEvents.Count; } /// /// Publishes an event alerting that search was not triggered for an item /// public async Task PublishSearchNotTriggered(string hash, string itemName) { await PublishManualAsync( ManualEventType.SearchNotTriggered, "Replacement search was not triggered after removal\nPlease trigger a manual search if needed", EventSeverity.Warning, configure: e => { e.ItemTitle = itemName; e.ItemHash = hash; } ); } private async Task NotifyClientsAsync(AppEvent appEventEntity) { try { await _eventNotifier.NotifyEventAsync(appEventEntity); } catch (Exception ex) { _logger.LogError(ex, "Failed to send event {EventId} to connected clients", appEventEntity.Id); } } private async Task NotifyClientsAsync(ManualEvent appEventEntity) { try { await _eventNotifier.NotifyManualEventAsync(appEventEntity); } catch (Exception ex) { _logger.LogError(ex, "Failed to send event {EventId} to connected clients", appEventEntity.Id); } } private async Task BroadcastStrikeAsync(Guid strikeId, StrikeType strikeType, string hash, string itemName, bool isDryRun) { try { await _eventNotifier.NotifyStrikeAsync(strikeId, strikeType, hash, itemName, isDryRun); } catch (Exception ex) { _logger.LogError(ex, "Failed to send strike to connected clients"); } } }