diff --git a/src/KArtSell.Host/Infrastructure/KisConnectionPool.cs b/src/KArtSell.Host/Infrastructure/KisConnectionPool.cs deleted file mode 100644 index d54ae971..00000000 --- a/src/KArtSell.Host/Infrastructure/KisConnectionPool.cs +++ /dev/null @@ -1,276 +0,0 @@ -using System.Collections.Concurrent; -using System.Collections.Generic; -using System.Linq; -using System.Threading; -using System.Threading.Tasks; -using KArtSell.BuildingBlocks.Data; -using Dapper; - -namespace KArtSell.Host.Infrastructure; - -/// -/// KIS (Korea Investment & Securities) Connection Pool -/// Manages 3-5 concurrent connections with OAuth2 token refresh (55-min interval) -/// Priority queue: BUY > SELL > CANCEL -/// Idempotent: same token refresh never double-authenticates -/// -public sealed class KisConnectionPool : IAsyncDisposable -{ - private readonly IDbConnectionFactory _connectionFactory; - private readonly string _apiKey; - private readonly int _poolSize; - private readonly ConcurrentDictionary _connections; - private readonly PriorityQueue _availableConnections; - private readonly SemaphoreSlim _poolLock; - private DateTime _nextTokenRefreshTime; - - private const int MaxPoolSize = 5; - private const int MinPoolSize = 3; - private const int TokenRefreshIntervalMinutes = 55; - - public KisConnectionPool( - IDbConnectionFactory connectionFactory, - string apiKey, - int? poolSize = null) - { - _connectionFactory = connectionFactory ?? throw new ArgumentNullException(nameof(connectionFactory)); - _apiKey = apiKey ?? throw new ArgumentNullException(nameof(apiKey)); - _poolSize = poolSize ?? 3; - - if (_poolSize < MinPoolSize || _poolSize > MaxPoolSize) - throw new ArgumentException($"Pool size must be between {MinPoolSize} and {MaxPoolSize}", nameof(poolSize)); - - _connections = new(); - _availableConnections = new(); - _poolLock = new(_poolSize, _poolSize); - _nextTokenRefreshTime = DateTime.UtcNow.AddMinutes(TokenRefreshIntervalMinutes); - } - - /// - /// Initialize connection pool - /// Creates MinPoolSize connections - /// - public async Task InitializeAsync(CancellationToken cancellationToken = default) - { - for (int i = 0; i < _poolSize; i++) - { - var connectionId = Guid.NewGuid(); - var state = new KisConnectionState - { - ConnectionId = connectionId, - State = "idle", - Priority = 2, // CANCEL - CreatedAt = DateTime.UtcNow - }; - - _connections.TryAdd(connectionId, state); - _availableConnections.Enqueue(connectionId, 2); // Lowest priority initially - } - - // Store initial state in database - using var conn = _connectionFactory.CreateConnection(); - await conn.OpenAsync(cancellationToken); - - foreach (var (connId, connState) in _connections) - { - const string sql = """ - INSERT INTO kis.connection_pool_state (connection_id, state, priority, created_at, published_at) - VALUES (@connectionId, @state, @priority, @createdAt, CURRENT_TIMESTAMP) - """; - - await conn.ExecuteAsync(sql, new - { - connectionId = connId, - state = connState.State, - priority = connState.Priority, - createdAt = connState.CreatedAt - }); - } - } - - /// - /// Get an available connection with specified priority - /// Priority: 0=BUY, 1=SELL, 2=CANCEL - /// Idempotent: Returns same connection for same priority - /// - public async Task GetConnectionAsync(int priority, CancellationToken cancellationToken = default) - { - if (priority < 0 || priority > 2) - throw new ArgumentException("Priority must be 0 (BUY), 1 (SELL), or 2 (CANCEL)", nameof(priority)); - - // Try token refresh if due - await RefreshTokenIfDueAsync(cancellationToken); - - // Wait for available connection - var acquiredTime = DateTime.UtcNow; - await _poolLock.WaitAsync(cancellationToken); - - try - { - if (!_availableConnections.TryDequeue(out var connectionId, out _)) - { - throw new InvalidOperationException("No available KIS connections"); - } - - var state = _connections[connectionId]; - state.State = "active"; - state.Priority = priority; - state.AcquiredAt = acquiredTime; - - await UpdateConnectionStateAsync(connectionId, state, cancellationToken); - - return connectionId; - } - finally - { - // Don't release yet; connection is still in use - } - } - - /// - /// Release connection back to pool - /// Safe to release same connection multiple times (idempotent) - /// - public async Task ReleaseConnectionAsync(Guid connectionId, CancellationToken cancellationToken = default) - { - if (!_connections.TryGetValue(connectionId, out var state)) - { - // Already released or invalid; no-op (idempotent) - return; - } - - state.State = "idle"; - state.ReleasedAt = DateTime.UtcNow; - - await UpdateConnectionStateAsync(connectionId, state, cancellationToken); - - // Return to available queue (lowest priority = CANCEL) - _availableConnections.Enqueue(connectionId, 2); - - // Release semaphore - _poolLock.Release(); - } - - /// - /// Refresh OAuth2 token (55-min interval) - /// Idempotent: same refresh won't double-authenticate - /// - public async Task RefreshTokenIfDueAsync(CancellationToken cancellationToken = default) - { - if (DateTime.UtcNow < _nextTokenRefreshTime) - return; // Not due yet - - // TODO: Implement actual OAuth2 token refresh - // For now, just update the next refresh time - _nextTokenRefreshTime = DateTime.UtcNow.AddMinutes(TokenRefreshIntervalMinutes); - - using var conn = _connectionFactory.CreateConnection(); - await conn.OpenAsync(cancellationToken); - - const string sql = """ - INSERT INTO kis.token_refresh_log (connection_id, refresh_at, status, executed_at, published_at) - SELECT id, CURRENT_TIMESTAMP, 'success', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP - FROM kis.connection_pool_state - WHERE state IN ('idle', 'active') - """; - - await conn.ExecuteAsync(sql); - } - - /// - /// Get pool statistics - /// - public async Task GetStatsAsync(CancellationToken cancellationToken = default) - { - using var conn = _connectionFactory.CreateConnection(); - await conn.OpenAsync(cancellationToken); - - const string sql = """ - SELECT - (SELECT COUNT(*) FROM kis.connection_pool_state WHERE state = 'idle') as idle_count, - (SELECT COUNT(*) FROM kis.connection_pool_state WHERE state = 'active') as active_count, - (SELECT COUNT(*) FROM kis.connection_pool_state WHERE state = 'closed') as closed_count - """; - - var result = await conn.QuerySingleAsync(sql); - - return new KisPoolStatsDto - { - TotalConnections = _connections.Count, - IdleConnections = result.idle_count ?? 0, - ActiveConnections = result.active_count ?? 0, - ClosedConnections = result.closed_count ?? 0, - NextTokenRefreshAt = _nextTokenRefreshTime - }; - } - - // Private methods - - private async Task UpdateConnectionStateAsync( - Guid connectionId, - KisConnectionState state, - CancellationToken cancellationToken) - { - using var conn = _connectionFactory.CreateConnection(); - await conn.OpenAsync(cancellationToken); - - const string sql = """ - UPDATE kis.connection_pool_state - SET state = @state, - priority = @priority, - released_at = @releasedAt, - updated_at = CURRENT_TIMESTAMP, - published_at = CURRENT_TIMESTAMP - WHERE connection_id = @connectionId - """; - - await conn.ExecuteAsync(sql, new - { - connectionId, - state = state.State, - priority = state.Priority, - releasedAt = state.ReleasedAt - }); - } - - public async ValueTask DisposeAsync() - { - _poolLock?.Dispose(); - - // Close all connections - foreach (var connectionId in _connections.Keys) - { - using var conn = _connectionFactory.CreateConnection(); - await conn.OpenAsync(); - - const string sql = """ - UPDATE kis.connection_pool_state - SET state = 'closed', - updated_at = CURRENT_TIMESTAMP, - published_at = CURRENT_TIMESTAMP - WHERE connection_id = @connectionId - """; - - await conn.ExecuteAsync(sql, new { connectionId }); - } - } -} - -public sealed class KisConnectionState -{ - public Guid ConnectionId { get; set; } - public string State { get; set; } = "idle"; // idle, active, closed - public int Priority { get; set; } // 0=BUY, 1=SELL, 2=CANCEL - public DateTime CreatedAt { get; set; } - public DateTime? AcquiredAt { get; set; } - public DateTime? ReleasedAt { get; set; } -} - -public sealed class KisPoolStatsDto -{ - public int TotalConnections { get; set; } - public int IdleConnections { get; set; } - public int ActiveConnections { get; set; } - public int ClosedConnections { get; set; } - public DateTime NextTokenRefreshAt { get; set; } -} diff --git a/src/KArtSell.Host/Infrastructure/RateLimiterService.cs b/src/KArtSell.Host/Infrastructure/RateLimiterService.cs deleted file mode 100644 index 53baf1a5..00000000 --- a/src/KArtSell.Host/Infrastructure/RateLimiterService.cs +++ /dev/null @@ -1,340 +0,0 @@ -using KArtSell.BuildingBlocks.Data; -using Dapper; -using System.Collections.Concurrent; - -namespace KArtSell.Host.Infrastructure; - -/// -/// Central Rate Limiter Service -/// Token bucket pattern for KRX, OpenDart, KIS APIs -/// Per-API quota tracking, atomic operations (no partial success) -/// -public sealed class RateLimiterService -{ - private readonly IDbConnectionFactory _connectionFactory; - private readonly IDistributedCache _cache; - private readonly ConcurrentDictionary _quotas; - - // API-specific quotas (per CLAUDE.md requirements) - private static readonly Dictionary ApiQuotas = new() - { - { "krx", (100, 60) }, // 100/minute - { "opendart", (1000, 86400) }, // 1000/day (24h) - { "kis", (50, 1) }, // 50/second - }; - - public RateLimiterService( - IDbConnectionFactory connectionFactory, - IDistributedCache cache) - { - _connectionFactory = connectionFactory ?? throw new ArgumentNullException(nameof(connectionFactory)); - _cache = cache ?? throw new ArgumentNullException(nameof(cache)); - _quotas = new(); - } - - /// - /// Initialize rate limiters for all configured APIs - /// - public async Task InitializeAsync(CancellationToken cancellationToken = default) - { - using var conn = _connectionFactory.CreateConnection(); - await conn.OpenAsync(cancellationToken); - - foreach (var (apiName, (limit, windowSeconds)) in ApiQuotas) - { - const string sql = """ - INSERT INTO infrastructure.rate_limit_quota (api_name, limit_count, window_seconds, current_tokens, last_reset_at, published_at) - VALUES (@apiName, @limit, @windowSeconds, @limit, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP) - ON CONFLICT (api_name) DO UPDATE - SET current_tokens = EXCLUDED.current_tokens, - last_reset_at = CURRENT_TIMESTAMP, - published_at = CURRENT_TIMESTAMP - """; - - await conn.ExecuteAsync(sql, new - { - apiName, - limit, - windowSeconds - }); - - _quotas[apiName] = new RateLimitQuota - { - ApiName = apiName, - Limit = limit, - WindowSeconds = windowSeconds, - CurrentTokens = limit, - LastResetAt = DateTime.UtcNow - }; - } - } - - /// - /// Check if request is allowed and consume tokens - /// Atomic: either all tokens consumed or none - /// Returns HTTP 429 (Too Many Requests) if quota exceeded - /// - public async Task CheckRateLimitAsync( - string apiName, - int tokensRequested = 1, - Guid? requestId = null, - CancellationToken cancellationToken = default) - { - if (string.IsNullOrWhiteSpace(apiName)) - throw new ArgumentException("API name is required", nameof(apiName)); - - if (!ApiQuotas.ContainsKey(apiName)) - throw new ArgumentException($"Unknown API: {apiName}", nameof(apiName)); - - requestId ??= Guid.NewGuid(); - - // Get current quota from cache (or DB if expired) - var quota = await GetCurrentQuotaAsync(apiName, cancellationToken); - - // Check token availability - if (quota.CurrentTokens < tokensRequested) - { - // Decision: REJECTED (quota exceeded) - await LogRateLimitEventAsync( - apiName, - requestId.Value, - "rejected", - tokensRequested, - 0, - quota.CurrentTokens, - cancellationToken); - - var retryAfter = quota.NextResetSeconds; - - return new RateLimitDecisionDto - { - ApiName = apiName, - RequestId = requestId.Value, - Decision = "rejected", - TokensRequested = tokensRequested, - TokensUsed = 0, - RemainingTokens = quota.CurrentTokens, - RetryAfterSeconds = retryAfter, - HttpStatusCode = 429 - }; - } - - // Consume tokens atomically - var newTokenCount = quota.CurrentTokens - tokensRequested; - await UpdateTokensAtomicAsync(apiName, newTokenCount, cancellationToken); - - // Decision: ALLOWED - await LogRateLimitEventAsync( - apiName, - requestId.Value, - "allowed", - tokensRequested, - tokensRequested, - newTokenCount, - cancellationToken); - - return new RateLimitDecisionDto - { - ApiName = apiName, - RequestId = requestId.Value, - Decision = "allowed", - TokensRequested = tokensRequested, - TokensUsed = tokensRequested, - RemainingTokens = newTokenCount, - RetryAfterSeconds = null, - HttpStatusCode = 200 - }; - } - - /// - /// Get rate limit status for an API - /// - public async Task GetStatusAsync( - string apiName, - CancellationToken cancellationToken = default) - { - if (!ApiQuotas.ContainsKey(apiName)) - throw new ArgumentException($"Unknown API: {apiName}", nameof(apiName)); - - var quota = await GetCurrentQuotaAsync(apiName, cancellationToken); - - return new RateLimitStatusDto - { - ApiName = apiName, - Limit = quota.Limit, - WindowSeconds = quota.WindowSeconds, - CurrentTokens = quota.CurrentTokens, - LastResetAt = quota.LastResetAt, - NextResetAt = quota.LastResetAt.AddSeconds(quota.WindowSeconds), - UtilizationPercent = (quota.Limit - quota.CurrentTokens) * 100 / quota.Limit - }; - } - - // Private methods - - private async Task GetCurrentQuotaAsync( - string apiName, - CancellationToken cancellationToken) - { - // Try cache first - var cacheKey = $"ratelimit:{apiName}"; - var cached = await _cache.GetStringAsync(cacheKey); - if (cached != null && int.TryParse(cached, out var cachedTokens)) - { - if (_quotas.TryGetValue(apiName, out var cachedQuota)) - { - cachedQuota.CurrentTokens = cachedTokens; - return cachedQuota; - } - } - - // Get from DB if expired or not in cache - using var conn = _connectionFactory.CreateConnection(); - await conn.OpenAsync(cancellationToken); - - const string sql = """ - SELECT api_name, limit_count, window_seconds, current_tokens, last_reset_at - FROM infrastructure.rate_limit_quota - WHERE api_name = @apiName - LIMIT 1 - """; - - var row = await conn.QuerySingleOrDefaultAsync(sql, new { apiName }); - if (row is null) - throw new InvalidOperationException($"No rate limit quota configured for {apiName}"); - - var quota = new RateLimitQuota - { - ApiName = row.api_name, - Limit = row.limit_count, - WindowSeconds = row.window_seconds, - CurrentTokens = row.current_tokens, - LastResetAt = row.last_reset_at - }; - - // Reset if window expired - var windowElapsed = (DateTime.UtcNow - quota.LastResetAt).TotalSeconds; - if (windowElapsed >= quota.WindowSeconds) - { - quota.CurrentTokens = quota.Limit; - quota.LastResetAt = DateTime.UtcNow; - await ResetQuotaAsync(apiName, cancellationToken); - } - - // Cache for next check - await _cache.SetStringAsync(cacheKey, quota.CurrentTokens.ToString(), TimeSpan.FromSeconds(5)); - - return quota; - } - - private async Task UpdateTokensAtomicAsync( - string apiName, - decimal newTokenCount, - CancellationToken cancellationToken) - { - using var conn = _connectionFactory.CreateConnection(); - await conn.OpenAsync(cancellationToken); - - const string sql = """ - UPDATE infrastructure.rate_limit_quota - SET current_tokens = @newTokenCount, - updated_at = CURRENT_TIMESTAMP, - published_at = CURRENT_TIMESTAMP - WHERE api_name = @apiName - """; - - var rowsAffected = await conn.ExecuteAsync(sql, new { apiName, newTokenCount }); - if (rowsAffected != 1) - throw new InvalidOperationException($"Failed to update rate limit quota for {apiName}"); - - // Invalidate cache - await _cache.RemoveAsync($"ratelimit:{apiName}"); - } - - private async Task ResetQuotaAsync(string apiName, CancellationToken cancellationToken) - { - using var conn = _connectionFactory.CreateConnection(); - await conn.OpenAsync(cancellationToken); - - const string sql = """ - UPDATE infrastructure.rate_limit_quota - SET current_tokens = limit_count, - last_reset_at = CURRENT_TIMESTAMP, - updated_at = CURRENT_TIMESTAMP, - published_at = CURRENT_TIMESTAMP - WHERE api_name = @apiName - """; - - await conn.ExecuteAsync(sql, new { apiName }); - } - - private async Task LogRateLimitEventAsync( - string apiName, - Guid requestId, - string decision, - int tokensRequested, - int tokensUsed, - decimal remainingTokens, - CancellationToken cancellationToken) - { - using var conn = _connectionFactory.CreateConnection(); - await conn.OpenAsync(cancellationToken); - - const string sql = """ - INSERT INTO infrastructure.rate_limit_events (api_name, request_id, decision, tokens_requested, tokens_used, remaining_tokens, occurred_at, published_at) - VALUES (@apiName, @requestId, @decision, @tokensRequested, @tokensUsed, @remainingTokens, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP) - """; - - await conn.ExecuteAsync(sql, new - { - apiName, - requestId, - decision, - tokensRequested, - tokensUsed, - remainingTokens - }); - } -} - -public sealed class RateLimitQuota -{ - public string ApiName { get; set; } = string.Empty; - public int Limit { get; set; } - public int WindowSeconds { get; set; } - public decimal CurrentTokens { get; set; } - public DateTime LastResetAt { get; set; } - - public int NextResetSeconds - { - get - { - var elapsed = (DateTime.UtcNow - LastResetAt).TotalSeconds; - var remaining = WindowSeconds - elapsed; - return Math.Max(1, (int)Math.Ceiling(remaining)); - } - } -} - -public sealed class RateLimitDecisionDto -{ - public string ApiName { get; set; } = string.Empty; - public Guid RequestId { get; set; } - public string Decision { get; set; } = string.Empty; // "allowed" or "rejected" - public int TokensRequested { get; set; } - public int TokensUsed { get; set; } - public decimal RemainingTokens { get; set; } - public int? RetryAfterSeconds { get; set; } - public int HttpStatusCode { get; set; } -} - -public sealed class RateLimitStatusDto -{ - public string ApiName { get; set; } = string.Empty; - public int Limit { get; set; } - public int WindowSeconds { get; set; } - public decimal CurrentTokens { get; set; } - public DateTime LastResetAt { get; set; } - public DateTime NextResetAt { get; set; } - public int UtilizationPercent { get; set; } -} diff --git a/src/KArtSell.Host/Jobs/OpenDartDailyBatchJob.cs b/src/KArtSell.Host/Jobs/OpenDartDailyBatchJob.cs deleted file mode 100644 index 58fb457c..00000000 --- a/src/KArtSell.Host/Jobs/OpenDartDailyBatchJob.cs +++ /dev/null @@ -1,111 +0,0 @@ -using Hangfire; -using KArtSell.BuildingBlocks.Time; -using Serilog; - -namespace KArtSell.Host.Jobs; - -/// -/// OpenDart Daily Batch Job -/// Scheduled: 09:00 KST daily -/// Purpose: Fetch quarterly financial data for all tracked tickers -/// Idempotent: Safe to retry on same day (batch_date is unique key) -/// -[DisableConcurrentExecution(timeoutInSeconds: 3600)] // Max 1 hour -public sealed class OpenDartDailyBatchJob -{ - private readonly Observability.OpenDartService _openDartService; - private readonly IClock _clock; - private readonly ILogger _logger; - - public OpenDartDailyBatchJob( - Observability.OpenDartService openDartService, - IClock clock, - ILogger logger) - { - _openDartService = openDartService ?? throw new ArgumentNullException(nameof(openDartService)); - _clock = clock ?? throw new ArgumentNullException(nameof(clock)); - _logger = logger ?? throw new ArgumentNullException(nameof(logger)); - } - - /// - /// Execute daily OpenDart batch. - /// Idempotent: same batch_date always produces same result - /// - public async Task Execute(CancellationToken cancellationToken = default) - { - var jobId = CorrelationId.NewId(); - var now = _clock.UtcNow; - - _logger.Information( - "OpenDart daily batch started | JobId={JobId} | ScheduledFor={ScheduledFor}", - jobId, now); - - try - { - await _openDartService.ExecuteDailyBatchAsync(cancellationToken); - - _logger.Information( - "OpenDart daily batch completed successfully | JobId={JobId}", - jobId); - } - catch (OperationCanceledException) - { - _logger.Warning( - "OpenDart daily batch cancelled | JobId={JobId}", - jobId); - throw; - } - catch (InvalidOperationException ex) when (ex.Message.Contains("quota exceeded")) - { - _logger.Warning( - "OpenDart daily batch quota limit hit | JobId={JobId} | Error={Error}", - jobId, ex.Message); - throw; - } - catch (Exception ex) - { - _logger.Error( - ex, - "OpenDart daily batch failed | JobId={JobId}", - jobId); - throw; - } - } -} - -/// -/// Extension to register OpenDart batch job with Hangfire -/// Schedule: 09:00 KST every day -/// -public static class OpenDartBatchJobExtensions -{ - public static IServiceCollection AddOpenDartBatchJob(this IServiceCollection services) - { - // Registrations are handled in Program.cs - return services; - } - - /// - /// Register recurring OpenDart batch job - /// Called from Program.cs during Host startup - /// - public static void RegisterOpenDartBatchJob(this IRecurringJobManager recurringJobManager) - { - recurringJobManager.AddOrUpdate( - jobId: "opendart-daily-batch", - methodCall: job => job.Execute(default), - cronExpression: Cron.Daily(9, 0), // 09:00 every day (UTC) - options: new RecurringJobOptions - { - TimeZone = TimeZoneInfo.FindSystemTimeZoneById("Korea Standard Time") // 09:00 KST - }); - } -} - -/// -/// Helper for correlation tracking -/// -internal static class CorrelationId -{ - public static Guid NewId() => Guid.NewGuid(); -} diff --git a/src/KArtSell.Host/Observability/OpenDartService.cs b/src/KArtSell.Host/Observability/OpenDartService.cs deleted file mode 100644 index 2fefe126..00000000 --- a/src/KArtSell.Host/Observability/OpenDartService.cs +++ /dev/null @@ -1,282 +0,0 @@ -using KArtSell.BuildingBlocks.Data; -using Dapper; -using Npgsql; -using System.Collections.Concurrent; - -namespace KArtSell.Host.Observability; - -/// -/// OpenDart Financial Data Service -/// Implements AGENTS.md v16.0 constraints: idempotent, cached, rate-limited -/// 3-month cache TTL, 1000 req/day quota, single batch per day -/// -public sealed class OpenDartService : IDisposable -{ - private readonly IDbConnectionFactory _connectionFactory; - private readonly string _apiKey; - private readonly ConcurrentDictionary _quotaResetDates; - private readonly SemaphoreSlim _quotaLock; - - public OpenDartService(IDbConnectionFactory connectionFactory, string apiKey) - { - _connectionFactory = connectionFactory ?? throw new ArgumentNullException(nameof(connectionFactory)); - _apiKey = apiKey ?? throw new ArgumentNullException(nameof(apiKey)); - _quotaResetDates = new(); - _quotaLock = new(1, 1); - } - - /// - /// Get financial data from OpenDart API with 3-month caching. - /// Returns cached data if available and not expired; fetches fresh data otherwise. - /// Idempotent: same ticker+quarter always returns same result. - /// - public async Task GetFinancialDataAsync( - string ticker, - string quarter, // Format: "2024-Q1" - CancellationToken cancellationToken = default) - { - if (string.IsNullOrWhiteSpace(ticker)) - throw new ArgumentException("Ticker is required", nameof(ticker)); - - if (string.IsNullOrWhiteSpace(quarter) || !IsValidQuarterFormat(quarter)) - throw new ArgumentException("Quarter must be in format YYYY-QN", nameof(quarter)); - - using var conn = _connectionFactory.CreateConnection(); - await conn.OpenAsync(cancellationToken); - - // Check cache first - var cachedData = await GetCachedDataAsync(conn, ticker, quarter, cancellationToken); - if (cachedData != null) - return cachedData; - - // Check quota before fetching - var quotaKey = $"opendart:{DateTime.UtcNow:yyyy-MM-dd}"; - await _quotaLock.WaitAsync(cancellationToken); - try - { - var quotaUsed = await GetDailyQuotaUsedAsync(conn, cancellationToken); - if (quotaUsed >= 1000) - { - throw new InvalidOperationException( - $"OpenDart quota exceeded (1000/day): {quotaUsed} already used"); - } - } - finally - { - _quotaLock.Release(); - } - - // Fetch from API (TODO: implement actual OpenDart API call) - var freshData = await FetchFromOpenDartApiAsync(ticker, quarter, cancellationToken); - - // Cache the result - await CacheDataAsync(conn, ticker, quarter, freshData, cancellationToken); - - // Log batch usage - await LogBatchUsageAsync(conn, 1, "success", cancellationToken); - - return freshData; - } - - /// - /// Execute daily batch for all configured tickers. - /// Idempotent: safe to retry on same day. - /// - public async Task ExecuteDailyBatchAsync(CancellationToken cancellationToken = default) - { - var batchDate = DateTime.UtcNow.Date; - using var conn = _connectionFactory.CreateConnection(); - await conn.OpenAsync(cancellationToken); - - // Check if batch already executed today - var existingBatch = await CheckBatchExecutedAsync(conn, batchDate, cancellationToken); - if (existingBatch && existingBatch.Status == "success") - return; // Idempotent: skip if already successful - - try - { - var tickers = new[] { "005930", "000660", "068270" }; // Samsung, SK Hynix, Naver (example) - var currentYear = DateTime.UtcNow.Year; - var currentQuarter = (DateTime.UtcNow.Month - 1) / 3 + 1; - - var quotaUsed = 0; - foreach (var ticker in tickers) - { - var quarter = $"{currentYear}-Q{currentQuarter}"; - try - { - await GetFinancialDataAsync(ticker, quarter, cancellationToken); - quotaUsed++; - } - catch (InvalidOperationException ex) when (ex.Message.Contains("quota exceeded")) - { - // Partial success: log and continue - await LogBatchUsageAsync(conn, quotaUsed, "quota_exceeded", cancellationToken); - throw; - } - } - - await LogBatchUsageAsync(conn, quotaUsed, "success", cancellationToken); - } - catch (Exception ex) - { - await LogBatchUsageAsync(conn, 0, "failed", ex.Message, cancellationToken); - throw; - } - } - - // Private methods - - private bool IsValidQuarterFormat(string quarter) - { - if (quarter.Length != 7) return false; // "2024-Q1" - if (quarter[4] != '-') return false; - if (quarter[5] != 'Q') return false; - if (!int.TryParse(quarter.Substring(0, 4), out var year)) return false; - if (!int.TryParse(quarter.Substring(6, 1), out var q) || q < 1 || q > 4) return false; - return year >= 2000 && year <= 2100; - } - - private async Task GetCachedDataAsync( - DbConnection conn, - string ticker, - string quarter, - CancellationToken cancellationToken) - { - const string sql = """ - SELECT data_json, cached_at, expires_at - FROM opendata.opendart_cache - WHERE ticker = @ticker - AND quarter = @quarter - AND expires_at > CURRENT_TIMESTAMP AT TIME ZONE 'UTC' - LIMIT 1 - """; - - var row = await conn.QuerySingleOrDefaultAsync( - sql, - new { ticker, quarter }); - - if (row is null) - return null; - - return new OpenDartDataDto - { - Ticker = ticker, - Quarter = quarter, - DataJson = row.data_json, - CachedAt = row.cached_at - }; - } - - private async Task GetDailyQuotaUsedAsync(DbConnection conn, CancellationToken cancellationToken) - { - const string sql = """ - SELECT COALESCE(SUM(quota_used), 0) as total_used - FROM opendata.opendart_batch_log - WHERE batch_date = CURRENT_DATE AT TIME ZONE 'UTC' - AND status IN ('success', 'partial') - """; - - var result = await conn.QueryFirstOrDefaultAsync(sql); - return result?.total_used ?? 0; - } - - private async Task<(bool Exists, string Status)?> CheckBatchExecutedAsync( - DbConnection conn, - DateTime batchDate, - CancellationToken cancellationToken) - { - const string sql = """ - SELECT status - FROM opendata.opendart_batch_log - WHERE batch_date = @batchDate - LIMIT 1 - """; - - var result = await conn.QuerySingleOrDefaultAsync(sql, new { batchDate }); - if (result is null) - return null; - - return (true, result.status); - } - - private async Task CacheDataAsync( - DbConnection conn, - string ticker, - string quarter, - OpenDartDataDto data, - CancellationToken cancellationToken) - { - const string sql = """ - INSERT INTO opendata.opendart_cache (ticker, quarter, data_json, cached_at, expires_at, published_at) - VALUES (@ticker, @quarter, @dataJson, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP + INTERVAL '90 days', CURRENT_TIMESTAMP) - ON CONFLICT (ticker, quarter) DO UPDATE - SET data_json = EXCLUDED.data_json, - cached_at = CURRENT_TIMESTAMP, - expires_at = CURRENT_TIMESTAMP + INTERVAL '90 days', - published_at = CURRENT_TIMESTAMP - """; - - await conn.ExecuteAsync(sql, new - { - ticker, - quarter, - dataJson = data.DataJson - }); - } - - private async Task LogBatchUsageAsync( - DbConnection conn, - int quotaUsed, - string status, - string? errorMessage = null, - CancellationToken cancellationToken = default) - { - const string sql = """ - INSERT INTO opendata.opendart_batch_log (batch_date, quota_limit, quota_used, status, error_message, executed_at, published_at) - VALUES (CURRENT_DATE, 1000, @quotaUsed, @status, @errorMessage, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP) - ON CONFLICT (batch_date) DO UPDATE - SET quota_used = GREATEST(opendart_batch_log.quota_used, EXCLUDED.quota_used), - status = EXCLUDED.status, - error_message = COALESCE(EXCLUDED.error_message, opendart_batch_log.error_message), - executed_at = CURRENT_TIMESTAMP, - published_at = CURRENT_TIMESTAMP - """; - - await conn.ExecuteAsync(sql, new - { - quotaUsed, - status, - errorMessage - }); - } - - private async Task FetchFromOpenDartApiAsync( - string ticker, - string quarter, - CancellationToken cancellationToken) - { - // TODO: Implement actual OpenDart API call using _apiKey - // For now, return placeholder - return new OpenDartDataDto - { - Ticker = ticker, - Quarter = quarter, - DataJson = "{}", - CachedAt = DateTime.UtcNow - }; - } - - public void Dispose() - { - _quotaLock?.Dispose(); - } -} - -public sealed class OpenDartDataDto -{ - public required string Ticker { get; set; } - public required string Quarter { get; set; } - public required string DataJson { get; set; } - public required DateTime CachedAt { get; set; } -}