using System.Text.Json; using Dapper; using FastEndpoints; using Npgsql; using Shared.Problems; namespace Modules.OMS.Orders.Ship; public sealed record ShipOrdersFilter( DateOnly From, DateOnly To, Guid? ChannelId, string? Status, string? Keyword, bool ExceptionOnly = false); public sealed record ShipOrdersRequest( string Mode, IReadOnlyList? Ids, ShipOrdersFilter? Filter, IReadOnlyList? ExcludedIds); public sealed record ShipOrdersResponse(int Requested, int Accepted, int Rejected); public sealed class Endpoint(NpgsqlDataSource dataSource) : Endpoint { public override void Configure() { Post("/api/oms/orders/ship"); Permissions("oms.order.ship"); } public override async Task HandleAsync(ShipOrdersRequest req, CancellationToken ct) { var key = HttpContext.Request.Headers["Idempotency-Key"].FirstOrDefault(); if (string.IsNullOrWhiteSpace(key)) { await Send.ResponseAsync(KbxValidationProblem.Create( new KbxValidationError(null, null, "IDEMPOTENCY_KEY_REQUIRED", "안전한 재처리를 위해 Idempotency-Key가 필요합니다.")), 400, cancellation: ct); return; } var mode = req.Mode?.Trim().ToLowerInvariant(); if (mode is not ("ids" or "filter") || (mode == "ids" && (req.Ids is null || req.Ids.Count == 0)) || (mode == "filter" && req.Filter is null)) { await Send.ResponseAsync(KbxValidationProblem.Create( new KbxValidationError("selection", null, "ORDER_SELECTION_REQUIRED", "출고지시할 주문을 선택하세요.")), 400, cancellation: ct); return; } await using var connection = await dataSource.OpenConnectionAsync(ct); await using var tx = await connection.BeginTransactionAsync(ct); var replay = await connection.QuerySingleOrDefaultAsync(new CommandDefinition( "select response_json::text from kbx.command_receipts where operation_id=@OperationId and idempotency_key=@Key", new { OperationId = "oms.orders.ship", Key = key }, tx, cancellationToken: ct)); if (replay is not null) { await tx.RollbackAsync(ct); await Send.OkAsync(JsonSerializer.Deserialize(replay)!, ct); return; } const string idsTarget = "select distinct unnest(@Ids::uuid[]) as id"; const string filterTarget = """ select o.id from oms_order_search_projection o where o.ordered_at >= @From and o.ordered_at < @ToExclusive and (@ChannelId is null or o.channel_id = @ChannelId) and (@Status is null or o.shipment_status = @Status) and (@Keyword is null or o.search_text ilike '%' || @Keyword || '%') and (@ExceptionOnly = false or o.exception_count > 0) and not (o.id = any(@ExcludedIds)) """; var targetSql = mode == "ids" ? idsTarget : filterTarget; var filter = req.Filter; var args = new { Ids = (req.Ids ?? Array.Empty()).ToArray(), From = filter is null ? (DateTime?)null : filter.From.ToDateTime(TimeOnly.MinValue), ToExclusive = filter is null ? (DateTime?)null : filter.To.AddDays(1).ToDateTime(TimeOnly.MinValue), ChannelId = filter?.ChannelId, Status = string.IsNullOrWhiteSpace(filter?.Status) ? null : filter!.Status, Keyword = string.IsNullOrWhiteSpace(filter?.Keyword) ? null : filter!.Keyword!.Trim(), ExceptionOnly = filter?.ExceptionOnly ?? false, ExcludedIds = (req.ExcludedIds ?? Array.Empty()).ToArray(), }; var result = await connection.QuerySingleAsync(new CommandDefinition($""" with target as materialized ( {targetSql} ), accepted as materialized ( update oms.orders set status='CONFIRMED', version=version+1, updated_at=now(), updated_by='web' where id in (select id from target) and status in ('NEW','DRAFT','READY') returning id ), audit_insert as ( insert into audit.entries(id,aggregate_type,aggregate_id,action,actor,occurred_at,data) select gen_random_uuid(),'Order',id,'ShipRequested','web',now(),'{}'::jsonb from accepted returning 1 ), outbox_insert as ( insert into integration.outbox(id,event_type,aggregate_id,payload,occurred_at,status) select gen_random_uuid(),'OmsOrderShipRequested',id,jsonb_build_object('orderId',id),now(),'PENDING' from accepted returning 1 ) select (select count(*)::int from target) as Requested, (select count(*)::int from accepted) as Accepted; """, args, tx, cancellationToken: ct)); if (result.Requested == 0) { await tx.RollbackAsync(ct); await Send.ResponseAsync(KbxValidationProblem.Create( new KbxValidationError("selection", null, "ORDER_SELECTION_EMPTY", "현재 검색조건에서 처리할 주문이 없습니다.")), 400, cancellation: ct); return; } var response = new ShipOrdersResponse(result.Requested, result.Accepted, result.Requested - result.Accepted); await connection.ExecuteAsync(new CommandDefinition(""" insert into kbx.command_receipts(operation_id,idempotency_key,response_json) values(@OperationId,@Key,cast(@Response as jsonb)); """, new { OperationId = "oms.orders.ship", Key = key, Response = JsonSerializer.Serialize(response) }, tx, cancellationToken: ct)); await tx.CommitAsync(ct); await Send.OkAsync(response, ct); } private sealed record ShipMutationResult(int Requested, int Accepted); }