using Application.Abstractions.Data;
using Application.Abstractions.Hub;
using Application.Abstractions.YouTube;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using StackExchange.Redis;
namespace Infrastructure.YouTube;
///
/// 30초 주기로 라이브 중인 채널의 concurrentViewers 를 일괄 조회하여 Redis 에 갱신.
///
/// Quota 예산:
/// - videos.list (id=v1,v2,...50) = 1 unit / 50 videos
/// - 30초 주기 + batch 50 = 분당 최대 2 call = 하루 최대 2,880 unit (기본 quota 10,000 대비 29%)
/// - 라이브 채널 50개 미만이면 비용 동일 (1 batch 내 처리)
///
/// 분산 락:
/// - Redis SETNX 기반 락으로 멀티 인스턴스 중복 실행 방지 (quota 2배 소모 방지)
/// - 락 TTL 은 PollInterval 보다 약간 짧게 설정
///
/// SignalR 연동:
/// - 시청자 수 갱신마다 ReceiveChannelStatus 브로드캐스트 → ChannelSidebar / WatchView 실시간 반영
/// - concurrentViewers 가 응답에 없으면 라이브 종료로 판단 → ClearLiveAsync + IsLive=false 브로드캐스트
///
internal sealed class YouTubeLiveViewerPoller(
IYouTubeLiveStateStore liveStateStore,
IYouTubeApiService youTubeApi,
IChannelStatusBroadcaster statusBroadcaster,
IServiceScopeFactory scopeFactory,
IConnectionMultiplexer redis,
ILogger logger
) : BackgroundService
{
private const string LockKey = "youtube:viewer-poller:lock";
private static readonly TimeSpan InitialDelay = TimeSpan.FromSeconds(30);
private static readonly TimeSpan PollInterval = TimeSpan.FromSeconds(30);
private static readonly TimeSpan LockDuration = TimeSpan.FromSeconds(25);
private readonly string _instanceId = Guid.NewGuid().ToString("N");
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
await Task.Delay(InitialDelay, stoppingToken);
logger.LogInformation("[ViewerPoller] 서비스 시작 — 폴링 주기: {Interval}초, 인스턴스: {InstanceId}",
PollInterval.TotalSeconds, _instanceId);
while (!stoppingToken.IsCancellationRequested)
{
try
{
await PollOnceAsync(stoppingToken);
}
catch (OperationCanceledException)
{
break;
}
catch (Exception ex)
{
logger.LogError(ex, "[ViewerPoller] Loop 예외");
}
try
{
await Task.Delay(PollInterval, stoppingToken);
}
catch (OperationCanceledException)
{
break;
}
}
}
private async Task PollOnceAsync(CancellationToken ct)
{
var db = redis.GetDatabase();
// 분산 락 획득 (멀티 인스턴스 환경에서 중복 실행 방지)
var acquired = await db.StringSetAsync(LockKey, _instanceId, LockDuration, When.NotExists);
if (!acquired)
{
logger.LogDebug("[ViewerPoller] 다른 인스턴스가 락 보유 중 — skip");
return;
}
try
{
var allLive = await liveStateStore.GetAllLiveAsync();
var liveList = allLive.Where(l => l.IsLive).ToList();
if (liveList.Count == 0)
{
logger.LogDebug("[ViewerPoller] 라이브 채널 없음 — skip");
return;
}
var videoIds = liveList.Select(l => l.VideoId).ToList();
var viewers = await youTubeApi.GetConcurrentViewersAsync(videoIds, ct);
// 🛡️ Quota 초과 방어: viewers 응답이 통째로 비어있으면 API 실패(quotaExceeded 등)로 판단,
// ended 오인 처리 안 하고 이번 cycle skip. 다음 cycle 또는 다음 PubSub에서 재처리됨.
// 단일 video만 missing이면 진짜 ended로 처리 (정상 분기).
if (viewers.Count == 0)
{
logger.LogWarning("[ViewerPoller] viewers 응답 empty (live={LiveCount}) — API 실패 추정 (quota?). ended 오인 방지 위해 skip",
liveList.Count);
return;
}
// YouTube channelId → 내부 Channel.SID 매핑 (SignalR 브로드캐스트용)
var ytChannelIds = liveList.Select(l => l.ChannelId).Distinct().ToList();
using var scope = scopeFactory.CreateScope();
var dbCtx = scope.ServiceProvider.GetRequiredService();
var sidMap = await dbCtx.Channel.AsNoTracking()
.Where(c => c.YouTubeChannelID != null && ytChannelIds.Contains(c.YouTubeChannelID))
.Select(c => new { c.YouTubeChannelID, c.SID })
.ToDictionaryAsync(x => x.YouTubeChannelID!, x => x.SID, ct);
var now = DateTime.UtcNow;
var updated = 0;
var ended = 0;
foreach (var info in liveList)
{
if (ct.IsCancellationRequested)
{
break;
}
if (!viewers.TryGetValue(info.VideoId, out var cv))
{
// concurrentViewers 미제공 = 라이브 종료로 판단 (Redis 정리 + 브로드캐스트)
// 만약 일시적 0명 상태였다면 다음 PubSub/FeedPoll 사이클에서 다시 마킹됨
await liveStateStore.ClearLiveAsync(info.ChannelId);
if (sidMap.TryGetValue(info.ChannelId, out var endedSid))
{
await statusBroadcaster.BroadcastAsync(endedSid, isLive: false, viewerCount: 0, videoId: null, ct);
}
ended++;
continue;
}
await liveStateStore.UpdateViewerCountAsync(info.ChannelId, cv, now);
if (sidMap.TryGetValue(info.ChannelId, out var sid))
{
await statusBroadcaster.BroadcastAsync(sid, isLive: true, viewerCount: cv, videoId: info.VideoId, ct);
}
updated++;
}
logger.LogInformation("[ViewerPoller] 완료 — live={Live}, updated={Updated}, ended={Ended}",
liveList.Count, updated, ended);
}
finally
{
// 내 인스턴스가 잡은 락만 해제 (만료된 락을 다른 인스턴스가 잡았을 경우 보호)
var current = await db.StringGetAsync(LockKey);
if (current == _instanceId)
{
await db.KeyDeleteAsync(LockKey);
}
}
}
}