269 lines
11 KiB
C#
269 lines
11 KiB
C#
using System.Text.Json;
|
|
using Dapper;
|
|
using Modules.WMS.Picking.Shared;
|
|
using Npgsql;
|
|
using Shared.Problems;
|
|
|
|
namespace Modules.WMS.Picking.Scan;
|
|
|
|
public sealed class Handler(NpgsqlDataSource dataSource)
|
|
{
|
|
public sealed record HandlerResult(int StatusCode, object Body);
|
|
|
|
public async Task<HandlerResult> HandleAsync(
|
|
Guid taskId,
|
|
Request request,
|
|
string actor,
|
|
CancellationToken ct)
|
|
{
|
|
if (string.IsNullOrWhiteSpace(request.IdempotencyKey))
|
|
return new(400, KbxValidationProblem.Create(
|
|
new KbxValidationError("idempotencyKey", null, "IDEMPOTENCY_REQUIRED", "Idempotency key가 필요합니다.")));
|
|
|
|
var barcode = request.Barcode.Trim();
|
|
if (barcode.Length < 3)
|
|
return new(400, KbxValidationProblem.Create(
|
|
new KbxValidationError("barcode", null, "BARCODE_INVALID", "바코드를 다시 스캔하세요.")));
|
|
|
|
await using var connection = await dataSource.OpenConnectionAsync(ct);
|
|
await using var tx = await connection.BeginTransactionAsync(ct);
|
|
|
|
// One picking task is a sequence. Serialize scanner commands for this task.
|
|
var taskRow = await connection.QuerySingleOrDefaultAsync<TaskRow>(new CommandDefinition("""
|
|
select id, task_no as TaskNo, status, version
|
|
from wms.picking_tasks
|
|
where id=@TaskId
|
|
for update;
|
|
""", new { TaskId = taskId }, tx, cancellationToken: ct));
|
|
|
|
if (taskRow is null)
|
|
{
|
|
await tx.RollbackAsync(ct);
|
|
return new(404, new { type = "not-found", title = "피킹 작업을 찾을 수 없습니다." });
|
|
}
|
|
|
|
// Idempotency is checked before version. A response-lost retry must replay the committed result.
|
|
var existingJson = await connection.QuerySingleOrDefaultAsync<string>(new CommandDefinition("""
|
|
select response_payload::text
|
|
from wms.scan_receipts
|
|
where task_id=@TaskId and idempotency_key=@Key;
|
|
""", new { TaskId = taskId, Key = request.IdempotencyKey }, tx, cancellationToken: ct));
|
|
if (existingJson is not null)
|
|
{
|
|
var existing = JsonSerializer.Deserialize<Response>(existingJson, JsonOptions)!;
|
|
await tx.RollbackAsync(ct);
|
|
return new(200, existing with { Duplicate = true });
|
|
}
|
|
|
|
if (taskRow.Version != request.ExpectedVersion)
|
|
{
|
|
await tx.RollbackAsync(ct);
|
|
return new(409, KbxConflictProblem.Version(taskRow.Version));
|
|
}
|
|
|
|
if (taskRow.Status != "IN_PROGRESS")
|
|
{
|
|
await tx.RollbackAsync(ct);
|
|
return new(409, KbxBusinessProblem.Create(
|
|
"PICKING_NOT_ACTIVE",
|
|
"현재 피킹할 수 없는 작업입니다.",
|
|
$"현재 상태: {taskRow.Status}"));
|
|
}
|
|
|
|
var line = await connection.QuerySingleOrDefaultAsync<LineRow>(new CommandDefinition("""
|
|
select l.id as LineId,
|
|
l.line_no as LineNo,
|
|
l.location_code as LocationCode,
|
|
l.location_barcode as LocationBarcode,
|
|
l.location_confirmed as LocationConfirmed,
|
|
l.item_id as ItemId,
|
|
l.barcode as Barcode,
|
|
l.required_qty as RequiredQty,
|
|
l.picked_qty as PickedQty
|
|
from wms.picking_lines l
|
|
where l.task_id=@TaskId and l.status <> 'COMPLETED'
|
|
order by l.line_no
|
|
limit 1
|
|
for update;
|
|
""", new { TaskId = taskId }, tx, cancellationToken: ct));
|
|
|
|
if (line is null)
|
|
{
|
|
await CompleteTaskAsync(connection, tx, taskId, ct);
|
|
var completed = await BuildAndPersistResponse(
|
|
connection, tx, taskId, request, actor, true, "success", "피킹을 완료했습니다.", "TaskRecoveredAsCompleted", ct);
|
|
await tx.CommitAsync(ct);
|
|
return new(200, completed);
|
|
}
|
|
|
|
var decision = PickingScanStateMachine.Decide(new PickingScanState(
|
|
line.LocationConfirmed,
|
|
line.LocationCode,
|
|
line.LocationBarcode,
|
|
line.Barcode,
|
|
line.RequiredQty,
|
|
line.PickedQty), barcode);
|
|
|
|
switch (decision.Kind)
|
|
{
|
|
case PickingScanDecisionKind.RejectLocation:
|
|
case PickingScanDecisionKind.RejectItem:
|
|
{
|
|
var rejected = await BuildAndPersistResponse(
|
|
connection, tx, taskId, request, actor,
|
|
decision.Accepted, decision.Feedback, decision.Message, decision.AuditAction, ct);
|
|
await tx.CommitAsync(ct);
|
|
return new(422, rejected);
|
|
}
|
|
|
|
case PickingScanDecisionKind.AcceptLocation:
|
|
{
|
|
await connection.ExecuteAsync(new CommandDefinition("""
|
|
update wms.picking_lines
|
|
set location_confirmed=true, updated_at=now()
|
|
where id=@LineId;
|
|
update wms.picking_tasks
|
|
set version=version+1, updated_at=now()
|
|
where id=@TaskId;
|
|
""", new { line.LineId, TaskId = taskId }, tx, cancellationToken: ct));
|
|
|
|
var accepted = await BuildAndPersistResponse(
|
|
connection, tx, taskId, request, actor,
|
|
true, decision.Feedback, decision.Message, decision.AuditAction, ct);
|
|
await tx.CommitAsync(ct);
|
|
return new(200, accepted);
|
|
}
|
|
|
|
case PickingScanDecisionKind.AcceptItem:
|
|
{
|
|
await ApplyItemScanAsync(connection, tx, taskId, line, decision, ct);
|
|
|
|
var nextExists = await HasIncompleteLineAsync(connection, tx, taskId, ct);
|
|
if (!nextExists)
|
|
await CompleteTaskAsync(connection, tx, taskId, ct);
|
|
|
|
var message = nextExists ? decision.Message : "피킹을 완료했습니다.";
|
|
var accepted = await BuildAndPersistResponse(
|
|
connection, tx, taskId, request, actor,
|
|
true, decision.Feedback, message, decision.AuditAction, ct);
|
|
await tx.CommitAsync(ct);
|
|
return new(200, accepted);
|
|
}
|
|
|
|
default:
|
|
throw new InvalidOperationException($"Unsupported picking scan decision: {decision.Kind}");
|
|
}
|
|
}
|
|
|
|
private static async Task ApplyItemScanAsync(
|
|
NpgsqlConnection connection,
|
|
NpgsqlTransaction tx,
|
|
Guid taskId,
|
|
LineRow line,
|
|
PickingScanDecision decision,
|
|
CancellationToken ct)
|
|
{
|
|
await connection.ExecuteAsync(new CommandDefinition("""
|
|
update wms.picking_lines
|
|
set picked_qty=@PickedQty,
|
|
status=case when @Completed then 'COMPLETED' else 'IN_PROGRESS' end,
|
|
completed_at=case when @Completed then now() else completed_at end,
|
|
updated_at=now()
|
|
where id=@LineId;
|
|
|
|
update wms.picking_tasks set version=version+1, updated_at=now() where id=@TaskId;
|
|
""", new {
|
|
PickedQty = decision.NewPickedQty!.Value,
|
|
Completed = decision.LineCompleted,
|
|
line.LineId,
|
|
TaskId = taskId,
|
|
}, tx, cancellationToken: ct));
|
|
|
|
if (decision.LineCompleted)
|
|
await PickingTransitions.AutoConfirmNextSameLocationAsync(connection, tx, taskId, line.LocationCode, ct);
|
|
}
|
|
|
|
private static Task<bool> HasIncompleteLineAsync(
|
|
NpgsqlConnection connection,
|
|
NpgsqlTransaction tx,
|
|
Guid taskId,
|
|
CancellationToken ct) => connection.ExecuteScalarAsync<bool>(new CommandDefinition("""
|
|
select exists(select 1 from wms.picking_lines where task_id=@TaskId and status <> 'COMPLETED');
|
|
""", new { TaskId = taskId }, tx, cancellationToken: ct));
|
|
|
|
private static async Task CompleteTaskAsync(
|
|
NpgsqlConnection connection,
|
|
NpgsqlTransaction tx,
|
|
Guid taskId,
|
|
CancellationToken ct)
|
|
{
|
|
await connection.ExecuteAsync(new CommandDefinition("""
|
|
update wms.picking_tasks
|
|
set status='COMPLETED', version=version+1, completed_at=coalesce(completed_at,now()), updated_at=now()
|
|
where id=@TaskId and status <> 'COMPLETED';
|
|
|
|
insert into integration.outbox(id,event_type,aggregate_id,payload,occurred_at,status)
|
|
select @EventId,'WmsPickingCompleted',@TaskId,jsonb_build_object('taskId',@TaskId),now(),'PENDING'
|
|
where not exists (
|
|
select 1 from integration.outbox
|
|
where aggregate_id=@TaskId and event_type='WmsPickingCompleted'
|
|
);
|
|
""", new { TaskId = taskId, EventId = Guid.NewGuid() }, tx, cancellationToken: ct));
|
|
}
|
|
|
|
private static async Task<Response> BuildAndPersistResponse(
|
|
NpgsqlConnection connection,
|
|
NpgsqlTransaction tx,
|
|
Guid taskId,
|
|
Request request,
|
|
string actor,
|
|
bool accepted,
|
|
string feedback,
|
|
string message,
|
|
string auditAction,
|
|
CancellationToken ct)
|
|
{
|
|
var current = await PickingTaskQueries.GetAsync(connection, tx, taskId, ct)
|
|
?? throw new InvalidOperationException("Picking task disappeared during transaction.");
|
|
var response = new Response(accepted, false, current, feedback, message);
|
|
var json = JsonSerializer.Serialize(response, JsonOptions);
|
|
|
|
await connection.ExecuteAsync(new CommandDefinition("""
|
|
insert into wms.scan_receipts(
|
|
id, task_id, idempotency_key, barcode, source, occurred_at, actor, response_payload)
|
|
values (@Id,@TaskId,@Key,@Barcode,@Source,@OccurredAt,@Actor,cast(@Response as jsonb));
|
|
|
|
insert into audit.entries(id,aggregate_type,aggregate_id,action,actor,occurred_at,data)
|
|
values (@AuditId,'WmsPickingTask',@TaskId,@Action,@Actor,now(),
|
|
jsonb_build_object('barcode',@Barcode,'source',@Source,'idempotencyKey',@Key));
|
|
""", new {
|
|
Id = Guid.NewGuid(),
|
|
AuditId = Guid.NewGuid(),
|
|
TaskId = taskId,
|
|
Key = request.IdempotencyKey,
|
|
request.Barcode,
|
|
request.Source,
|
|
request.OccurredAt,
|
|
Actor = actor,
|
|
Response = json,
|
|
Action = auditAction,
|
|
}, tx, cancellationToken: ct));
|
|
|
|
return response;
|
|
}
|
|
|
|
private static readonly JsonSerializerOptions JsonOptions = new(JsonSerializerDefaults.Web);
|
|
|
|
private sealed record TaskRow(Guid Id, string TaskNo, string Status, long Version);
|
|
private sealed record LineRow(
|
|
Guid LineId,
|
|
int LineNo,
|
|
string LocationCode,
|
|
string LocationBarcode,
|
|
bool LocationConfirmed,
|
|
Guid ItemId,
|
|
string Barcode,
|
|
decimal RequiredQty,
|
|
decimal PickedQty);
|
|
}
|