722 lines
29 KiB
C#
722 lines
29 KiB
C#
using System.Collections.Concurrent;
|
|
using LiveRecorder.Application.Common;
|
|
using LiveRecorder.Application.Abstractions.Logging;
|
|
using LiveRecorder.Application.Abstractions.Notifications;
|
|
using LiveRecorder.Application.Abstractions.Platforms;
|
|
using LiveRecorder.Application.Abstractions.Recording;
|
|
using LiveRecorder.Application.Abstractions.Settings;
|
|
using LiveRecorder.Application.Abstractions.Storage;
|
|
using LiveRecorder.Application.Models.Settings;
|
|
using LiveRecorder.Application.Models.RecordTasks;
|
|
using LiveRecorder.Application.Services;
|
|
using LiveRecorder.Domain.Enums;
|
|
using LiveRecorder.Infrastructure.Persistence;
|
|
using Microsoft.Data.Sqlite;
|
|
using Microsoft.EntityFrameworkCore;
|
|
using Microsoft.Extensions.DependencyInjection;
|
|
using Microsoft.Extensions.Hosting;
|
|
using Microsoft.Extensions.Logging;
|
|
|
|
namespace LiveRecorder.Infrastructure.Services;
|
|
|
|
public sealed class LiveRoomPollingBackgroundService : BackgroundService
|
|
{
|
|
private static readonly TimeSpan OfflineGracefulStopTimeout = TimeSpan.FromSeconds(20);
|
|
private static readonly TimeSpan OfflineForcedStopTimeout = TimeSpan.FromSeconds(8);
|
|
private static readonly TimeSpan ExceptionEmailCooldown = TimeSpan.FromHours(6);
|
|
private static readonly TimeSpan PollDispatchSpacing = TimeSpan.FromMilliseconds(400);
|
|
private static readonly TimeSpan MinimumIdleDelay = TimeSpan.FromSeconds(2);
|
|
private const int MaxConcurrentLiveRoomPolls = 2;
|
|
|
|
private readonly IServiceScopeFactory _serviceScopeFactory;
|
|
private readonly ILogger<LiveRoomPollingBackgroundService> _logger;
|
|
private readonly ConcurrentDictionary<string, DateTimeOffset> _exceptionEmailSentAt = new(StringComparer.OrdinalIgnoreCase);
|
|
private readonly ConcurrentDictionary<Guid, DateTimeOffset> _nextPollDueAt = new();
|
|
|
|
public LiveRoomPollingBackgroundService(
|
|
IServiceScopeFactory serviceScopeFactory,
|
|
ILogger<LiveRoomPollingBackgroundService> logger)
|
|
{
|
|
_serviceScopeFactory = serviceScopeFactory;
|
|
_logger = logger;
|
|
}
|
|
|
|
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
|
{
|
|
while (!stoppingToken.IsCancellationRequested)
|
|
{
|
|
var delay = TimeSpan.FromSeconds(60);
|
|
|
|
try
|
|
{
|
|
using var scope = _serviceScopeFactory.CreateScope();
|
|
var settingsService = scope.ServiceProvider.GetRequiredService<ISystemSettingsService>();
|
|
var settings = await settingsService.GetAsync(stoppingToken);
|
|
var ffmpegService = scope.ServiceProvider.GetRequiredService<IFfmpegService>();
|
|
var storageGuardService = scope.ServiceProvider.GetRequiredService<IStorageGuardService>();
|
|
if (storageGuardService.CheckCanStartOrResume(settings).HasEnoughSpace)
|
|
{
|
|
await ffmpegService.ResumePausedFinalizationsAsync(stoppingToken);
|
|
}
|
|
|
|
if (!settings.EnableBackgroundPolling)
|
|
{
|
|
await DelayAsync(TimeSpan.FromSeconds(Math.Clamp(settings.PollingIntervalSeconds, 10, 3600)), stoppingToken);
|
|
continue;
|
|
}
|
|
|
|
var dbContext = scope.ServiceProvider.GetRequiredService<LiveRecorderDbContext>();
|
|
var now = DateTimeOffset.UtcNow;
|
|
var candidates = (await dbContext.LiveRooms
|
|
.AsNoTracking()
|
|
.Where(static item => item.IsEnabled)
|
|
.Select(static item => new
|
|
{
|
|
item.Id,
|
|
item.UpdatedAt,
|
|
item.LastCheckedAt,
|
|
item.IsPriority,
|
|
item.PollingIntervalSecondsOverride
|
|
})
|
|
.ToListAsync(stoppingToken))
|
|
.ToList();
|
|
|
|
var pollCandidates = candidates
|
|
.Select(item =>
|
|
{
|
|
var intervalSeconds = GetEffectivePollingIntervalSeconds(settings.PollingIntervalSeconds, item.PollingIntervalSecondsOverride);
|
|
var baselineDueAt = item.LastCheckedAt?.AddSeconds(intervalSeconds) ?? DateTimeOffset.MinValue;
|
|
var dueAt = _nextPollDueAt.TryGetValue(item.Id, out var scheduledDueAt) && scheduledDueAt > baselineDueAt
|
|
? scheduledDueAt
|
|
: baselineDueAt;
|
|
return new PollCandidate(item.Id, dueAt, item.IsPriority, item.UpdatedAt, intervalSeconds);
|
|
})
|
|
.ToList();
|
|
|
|
if (pollCandidates.Count == 0)
|
|
{
|
|
delay = TimeSpan.FromSeconds(Math.Clamp(settings.PollingIntervalSeconds, 10, 3600));
|
|
}
|
|
else
|
|
{
|
|
var dueCandidates = pollCandidates
|
|
.Where(item => item.DueAt <= now)
|
|
.OrderByDescending(static item => item.IsPriority)
|
|
.ThenBy(static item => item.DueAt)
|
|
.ThenBy(static item => item.UpdatedAt)
|
|
.ToList();
|
|
|
|
if (dueCandidates.Count == 0)
|
|
{
|
|
var nextDueAt = pollCandidates.Min(static item => item.DueAt);
|
|
delay = nextDueAt <= now
|
|
? MinimumIdleDelay
|
|
: ClampDelay(nextDueAt - now);
|
|
}
|
|
else
|
|
{
|
|
await PollLiveRoomsAsync(dueCandidates, settings, stoppingToken);
|
|
delay = MinimumIdleDelay;
|
|
}
|
|
}
|
|
}
|
|
catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
|
|
{
|
|
break;
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogError(ex, "Background live room polling failed");
|
|
|
|
try
|
|
{
|
|
using var notificationScope = _serviceScopeFactory.CreateScope();
|
|
var logService = notificationScope.ServiceProvider.GetRequiredService<ISystemLogService>();
|
|
var emailNotificationService = notificationScope.ServiceProvider.GetRequiredService<IEmailNotificationService>();
|
|
var webhookNotificationService = notificationScope.ServiceProvider.GetRequiredService<IWebhookNotificationService>();
|
|
await logService.WriteAsync(
|
|
SystemLogLevel.Error,
|
|
"Scheduler",
|
|
"Background live room polling failed.",
|
|
ex.ToString(),
|
|
cancellationToken: CancellationToken.None);
|
|
if (ShouldSendExceptionEmail(BuildExceptionEmailKey("background-loop", ex)))
|
|
{
|
|
await emailNotificationService.SendExceptionAsync(
|
|
"Scheduler",
|
|
"Background live room polling failed.",
|
|
ex.ToString(),
|
|
cancellationToken: stoppingToken);
|
|
await webhookNotificationService.SendExceptionAsync(
|
|
"Scheduler",
|
|
"Background live room polling failed.",
|
|
ex.ToString(),
|
|
cancellationToken: stoppingToken);
|
|
}
|
|
}
|
|
catch (Exception notificationEx)
|
|
{
|
|
_logger.LogWarning(notificationEx, "Scheduler failure notification send failed");
|
|
}
|
|
}
|
|
|
|
await DelayAsync(delay, stoppingToken);
|
|
}
|
|
}
|
|
|
|
private static Task DelayAsync(TimeSpan delay, CancellationToken cancellationToken) =>
|
|
Task.Delay(ClampDelay(delay), cancellationToken);
|
|
|
|
private static TimeSpan ClampDelay(TimeSpan delay)
|
|
{
|
|
if (delay <= TimeSpan.Zero)
|
|
{
|
|
return MinimumIdleDelay;
|
|
}
|
|
|
|
return delay < MinimumIdleDelay
|
|
? MinimumIdleDelay
|
|
: delay > TimeSpan.FromHours(1)
|
|
? TimeSpan.FromHours(1)
|
|
: delay;
|
|
}
|
|
|
|
private static int GetEffectivePollingIntervalSeconds(int globalIntervalSeconds, int? overrideIntervalSeconds) =>
|
|
Math.Clamp(overrideIntervalSeconds ?? globalIntervalSeconds, 10, 3600);
|
|
|
|
private async Task PollLiveRoomsAsync(
|
|
IReadOnlyList<PollCandidate> liveRooms,
|
|
SystemSettingsDto settings,
|
|
CancellationToken cancellationToken)
|
|
{
|
|
if (liveRooms.Count == 0)
|
|
{
|
|
return;
|
|
}
|
|
|
|
using var semaphore = new SemaphoreSlim(Math.Min(MaxConcurrentLiveRoomPolls, liveRooms.Count));
|
|
var tasks = new List<Task>(liveRooms.Count);
|
|
|
|
for (var index = 0; index < liveRooms.Count; index++)
|
|
{
|
|
cancellationToken.ThrowIfCancellationRequested();
|
|
|
|
await semaphore.WaitAsync(cancellationToken);
|
|
var liveRoom = liveRooms[index];
|
|
tasks.Add(PollLiveRoomWithReleaseAsync(liveRoom, settings, semaphore, cancellationToken));
|
|
|
|
if (index < liveRooms.Count - 1)
|
|
{
|
|
await Task.Delay(PollDispatchSpacing, cancellationToken);
|
|
}
|
|
}
|
|
|
|
await Task.WhenAll(tasks);
|
|
}
|
|
|
|
private async Task PollLiveRoomWithReleaseAsync(
|
|
PollCandidate liveRoom,
|
|
SystemSettingsDto settings,
|
|
SemaphoreSlim semaphore,
|
|
CancellationToken cancellationToken)
|
|
{
|
|
try
|
|
{
|
|
await PollLiveRoomAsync(liveRoom.LiveRoomId, liveRoom.IntervalSeconds, settings, cancellationToken);
|
|
}
|
|
finally
|
|
{
|
|
semaphore.Release();
|
|
}
|
|
}
|
|
|
|
private async Task PollLiveRoomAsync(
|
|
Guid liveRoomId,
|
|
int intervalSeconds,
|
|
SystemSettingsDto settings,
|
|
CancellationToken cancellationToken)
|
|
{
|
|
using var scope = _serviceScopeFactory.CreateScope();
|
|
var dbContext = scope.ServiceProvider.GetRequiredService<LiveRecorderDbContext>();
|
|
var adapterFactory = scope.ServiceProvider.GetRequiredService<ILivePlatformAdapterFactory>();
|
|
var ffmpegService = scope.ServiceProvider.GetRequiredService<IFfmpegService>();
|
|
var recordService = scope.ServiceProvider.GetRequiredService<RecordService>();
|
|
var liveRoomStatusService = scope.ServiceProvider.GetRequiredService<LiveRoomStatusService>();
|
|
var logService = scope.ServiceProvider.GetRequiredService<ISystemLogService>();
|
|
var emailNotificationService = scope.ServiceProvider.GetRequiredService<IEmailNotificationService>();
|
|
var webhookNotificationService = scope.ServiceProvider.GetRequiredService<IWebhookNotificationService>();
|
|
var storageGuardService = scope.ServiceProvider.GetRequiredService<IStorageGuardService>();
|
|
|
|
var liveRoom = await dbContext.LiveRooms.FirstOrDefaultAsync(item => item.Id == liveRoomId, cancellationToken);
|
|
if (liveRoom is null || !liveRoom.IsEnabled)
|
|
{
|
|
_nextPollDueAt.TryRemove(liveRoomId, out _);
|
|
return;
|
|
}
|
|
|
|
try
|
|
{
|
|
var adapter = adapterFactory.GetByPlatform(liveRoom.Platform);
|
|
var liveStatus = await adapter.GetLiveStatusAsync(liveRoom.RoomId, cancellationToken);
|
|
var now = DateTimeOffset.UtcNow;
|
|
|
|
await liveRoomStatusService.ApplySnapshotAsync(liveRoom, liveStatus, now, cancellationToken);
|
|
await SaveChangesWithRetryAsync(dbContext, cancellationToken);
|
|
|
|
if (!liveStatus.IsLive)
|
|
{
|
|
await CompleteActiveSessionsForOfflineRoomAsync(
|
|
dbContext,
|
|
ffmpegService,
|
|
logService,
|
|
liveRoom.Id,
|
|
cancellationToken);
|
|
return;
|
|
}
|
|
|
|
var pauseCheck = storageGuardService.CheckShouldPause(settings);
|
|
if (!pauseCheck.HasEnoughSpace)
|
|
{
|
|
await PauseActiveSessionsForLowStorageAsync(
|
|
dbContext,
|
|
ffmpegService,
|
|
logService,
|
|
liveRoom.Id,
|
|
pauseCheck.Message,
|
|
cancellationToken);
|
|
}
|
|
|
|
if (!settings.AutoStartRecordingOnLive)
|
|
{
|
|
await UpdateAutoStartDecisionAsync(
|
|
dbContext,
|
|
liveRoom,
|
|
AutoStartDecisionCodes.SkippedDisabled,
|
|
"Auto-start skipped because automatic start is disabled.",
|
|
detail: null,
|
|
cancellationToken);
|
|
return;
|
|
}
|
|
|
|
var startCheck = storageGuardService.CheckCanStartOrResume(settings);
|
|
if (!startCheck.HasEnoughSpace)
|
|
{
|
|
await UpdateAutoStartDecisionAsync(
|
|
dbContext,
|
|
liveRoom,
|
|
AutoStartDecisionCodes.SkippedStorage,
|
|
"Auto-start skipped because storage is below threshold.",
|
|
startCheck.Message,
|
|
cancellationToken);
|
|
await logService.WriteAsync(
|
|
SystemLogLevel.Warning,
|
|
"Storage",
|
|
"Auto-start recording skipped because storage is below resume threshold.",
|
|
startCheck.Message,
|
|
liveRoomId: liveRoom.Id,
|
|
cancellationToken: cancellationToken);
|
|
return;
|
|
}
|
|
|
|
var reconciledStaleSessionIds = await ReconcileStaleActiveSessionsAsync(
|
|
dbContext,
|
|
ffmpegService,
|
|
liveRoom.Id,
|
|
cancellationToken);
|
|
foreach (var reconciledSessionId in reconciledStaleSessionIds)
|
|
{
|
|
await logService.WriteAsync(
|
|
SystemLogLevel.Warning,
|
|
"Scheduler",
|
|
"Recovered a stale active recording session before auto-start check.",
|
|
liveRoomId: liveRoom.Id,
|
|
recordSessionId: reconciledSessionId,
|
|
cancellationToken: cancellationToken);
|
|
}
|
|
|
|
var hasRunningSession = await dbContext.RecordSessions.AnyAsync(
|
|
item => item.LiveRoomId == liveRoom.Id &&
|
|
(item.Status == RecordSessionStatus.Starting ||
|
|
item.Status == RecordSessionStatus.Running ||
|
|
item.Status == RecordSessionStatus.Stopping),
|
|
cancellationToken);
|
|
|
|
if (hasRunningSession)
|
|
{
|
|
await UpdateAutoStartDecisionAsync(
|
|
dbContext,
|
|
liveRoom,
|
|
AutoStartDecisionCodes.SkippedActiveSession,
|
|
"Auto-start skipped because an active recording session already exists.",
|
|
detail: null,
|
|
cancellationToken);
|
|
await logService.WriteAsync(
|
|
SystemLogLevel.Info,
|
|
"Scheduler",
|
|
"Auto-start skipped because an active recording session already exists.",
|
|
liveRoomId: liveRoom.Id,
|
|
cancellationToken: cancellationToken);
|
|
return;
|
|
}
|
|
|
|
_logger.LogInformation("Auto-start recording for live room {RoomId}", liveRoom.RoomId);
|
|
await logService.WriteAsync(
|
|
SystemLogLevel.Info,
|
|
"Scheduler",
|
|
"Live detected by background poller. Auto-starting recording task.",
|
|
liveRoomId: liveRoom.Id,
|
|
cancellationToken: cancellationToken);
|
|
|
|
await recordService.StartAsync(
|
|
new StartRecordTaskRequest
|
|
{
|
|
LiveRoomId = liveRoom.Id
|
|
},
|
|
trackAutoStartDecision: true,
|
|
cancellationToken);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogWarning(ex, "Polling room {RoomId} failed", liveRoom.RoomId);
|
|
|
|
var isTransient = IsTransientPollingException(ex, cancellationToken);
|
|
await logService.WriteAsync(
|
|
isTransient ? SystemLogLevel.Warning : SystemLogLevel.Error,
|
|
"Scheduler",
|
|
isTransient
|
|
? "Transient background polling failure. The room will be retried on the next cycle."
|
|
: "Background polling failed for a live room.",
|
|
ex.ToString(),
|
|
liveRoomId: liveRoom.Id,
|
|
cancellationToken: cancellationToken);
|
|
|
|
if (!isTransient && ShouldSendExceptionEmail(BuildExceptionEmailKey("live-room-poll", ex, liveRoom.Id)))
|
|
{
|
|
await emailNotificationService.SendExceptionAsync(
|
|
"Scheduler",
|
|
"Background polling failed for a live room.",
|
|
ex.ToString(),
|
|
liveRoom,
|
|
cancellationToken: cancellationToken);
|
|
await webhookNotificationService.SendExceptionAsync(
|
|
"Scheduler",
|
|
"Background polling failed for a live room.",
|
|
ex.ToString(),
|
|
liveRoom,
|
|
cancellationToken: cancellationToken);
|
|
}
|
|
}
|
|
finally
|
|
{
|
|
_nextPollDueAt[liveRoomId] = DateTimeOffset.UtcNow.AddSeconds(Math.Clamp(intervalSeconds, 10, 3600));
|
|
}
|
|
}
|
|
|
|
private static async Task CompleteActiveSessionsForOfflineRoomAsync(
|
|
LiveRecorderDbContext dbContext,
|
|
IFfmpegService ffmpegService,
|
|
ISystemLogService logService,
|
|
Guid liveRoomId,
|
|
CancellationToken cancellationToken)
|
|
{
|
|
var activeSessions = await dbContext.RecordSessions
|
|
.Where(item => item.LiveRoomId == liveRoomId &&
|
|
(item.Status == RecordSessionStatus.Starting ||
|
|
item.Status == RecordSessionStatus.Running ||
|
|
item.Status == RecordSessionStatus.Stopping))
|
|
.ToListAsync(cancellationToken);
|
|
|
|
activeSessions = activeSessions
|
|
.OrderBy(item => item.CreatedAt)
|
|
.ToList();
|
|
|
|
foreach (var activeSession in activeSessions)
|
|
{
|
|
if (ffmpegService.IsRunning(activeSession.Id))
|
|
{
|
|
var stopped = await ffmpegService.StopAndWaitAsync(
|
|
activeSession.Id,
|
|
markAsCompletedOnExit: true,
|
|
OfflineGracefulStopTimeout,
|
|
cancellationToken);
|
|
|
|
if (!stopped)
|
|
{
|
|
await logService.WriteAsync(
|
|
SystemLogLevel.Warning,
|
|
"Scheduler",
|
|
"Live room is offline, but the recorder did not stop gracefully in time. Force killing the ffmpeg process.",
|
|
liveRoomId: liveRoomId,
|
|
recordSessionId: activeSession.Id,
|
|
cancellationToken: cancellationToken);
|
|
|
|
var killed = await ffmpegService.KillAndWaitAsync(
|
|
activeSession.Id,
|
|
OfflineForcedStopTimeout,
|
|
cancellationToken);
|
|
if (!killed)
|
|
{
|
|
await logService.WriteAsync(
|
|
SystemLogLevel.Error,
|
|
"Scheduler",
|
|
"Live room is offline, but the active recording session is still shutting down.",
|
|
liveRoomId: liveRoomId,
|
|
recordSessionId: activeSession.Id,
|
|
cancellationToken: cancellationToken);
|
|
continue;
|
|
}
|
|
}
|
|
|
|
await ffmpegService.TryReconcileInactiveSessionAsync(activeSession.Id, cancellationToken);
|
|
await logService.WriteAsync(
|
|
SystemLogLevel.Info,
|
|
"Scheduler",
|
|
"Live room is offline. Completing the active recording session.",
|
|
liveRoomId: liveRoomId,
|
|
recordSessionId: activeSession.Id,
|
|
cancellationToken: cancellationToken);
|
|
continue;
|
|
}
|
|
|
|
if (await ffmpegService.TryReconcileInactiveSessionAsync(activeSession.Id, cancellationToken))
|
|
{
|
|
await logService.WriteAsync(
|
|
SystemLogLevel.Warning,
|
|
"Scheduler",
|
|
"Recovered a stale active recording session after the room was detected offline.",
|
|
liveRoomId: liveRoomId,
|
|
recordSessionId: activeSession.Id,
|
|
cancellationToken: cancellationToken);
|
|
}
|
|
}
|
|
}
|
|
|
|
private static async Task PauseActiveSessionsForLowStorageAsync(
|
|
LiveRecorderDbContext dbContext,
|
|
IFfmpegService ffmpegService,
|
|
ISystemLogService logService,
|
|
Guid liveRoomId,
|
|
string detail,
|
|
CancellationToken cancellationToken)
|
|
{
|
|
var activeSessions = await dbContext.RecordSessions
|
|
.Where(item => item.LiveRoomId == liveRoomId &&
|
|
(item.Status == RecordSessionStatus.Starting ||
|
|
item.Status == RecordSessionStatus.Running ||
|
|
item.Status == RecordSessionStatus.Stopping))
|
|
.ToListAsync(cancellationToken);
|
|
|
|
foreach (var activeSession in activeSessions.OrderBy(static item => item.CreatedAt))
|
|
{
|
|
await logService.WriteAsync(
|
|
SystemLogLevel.Warning,
|
|
"Storage",
|
|
"Storage is below threshold. Pausing active recording session.",
|
|
detail,
|
|
liveRoomId: liveRoomId,
|
|
recordSessionId: activeSession.Id,
|
|
cancellationToken: cancellationToken);
|
|
|
|
if (ffmpegService.IsRunning(activeSession.Id))
|
|
{
|
|
var stopped = await ffmpegService.StopAndWaitAsync(
|
|
activeSession.Id,
|
|
markAsCompletedOnExit: false,
|
|
OfflineGracefulStopTimeout,
|
|
cancellationToken);
|
|
if (!stopped)
|
|
{
|
|
await logService.WriteAsync(
|
|
SystemLogLevel.Warning,
|
|
"Storage",
|
|
"Recorder did not stop gracefully after low storage pause. Force killing the ffmpeg process.",
|
|
detail,
|
|
liveRoomId,
|
|
activeSession.Id,
|
|
cancellationToken: cancellationToken);
|
|
await ffmpegService.KillAndWaitAsync(activeSession.Id, OfflineForcedStopTimeout, cancellationToken);
|
|
}
|
|
}
|
|
|
|
await ffmpegService.TryReconcileInactiveSessionAsync(activeSession.Id, cancellationToken);
|
|
}
|
|
}
|
|
|
|
private static async Task<IReadOnlyList<Guid>> ReconcileStaleActiveSessionsAsync(
|
|
LiveRecorderDbContext dbContext,
|
|
IFfmpegService ffmpegService,
|
|
Guid liveRoomId,
|
|
CancellationToken cancellationToken)
|
|
{
|
|
var activeSessionIds = await dbContext.RecordSessions
|
|
.Where(item => item.LiveRoomId == liveRoomId &&
|
|
(item.Status == RecordSessionStatus.Starting ||
|
|
item.Status == RecordSessionStatus.Running ||
|
|
item.Status == RecordSessionStatus.Stopping))
|
|
.ToListAsync(cancellationToken);
|
|
|
|
var orderedActiveSessionIds = activeSessionIds
|
|
.OrderBy(item => item.CreatedAt)
|
|
.Select(item => item.Id)
|
|
.ToList();
|
|
|
|
if (orderedActiveSessionIds.Count == 0)
|
|
{
|
|
return [];
|
|
}
|
|
|
|
var reconciledSessionIds = new List<Guid>();
|
|
foreach (var activeSessionId in orderedActiveSessionIds)
|
|
{
|
|
if (await ffmpegService.TryReconcileInactiveSessionAsync(activeSessionId, cancellationToken))
|
|
{
|
|
reconciledSessionIds.Add(activeSessionId);
|
|
}
|
|
}
|
|
|
|
return reconciledSessionIds;
|
|
}
|
|
|
|
private static bool IsTransientPollingException(Exception exception, CancellationToken cancellationToken)
|
|
{
|
|
if (exception is OperationCanceledException && cancellationToken.IsCancellationRequested)
|
|
{
|
|
return false;
|
|
}
|
|
|
|
if (exception is HttpRequestException or IOException or TimeoutException)
|
|
{
|
|
return true;
|
|
}
|
|
|
|
if (exception is DbUpdateException dbUpdateException &&
|
|
IsSqliteLockException(dbUpdateException))
|
|
{
|
|
return true;
|
|
}
|
|
|
|
if (exception is SqliteException sqliteException &&
|
|
IsSqliteLockException(sqliteException))
|
|
{
|
|
return true;
|
|
}
|
|
|
|
return exception.InnerException is not null &&
|
|
IsTransientPollingException(exception.InnerException, cancellationToken);
|
|
}
|
|
|
|
private bool ShouldSendExceptionEmail(string key)
|
|
{
|
|
var now = DateTimeOffset.UtcNow;
|
|
while (true)
|
|
{
|
|
if (_exceptionEmailSentAt.TryGetValue(key, out var lastSentAt))
|
|
{
|
|
if (now - lastSentAt < ExceptionEmailCooldown)
|
|
{
|
|
_logger.LogWarning(
|
|
"Suppressed repeated scheduler exception email. Key={Key}; CooldownMinutes={CooldownMinutes}",
|
|
key,
|
|
ExceptionEmailCooldown.TotalMinutes);
|
|
return false;
|
|
}
|
|
|
|
if (_exceptionEmailSentAt.TryUpdate(key, now, lastSentAt))
|
|
{
|
|
return true;
|
|
}
|
|
|
|
continue;
|
|
}
|
|
|
|
if (_exceptionEmailSentAt.TryAdd(key, now))
|
|
{
|
|
return true;
|
|
}
|
|
}
|
|
}
|
|
|
|
private static string BuildExceptionEmailKey(string scope, Exception exception, Guid? liveRoomId = null)
|
|
{
|
|
if (IsSqliteStorageFullException(exception))
|
|
{
|
|
return $"{scope}:sqlite-storage-full";
|
|
}
|
|
|
|
var root = exception.GetBaseException();
|
|
var message = root.Message.Length > 160 ? root.Message[..160] : root.Message;
|
|
return $"{scope}:{liveRoomId?.ToString() ?? "global"}:{root.GetType().FullName}:{message}";
|
|
}
|
|
|
|
private static bool IsSqliteStorageFullException(Exception exception)
|
|
{
|
|
if (exception is SqliteException sqliteException &&
|
|
(sqliteException.SqliteErrorCode == 13 ||
|
|
sqliteException.Message.Contains("database or disk is full", StringComparison.OrdinalIgnoreCase)))
|
|
{
|
|
return true;
|
|
}
|
|
|
|
return exception.InnerException is not null && IsSqliteStorageFullException(exception.InnerException);
|
|
}
|
|
|
|
private static async Task UpdateAutoStartDecisionAsync(
|
|
LiveRecorderDbContext dbContext,
|
|
Domain.Entities.LiveRoom liveRoom,
|
|
string code,
|
|
string summary,
|
|
string? detail,
|
|
CancellationToken cancellationToken)
|
|
{
|
|
liveRoom.SetLastAutoStartDecision(
|
|
Truncate(code, 64),
|
|
Truncate(summary, 256),
|
|
Truncate(detail, 2048),
|
|
DateTimeOffset.UtcNow);
|
|
await SaveChangesWithRetryAsync(dbContext, cancellationToken);
|
|
}
|
|
|
|
private static async Task SaveChangesWithRetryAsync(LiveRecorderDbContext dbContext, CancellationToken cancellationToken)
|
|
{
|
|
for (var attempt = 1; attempt <= 5; attempt++)
|
|
{
|
|
try
|
|
{
|
|
await dbContext.SaveChangesAsync(cancellationToken);
|
|
return;
|
|
}
|
|
catch (DbUpdateException ex) when (attempt < 5 && IsSqliteLockException(ex))
|
|
{
|
|
await Task.Delay(TimeSpan.FromMilliseconds(300 * Math.Pow(2, attempt - 1)), cancellationToken);
|
|
}
|
|
}
|
|
}
|
|
|
|
private static bool IsSqliteLockException(DbUpdateException exception) =>
|
|
exception.InnerException is SqliteException sqliteException &&
|
|
IsSqliteLockException(sqliteException);
|
|
|
|
private static bool IsSqliteLockException(SqliteException exception) =>
|
|
exception.SqliteErrorCode is 5 or 6 ||
|
|
exception.Message.Contains("database is locked", StringComparison.OrdinalIgnoreCase) ||
|
|
exception.Message.Contains("database table is locked", StringComparison.OrdinalIgnoreCase);
|
|
|
|
private static string? Truncate(string? value, int maxLength)
|
|
{
|
|
if (string.IsNullOrWhiteSpace(value))
|
|
{
|
|
return null;
|
|
}
|
|
|
|
var trimmed = value.Trim();
|
|
return trimmed.Length <= maxLength ? trimmed : trimmed[..maxLength];
|
|
}
|
|
|
|
private sealed record PollCandidate(
|
|
Guid LiveRoomId,
|
|
DateTimeOffset DueAt,
|
|
bool IsPriority,
|
|
DateTimeOffset UpdatedAt,
|
|
int IntervalSeconds);
|
|
}
|