using FastEndpoints; using Hangfire; using Npgsql; using System.Text.Json; using KArtSell.BuildingBlocks.Time; using KArtSell.Modules.ModelOperations.Domain; namespace KArtSell.Host.Features.MarketData; /// /// VS-03 BE: Market Data Ingestion Endpoints /// POST /api/market/ingest - Trigger data ingestion /// GET /api/market/ingest/{jobId} - Check job status /// /// Schedules market data collection from KRX/OpenDart /// - Idempotent by date range + data source /// - Returns 202 Accepted (async processing) /// - Audit trail with correlation ID /// public sealed class IngestionRequest { public string DataSource { get; set; } = "KRX"; // "KRX", "OpenDart", "Stub" public string FromDate { get; set; } = ""; // "2026-01-01" public string ToDate { get; set; } = ""; // "2026-12-31" } public sealed class IngestionResponse { public Guid JobId { get; set; } public string Status { get; set; } = "Queued"; public int ExpectedRowCount { get; set; } public DateTime QueuedAt { get; set; } } public sealed class IngestionStatusResponse { public Guid JobId { get; set; } public string Status { get; set; } = "Running"; public int RowsProcessed { get; set; } public int RowsFailed { get; set; } public int RowsSkipped { get; set; } public DateTime? CompletedAt { get; set; } public int? DurationSeconds { get; set; } public string? ErrorMessage { get; set; } } public sealed class TriggerIngestionEndpoint : Endpoint { private readonly IMarketDataIngestionService _ingestionService; private readonly IClock _clock; public TriggerIngestionEndpoint(IMarketDataIngestionService ingestionService, IClock clock) { _ingestionService = ingestionService; _clock = clock; } public override void Configure() { Post("/api/market/ingest"); Roles("DataAdmin"); AllowAnonymous(); } public override async Task HandleAsync(IngestionRequest req, CancellationToken ct) { if (!DateOnly.TryParse(req.FromDate, out var fromDate)) { ThrowError("Invalid FromDate format. Use YYYY-MM-DD"); } if (!DateOnly.TryParse(req.ToDate, out var toDate)) { ThrowError("Invalid ToDate format. Use YYYY-MM-DD"); } if (fromDate > toDate) { ThrowError("FromDate must be <= ToDate"); } var correlationId = HttpContext.TraceIdentifier; var (jobId, expectedCount) = await _ingestionService.ScheduleIngestionAsync( dataSource: req.DataSource, fromDate: fromDate, toDate: toDate, correlationId: correlationId, cancellationToken: ct); HttpContext.Response.StatusCode = StatusCodes.Status202Accepted; HttpContext.Response.ContentType = "application/json"; await HttpContext.Response.WriteAsync(JsonSerializer.Serialize(new IngestionResponse { JobId = jobId, Status = "Queued", ExpectedRowCount = expectedCount, QueuedAt = _clock.UtcNow.DateTime, }), ct); } } public sealed class GetIngestionStatusEndpoint : EndpointWithoutRequest { private readonly IMarketDataIngestionService _ingestionService; public GetIngestionStatusEndpoint(IMarketDataIngestionService ingestionService) { _ingestionService = ingestionService; } public override void Configure() { Get("/api/market/ingest/{jobId}"); AllowAnonymous(); } public override async Task HandleAsync(CancellationToken ct) { var jobIdStr = Route("jobId"); if (!Guid.TryParse(jobIdStr, out var jobId)) { ThrowError("Invalid job ID format"); } var status = await _ingestionService.GetIngestionStatusAsync(jobId, ct); if (status == null) { ThrowError("Job not found"); } HttpContext.Response.StatusCode = StatusCodes.Status200OK; HttpContext.Response.ContentType = "application/json"; await HttpContext.Response.WriteAsync(JsonSerializer.Serialize(new IngestionStatusResponse { JobId = status.JobId, Status = status.Status, RowsProcessed = status.RowsProcessed, RowsFailed = status.RowsFailed, RowsSkipped = status.RowsSkipped, CompletedAt = status.CompletedAt, DurationSeconds = status.DurationSeconds, ErrorMessage = status.ErrorMessage, }), ct); } } /// /// VS-03 Application Handler: Orchestrates ingestion /// /// Responsibilities: /// - Schedule ingestion job (Hangfire) /// - Validate date range /// - Check idempotency (same date range = no re-run) /// - Audit logging /// public interface IMarketDataIngestionService { Task<(Guid JobId, int ExpectedRowCount)> ScheduleIngestionAsync( string dataSource, DateOnly fromDate, DateOnly toDate, string correlationId, CancellationToken cancellationToken); Task GetIngestionStatusAsync(Guid jobId, CancellationToken cancellationToken); } public record IngestionJobStatus( Guid JobId, string Status, int RowsProcessed, int RowsFailed, int RowsSkipped, DateTime? CompletedAt, int? DurationSeconds, string? ErrorMessage); public class MarketDataIngestionService : IMarketDataIngestionService { private readonly NpgsqlDataSource _dataSource; private readonly IBackgroundJobClient _jobClient; public MarketDataIngestionService(NpgsqlDataSource dataSource, IBackgroundJobClient jobClient) { _dataSource = dataSource; _jobClient = jobClient; } public async Task<(Guid JobId, int ExpectedRowCount)> ScheduleIngestionAsync( string dataSource, DateOnly fromDate, DateOnly toDate, string correlationId, CancellationToken cancellationToken) { var jobId = Guid.NewGuid(); // Check idempotency: Is there already a job for this date range? const string checkSql = """ SELECT job_id FROM market_data.ingestion_jobs WHERE data_source = @source AND from_date = @fromDate AND to_date = @toDate AND status IN ('Running', 'Completed') LIMIT 1; """; await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken); await using var checkCmd = connection.CreateCommand(); checkCmd.CommandText = checkSql; checkCmd.Parameters.AddWithValue("@source", dataSource); checkCmd.Parameters.AddWithValue("@fromDate", fromDate.ToDateTime(TimeOnly.MinValue)); checkCmd.Parameters.AddWithValue("@toDate", toDate.ToDateTime(TimeOnly.MinValue)); var existingJobId = await checkCmd.ExecuteScalarAsync(cancellationToken); if (existingJobId != null) { return ((Guid)existingJobId, 0); } // Estimate row count (rough: days * ~2000 stocks) var days = (toDate.DayNumber - fromDate.DayNumber) + 1; var expectedCount = days * 2000; // Stub estimate // Insert job record const string insertSql = """ INSERT INTO market_data.ingestion_jobs (job_id, data_source, from_date, to_date, status, correlation_id, triggered_by) VALUES (@jobId, @source, @fromDate, @toDate, 'Queued', @correlationId, 'API'); """; await using var insertCmd = connection.CreateCommand(); insertCmd.CommandText = insertSql; insertCmd.Parameters.AddWithValue("@jobId", jobId); insertCmd.Parameters.AddWithValue("@source", dataSource); insertCmd.Parameters.AddWithValue("@fromDate", fromDate.ToDateTime(TimeOnly.MinValue)); insertCmd.Parameters.AddWithValue("@toDate", toDate.ToDateTime(TimeOnly.MinValue)); insertCmd.Parameters.AddWithValue("@correlationId", correlationId); await insertCmd.ExecuteNonQueryAsync(cancellationToken); // Schedule Hangfire job _jobClient.Enqueue(j => j.ExecuteAsync(jobId, dataSource, fromDate, toDate, correlationId, CancellationToken.None)); return (jobId, expectedCount); } public async Task GetIngestionStatusAsync(Guid jobId, CancellationToken cancellationToken) { const string sql = """ SELECT job_id, status, rows_processed, rows_failed, rows_skipped, completed_at, duration_seconds, last_error_message FROM market_data.ingestion_jobs WHERE job_id = @jobId; """; await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken); await using var cmd = connection.CreateCommand(); cmd.CommandText = sql; cmd.Parameters.AddWithValue("@jobId", jobId); await using var reader = await cmd.ExecuteReaderAsync(cancellationToken); if (!await reader.ReadAsync(cancellationToken)) return null; return new IngestionJobStatus( JobId: reader.GetGuid(0), Status: reader.GetString(1), RowsProcessed: reader.IsDBNull(2) ? 0 : reader.GetInt32(2), RowsFailed: reader.IsDBNull(3) ? 0 : reader.GetInt32(3), RowsSkipped: reader.IsDBNull(4) ? 0 : reader.GetInt32(4), CompletedAt: reader.IsDBNull(5) ? null : reader.GetDateTime(5), DurationSeconds: reader.IsDBNull(6) ? null : reader.GetInt32(6), ErrorMessage: reader.IsDBNull(7) ? null : reader.GetString(7)); } } /// /// Abstraction: Market data ingestion job (Hangfire worker) /// public interface IMarketDataIngestionJob { Task ExecuteAsync(Guid jobId, string dataSource, DateOnly fromDate, DateOnly toDate, string correlationId, CancellationToken ct); }