ANTOOZA-BACKEND/Infrastructure/StockData/KrxEsgSyncService.cs

using Application.Abstractions.Data;
using Application.Helpers;
using Domain.Entities.Stocks;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using SharedKernel;

namespace Infrastructure.StockData;

/// <summary>
/// ESG 데이터(ESG 증권상품 + ESG 지수 + 사회책임투자채권) 수집 — KRX OpenAPI (esg/esg_etp_info + esg_index_info + sri_bond_info).
/// 세 엔드포인트 모두 basDd 파라미터로 그날 전량(스냅샷)을 반환하므로 공용 KrxBackfill 로 최근 BackfillYears(기본 3)년치를 endDate(직전 영업일)부터
/// 과거로 훑으며 미적재일만 채운다. quota 보호를 위해 1회 실행당 BackfillMaxPerRun(기본 60)일까지만 fetch → 여러 날에 걸쳐 3년치를 메우고 이후엔 최신만 유지.
/// 각 날짜는 3개 엔드포인트를 모두 수집해 EsgSecurity(UQ Name+TradeDate) / EsgIndexDailyPrice(UQ IndexName+TradeDate) / SriBond(UQ Code+TradeDate) upsert.
/// 세 테이블 모두 해당 날짜가 있어야 "적재됨"으로 간주(부분 적재 방어). 기본 19:30 KST 실행(일반상품 19:15 뒤 여유). ApiKey 미설정 시 skip.
/// </summary>
internal sealed class KrxEsgSyncService(
    IServiceScopeFactory scopeFactory,
    IHttpClientFactory httpClientFactory,
    IOptions<AppSettings> settings,
    ILogger<KrxEsgSyncService> logger
) : DailyScheduledService(logger)
{
    private const string EtpPath = "/svc/apis/esg/esg_etp_info";
    private const string IndexPath = "/svc/apis/esg/esg_index_info";
    private const string SriBondPath = "/svc/apis/esg/sri_bond_info";

    protected override string JobName => "KrxEsgSync";

    protected override TimeOnly TargetTime => ParseTime(settings.Value.KRXCoKr.EsgSyncTime, new TimeOnly(19, 30));

    protected override int MaxRetryCount => 2;

    protected override TimeSpan RetryDelay => TimeSpan.FromHours(2);

    protected override async Task<bool> RunOnceAsync(DateOnly todayKst, CancellationToken ct)
    {
        var cfg = settings.Value.KRXCoKr;

        using var scope = scopeFactory.CreateScope();
        var collectorSettings = scope.ServiceProvider.GetRequiredService<ICollectorSettingsProvider>();
        if (!await collectorSettings.IsEnabledAsync(CollectorFlag.KrxEsg, ct))
        {
            return true;
        }
        cfg = cfg with { ApiKey = await collectorSettings.GetKeyAsync(CollectorKey.Krx, ct) ?? cfg.ApiKey };
        if (string.IsNullOrWhiteSpace(cfg.ApiKey))
        {
            Logger.LogWarning("[{Job}] KRXCoKr:ApiKey 미설정 — 수집 skip", JobName);
            return true;
        }
        var db = scope.ServiceProvider.GetRequiredService<IAppDbContext>();
        var client = httpClientFactory.CreateClient(KrxCoKrHttp.ClientName);

        var endDate = await MarketCalendar.GetPreviousBusinessDayAsync(db, todayKst.AddDays(1), ct);
        var years = cfg.BackfillYears > 0 ? cfg.BackfillYears : 3;
        var startDate = todayKst.AddYears(-years);

        var holidays = (await db.MarketHoliday.AsNoTracking()
            .Where(c => c.Date >= startDate && c.Date <= endDate)
            .Select(c => c.Date)
            .ToListAsync(ct)).ToHashSet();

        var maxPerRun = cfg.BackfillMaxPerRun > 0 ? cfg.BackfillMaxPerRun : 60;

        var fetched = await KrxBackfill.RunAsync(
            existsForDate: (day, token) => AllPresentForDateAsync(db, day, token),
            fetchAndUpsertForDate: (day, token) => FetchAndUpsertAsync(db, client, cfg.BaseUrl, cfg.ApiKey, day, token),
            startDate: startDate,
            endDate: endDate,
            holidays: holidays,
            maxPerRun: maxPerRun,
            delayMs: 300,
            ct: ct);

        Logger.LogInformation("[{Job}] 완료 — 창=[{Start}~{End}], 이번 실행 fetch={Fetched}일 (maxPerRun={Max})",
            JobName, startDate, endDate, fetched, maxPerRun);

        // 최신 영업일(endDate) 데이터가 이번 실행에서도 미적재면 false → 베이스가 RetryDelay 후 재시도 (KRX T+0 마감 데이터 미반영 대비)
        if (!await AllPresentForDateAsync(db, endDate, ct))
        {
            Logger.LogWarning("[{Job}] 최신 영업일 {End} 데이터 미적재 — {Delay} 후 재시도 (최대 {Max}회)", JobName, endDate, RetryDelay, MaxRetryCount);
            return false;
        }

        return true;
    }

    /// <summary>세 테이블 모두 해당 날짜가 적재되어 있는지 (부분 적재 시 refetch 하도록 AND 조건).</summary>
    private static async Task<bool> AllPresentForDateAsync(IAppDbContext db, DateOnly day, CancellationToken ct)
    {
        var etp = await db.EsgSecurity.AsNoTracking().AnyAsync(c => c.TradeDate == day, ct);
        var index = await db.EsgIndexDailyPrice.AsNoTracking().AnyAsync(c => c.TradeDate == day, ct);
        var sri = await db.SriBond.AsNoTracking().AnyAsync(c => c.TradeDate == day, ct);
        return etp && index && sri;
    }

    /// <summary>한 날짜에 대해 3개 ESG 엔드포인트를 모두 수집하고 각 테이블 upsert.</summary>
    private async Task FetchAndUpsertAsync(IAppDbContext db, HttpClient client, string baseUrl, string apiKey, DateOnly day, CancellationToken ct)
    {
        var trimmed = baseUrl.TrimEnd('/');

        await UpsertEsgSecuritiesAsync(db, client, trimmed, apiKey, day, ct);
        await UpsertEsgIndicesAsync(db, client, trimmed, apiKey, day, ct);
        await UpsertSriBondsAsync(db, client, trimmed, apiKey, day, ct);
    }

    private async Task UpsertEsgSecuritiesAsync(IAppDbContext db, HttpClient client, string baseUrl, string apiKey, DateOnly day, CancellationToken ct)
    {
        var url = $"{baseUrl}{EtpPath}?basDd={day:yyyyMMdd}";
        var json = await KrxCoKrHttp.GetStringWithRetryAsync(client, url, apiKey, Logger, ct);
        var rows = KrxEsgSecurityParser.ParseDaily(json);

        Logger.LogInformation("[{Job}] ESG증권상품 basDd={Day} rows={Rows}", JobName, day, rows.Count);

        if (rows.Count == 0)
        {
            return;
        }

        var existing = await db.EsgSecurity.Where(c => c.TradeDate == day).ToListAsync(ct);
        var existingByKey = existing.ToDictionary(c => c.Name);

        foreach (var row in rows)
        {
            if (existingByKey.TryGetValue(row.Name, out var entity))
            {
                entity.Update(row.Close, row.ChangeAmount, row.ChangeRate, row.ListedShares, row.Volume, row.TradeValue);
            }
            else
            {
                var created = EsgSecurity.Create(row.Name, row.TradeDate, row.Close, row.ChangeAmount, row.ChangeRate, row.ListedShares, row.Volume, row.TradeValue);
                await db.EsgSecurity.AddAsync(created, ct);
                existingByKey[row.Name] = created;
            }
        }

        await db.SaveChangesAsync(ct);
    }

    private async Task UpsertEsgIndicesAsync(IAppDbContext db, HttpClient client, string baseUrl, string apiKey, DateOnly day, CancellationToken ct)
    {
        var url = $"{baseUrl}{IndexPath}?basDd={day:yyyyMMdd}";
        var json = await KrxCoKrHttp.GetStringWithRetryAsync(client, url, apiKey, Logger, ct);
        var rows = KrxEsgIndexParser.ParseDaily(json);

        Logger.LogInformation("[{Job}] ESG지수 basDd={Day} rows={Rows}", JobName, day, rows.Count);

        if (rows.Count == 0)
        {
            return;
        }

        var existing = await db.EsgIndexDailyPrice.Where(c => c.TradeDate == day).ToListAsync(ct);
        var existingByKey = existing.ToDictionary(c => c.IndexName);

        foreach (var row in rows)
        {
            if (existingByKey.TryGetValue(row.IndexName, out var entity))
            {
                entity.Update(row.Close, row.ChangeVal, row.FlucRateBp, row.ConstituentCount, row.Volume, row.Value);
            }
            else
            {
                var created = EsgIndexDailyPrice.Create(row.IndexName, row.TradeDate, row.Close, row.ChangeVal, row.FlucRateBp, row.ConstituentCount, row.Volume, row.Value);
                await db.EsgIndexDailyPrice.AddAsync(created, ct);
                existingByKey[row.IndexName] = created;
            }
        }

        await db.SaveChangesAsync(ct);
    }

    private async Task UpsertSriBondsAsync(IAppDbContext db, HttpClient client, string baseUrl, string apiKey, DateOnly day, CancellationToken ct)
    {
        var url = $"{baseUrl}{SriBondPath}?basDd={day:yyyyMMdd}";
        var json = await KrxCoKrHttp.GetStringWithRetryAsync(client, url, apiKey, Logger, ct);
        var rows = KrxSriBondParser.ParseDaily(json);

        Logger.LogInformation("[{Job}] 사회책임투자채권 basDd={Day} rows={Rows}", JobName, day, rows.Count);

        if (rows.Count == 0)
        {
            return;
        }

        var existing = await db.SriBond.Where(c => c.TradeDate == day).ToListAsync(ct);
        var existingByKey = existing.ToDictionary(c => c.Code);

        foreach (var row in rows)
        {
            if (existingByKey.TryGetValue(row.Code, out var entity))
            {
                entity.Update(row.IssuerName, row.SriBondType, row.Name, row.ListDate, row.IssueDate, row.RedemptionDate, row.CouponRate, row.IssueAmount, row.ListAmount, row.BondType);
            }
            else
            {
                var created = SriBond.Create(row.Code, row.TradeDate, row.IssuerName, row.SriBondType, row.Name, row.ListDate, row.IssueDate, row.RedemptionDate, row.CouponRate, row.IssueAmount, row.ListAmount, row.BondType);
                await db.SriBond.AddAsync(created, ct);
                existingByKey[row.Code] = created;
            }
        }

        await db.SaveChangesAsync(ct);
    }
}